feat(outbound): close the provider seam for every sender (B7)

cmd/e2a/main.go

@@ -121,6 +121,7 @@ func main() {
 	flag.IntVar(&spFlags.activeBillingContract, "active-billing-contract", -1, "verified active billing contract level")
 	flag.StringVar(&spFlags.rollbackBillingDigest, "rollback-billing-digest", "", "verified rollback billing image digest")
 	flag.IntVar(&spFlags.rollbackBillingContract, "rollback-billing-contract", -1, "verified rollback billing contract level")
+	flag.BoolVar(&spFlags.reconcile, "reconcile-legacy-sending-jobs", false, "stamp a sending operation reference onto every pending provider-submitting job enqueued without one (cancelling orphans whose source row is gone), print counts, then exit; nonzero unless every job was decided")
 	flag.BoolVar(&spFlags.capabilities, "print-capabilities", false, "print the machine-readable capability marker (contract level, policy source, operator commitments), then exit")
 	flag.StringVar(&spFlags.reason, "reason", "", "nonblank reason recorded in the audit row of a sending-protection mutation")
 	flag.Parse()
@@ -365,6 +366,9 @@ func main() {
 	})
 	outboundJobs := outboundSending.jobs
 	registrars = append(registrars, outboundJobs)
+	// Platform mail the API sends itself (public feedback) crosses the same
+	// seam with tokens from the same gate.
+	sendingGate, providerSubmitter := outboundSending.gate, outboundSending.submitter
 	registrars = append(registrars, sendramp.NewMaintenanceJobs(rampStore))
 	// Queue depth/age gauges: a 30s maintenance periodic sampling river_job
 	// per queue+state (docs/observability.md).
@@ -389,10 +393,17 @@ func main() {
 	// later via SetDeliverer — mirrors inbound's late-bound Processor. Gated on the
 	// same relay+public-URL config as the notifier itself; when unconfigured, no jobs
 	// register and the hold takes the plain path (no notification).
-	var notifyJobs *hitlnotify.Jobs
 	notifierEnabled := cfg.OutboundSMTP.FromDomain != "" && cfg.HTTP.PublicURL != ""
-	if notifierEnabled {
-		notifyJobs = hitlnotify.NewJobs(store)
+	notification := newNotificationJobs(notificationDeps{
+		store:          store,
+		pool:           pool,
+		gate:           sendingGate,
+		metrics:        metrics,
+		hitlEnabled:    notifierEnabled,
+		webhookEnabled: cfg.OutboundSMTP.FromDomain != "",
+	})
+	notifyJobs := notification.hitl
+	if notifyJobs != nil {
 		registrars = append(registrars, notifyJobs)
 	}
 
@@ -405,9 +416,8 @@ func main() {
 	// (generic dashboard copy instead of a link). When unconfigured, no jobs
 	// register and the sweep transitions state without notifications
 	// (pre-feature behavior).
-	var webhookNotifyJobs *webhooknotify.Jobs
-	if cfg.OutboundSMTP.FromDomain != "" {
-		webhookNotifyJobs = webhooknotify.NewJobs(store).WithMetrics(metrics)
+	webhookNotifyJobs := notification.webhook
+	if webhookNotifyJobs != nil {
 		registrars = append(registrars, webhookNotifyJobs)
 	}
 
@@ -697,7 +707,7 @@ func main() {
 		// unreachable in practice — kept as a defensive guard against future drift.
 		log.Printf("[hitl] notifier disabled: notification job pipeline not registered")
 	} else {
-		notifier := hitlnotify.New(store, smtpRelay, approvalSigner, cfg.OutboundSMTP.FromDomain, cfg.Notifications.FromAddress, cfg.Notifications.ReplyTo, cfg.HTTP.PublicURL).WithDKIM(store)
+		notifier := hitlnotify.New(store, providerSubmitter, approvalSigner, cfg.OutboundSMTP.FromDomain, cfg.Notifications.FromAddress, cfg.Notifications.ReplyTo, cfg.HTTP.PublicURL).WithDKIM(store)
 		// Late-bind the concrete Deliverer onto the registered NotifyWorker (which
 		// has been running since jobsClient.Start; jobs enqueued before this bind
 		// simply retry) and give the hold path its accept-tx enqueuer. The HTTP
@@ -718,7 +728,7 @@ func main() {
 	// a BYODKIM custom from-address domain is signed here or not at all.
 	// Fail-open — no stored key (self-host default) sends unsigned.
 	if webhookNotifyJobs != nil {
-		whNotifier := webhooknotify.New(store, smtpRelay, cfg.OutboundSMTP.FromDomain, cfg.Notifications.FromAddress, cfg.Notifications.ReplyTo, cfg.HTTP.PublicURL).WithDKIM(store)
+		whNotifier := webhooknotify.New(store, providerSubmitter, cfg.OutboundSMTP.FromDomain, cfg.Notifications.FromAddress, cfg.Notifications.ReplyTo, cfg.HTTP.PublicURL).WithDKIM(store)
 		webhookNotifyJobs.SetDeliverer(whNotifier)
 		log.Printf("[webhook-notify] enabled (from=%s)", whNotifier.FromAddress())
 	} else {
@@ -833,6 +843,7 @@ func main() {
 	// The outbound accept-tx enqueuer is mandatory: DeliverOutbound always
 	// persists+enqueues and returns accepted before provider submission.
 	api.SetOutboundEnqueuer(outboundJobs)
+	outboundSending.armAPI(api)
 	// Slices 6 + 7: customer-facing events API needs the raw pool to
 	// query webhook_events and write webhook_subscriber_deliveries on
 	// replay. Kept as a separate setter so a future refactor can route

cmd/e2a/outbound_wiring.go

@@ -4,9 +4,12 @@ import (
 	"github.com/jackc/pgx/v5/pgxpool"
 
 	"github.com/tokencanopy/e2a/internal/agent"
+	"github.com/tokencanopy/e2a/internal/hitlnotify"
+	"github.com/tokencanopy/e2a/internal/identity"
 	"github.com/tokencanopy/e2a/internal/outbound"
 	"github.com/tokencanopy/e2a/internal/outboundsend"
 	"github.com/tokencanopy/e2a/internal/sendingpolicy"
+	"github.com/tokencanopy/e2a/internal/webhooknotify"
 )
 
 // outboundSendingDeps is everything the outbound composition root needs. It
@@ -49,3 +52,45 @@ func newOutboundSending(d outboundSendingDeps) outboundSending {
 		WithRateGate(d.rate)
 	return outboundSending{gate: gate, submitter: submitter, jobs: jobs}
 }
+
+// notificationDeps is what the notification composition needs: the same gate
+// and pool the customer path uses, plus the two config gates main applies.
+type notificationDeps struct {
+	store          *identity.Store
+	pool           *pgxpool.Pool
+	gate           sendingpolicy.Gate
+	metrics        webhooknotify.Metrics
+	hitlEnabled    bool // outbound_smtp.from_domain and http.public_url set
+	webhookEnabled bool // outbound_smtp.from_domain set
+}
+
+// notificationJobs are the two notification job bundles, nil when their
+// feature is unconfigured (no worker registers, the sweep/hold take the
+// plain path).
+type notificationJobs struct {
+	hitl    *hitlnotify.Jobs
+	webhook *webhooknotify.Jobs
+}
+
+// newNotificationJobs composes the notification bundles over the ONE gate.
+// Every enqueue prepares a customer_notification operation in the source
+// transaction and every worker execution authorizes through the gate; a
+// bundle built any other way would fail closed at runtime (empty token) with
+// an error that says nothing about wiring, which is why the composition is
+// factored here and pinned by the wiring test.
+func newNotificationJobs(d notificationDeps) notificationJobs {
+	var n notificationJobs
+	if d.hitlEnabled {
+		n.hitl = hitlnotify.NewJobs(d.store).WithGate(d.gate, d.pool)
+	}
+	if d.webhookEnabled {
+		n.webhook = webhooknotify.NewJobs(d.store).WithMetrics(d.metrics).WithGate(d.gate, d.pool)
+	}
+	return n
+}
+
+// armAPI hands the API the authorized seam for the platform mail it sends
+// itself (public feedback).
+func (s outboundSending) armAPI(api *agent.API) {
+	api.SetProviderSubmitter(s.submitter, s.gate)
+}

cmd/e2a/sending_policy.go

@@ -24,6 +24,7 @@ type sendingProtectionFlags struct {
 	register     bool
 	attest       bool
 	capabilities bool
+	reconcile    bool
 
 	expectedGeneration int64
 	expectedPolicySHA  string
@@ -40,12 +41,12 @@ type sendingProtectionFlags struct {
 }
 
 func (f *sendingProtectionFlags) commandRequested() bool {
-	return f.inspect || f.activate || f.register || f.attest || f.capabilities
+	return f.inspect || f.activate || f.register || f.attest || f.capabilities || f.reconcile
 }
 
 func (f *sendingProtectionFlags) selectedCount() int {
 	n := 0
-	for _, set := range []bool{f.inspect, f.activate, f.register, f.attest, f.capabilities} {
+	for _, set := range []bool{f.inspect, f.activate, f.register, f.attest, f.capabilities, f.reconcile} {
 		if set {
 			n++
 		}
@@ -105,6 +106,8 @@ func runSendingProtectionCommand(ctx context.Context, cfg *config.Config, pool *
 		return runRuntimeAttest(ctx, module, f, stdout)
 	case f.capabilities:
 		return runPrintCapabilities(source, secrets, stdout)
+	case f.reconcile:
+		return runReconcileLegacySendingJobs(ctx, pool, sendingpolicy.NewGate(pool, secrets, source, policy), stdout)
 	}
 	return errors.New("no sending-protection command selected")
 }

cmd/e2a/sending_policy_test.go

@@ -312,6 +312,17 @@ func TestSendingProtectionCommands(t *testing.T) {
 		}
 	})
 
+	t.Run("reconcile-legacy-sending-jobs dispatches", func(t *testing.T) {
+		resetRiverJobs(t, pool)
+		out, err := run(&sendingProtectionFlags{reconcile: true})
+		if err != nil {
+			t.Fatalf("reconcile: %v", err)
+		}
+		if !strings.Contains(out, "scanned:   0") || !strings.Contains(out, "remaining: 0") {
+			t.Errorf("reconcile output = %q", out)
+		}
+	})
+
 	t.Run("print-capabilities", func(t *testing.T) {
 		clearEnvForTest(t)
 		out, err := run(&sendingProtectionFlags{capabilities: true})

cmd/e2a/sending_policy_wiring_test.go

@@ -8,10 +8,12 @@ import (
 
 	"github.com/riverqueue/river"
 
+	"github.com/tokencanopy/e2a/internal/agent"
 	"github.com/tokencanopy/e2a/internal/config"
 	"github.com/tokencanopy/e2a/internal/outbound"
 	"github.com/tokencanopy/e2a/internal/sendingpolicy"
 	"github.com/tokencanopy/e2a/internal/testutil/testdb"
+	"github.com/tokencanopy/e2a/internal/usage"
 )
 
 // TestSendingPolicyWiring builds the production outbound composition from
@@ -76,3 +78,50 @@ func TestSendingPolicyWiring(t *testing.T) {
 		t.Fatal("a never-prepared operation resolved")
 	}
 }
+
+// TestNotificationAndPlatformMailWiring pins the three composition-root
+// edges the AST closure guard cannot see: both notification bundles hold the
+// gate (so their enqueues prepare operations and their workers authorize),
+// and the API holds the submitter + gate for public feedback. Dropping any
+// of them fails closed at runtime with an opaque "authorization required"
+// error; this is where it fails loudly instead.
+func TestNotificationAndPlatformMailWiring(t *testing.T) {
+	pool := testdb.TestDB(t)
+	relay := outbound.NewSMTPRelay(&config.OutboundSMTPConfig{Host: "relay.invalid", Port: 587, FromDomain: "test.e2a.dev"})
+	composed := newOutboundSending(outboundSendingDeps{
+		pool:    pool,
+		relay:   relay,
+		secrets: sendingpolicy.Secrets{},
+		source:  sendingpolicy.PolicySourceConfig,
+		policy:  sendingpolicy.DisabledPolicy(),
+	})
+
+	n := newNotificationJobs(notificationDeps{pool: pool, gate: composed.gate, hitlEnabled: true, webhookEnabled: true})
+	if n.hitl == nil || n.hitl.Gate() != composed.gate {
+		t.Fatal("hitl notification bundle does not hold the composed gate")
+	}
+	if n.webhook == nil || n.webhook.Gate() != composed.gate {
+		t.Fatal("webhook notification bundle does not hold the composed gate")
+	}
+	// The registered workers are what run; they must carry the gate too.
+	if w := n.hitl.NotifyWorker(); w == nil || w.Gate() != composed.gate {
+		t.Fatal("hitl notify worker registered without the gate")
+	}
+	if w := n.webhook.NotifyWorker(); w == nil || w.Gate() != composed.gate {
+		t.Fatal("webhook notify worker registered without the gate")
+	}
+
+	off := newNotificationJobs(notificationDeps{pool: pool, gate: composed.gate})
+	if off.hitl != nil || off.webhook != nil {
+		t.Fatal("unconfigured notifications must register nothing")
+	}
+
+	api := agent.NewAPI(nil, nil, relay, nil, usage.NewNoopUsageTracker(), "e2a.dev", "test.e2a.dev", "agents.e2a.dev", "", false)
+	if api.ProviderSubmitterWired() {
+		t.Fatal("a fresh API must not claim a submitter")
+	}
+	composed.armAPI(api)
+	if !api.ProviderSubmitterWired() {
+		t.Fatal("armAPI did not hand the API the submitter and gate")
+	}
+}