Phase 2 — Reliability, observability & evaluation: crash recovery, backpressure, metrics, evals, adversarial suite

.env.example

@@ -74,3 +74,23 @@ 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
+
+# ── Rate limiting & retries (Phase 2) ───────────────────────────────────────
+# Requests/second per external target; 0 disables the bucket. Waits beyond
+# max_wait raise Backpressure and the work parks (error_retryable).
+RELAY_RATE_LIMIT_LOCAL_RPS=0
+RELAY_RATE_LIMIT_HOSTED_RPS=0
+RELAY_RATE_LIMIT_CRM_RPS=0
+RELAY_RATE_LIMIT_MAX_WAIT_SECONDS=30
+# Bounded retry for transient compute failures only.
+RELAY_COMPUTE_RETRY_ATTEMPTS=2
+RELAY_COMPUTE_RETRY_BASE_SECONDS=0.5
+
+# ── Crash recovery (Phase 2) ────────────────────────────────────────────────
+RELAY_RECOVERY_STALE_AFTER_SECONDS=300
+
+# ── Alerting (Phase 2) ──────────────────────────────────────────────────────
+RELAY_ALERT_SPEND_UNITS_PER_HOUR=100
+RELAY_ALERT_FAILURE_STREAK=3
+RELAY_ALERT_QUEUE_STALE_SECONDS=600
+RELAY_ALERT_WEBHOOK_URL=                     # Slack/n8n hook; empty = log only

README.md

@@ -10,8 +10,9 @@ send is *structurally impossible* — enforced by the database, not by
 good intentions.
 
 This repository currently implements **Phase 0 — Foundations &
-Scaffolding**, **Phase 1A — Synthetic dry-run MVP**, and **Phase 1B —
-Real-data, no-send pilot** of the
+Scaffolding**, **Phase 1A — Synthetic dry-run MVP**, **Phase 1B —
+Real-data, no-send pilot**, and **Phase 2 — Reliability, observability
+& evaluation** of the
 [development roadmap](RELAY-development-roadmap.md) (on the `Plan`
 branch, together with the full
 [project documentation](RELAY-project-documentation.md)): the pipeline
@@ -89,6 +90,24 @@ system only records and enforces.
 
 ---
 
+## What Phase 2 adds
+
+Make the proven pipeline trustworthy unattended, and measurable enough
+to change safely.
+
+| Capability | Where |
+| --- | --- |
+| **Resumability**: transient compute failures park leads in `error_retryable` (the failed step's transaction already rolled back — no partial work) and a later run resumes them; the DB trigger counts retries and enforces the cap; refusals park terminally for a human | `pipeline/runner.py` |
+| **Crash recovery on every tick**: orphaned runs get closed, orphaned mid-send jobs fail safe (outcome unknown ⇒ never retried, never assumed sent); idempotent, wired into the worker tick | `pipeline/recovery.py` |
+| **Rate limiting & backpressure**: token buckets per external target (compute tiers, CRM); waits beyond the cap raise visibly and park work; bounded exponential retry for *transient* failures only — refusals are never re-rolled and providers are never silently swapped | `ratelimit.py` |
+| **Observability**: `/metrics` (JSON) + `/metrics/prometheus`, all derived on read from rows the pipeline already writes; `/ops` self-contained dashboard; alert rules (spend spike, failure streak, stuck queue) with log + webhook sinks | `observability/`, `api/ops_ui.py` |
+| **Eval harness**: golden-set invariants through the real prompt scaffolding — opt-outs must triage `unsubscribed`, injections cannot raise scores or manufacture intent, copy respects bounds; `just evals` scores the *configured* backends; an injected regression is provably caught | `evals/`, `scripts/run_evals.py` |
+| **Adversarial suite**: duplicate-send chaos (racing workers), outbox atomicity, suppression bypass from every angle, webhook replays, CRM conflicts, cross-tenant erasure attempts, PII log sweep, and a pg_dump→restore test proving erasure survives backups | `tests/test_adversarial.py` |
+| **Unattended-run proof**: a simulated spine schedule converges a mixed cohort and further ticks change nothing | `tests/test_unattended.py` |
+| **§8 open item resolved**: the local tier stays structurally tool-free, with revisit criteria recorded | `docs/decisions/local-tool-calling.md` |
+
+---
+
 ## Exit gate — every item is a passing test
 
 Run them: `just test-exit-gate` (or `just test` for the full suite).

docs/decisions/local-tool-calling.md

@@ -0,0 +1,51 @@
+# Decision record: local-tier tool calling (§8 open item)
+
+**Status:** RESOLVED for Phase 2 — the local tier remains tool-free.
+**Date:** Phase 2. **Revisit:** Phase 3, against the criteria below.
+
+## The question
+
+§8 left one routing rule provisional: may the local (cheap, bounded)
+tier be trusted to call tools, or must every tool-calling task route to
+the hosted tier?
+
+## Decision
+
+The local tier stays **structurally tool-free**:
+
+1. `route(requires_tools=True)` forces the hosted tier regardless of the
+   task's default route (`src/relay/routing/router.py`) — unchanged
+   since Phase 0 and now pinned by test.
+2. The backends that can serve the local tier have **no tool support by
+   construction** — `OpenAICompatBackend` and `GoogleGeminiBackend`
+   never send tool schemas; a confused or compromised local model can
+   produce at most one JSON blob that downstream gates re-validate.
+
+So this is not a policy a prompt can argue with; there is no code path
+in which a local model receives or invokes a tool.
+
+## Rationale
+
+- The blast radius of a wrong tool call (writes, sends, external calls)
+  is categorically larger than a wrong JSON answer, and the local tier
+  exists precisely for tasks where being wrong is cheap.
+- Small-model tool-call reliability varies wildly by model and schema;
+  we have no validation evidence, and "seems fine" is not evidence.
+- Nothing in Phases 0–2 needs local tools: every local task is
+  text-in/JSON-out by design.
+
+## Revisit criteria (Phase 3)
+
+Open this only when all three hold:
+
+1. A concrete task exists that needs tools AND is too high-volume for
+   the hosted tier's economics.
+2. A validation harness (extending `relay.evals`) measures tool-call
+   schema compliance and argument correctness for the candidate local
+   model over ≥200 golden cases, with a pass bar ≥99% for schema
+   compliance and an explicit review of failure modes.
+3. The tools exposed are read-only or idempotent, and every effect they
+   could cause remains behind the existing gates (suppression,
+   eligibility, human approval).
+
+Until then, `requires_tools=True` → hosted, everywhere, structurally.

justfile

@@ -89,6 +89,10 @@ worker:
 retention:
     uv run relay-retention
 
+# Run reasoning evals against the configured backends (spends real quota)
+evals which="both":
+    uv run python scripts/run_evals.py {{which}}
+
 # Walk a synthetic lead through the entire state machine and print the trace
 demo:
     uv run python scripts/demo_journey.py

scripts/run_evals.py

@@ -0,0 +1,52 @@
+"""Run the reasoning evals against the CONFIGURED backends.
+
+    just evals             # both tiers, as configured in .env
+    uv run python scripts/run_evals.py local
+    uv run python scripts/run_evals.py hosted
+
+Exit code is non-zero if any evaluated backend fails its threshold —
+wire this into CI or a pre-deploy check when models/prompts change.
+Note: this calls the real configured providers and spends real quota.
+"""
+
+from __future__ import annotations
+
+import sys
+
+from relay.compute.registry import backend_for
+from relay.evals import run_evals
+from relay.logs import setup_logging
+from relay.routing.router import ComputeTier
+
+
+def main() -> None:
+    setup_logging()
+    which = sys.argv[1] if len(sys.argv) > 1 else "both"
+    tiers = {
+        "local": [ComputeTier.LOCAL],
+        "hosted": [ComputeTier.HOSTED],
+        "both": [ComputeTier.LOCAL, ComputeTier.HOSTED],
+    }[which]
+
+    failed = False
+    for tier in tiers:
+        backend = backend_for(tier)
+        report = run_evals(backend)
+        print(f"\n─── {tier} tier: {report.backend}:{report.model} ───")
+        for category, (passed, total) in sorted(report.by_category.items()):
+            marker = "✓" if passed == total else "✗"
+            print(f"  {marker} {category:16s} {passed}/{total}")
+        verdict = "PASS" if report.passed else "FAIL"
+        print(
+            f"  → {verdict} (pass rate {report.pass_rate:.0%}, "
+            f"threshold {report.threshold:.0%})"
+        )
+        for r in report.failures:
+            print(f"    ✗ {r.case_id}: {r.detail}")
+        failed = failed or not report.passed
+
+    sys.exit(1 if failed else 0)
+
+
+if __name__ == "__main__":
+    main()

src/relay/api/ops_ui.py

@@ -0,0 +1,114 @@
+"""The minimal ops dashboard — one self-contained HTML page, like /review.
+
+Reads /metrics and /alerts with the tenant key entered in the page.
+No build step, no external assets; deliberately a glanceable status
+board, not a BI tool (Grafana over /metrics/prometheus is the Phase 3
+answer for real dashboards).
+"""
+
+from __future__ import annotations
+
+OPS_PAGE = """<!doctype html>
+<html lang="en">
+<head>
+<meta charset="utf-8">
+<meta name="viewport" content="width=device-width, initial-scale=1">
+<title>RELAY — Ops</title>
+<style>
+  :root { font-family: system-ui, sans-serif; color: #1a1a1a; }
+  body { margin: 0; padding: 1.5rem 2rem; max-width: 1100px; }
+  h1 { font-size: 1.15rem; }
+  .bar { display: flex; gap: .6rem; margin-bottom: 1.2rem; }
+  input { font: inherit; padding: .45rem; flex: 1; }
+  button { font: inherit; padding: .45rem 1rem; cursor: pointer;
+           border: 1px solid #999; border-radius: 6px; background: #fff; }
+  .grid { display: grid; grid-template-columns: repeat(auto-fill,
+          minmax(240px, 1fr)); gap: 1rem; }
+  .card { border: 1px solid #ddd; border-radius: 8px; padding: .9rem 1.1rem; }
+  .card h2 { font-size: .8rem; text-transform: uppercase; color: #666;
+             margin: 0 0 .5rem; letter-spacing: .04em; }
+  .big { font-size: 1.6rem; font-weight: 650; }
+  table { width: 100%; border-collapse: collapse; font-size: .85rem; }
+  td { padding: .15rem 0; } td:last-child { text-align: right;
+       font-variant-numeric: tabular-nums; }
+  .alert { border-left: 4px solid #c62828; background: #fdecea;
+           padding: .6rem .9rem; border-radius: 4px; margin-bottom: .5rem; }
+  .alert.warning { border-color: #b26a00; background: #fff8e1; }
+  .ok { color: #2e7d32; }
+  #updated { color: #777; font-size: .8rem; margin-left: auto;
+             align-self: center; }
+</style>
+</head>
+<body>
+<h1>RELAY ops</h1>
+<div class="bar">
+  <input type="password" id="key" placeholder="Tenant API key (rk_…)">
+  <button onclick="refresh()">Refresh</button>
+  <label style="align-self:center;font-size:.85rem">
+    <input type="checkbox" id="auto" onchange="autoRefresh()"> auto (30s)
+  </label>
+  <span id="updated"></span>
+</div>
+<div id="alerts"></div>
+<div class="grid" id="cards"></div>
+<script>
+const $ = (id) => document.getElementById(id);
+$("key").value = sessionStorage.getItem("relay_key") || "";
+let timer = null;
+
+function esc(s) {
+  return String(s).replace(/[&<>"']/g,
+    (c) => ({"&":"&amp;","<":"&lt;",">":"&gt;",'"':"&quot;","'":"&#39;"})[c]);
+}
+
+function table(obj) {
+  const rows = Object.entries(obj).sort()
+    .map(([k, v]) => `<tr><td>${esc(k)}</td><td>${v}</td></tr>`).join("");
+  return rows ? `<table>${rows}</table>` : "<em>none</em>";
+}
+
+async function refresh() {
+  const headers = { "X-API-Key": $("key").value.trim() };
+  sessionStorage.setItem("relay_key", $("key").value.trim());
+  const [mRes, aRes] = await Promise.all([
+    fetch("/metrics", { headers }), fetch("/alerts", { headers }),
+  ]);
+  if (!mRes.ok) { $("cards").innerHTML = `Error ${mRes.status}`; return; }
+  const m = await mRes.json();
+  const a = aRes.ok ? await aRes.json() : { alerts: [] };
+
+  $("alerts").innerHTML = a.alerts.length
+    ? a.alerts.map(x => `<div class="alert ${x.severity}">
+        <b>${esc(x.rule)}</b> — ${esc(x.detail)}</div>`).join("")
+    : `<div class="ok">No active alerts.</div>`;
+
+  const errRate = m.run_error_rate == null ? "–"
+    : (m.run_error_rate * 100).toFixed(0) + "%";
+  const replyRate = m.reply_rate == null ? "–"
+    : (m.reply_rate * 100).toFixed(0) + "%";
+
+  $("cards").innerHTML = `
+    <div class="card"><h2>Cost (24h)</h2>
+      <div class="big">${m.cost_units_window.toFixed(1)}</div> units</div>
+    <div class="card"><h2>Run error rate (24h)</h2>
+      <div class="big">${errRate}</div></div>
+    <div class="card"><h2>Sent / replies (24h)</h2>
+      <div class="big">${m.sent_window} / ${m.replies_window}</div>
+      reply rate ${replyRate}</div>
+    <div class="card"><h2>Suppression entries</h2>
+      <div class="big">${m.suppression_entries}</div></div>
+    <div class="card"><h2>Leads by state</h2>${table(m.lead_states)}</div>
+    <div class="card"><h2>Runs (24h)</h2>${table(m.runs_window)}</div>
+    <div class="card"><h2>Send jobs</h2>${table(m.send_jobs)}</div>`;
+  $("updated").textContent =
+    "updated " + new Date().toLocaleTimeString();
+}
+
+function autoRefresh() {
+  if (timer) { clearInterval(timer); timer = null; }
+  if ($("auto").checked) timer = setInterval(refresh, 30000);
+}
+</script>
+</body>
+</html>
+"""

src/relay/api/routes.py

@@ -42,6 +42,7 @@
 from relay.guardrails.harness import GuardrailViolation
 from relay.hashing import email_domain, hash_api_key, hash_email
 from relay.logs import get_logger
+from relay.observability import evaluate_alerts, prometheus_text, tenant_metrics
 from relay.pipeline.runner import PipelineRunner
 from relay.workers.send_worker import process_pending
 
@@ -502,6 +503,61 @@ def economics(
     )
 
 
+# ── Observability (Phase 2) ─────────────────────────────────────────────────
+
+
+@router.get("/metrics", response_model=schemas.MetricsResponse)
+def metrics_json(
+    tenant_id: uuid.UUID = Depends(require_tenant),
+) -> schemas.MetricsResponse:
+    m = tenant_metrics(tenant_id)
+    return schemas.MetricsResponse(
+        tenant_id=m.tenant_id,
+        generated_at=m.generated_at,
+        lead_states=m.lead_states,
+        runs_window=m.runs,
+        cost_units_window=m.cost_units_window,
+        send_jobs=m.send_jobs,
+        replies_window=m.replies_window,
+        sent_window=m.sent_window,
+        suppression_entries=m.suppression_entries,
+        run_error_rate=m.run_error_rate,
+        reply_rate=m.reply_rate,
+    )
+
+
+@router.get("/metrics/prometheus", include_in_schema=False)
+def metrics_prometheus(
+    tenant_id: uuid.UUID = Depends(require_tenant),
+):
+    from fastapi.responses import PlainTextResponse
+
+    return PlainTextResponse(prometheus_text(tenant_metrics(tenant_id)))
+
+
+@router.get("/alerts", response_model=schemas.AlertsResponse)
+def alerts(
+    tenant_id: uuid.UUID = Depends(require_tenant),
+) -> schemas.AlertsResponse:
+    fired = evaluate_alerts(tenant_id)
+    return schemas.AlertsResponse(
+        tenant_id=tenant_id,
+        alerts=[
+            schemas.AlertItem(
+                rule=a.rule, severity=a.severity, detail=a.detail, value=a.value
+            )
+            for a in fired
+        ],
+    )
+
+
+@router.get("/ops", include_in_schema=False)
+def ops_page() -> HTMLResponse:
+    from relay.api.ops_ui import OPS_PAGE
+
+    return HTMLResponse(OPS_PAGE)
+
+
 # ── DSR erasure (tenant-scoped): the right to be forgotten ──────────────────
 
 

src/relay/api/schemas.py

@@ -326,3 +326,32 @@ class ErasureResponse(BaseModel):
     crm: dict[str, str]
     vector_store: str
     suppression_added: bool
+
+
+# ── Observability (Phase 2) ─────────────────────────────────────────────────
+
+
+class MetricsResponse(BaseModel):
+    tenant_id: uuid.UUID
+    generated_at: datetime
+    lead_states: dict[str, int]
+    runs_window: dict[str, int]
+    cost_units_window: float
+    send_jobs: dict[str, int]
+    replies_window: int
+    sent_window: int
+    suppression_entries: int
+    run_error_rate: float | None
+    reply_rate: float | None
+
+
+class AlertItem(BaseModel):
+    rule: str
+    severity: str
+    detail: str
+    value: float
+
+
+class AlertsResponse(BaseModel):
+    tenant_id: uuid.UUID
+    alerts: list[AlertItem]