Phase 3: human-in-the-loop at scale, secrets rotation, reputation monitoring

.env.example

@@ -74,6 +74,9 @@ RELAY_ESPO_API_KEY=
 # Master key for per-tenant key derivation (HKDF). Dev value only; production
 # uses a KMS-managed key (Phase 3).
 RELAY_MASTER_KEY=dev-master-key-not-for-production
+# Rotation window: set the OLD master key here while rotating so signed
+# unsubscribe links in already-delivered mail keep verifying; clear after.
+RELAY_MASTER_KEY_PREVIOUS=
 
 # ── Rate limiting & retries (Phase 2) ───────────────────────────────────────
 # Requests/second per external target; 0 disables the bucket. Waits beyond
@@ -126,6 +129,10 @@ RELAY_WARMUP_DAILY_START=0                   # day-0 daily cap (0 = no ramp)
 RELAY_WARMUP_DAILY_INCREMENT=0               # daily cap growth per day
 RELAY_BOUNCE_COMPLAINT_WINDOW_DAYS=7
 RELAY_MAX_BOUNCES_COMPLAINTS_IN_WINDOW=2
+# Reputation alert: fire when 24h hard-bounce rate exceeds the fraction,
+# once at least MIN_SENDS went out (1/1 is noise, not reputation).
+RELAY_ALERT_BOUNCE_RATE=0.05
+RELAY_ALERT_BOUNCE_RATE_MIN_SENDS=5
 # SES event ingestion: webhook token (SNS HTTPS) and/or SQS queue (polling).
 # The relay-ses-events queue is Part 2 — leave empty until it exists.
 RELAY_SES_WEBHOOK_TOKEN=

README.md

@@ -130,6 +130,13 @@ to change safely.
 | --- | --- |
 | **One-click unsubscribe (RFC 8058)**: every real send embeds a per-job signed-token URL in its List-Unsubscribe header (beside the mailto). `GET /unsubscribe` renders a confirm page and never mutates state (mail clients and scanners prefetch links); the `POST` honors it idempotently — the lead transitions to `unsubscribed` where the state machine allows, and the do-not-contact suppression entry ALWAYS lands, decoupled, same pattern as bounces. Tokens are HMAC-signed with a per-tenant derived key and carry no PII | `ingest/unsubscribe.py`, `api/routes.py`, `senders/ses.py` |
 | **Deliverability pacing**: per-mailbox rolling-hour cap, minimum spacing between sends, and a warmup ramp that grows the effective daily cap from the tenant's first real send (`min(cap, start + increment·day)`). Pacing is execution-time only and **defers** — a paced-out job stays queued for a later tick, its lead untouched; it is never terminally blocked over a temporal condition. Evaluated under the same per-tenant advisory lock as the daily cap, so racing workers cannot both pass at a pace boundary. All off by default (`RELAY_REAL_SEND_HOURLY_CAP`, `RELAY_REAL_SEND_MIN_SPACING_SECONDS`, `RELAY_WARMUP_DAILY_*`) | `domain/eligibility.py`, `workers/send_worker.py` |
+| **Human-in-the-loop at scale**: the review queue is confidence-ordered (highest `fit_score` first — the batchable tail on top, reviewer attention at the bottom); a batch-review endpoint processes up to 100 rubric decisions per call, each in its own transaction so one stale item fails alone; the edit rate (`approved_with_edits` share) is a first-class metric — edits-as-signal for prompt iteration | `api/routes.py`, `observability/metrics.py` |
+| **Reputation monitoring**: 24h bounce/complaint rates and per-reason suppression counts in `/metrics` and the Prometheus export; a `bounce_rate_high` critical alert fires past `RELAY_ALERT_BOUNCE_RATE` — with a `_MIN_SENDS` floor so 1-of-1 noise never pages — BEFORE the eligibility threshold silently pauses sending | `observability/metrics.py`, `observability/alerts.py` |
+| **Secrets rotation**: `POST /internal/tenants/{id}/rotate-key` (admin) issues a new tenant API key, kills the old one instantly, and audits the rotation; `RELAY_MASTER_KEY_PREVIOUS` gives master-key rotation a verify-only window so unsubscribe links already sitting in delivered mail keep working — a dead unsubscribe link is a compliance failure | `api/routes.py`, `ingest/unsubscribe.py` |
+
+What code cannot close — the production-posture, legal, and review
+items, plus the three deliberately parked decisions — is recorded in
+[docs/phase3-readiness.md](docs/phase3-readiness.md).
 
 ---
 

docs/phase3-readiness.md

@@ -0,0 +1,55 @@
+# Phase 3 — Production Readiness: status against the exit gate
+
+Phase 3's exit gate cannot be closed by code alone: it requires real
+outbound volume, monitored deliverability over time, and a human
+security + compliance review. This document separates what the codebase
+now **structurally provides** from what remains an **operator/legal
+deliverable**, and records the deliberately parked decisions.
+
+## What the code provides (each item pinned by tests)
+
+| Exit-gate concern | Mechanism | Where |
+| --- | --- | --- |
+| Suppression before every send | eligibility gate + DB trigger on queue AND claim | `fn_is_suppressed`, `fn_send_jobs_guard` |
+| Permanent unsubscribe, incl. one-click | RFC 8058 signed-token endpoint; suppression always lands, decoupled from lead state; tokens survive master-key rotation | `ingest/unsubscribe.py` |
+| Bounce/complaint handling with automatic pausing | SNS-verified ingestion → auto-suppress → `campaign_below_thresholds` blocks further real sends | `ingest/ses_events.py`, `domain/eligibility.py` |
+| Volume caps, warmup, pacing | daily cap (race-proof, advisory-lock serialized), hourly cap, min spacing, warmup ramp; pacing defers rather than blocks | `domain/eligibility.py`, `workers/send_worker.py` |
+| Reputation monitoring | bounce/complaint rates in `/metrics` (+ Prometheus), `bounce_rate_high` critical alert with a min-sends floor | `observability/` |
+| Human-in-the-loop at scale | confidence-ordered review queue, batch review endpoint (per-item transactions), edit-rate as a first-class metric | `api/routes.py`, `observability/metrics.py` |
+| Retention / deletion / DSR | erasure leaves only the hashed do-not-contact entry; retention purge never fabricates an opt-out | `domain/dsr.py`, `workers/retention_worker.py` |
+| DR: tested restore, in-flight durability | pg_dump→restore test proves erasure survives backups; crash recovery closes orphans on every tick | `tests/test_adversarial.py`, `pipeline/recovery.py` |
+| Audit trail | append-only, redacted, every consequential action | `audit.py`, DB triggers |
+| Secrets rotation | tenant API key rotation endpoint (old key dies instantly, audited); `RELAY_MASTER_KEY_PREVIOUS` verify-only rotation window | `api/routes.py`, `config.py` |
+| Tenant isolation | FORCEd RLS on every tenant-bearing table, tested cross-tenant | `db/sql/004_rls.sql` |
+
+## Operator / legal deliverables (code cannot close these)
+
+- **Production sending posture** — leaving the SES sandbox, dedicated
+  authenticated domains at volume, DMARC report review cadence,
+  inbox-placement monitoring. Gated by the §6 revisit criteria in
+  [the sending-provider decision record](decisions/sending-provider.md).
+- **Region-specific suppression / lawful-basis rules** — the
+  `lawful_send_basis` check is a named seam awaiting the Legal/Data
+  Preflight's jurisdiction matrix (GDPR / CASL / CAN-SPAM, verified
+  current at build time). Code must not invent this.
+- **Client contract / DPA, subprocessor list, incident-response
+  process, abuse-prevention policy** — human/legal documents.
+- **KMS-managed master key** — the derivation seam is ready
+  (`derive_tenant_key`); swapping the dev master key for KMS is a
+  deployment change plus the parked pepper decision below.
+- **Human security + compliance review** — the exit gate requires it
+  explicitly; an automated audit is input to it, not a substitute.
+
+## Parked decisions (deliberate, awaiting the operator)
+
+1. **Email-hash HMAC pepper** — `hash_email` is unkeyed SHA-256; a
+   DSR-erased suppression hash is theoretically reversible by guessing
+   a known address. Peppering changes every stored digest, so it needs
+   a migration plan; fold into the KMS/master-key work.
+2. **Global-scope suppression cross-tenant asymmetry** — any tenant can
+   insert a `scope='global'` row that silently blocks every other
+   tenant's sends, which those tenants can neither see nor remove.
+   Over-suppression is the safe direction, but the asymmetry needs a
+   deliberate multi-tenant decision before Phase 4.
+3. **`sequence_step == 1` hardcoded** in the idempotency/duplicate
+   check — must be generalized before multi-step sequences ship.

src/relay/api/routes.py

@@ -83,6 +83,37 @@ def create_tenant(
     return schemas.TenantCreateResponse(id=tenant_id, name=body.name, api_key=api_key)
 
 
+@router.post(
+    "/internal/tenants/{tenant_id}/rotate-key",
+    response_model=schemas.TenantKeyRotateResponse,
+    dependencies=[Depends(require_admin)],
+)
+def rotate_tenant_key(tenant_id: uuid.UUID) -> schemas.TenantKeyRotateResponse:
+    """Rotate a tenant's API key (Phase 3: secrets rotation).
+
+    The old key stops working the moment this commits — rotation is for
+    suspected exposure, so a grace overlap would defeat the point. The
+    new key is returned exactly once; only its hash is stored.
+    """
+    api_key = f"rk_{secrets.token_urlsafe(32)}"
+    with admin_session() as session:
+        tenant = session.get(Tenant, tenant_id)
+        if tenant is None:
+            raise HTTPException(status.HTTP_404_NOT_FOUND, "tenant not found")
+        tenant.api_key_hash = hash_api_key(api_key)
+        audit.record(
+            session,
+            tenant_id=tenant_id,
+            actor_type="human",
+            actor_id="admin",
+            action="tenant.rotate_key",
+            entity_type="tenant",
+            entity_id=str(tenant_id),
+            payload={"note": "api key rotated; old key invalidated"},
+        )
+    return schemas.TenantKeyRotateResponse(id=tenant_id, api_key=api_key)
+
+
 # ── Lead source register ───────────────────────────────────────────────────
 
 
@@ -402,7 +433,12 @@ def reject(
 def pending_drafts(
     tenant_id: uuid.UUID = Depends(require_tenant),
 ) -> schemas.PendingDraftsResponse:
-    """The reviewer's queue: drafts waiting at the human gate."""
+    """The reviewer's queue: drafts waiting at the human gate.
+
+    Confidence-ordered (highest fit score first, FIFO within a score):
+    the top of the queue is the batchable tail, the bottom is where
+    reviewer attention belongs.
+    """
     with tenant_session(tenant_id) as session:
         rows = session.execute(
             select(OutreachDraft, Lead)
@@ -412,7 +448,7 @@ def pending_drafts(
                 & (Lead.id == OutreachDraft.lead_id),
             )
             .where(OutreachDraft.status == "pending_approval")
-            .order_by(OutreachDraft.created_at)
+            .order_by(Lead.fit_score.desc().nulls_last(), OutreachDraft.created_at)
         ).all()
         return schemas.PendingDraftsResponse(
             drafts=[
@@ -427,6 +463,9 @@ def pending_drafts(
                     lead_first_name=lead.first_name,
                     lead_company=lead.company_name,
                     lead_state=lead.state,
+                    fit_score=(
+                        float(lead.fit_score) if lead.fit_score is not None else None
+                    ),
                     created_at=draft.created_at,
                 )
                 for draft, lead in rows
@@ -476,6 +515,78 @@ def review(
         )
 
 
+@router.post(
+    "/outreach-drafts/batch-review",
+    response_model=schemas.BatchReviewResponse,
+)
+def batch_review(
+    body: schemas.BatchReviewRequest,
+    tenant_id: uuid.UUID = Depends(require_tenant),
+) -> schemas.BatchReviewResponse:
+    """Review many drafts in one call (Phase 3: human-in-the-loop at scale).
+
+    Each item is processed in ITS OWN transaction through the same rubric
+    path as the single-draft endpoint — one bad item (stale draft, wrong
+    state) fails alone and the rest of the batch still lands. Like every
+    review surface, this never sends.
+    """
+    results: list[schemas.BatchReviewResultItem] = []
+    counts = {"approved": 0, "approved_with_edits": 0, "rejected": 0}
+    failed = 0
+    for item in body.items:
+        try:
+            with tenant_session(tenant_id) as session:
+                draft = session.get(OutreachDraft, item.draft_id)
+                if draft is None:
+                    raise ApprovalError("draft not found")
+                outcome = review_draft(
+                    session,
+                    draft=draft,
+                    reviewer=body.reviewer,
+                    decision=item.decision,
+                    reasons=item.reasons,
+                    notes=item.notes,
+                    edited_subject=item.edited_subject,
+                    edited_body=item.edited_body,
+                )
+                lead = session.get(Lead, draft.lead_id)
+                results.append(
+                    schemas.BatchReviewResultItem(
+                        draft_id=item.draft_id,
+                        ok=True,
+                        decision=item.decision,
+                        active_draft_id=outcome.active_draft_id,
+                        lead_state=lead.state if lead else None,
+                    )
+                )
+                counts[str(item.decision)] += 1
+        except (ApprovalError, ValueError, TransitionError, IntegrityError) as exc:
+            failed += 1
+            results.append(
+                schemas.BatchReviewResultItem(
+                    draft_id=item.draft_id,
+                    ok=False,
+                    decision=item.decision,
+                    error=str(exc)[:500],
+                )
+            )
+    log.info(
+        "batch review processed",
+        reviewer=body.reviewer,
+        approved=counts["approved"],
+        edited=counts["approved_with_edits"],
+        rejected=counts["rejected"],
+        failed=failed,
+    )
+    return schemas.BatchReviewResponse(
+        results=results,
+        approved=counts["approved"],
+        edited=counts["approved_with_edits"],
+        rejected=counts["rejected"],
+        failed=failed,
+    )
+
+
 # ── The approval UI: a static page, credentials stay client-side ────────────
 
 
@@ -528,8 +639,13 @@ def metrics_json(
         replies_window=m.replies_window,
         sent_window=m.sent_window,
         suppression_entries=m.suppression_entries,
+        suppressions_window=m.suppressions_window,
+        reviews_window=m.reviews_window,
         run_error_rate=m.run_error_rate,
         reply_rate=m.reply_rate,
+        bounce_rate=m.bounce_rate,
+        complaint_rate=m.complaint_rate,
+        edit_rate=m.edit_rate,
     )
 
 

src/relay/api/schemas.py

@@ -51,6 +51,13 @@ class TenantCreateResponse(BaseModel):
     api_key: str
 
 
+class TenantKeyRotateResponse(BaseModel):
+    id: uuid.UUID
+    #: The NEW key, shown exactly once; the old key stops working
+    #: immediately (only the new hash is stored).
+    api_key: str
+
+
 # ── Lead source register (§7) ───────────────────────────────────────────────
 
 
@@ -264,13 +271,54 @@ class PendingDraftItem(BaseModel):
     lead_first_name: str | None
     lead_company: str | None
     lead_state: str
+    #: The scoring confidence for this lead — the queue is ordered by it
+    #: (highest first) so reviewers can batch the confident tail and
+    #: spend their attention on the uncertain one.
+    fit_score: float | None = None
     created_at: datetime
 
 
 class PendingDraftsResponse(BaseModel):
     drafts: list[PendingDraftItem]
 
 
+# ── Batched review (Phase 3 human-in-the-loop at scale) ─────────────────────
+
+
+class BatchReviewItem(BaseModel):
+    draft_id: uuid.UUID
+    decision: ReviewDecision
+    reasons: list[ReviewReason] = Field(default_factory=list)
+    notes: str | None = Field(default=None, max_length=2000)
+    edited_subject: str | None = Field(default=None, max_length=200)
+    edited_body: str | None = Field(default=None, max_length=5000)
+
+
+class BatchReviewRequest(BaseModel):
+    reviewer: str = Field(min_length=1, max_length=200)
+    items: list[BatchReviewItem] = Field(min_length=1, max_length=100)
+
+
+class BatchReviewResultItem(BaseModel):
+    draft_id: uuid.UUID
+    ok: bool
+    decision: ReviewDecision
+    #: The approved draft after this item (None unless approved).
+    active_draft_id: uuid.UUID | None = None
+    lead_state: str | None = None
+    error: str | None = None
+
+
+class BatchReviewResponse(BaseModel):
+    results: list[BatchReviewResultItem]
+    approved: int
+    edited: int
+    rejected: int
+    failed: int
+    #: Always false: review/approval never sends (§10).
+    sent: Literal[False] = False
+
+
 # ── Economics (Phase 1A gate) ───────────────────────────────────────────────
 
 
@@ -342,8 +390,15 @@ class MetricsResponse(BaseModel):
     replies_window: int
     sent_window: int
     suppression_entries: int
+    #: New suppression entries in the window per reason (reputation signal).
+    suppressions_window: dict[str, int] = {}
+    #: Rubric reviews in the window per decision (edits-as-signal).
+    reviews_window: dict[str, int] = {}
     run_error_rate: float | None
     reply_rate: float | None
+    bounce_rate: float | None = None
+    complaint_rate: float | None = None
+    edit_rate: float | None = None
 
 
 class AlertItem(BaseModel):

src/relay/config.py

@@ -97,6 +97,11 @@ class Settings(BaseSettings):
     #: start=0 disables the ramp entirely.
     warmup_daily_start: int = Field(default=0, ge=0)
     warmup_daily_increment: int = Field(default=0, ge=0)
+    #: Reputation alert: fire when the 24h hard-bounce rate exceeds this
+    #: fraction — but only once at least min_sends went out (1/1 is not a
+    #: reputation signal, it is noise).
+    alert_bounce_rate: float = Field(default=0.05, ge=0)
+    alert_bounce_rate_min_sends: int = Field(default=5, ge=1)
     # SNS event ingestion (webhook token and/or SQS polling).
     ses_webhook_token: SecretStr | None = None
     sqs_queue_url: str = ""
@@ -181,6 +186,11 @@ class Settings(BaseSettings):
     # ── Tenancy primitives ──────────────────────────────────────────────────
     # Dev default only; production uses a KMS-managed key (Phase 3).
     master_key: SecretStr = SecretStr("dev-master-key-not-for-production")
+    #: Rotation seam: during a master-key rotation, set the OLD key here so
+    #: signatures minted with it (unsubscribe tokens already sitting in
+    #: delivered mail) keep verifying. New signatures always use master_key.
+    #: Clear it once the rotation window closes.
+    master_key_previous: SecretStr | None = None
 
     def pilot_recipient_addresses(self) -> tuple[str, ...]:
         """The parsed pilot allowlist (comma-separated, trimmed, no blanks)."""