From 7c10529b6f655d78098b7f9fb0b2a97eaad7fdbd Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 5 Jul 2026 07:35:14 +0000 Subject: [PATCH] Phase 3: human-in-the-loop at scale, secrets rotation, reputation monitoring Completes the code-buildable remainder of Phase 3 (production readiness); docs/phase3-readiness.md records what code provides vs the operator/legal deliverables and the three parked decisions. Human-in-the-loop at scale: - The review queue is confidence-ordered (highest fit_score first, FIFO within a score) and carries fit_score, so reviewers batch the confident tail and spend attention on the uncertain bottom. - POST /outreach-drafts/batch-review: up to 100 rubric decisions per call through the same domain review path, each item in its own transaction - one stale/bogus item fails alone, the rest land. Never sends, like every review surface. - Edit rate (approved_with_edits share of window reviews) is a first-class metric: edits-as-signal for prompt iteration. Secrets rotation: - POST /internal/tenants/{id}/rotate-key (admin token): issues a new tenant API key, old key dies at commit, rotation audited, key returned exactly once. - RELAY_MASTER_KEY_PREVIOUS: verify-only rotation window for the master key, so signed unsubscribe tokens already sitting in delivered mail keep working across a rotation - a dead unsubscribe link is a compliance failure. New tokens always sign with the current key. Reputation monitoring: - 24h suppressions-by-reason and reviews-by-decision in tenant metrics; bounce_rate / complaint_rate / edit_rate properties wired through /metrics and the Prometheus export. - bounce_rate_high critical alert (RELAY_ALERT_BOUNCE_RATE, default 5%) with a min-sends floor (RELAY_ALERT_BOUNCE_RATE_MIN_SENDS) so a 1-of-1 bounce never pages - it fires BEFORE the eligibility threshold silently pauses sending. Tests (+7, suite now 299): batch review item isolation and terminal states, confidence ordering, key rotation kills the old key instantly (and 422/403/404 edges), master-key rotation keeps old tokens alive, reputation + edit-signal metrics end to end, bounce alert honors the min-sends floor then fires critical. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01ASAVj4XgJCH3UcHkZZaYzM --- .env.example | 7 + README.md | 7 + docs/phase3-readiness.md | 55 +++++++ src/relay/api/routes.py | 120 +++++++++++++- src/relay/api/schemas.py | 55 +++++++ src/relay/config.py | 10 ++ src/relay/ingest/unsubscribe.py | 37 +++-- src/relay/observability/alerts.py | 33 +++- src/relay/observability/metrics.py | 62 +++++++- tests/test_phase3_scale.py | 246 +++++++++++++++++++++++++++++ 10 files changed, 617 insertions(+), 15 deletions(-) create mode 100644 docs/phase3-readiness.md create mode 100644 tests/test_phase3_scale.py diff --git a/.env.example b/.env.example index a3f720b..6e33e2e 100644 --- a/.env.example +++ b/.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= diff --git a/README.md b/README.md index 71891db..9752d0b 100644 --- a/README.md +++ b/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). --- diff --git a/docs/phase3-readiness.md b/docs/phase3-readiness.md new file mode 100644 index 0000000..a9f30c2 --- /dev/null +++ b/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. diff --git a/src/relay/api/routes.py b/src/relay/api/routes.py index f0cfcdc..371678d 100644 --- a/src/relay/api/routes.py +++ b/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, ) diff --git a/src/relay/api/schemas.py b/src/relay/api/schemas.py index e99d43c..ac6d859 100644 --- a/src/relay/api/schemas.py +++ b/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,6 +271,10 @@ 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 @@ -271,6 +282,43 @@ 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): diff --git a/src/relay/config.py b/src/relay/config.py index 77afeaa..ac4c04a 100644 --- a/src/relay/config.py +++ b/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).""" diff --git a/src/relay/ingest/unsubscribe.py b/src/relay/ingest/unsubscribe.py index 289ddd4..dd9b470 100644 --- a/src/relay/ingest/unsubscribe.py +++ b/src/relay/ingest/unsubscribe.py @@ -47,22 +47,33 @@ class UnsubscribeRejected(Exception): """The token failed authentication or was structurally invalid.""" -def _sign(tenant_id: uuid.UUID, lead_id: uuid.UUID, job_id: uuid.UUID) -> str: - key = derive_tenant_key( - get_settings().master_key.get_secret_value(), str(tenant_id), "unsubscribe" - ) +def _sign( + master_key: str, tenant_id: uuid.UUID, lead_id: uuid.UUID, job_id: uuid.UUID +) -> str: + key = derive_tenant_key(master_key, str(tenant_id), "unsubscribe") payload = f"{_VERSION}.{tenant_id.hex}.{lead_id.hex}.{job_id.hex}" return hmac.new(key, payload.encode("utf-8"), "sha256").hexdigest() def build_token(tenant_id: uuid.UUID, lead_id: uuid.UUID, job_id: uuid.UUID) -> str: - """Mint the signed token embedded in a send's List-Unsubscribe URL.""" - sig = _sign(tenant_id, lead_id, job_id) + """Mint the signed token embedded in a send's List-Unsubscribe URL. + + Always signs with the CURRENT master key — the previous key (if any) + is verify-only during a rotation window. + """ + key = get_settings().master_key.get_secret_value() + sig = _sign(key, tenant_id, lead_id, job_id) return f"{_VERSION}.{tenant_id.hex}.{lead_id.hex}.{job_id.hex}.{sig}" def verify_token(token: str) -> tuple[uuid.UUID, uuid.UUID, uuid.UUID]: - """Return (tenant_id, lead_id, job_id) for a valid token, else raise.""" + """Return (tenant_id, lead_id, job_id) for a valid token, else raise. + + Verification accepts the current master key and — during a rotation + window — RELAY_MASTER_KEY_PREVIOUS, so unsubscribe links already + sitting in delivered mail keep working across a rotation. An + unsubscribe link that silently dies IS a compliance failure. + """ parts = token.split(".") if len(parts) != 5 or parts[0] != _VERSION: raise UnsubscribeRejected("malformed unsubscribe token") @@ -72,10 +83,14 @@ def verify_token(token: str) -> tuple[uuid.UUID, uuid.UUID, uuid.UUID]: job_id = uuid.UUID(hex=parts[3]) except ValueError as exc: raise UnsubscribeRejected("malformed unsubscribe token") from exc - expected = _sign(tenant_id, lead_id, job_id) - if not hmac.compare_digest(parts[4], expected): - raise UnsubscribeRejected("bad unsubscribe token signature") - return tenant_id, lead_id, job_id + settings = get_settings() + keys = [settings.master_key.get_secret_value()] + if settings.master_key_previous is not None: + keys.append(settings.master_key_previous.get_secret_value()) + for master_key in keys: + if hmac.compare_digest(parts[4], _sign(master_key, tenant_id, lead_id, job_id)): + return tenant_id, lead_id, job_id + raise UnsubscribeRejected("bad unsubscribe token signature") def process_unsubscribe(token: str) -> bool: diff --git a/src/relay/observability/alerts.py b/src/relay/observability/alerts.py index 544d1a9..f51c9ac 100644 --- a/src/relay/observability/alerts.py +++ b/src/relay/observability/alerts.py @@ -19,7 +19,7 @@ from relay.config import get_settings from relay.db.engine import tenant_session -from relay.db.models import PipelineRun, SendJob +from relay.db.models import PipelineRun, SendJob, Suppression from relay.logs import get_logger log = get_logger(__name__) @@ -102,6 +102,37 @@ def evaluate_alerts(tenant_id: uuid.UUID) -> list[Alert]: ) ) + # ── Reputation: hard-bounce rate over the last 24h ────────────────── + # Providers cut senders off for exactly this number; it must get + # loud BEFORE the eligibility threshold silently pauses sending. + day_cutoff = now - timedelta(hours=24) + sent_24h = session.execute( + select(func.count()).where( + SendJob.status == "sent", SendJob.completed_at >= day_cutoff + ) + ).scalar_one() + if sent_24h >= settings.alert_bounce_rate_min_sends: + bounces_24h = session.execute( + select(func.count()).where( + Suppression.reason == "hard_bounce", + Suppression.created_at >= day_cutoff, + ) + ).scalar_one() + rate = bounces_24h / sent_24h + if rate > settings.alert_bounce_rate: + alerts.append( + Alert( + rule="bounce_rate_high", + severity="critical", + detail=( + f"{bounces_24h} hard bounces over {sent_24h} sends " + f"in 24h ({rate:.1%}, threshold " + f"{settings.alert_bounce_rate:.1%})" + ), + value=rate, + ) + ) + for alert in alerts: log.warning( "ALERT", diff --git a/src/relay/observability/metrics.py b/src/relay/observability/metrics.py index d74cd97..a83afa1 100644 --- a/src/relay/observability/metrics.py +++ b/src/relay/observability/metrics.py @@ -9,7 +9,14 @@ from sqlalchemy import func, select from relay.db.engine import tenant_session -from relay.db.models import Lead, PipelineRun, Reply, SendJob, Suppression +from relay.db.models import ( + DraftReview, + Lead, + PipelineRun, + Reply, + SendJob, + Suppression, +) #: The rolling window for rate-style metrics. WINDOW = timedelta(hours=24) @@ -30,6 +37,11 @@ class TenantMetrics: replies_window: int = 0 sent_window: int = 0 suppression_entries: int = 0 + #: New suppression entries in the window, per reason — the reputation + #: signal (hard_bounce / complaint / unsubscribe). + suppressions_window: dict[str, int] = field(default_factory=dict) + #: Rubric reviews in the window, per decision — edits-as-signal. + reviews_window: dict[str, int] = field(default_factory=dict) @property def run_error_rate(self) -> float | None: @@ -49,6 +61,28 @@ def reply_rate(self) -> float | None: return None return self.replies_window / self.sent_window + @property + def bounce_rate(self) -> float | None: + """Hard bounces per send in the window (reputation).""" + if not self.sent_window: + return None + return self.suppressions_window.get("hard_bounce", 0) / self.sent_window + + @property + def complaint_rate(self) -> float | None: + if not self.sent_window: + return None + return self.suppressions_window.get("complaint", 0) / self.sent_window + + @property + def edit_rate(self) -> float | None: + """Share of window reviews where the human had to edit — the + edits-as-signal number that steers prompt iteration.""" + total = sum(self.reviews_window.values()) + if not total: + return None + return self.reviews_window.get("approved_with_edits", 0) / total + def tenant_metrics(tenant_id: uuid.UUID) -> TenantMetrics: cutoff = datetime.now(tz=UTC) - WINDOW @@ -86,6 +120,20 @@ def tenant_metrics(tenant_id: uuid.UUID) -> TenantMetrics: suppression = session.execute( select(func.count()).select_from(Suppression) ).scalar_one() + suppressions_window = dict( + session.execute( + select(Suppression.reason, func.count()) + .where(Suppression.created_at >= cutoff) + .group_by(Suppression.reason) + ).all() + ) + reviews_window = dict( + session.execute( + select(DraftReview.decision, func.count()) + .where(DraftReview.created_at >= cutoff) + .group_by(DraftReview.decision) + ).all() + ) return TenantMetrics( tenant_id=tenant_id, @@ -97,6 +145,8 @@ def tenant_metrics(tenant_id: uuid.UUID) -> TenantMetrics: replies_window=replies_window, sent_window=sent_window, suppression_entries=suppression, + suppressions_window=suppressions_window, + reviews_window=reviews_window, ) @@ -129,5 +179,15 @@ def prometheus_text(metrics: TenantMetrics) -> str: f'relay_sent_window{{tenant="{t}"}} {metrics.sent_window}', "# TYPE relay_suppression_entries gauge", f'relay_suppression_entries{{tenant="{t}"}} {metrics.suppression_entries}', + "# TYPE relay_suppressions_window counter", + *( + f'relay_suppressions_window{{tenant="{t}",reason="{r}"}} {n}' + for r, n in sorted(metrics.suppressions_window.items()) + ), + "# TYPE relay_reviews_window counter", + *( + f'relay_reviews_window{{tenant="{t}",decision="{d}"}} {n}' + for d, n in sorted(metrics.reviews_window.items()) + ), ] return "\n".join(lines) + "\n" diff --git a/tests/test_phase3_scale.py b/tests/test_phase3_scale.py new file mode 100644 index 0000000..9f67792 --- /dev/null +++ b/tests/test_phase3_scale.py @@ -0,0 +1,246 @@ +"""Phase 3 production readiness: human-in-the-loop at scale, secrets +rotation, and reputation monitoring.""" + +from __future__ import annotations + +import uuid + +import pytest +from sqlalchemy import select + +from relay.config import get_settings +from relay.db.engine import tenant_session +from relay.db.models import Lead +from relay.domain.suppression import add_suppression +from relay.ingest.unsubscribe import ( + UnsubscribeRejected, + build_token, + verify_token, +) +from relay.observability import evaluate_alerts, tenant_metrics +from tests.conftest import ADMIN, run_to_approval, walk_to_sent + +pytestmark = pytest.mark.exit_gate + + +# ── Batched review: human-in-the-loop at scale ────────────────────────────── + + +def _pending_draft_id(tenant_id, lead_id) -> uuid.UUID: + from relay.db.models import OutreachDraft + + with tenant_session(tenant_id) as session: + return session.execute( + select(OutreachDraft.id).where( + OutreachDraft.lead_id == lead_id, + OutreachDraft.status == "pending_approval", + ) + ).scalar_one() + + +def test_batch_review_processes_items_independently(client, tenant_a, factory_a): + """One batch call: an approval, a rejection, and a bogus draft id. + The bad item fails alone; the good ones land; nothing sends.""" + tenant_id, api_key = tenant_a + leads = [factory_a.lead() for _ in range(2)] + for lead_id in leads: + run_to_approval(tenant_id, lead_id) + approve_id = _pending_draft_id(tenant_id, leads[0]) + reject_id = _pending_draft_id(tenant_id, leads[1]) + + response = client.post( + "/outreach-drafts/batch-review", + headers={"X-API-Key": api_key}, + json={ + "reviewer": "batch-reviewer", + "items": [ + {"draft_id": str(approve_id), "decision": "approved"}, + { + "draft_id": str(reject_id), + "decision": "rejected", + "reasons": ["tone"], + }, + {"draft_id": str(uuid.uuid4()), "decision": "approved"}, + ], + }, + ) + assert response.status_code == 200, response.text + body = response.json() + assert body["sent"] is False + assert (body["approved"], body["rejected"], body["failed"]) == (1, 1, 1) + ok_by_id = {r["draft_id"]: r["ok"] for r in body["results"]} + assert ok_by_id[str(approve_id)] and ok_by_id[str(reject_id)] + + with tenant_session(tenant_id) as session: + states = {str(lead_id): session.get(Lead, lead_id).state for lead_id in leads} + assert states[str(leads[0])] == "approved" + assert states[str(leads[1])] == "rejected_by_human" + + +def test_review_queue_is_confidence_ordered(client, tenant_a, factory_a): + """The queue surfaces the highest-confidence drafts first and carries + fit_score so a reviewer can split batch-tail from careful-review.""" + tenant_id, api_key = tenant_a + leads = [factory_a.lead() for _ in range(3)] + for lead_id in leads: + run_to_approval(tenant_id, lead_id) + scores = {leads[0]: 0.31, leads[1]: 0.97, leads[2]: 0.55} + with tenant_session(tenant_id) as session: + for lead_id, score in scores.items(): + session.get(Lead, lead_id).fit_score = score + + response = client.get("/outreach-drafts/pending", headers={"X-API-Key": api_key}) + assert response.status_code == 200 + listed = [ + (d["lead_id"], d["fit_score"]) + for d in response.json()["drafts"] + if uuid.UUID(d["lead_id"]) in scores + ] + assert [s for _, s in listed] == sorted((s for s in scores.values()), reverse=True) + + +# ── Secrets rotation ──────────────────────────────────────────────────────── + + +def test_tenant_api_key_rotation_invalidates_old_key(client, api_tenant): + old_key = api_tenant["api_key"] + assert ( + client.get( + "/outreach-drafts/pending", headers={"X-API-Key": old_key} + ).status_code + == 200 + ) + + response = client.post( + f"/internal/tenants/{api_tenant['id']}/rotate-key", headers=ADMIN + ) + assert response.status_code == 200, response.text + new_key = response.json()["api_key"] + assert new_key != old_key + + # The old key dies immediately; the new one works. + assert ( + client.get( + "/outreach-drafts/pending", headers={"X-API-Key": old_key} + ).status_code + == 401 + ) + assert ( + client.get( + "/outreach-drafts/pending", headers={"X-API-Key": new_key} + ).status_code + == 200 + ) + + +def test_rotate_key_requires_admin_and_known_tenant(client, api_tenant): + url = f"/internal/tenants/{api_tenant['id']}/rotate-key" + assert client.post(url).status_code == 422 # header missing entirely + assert client.post(url, headers={"X-Admin-Token": "wrong-token"}).status_code == 403 + assert ( + client.post( + f"/internal/tenants/{uuid.uuid4()}/rotate-key", headers=ADMIN + ).status_code + == 404 + ) + + +def test_master_key_rotation_keeps_old_unsubscribe_tokens_alive(tenant_a, monkeypatch): + """Unsubscribe links already sitting in delivered mail MUST keep + working across a master-key rotation (a dead unsubscribe link is a + compliance failure) — via RELAY_MASTER_KEY_PREVIOUS, verify-only.""" + tenant_id, _ = tenant_a + lead_id, job_id = uuid.uuid4(), uuid.uuid4() + old_token = build_token(tenant_id, lead_id, job_id) + + monkeypatch.setenv("RELAY_MASTER_KEY", "rotated-master-key") + get_settings.cache_clear() + # Without the previous key configured, the old link would die… + with pytest.raises(UnsubscribeRejected): + verify_token(old_token) + + monkeypatch.setenv("RELAY_MASTER_KEY_PREVIOUS", "dev-master-key-not-for-production") + get_settings.cache_clear() + # …with it, the old link verifies AND new tokens use the new key. + assert verify_token(old_token) == (tenant_id, lead_id, job_id) + new_token = build_token(tenant_id, lead_id, job_id) + assert new_token != old_token + assert verify_token(new_token) == (tenant_id, lead_id, job_id) + + monkeypatch.delenv("RELAY_MASTER_KEY_PREVIOUS") + monkeypatch.delenv("RELAY_MASTER_KEY") + get_settings.cache_clear() + + +# ── Reputation monitoring ─────────────────────────────────────────────────── + + +def test_metrics_expose_reputation_and_edit_signal(client, tenant_a, factory_a): + tenant_id, api_key = tenant_a + sent_lead = factory_a.lead() + walk_to_sent(tenant_id, sent_lead) + with tenant_session(tenant_id) as session: + add_suppression( + session, + tenant_id=tenant_id, + reason="hard_bounce", + source="provider_webhook", + created_by="test", + email=f"dead-{uuid.uuid4().hex[:6]}@example.test", + ) + + # An edits-as-signal data point: review a pending draft with edits. + edit_lead = factory_a.lead() + run_to_approval(tenant_id, edit_lead) + draft_id = _pending_draft_id(tenant_id, edit_lead) + response = client.post( + f"/outreach-drafts/{draft_id}/review", + headers={"X-API-Key": api_key}, + json={ + "reviewer": "editor", + "decision": "approved_with_edits", + "reasons": ["tone"], + "edited_body": "Hand-tuned body.", + }, + ) + assert response.status_code == 200, response.text + + m = tenant_metrics(tenant_id) + assert m.suppressions_window.get("hard_bounce", 0) >= 1 + assert m.bounce_rate is not None and m.bounce_rate > 0 + assert m.reviews_window.get("approved_with_edits", 0) >= 1 + assert m.edit_rate is not None and m.edit_rate > 0 + + body = client.get("/metrics", headers={"X-API-Key": api_key}).json() + assert body["bounce_rate"] == pytest.approx(m.bounce_rate) + assert body["edit_rate"] == pytest.approx(m.edit_rate) + prom = client.get("/metrics/prometheus", headers={"X-API-Key": api_key}).text + assert "relay_suppressions_window{" in prom + assert "relay_reviews_window{" in prom + + +def test_bounce_rate_alert_fires_past_threshold(tenant_a, factory_a, monkeypatch): + """One bounce over one send is 100% — over any sane threshold — but + the rule stays quiet until min_sends is met, then fires.""" + tenant_id, _ = tenant_a + monkeypatch.setenv("RELAY_ALERT_BOUNCE_RATE", "0.05") + monkeypatch.setenv("RELAY_ALERT_BOUNCE_RATE_MIN_SENDS", "2") + get_settings.cache_clear() + + walk_to_sent(tenant_id, factory_a.lead()) + with tenant_session(tenant_id) as session: + add_suppression( + session, + tenant_id=tenant_id, + reason="hard_bounce", + source="provider_webhook", + created_by="test", + email=f"dead-{uuid.uuid4().hex[:6]}@example.test", + ) + # 1 send < min_sends: noise, not reputation — no alert. + assert not any(a.rule == "bounce_rate_high" for a in evaluate_alerts(tenant_id)) + + walk_to_sent(tenant_id, factory_a.lead()) + fired = [a for a in evaluate_alerts(tenant_id) if a.rule == "bounce_rate_high"] + assert len(fired) == 1 and fired[0].severity == "critical" + assert fired[0].value == pytest.approx(0.5)