diff --git a/internal/outbound/provider_submit.go b/internal/outbound/provider_submit.go
new file mode 100644
index 000000000..79668a82b
--- /dev/null
+++ b/internal/outbound/provider_submit.go
@@ -0,0 +1,332 @@
+package outbound
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/tokencanopy/e2a/internal/delivery"
+	"github.com/tokencanopy/e2a/internal/sendingpolicy"
+)
+
+// This file is the provider seam: the one place a customer-bound message
+// becomes an SMTP transaction with SES.
+//
+// Everything above it — composition, DKIM, footers, recipient normalization —
+// is deliberately token-free, because none of it exposes the shared SES
+// reputation. Opening the socket does. So the socket is the thing that
+// requires a sendingpolicy.ProviderAuthorization, and the adapter redeems that
+// single-use token immediately before dialing, not when the job was picked up
+// and not when the message was composed. A decision that went stale in between
+// — a pause, a plan change, a policy rotation, a duplicate worker — invalidates
+// the token instead of being raced.
+//
+// The adapter also owns three headers SES reads for provider-side isolation and
+// attribution: X-SES-TENANT, X-E2A-Provider-Attempt, and
+// X-SES-CONFIGURATION-SET (plus the stable X-E2A-Message-ID correlation marker
+// beside them). Their values come only from the token and this deployment's
+// configuration. Whatever the composed MIME already carried under those names
+// is removed first, every occurrence, so neither a customer nor an upstream
+// compose bug can smuggle or duplicate a tenant, attempt, or configuration-set
+// selector.
+
+// ProviderAttemptHeader carries the random attempt correlation id SES echoes
+// back in delivery feedback. It is the fallback lookup when the worker died
+// between SES accepting the message and the provider id being stored.
+const ProviderAttemptHeader = "X-E2A-Provider-Attempt"
+
+// SESTenantHeader names the SES tenant a submission is attributed to.
+const SESTenantHeader = "X-SES-TENANT"
+
+// SESConfigurationSetHeader selects the SES configuration set (delivery
+// feedback destination) for a submission.
+const SESConfigurationSetHeader = "X-SES-CONFIGURATION-SET"
+
+// Sentinel errors for the provider seam. Every one of them is returned before
+// any network I/O and before the token is redeemed.
+var (
+	// ErrAuthorizationRequired means SubmitOnce was called without a token.
+	// There is no tokenless path to the provider by construction; a caller
+	// hitting this has bypassed the gate.
+	ErrAuthorizationRequired = errors.New("outbound: provider submission requires an authorization")
+	// ErrTenantNameMissing means the token demands a tenant header but carries
+	// no tenant name. The gate refuses to mint such a token; the adapter checks
+	// again because the header is the provider-side isolation boundary and
+	// must never be emitted empty.
+	ErrTenantNameMissing = errors.New("outbound: authorization requires a tenant header but names no tenant")
+	// ErrProviderHeaderValue means a provider-owned header value carries a
+	// line break. The values come from the gate, so this is a defect, not
+	// input — and a defect here is a header injection, so it fails closed
+	// rather than being sanitized silently.
+	ErrProviderHeaderValue = errors.New("outbound: provider header value contains a line break")
+	// ErrMalformedHeaderSection means the composed MIME's header section holds
+	// a bare carriage return. Receivers disagree on whether a lone CR ends a
+	// line, so a header hidden behind one might survive stripping here and
+	// still be honoured by the provider. The composer never emits one — every
+	// header value is sanitized — so this only fires on a compose defect, and
+	// it fails the send rather than guess.
+	ErrMalformedHeaderSection = errors.New("outbound: message header section contains a bare carriage return")
+)
+
+// providerOwnedHeaders are removed from the composed MIME before submission,
+// matched case-insensitively, folded continuations included.
+var providerOwnedHeaders = map[string]struct{}{
+	strings.ToLower(SESTenantHeader):           {},
+	strings.ToLower(ProviderAttemptHeader):     {},
+	strings.ToLower(SESConfigurationSetHeader): {},
+	strings.ToLower(delivery.MessageIDHeader):  {},
+}
+
+// Envelope is what a caller hands the provider seam: the SMTP envelope and the
+// composed wire bytes.
+//
+// There is deliberately no message id here. The stable X-E2A-Message-ID marker
+// that delivery feedback keys on is derived from the token — a customer
+// message's operation IS its message id — so a caller cannot stamp one
+// message's id on another's send and misroute its bounces.
+//
+// The sender is the one envelope field the token does not bind: the
+// authorization carries recipients and tenant, not MAIL FROM. SES enforces
+// identity ownership of the sender domain on its side, and the caller here is
+// the trusted worker that composed the message, so the seam only insists the
+// sender is present. Binding it would need the gate to learn the composed
+// sender at acceptance; that is a gate change, not an adapter one.
+type Envelope struct {
+	// From is the SMTP MAIL FROM address.
+	From string
+	// Recipients is the exact final envelope: one entry per distinct mailbox,
+	// and exactly the set the authorization was minted for. Order is free.
+	Recipients []string
+	// Message is the composed MIME. Provider-owned headers in it are stripped.
+	Message []byte
+}
+
+// ProviderResult reports one accepted provider submission.
+type ProviderResult struct {
+	// ProviderMessageID is the id SES assigned on acceptance.
+	ProviderMessageID string
+	// Attempt is the durable attempt that was redeemed for this call.
+	Attempt sendingpolicy.AttemptRef
+	// SettlementErr is set when SES accepted the message but the local
+	// settlement did not commit. The send HAPPENED; the caller must retry
+	// SettleProvider with the same attempt and provider id, and must never
+	// resubmit. Delivery feedback carrying the attempt header is the fallback
+	// if it never does.
+	//
+	// One value is not a retry: errors.Is(SettlementErr,
+	// sendingpolicy.ErrProviderMessageIDConflict) means this attempt was
+	// already settled with a DIFFERENT provider id — two physical sends for one
+	// charge. Retrying settlement cannot fix that; it must be surfaced as an
+	// invariant violation, not absorbed as a transient.
+	SettlementErr error
+}
+
+// ProviderSubmitter is the token-requiring adapter over the SMTP relay.
+type ProviderSubmitter struct {
+	relay        *SMTPRelay
+	gate         sendingpolicy.Gate
+	sesConfigSet string
+}
+
+// NewProviderSubmitter binds the relay to the gate whose tokens it honors.
+func NewProviderSubmitter(relay *SMTPRelay, gate sendingpolicy.Gate) *ProviderSubmitter {
+	return &ProviderSubmitter{relay: relay, gate: gate}
+}
+
+// SetSESConfigurationSet names the configuration set every submission is
+// tagged with. Empty means no header (dev/self-host without SES).
+func (s *ProviderSubmitter) SetSESConfigurationSet(name string) { s.sesConfigSet = name }
+
+// SubmitOnce makes exactly one provider call for one authorized attempt.
+//
+// The sequence is fixed and every early exit is I/O-free: prove the envelope is
+// the authorized one, derive the provider headers from the token, rewrite the
+// wire bytes, redeem the token, THEN dial. A relay-level retry is not offered
+// here on purpose — each physical submission exposes SES once and must be
+// charged once, so a retry is a new attempt with a new token, obtained by the
+// caller through the gate.
+//
+// Outcomes: a definite permanent rejection is settled as such and returned; an
+// acceptance is settled with the provider id and returned as a result. Anything
+// ambiguous (4xx, connection loss, cancellation) is returned unsettled, because
+// a message that might have been delivered must not release anything.
+func (s *ProviderSubmitter) SubmitOnce(ctx context.Context, auth sendingpolicy.ProviderAuthorization, env Envelope) (ProviderResult, error) {
+	if s == nil || s.relay == nil || s.gate == nil {
+		return ProviderResult{}, errors.New("outbound: provider submitter is not wired")
+	}
+	if auth.IsZero() {
+		return ProviderResult{}, ErrAuthorizationRequired
+	}
+	if strings.TrimSpace(env.From) == "" {
+		return ProviderResult{}, errors.New("outbound: envelope sender is empty")
+	}
+	headers, err := auth.ValidateEnvelope(env.Recipients)
+	if err != nil {
+		return ProviderResult{}, err
+	}
+	provider, err := providerHeaderLines(headers, s.sesConfigSet, correlationMessageID(auth))
+	if err != nil {
+		return ProviderResult{}, err
+	}
+	stripped, err := stripProviderHeaders(env.Message)
+	if err != nil {
+		return ProviderResult{}, err
+	}
+	// Refuse a misconfigured relay before the token is spent. Redeeming first
+	// would invalidate the attempt for a failure that had nothing to do with
+	// the message, and the caller would burn a fresh ordinal per retry.
+	if !s.relay.Configured() {
+		return ProviderResult{}, fmt.Errorf("outbound SMTP relay not configured")
+	}
+	wire := append(provider, stripped...)
+
+	if err := s.gate.RedeemProviderCall(ctx, auth); err != nil {
+		return ProviderResult{}, fmt.Errorf("provider authorization: %w", err)
+	}
+
+	// RCPT TO is issued from the token's canonical envelope, not the caller's
+	// spelling of it. ValidateEnvelope has just proved the two name the same
+	// mailboxes; what goes on the wire is the normalized set the budget priced,
+	// so a padded or upper-cased entry cannot turn into an SMTP grammar error
+	// that downstream classifies as the message's own permanent failure.
+	//
+	// A failure after the body was fully written (ErrProviderAcceptanceUnknown)
+	// is neither accepted nor rejected here: it is returned unsettled, because
+	// the provider may hold the message and only its feedback can say.
+	providerID, sendErr := s.relay.SendOnceContext(ctx, env.From, auth.AuthorizedRecipients(), wire)
+	if sendErr != nil {
+		// IsPermanentSMTPError is the worker's retry classifier: any 5xx,
+		// including one raised before DATA (an AUTH 535, say). Settling such a
+		// failure as a rejection is conservative in the only direction that
+		// matters — the provider never took the message, so giving its
+		// capacity back is correct — and it keeps this seam's verdict identical
+		// to the one the worker already acts on.
+		if IsPermanentSMTPError(sendErr) {
+			if err := s.gate.SettleProvider(ctx, sendingpolicy.ProviderSettlement{
+				Attempt: auth.Attempt(),
+				Outcome: sendingpolicy.SettlementProviderPermanentlyRejected,
+			}); err != nil {
+				return ProviderResult{}, errors.Join(sendErr, fmt.Errorf("settle rejection: %w", err))
+			}
+		}
+		return ProviderResult{}, sendErr
+	}
+
+	result := ProviderResult{ProviderMessageID: providerID, Attempt: auth.Attempt()}
+	if err := s.gate.SettleProvider(ctx, sendingpolicy.ProviderSettlement{
+		Attempt:           auth.Attempt(),
+		Outcome:           sendingpolicy.SettlementProviderAccepted,
+		ProviderMessageID: providerID,
+	}); err != nil {
+		result.SettlementErr = fmt.Errorf("settle acceptance: %w", err)
+	}
+	return result, nil
+}
+
+// correlationMessageID is the value of the stable X-E2A-Message-ID marker for
+// a token: the message id for a customer message, nothing for every other
+// purpose (operational mail has no message row for feedback to land on).
+func correlationMessageID(auth sendingpolicy.ProviderAuthorization) string {
+	if auth.Purpose() == sendingpolicy.PurposeCustomerMessage {
+		return auth.Attempt().OperationID()
+	}
+	return ""
+}
+
+// providerHeaderLines renders the provider-owned header block from the token's
+// view and the deployment configuration — and from nothing else.
+//
+// Callers cannot pass a tenant name, correlation id, or message id; the only
+// way to change what SES receives is to change what the gate authorized. The
+// order — configuration set, then message id — matches what Sender.SubmitOnce
+// emits today, so swapping the worker onto this seam is byte-identical for
+// the headers both paths share.
+func providerHeaderLines(h sendingpolicy.ProviderHeaders, sesConfigSet, messageID string) ([]byte, error) {
+	if h.TenantRequired && strings.TrimSpace(h.TenantName) == "" {
+		return nil, ErrTenantNameMissing
+	}
+	for _, v := range []string{h.AttemptCorrelationID, h.TenantName, sesConfigSet, messageID} {
+		if strings.ContainsAny(v, "\r\n") {
+			return nil, ErrProviderHeaderValue
+		}
+	}
+	var b bytes.Buffer
+	if sesConfigSet != "" {
+		b.WriteString(SESConfigurationSetHeader + ": " + sesConfigSet + "\r\n")
+	}
+	if messageID != "" {
+		b.WriteString(delivery.MessageIDHeader + ": " + messageID + "\r\n")
+	}
+	if h.AttemptCorrelationID != "" {
+		b.WriteString(ProviderAttemptHeader + ": " + h.AttemptCorrelationID + "\r\n")
+	}
+	if h.TenantRequired {
+		b.WriteString(SESTenantHeader + ": " + h.TenantName + "\r\n")
+	}
+	return b.Bytes(), nil
+}
+
+// stripProviderHeaders removes every provider-owned header field from the
+// header section of a message, leaving every other byte — including the body
+// and the original line endings — untouched.
+//
+// It walks the header section line by line rather than parsing it as a
+// message: the bytes were composed by this package or arrived from a customer,
+// and a parser that normalized them would change what DKIM signed. A field is
+// its name line plus every folded continuation (a line starting with space or
+// tab); dropping a field drops its continuations with it. Matching is on the
+// lowercased name, so mixed-case spellings are not an evasion.
+//
+// Lines end in LF or CRLF. A carriage return anywhere else in the header
+// section is refused (ErrMalformedHeaderSection): a receiver that treats a
+// bare CR as a line break would see a header this walker did not, and the
+// composer never produces one.
+func stripProviderHeaders(msg []byte) ([]byte, error) {
+	// A message whose first line is a folded continuation has nothing to
+	// continue — except the provider header this adapter is about to prepend,
+	// whose value it would silently extend. Refuse it.
+	if len(msg) > 0 && (msg[0] == ' ' || msg[0] == '\t') {
+		return nil, ErrMalformedHeaderSection
+	}
+	out := make([]byte, 0, len(msg))
+	rest := msg
+	dropping := false
+	for len(rest) > 0 {
+		nl := bytes.IndexByte(rest, '\n')
+		var line []byte
+		if nl < 0 {
+			line, rest = rest, nil
+		} else {
+			line, rest = rest[:nl+1], rest[nl+1:]
+		}
+		if raw := bytes.TrimSuffix(bytes.TrimSuffix(line, []byte("\n")), []byte("\r")); bytes.IndexByte(raw, '\r') >= 0 {
+			return nil, ErrMalformedHeaderSection
+		}
+		trimmed := bytes.TrimRight(line, "\r\n")
+		if len(trimmed) == 0 {
+			// End of the header section: emit the separator and the body
+			// verbatim.
+			out = append(out, line...)
+			out = append(out, rest...)
+			return out, nil
+		}
+		if trimmed[0] == ' ' || trimmed[0] == '\t' {
+			if !dropping {
+				out = append(out, line...)
+			}
+			continue
+		}
+		dropping = false
+		if colon := bytes.IndexByte(trimmed, ':'); colon > 0 {
+			name := strings.ToLower(strings.TrimSpace(string(trimmed[:colon])))
+			if _, owned := providerOwnedHeaders[name]; owned {
+				dropping = true
+				continue
+			}
+		}
+		out = append(out, line...)
+	}
+	return out, nil
+}
diff --git a/internal/outbound/provider_submit_internal_test.go b/internal/outbound/provider_submit_internal_test.go
new file mode 100644
index 000000000..56b8238e5
--- /dev/null
+++ b/internal/outbound/provider_submit_internal_test.go
@@ -0,0 +1,130 @@
+package outbound
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/tokencanopy/e2a/internal/sendingpolicy"
+)
+
+// Pure unit tests over the adapter's two rewriting steps. The DB-backed
+// tests live in the external test package, because the shared test database
+// helper transitively imports this package.
+
+// SmuggledMIME is exported for the external test package.
+var SmuggledMIME = smuggledMIME
+
+// smuggledMIME is customer-shaped wire bytes that try every spelling of the
+// provider-owned headers, including a folded one and a body decoy.
+func smuggledMIME() []byte {
+	return []byte("From: agent@agents.e2a.dev\r\n" +
+		"x-ses-tenant: smuggled-lower\r\n" +
+		"X-SES-TENANT: smuggled-upper\r\n" +
+		"X-Ses-Tenant: smuggled-\r\n folded\r\n" +
+		"X-E2A-Provider-Attempt: cor_forged\r\n" +
+		"x-e2a-provider-attempt: cor_forged_two\r\n" +
+		"X-SES-CONFIGURATION-SET: attacker-set\r\n" +
+		"x-ses-configuration-set:\r\n\tattacker-folded\r\n" +
+		"X-E2A-Message-ID: msg_forged\r\n" +
+		"Subject: hello\r\n" +
+		"\r\n" +
+		"X-SES-TENANT: body-decoy\r\n" +
+		"body line\r\n")
+}
+
+func TestProviderHeaderLinesFailClosed(t *testing.T) {
+	for name, tc := range map[string]struct {
+		h       sendingpolicy.ProviderHeaders
+		set, id string
+		wantErr error
+		want    string
+	}{
+		"tenant required but empty": {
+			h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1", TenantRequired: true}, wantErr: ErrTenantNameMissing,
+		},
+		"tenant required but blank": {
+			h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1", TenantRequired: true, TenantName: " \t"}, wantErr: ErrTenantNameMissing,
+		},
+		"line break in tenant": {
+			h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1", TenantRequired: true, TenantName: "t\r\nBcc: x"}, wantErr: ErrProviderHeaderValue,
+		},
+		"line break in config set": {
+			h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1"}, set: "a\nb", wantErr: ErrProviderHeaderValue,
+		},
+		"line break in message id": {
+			h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1"}, id: "m\r", wantErr: ErrProviderHeaderValue,
+		},
+		"no tenant, no set, no id": {
+			h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1"}, want: "X-E2A-Provider-Attempt: cor_1\r\n",
+		},
+		"everything, in the legacy path's order": {
+			h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1", TenantRequired: true, TenantName: "tenant_a"}, set: "cs", id: "msg_1",
+			want: "X-SES-CONFIGURATION-SET: cs\r\nX-E2A-Message-ID: msg_1\r\nX-E2A-Provider-Attempt: cor_1\r\nX-SES-TENANT: tenant_a\r\n",
+		},
+	} {
+		got, err := providerHeaderLines(tc.h, tc.set, tc.id)
+		if !errors.Is(err, tc.wantErr) {
+			t.Errorf("%s: err = %v, want %v", name, err, tc.wantErr)
+		}
+		if string(got) != tc.want {
+			t.Errorf("%s: headers = %q, want %q", name, got, tc.want)
+		}
+	}
+}
+
+func TestStripProviderHeaders(t *testing.T) {
+	for name, tc := range map[string]struct {
+		in, want string
+		wantErr  error
+	}{
+		"mixed case, duplicates, folded": {
+			in:   string(smuggledMIME()),
+			want: "From: agent@agents.e2a.dev\r\nSubject: hello\r\n\r\nX-SES-TENANT: body-decoy\r\nbody line\r\n",
+		},
+		"lf-only line endings": {
+			in:   "X-SES-TENANT: a\nSubject: s\nx-ses-configuration-set: b\n c\n\nbody\n",
+			want: "Subject: s\n\nbody\n",
+		},
+		"headers only, no body separator": {
+			in:   "Subject: s\r\nX-E2A-Provider-Attempt: cor\r\n\tfolded\r\n",
+			want: "Subject: s\r\n",
+		},
+		"nothing to strip": {
+			in:   "Subject: s\r\nTo: a@example.test\r\n\r\nbody\r\n",
+			want: "Subject: s\r\nTo: a@example.test\r\n\r\nbody\r\n",
+		},
+		"name prefix is not a match": {
+			in:   "X-SES-TENANT-EXTRA: keep\r\nX-SES-TENANTS: keep\r\n\r\n",
+			want: "X-SES-TENANT-EXTRA: keep\r\nX-SES-TENANTS: keep\r\n\r\n",
+		},
+		"whitespace before colon still matches": {
+			in:   "X-SES-TENANT : a\r\nSubject: s\r\n\r\n",
+			want: "Subject: s\r\n\r\n",
+		},
+		"continuation after a kept header is kept": {
+			in:   "Subject: long\r\n subject\r\nX-SES-TENANT: a\r\n b\r\nTo: x@example.test\r\n\r\n",
+			want: "Subject: long\r\n subject\r\nTo: x@example.test\r\n\r\n",
+		},
+		"body may contain bare CR": {
+			in:   "Subject: s\r\n\r\nbinary\rbody\r\n",
+			want: "Subject: s\r\n\r\nbinary\rbody\r\n",
+		},
+		"bare CR hiding a header is refused": {
+			in:      "Subject: a\rX-SES-TENANT: evil\r\n\r\nbody\r\n",
+			wantErr: ErrMalformedHeaderSection,
+		},
+		"CR CR LF pseudo-separator is refused": {
+			in:      "Subject: s\r\n\r\r\nX-SES-TENANT: evil\r\n\r\nbody",
+			wantErr: ErrMalformedHeaderSection,
+		},
+		"empty": {in: "", want: ""},
+	} {
+		got, err := stripProviderHeaders([]byte(tc.in))
+		if !errors.Is(err, tc.wantErr) {
+			t.Errorf("%s: err = %v, want %v", name, err, tc.wantErr)
+		}
+		if string(got) != tc.want {
+			t.Errorf("%s:\n got %q\nwant %q", name, got, tc.want)
+		}
+	}
+}
diff --git a/internal/outbound/provider_submit_test.go b/internal/outbound/provider_submit_test.go
new file mode 100644
index 000000000..c33a3e834
--- /dev/null
+++ b/internal/outbound/provider_submit_test.go
@@ -0,0 +1,925 @@
+package outbound_test
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"fmt"
+	"math/rand"
+	"net"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/tokencanopy/e2a/internal/config"
+	"github.com/tokencanopy/e2a/internal/delivery"
+	"github.com/tokencanopy/e2a/internal/outbound"
+	"github.com/tokencanopy/e2a/internal/sendingpolicy"
+	"github.com/tokencanopy/e2a/internal/testutil"
+)
+
+// These tests drive the real gate against real Postgres and a real (fake)
+// SMTP listener. Every "no network call" claim is asserted against a socket
+// counter, never against the absence of an error, because the failure that
+// matters is a connection that happened anyway. Every address is synthetic.
+
+const (
+	psHMAC     = `{"active":1,"keys":{"1":"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"}}`
+	psOperator = `{"commitment_key":"AgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgI","recipients":{"1":"submit-operator@example.test"}}`
+)
+
+type gateFixture struct {
+	t      *testing.T
+	ctx    context.Context
+	pool   *pgxpool.Pool
+	gate   sendingpolicy.Gate
+	userID string
+	agent  string
+	tenant string
+}
+
+var psSeq int
+
+func psID(prefix string) string {
+	psSeq++
+	return fmt.Sprintf("%s_ps_%d_%x", prefix, psSeq, rand.Uint32())
+}
+
+// newGateFixture builds an enforcing config-source gate with one standard
+// account owning one shared-domain agent. mutate adjusts the policy.
+func newGateFixture(t *testing.T, mutate func(*sendingpolicy.RuntimePolicy)) *gateFixture {
+	t.Helper()
+	ctx := context.Background()
+	pool := testutil.TestDB(t)
+	keyring, err := sendingpolicy.LoadKeyring(psHMAC)
+	if err != nil {
+		t.Fatalf("load keyring: %v", err)
+	}
+	recipients, err := sendingpolicy.LoadOperatorRecipients(psOperator)
+	if err != nil {
+		t.Fatalf("load operator map: %v", err)
+	}
+	secrets := sendingpolicy.Secrets{Keyring: keyring, Recipients: recipients}
+	if _, err := sendingpolicy.NewModule(pool, secrets).RegisterOperatorRecipients(ctx, "fixture", "submit test bootstrap"); err != nil {
+		t.Fatalf("register operator recipients: %v", err)
+	}
+	policy := sendingpolicy.DisabledPolicy()
+	policy.BudgetMode = sendingpolicy.ModeEnforce
+	if mutate != nil {
+		mutate(&policy)
+	}
+	f := &gateFixture{
+		t: t, ctx: ctx, pool: pool,
+		gate:   sendingpolicy.NewGate(pool, secrets, sendingpolicy.PolicySourceConfig, policy),
+		userID: psID("usr"),
+		agent:  psID("agt"),
+		tenant: psID("tenant"),
+	}
+	if _, err := pool.Exec(ctx,
+		`INSERT INTO users (id, email, google_subject, account_class) VALUES ($1, $2, $3, 'standard')`,
+		f.userID, f.userID+"@example.test", "sub_"+f.userID,
+	); err != nil {
+		t.Fatalf("insert user: %v", err)
+	}
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO account_sending_controls (user_id, ses_tenant_name, ses_tenant_ready, ses_tenant_ready_at)
+		VALUES ($1, $2, true, now())
+		ON CONFLICT (user_id) DO UPDATE
+		    SET ses_tenant_name = EXCLUDED.ses_tenant_name, ses_tenant_ready = true, ses_tenant_ready_at = now()`,
+		f.userID, f.tenant,
+	); err != nil {
+		t.Fatalf("provision tenant: %v", err)
+	}
+	if _, err := pool.Exec(ctx,
+		`INSERT INTO agent_identities (id, user_id, registered_domain, name) VALUES ($1, $2, 'agents.e2a.dev', $1)`,
+		f.agent, f.userID,
+	); err != nil {
+		t.Fatalf("insert agent: %v", err)
+	}
+	return f
+}
+
+// message inserts an outbound message with `count` distinct recipients.
+func (f *gateFixture) message(count int) (string, []string) {
+	f.t.Helper()
+	id := psID("msg")
+	to := make([]string, count)
+	for i := range to {
+		to[i] = fmt.Sprintf("rcpt-%s-%d@example.test", id, i)
+	}
+	if _, err := f.pool.Exec(f.ctx,
+		`INSERT INTO messages (id, agent_id, direction, to_recipients, sent_as, status)
+		 VALUES ($1, $2, 'outbound', $3, 'own_address', 'sent')`, id, f.agent, to,
+	); err != nil {
+		f.t.Fatalf("insert message: %v", err)
+	}
+	return id, to
+}
+
+// prepare runs the acceptance half the way an API handler does.
+func (f *gateFixture) prepare(messageID string) sendingpolicy.OperationRef {
+	f.t.Helper()
+	tx, err := f.pool.Begin(f.ctx)
+	if err != nil {
+		f.t.Fatalf("begin: %v", err)
+	}
+	accept, ref, err := f.gate.PrepareExternalTx(f.ctx, tx, messageID)
+	if err != nil {
+		_ = tx.Rollback(f.ctx)
+		f.t.Fatalf("prepare: %v", err)
+	}
+	if accept != sendingpolicy.AcceptanceAccept {
+		_ = tx.Rollback(f.ctx)
+		f.t.Fatalf("prepare decision = %v, want accept", accept)
+	}
+	if err := tx.Commit(f.ctx); err != nil {
+		f.t.Fatalf("commit: %v", err)
+	}
+	return ref
+}
+
+// authorize runs the worker's sequence — Reserve, then ConsumeAttempt — and
+// fails the test on a hold, because every test here is about what happens
+// AFTER a token exists.
+func (f *gateFixture) authorize(ref sendingpolicy.OperationRef) sendingpolicy.ProviderAuthorization {
+	f.t.Helper()
+	early, attempt, err := f.gate.Reserve(f.ctx, ref)
+	if err != nil {
+		f.t.Fatalf("reserve: %v", err)
+	}
+	if !early.Allow {
+		f.t.Fatalf("reserve held: %+v", early)
+	}
+	decision, auth, err := f.gate.ConsumeAttempt(f.ctx, attempt)
+	if err != nil {
+		f.t.Fatalf("consume: %v", err)
+	}
+	if !decision.Allow || auth == nil {
+		f.t.Fatalf("consume held: %+v (token=%v)", decision, auth != nil)
+	}
+	return *auth
+}
+
+func (f *gateFixture) callState(operationID string, attempt int) string {
+	f.t.Helper()
+	var state string
+	if err := f.pool.QueryRow(f.ctx, `
+		SELECT call_state FROM sending_budget_reservations
+		 WHERE operation_id = $1 AND submission_attempt = $2`, operationID, attempt,
+	).Scan(&state); err != nil {
+		f.t.Fatalf("read call_state: %v", err)
+	}
+	return state
+}
+
+func (f *gateFixture) correlation(operationID string, attempt int) (correlationID string, providerMessageID *string) {
+	f.t.Helper()
+	err := f.pool.QueryRow(f.ctx, `
+		SELECT correlation_id, provider_message_id FROM sending_feedback_correlations
+		 WHERE operation_id = $1 AND submission_attempt = $2`, operationID, attempt,
+	).Scan(&correlationID, &providerMessageID)
+	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
+		f.t.Fatalf("read correlation: %v", err)
+	}
+	return correlationID, providerMessageID
+}
+
+// countingListener accepts and immediately drops connections, counting them.
+// A relay pointed at it can never complete a transaction, so any nonzero
+// count is a socket that should not have been opened.
+func countingListener(t *testing.T) (*outbound.SMTPRelay, func() int) {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = listener.Close() })
+	var mu sync.Mutex
+	count := 0
+	go func() {
+		for {
+			conn, err := listener.Accept()
+			if err != nil {
+				return
+			}
+			mu.Lock()
+			count++
+			mu.Unlock()
+			_ = conn.Close()
+		}
+	}()
+	addr := listener.Addr().(*net.TCPAddr)
+	relay := outbound.NewSMTPRelay(&config.OutboundSMTPConfig{Host: addr.IP.String(), Port: addr.Port})
+	// The accept goroutine runs after the client's dial returns, so a caller
+	// that reads immediately could miss a connection that did happen. Give a
+	// connection time to be observed: return as soon as one is, or after a
+	// grace period that is long compared to a loopback accept.
+	return relay, func() int {
+		deadline := time.Now().Add(300 * time.Millisecond)
+		for {
+			mu.Lock()
+			c := count
+			mu.Unlock()
+			if c > 0 || time.Now().After(deadline) {
+				return c
+			}
+			time.Sleep(10 * time.Millisecond)
+		}
+	}
+}
+
+// acceptingRelay fronts testutil's fake SMTP server.
+func acceptingRelay(t *testing.T) (*outbound.SMTPRelay, func() []testutil.SMTPMessage) {
+	t.Helper()
+	addr, messages := testutil.FakeSMTPServer(t)
+	return outbound.NewSMTPRelay(&config.OutboundSMTPConfig{Host: addr.Host, Port: addr.Port}), messages
+}
+
+// rejectingRelay fronts a server that answers every RCPT TO with `reply`.
+func rejectingRelay(t *testing.T, reply string) *outbound.SMTPRelay {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = listener.Close() })
+	go func() {
+		for {
+			conn, err := listener.Accept()
+			if err != nil {
+				return
+			}
+			go func(conn net.Conn) {
+				defer conn.Close()
+				r := bufio.NewReader(conn)
+				fmt.Fprint(conn, "220 reject ready\r\n")
+				for {
+					line, err := r.ReadString('\n')
+					if err != nil {
+						return
+					}
+					switch upper := strings.ToUpper(strings.TrimSpace(line)); {
+					case strings.HasPrefix(upper, "RCPT TO:"):
+						fmt.Fprint(conn, reply+"\r\n")
+					case upper == "QUIT":
+						fmt.Fprint(conn, "221 Bye\r\n")
+						return
+					default:
+						fmt.Fprint(conn, "250 OK\r\n")
+					}
+				}
+			}(conn)
+		}
+	}()
+	addr := listener.Addr().(*net.TCPAddr)
+	return outbound.NewSMTPRelay(&config.OutboundSMTPConfig{Host: addr.IP.String(), Port: addr.Port})
+}
+
+// headerValues returns every value of `name` in the header section of a
+// captured message, case-insensitively, with folded continuations unfolded.
+func headerValues(data, name string) []string {
+	var out []string
+	current := -1
+	for _, line := range strings.Split(data, "\n") {
+		line = strings.TrimRight(line, "\r")
+		if line == "" {
+			break
+		}
+		if line[0] == ' ' || line[0] == '\t' {
+			if current >= 0 {
+				out[current] += " " + strings.TrimSpace(line)
+			}
+			continue
+		}
+		current = -1
+		if i := strings.IndexByte(line, ':'); i > 0 && strings.EqualFold(strings.TrimSpace(line[:i]), name) {
+			out = append(out, strings.TrimSpace(line[i+1:]))
+			current = len(out) - 1
+		}
+	}
+	return out
+}
+
+func body(data string) string {
+	if i := strings.Index(data, "\n\n"); i >= 0 {
+		return data[i+2:]
+	}
+	return ""
+}
+
+func TestProviderSubmitterZeroNetworkWithoutAuthorization(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, sockets := countingListener(t)
+	s := outbound.NewProviderSubmitter(relay, f.gate)
+
+	_, err := s.SubmitOnce(f.ctx, sendingpolicy.ProviderAuthorization{}, outbound.Envelope{
+		From: "agent@agents.e2a.dev", Recipients: []string{"someone@example.test"}, Message: []byte("Subject: x\r\n\r\nbody"),
+	})
+	if !errors.Is(err, outbound.ErrAuthorizationRequired) {
+		t.Fatalf("err = %v, want outbound.ErrAuthorizationRequired", err)
+	}
+	if sockets() != 0 {
+		t.Fatalf("sockets = %d, want 0", sockets())
+	}
+}
+
+// TestProviderSubmitterZeroNetworkOnEnvelopeMismatch proves the envelope check
+// runs before redemption: a wrong envelope costs nothing, sends nothing, and
+// leaves the token spendable for the right one.
+func TestProviderSubmitterZeroNetworkOnEnvelopeMismatch(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, sockets := countingListener(t)
+	s := outbound.NewProviderSubmitter(relay, f.gate)
+	messageID, to := f.message(2)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+
+	for name, envelope := range map[string][]string{
+		"extra recipient":    append(append([]string(nil), to...), "attacker@example.test"),
+		"swapped recipient":  {to[0], "attacker@example.test"},
+		"dropped recipient":  {to[0]},
+		"duplicated mailbox": {to[0], strings.ToUpper(to[0]), to[1]},
+		"empty":              nil,
+	} {
+		_, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: envelope, Message: []byte("Subject: x\r\n\r\nbody")})
+		if err == nil {
+			t.Fatalf("%s: submitted, want refusal", name)
+		}
+		if state := f.callState(ref.ID(), 1); state != "authorized" {
+			t.Fatalf("%s: call_state = %s, want authorized (token must survive a mismatch)", name, state)
+		}
+	}
+	if sockets() != 0 {
+		t.Fatalf("sockets = %d, want 0", sockets())
+	}
+}
+
+// TestProviderSubmitterUnconfiguredRelayLeavesTokenIntact: with no relay host
+// there is nothing to dial and nothing to count; what matters is that the
+// token survives, because the failure is the deployment's, not the message's.
+func TestProviderSubmitterUnconfiguredRelayLeavesTokenIntact(t *testing.T) {
+	f := newGateFixture(t, nil)
+	s := outbound.NewProviderSubmitter(outbound.NewSMTPRelay(&config.OutboundSMTPConfig{}), f.gate)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+
+	if _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}); err == nil {
+		t.Fatal("submitted through an unconfigured relay")
+	}
+	// A misconfigured relay is not the message's fault: the token is intact.
+	if state := f.callState(ref.ID(), 1); state != "authorized" {
+		t.Fatalf("call_state = %s, want authorized", state)
+	}
+}
+
+// TestProviderSubmitterAuthorizedTokenIsSingleUse proves the token is spent by
+// the call, not merely checked: a second submission with the same token opens
+// no socket.
+func TestProviderSubmitterAuthorizedTokenIsSingleUse(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, captured := acceptingRelay(t)
+	s := outbound.NewProviderSubmitter(relay, f.gate)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+	env := outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}
+
+	res, err := s.SubmitOnce(f.ctx, auth, env)
+	if err != nil || res.SettlementErr != nil || res.ProviderMessageID == "" {
+		t.Fatalf("first submit: res=%+v err=%v", res, err)
+	}
+	if state := f.callState(ref.ID(), 1); state != "started" {
+		t.Fatalf("call_state = %s, want started", state)
+	}
+
+	_, err = s.SubmitOnce(f.ctx, auth, env)
+	if !errors.Is(err, sendingpolicy.ErrAuthorizationInvalid) {
+		t.Fatalf("second submit err = %v, want ErrAuthorizationInvalid", err)
+	}
+	if n := len(captured()); n != 1 {
+		t.Fatalf("provider received %d messages for one token, want 1", n)
+	}
+}
+
+// TestProviderSubmitterZeroNetworkForStaleAttempt: a token whose ordinal has
+// been superseded — the worker died and a later execution re-reserved — opens
+// no socket.
+func TestProviderSubmitterZeroNetworkForStaleAttempt(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, sockets := countingListener(t)
+	s := outbound.NewProviderSubmitter(relay, f.gate)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	stale := f.authorize(ref)
+	if _, next, err := f.gate.Reserve(f.ctx, ref); err != nil || next.Attempt() != 2 {
+		t.Fatalf("re-reserve: attempt=%v err=%v", next, err)
+	}
+
+	_, err := s.SubmitOnce(f.ctx, stale, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")})
+	if !errors.Is(err, sendingpolicy.ErrAuthorizationInvalid) {
+		t.Fatalf("err = %v, want ErrAuthorizationInvalid", err)
+	}
+	if sockets() != 0 {
+		t.Fatalf("sockets = %d, want 0", sockets())
+	}
+}
+
+// TestProviderSubmitterAttemptHeaderDerivesOnlyFromToken proves the wire
+// carries exactly one attempt header, its value is the gate's correlation id,
+// every smuggled spelling is gone, and the body is untouched.
+func TestProviderSubmitterAttemptHeaderDerivesOnlyFromToken(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, captured := acceptingRelay(t)
+	s := outbound.NewProviderSubmitter(relay, f.gate)
+	s.SetSESConfigurationSet("e2a-delivery-test")
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+
+	res, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: outbound.SmuggledMIME()})
+	if err != nil || res.SettlementErr != nil {
+		t.Fatalf("submit: res=%+v err=%v", res, err)
+	}
+	msgs := captured()
+	if len(msgs) != 1 {
+		t.Fatalf("captured %d messages, want 1", len(msgs))
+	}
+	data := msgs[0].Data
+	corr, _ := f.correlation(ref.ID(), 1)
+	if corr == "" {
+		t.Fatal("no correlation row for the attempt")
+	}
+	for name, want := range map[string][]string{
+		outbound.ProviderAttemptHeader:     {corr},
+		outbound.SESConfigurationSetHeader: {"e2a-delivery-test"},
+		delivery.MessageIDHeader:           {messageID},
+		outbound.SESTenantHeader:           nil, // policy has the tenant header disabled
+	} {
+		if got := headerValues(data, name); strings.Join(got, "|") != strings.Join(want, "|") {
+			t.Errorf("%s = %q, want %q", name, got, want)
+		}
+	}
+	if got := headerValues(data, "Subject"); len(got) != 1 || got[0] != "hello" {
+		t.Errorf("customer headers disturbed: Subject = %q", got)
+	}
+	if b := body(data); !strings.Contains(b, "X-SES-TENANT: body-decoy") || !strings.Contains(b, "body line") {
+		t.Errorf("body was rewritten: %q", b)
+	}
+	if strings.Contains(data, "smuggled") || strings.Contains(data, "forged") || strings.Contains(data, "attacker") {
+		t.Errorf("a smuggled header value survived:\n%s", data)
+	}
+}
+
+// TestProviderSubmitterTenantHeaderIsExactAndSingle: under an enforcing tenant
+// policy the wire carries exactly one X-SES-TENANT, whose value is the tenant
+// the gate read under lock — not anything the MIME said.
+func TestProviderSubmitterTenantHeaderIsExactAndSingle(t *testing.T) {
+	f := newGateFixture(t, func(p *sendingpolicy.RuntimePolicy) {
+		p.TenantHeaderMode = sendingpolicy.TenantHeaderEnforce
+	})
+	relay, captured := acceptingRelay(t)
+	s := outbound.NewProviderSubmitter(relay, f.gate)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+
+	if _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: outbound.SmuggledMIME()}); err != nil {
+		t.Fatalf("submit: %v", err)
+	}
+	msgs := captured()
+	if len(msgs) != 1 {
+		t.Fatalf("captured %d messages, want 1", len(msgs))
+	}
+	if got := headerValues(msgs[0].Data, outbound.SESTenantHeader); len(got) != 1 || got[0] != f.tenant {
+		t.Fatalf("%s = %q, want exactly [%q]", outbound.SESTenantHeader, got, f.tenant)
+	}
+}
+
+// TestProviderSubmitterRetryRedeemsADistinctAttempt: a physical retry is a new
+// ordinal with a new token and a new attempt header, never a resubmission
+// under the old one.
+func TestProviderSubmitterRetryRedeemsADistinctAttempt(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, captured := acceptingRelay(t)
+	s := outbound.NewProviderSubmitter(relay, f.gate)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	env := outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}
+
+	first := f.authorize(ref)
+	if _, err := s.SubmitOnce(f.ctx, first, env); err != nil {
+		t.Fatalf("first: %v", err)
+	}
+	second := f.authorize(ref)
+	if second.Attempt().Attempt() != 2 {
+		t.Fatalf("second ordinal = %d, want 2", second.Attempt().Attempt())
+	}
+	if _, err := s.SubmitOnce(f.ctx, second, env); err != nil {
+		t.Fatalf("second: %v", err)
+	}
+	msgs := captured()
+	if len(msgs) != 2 {
+		t.Fatalf("captured %d messages, want 2", len(msgs))
+	}
+	c1, _ := f.correlation(ref.ID(), 1)
+	c2, _ := f.correlation(ref.ID(), 2)
+	h1 := headerValues(msgs[0].Data, outbound.ProviderAttemptHeader)
+	h2 := headerValues(msgs[1].Data, outbound.ProviderAttemptHeader)
+	if len(h1) != 1 || len(h2) != 1 || h1[0] != c1 || h2[0] != c2 || c1 == c2 {
+		t.Fatalf("attempt headers %v / %v, want distinct correlations %q / %q", h1, h2, c1, c2)
+	}
+}
+
+func TestProviderSubmitterBindsProviderMessageIDOnAcceptance(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, _ := acceptingRelay(t)
+	s := outbound.NewProviderSubmitter(relay, f.gate)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+
+	res, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")})
+	if err != nil || res.SettlementErr != nil {
+		t.Fatalf("submit: res=%+v err=%v", res, err)
+	}
+	if res.Attempt.Attempt() != 1 {
+		t.Errorf("result attempt = %d, want 1", res.Attempt.Attempt())
+	}
+	_, bound := f.correlation(ref.ID(), 1)
+	want := sendingpolicy.NormalizeProviderMessageID(res.ProviderMessageID)
+	if bound == nil || *bound != want {
+		t.Fatalf("correlation provider_message_id = %v, want %q (normalized from %q)", bound, want, res.ProviderMessageID)
+	}
+}
+
+// TestProviderSubmitterPermanentRejectionIsSettledNotRetried: a definite 5xx
+// consumed the attempt (the socket opened), settles as rejected, binds no
+// provider id, and surfaces as a permanent error the worker can classify.
+func TestProviderSubmitterPermanentRejectionIsSettledNotRetried(t *testing.T) {
+	f := newGateFixture(t, nil)
+	spy := &spyGate{Gate: f.gate}
+	s := outbound.NewProviderSubmitter(rejectingRelay(t, "550 5.1.1 no such user"), spy)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+
+	_, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")})
+	if err == nil || !outbound.IsPermanentSMTPError(err) {
+		t.Fatalf("err = %v, want a permanent SMTP error", err)
+	}
+	if state := f.callState(ref.ID(), 1); state != "started" {
+		t.Fatalf("call_state = %s, want started (the socket did open)", state)
+	}
+	if _, bound := f.correlation(ref.ID(), 1); bound != nil {
+		t.Fatalf("provider_message_id = %q bound on a rejection", *bound)
+	}
+	got := spy.settled()
+	if len(got) != 1 || got[0].Outcome != sendingpolicy.SettlementProviderPermanentlyRejected || got[0].ProviderMessageID != "" {
+		t.Fatalf("settlements = %+v, want exactly one permanent rejection without a provider id", got)
+	}
+}
+
+// spyGate records settlements and can be made to fail them, so a test can see
+// the one effect of SubmitOnce that has no observable row yet.
+type spyGate struct {
+	sendingpolicy.Gate
+	mu        sync.Mutex
+	settleErr error
+	calls     []sendingpolicy.ProviderSettlement
+}
+
+func (g *spyGate) SettleProvider(ctx context.Context, s sendingpolicy.ProviderSettlement) error {
+	g.mu.Lock()
+	g.calls = append(g.calls, s)
+	g.mu.Unlock()
+	if g.settleErr != nil {
+		return g.settleErr
+	}
+	return g.Gate.SettleProvider(ctx, s)
+}
+
+func (g *spyGate) settled() []sendingpolicy.ProviderSettlement {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	return append([]sendingpolicy.ProviderSettlement(nil), g.calls...)
+}
+
+// TestProviderSubmitterAmbiguousOutcomeIsNotSettled: a 4xx might still be
+// delivered on retry, so nothing is settled and the error is not permanent.
+func TestProviderSubmitterAmbiguousOutcomeIsNotSettled(t *testing.T) {
+	f := newGateFixture(t, nil)
+	spy := &spyGate{Gate: f.gate}
+	s := outbound.NewProviderSubmitter(rejectingRelay(t, "451 4.3.0 try again later"), spy)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+
+	_, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")})
+	if err == nil || outbound.IsPermanentSMTPError(err) {
+		t.Fatalf("err = %v, want a non-permanent SMTP error", err)
+	}
+	if got := spy.settled(); len(got) != 0 {
+		t.Fatalf("settlements = %+v, want none for an ambiguous outcome", got)
+	}
+	if state := f.callState(ref.ID(), 1); state != "started" {
+		t.Fatalf("call_state = %s, want started (the socket did open; a retry needs a new ordinal)", state)
+	}
+}
+
+// TestProviderSubmitterAcceptedButUnsettledIsReportedNotRetried pins the
+// at-least-once contract: SES took the message, so the failure to settle is
+// carried on the result with a nil error. A caller that resubmitted here would
+// send the message twice.
+func TestProviderSubmitterAcceptedButUnsettledIsReportedNotRetried(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, captured := acceptingRelay(t)
+	spy := &spyGate{Gate: f.gate, settleErr: errors.New("settle: database unavailable")}
+	s := outbound.NewProviderSubmitter(relay, spy)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+
+	res, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")})
+	if err != nil {
+		t.Fatalf("err = %v, want nil — the message was accepted", err)
+	}
+	if res.SettlementErr == nil || res.ProviderMessageID == "" || res.Attempt.Attempt() != 1 {
+		t.Fatalf("result = %+v, want provider id, attempt 1, and a settlement error", res)
+	}
+	if n := len(captured()); n != 1 {
+		t.Fatalf("provider received %d messages, want 1", n)
+	}
+	got := spy.settled()
+	if len(got) != 1 || got[0].Outcome != sendingpolicy.SettlementProviderAccepted || got[0].ProviderMessageID != res.ProviderMessageID {
+		t.Fatalf("settlement attempted = %+v, want one acceptance carrying %q", got, res.ProviderMessageID)
+	}
+	// The caller's recovery is to settle again, idempotently — never to resubmit.
+	if err := f.gate.SettleProvider(f.ctx, got[0]); err != nil {
+		t.Fatalf("late settlement: %v", err)
+	}
+	if _, bound := f.correlation(ref.ID(), 1); bound == nil || *bound != sendingpolicy.NormalizeProviderMessageID(res.ProviderMessageID) {
+		t.Fatalf("bound = %v, want the normalized provider id", bound)
+	}
+}
+
+// TestProviderSubmitterSocketCounterObservesADial is the positive control for
+// every zero-network assertion above: a redeemed token that reaches the dial
+// is seen by the counter exactly once.
+func TestProviderSubmitterSocketCounterObservesADial(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, sockets := countingListener(t)
+	s := outbound.NewProviderSubmitter(relay, f.gate)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+
+	if _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}); err == nil {
+		t.Fatal("a dropped connection was reported as success")
+	}
+	if sockets() != 1 {
+		t.Fatalf("sockets = %d, want exactly 1", sockets())
+	}
+	if state := f.callState(ref.ID(), 1); state != "started" {
+		t.Fatalf("call_state = %s, want started", state)
+	}
+}
+
+// vanishingRelay fronts a server that takes the whole DATA body and then
+// closes without a 250 — the lost-acceptance shape.
+func vanishingRelay(t *testing.T) (*outbound.SMTPRelay, func() int) {
+	return afterDotRelay(t, "")
+}
+
+// afterDotRelay fronts a server that takes the whole DATA body and then does
+// `then`: "" closes silently, "stall" never answers, anything else is sent as
+// the final reply line.
+func afterDotRelay(t *testing.T, then string) (*outbound.SMTPRelay, func() int) {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	stop := make(chan struct{})
+	t.Cleanup(func() { close(stop); _ = listener.Close() })
+	var mu sync.Mutex
+	bodies := 0
+	go func() {
+		for {
+			conn, err := listener.Accept()
+			if err != nil {
+				return
+			}
+			go func(conn net.Conn) {
+				defer conn.Close()
+				r := bufio.NewReader(conn)
+				fmt.Fprint(conn, "220 vanish ready\r\n")
+				inData := false
+				for {
+					line, err := r.ReadString('\n')
+					if err != nil {
+						return
+					}
+					if inData {
+						if strings.TrimRight(line, "\r\n") == "." {
+							mu.Lock()
+							bodies++
+							mu.Unlock()
+							switch then {
+							case "":
+								return // no 250: the connection just dies
+							case "stall":
+								<-stop // hold the connection open, silently
+								return
+							default:
+								fmt.Fprint(conn, then+"\r\n")
+								inData = false
+								continue
+							}
+						}
+						continue
+					}
+					switch upper := strings.ToUpper(strings.TrimSpace(line)); {
+					case upper == "DATA":
+						inData = true
+						fmt.Fprint(conn, "354 Go ahead\r\n")
+					case upper == "QUIT":
+						fmt.Fprint(conn, "221 Bye\r\n")
+						return
+					default:
+						fmt.Fprint(conn, "250 OK\r\n")
+					}
+				}
+			}(conn)
+		}
+	}()
+	addr := listener.Addr().(*net.TCPAddr)
+	return outbound.NewSMTPRelay(&config.OutboundSMTPConfig{Host: addr.IP.String(), Port: addr.Port}),
+		func() int { mu.Lock(); defer mu.Unlock(); return bodies }
+}
+
+// TestProviderSubmitterPauseBetweenConsumeAndSubmitOpensNoSocket: the abuse
+// pause is re-proved at redemption, so a pause that lands after the token was
+// minted still stops the send.
+func TestProviderSubmitterPauseBetweenConsumeAndSubmitOpensNoSocket(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, sockets := countingListener(t)
+	s := outbound.NewProviderSubmitter(relay, f.gate)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+	if _, err := f.pool.Exec(f.ctx, `UPDATE account_sending_controls SET state = 'paused' WHERE user_id = $1`, f.userID); err != nil {
+		t.Fatal(err)
+	}
+
+	_, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")})
+	if !errors.Is(err, sendingpolicy.ErrAuthorizationInvalid) {
+		t.Fatalf("err = %v, want ErrAuthorizationInvalid", err)
+	}
+	if sockets() != 0 {
+		t.Fatalf("sockets = %d, want 0", sockets())
+	}
+}
+
+// TestProviderSubmitterMalformedHeaderSectionOpensNoSocket: a bare CR in the
+// header section, or a leading continuation, is refused before redemption.
+func TestProviderSubmitterMalformedHeaderSectionOpensNoSocket(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, sockets := countingListener(t)
+	s := outbound.NewProviderSubmitter(relay, f.gate)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+
+	for name, mime := range map[string]string{
+		"bare CR hides a header":    "Subject: a\rX-SES-TENANT: evil\r\n\r\nbody",
+		"CR CR LF pseudo-separator": "Subject: s\r\n\r\r\nX-SES-TENANT: evil\r\n\r\nbody",
+		"leading continuation":      " evil-suffix\r\nSubject: s\r\n\r\nbody",
+	} {
+		_, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte(mime)})
+		if !errors.Is(err, outbound.ErrMalformedHeaderSection) {
+			t.Errorf("%s: err = %v, want ErrMalformedHeaderSection", name, err)
+		}
+		if state := f.callState(ref.ID(), 1); state != "authorized" {
+			t.Errorf("%s: call_state = %s, want authorized", name, state)
+		}
+	}
+	if sockets() != 0 {
+		t.Fatalf("sockets = %d, want 0", sockets())
+	}
+}
+
+// TestProviderSubmitterSubmitsTheCanonicalEnvelope: the caller's spelling of a
+// recipient is validated but never sent; RCPT TO carries the normalized
+// address the token was priced for.
+func TestProviderSubmitterSubmitsTheCanonicalEnvelope(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, captured := acceptingRelay(t)
+	s := outbound.NewProviderSubmitter(relay, f.gate)
+	messageID, to := f.message(2)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+
+	padded := []string{"  " + strings.ToUpper(to[1]) + "  ", to[0]}
+	if _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: padded, Message: []byte("Subject: x\r\n\r\nbody")}); err != nil {
+		t.Fatalf("submit: %v", err)
+	}
+	msgs := captured()
+	if len(msgs) != 1 {
+		t.Fatalf("captured %d messages, want 1", len(msgs))
+	}
+	want := auth.AuthorizedRecipients()
+	if strings.Join(msgs[0].Recipients, ",") != strings.Join(want, ",") {
+		t.Fatalf("RCPT TO = %q, want the canonical %q", msgs[0].Recipients, want)
+	}
+}
+
+// TestProviderSubmitterLostAcceptanceIsUnsettledAndMarked: the body was
+// delivered and the 250 never came. Nothing is settled, and the error carries
+// the marker so the worker can tell "maybe sent" from "not sent".
+func TestProviderSubmitterLostAcceptanceIsUnsettledAndMarked(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, bodies := vanishingRelay(t)
+	spy := &spyGate{Gate: f.gate}
+	s := outbound.NewProviderSubmitter(relay, spy)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+
+	_, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")})
+	if !errors.Is(err, outbound.ErrProviderAcceptanceUnknown) {
+		t.Fatalf("err = %v, want ErrProviderAcceptanceUnknown", err)
+	}
+	if outbound.IsPermanentSMTPError(err) {
+		t.Fatalf("err = %v classified permanent; a delivered body must never be", err)
+	}
+	if bodies() != 1 {
+		t.Fatalf("provider took %d bodies, want 1", bodies())
+	}
+	if got := spy.settled(); len(got) != 0 {
+		t.Fatalf("settlements = %+v, want none while acceptance is unknown", got)
+	}
+	if state := f.callState(ref.ID(), 1); state != "started" {
+		t.Fatalf("call_state = %s, want started", state)
+	}
+}
+
+// TestProviderSubmitterLostAcceptanceSurvivesTheDeadline: the likeliest way
+// to lose a 250 is the caller's deadline. The marker must survive the relay's
+// context remap, or the worker cannot tell "maybe sent" from "not sent".
+func TestProviderSubmitterLostAcceptanceSurvivesTheDeadline(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, bodies := afterDotRelay(t, "stall")
+	spy := &spyGate{Gate: f.gate}
+	s := outbound.NewProviderSubmitter(relay, spy)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+
+	ctx, cancel := context.WithTimeout(f.ctx, 400*time.Millisecond)
+	defer cancel()
+	_, err := s.SubmitOnce(ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")})
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("err = %v, want the deadline", err)
+	}
+	if !errors.Is(err, outbound.ErrProviderAcceptanceUnknown) {
+		t.Fatalf("err = %v, want ErrProviderAcceptanceUnknown preserved through the remap", err)
+	}
+	if bodies() != 1 {
+		t.Fatalf("provider took %d bodies, want 1", bodies())
+	}
+	if got := spy.settled(); len(got) != 0 {
+		t.Fatalf("settlements = %+v, want none", got)
+	}
+}
+
+// TestProviderSubmitterPostDataRejectionIsDefinite: SES answers a content
+// rejection AFTER the body with an ordinary 554. That is a definite answer —
+// settled as rejected, classified permanent, and never marked ambiguous.
+func TestProviderSubmitterPostDataRejectionIsDefinite(t *testing.T) {
+	f := newGateFixture(t, nil)
+	relay, bodies := afterDotRelay(t, "554 5.6.0 Message rejected")
+	spy := &spyGate{Gate: f.gate}
+	s := outbound.NewProviderSubmitter(relay, spy)
+	messageID, to := f.message(1)
+	ref := f.prepare(messageID)
+	auth := f.authorize(ref)
+
+	_, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")})
+	if err == nil || !outbound.IsPermanentSMTPError(err) {
+		t.Fatalf("err = %v, want permanent", err)
+	}
+	if errors.Is(err, outbound.ErrProviderAcceptanceUnknown) {
+		t.Fatalf("err = %v carries the ambiguity marker on a definite reply", err)
+	}
+	if bodies() != 1 {
+		t.Fatalf("provider took %d bodies, want 1", bodies())
+	}
+	got := spy.settled()
+	if len(got) != 1 || got[0].Outcome != sendingpolicy.SettlementProviderPermanentlyRejected {
+		t.Fatalf("settlements = %+v, want one permanent rejection", got)
+	}
+}
diff --git a/internal/outbound/smtp_relay.go b/internal/outbound/smtp_relay.go
index d35eda183..b67753423 100644
--- a/internal/outbound/smtp_relay.go
+++ b/internal/outbound/smtp_relay.go
@@ -19,6 +19,15 @@ import (
 
 var smtpRetryBackoffs = []time.Duration{1 * time.Second, 5 * time.Second, 15 * time.Second}
 
+// ErrProviderAcceptanceUnknown marks a failure that happened AFTER the whole
+// message body was handed to the provider: the terminating dot was written and
+// the 250 never arrived. The provider may have accepted the message. No
+// classifier can call this permanent or transient, and a caller that retries
+// it as if nothing was sent will deliver the message twice. The sending
+// protection seam leaves such an attempt unsettled; delivery feedback carrying
+// the attempt header is the only authoritative answer.
+var ErrProviderAcceptanceUnknown = errors.New("outbound smtp: message body delivered, provider acceptance unknown")
+
 type SMTPRelay struct {
 	cfg *config.OutboundSMTPConfig
 }
@@ -178,16 +187,26 @@ func (r *SMTPRelay) sendOnceContext(ctx context.Context, envelopeFrom string, re
 		if err == nil {
 			return
 		}
-		if ctx.Err() != nil {
+		// Whether the body had already been handed over must survive the
+		// remaps below: a cancellation or deadline is the LIKELIEST way to
+		// lose the 250, and the caller's contract for that shape is "maybe
+		// sent", not "not sent".
+		unknown := errors.Is(err, ErrProviderAcceptanceUnknown)
+		switch {
+		case ctx.Err() != nil:
 			err = ctx.Err()
-			return
+		default:
+			// The conn deadline is set to the ctx deadline below, so the net
+			// poller's timer races the context's own timer to the same
+			// instant. When the poller wins, the I/O error surfaces while
+			// ctx.Err() is still nil — map it to the deadline error the
+			// caller contracted for.
+			if d, ok := ctx.Deadline(); ok && !time.Now().Before(d) && errors.Is(err, os.ErrDeadlineExceeded) {
+				err = context.DeadlineExceeded
+			}
 		}
-		// The conn deadline is set to the ctx deadline below, so the net
-		// poller's timer races the context's own timer to the same instant.
-		// When the poller wins, the I/O error surfaces while ctx.Err() is
-		// still nil — map it to the deadline error the caller contracted for.
-		if d, ok := ctx.Deadline(); ok && !time.Now().Before(d) && errors.Is(err, os.ErrDeadlineExceeded) {
-			err = context.DeadlineExceeded
+		if unknown && !errors.Is(err, ErrProviderAcceptanceUnknown) {
+			err = errors.Join(ErrProviderAcceptanceUnknown, err)
 		}
 	}()
 
@@ -289,6 +308,13 @@ func (r *SMTPRelay) sendOnceContext(ctx context.Context, envelopeFrom string, re
 	// 250 response is waiting in the buffer. Read it directly.
 	_, msg, err := text.ReadResponse(250)
 	if err != nil {
+		// A coded reply here is the provider's definite answer to the whole
+		// message (SES's post-DATA content rejection is an ordinary 554) and
+		// classifies like any other. Only a reply that never came is
+		// ambiguous, and only that carries the marker.
+		if _, coded := smtpCode(err); !coded {
+			err = errors.Join(ErrProviderAcceptanceUnknown, err)
+		}
 		return "", fmt.Errorf("data final: %w", err)
 	}
 
diff --git a/internal/sendingpolicy/gate.go b/internal/sendingpolicy/gate.go
index c606b85e9..1bf71b0d8 100644
--- a/internal/sendingpolicy/gate.go
+++ b/internal/sendingpolicy/gate.go
@@ -720,8 +720,12 @@ func (m *Module) readAuthState(ctx context.Context, tx pgx.Tx, ref AttemptRef) (
 	}
 	// A required header with no name to put in it is not a header. The adapter
 	// would have to either omit it or send an empty value, and both defeat the
-	// isolation the header exists for, so the send waits for a real tenant.
-	if st.tenantMode == TenantModeRequired && strings.TrimSpace(st.tenantName) == "" {
+	// isolation the header exists for, so the send waits for a real tenant. A
+	// name that cannot be a header VALUE — control characters, whitespace — is
+	// refused here for the same reason: the adapter would fail closed on it
+	// anyway, but silently and unretryably, whereas a hold with this reason is
+	// something an operator can see and repair.
+	if st.tenantMode == TenantModeRequired && !validTenantName(st.tenantName) {
 		return st, holdDecision(ReasonTenantUnnamed, time.Time{}), nil
 	}
 
@@ -793,6 +797,20 @@ func (m *Module) readAuthState(ctx context.Context, tx pgx.Tx, ref AttemptRef) (
 	return st, allowDecision(), nil
 }
 
+// validTenantName reports whether a tenant name can travel as an SMTP header
+// value: non-empty, printable ASCII, no whitespace.
+func validTenantName(name string) bool {
+	if strings.TrimSpace(name) == "" {
+		return false
+	}
+	for _, r := range name {
+		if r <= ' ' || r > '~' {
+			return false
+		}
+	}
+	return true
+}
+
 // tenantModeFor resolves the tenant-header mode for one account.
 //
 // Canary is an explicit account list rather than a percentage: a tenant header
@@ -1506,6 +1524,34 @@ func (m *Module) RedeemProviderCall(ctx context.Context, auth ProviderAuthorizat
 		return m.invalidate(ctx, tx, auth.attempt)
 	}
 
+	// Re-prove the abuse pause. ConsumeAttempt linearized it under the
+	// account-control lock, but that transaction has committed and a pause
+	// can land in the gap before the socket opens; a pause is the one control
+	// whose whole point is "stop now". The read is deliberately UNLOCKED: the
+	// account-control key precedes the operation key in the normative order
+	// and this transaction already holds the operation, so locking it here
+	// would invert the order. A committed pause is visible to a plain read
+	// because the transaction runs READ COMMITTED (each statement sees a
+	// fresh snapshot; under REPEATABLE READ this re-check would silently
+	// read the snapshot taken at effectivePolicy and prove nothing), and the
+	// check can only refuse, never widen. The window that remains runs from
+	// this read to the dial — the reservation update, commit, and return.
+	//
+	// Protection notices are exempt: the notice telling an account it was
+	// paused is SOURCED from that paused account, and it must go out.
+	if op.Purpose.isCustomer() && op.SourceAccountRef != nil {
+		var state string
+		err := tx.QueryRow(ctx,
+			`SELECT state FROM account_sending_controls WHERE user_id = $1`, *op.SourceAccountRef,
+		).Scan(&state)
+		if err != nil && !errors.Is(err, pgx.ErrNoRows) {
+			return fmt.Errorf("sendingpolicy: read account control: %w", err)
+		}
+		if state == "paused" {
+			return m.invalidate(ctx, tx, auth.attempt)
+		}
+	}
+
 	stored, err := lockReservation(ctx, tx, auth.attempt.operationID, auth.attempt.attempt)
 	if err != nil {
 		return err
@@ -1788,10 +1834,20 @@ func (m *Module) SettleProvider(ctx context.Context, settlement ProviderSettleme
 	// attempt that was authorized to reach the provider. Accepting a reserved
 	// or released attempt would advance ramp progress for a send that never
 	// happened.
-	if stored.State != "confirmed" {
+	if stored.State != "confirmed" || stored.CallState != "started" {
+		// `confirmed` says capacity was charged; `started` says the token was
+		// redeemed and the socket opened. Only the second proves the provider
+		// could have seen the message, and a provider id bound to an attempt
+		// that never dialed is a ledger claim about a send that did not
+		// happen.
 		return ErrAttemptStale
 	}
 
+	if err := bindProviderMessageID(ctx, tx, settlement); err != nil {
+		return err
+	}
+	// Ramp keys come last in the normative order, after the correlation row,
+	// which is keyed by this operation and already held under its lock.
 	if op.Purpose == PurposeCustomerMessage {
 		if err := m.rampSettle(ctx, tx, op.OperationID, settlement.Outcome); err != nil {
 			return err
@@ -1799,3 +1855,53 @@ func (m *Module) SettleProvider(ctx context.Context, settlement ProviderSettleme
 	}
 	return m.commit(ctx, tx, "settle")
 }
+
+// bindProviderMessageID records the provider's id on the attempt's feedback
+// correlation, exactly once.
+//
+// It runs after the operation and reservation locks, which is the only place
+// the correlation row is ever written after its insert, so no additional key
+// joins the normative order. Replaying the same id is a no-op — the
+// synchronous success branch and the delayed feedback finalizer both settle
+// the same attempt — while a different id is refused outright. A correlation
+// that has already aged out of retention is left alone: there is nothing to
+// attribute feedback to any more, and failing a late settlement over it would
+// only make the caller retry forever.
+func bindProviderMessageID(ctx context.Context, tx pgx.Tx, settlement ProviderSettlement) error {
+	id := NormalizeProviderMessageID(settlement.ProviderMessageID)
+	if id == "" {
+		return nil
+	}
+	if settlement.Outcome != SettlementProviderAccepted {
+		return fmt.Errorf("sendingpolicy: a %q settlement cannot carry a provider message id", settlement.Outcome)
+	}
+	var bound *string
+	err := tx.QueryRow(ctx, `
+		SELECT provider_message_id
+		  FROM sending_feedback_correlations
+		 WHERE operation_id = $1 AND submission_attempt = $2
+		   FOR UPDATE`,
+		settlement.Attempt.operationID, settlement.Attempt.attempt,
+	).Scan(&bound)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("sendingpolicy: lock correlation: %w", err)
+	}
+	if bound != nil {
+		if NormalizeProviderMessageID(*bound) == id {
+			return nil
+		}
+		return ErrProviderMessageIDConflict
+	}
+	if _, err := tx.Exec(ctx, `
+		UPDATE sending_feedback_correlations
+		   SET provider_message_id = $3
+		 WHERE operation_id = $1 AND submission_attempt = $2`,
+		settlement.Attempt.operationID, settlement.Attempt.attempt, id,
+	); err != nil {
+		return fmt.Errorf("sendingpolicy: bind provider message id: %w", err)
+	}
+	return nil
+}
diff --git a/internal/sendingpolicy/provider_attempt_integration_test.go b/internal/sendingpolicy/provider_attempt_integration_test.go
index f3b3c495f..b480cce7a 100644
--- a/internal/sendingpolicy/provider_attempt_integration_test.go
+++ b/internal/sendingpolicy/provider_attempt_integration_test.go
@@ -899,9 +899,15 @@ func TestSettleProviderValidatesAndIsIdempotent(t *testing.T) {
 	user := f.user("standard")
 	agent := f.agent(user)
 	_, attempt := f.prepareAndReserve(g, agent, 2)
-	if _, auth, err := g.ConsumeAttempt(f.ctx, attempt); err != nil || auth == nil {
+	_, auth, err := g.ConsumeAttempt(f.ctx, attempt)
+	if err != nil || auth == nil {
 		t.Fatalf("authorize: auth=%v err=%v", auth, err)
 	}
+	// Settlement reports what the provider did, so the attempt must have
+	// reached the provider: redeem, as the adapter does before it dials.
+	if err := g.RedeemProviderCall(f.ctx, *auth); err != nil {
+		t.Fatalf("redeem: %v", err)
+	}
 
 	for i := 0; i < 2; i++ {
 		if err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{
@@ -1527,9 +1533,14 @@ func TestSettlementRejectsAnAttemptThatWasNeverAuthorized(t *testing.T) {
 	}
 
 	// Authorized: now it settles.
-	if _, auth, err := g.ConsumeAttempt(f.ctx, attempt); err != nil || auth == nil {
+	_, auth, err := g.ConsumeAttempt(f.ctx, attempt)
+	if err != nil || auth == nil {
 		t.Fatalf("authorize: auth=%v err=%v", auth, err)
 	}
+	// Authorized is not enough: settlement needs the socket to have opened.
+	if err := g.RedeemProviderCall(f.ctx, *auth); err != nil {
+		t.Fatalf("redeem: %v", err)
+	}
 	if err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{
 		Attempt: attempt, Outcome: sendingpolicy.SettlementProviderAccepted,
 	}); err != nil {
diff --git a/internal/sendingpolicy/provider_token_test.go b/internal/sendingpolicy/provider_token_test.go
new file mode 100644
index 000000000..1e3d2d696
--- /dev/null
+++ b/internal/sendingpolicy/provider_token_test.go
@@ -0,0 +1,271 @@
+package sendingpolicy_test
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/tokencanopy/e2a/internal/sendingpolicy"
+)
+
+// These tests are about what a settlement may bind to the attempt's feedback
+// correlation. The provider id is how most delivery feedback finds its
+// attempt, so it has to be written exactly once, by an acceptance, and never
+// rewritten.
+
+func (f *fixture) providerMessageID(operationID string, attempt int) *string {
+	f.t.Helper()
+	var id *string
+	if err := f.pool.QueryRow(f.ctx, `
+		SELECT provider_message_id FROM sending_feedback_correlations
+		 WHERE operation_id = $1 AND submission_attempt = $2`, operationID, attempt,
+	).Scan(&id); err != nil {
+		f.t.Fatalf("read provider_message_id: %v", err)
+	}
+	return id
+}
+
+// redeemed mints a token and redeems it, leaving the attempt in the state a
+// settlement expects.
+func (f *fixture) redeemed(g sendingpolicy.Gate, agent string) (sendingpolicy.OperationRef, *sendingpolicy.ProviderAuthorization) {
+	f.t.Helper()
+	ref, attempt := f.prepareAndReserve(g, agent, 1)
+	_, auth, err := g.ConsumeAttempt(f.ctx, attempt)
+	if err != nil || auth == nil {
+		f.t.Fatalf("authorize: auth=%v err=%v", auth, err)
+	}
+	if err := g.RedeemProviderCall(f.ctx, *auth); err != nil {
+		f.t.Fatalf("redeem: %v", err)
+	}
+	return ref, auth
+}
+
+func TestProviderTokenSettlementBindsProviderMessageIDOnce(t *testing.T) {
+	f := newFixture(t)
+	g := f.gate(enforcingPolicy(nil))
+	ref, auth := f.redeemed(g, f.agent(f.user("standard")))
+
+	accepted := sendingpolicy.ProviderSettlement{
+		Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderAccepted, ProviderMessageID: "ses-id-one",
+	}
+	if err := g.SettleProvider(f.ctx, accepted); err != nil {
+		t.Fatalf("settle: %v", err)
+	}
+	if got := f.providerMessageID(ref.ID(), 1); got == nil || *got != "ses-id-one" {
+		t.Fatalf("bound = %v, want ses-id-one", got)
+	}
+
+	// The delayed feedback finalizer settles the same attempt again.
+	if err := g.SettleProvider(f.ctx, accepted); err != nil {
+		t.Fatalf("replay: %v", err)
+	}
+
+	// A different id for the same attempt is two physical sends for one
+	// charge. Refuse it and keep the first.
+	conflicting := accepted
+	conflicting.ProviderMessageID = "ses-id-two"
+	if err := g.SettleProvider(f.ctx, conflicting); !errors.Is(err, sendingpolicy.ErrProviderMessageIDConflict) {
+		t.Fatalf("conflict err = %v, want ErrProviderMessageIDConflict", err)
+	}
+	if got := f.providerMessageID(ref.ID(), 1); got == nil || *got != "ses-id-one" {
+		t.Fatalf("bound after conflict = %v, want ses-id-one kept", got)
+	}
+}
+
+func TestProviderTokenRejectionNeverBindsProviderMessageID(t *testing.T) {
+	f := newFixture(t)
+	g := f.gate(enforcingPolicy(nil))
+	ref, auth := f.redeemed(g, f.agent(f.user("standard")))
+
+	// A rejection carrying an id is a caller bug: nothing was accepted, so
+	// there is nothing to attribute feedback to.
+	err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{
+		Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderPermanentlyRejected, ProviderMessageID: "ses-id",
+	})
+	if err == nil {
+		t.Fatal("a rejection with a provider id was accepted")
+	}
+	if got := f.providerMessageID(ref.ID(), 1); got != nil {
+		t.Fatalf("bound = %q on a refused settlement", *got)
+	}
+
+	if err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{
+		Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderPermanentlyRejected,
+	}); err != nil {
+		t.Fatalf("plain rejection: %v", err)
+	}
+	if got := f.providerMessageID(ref.ID(), 1); got != nil {
+		t.Fatalf("bound = %q after a rejection", *got)
+	}
+}
+
+// TestProviderTokenAcceptanceWithoutIDStillSettles keeps the pre-existing
+// contract: an acceptance whose id was lost (crash between DATA and the
+// result) settles normally and leaves the correlation open for the feedback
+// path to bind by attempt header.
+func TestProviderTokenAcceptanceWithoutIDStillSettles(t *testing.T) {
+	f := newFixture(t)
+	g := f.gate(enforcingPolicy(nil))
+	ref, auth := f.redeemed(g, f.agent(f.user("standard")))
+
+	if err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{
+		Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderAccepted,
+	}); err != nil {
+		t.Fatalf("settle: %v", err)
+	}
+	if got := f.providerMessageID(ref.ID(), 1); got != nil {
+		t.Fatalf("bound = %q, want nothing bound", *got)
+	}
+	// And a later settlement that does know the id may still bind it.
+	if err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{
+		Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderAccepted, ProviderMessageID: "ses-late",
+	}); err != nil {
+		t.Fatalf("late bind: %v", err)
+	}
+	if got := f.providerMessageID(ref.ID(), 1); got == nil || *got != "ses-late" {
+		t.Fatalf("bound = %v, want ses-late", got)
+	}
+}
+
+// TestProviderTokenSettlementNormalizesProviderMessageID: the relay reports
+// SES's id qualified (<id@region.amazonses.com>) and SES's feedback reports it
+// bare. Both spellings must be one binding, or the two writers that settle the
+// same attempt will refuse each other.
+func TestProviderTokenSettlementNormalizesProviderMessageID(t *testing.T) {
+	f := newFixture(t)
+	g := f.gate(enforcingPolicy(nil))
+	ref, auth := f.redeemed(g, f.agent(f.user("standard")))
+
+	qualified := sendingpolicy.ProviderSettlement{
+		Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderAccepted,
+		ProviderMessageID: "<010f0193abcdef00-000000@us-east-2.amazonses.com>",
+	}
+	if err := g.SettleProvider(f.ctx, qualified); err != nil {
+		t.Fatalf("settle qualified: %v", err)
+	}
+	if got := f.providerMessageID(ref.ID(), 1); got == nil || *got != "010f0193abcdef00-000000" {
+		t.Fatalf("bound = %v, want the bare id", got)
+	}
+	bare := qualified
+	bare.ProviderMessageID = "010f0193abcdef00-000000"
+	if err := g.SettleProvider(f.ctx, bare); err != nil {
+		t.Fatalf("settle bare after qualified: %v (want idempotent)", err)
+	}
+	bracketed := qualified
+	bracketed.ProviderMessageID = "<010f0193abcdef00-000000>"
+	if err := g.SettleProvider(f.ctx, bracketed); err != nil {
+		t.Fatalf("settle bracketed after qualified: %v (want idempotent)", err)
+	}
+	other := qualified
+	other.ProviderMessageID = "<010f0193abcdef00-000001@us-east-2.amazonses.com>"
+	if err := g.SettleProvider(f.ctx, other); !errors.Is(err, sendingpolicy.ErrProviderMessageIDConflict) {
+		t.Fatalf("different id err = %v, want ErrProviderMessageIDConflict", err)
+	}
+}
+
+// TestProviderTokenSettlementRequiresTheSocketToHaveOpened: an attempt that
+// was authorized but never redeemed cannot be settled as accepted — nothing
+// reached the provider, so there is no provider outcome to record.
+func TestProviderTokenSettlementRequiresTheSocketToHaveOpened(t *testing.T) {
+	f := newFixture(t)
+	g := f.gate(enforcingPolicy(nil))
+	ref, attempt := f.prepareAndReserve(g, f.agent(f.user("standard")), 1)
+	if _, auth, err := g.ConsumeAttempt(f.ctx, attempt); err != nil || auth == nil {
+		t.Fatalf("authorize: auth=%v err=%v", auth, err)
+	}
+
+	err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{
+		Attempt: attempt, Outcome: sendingpolicy.SettlementProviderAccepted, ProviderMessageID: "ses-id-never-sent",
+	})
+	if !errors.Is(err, sendingpolicy.ErrAttemptStale) {
+		t.Fatalf("settle without redeem err = %v, want ErrAttemptStale", err)
+	}
+	if got := f.providerMessageID(ref.ID(), 1); got != nil {
+		t.Fatalf("bound = %q for an attempt that never dialed", *got)
+	}
+	if _, callState := f.reservationState(ref.ID(), 1); callState != "authorized" {
+		t.Fatalf("call_state = %s, want authorized (untouched)", callState)
+	}
+}
+
+// TestProviderTokenHoldsOnATenantNameThatCannotBeAHeader: a tenant name with a
+// line break would be refused by the adapter, silently and forever. The gate
+// holds the send with a visible reason instead.
+func TestProviderTokenHoldsOnATenantNameThatCannotBeAHeader(t *testing.T) {
+	f := newFixture(t)
+	g := f.gate(enforcingPolicy(func(p *sendingpolicy.RuntimePolicy) {
+		p.TenantHeaderMode = sendingpolicy.TenantHeaderEnforce
+	}))
+	user := f.user("standard")
+	agent := f.agent(user)
+	if _, err := f.pool.Exec(f.ctx, `
+		INSERT INTO account_sending_controls (user_id, ses_tenant_name, ses_tenant_ready, ses_tenant_ready_at)
+		VALUES ($1, $2, true, now())
+		ON CONFLICT (user_id) DO UPDATE
+		    SET ses_tenant_name = EXCLUDED.ses_tenant_name, ses_tenant_ready = true, ses_tenant_ready_at = now()`,
+		user, "good\r\nX-SES-CONFIGURATION-SET: attacker-set",
+	); err != nil {
+		t.Fatal(err)
+	}
+	_, ref := f.prepareMessage(g, f.message(agent, "own_address", 1))
+	d := f.authorize(g, ref)
+	if d.Allow || d.Reason != sendingpolicy.ReasonTenantUnnamed {
+		t.Fatalf("decision = %+v, want a hold with reason %q", d, sendingpolicy.ReasonTenantUnnamed)
+	}
+}
+
+// TestProviderTokenPauseNoticeToAPausedOwnerStillRedeems pins the customer-only
+// guard on RedeemProviderCall's pause re-check: the notice telling an account
+// it was paused is SOURCED from that paused account, so an unguarded re-read
+// would refuse the one email the pause exists to send.
+func TestProviderTokenPauseNoticeToAPausedOwnerStillRedeems(t *testing.T) {
+	f := newFixture(t)
+	g := f.gate(enforcingPolicy(nil))
+	user := f.user("standard")
+	f.pause(user)
+	eventID := f.pauseNotice(user)
+
+	var ref sendingpolicy.OperationRef
+	f.inTx(func(tx pgx.Tx) error {
+		var err error
+		ref, err = g.PrepareProtectionNoticeTx(f.ctx, tx,
+			sendingpolicy.NewProtectionNoticeRef(eventID, sendingpolicy.AudienceOwner))
+		return err
+	})
+	_, attempt, err := g.Reserve(f.ctx, ref)
+	if err != nil {
+		t.Fatalf("reserve: %v", err)
+	}
+	decision, auth, err := g.ConsumeAttempt(f.ctx, attempt)
+	if err != nil || auth == nil {
+		t.Fatalf("consume: decision=%+v auth=%v err=%v", decision, auth, err)
+	}
+	if err := g.RedeemProviderCall(f.ctx, *auth); err != nil {
+		t.Fatalf("redeem of a pause notice to a PAUSED owner: %v — the notice must still go out", err)
+	}
+}
+
+// TestProviderTokenSettlementComparesNormalizedProviderMessageID: a row bound
+// by another writer in the qualified spelling must not refuse a bare replay.
+func TestProviderTokenSettlementComparesNormalizedProviderMessageID(t *testing.T) {
+	f := newFixture(t)
+	g := f.gate(enforcingPolicy(nil))
+	ref, auth := f.redeemed(g, f.agent(f.user("standard")))
+	if _, err := f.pool.Exec(f.ctx, `
+		UPDATE sending_feedback_correlations SET provider_message_id = $3
+		 WHERE operation_id = $1 AND submission_attempt = $2`,
+		ref.ID(), 1, "<010f0193abcdef00-000000@us-east-2.amazonses.com>",
+	); err != nil {
+		t.Fatal(err)
+	}
+	if err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{
+		Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderAccepted, ProviderMessageID: "010f0193abcdef00-000000",
+	}); err != nil {
+		t.Fatalf("bare replay over a qualified binding: %v", err)
+	}
+	if err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{
+		Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderAccepted, ProviderMessageID: "other-000000",
+	}); !errors.Is(err, sendingpolicy.ErrProviderMessageIDConflict) {
+		t.Fatalf("different id err = %v, want ErrProviderMessageIDConflict", err)
+	}
+}
diff --git a/internal/sendingpolicy/runtime_attestation_response_loss_test.go b/internal/sendingpolicy/runtime_attestation_response_loss_test.go
index 9759bbbef..56e642d7f 100644
--- a/internal/sendingpolicy/runtime_attestation_response_loss_test.go
+++ b/internal/sendingpolicy/runtime_attestation_response_loss_test.go
@@ -8,7 +8,7 @@ import (
 	"time"
 
 	"github.com/jackc/pgx/v5"
-	"github.com/tokencanopy/e2a/internal/testutil"
+	"github.com/tokencanopy/e2a/internal/testutil/testdb"
 )
 
 var errSyntheticCommitResponseLoss = errors.New("synthetic commit response loss")
@@ -19,7 +19,7 @@ func responseLossDigest(hexDigit string) string {
 
 func TestRuntimeAttestationCommitResponseLossRereadsExactSuccess(t *testing.T) {
 	ctx := context.Background()
-	pool := testutil.TestDB(t)
+	pool := testdb.TestDB(t)
 	m := NewModule(pool, Secrets{})
 	current, err := m.InspectAttestation(ctx)
 	if err != nil {
@@ -58,7 +58,7 @@ func TestRuntimeAttestationCommitResponseLossRereadsExactSuccess(t *testing.T) {
 func TestRuntimeAttestationCommitResponseLossRereadsAfterCallerCancellation(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	pool := testutil.TestDB(t)
+	pool := testdb.TestDB(t)
 	m := NewModule(pool, Secrets{})
 	current, err := m.InspectAttestation(ctx)
 	if err != nil {
@@ -97,7 +97,7 @@ func TestRuntimeAttestationCommitResponseLossRereadsAfterCallerCancellation(t *t
 
 func TestRuntimeAttestationCommitFailureClassifiesUnchangedState(t *testing.T) {
 	ctx := context.Background()
-	pool := testutil.TestDB(t)
+	pool := testdb.TestDB(t)
 	m := NewModule(pool, Secrets{})
 	current, err := m.InspectAttestation(ctx)
 	if err != nil {
@@ -139,7 +139,7 @@ func TestRuntimeAttestationCommitFailureClassifiesUnchangedState(t *testing.T) {
 
 func TestRuntimeAttestationCommitResponseLossClassifiesHigherRevisionStale(t *testing.T) {
 	ctx := context.Background()
-	pool := testutil.TestDB(t)
+	pool := testdb.TestDB(t)
 	m := NewModule(pool, Secrets{})
 	other := NewModule(pool, Secrets{})
 	current, err := m.InspectAttestation(ctx)
@@ -199,7 +199,7 @@ func TestRuntimeAttestationConcurrentAbortFenceBothLockOrders(t *testing.T) {
 
 	runRace := func(t *testing.T, firstIsFence bool) {
 		t.Helper()
-		pool := testutil.TestDB(t)
+		pool := testdb.TestDB(t)
 		first := NewModule(pool, Secrets{})
 		second := NewModule(pool, Secrets{})
 		prior, err := first.InspectAttestation(ctx)
diff --git a/internal/sendingpolicy/types.go b/internal/sendingpolicy/types.go
index 9ea1974f2..dad75ab5e 100644
--- a/internal/sendingpolicy/types.go
+++ b/internal/sendingpolicy/types.go
@@ -412,6 +412,38 @@ func (o SettlementOutcome) valid() bool {
 type ProviderSettlement struct {
 	Attempt AttemptRef
 	Outcome SettlementOutcome
+	// ProviderMessageID is the id SES assigned when it accepted the message.
+	// It is bound to the attempt's feedback correlation so delivery feedback
+	// that arrives by provider id — the common case — resolves to the same
+	// attempt as feedback that arrives by the random attempt header. Only an
+	// accepted settlement may carry one; a rejection has nothing to bind.
+	ProviderMessageID string
+}
+
+// ErrProviderMessageIDConflict means an attempt is being settled with a
+// different provider message id than the one already bound to it. One attempt
+// is exactly one DATA transaction and SES assigns exactly one id to it, so a
+// second, different id is evidence of two physical sends for one charged
+// attempt — the invariant this module exists to hold — and is never absorbed.
+var ErrProviderMessageIDConflict = errors.New("sendingpolicy: attempt already settled with a different provider message id")
+
+// NormalizeProviderMessageID reduces a provider message id to the bare form the
+// provider itself reports in delivery feedback.
+//
+// The SMTP relay returns SES's id angle-bracketed and qualified with the
+// region domain (<id@us-east-2.amazonses.com>) because that is the on-wire
+// Message-ID replies must anchor on; SES's SNS feedback carries the same id
+// BARE. The correlation row exists so feedback can find its attempt, and two
+// writers — the synchronous worker and the delayed feedback finalizer — must
+// agree on one spelling or the second one is refused as a conflict. Every
+// write and comparison goes through this function; readers should too.
+func NormalizeProviderMessageID(id string) string {
+	id = strings.TrimSpace(id)
+	id = strings.TrimSuffix(strings.TrimPrefix(id, "<"), ">")
+	if at := strings.IndexByte(id, '@'); at >= 0 {
+		id = id[:at]
+	}
+	return strings.TrimSpace(id)
 }
 
 // TenantMode is the closed tenant-header state carried by an authorization.
diff --git a/internal/testutil/contract_server.go b/internal/testutil/contract_server.go
index eba420de8..86a3e3835 100644
--- a/internal/testutil/contract_server.go
+++ b/internal/testutil/contract_server.go
@@ -18,6 +18,7 @@ import (
 	"github.com/tokencanopy/e2a/internal/outbound"
 	"github.com/tokencanopy/e2a/internal/outboundsend"
 	"github.com/tokencanopy/e2a/internal/relay"
+	"github.com/tokencanopy/e2a/internal/testutil/testdb"
 	"github.com/tokencanopy/e2a/internal/unsubscribe"
 	"github.com/tokencanopy/e2a/internal/usage"
 	"github.com/tokencanopy/e2a/internal/webhook"
@@ -281,7 +282,7 @@ func (s *ContractServer) Close(ctx context.Context) error {
 		firstErr = err
 	}
 	s.WSHub.Close()
-	if err := truncateAll(ctx, s.DBPool); err != nil && firstErr == nil {
+	if err := testdb.Truncate(ctx, s.DBPool); err != nil && firstErr == nil {
 		firstErr = err
 	}
 	s.DBPool.Close()
diff --git a/internal/testutil/contract_server_river_test.go b/internal/testutil/contract_server_river_test.go
index cdccb2d7a..dd33982c4 100644
--- a/internal/testutil/contract_server_river_test.go
+++ b/internal/testutil/contract_server_river_test.go
@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/tokencanopy/e2a/internal/testutil/testdb"
 )
 
 const (
@@ -113,7 +114,7 @@ func requireReachableContractTestDB(t *testing.T) string {
 		contractDBReachabilityTimeout,
 		contractDBPreparationTimeout,
 		func(ctx context.Context) error {
-			probe, err := pgxpool.New(ctx, baseTestDBURL())
+			probe, err := pgxpool.New(ctx, testdb.BaseTestDBURL())
 			if err != nil {
 				return err
 			}
diff --git a/internal/testutil/db.go b/internal/testutil/db.go
index cee81312e..c61c49348 100644
--- a/internal/testutil/db.go
+++ b/internal/testutil/db.go
@@ -2,384 +2,35 @@ package testutil
 
 import (
 	"context"
-	"crypto/sha256"
-	"encoding/hex"
-	"errors"
-	"fmt"
-	"net/url"
-	"os"
-	"path/filepath"
-	"strings"
 	"testing"
 
-	"github.com/jackc/pgx/v5"
-	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
-	"github.com/tokencanopy/e2a/internal/identity"
-	"github.com/tokencanopy/e2a/migrations"
+	"github.com/tokencanopy/e2a/internal/testutil/testdb"
 )
 
-const defaultTestDBURL = "postgres://e2a:e2a@localhost:5433/e2a_test?sslmode=disable"
+// The database helpers live in the leaf package testdb so that packages this
+// package depends on — outbound, and through it sendingpolicy — can use them
+// from INTERNAL test files without an import cycle. These wrappers keep every
+// existing caller working unchanged; new internal tests in those packages
+// should import testdb directly.
 
-type testDBPreparationError struct {
-	stage string
-	err   error
-}
-
-func (e *testDBPreparationError) Error() string {
-	return fmt.Sprintf("%s: %v", e.stage, e.err)
-}
-
-func (e *testDBPreparationError) Unwrap() error {
-	return e.err
-}
-
-// baseTestDBURL is the configured URL before per-package derivation: the
-// E2A_TEST_DATABASE_URL override or the local-dev default. Also the admin
-// connection target for creating missing package databases.
-func baseTestDBURL() string {
-	if dbURL := os.Getenv("E2A_TEST_DATABASE_URL"); dbURL != "" {
-		return dbURL
-	}
-	return defaultTestDBURL
-}
-
-// TestDBURL returns the database URL tests should use. Inside a `go test`
-// binary it derives a PER-PACKAGE database name (<base>_pkg_<package>) so
-// packages can run in parallel: the harness truncates tables between tests,
-// which made one shared database the documented cross-package flake source
-// and forced -p 1 on every DB-backed run. The suffix comes from the test
-// binary's name (os.Args[0] = <package>.test — unique per package in this
-// repo), so every URL consumer in one test binary — TestDB, hand-built
-// pools, the in-process contract server — lands on the same database.
-// Non-test binaries (cmd/e2a-contract-server) and E2A_TEST_DB_SHARED=1 get
-// the base URL verbatim. Missing databases self-provision on first open
-// (see OpenPreparedTestDB). Concurrent sessions, agents, and worktrees are
-// isolated by the per-workspace component below, so handing each runner its
-// own base URL is no longer required — only useful for pointing a run at an
-// entirely separate server.
-func TestDBURL() string {
-	base := baseTestDBURL()
-	suffix := derivedDBSuffix()
-	if suffix == "" {
-		return base
-	}
-	u, err := url.Parse(base)
-	// Only derive on genuine postgres:// URLs. DSN keyword/value form
-	// ("host=… dbname=…") "parses" into u.Path and would be mangled into
-	// garbage; pass anything unrecognizable through verbatim.
-	if err != nil || (u.Scheme != "postgres" && u.Scheme != "postgresql") ||
-		strings.TrimPrefix(u.Path, "/") == "" {
-		return base
-	}
-	// Idempotent: a child .test process handed an already-derived URL (the
-	// harness's own re-exec tests do this) must not double-suffix it.
-	if strings.HasSuffix(strings.TrimSuffix(u.Path, "/"), suffix) {
-		return base
-	}
-	u.Path = u.Path + suffix
-	// Postgres truncates identifiers past maxPostgresIdentifier bytes, and it does
-	// so SILENTLY. Truncation lands at the END — inside the package component — so
-	// sibling packages sharing a prefix collapse onto ONE database: internal/identity
-	// and internal/idempotency both become ..._pkg_ide, then truncate each other's
-	// tables under -p 4. That is exactly the corruption this derivation exists to
-	// prevent, so a base too long to derive from has to fail loudly rather than
-	// quietly reintroduce it.
-	if name := strings.TrimPrefix(u.Path, "/"); len(name) > maxPostgresIdentifier {
-		panic(fmt.Sprintf("testutil: derived test database name %q is %d bytes, over Postgres's "+
-			"%d-byte identifier limit — Postgres would truncate it silently and collide sibling "+
-			"packages onto one database. Shorten the base in E2A_TEST_DATABASE_URL; the derived "+
-			"suffix needs %d bytes.", name, len(name), maxPostgresIdentifier, len(suffix)))
-	}
-	return u.String()
-}
-
-// maxPostgresIdentifier is Postgres's NAMEDATALEN-1 ceiling for identifiers,
-// database names included. Measured in BYTES, which is what len() reports.
-const maxPostgresIdentifier = 63
-
-// derivedDBSuffix derives the database-name suffix beneath the configured base:
-// a per-WORKSPACE component plus a per-PACKAGE component, or "" when the process
-// is not a test binary or sharing is forced.
-//
-// Two dimensions, because per-package alone was not enough. It stops packages in
-// ONE run from truncating each other, but every checkout computed the same names,
-// so two agents (or two worktrees, or a second terminal) running the same package
-// shared a database and corrupted each other. AGENTS.md asked people to hand out
-// their own base URL; that is convention, and convention does not scale across
-// callers who do not know about each other. Deriving from the module root path
-// makes the isolation structural: two checkouts cannot collide even when nobody
-// configures anything.
-//
-// Name length: <base>_ws<8>_pkg_<package> runs ~40 chars for this repo's longest
-// package names, well inside Postgres's 63-byte identifier limit. A much longer
-// custom base could push past it, where Postgres truncates silently — keep bases
-// short.
-func derivedDBSuffix() string {
-	switch strings.ToLower(os.Getenv("E2A_TEST_DB_SHARED")) {
-	case "1", "true", "yes":
-		return ""
-	}
-	bin := filepath.Base(os.Args[0])
-	if !strings.HasSuffix(bin, ".test") {
-		return ""
-	}
-	name := strings.ToLower(strings.TrimSuffix(bin, ".test"))
-	sanitized := make([]rune, 0, len(name))
-	for _, r := range name {
-		switch {
-		case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
-			sanitized = append(sanitized, r)
-		default:
-			sanitized = append(sanitized, '_')
-		}
-	}
-	return workspaceSuffix(moduleRootDir()) + "_pkg_" + string(sanitized)
-}
-
-// workspaceSuffix is the per-checkout component: a short, stable digest of the
-// module root's absolute path. Empty when the root cannot be resolved, which
-// degrades to the previous per-package-only behavior rather than failing.
-//
-// Pure and path-taking so it is directly testable: the same path must always
-// give the same suffix, and different paths must differ.
-func workspaceSuffix(moduleRoot string) string {
-	if moduleRoot == "" {
-		return ""
-	}
-	sum := sha256.Sum256([]byte(filepath.Clean(moduleRoot)))
-	return "_ws" + hex.EncodeToString(sum[:])[:8]
-}
-
-// moduleRootDir returns the directory holding go.mod at or above the working
-// directory, or "" if there is none. Symlinks are resolved so two paths that
-// reach the same checkout derive the same workspace suffix.
-func moduleRootDir() string {
-	dir, err := os.Getwd()
-	if err != nil {
-		return ""
-	}
-	if resolved, rerr := filepath.EvalSymlinks(dir); rerr == nil {
-		dir = resolved
-	}
-	for {
-		if _, statErr := os.Stat(filepath.Join(dir, "go.mod")); statErr == nil {
-			return dir
-		}
-		parent := filepath.Dir(dir)
-		if parent == dir {
-			return ""
-		}
-		dir = parent
-	}
-}
+// TestDBURL returns the per-workspace, per-package test database URL.
+func TestDBURL() string { return testdb.TestDBURL() }
 
+// OpenPreparedTestDB opens and prepares the database at dbURL.
 func OpenPreparedTestDB(ctx context.Context, dbURL string) (*pgxpool.Pool, error) {
-	if dbURL == "" {
-		dbURL = defaultTestDBURL
-	}
-
-	pool, err := pgxpool.New(ctx, dbURL)
-	if err != nil {
-		return nil, err
-	}
-
-	if err := pool.Ping(ctx); err != nil {
-		pool.Close()
-		// SQLSTATE 3D000 (invalid_catalog_name): the server is up but this
-		// per-package database doesn't exist yet — self-provision it from
-		// the base URL's server and retry once. Any other error (server
-		// down, bad credentials) keeps the caller's skip-vs-fail semantics.
-		var pgErr *pgconn.PgError
-		if !errors.As(err, &pgErr) || pgErr.Code != "3D000" {
-			return nil, err
-		}
-		if cerr := createTestDatabase(ctx, dbURL); cerr != nil {
-			return nil, cerr
-		}
-		pool, err = pgxpool.New(ctx, dbURL)
-		if err != nil {
-			return nil, err
-		}
-		if err := pool.Ping(ctx); err != nil {
-			pool.Close()
-			return nil, err
-		}
-	}
-
-	if err := runMigrations(ctx, pool); err != nil {
-		pool.Close()
-		return nil, &testDBPreparationError{stage: "run migrations", err: err}
-	}
-
-	if err := truncateAll(ctx, pool); err != nil {
-		pool.Close()
-		return nil, &testDBPreparationError{stage: "truncate tables", err: err}
-	}
-
-	return pool, nil
+	return testdb.OpenPreparedTestDB(ctx, dbURL)
 }
 
+// TestDB returns a migrated, truncated pool for this test, skipping when no
+// database is reachable.
 func TestDB(t *testing.T) *pgxpool.Pool {
 	t.Helper()
-
-	ctx := context.Background()
-	pool, err := OpenPreparedTestDB(ctx, TestDBURL())
-	if err != nil {
-		var preparationErr *testDBPreparationError
-		if errors.As(err, &preparationErr) {
-			t.Fatalf("failed to prepare test database: %v", err)
-		}
-		t.Skipf("test database not available: %v", err)
-	}
-
-	t.Cleanup(func() {
-		TruncateAll(t, pool)
-		pool.Close()
-	})
-
-	return pool
+	return testdb.TestDB(t)
 }
 
+// TruncateAll empties every application table.
 func TruncateAll(t *testing.T, pool *pgxpool.Pool) {
 	t.Helper()
-	err := truncateAll(context.Background(), pool)
-	if err != nil {
-		t.Fatalf("failed to truncate tables: %v", err)
-	}
-}
-
-// createTestDatabase creates dbURL's database via the base URL's server.
-// A concurrent creator racing us is success — the database exists either
-// way. Postgres reports that race two ways: 42P04 (duplicate_database, the
-// already-committed case) and 23505 (unique violation on
-// pg_database_datname_index, the losing side of a true concurrent race —
-// empirically what 8 parallel same-name creates produce on PG16).
-//
-// Error classification is load-bearing for skip-vs-fail: a failure to
-// CONNECT to the base URL keeps the caller's "DB unavailable → skip"
-// semantics, but a failure to CREATE on a reachable server (e.g. a role
-// without CREATEDB) is a preparation error — TestDB must FAIL loudly, not
-// silently skip the entire DB tier green.
-func createTestDatabase(ctx context.Context, dbURL string) error {
-	target, err := url.Parse(dbURL)
-	if err != nil {
-		return &testDBPreparationError{stage: "parse target db url", err: err}
-	}
-	name := strings.TrimPrefix(target.Path, "/")
-	if name == "" {
-		return &testDBPreparationError{stage: "derive database name", err: fmt.Errorf("no database name in %s", dbURL)}
-	}
-	conn, err := pgx.Connect(ctx, baseTestDBURL())
-	if err != nil {
-		return fmt.Errorf("connect base db to create %s: %w", name, err)
-	}
-	defer conn.Close(ctx)
-	if _, err := conn.Exec(ctx, "CREATE DATABASE "+pgx.Identifier{name}.Sanitize()); err != nil {
-		var pgErr *pgconn.PgError
-		if errors.As(err, &pgErr) && (pgErr.Code == "42P04" || pgErr.Code == "23505") {
-			return nil
-		}
-		return &testDBPreparationError{stage: "create database " + name, err: err}
-	}
-	return nil
-}
-
-func runMigrations(ctx context.Context, pool *pgxpool.Pool) error {
-	return identity.RunMigrations(ctx, pool, migrations.FS, identity.ModeAuto)
-}
-
-// truncateAll resets the DB between tests. Most tables are reached implicitly by
-// TRUNCATE ... CASCADE via their FK path to users/messages/webhooks, so they need
-// no explicit mention. Tables with NO foreign key at all cannot be reached by
-// CASCADE, so they need explicit cleanup. Currently that is:
-//
-//   - inbound_intake: written at the SMTP edge BEFORE the agent lookup, so it
-//     deliberately has no FK. Omitting it left stale dedup rows behind and made
-//     TestInboundIntake_InsertLoadDedup / _StampProcessAndFail fail on a re-run
-//     (the "insert must be new" assertions saw the previous run's rows).
-//   - sender_identity_managed_domains: deliberately survives domain deletion
-//     until asynchronous provider teardown is confirmed.
-//   - sending-protection security ledgers: provider operations, budget rows,
-//     control audit, notice outbox, and feedback provenance deliberately have no
-//     customer-tree FK so account/message deletion cannot erase them.
-//   - sending-protection policy state: the event/marker tables have no FK, while
-//     the runtime-policy and attestation singletons must be restored to their
-//     migration-owned generation-zero sentinels between tests.
-//
-// Use DELETE for FK-less tables instead of adding them to TRUNCATE. The test suite
-// calls this helper hundreds of times; repeatedly truncating inbound_intake also
-// recreates and fsyncs its three indexes and requires an ACCESS EXCLUSIVE lock.
-// Any future FK-less table MUST be added to the DELETE section here.
-// truncateAllLockTimeout bounds how long cleanup will WAIT ON A LOCK — not how
-// long it may take. Cleanup is expected to be lock-free (inbound_intake is
-// DELETEd precisely so a concurrent reader's ACCESS SHARE cannot block it), so a
-// wait this long means something genuinely holds a conflicting lock. Failing
-// fast with SQLSTATE 55P03 (lock_not_available) makes that case
-// self-identifying, instead of hanging until the caller's context expires and
-// reporting an indistinguishable deadline error.
-//
-// Deliberately NOT a statement timeout: cleanup is legitimately slow under a
-// loaded parallel run (`-p 4` across every package), and slowness must not be
-// conflated with a lock conflict — that conflation is what made
-// TestTruncateAll_CleansInboundIntakeWithoutExclusiveTableLock flaky in CI.
-const truncateAllLockTimeout = "5s"
-
-func truncateAll(ctx context.Context, pool *pgxpool.Pool) error {
-	_, err := pool.Exec(ctx, `
-		SET LOCAL lock_timeout = '`+truncateAllLockTimeout+`';
-
-		DELETE FROM inbound_intake;
-		DELETE FROM sender_identity_managed_domains;
-		DELETE FROM sending_protection_notice_deliveries;
-		DELETE FROM sending_protection_notice_events;
-		DELETE FROM sending_feedback_recipients;
-		DELETE FROM sending_feedback_events;
-		DELETE FROM sending_feedback_correlations;
-		DELETE FROM sending_budget_reservations;
-		DELETE FROM sending_budget_counters;
-		DELETE FROM sending_provider_operations;
-		DELETE FROM account_sending_control_events;
-		DELETE FROM sending_protection_policy_events;
-		DELETE FROM sending_protection_runtime_attestation_events;
-		DELETE FROM sending_ramp_grandfathering;
-
-		-- This registry is append-only in application/migration use; its
-		-- unconditional trigger intentionally rejects DELETE. The disposable
-		-- test database bypasses user triggers for this one row-lock-scoped
-		-- cleanup instead of using TRUNCATE's ACCESS EXCLUSIVE table lock.
-		SET LOCAL session_replication_role = replica;
-		DELETE FROM sending_operator_recipient_versions;
-		SET LOCAL session_replication_role = origin;
-
-		DELETE FROM sending_protection_runtime_policy;
-		INSERT INTO sending_protection_runtime_policy
-		    (singleton, generation, schema_version, policy, policy_sha256, activated_at, activated_by)
-		VALUES (
-		    true, 0, 1,
-		    '{"all_customer_global_daily_recipients":5000,"bounce_min_outcomes":50,"bounce_pause_basis_points":400,"budget_hold_max_days":7,"budget_mode":"disabled","complaint_pause_basis_points":8,"critical_operational_daily_recipients":100,"daily_unlimited_plan_codes":["starter","pro","scale"],"default_account_daily_recipients":100,"detector_interval_seconds":300,"detector_mode":"disabled","detector_window_days":7,"operator_notice_recipient_version":1,"probation_global_daily_recipients":150,"ramp_days":30,"ramp_enabled":false,"ramp_start_daily":150,"ramp_target_daily":2000,"sending_control_audit_retention_days":90,"sending_feedback_post_account_retention_days":30,"shared_domain_account_daily_recipients":50,"shared_reputation_bounce_min_outcomes":1,"tenant_header_canary_account_ids":[],"tenant_header_mode":"disabled","tenant_provisioning_mode":"disabled","tenant_suppression_sync_mode":"disabled","violation_operational_daily_recipients":100}'::jsonb,
-		    '198d8cfb3220b6094a3b8dfe13cb0e2ff97c512ad87ae14609e580ae335c9ce6',
-		    now(), 'migration'
-		);
-
-		DELETE FROM sending_protection_runtime_attestation;
-		INSERT INTO sending_protection_runtime_attestation
-		    (singleton, revision, active_billing_digest, active_billing_contract,
-		     rollback_billing_digest, rollback_billing_contract, updated_by)
-		VALUES (true, 0, '', 0, '', 0, 'migration');
-
-		TRUNCATE oauth_pkce_requests, oauth_refresh_tokens, oauth_access_tokens,
-		         oauth_auth_codes, oauth_clients,
-		         usage_summaries, usage_events, webhook_deliveries,
-		         send_attempts, protection_events, messages,
-		         idempotency_keys, api_keys,
-		         agent_identities, domains,
-		         user_sessions, users CASCADE
-	`)
-	if err != nil {
-		return err
-	}
-	// Re-seed shared domain (migration seeds it but truncation removes it)
-	pool.Exec(ctx, `INSERT INTO domains (domain, user_id, verified, verified_at) VALUES ('agents.e2a.dev', NULL, true, now()) ON CONFLICT DO NOTHING`)
-	return nil
+	testdb.TruncateAll(t, pool)
 }
diff --git a/internal/testutil/testdb/db.go b/internal/testutil/testdb/db.go
new file mode 100644
index 000000000..2f507690f
--- /dev/null
+++ b/internal/testutil/testdb/db.go
@@ -0,0 +1,393 @@
+package testdb
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/tokencanopy/e2a/internal/identity"
+	"github.com/tokencanopy/e2a/migrations"
+)
+
+const defaultTestDBURL = "postgres://e2a:e2a@localhost:5433/e2a_test?sslmode=disable"
+
+type testDBPreparationError struct {
+	stage string
+	err   error
+}
+
+func (e *testDBPreparationError) Error() string {
+	return fmt.Sprintf("%s: %v", e.stage, e.err)
+}
+
+func (e *testDBPreparationError) Unwrap() error {
+	return e.err
+}
+
+// baseTestDBURL is the configured URL before per-package derivation: the
+// E2A_TEST_DATABASE_URL override or the local-dev default. Also the admin
+// connection target for creating missing package databases.
+func baseTestDBURL() string {
+	if dbURL := os.Getenv("E2A_TEST_DATABASE_URL"); dbURL != "" {
+		return dbURL
+	}
+	return defaultTestDBURL
+}
+
+// TestDBURL returns the database URL tests should use. Inside a `go test`
+// binary it derives a PER-PACKAGE database name (<base>_pkg_<package>) so
+// packages can run in parallel: the harness truncates tables between tests,
+// which made one shared database the documented cross-package flake source
+// and forced -p 1 on every DB-backed run. The suffix comes from the test
+// binary's name (os.Args[0] = <package>.test — unique per package in this
+// repo), so every URL consumer in one test binary — TestDB, hand-built
+// pools, the in-process contract server — lands on the same database.
+// Non-test binaries (cmd/e2a-contract-server) and E2A_TEST_DB_SHARED=1 get
+// the base URL verbatim. Missing databases self-provision on first open
+// (see OpenPreparedTestDB). Concurrent sessions, agents, and worktrees are
+// isolated by the per-workspace component below, so handing each runner its
+// own base URL is no longer required — only useful for pointing a run at an
+// entirely separate server.
+func TestDBURL() string {
+	base := baseTestDBURL()
+	suffix := derivedDBSuffix()
+	if suffix == "" {
+		return base
+	}
+	u, err := url.Parse(base)
+	// Only derive on genuine postgres:// URLs. DSN keyword/value form
+	// ("host=… dbname=…") "parses" into u.Path and would be mangled into
+	// garbage; pass anything unrecognizable through verbatim.
+	if err != nil || (u.Scheme != "postgres" && u.Scheme != "postgresql") ||
+		strings.TrimPrefix(u.Path, "/") == "" {
+		return base
+	}
+	// Idempotent: a child .test process handed an already-derived URL (the
+	// harness's own re-exec tests do this) must not double-suffix it.
+	if strings.HasSuffix(strings.TrimSuffix(u.Path, "/"), suffix) {
+		return base
+	}
+	u.Path = u.Path + suffix
+	// Postgres truncates identifiers past maxPostgresIdentifier bytes, and it does
+	// so SILENTLY. Truncation lands at the END — inside the package component — so
+	// sibling packages sharing a prefix collapse onto ONE database: internal/identity
+	// and internal/idempotency both become ..._pkg_ide, then truncate each other's
+	// tables under -p 4. That is exactly the corruption this derivation exists to
+	// prevent, so a base too long to derive from has to fail loudly rather than
+	// quietly reintroduce it.
+	if name := strings.TrimPrefix(u.Path, "/"); len(name) > maxPostgresIdentifier {
+		panic(fmt.Sprintf("testdb: derived test database name %q is %d bytes, over Postgres's "+
+			"%d-byte identifier limit — Postgres would truncate it silently and collide sibling "+
+			"packages onto one database. Shorten the base in E2A_TEST_DATABASE_URL; the derived "+
+			"suffix needs %d bytes.", name, len(name), maxPostgresIdentifier, len(suffix)))
+	}
+	return u.String()
+}
+
+// maxPostgresIdentifier is Postgres's NAMEDATALEN-1 ceiling for identifiers,
+// database names included. Measured in BYTES, which is what len() reports.
+const maxPostgresIdentifier = 63
+
+// derivedDBSuffix derives the database-name suffix beneath the configured base:
+// a per-WORKSPACE component plus a per-PACKAGE component, or "" when the process
+// is not a test binary or sharing is forced.
+//
+// Two dimensions, because per-package alone was not enough. It stops packages in
+// ONE run from truncating each other, but every checkout computed the same names,
+// so two agents (or two worktrees, or a second terminal) running the same package
+// shared a database and corrupted each other. AGENTS.md asked people to hand out
+// their own base URL; that is convention, and convention does not scale across
+// callers who do not know about each other. Deriving from the module root path
+// makes the isolation structural: two checkouts cannot collide even when nobody
+// configures anything.
+//
+// Name length: <base>_ws<8>_pkg_<package> runs ~40 chars for this repo's longest
+// package names, well inside Postgres's 63-byte identifier limit. A much longer
+// custom base could push past it, where Postgres truncates silently — keep bases
+// short.
+func derivedDBSuffix() string {
+	switch strings.ToLower(os.Getenv("E2A_TEST_DB_SHARED")) {
+	case "1", "true", "yes":
+		return ""
+	}
+	bin := filepath.Base(os.Args[0])
+	if !strings.HasSuffix(bin, ".test") {
+		return ""
+	}
+	name := strings.ToLower(strings.TrimSuffix(bin, ".test"))
+	sanitized := make([]rune, 0, len(name))
+	for _, r := range name {
+		switch {
+		case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
+			sanitized = append(sanitized, r)
+		default:
+			sanitized = append(sanitized, '_')
+		}
+	}
+	return workspaceSuffix(moduleRootDir()) + "_pkg_" + string(sanitized)
+}
+
+// workspaceSuffix is the per-checkout component: a short, stable digest of the
+// module root's absolute path. Empty when the root cannot be resolved, which
+// degrades to the previous per-package-only behavior rather than failing.
+//
+// Pure and path-taking so it is directly testable: the same path must always
+// give the same suffix, and different paths must differ.
+func workspaceSuffix(moduleRoot string) string {
+	if moduleRoot == "" {
+		return ""
+	}
+	sum := sha256.Sum256([]byte(filepath.Clean(moduleRoot)))
+	return "_ws" + hex.EncodeToString(sum[:])[:8]
+}
+
+// moduleRootDir returns the directory holding go.mod at or above the working
+// directory, or "" if there is none. Symlinks are resolved so two paths that
+// reach the same checkout derive the same workspace suffix.
+func moduleRootDir() string {
+	dir, err := os.Getwd()
+	if err != nil {
+		return ""
+	}
+	if resolved, rerr := filepath.EvalSymlinks(dir); rerr == nil {
+		dir = resolved
+	}
+	for {
+		if _, statErr := os.Stat(filepath.Join(dir, "go.mod")); statErr == nil {
+			return dir
+		}
+		parent := filepath.Dir(dir)
+		if parent == dir {
+			return ""
+		}
+		dir = parent
+	}
+}
+
+func OpenPreparedTestDB(ctx context.Context, dbURL string) (*pgxpool.Pool, error) {
+	if dbURL == "" {
+		dbURL = defaultTestDBURL
+	}
+
+	pool, err := pgxpool.New(ctx, dbURL)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := pool.Ping(ctx); err != nil {
+		pool.Close()
+		// SQLSTATE 3D000 (invalid_catalog_name): the server is up but this
+		// per-package database doesn't exist yet — self-provision it from
+		// the base URL's server and retry once. Any other error (server
+		// down, bad credentials) keeps the caller's skip-vs-fail semantics.
+		var pgErr *pgconn.PgError
+		if !errors.As(err, &pgErr) || pgErr.Code != "3D000" {
+			return nil, err
+		}
+		if cerr := createTestDatabase(ctx, dbURL); cerr != nil {
+			return nil, cerr
+		}
+		pool, err = pgxpool.New(ctx, dbURL)
+		if err != nil {
+			return nil, err
+		}
+		if err := pool.Ping(ctx); err != nil {
+			pool.Close()
+			return nil, err
+		}
+	}
+
+	if err := runMigrations(ctx, pool); err != nil {
+		pool.Close()
+		return nil, &testDBPreparationError{stage: "run migrations", err: err}
+	}
+
+	if err := truncateAll(ctx, pool); err != nil {
+		pool.Close()
+		return nil, &testDBPreparationError{stage: "truncate tables", err: err}
+	}
+
+	return pool, nil
+}
+
+func TestDB(t *testing.T) *pgxpool.Pool {
+	t.Helper()
+
+	ctx := context.Background()
+	pool, err := OpenPreparedTestDB(ctx, TestDBURL())
+	if err != nil {
+		var preparationErr *testDBPreparationError
+		if errors.As(err, &preparationErr) {
+			t.Fatalf("failed to prepare test database: %v", err)
+		}
+		t.Skipf("test database not available: %v", err)
+	}
+
+	t.Cleanup(func() {
+		TruncateAll(t, pool)
+		pool.Close()
+	})
+
+	return pool
+}
+
+// Truncate empties every application table, for callers that own their own
+// lifecycle (the contract server's Close) rather than a *testing.T.
+func Truncate(ctx context.Context, pool *pgxpool.Pool) error { return truncateAll(ctx, pool) }
+
+// BaseTestDBURL returns the configured base URL without the per-workspace,
+// per-package suffix — the database an administrative probe connects to.
+func BaseTestDBURL() string { return baseTestDBURL() }
+
+func TruncateAll(t *testing.T, pool *pgxpool.Pool) {
+	t.Helper()
+	err := truncateAll(context.Background(), pool)
+	if err != nil {
+		t.Fatalf("failed to truncate tables: %v", err)
+	}
+}
+
+// createTestDatabase creates dbURL's database via the base URL's server.
+// A concurrent creator racing us is success — the database exists either
+// way. Postgres reports that race two ways: 42P04 (duplicate_database, the
+// already-committed case) and 23505 (unique violation on
+// pg_database_datname_index, the losing side of a true concurrent race —
+// empirically what 8 parallel same-name creates produce on PG16).
+//
+// Error classification is load-bearing for skip-vs-fail: a failure to
+// CONNECT to the base URL keeps the caller's "DB unavailable → skip"
+// semantics, but a failure to CREATE on a reachable server (e.g. a role
+// without CREATEDB) is a preparation error — TestDB must FAIL loudly, not
+// silently skip the entire DB tier green.
+func createTestDatabase(ctx context.Context, dbURL string) error {
+	target, err := url.Parse(dbURL)
+	if err != nil {
+		return &testDBPreparationError{stage: "parse target db url", err: err}
+	}
+	name := strings.TrimPrefix(target.Path, "/")
+	if name == "" {
+		return &testDBPreparationError{stage: "derive database name", err: fmt.Errorf("no database name in %s", dbURL)}
+	}
+	conn, err := pgx.Connect(ctx, baseTestDBURL())
+	if err != nil {
+		return fmt.Errorf("connect base db to create %s: %w", name, err)
+	}
+	defer conn.Close(ctx)
+	if _, err := conn.Exec(ctx, "CREATE DATABASE "+pgx.Identifier{name}.Sanitize()); err != nil {
+		var pgErr *pgconn.PgError
+		if errors.As(err, &pgErr) && (pgErr.Code == "42P04" || pgErr.Code == "23505") {
+			return nil
+		}
+		return &testDBPreparationError{stage: "create database " + name, err: err}
+	}
+	return nil
+}
+
+func runMigrations(ctx context.Context, pool *pgxpool.Pool) error {
+	return identity.RunMigrations(ctx, pool, migrations.FS, identity.ModeAuto)
+}
+
+// truncateAll resets the DB between tests. Most tables are reached implicitly by
+// TRUNCATE ... CASCADE via their FK path to users/messages/webhooks, so they need
+// no explicit mention. Tables with NO foreign key at all cannot be reached by
+// CASCADE, so they need explicit cleanup. Currently that is:
+//
+//   - inbound_intake: written at the SMTP edge BEFORE the agent lookup, so it
+//     deliberately has no FK. Omitting it left stale dedup rows behind and made
+//     TestInboundIntake_InsertLoadDedup / _StampProcessAndFail fail on a re-run
+//     (the "insert must be new" assertions saw the previous run's rows).
+//   - sender_identity_managed_domains: deliberately survives domain deletion
+//     until asynchronous provider teardown is confirmed.
+//   - sending-protection security ledgers: provider operations, budget rows,
+//     control audit, notice outbox, and feedback provenance deliberately have no
+//     customer-tree FK so account/message deletion cannot erase them.
+//   - sending-protection policy state: the event/marker tables have no FK, while
+//     the runtime-policy and attestation singletons must be restored to their
+//     migration-owned generation-zero sentinels between tests.
+//
+// Use DELETE for FK-less tables instead of adding them to TRUNCATE. The test suite
+// calls this helper hundreds of times; repeatedly truncating inbound_intake also
+// recreates and fsyncs its three indexes and requires an ACCESS EXCLUSIVE lock.
+// Any future FK-less table MUST be added to the DELETE section here.
+// truncateAllLockTimeout bounds how long cleanup will WAIT ON A LOCK — not how
+// long it may take. Cleanup is expected to be lock-free (inbound_intake is
+// DELETEd precisely so a concurrent reader's ACCESS SHARE cannot block it), so a
+// wait this long means something genuinely holds a conflicting lock. Failing
+// fast with SQLSTATE 55P03 (lock_not_available) makes that case
+// self-identifying, instead of hanging until the caller's context expires and
+// reporting an indistinguishable deadline error.
+//
+// Deliberately NOT a statement timeout: cleanup is legitimately slow under a
+// loaded parallel run (`-p 4` across every package), and slowness must not be
+// conflated with a lock conflict — that conflation is what made
+// TestTruncateAll_CleansInboundIntakeWithoutExclusiveTableLock flaky in CI.
+const truncateAllLockTimeout = "5s"
+
+func truncateAll(ctx context.Context, pool *pgxpool.Pool) error {
+	_, err := pool.Exec(ctx, `
+		SET LOCAL lock_timeout = '`+truncateAllLockTimeout+`';
+
+		DELETE FROM inbound_intake;
+		DELETE FROM sender_identity_managed_domains;
+		DELETE FROM sending_protection_notice_deliveries;
+		DELETE FROM sending_protection_notice_events;
+		DELETE FROM sending_feedback_recipients;
+		DELETE FROM sending_feedback_events;
+		DELETE FROM sending_feedback_correlations;
+		DELETE FROM sending_budget_reservations;
+		DELETE FROM sending_budget_counters;
+		DELETE FROM sending_provider_operations;
+		DELETE FROM account_sending_control_events;
+		DELETE FROM sending_protection_policy_events;
+		DELETE FROM sending_protection_runtime_attestation_events;
+		DELETE FROM sending_ramp_grandfathering;
+
+		-- This registry is append-only in application/migration use; its
+		-- unconditional trigger intentionally rejects DELETE. The disposable
+		-- test database bypasses user triggers for this one row-lock-scoped
+		-- cleanup instead of using TRUNCATE's ACCESS EXCLUSIVE table lock.
+		SET LOCAL session_replication_role = replica;
+		DELETE FROM sending_operator_recipient_versions;
+		SET LOCAL session_replication_role = origin;
+
+		DELETE FROM sending_protection_runtime_policy;
+		INSERT INTO sending_protection_runtime_policy
+		    (singleton, generation, schema_version, policy, policy_sha256, activated_at, activated_by)
+		VALUES (
+		    true, 0, 1,
+		    '{"all_customer_global_daily_recipients":5000,"bounce_min_outcomes":50,"bounce_pause_basis_points":400,"budget_hold_max_days":7,"budget_mode":"disabled","complaint_pause_basis_points":8,"critical_operational_daily_recipients":100,"daily_unlimited_plan_codes":["starter","pro","scale"],"default_account_daily_recipients":100,"detector_interval_seconds":300,"detector_mode":"disabled","detector_window_days":7,"operator_notice_recipient_version":1,"probation_global_daily_recipients":150,"ramp_days":30,"ramp_enabled":false,"ramp_start_daily":150,"ramp_target_daily":2000,"sending_control_audit_retention_days":90,"sending_feedback_post_account_retention_days":30,"shared_domain_account_daily_recipients":50,"shared_reputation_bounce_min_outcomes":1,"tenant_header_canary_account_ids":[],"tenant_header_mode":"disabled","tenant_provisioning_mode":"disabled","tenant_suppression_sync_mode":"disabled","violation_operational_daily_recipients":100}'::jsonb,
+		    '198d8cfb3220b6094a3b8dfe13cb0e2ff97c512ad87ae14609e580ae335c9ce6',
+		    now(), 'migration'
+		);
+
+		DELETE FROM sending_protection_runtime_attestation;
+		INSERT INTO sending_protection_runtime_attestation
+		    (singleton, revision, active_billing_digest, active_billing_contract,
+		     rollback_billing_digest, rollback_billing_contract, updated_by)
+		VALUES (true, 0, '', 0, '', 0, 'migration');
+
+		TRUNCATE oauth_pkce_requests, oauth_refresh_tokens, oauth_access_tokens,
+		         oauth_auth_codes, oauth_clients,
+		         usage_summaries, usage_events, webhook_deliveries,
+		         send_attempts, protection_events, messages,
+		         idempotency_keys, api_keys,
+		         agent_identities, domains,
+		         user_sessions, users CASCADE
+	`)
+	if err != nil {
+		return err
+	}
+	// Re-seed shared domain (migration seeds it but truncation removes it)
+	pool.Exec(ctx, `INSERT INTO domains (domain, user_id, verified, verified_at) VALUES ('agents.e2a.dev', NULL, true, now()) ON CONFLICT DO NOTHING`)
+	return nil
+}
diff --git a/internal/testutil/db_test.go b/internal/testutil/testdb/db_test.go
similarity index 98%
rename from internal/testutil/db_test.go
rename to internal/testutil/testdb/db_test.go
index 8bb29ac93..4e46dafd4 100644
--- a/internal/testutil/db_test.go
+++ b/internal/testutil/testdb/db_test.go
@@ -1,4 +1,4 @@
-package testutil
+package testdb
 
 import (
 	"context"
@@ -352,8 +352,8 @@ func TestTestDBURLIsUniquePerWorkspaceAndPackage(t *testing.T) {
 	if !strings.Contains(name, "_ws") {
 		t.Errorf("dbname = %q, want a _ws<hash> workspace component", name)
 	}
-	if !strings.HasSuffix(name, "_pkg_testutil") {
-		t.Errorf("dbname = %q, want the _pkg_testutil suffix retained", name)
+	if !strings.HasSuffix(name, "_pkg_testdb") {
+		t.Errorf("dbname = %q, want the _pkg_testdb suffix retained", name)
 	}
 	if ws := workspaceSuffix(moduleRootDir()); ws == "" || !strings.Contains(name, ws) {
 		t.Errorf("dbname = %q, want it to contain this checkout's suffix %q", name, ws)
@@ -396,14 +396,14 @@ func TestTestDBURLDerivesPerPackageDatabase(t *testing.T) {
 	// appends a per-package suffix to the base database name so packages
 	// running in parallel (-p N) cannot truncate each other's rows — the
 	// harness truncates between tests, which made a shared DB the
-	// documented cross-package flake source. This binary is testutil.test,
-	// so the derived name is <base>_pkg_testutil.
+	// documented cross-package flake source. This binary is testdb.test,
+	// so the derived name is <base>_pkg_testdb.
 	u, err := url.Parse(TestDBURL())
 	if err != nil {
 		t.Fatalf("parse TestDBURL: %v", err)
 	}
-	if got := strings.TrimPrefix(u.Path, "/"); !strings.HasSuffix(got, "_pkg_testutil") {
-		t.Errorf("TestDBURL dbname = %q, want *_pkg_testutil suffix", got)
+	if got := strings.TrimPrefix(u.Path, "/"); !strings.HasSuffix(got, "_pkg_testdb") {
+		t.Errorf("TestDBURL dbname = %q, want *_pkg_testdb suffix", got)
 	}
 
 	// E2A_TEST_DB_SHARED=1 restores the verbatim single-DB behavior (escape
@@ -469,7 +469,7 @@ func TestTestDBURLDerivationIsIdempotentAndURLOnly(t *testing.T) {
 	if err != nil {
 		t.Fatalf("parse: %v", err)
 	}
-	if got := strings.TrimPrefix(u.Path, "/"); strings.Contains(got, "_pkg_testutil_pkg_") {
+	if got := strings.TrimPrefix(u.Path, "/"); strings.Contains(got, "_pkg_testdb_pkg_") {
 		t.Errorf("double-derived dbname %q", got)
 	}