Isolate admitted transactional callers and retain retry policy

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a119cf-56d0-7fa3-840b-eb8732e1db0b

docs/caller-admission-2026-10-08.md

@@ -0,0 +1,75 @@
+# Scoped transactional callers — source review, 2026-10-08
+
+The receiver can isolate explicitly admitted callers while retaining the existing
+single-caller startup path until deployment admission. No live credentials,
+mailboxes, sender, chart, image, ledger or timer changed in this source increment.
+
+Set EMAIL_CONNECT_CALLERS_FILE to an owner-controlled read-only JSON policy file.
+If present, invalid policy fails startup before ledger migration; there is no
+fallback to EMAIL_CONNECT_INGEST_TOKEN. Schema version1 has a callers list (1–32).
+Each entry needs caller_id, source, absolute token_file, bindings (exact
+[tenant, client] pairs), operations, templates, rate_per_minute, burst and
+max_payload_bytes. Credentials are independently delivered files; inline tokens,
+shared tokens and duplicate caller IDs are rejected. Tokens never enter policy
+receipts or repr output. Limits are positive integers: rate <=6000/minute,
+burst <=600 requests, payload <=128KiB. These are implementation ceilings,
+not approved runtime quotas.
+
+Operations: send, delivery-status, status, verify, consume, cancel, suppress,
+unsuppress. Send templates are only family_member.invited and
+family_invitation.resent. Payload source must match the authenticated policy;
+verification and invitation tenant/client pairs must be explicitly admitted.
+Each caller's invitation IDs, verification reservations and handles are isolated.
+An event's immutable payload cannot change on retry. Status exposes a caller-local
+opaque reference and transport evidence, never raw provider diagnostics.
+Confirmed temporary non-acceptance can retry; permanent rejection is terminal;
+uncertain acceptance remains terminal for automatic retry across restart.
+Historical failed invitations lacking retry evidence also remain terminal.
+
+A persisted atomic SQLite token bucket counts authorized requests, including
+status, validation failures and duplicates. Concurrent connections and restarts
+share its allowance. Changing caller identity resets its budget; admission must
+treat that as a quota/security change. Backward wall-clock movement grants no
+extra tokens. Oversized requests are rejected before reading their bodies.
+Public health only proves receiver liveness; authenticated status is separately
+admitted and consumes allowance.
+
+Suppression is global recipient protection. suppress/unsuppress are privileged
+platform operations, to admit only for a named operations owner, never generic
+application callers. They do not inspect mailboxes or automatically classify
+bounces. Sender identity/TLS remain governed by the existing SMTP provider;
+caller policy does not grant arbitrary subjects/bodies/From or a bulk-mail API.
+
+## Existing original caller and rollback
+
+Only caller_id/source user-engine may explicitly adopt legacy verification rows
+(adopt_legacy_verifications=true). Its granted binding pairs still constrain
+consume/cancel. legacy_invitation_binding may explicitly admit one binding for
+old invitation payloads lacking tenant/client, and preserves old event keys and
+accepted receipts. Reserved ec2_ input IDs cannot collide with scoped storage IDs.
+There is no inferred legacy adoption. Removing or narrowing a policy immediately
+denies its operations without deleting transport/idempotency evidence.
+
+Startup adds caller_id to verification rows, error/retryable to send_attempts,
+and caller_buckets. Capture fingerprints include all six tables; old five-table
+captures remain readable. Rehearsal source preserves scoped accepted/uncertain
+verification reservations and budgets without provider calls. Helpers remain
+pinned to the current admitted image: repin capture and rehearsal together to
+the candidate's immutable image before new-schema capture or scoped activation.
+Do not downgrade after new-schema rows exist without an admitted compatible
+recovery plan; old receiver code can lose retry restrictions or fail inserts.
+
+## Validation and outstanding admission
+
+Synthetic caller/legacy tests:66 passed, zero skips, network-denied app gate.
+Full owner suite including loopback GreenMail and scoped ledger rehearsal:
+148 passed, zero skips. Receiver/caller module and caller-test Ruff checks pass.
+No live recipient, identity, token, handle or message body is a fixture.
+
+EMAIL-WP-0008-T03 retains admission: choose actual caller/binding/template and
+operation sets, quotas, separate token custody/delivery, original-user-engine
+transition, platform sender/reply/bounce responsibilities, exact image/schema
+rollout and rollback, and independent ledger offsite recovery. Existing live
+EMAIL-WP-0007-T03/T04 and RPF-WP-0053-T05/T06 retain archive admission and atomic
+credential-pair custody. CSOC-WP-0010-T04/T05/T06 retain productive flow and
+owner-spanning browser acceptance. Source checks alone close none of these gates.

scripts/ledger_archive.py

@@ -72,7 +72,7 @@ def fingerprint(path):
                 "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
             )
         }
-        if tables != TABLES:
+        if tables not in (TABLES, TABLES | {"caller_buckets"}):
             raise Refused("unexpected_ledger_schema")
         schema = list(
             database.execute("SELECT type,name,tbl_name,sql FROM sqlite_master ORDER BY type,name")
@@ -82,7 +82,7 @@ def fingerprint(path):
             "user_version": database.execute("PRAGMA user_version").fetchone()[0],
             "tables": {},
         }
-        for name in sorted(TABLES):
+        for name in sorted(tables):
             rows = [
                 json.dumps(
                     row, separators=(",", ":"), default=lambda value: {"bytes_hex": value.hex()}

scripts/rehearse-ledger.py

@@ -41,10 +41,13 @@ class ForbiddenProvider:
 app=TransactionalApplication(store,ForbiddenProvider(),'synthetic-recovery-token','https://issuer.example.test')
 checks={'accepted_verifications':0,'unknown_verifications':0,'invalidated_handles':0,
         'suppressions':0,'accepted_deliveries':0,'unknown_sends':0}
-query=("SELECT v.registration_id,v.email,v.username,v.client_id,v.tenant,v.display_name,a.state "
+caller_column="v.caller_id" if "caller_id" in {r[1] for r in store.db.execute("PRAGMA table_info(verifications)")} else "''"
+query=("SELECT v.registration_id,v.email,v.username,v.client_id,v.tenant,v.display_name,a.state,"+caller_column+" "
        "FROM verifications v JOIN verification_attempts a ON a.request_id=v.request_id "
        "WHERE a.state IN ('accepted','in_flight')")
 for row in store.db.execute(query).fetchall():
+    app=TransactionalApplication(store,ForbiddenProvider(),'synthetic-recovery-token',
+        'https://issuer.example.test',**({'verification_namespace':row[7]} if row[7] else {}))
     payload=dict(zip(['registration_id','normalized_email','preferred_username','client_id','tenant','display_name'],row[:6]))
     payload['correlation_id']='synthetic-recovery-probe'
     raw=json.dumps(payload).encode();response={}

src/email_connect/callers.py

@@ -0,0 +1,368 @@
+"""Explicitly admitted caller policies for fixed transactional templates.
+
+No implicit shared-bearer fallback exists in this mode. The old receiver remains
+the default until the owning deployment selects an admitted policy file.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import io
+import json
+import math
+import re
+import secrets
+import time
+from dataclasses import dataclass, field
+from http import HTTPStatus
+from pathlib import Path
+
+IDENTIFIER = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}\Z")
+OPERATIONS = {
+    "/v1/send": "send",
+    "/v1/delivery-status": "delivery-status",
+    "/v1/status": "status",
+    "/v1/registration-verifications": "verify",
+    "/v1/registration-verifications/consume": "consume",
+    "/v1/registration-verifications/cancel": "cancel",
+    "/v1/suppressions": "suppress",
+    "/v1/suppressions/remove": "unsuppress",
+}
+TEMPLATES = frozenset({"family_member.invited", "family_invitation.resent"})
+ERRORS = frozenset(
+    {
+        "template_not_allowed",
+        "invalid_recipient",
+        "recipient_suppressed",
+        "invitation_id_required",
+        "idempotency_key_mismatch",
+        "idempotency_payload_mismatch",
+        "invalid_verification_request",
+        "verification_invalid",
+        "verification_expired",
+        "delivery_outcome_unknown",
+        "provider_timeout",
+        "provider_unavailable",
+        "temporary_deferral",
+        "permanent_rejection",
+        "invalid_size",
+    }
+)
+
+
+@dataclass(frozen=True)
+class CallerPolicy:
+    caller_id: str
+    token: str = field(repr=False)
+    source: str
+    bindings: frozenset[tuple[str, str]]
+    operations: frozenset[str]
+    templates: frozenset[str]
+    rate_per_minute: int
+    burst: int
+    max_payload_bytes: int
+    adopt_legacy_verifications: bool = False
+    legacy_invitation_binding: tuple[str, str] | None = None
+
+    def __post_init__(self):
+        if not isinstance(self.caller_id, str) or not IDENTIFIER.fullmatch(self.caller_id):
+            raise ValueError("invalid_caller_policy")
+        if not isinstance(self.source, str) or not IDENTIFIER.fullmatch(self.source):
+            raise ValueError("invalid_caller_policy")
+        if not isinstance(self.token, str) or not re.fullmatch(r"[!-~]{16,512}", self.token):
+            raise ValueError("invalid_caller_credential")
+        if (
+            not self.operations
+            or not self.operations <= set(OPERATIONS.values())
+            or not self.templates <= TEMPLATES
+            or not self.bindings
+        ):
+            raise ValueError("invalid_caller_policy")
+        for binding in self.bindings:
+            if len(binding) != 2 or any(
+                not isinstance(v, str) or not IDENTIFIER.fullmatch(v) for v in binding
+            ):
+                raise ValueError("invalid_caller_binding")
+        for value, maximum in (
+            (self.rate_per_minute, 6000),
+            (self.burst, 600),
+            (self.max_payload_bytes, 128 * 1024),
+        ):
+            if type(value) is not int or not 0 < value <= maximum:
+                raise ValueError("invalid_caller_limits")
+        if type(self.adopt_legacy_verifications) is not bool:
+            raise ValueError("invalid_legacy_admission")
+        if self.adopt_legacy_verifications or self.legacy_invitation_binding is not None:
+            if self.caller_id != "user-engine" or self.source != "user-engine":
+                raise ValueError("legacy_admission_requires_original_caller")
+        if (
+            self.legacy_invitation_binding is not None
+            and self.legacy_invitation_binding not in self.bindings
+        ):
+            raise ValueError("invalid_legacy_binding")
+
+
+def load_callers(path):
+    """Owner-controlled non-secret policy plus independently delivered token files."""
+    try:
+        data = json.loads(Path(path).read_text(encoding="utf-8"))
+        if data["schema_version"] != 1 or not isinstance(data["callers"], list):
+            raise ValueError
+        policies = []
+        fields = set(CallerPolicy.__dataclass_fields__) - {"token"}
+        for entry in data["callers"]:
+            if set(entry) - (fields | {"token_file"}) or "token" in entry:
+                raise ValueError
+            values = dict(entry)
+            token_path = Path(values.pop("token_file"))
+            if not token_path.is_absolute():
+                raise ValueError
+            # Kubernetes projections legitimately use symlinks. The projection
+            # and policy mounts are owner-controlled, read-only deployment inputs.
+            with token_path.open("rb") as stream:
+                raw_token = stream.read(515)
+                if len(raw_token) > 514:
+                    raise ValueError
+                token = raw_token.decode("ascii").strip()
+            values["token"] = token
+            values["bindings"] = frozenset(tuple(v) for v in values["bindings"])
+            for key in ("operations", "templates"):
+                values[key] = frozenset(values[key])
+            if values.get("legacy_invitation_binding") is not None:
+                values["legacy_invitation_binding"] = tuple(values["legacy_invitation_binding"])
+            policies.append(CallerPolicy(**values))
+        validate_callers(policies)
+        return policies
+    except (OSError, UnicodeError, ValueError, TypeError, KeyError):
+        raise ValueError("caller_configuration_invalid") from None
+
+
+def validate_callers(policies):
+    if not policies or len(policies) > 32:
+        raise ValueError("invalid_caller_count")
+    if len({p.caller_id for p in policies}) != len(policies):
+        raise ValueError("duplicate_caller_identity")
+    if len({p.token for p in policies}) != len(policies):
+        raise ValueError("shared_caller_credential")
+
+
+class ScopedTransactionalApplication:
+    def __init__(self, store, provider, portal_url, policies, *, clock=time.time):
+        from .transactional import TransactionalApplication
+
+        validate_callers(policies)
+        self.store, self.policies, self.clock = store, tuple(policies), clock
+        self.internal_token = secrets.token_urlsafe(32)
+        self.apps = {
+            p.caller_id: TransactionalApplication(
+                store,
+                provider,
+                self.internal_token,
+                portal_url,
+                verification_namespace="" if p.adopt_legacy_verifications else p.caller_id,
+                verification_bindings=p.bindings,
+            )
+            for p in policies
+        }
+
+    @staticmethod
+    def _json(start_response, status, payload, *, retry_after=None):
+        body = json.dumps(payload).encode()
+        headers = [
+            ("Content-Type", "application/json"),
+            ("Content-Length", str(len(body))),
+            ("Cache-Control", "private, no-store"),
+        ]
+        if retry_after is not None:
+            headers.append(("Retry-After", str(retry_after)))
+        start_response(f"{status.value} {status.phrase}", headers)
+        return [body]
+
+    @staticmethod
+    def _event_key(policy, event_id):
+        if (
+            not isinstance(event_id, str)
+            or not IDENTIFIER.fullmatch(event_id)
+            or event_id.startswith("ec2_")
+        ):
+            raise ValueError("invalid_event_id")
+        if policy.legacy_invitation_binding is not None:
+            return event_id
+        return (
+            "ec2_" + hashlib.sha256(json.dumps([policy.caller_id, event_id]).encode()).hexdigest()
+        )
+
+    @staticmethod
+    def _reference(policy, event_id):
+        return (
+            "dvr_"
+            + hashlib.sha256(json.dumps([policy.caller_id, event_id]).encode()).hexdigest()[:32]
+        )
+
+    def __call__(self, environ, start_response):
+        path = environ.get("PATH_INFO")
+        if path in {"/healthz", "/readyz"} and environ.get("REQUEST_METHOD") in {"GET", "HEAD"}:
+            return self._json(start_response, HTTPStatus.OK, {"status": "ok"})
+        if path not in OPERATIONS or environ.get("REQUEST_METHOD") != "POST":
+            return self._json(start_response, HTTPStatus.NOT_FOUND, {"error": "not_found"})
+        authorization = str(environ.get("HTTP_AUTHORIZATION", ""))
+        if len(authorization) > 1024:
+            return self._json(start_response, HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"})
+        try:
+            matches = [
+                p
+                for p in self.policies
+                if hmac.compare_digest(
+                    authorization.encode("utf-8"),
+                    f"Bearer {p.token}".encode(),
+                )
+            ]
+        except UnicodeError:
+            matches = []
+        if len(matches) != 1:
+            return self._json(start_response, HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"})
+        policy = matches[0]
+        if OPERATIONS[path] not in policy.operations:
+            return self._json(
+                start_response, HTTPStatus.FORBIDDEN, {"error": "operation_not_allowed"}
+            )
+        try:
+            now = self.clock()
+            if not math.isfinite(now):
+                raise RuntimeError
+            allowed, wait = self.store.request_budget(
+                policy.caller_id,
+                rate=policy.rate_per_minute,
+                burst=policy.burst,
+                now=now,
+            )
+            if not allowed:
+                return self._json(
+                    start_response,
+                    HTTPStatus.TOO_MANY_REQUESTS,
+                    {"error": "rate_limited", "retryable": True},
+                    retry_after=wait,
+                )
+            length = int(environ.get("CONTENT_LENGTH") or 0)
+            if not 0 < length <= policy.max_payload_bytes:
+                return self._json(
+                    start_response, HTTPStatus.REQUEST_ENTITY_TOO_LARGE, {"error": "invalid_size"}
+                )
+            raw = environ["wsgi.input"].read(length)
+            if len(raw) != length:
+                raise ValueError("invalid_request")
+            payload = json.loads(raw)
+            if not isinstance(payload, dict):
+                raise ValueError("invalid_request")
+            return self._dispatch(policy, path, payload, environ, start_response)
+        except ValueError as exc:
+            code = str(exc)
+            permitted = ERRORS | {
+                "invalid_request",
+                "invalid_event_id",
+                "caller_scope_denied",
+                "invalid_suppression_request",
+            }
+            return self._json(
+                start_response,
+                HTTPStatus.BAD_REQUEST,
+                {"error": code if code in permitted else "invalid_request"},
+            )
+        except Exception:
+            return self._json(
+                start_response,
+                HTTPStatus.SERVICE_UNAVAILABLE,
+                {"error": "operation_unconfirmed", "retryable": False},
+            )
+
+    def _dispatch(self, policy, path, payload, environ, start_response):
+        from .transactional import _valid_address
+
+        external_id = None
+        if path == "/v1/delivery-status":
+            external_id = payload.get("event_id")
+            status = self.store.delivery_status(self._event_key(policy, external_id))
+            result = {"event_id": external_id, "state": status["state"]}
+            if status["state"] == "provider_accepted":
+                result["reference"] = self._reference(policy, external_id)
+                result["evidence_ceiling"] = "provider_accepted"
+            return self._json(start_response, HTTPStatus.OK, result)
+        if path in {"/v1/suppressions", "/v1/suppressions/remove"}:
+            recipient = payload.get("recipient")
+            if (
+                not isinstance(recipient, str)
+                or not _valid_address(recipient)
+                or len(recipient) > 254
+            ):
+                raise ValueError("invalid_suppression_request")
+            if path.endswith("/remove"):
+                self.store.unsuppress(recipient)
+            else:
+                reason = payload.get("reason", "manual")
+                if reason not in {"manual", "hard_bounce", "complaint", "unsubscribe"}:
+                    raise ValueError("invalid_suppression_request")
+                with self.store.lock:
+                    self.store.suppress(recipient, reason)
+            return self._json(start_response, HTTPStatus.OK, {"status": "updated"})
+        if path == "/v1/send":
+            external_id = payload.get("id")
+            event_key = self._event_key(policy, external_id)
+            if environ.get("HTTP_IDEMPOTENCY_KEY") != external_id:
+                raise ValueError("idempotency_key_mismatch")
+            if (
+                payload.get("source") != policy.source
+                or not isinstance(payload.get("type"), str)
+                or payload["type"] not in policy.templates
+            ):
+                raise ValueError("template_not_allowed")
+            binding = (payload.get("tenant"), payload.get("client_id"))
+            if binding == (None, None) and policy.legacy_invitation_binding is not None:
+                binding = policy.legacy_invitation_binding
+            if not all(isinstance(v, str) for v in binding) or binding not in policy.bindings:
+                raise ValueError("caller_scope_denied")
+            data = payload.get("data")
+            if (
+                not isinstance(data, dict)
+                or not isinstance(data.get("invitation_id"), str)
+                or not IDENTIFIER.fullmatch(data["invitation_id"])
+            ):
+                raise ValueError("invalid_request")
+            recipient = data.get("primary_email")
+            if not isinstance(recipient, str) or len(recipient) > 254:
+                raise ValueError("invalid_recipient")
+            payload = dict(payload, id=event_key, source="user-engine")
+        if path == "/v1/registration-verifications":
+            binding = (payload.get("tenant"), payload.get("client_id"))
+            if not all(isinstance(v, str) for v in binding) or binding not in policy.bindings:
+                raise ValueError("caller_scope_denied")
+        raw = json.dumps(payload).encode()
+        internal = dict(environ, CONTENT_LENGTH=str(len(raw)))
+        internal["wsgi.input"] = io.BytesIO(raw)
+        internal["HTTP_AUTHORIZATION"] = f"Bearer {self.internal_token}"
+        if external_id is not None:
+            internal["HTTP_IDEMPOTENCY_KEY"] = self._event_key(policy, external_id)
+        response = {}
+        body = json.loads(
+            b"".join(
+                self.apps[policy.caller_id](
+                    internal,
+                    lambda status, headers: response.update(status=int(status.split()[0])),
+                )
+            )
+        )
+        status = HTTPStatus(response["status"])
+        if "error" in body:
+            error = body["error"] if body["error"] in ERRORS else "invalid_request"
+            body = {
+                "error": error,
+                **({"retryable": body["retryable"]} if "retryable" in body else {}),
+            }
+        if external_id is not None and status in {HTTPStatus.OK, HTTPStatus.ACCEPTED}:
+            body = {
+                "event_id": external_id,
+                "status": body["status"],
+                "reference": self._reference(policy, external_id),
+                "evidence_ceiling": "provider_accepted",
+            }
+        return self._json(start_response, status, body)

src/email_connect/transactional.py

@@ -2,17 +2,18 @@
 
 from __future__ import annotations
 
-import hmac
 import hashlib
+import hmac
 import ipaddress
 import json
+import math
 import os
+import secrets
 import smtplib
+import sqlite3
 import ssl
 import time
-import sqlite3
-import secrets
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta
 from email.message import EmailMessage
 from email.utils import make_msgid, parseaddr
 from http import HTTPStatus
@@ -36,42 +37,85 @@ class SQLiteDeliveryStore:
         self.db = sqlite3.connect(path, check_same_thread=False)
         self.lock = RLock()
         self.db.execute(
-            "CREATE TABLE IF NOT EXISTS deliveries (event_id TEXT PRIMARY KEY, provider_ref TEXT NOT NULL)"
+            "CREATE TABLE IF NOT EXISTS deliveries (event_id TEXT PRIMARY KEY, "
+            "provider_ref TEXT NOT NULL)"
         )
         self.db.execute("""CREATE TABLE IF NOT EXISTS verifications (
           request_id TEXT PRIMARY KEY, handle_hash TEXT UNIQUE NOT NULL,
           registration_id TEXT NOT NULL, email TEXT NOT NULL, username TEXT NOT NULL,
           client_id TEXT NOT NULL, tenant TEXT NOT NULL, display_name TEXT,
-          expires_at TEXT NOT NULL, consumed_at TEXT, canceled_at TEXT
+          expires_at TEXT NOT NULL, consumed_at TEXT, canceled_at TEXT,
+          caller_id TEXT NOT NULL DEFAULT ''
         )""")
         self.db.execute("""CREATE TABLE IF NOT EXISTS suppressions (
           email TEXT PRIMARY KEY NOT NULL,
           reason TEXT NOT NULL,
           created_at TEXT NOT NULL
         )""")
-        columns = {
-            row[1] for row in self.db.execute("PRAGMA table_info(verifications)")
-        }
+        columns = {row[1] for row in self.db.execute("PRAGMA table_info(verifications)")}
         if "canceled_at" not in columns:
             self.db.execute("ALTER TABLE verifications ADD COLUMN canceled_at TEXT")
+        if "caller_id" not in columns:
             self.db.execute(
-            "CREATE TABLE IF NOT EXISTS send_attempts (event_id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, state TEXT NOT NULL)"
+                "ALTER TABLE verifications ADD COLUMN caller_id TEXT NOT NULL DEFAULT ''"
             )
+        self.db.execute(
+            "CREATE TABLE IF NOT EXISTS send_attempts (event_id TEXT PRIMARY KEY, "
+            "fingerprint TEXT NOT NULL, state TEXT NOT NULL)"
+        )
+        send_columns = {row[1] for row in self.db.execute("PRAGMA table_info(send_attempts)")}
+        for name, kind in (("error", "TEXT"), ("retryable", "INTEGER")):
+            if name not in send_columns:
+                self.db.execute(f"ALTER TABLE send_attempts ADD COLUMN {name} {kind}")
         self.db.execute("""CREATE TABLE IF NOT EXISTS verification_attempts (
           operation_key TEXT PRIMARY KEY, fingerprint TEXT NOT NULL,
           request_id TEXT NOT NULL, state TEXT NOT NULL,
           error TEXT, retryable INTEGER, provider_ref TEXT
         )""")
+        self.db.execute("""CREATE TABLE IF NOT EXISTS caller_buckets (
+          caller_id TEXT PRIMARY KEY, tokens REAL NOT NULL, updated_at REAL NOT NULL
+        )""")
         self.db.commit()
 
-    def reserve_verification(self, payload, handle_hash, expires_at):
+    def request_budget(self, caller_id, *, rate, burst, now):
+        """One durable token bucket shared by connections, restarts and operations."""
+        with self.lock, self.db:
+            self.db.execute("BEGIN IMMEDIATE")
+            row = self.db.execute(
+                "SELECT tokens,updated_at FROM caller_buckets WHERE caller_id=?",
+                (caller_id,),
+            ).fetchone()
+            tokens, previous = row if row else (float(burst), now)
+            tokens = min(float(burst), tokens + max(0, now - previous) * rate / 60)
+            allowed = tokens >= 1
+            self.db.execute(
+                "INSERT OR REPLACE INTO caller_buckets VALUES (?,?,?)",
+                (caller_id, tokens - int(allowed), max(now, previous)),
+            )
+            wait = max(1, math.ceil((1 - tokens) * 60 / rate)) if not allowed else 0
+            return allowed, wait
+
+    def unsuppress(self, email):
+        with self.lock, self.db:
+            self.db.execute("DELETE FROM suppressions WHERE email=?", (email.lower(),))
+
+    def reserve_verification(self, payload, handle_hash, expires_at, *, caller_id=""):
         # Registration identity, not a tracing ID, defines the logical operation.
         identity = [payload[key] for key in ("tenant", "client_id", "registration_id")]
+        if caller_id:
+            identity.insert(0, caller_id)
         operation_key = hashlib.sha256(json.dumps(identity).encode()).hexdigest()
-        binding = {key: payload.get(key) for key in (
-            "registration_id", "normalized_email", "preferred_username",
-            "client_id", "tenant", "display_name",
-        )}
+        binding = {
+            key: payload.get(key)
+            for key in (
+                "registration_id",
+                "normalized_email",
+                "preferred_username",
+                "client_id",
+                "tenant",
+                "display_name",
+            )
+        }
         fingerprint = hashlib.sha256(
             json.dumps(binding, sort_keys=True, separators=(",", ":")).encode()
         ).hexdigest()
@@ -79,12 +123,19 @@ class SQLiteDeliveryStore:
             self.db.execute("BEGIN IMMEDIATE")
             row = self.db.execute(
                 "SELECT fingerprint,request_id,state,error,retryable "
-                "FROM verification_attempts WHERE operation_key=?", (operation_key,),
+                "FROM verification_attempts WHERE operation_key=?",
+                (operation_key,),
             ).fetchone()
             if not row:
                 legacy = self.db.execute(
-                    "SELECT 1 FROM verifications WHERE registration_id=? AND client_id=? AND tenant=? LIMIT 1",
-                    (payload["registration_id"], payload["client_id"], payload["tenant"]),
+                    "SELECT 1 FROM verifications WHERE registration_id=? AND client_id=? AND "
+                    "tenant=? AND caller_id=? LIMIT 1",
+                    (
+                        payload["registration_id"],
+                        payload["client_id"],
+                        payload["tenant"],
+                        caller_id,
+                    ),
                 ).fetchone()
                 if legacy:
                     # Pre-upgrade challenges have no trustworthy send outcome.
@@ -99,7 +150,9 @@ class SQLiteDeliveryStore:
                     raise ProviderError("delivery_outcome_unknown", retryable=False)
                 if not row[4]:
                     raise ProviderError(row[3], retryable=False)
-            request_id = self._insert_verification(payload, handle_hash, expires_at)
+            request_id = self._insert_verification(
+                payload, handle_hash, expires_at, caller_id=caller_id
+            )
             self.db.execute(
                 "INSERT OR REPLACE INTO verification_attempts VALUES (?,?,?,?,NULL,NULL,NULL)",
                 (operation_key, fingerprint, request_id, "in_flight"),
@@ -110,7 +163,8 @@ class SQLiteDeliveryStore:
         with self.lock, self.db:
             self.db.execute(
                 "UPDATE verification_attempts SET state='accepted',provider_ref=? "
-                "WHERE request_id=? AND state='in_flight'", (reference, request_id),
+                "WHERE request_id=? AND state='in_flight'",
+                (reference, request_id),
             )
 
     def verification_failed(self, request_id, error):
@@ -120,7 +174,7 @@ class SQLiteDeliveryStore:
             self.db.execute("BEGIN IMMEDIATE")
             self.db.execute(
                 "UPDATE verifications SET canceled_at=? WHERE request_id=?",
-                (datetime.now(timezone.utc).isoformat(), request_id),
+                (datetime.now(UTC).isoformat(), request_id),
             )
             self.db.execute(
                 "UPDATE verification_attempts SET state='failed',error=?,retryable=? "
@@ -133,7 +187,7 @@ class SQLiteDeliveryStore:
             self.db.execute("BEGIN IMMEDIATE")
             try:
                 row = self.db.execute(
-                    "SELECT fingerprint,state FROM send_attempts WHERE event_id=?",
+                    "SELECT fingerprint,state,error,retryable FROM send_attempts WHERE event_id=?",
                     (event_id,),
                 ).fetchone()
                 if row and row[0] != fingerprint:
@@ -142,20 +196,23 @@ class SQLiteDeliveryStore:
                     return False
                 if row and row[1] == "in_flight":
                     raise ProviderError("delivery_outcome_unknown", retryable=False)
+                if row and row[1] == "failed" and not row[3]:
+                    raise ProviderError(row[2] or "delivery_outcome_unknown", retryable=False)
                 self.db.execute(
-                    "INSERT OR REPLACE INTO send_attempts VALUES (?,?,?)",
+                    "INSERT OR REPLACE INTO send_attempts (event_id,fingerprint,state) VALUES "
+                    "(?,?,?)",
                     (event_id, fingerprint, "in_flight"),
                 )
                 return True
             finally:
                 self.db.commit()
 
-    def send_failed(self, event_id, definitive):
+    def send_failed(self, event_id, definitive, error):
         if definitive:
             with self.db:
                 self.db.execute(
-                    "UPDATE send_attempts SET state='failed' WHERE event_id=?",
-                    (event_id,),
+                    "UPDATE send_attempts SET state='failed',error=?,retryable=? WHERE event_id=?",
+                    (error.code, int(error.retryable), event_id),
                 )
 
     def delivery_status(self, event_id):
@@ -167,9 +224,7 @@ class SQLiteDeliveryStore:
         ).fetchone()
         return {
             "state": (
-                "unknown"
-                if row and row[0] == "in_flight"
-                else "failed" if row else "not_found"
+                "unknown" if row and row[0] == "in_flight" else "failed" if row else "not_found"
             )
         }
 
@@ -181,9 +236,7 @@ class SQLiteDeliveryStore:
 
     def record(self, event_id: str, reference: str) -> None:
         with self.db:
-            self.db.execute(
-                "INSERT INTO deliveries VALUES (?, ?)", (event_id, reference)
-            )
+            self.db.execute("INSERT INTO deliveries VALUES (?, ?)", (event_id, reference))
 
     def is_suppressed(self, email: str) -> bool:
         row = self.db.execute(
@@ -195,22 +248,20 @@ class SQLiteDeliveryStore:
         with self.db:
             self.db.execute(
                 "INSERT OR REPLACE INTO suppressions VALUES (?,?,?)",
-                (email.lower(), reason, datetime.now(timezone.utc).isoformat()),
+                (email.lower(), reason, datetime.now(UTC).isoformat()),
             )
 
-    def create_verification(
-        self, payload: dict, handle_hash: str, expires_at: str
-    ) -> str:
+    def create_verification(self, payload: dict, handle_hash: str, expires_at: str) -> str:
         with self.lock, self.db:
             return self._insert_verification(payload, handle_hash, expires_at)
 
-    def _insert_verification(self, payload, handle_hash, expires_at):
+    def _insert_verification(self, payload, handle_hash, expires_at, *, caller_id=""):
         request_id = f"vrq_{secrets.token_hex(12)}"
         self.db.execute(
             """INSERT INTO verifications
           (request_id, handle_hash, registration_id, email, username,
-           client_id, tenant, display_name, expires_at, consumed_at, canceled_at)
-          VALUES (?,?,?,?,?,?,?,?,?,NULL,NULL)""",
+           client_id, tenant, display_name, expires_at, consumed_at, canceled_at, caller_id)
+          VALUES (?,?,?,?,?,?,?,?,?,NULL,NULL,?)""",
             (
                 request_id,
                 handle_hash,
@@ -221,11 +272,12 @@ class SQLiteDeliveryStore:
                 payload["tenant"],
                 payload.get("display_name"),
                 expires_at,
+                caller_id,
             ),
         )
         return request_id
 
-    def consume_verification(self, handle_hash: str) -> dict:
+    def consume_verification(self, handle_hash: str, *, caller_id="", bindings=None) -> dict:
         with self.lock, self.db:
             self.db.execute("BEGIN IMMEDIATE")
             row = self.db.execute(
@@ -233,11 +285,13 @@ class SQLiteDeliveryStore:
             ).fetchone()
             if row is None or row[9] is not None or row[10] is not None:
                 raise ValueError("verification_invalid")
-            if datetime.fromisoformat(row[8]) <= datetime.now(timezone.utc):
+            if row[11] != caller_id or (bindings is not None and (row[6], row[5]) not in bindings):
+                raise ValueError("verification_invalid")
+            if datetime.fromisoformat(row[8]) <= datetime.now(UTC):
                 raise ValueError("verification_expired")
             self.db.execute(
                 "UPDATE verifications SET consumed_at=? WHERE request_id=?",
-                (datetime.now(timezone.utc).isoformat(), row[0]),
+                (datetime.now(UTC).isoformat(), row[0]),
             )
         return {
             "request_id": row[0],
@@ -249,7 +303,7 @@ class SQLiteDeliveryStore:
             "display_name": row[7],
         }
 
-    def cancel_verification(self, handle_hash: str) -> dict:
+    def cancel_verification(self, handle_hash: str, *, caller_id="", bindings=None) -> dict:
         with self.lock, self.db:
             self.db.execute("BEGIN IMMEDIATE")
             row = self.db.execute(
@@ -257,11 +311,13 @@ class SQLiteDeliveryStore:
             ).fetchone()
             if row is None or row[9] is not None or row[10] is not None:
                 raise ValueError("verification_invalid")
-            if datetime.fromisoformat(row[8]) <= datetime.now(timezone.utc):
+            if row[11] != caller_id or (bindings is not None and (row[6], row[5]) not in bindings):
+                raise ValueError("verification_invalid")
+            if datetime.fromisoformat(row[8]) <= datetime.now(UTC):
                 raise ValueError("verification_expired")
             self.db.execute(
                 "UPDATE verifications SET canceled_at=? WHERE request_id=?",
-                (datetime.now(timezone.utc).isoformat(), row[0]),
+                (datetime.now(UTC).isoformat(), row[0]),
             )
         return {
             "request_id": row[0],
@@ -322,9 +378,7 @@ class SMTPProvider:
         # that ties an accepted send to a message later observed in a mailbox.
         # Without it the reference below fell back to hash(), which Python
         # randomizes per process and which collides for equal recipient/subject.
-        message["Message-ID"] = make_msgid(
-            domain=self.sender.rpartition("@")[2] or None
-        )
+        message["Message-ID"] = make_msgid(domain=self.sender.rpartition("@")[2] or None)
         message.set_content(text)
         submitted = False
         accepted = False
@@ -363,10 +417,19 @@ class SMTPProvider:
 
 class TransactionalApplication:
     def __init__(
-        self, store: SQLiteDeliveryStore, provider, bearer_token: str, portal_url: str
+        self,
+        store: SQLiteDeliveryStore,
+        provider,
+        bearer_token: str,
+        portal_url: str,
+        *,
+        verification_namespace="",
+        verification_bindings=None,
     ) -> None:
         self.store, self.provider, self.token = store, provider, bearer_token
         self.portal_url = portal_url.rstrip("/")
+        self.verification_namespace = verification_namespace
+        self.verification_bindings = verification_bindings
         self._status_cache = None
         self._status_time = 0.0
 
@@ -386,15 +449,11 @@ class TransactionalApplication:
             )
             or environ.get("REQUEST_METHOD") != "POST"
         ):
-            return self._json(
-                start_response, HTTPStatus.NOT_FOUND, {"error": "not_found"}
-            )
+            return self._json(start_response, HTTPStatus.NOT_FOUND, {"error": "not_found"})
         if not hmac.compare_digest(
             str(environ.get("HTTP_AUTHORIZATION", "")), f"Bearer {self.token}"
         ):
-            return self._json(
-                start_response, HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"}
-            )
+            return self._json(start_response, HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"})
         try:
             length = int(environ.get("CONTENT_LENGTH") or 0)
             if length <= 0 or length > 128 * 1024:
@@ -417,9 +476,7 @@ class TransactionalApplication:
             return self._send_invitation(start_response, environ, payload)
         except ProviderError as exc:
             status = (
-                HTTPStatus.SERVICE_UNAVAILABLE
-                if exc.retryable
-                else HTTPStatus.UNPROCESSABLE_ENTITY
+                HTTPStatus.SERVICE_UNAVAILABLE if exc.retryable else HTTPStatus.UNPROCESSABLE_ENTITY
             )
             return self._json(
                 start_response,
@@ -427,9 +484,7 @@ class TransactionalApplication:
                 {"error": exc.code, "retryable": exc.retryable},
             )
         except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
-            return self._json(
-                start_response, HTTPStatus.BAD_REQUEST, {"error": str(exc)}
-            )
+            return self._json(start_response, HTTPStatus.BAD_REQUEST, {"error": str(exc)})
         except (TimeoutError, OSError, smtplib.SMTPException):
             # Safety net: never leak provider exception text.
             return self._json(
@@ -454,7 +509,7 @@ class TransactionalApplication:
                 pass
             self._status_cache = {
                 "checks": checks,
-                "checked_at": datetime.now(timezone.utc).isoformat(),
+                "checked_at": datetime.now(UTC).isoformat(),
                 "message_sent": False,
                 "inbox_receipt_verified": False,
             }
@@ -465,10 +520,7 @@ class TransactionalApplication:
         event_id = str(payload["id"])
         if environ.get("HTTP_IDEMPOTENCY_KEY") != event_id:
             raise ValueError("idempotency_key_mismatch")
-        if (
-            payload.get("source") != "user-engine"
-            or payload.get("type") not in ALLOWED_EVENTS
-        ):
+        if payload.get("source") != "user-engine" or payload.get("type") not in ALLOWED_EVENTS:
             raise ValueError("template_not_allowed")
         recipient = str(payload.get("data", {}).get("primary_email", ""))
         if not _valid_address(recipient):
@@ -491,7 +543,8 @@ class TransactionalApplication:
             reference = self.provider.send(
                 recipient,
                 "Your NetKingdom invitation",
-                f"You have been invited. Continue securely at {self.portal_url}/invitations/{invitation_id}\n",
+                f"You have been invited. Continue securely at "
+                f"{self.portal_url}/invitations/{invitation_id}\n",
             )
             self.store.record(event_id, reference)
         except ProviderError as exc:
@@ -500,11 +553,9 @@ class TransactionalApplication:
                 "definitive",
                 exc.code in {"temporary_deferral", "permanent_rejection"},
             )
-            self.store.send_failed(event_id, definitive)
+            self.store.send_failed(event_id, definitive, exc)
             if not definitive:
-                raise ProviderError(
-                    "delivery_outcome_unknown", retryable=False
-                ) from None
+                raise ProviderError("delivery_outcome_unknown", retryable=False) from None
             raise
         except Exception:
             # Preserve the durable reservation: SMTP may have accepted the message.
@@ -530,17 +581,20 @@ class TransactionalApplication:
             "tenant",
             "correlation_id",
         )
-        if any(not isinstance(payload.get(key), str) or not payload[key] for key in required) or not _valid_address(
-            str(payload["normalized_email"])
-        ):
+        if any(
+            not isinstance(payload.get(key), str) or not payload[key] for key in required
+        ) or not _valid_address(str(payload["normalized_email"])):
             raise ValueError("invalid_verification_request")
         email = str(payload["normalized_email"])
         if self.store.is_suppressed(email):
             raise ValueError("recipient_suppressed")
         handle = secrets.token_urlsafe(32)
-        expires_at = datetime.now(timezone.utc) + timedelta(minutes=30)
+        expires_at = datetime.now(UTC) + timedelta(minutes=30)
         request_id, send = self.store.reserve_verification(
-            payload, hashlib.sha256(handle.encode()).hexdigest(), expires_at.isoformat()
+            payload,
+            hashlib.sha256(handle.encode()).hexdigest(),
+            expires_at.isoformat(),
+            caller_id=self.verification_namespace,
         )
         if send:
             try:
@@ -548,13 +602,15 @@ class TransactionalApplication:
                     email,
                     "Verify your NetKingdom registration",
                     f"Continue securely at {self.portal_url}/registration/verify?handle={handle}\n"
-                    f"Cancel this request at {self.portal_url}/registration/cancel?handle={handle}\n"
+                    f"Cancel this request at {self.portal_url}/registration/cancel"
+                    f"?handle={handle}\n"
                     "These links expire in 30 minutes.\n",
                 )
                 self.store.verification_accepted(request_id, reference)
             except ProviderError as exc:
                 definitive = getattr(
-                    exc, "definitive",
+                    exc,
+                    "definitive",
                     exc.code in {"temporary_deferral", "permanent_rejection"},
                 )
                 if not definitive:
@@ -577,7 +633,9 @@ class TransactionalApplication:
         if len(handle) < 32:
             raise ValueError("verification_invalid")
         evidence = self.store.consume_verification(
-            hashlib.sha256(handle.encode()).hexdigest()
+            hashlib.sha256(handle.encode()).hexdigest(),
+            caller_id=self.verification_namespace,
+            bindings=self.verification_bindings,
         )
         # mailbox_control is channel evidence only; never an authorization decision.
         return self._json(
@@ -599,7 +657,9 @@ class TransactionalApplication:
         if len(handle) < 32:
             raise ValueError("verification_invalid")
         evidence = self.store.cancel_verification(
-            hashlib.sha256(handle.encode()).hexdigest()
+            hashlib.sha256(handle.encode()).hexdigest(),
+            caller_id=self.verification_namespace,
+            bindings=self.verification_bindings,
         )
         return self._json(
             start_response,
@@ -639,15 +699,15 @@ def _is_loopback_host(host: str) -> bool:
 
 def _valid_address(value: str) -> bool:
     _, address = parseaddr(value)
-    return (
-        address == value
-        and "@" in address
-        and "\n" not in address
-        and "\r" not in address
-    )
+    return address == value and "@" in address and "\n" not in address and "\r" not in address
 
 
 def main() -> None:
+    policies = None
+    if "EMAIL_CONNECT_CALLERS_FILE" in os.environ:
+        from .callers import ScopedTransactionalApplication, load_callers
+
+        policies = load_callers(os.environ["EMAIL_CONNECT_CALLERS_FILE"])
     provider = SMTPProvider(
         os.environ["EMAIL_CONNECT_SMTP_HOST"],
         int(os.environ.get("EMAIL_CONNECT_SMTP_PORT", "587")),
@@ -656,10 +716,19 @@ def main() -> None:
         os.environ["EMAIL_CONNECT_SENDER"],
         security=os.environ.get("EMAIL_CONNECT_SMTP_SECURITY", "starttls"),
     )
-    app = TransactionalApplication(
-        SQLiteDeliveryStore(
+    store = SQLiteDeliveryStore(
         os.environ.get("EMAIL_CONNECT_DATABASE_PATH", "/data/email-connect.db")
-        ),
+    )
+    if policies is not None:
+        app = ScopedTransactionalApplication(
+            store,
+            provider,
+            os.environ["EMAIL_CONNECT_PORTAL_URL"],
+            policies,
+        )
+    else:
+        app = TransactionalApplication(
+            store,
             provider,
             os.environ["EMAIL_CONNECT_INGEST_TOKEN"].strip(),
             os.environ["EMAIL_CONNECT_PORTAL_URL"],

tests/test_callers.py

@@ -0,0 +1,414 @@
+"""Admitted caller contract: only synthetic tokens, providers, recipients and ledgers."""
+
+import io
+import json
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+from email_connect.callers import CallerPolicy, ScopedTransactionalApplication, load_callers
+from email_connect.transactional import ProviderError, SQLiteDeliveryStore, TransactionalApplication
+
+
+class Provider:
+    def __init__(self):
+        self.calls = []
+
+    def send(self, *args):
+        self.calls.append(args)
+        return "synthetic-reference-recipient=member@example.test-token=must-not-leak"
+
+    def check(self):
+        return None
+
+
+def policy(name="caller-a", **changes):
+    values = dict(
+        caller_id=name,
+        token="synthetic-token-" + name,
+        source=name,
+        bindings=frozenset({("tenant:a", "client:a")}),
+        operations=frozenset({"send", "delivery-status", "verify", "consume", "cancel"}),
+        templates=frozenset({"family_member.invited"}),
+        rate_per_minute=60,
+        burst=100,
+        max_payload_bytes=32768,
+    )
+    values.update(changes)
+    return CallerPolicy(**values)
+
+
+def event(caller="caller-a", **changes):
+    payload = {
+        "id": "evt-1",
+        "type": "family_member.invited",
+        "source": caller,
+        "tenant": "tenant:a",
+        "client_id": "client:a",
+        "data": {"primary_email": "member@example.test", "invitation_id": "invite-1"},
+    }
+    payload.update(changes)
+    return payload
+
+
+def verification(**changes):
+    result = {
+        "registration_id": "reg-1",
+        "normalized_email": "member@example.test",
+        "preferred_username": "member",
+        "tenant": "tenant:a",
+        "client_id": "client:a",
+        "correlation_id": "trace-1",
+    }
+    result.update(changes)
+    return result
+
+
+def invoke(app, payload, caller="caller-a", path="/v1/send", token=None, key=None):
+    raw = json.dumps(payload).encode()
+    response = {}
+    env = {
+        "PATH_INFO": path,
+        "REQUEST_METHOD": "POST",
+        "CONTENT_LENGTH": str(len(raw)),
+        "wsgi.input": io.BytesIO(raw),
+        "HTTP_AUTHORIZATION": "Bearer " + (token or "synthetic-token-" + caller),
+        "HTTP_IDEMPOTENCY_KEY": key if key is not None else payload.get("id", ""),
+    }
+    body = b"".join(
+        app(
+            env,
+            lambda status, headers: response.update(
+                status=int(status.split()[0]),
+                headers=dict(headers),
+            ),
+        )
+    )
+    return response["status"], json.loads(body), response["headers"]
+
+
+def application(tmp_path, policies=None, clock=None):
+    store = SQLiteDeliveryStore(str(tmp_path / "ledger.db"))
+    provider = Provider()
+    app = ScopedTransactionalApplication(
+        store,
+        provider,
+        "https://issuer.example.test",
+        policies or [policy(), policy("caller-b")],
+        **({"clock": clock} if clock is not None else {}),
+    )
+    return app, store, provider
+
+
+def test_credentials_determine_caller_and_status_is_scoped_and_redacted(tmp_path):
+    app, _, provider = application(tmp_path)
+    assert invoke(app, event(), token="wrong-credential")[0] == 401
+    assert invoke(app, event("caller-b"))[0] == 400
+    accepted, body, headers = invoke(app, event())
+    assert accepted == 202 and body["event_id"] == "evt-1" and body["reference"].startswith("dvr_")
+    assert headers["Cache-Control"] == "private, no-store"
+    status, record, _ = invoke(app, {"event_id": "evt-1"}, path="/v1/delivery-status")
+    assert status == 200 and record["state"] == "provider_accepted"
+    assert record["reference"] == body["reference"]
+    _, other, _ = invoke(app, {"event_id": "evt-1"}, caller="caller-b", path="/v1/delivery-status")
+    assert other["state"] == "not_found"
+    assert len(provider.calls) == 1
+    serialized = json.dumps([body, record, other])
+    assert "member@example.test" not in serialized and "must-not-leak" not in serialized
+
+
+@pytest.mark.parametrize(
+    "change",
+    [
+        {"tenant": "tenant:b"},
+        {"client_id": "client:b"},
+        {"tenant": ["tenant:a"]},
+        {"type": "unapproved-template"},
+        {"type": []},
+        {"source": "user-engine"},
+    ],
+)
+def test_scope_and_template_denials_never_submit(tmp_path, change):
+    app, _, provider = application(tmp_path)
+    assert invoke(app, event(**change))[0] == 400
+    assert provider.calls == []
+
+
+def test_retry_and_payload_identity_are_independent_across_callers_and_restart(tmp_path):
+    app, store, provider = application(tmp_path)
+    first = invoke(app, event())[1]
+    assert invoke(app, event())[1]["status"] == "duplicate"
+    assert (
+        invoke(
+            app, event(data={"primary_email": "other@example.test", "invitation_id": "invite-1"})
+        )[0]
+        == 400
+    )
+    assert invoke(app, event("caller-b"), caller="caller-b")[0] == 202
+    assert len(provider.calls) == 2
+    store.db.close()
+    restarted, _, empty_provider = application(tmp_path)
+    assert invoke(restarted, event())[1]["reference"] == first["reference"]
+    assert empty_provider.calls == []
+
+
+def test_unknown_submission_survives_restart_without_resend(tmp_path):
+    app, store, provider = application(tmp_path)
+
+    def unknown(*args):
+        provider.calls.append(args)
+        raise OSError("synthetic SMTP recipient/token must not leak")
+
+    provider.send = unknown
+    status, body, _ = invoke(app, event())
+    assert status == 422 and body == {"error": "delivery_outcome_unknown", "retryable": False}
+    store.db.close()
+    restarted, _, next_provider = application(tmp_path)
+    assert invoke(restarted, event())[1] == body
+    assert len(provider.calls) == 1 and next_provider.calls == []
+
+
+def test_rate_budget_survives_restart_and_backward_clock_and_is_per_caller(tmp_path):
+    now = [100.0]
+    policies = [policy(burst=1, rate_per_minute=2), policy("caller-b", burst=1, rate_per_minute=2)]
+    app, store, provider = application(tmp_path, policies, clock=lambda: now[0])
+    assert invoke(app, event())[0] == 202
+    status, _, headers = invoke(app, event(id="evt-2"))
+    assert status == 429 and headers["Retry-After"] == "30" and len(provider.calls) == 1
+    assert invoke(app, event("caller-b"), caller="caller-b")[0] == 202
+    store.db.close()
+    restarted, _, next_provider = application(tmp_path, policies, clock=lambda: now[0])
+    assert invoke(restarted, event(id="evt-2"))[0] == 429
+    now[0] = 90
+    assert invoke(restarted, event(id="evt-2"))[0] == 429
+    now[0] = 130
+    assert invoke(restarted, event(id="evt-2"))[0] == 202
+    assert len(next_provider.calls) == 1
+
+
+def test_concurrent_connections_cannot_exceed_request_burst(tmp_path):
+    first, store, provider = application(tmp_path, [policy(burst=1)])
+    second = ScopedTransactionalApplication(
+        SQLiteDeliveryStore(str(tmp_path / "ledger.db")),
+        provider,
+        "https://issuer.example.test",
+        [policy(burst=1)],
+    )
+    with ThreadPoolExecutor(max_workers=2) as workers:
+        responses = list(workers.map(lambda app: invoke(app, event()), [first, second]))
+    assert sorted(r[0] for r in responses) == [202, 429]
+    assert len(provider.calls) == 1
+
+
+def test_verification_owner_and_binding_checked_before_consumption_or_cancellation(tmp_path):
+    app, store, provider = application(tmp_path)
+    request = verification()
+    first = invoke(app, request, path="/v1/registration-verifications")[1]
+    assert invoke(app, request, path="/v1/registration-verifications")[1] == first
+    handle = provider.calls[0][2].split("handle=", 1)[1].splitlines()[0]
+    for path in ("/consume", "/cancel"):
+        assert (
+            invoke(
+                app,
+                {"handle": handle},
+                caller="caller-b",
+                path="/v1/registration-verifications" + path,
+            )[0]
+            == 400
+        )
+    assert invoke(app, request, caller="caller-b", path="/v1/registration-verifications")[0] == 202
+    assert len(provider.calls) == 2
+    legacy = TransactionalApplication(store, provider, "legacy", "https://issuer.example.test")
+    raw = json.dumps({"handle": handle}).encode()
+    response = {}
+    legacy(
+        {
+            "PATH_INFO": "/v1/registration-verifications/consume",
+            "REQUEST_METHOD": "POST",
+            "CONTENT_LENGTH": str(len(raw)),
+            "wsgi.input": io.BytesIO(raw),
+            "HTTP_AUTHORIZATION": "Bearer legacy",
+        },
+        lambda status, headers: response.update(status=int(status.split()[0])),
+    )
+    assert response["status"] == 400
+    limited = ScopedTransactionalApplication(
+        store,
+        provider,
+        "https://issuer.example.test",
+        [policy(bindings=frozenset({("tenant:b", "client:b")}))],
+    )
+    assert (
+        invoke(limited, {"handle": handle}, path="/v1/registration-verifications/consume")[0] == 400
+    )
+    assert invoke(app, {"handle": handle}, path="/v1/registration-verifications/consume")[0] == 200
+    assert invoke(app, {"handle": handle}, path="/v1/registration-verifications/cancel")[0] == 400
+
+
+def test_suppression_and_unsuppression_require_explicit_owner_operations(tmp_path):
+    ops = policy("mail-operator", operations=frozenset({"suppress", "unsuppress"}))
+    app, _, provider = application(tmp_path, [policy(), ops])
+    payload = {"recipient": "member@example.test", "reason": "complaint"}
+    assert invoke(app, payload, path="/v1/suppressions")[0] == 403
+    assert invoke(app, payload, caller="mail-operator", path="/v1/suppressions")[0] == 200
+    assert invoke(app, event())[1]["error"] == "recipient_suppressed"
+    assert not provider.calls
+    assert invoke(app, payload, caller="mail-operator", path="/v1/suppressions/remove")[0] == 200
+    assert invoke(app, event())[0] == 202
+
+
+def test_original_caller_legacy_reservations_and_links_need_explicit_adoption(tmp_path):
+    app, store, provider = application(tmp_path)
+    payload = verification()
+    legacy = TransactionalApplication(store, provider, "legacy", "https://issuer.example.test")
+    raw = json.dumps(payload).encode()
+    result = {}
+    legacy(
+        {
+            "PATH_INFO": "/v1/registration-verifications",
+            "REQUEST_METHOD": "POST",
+            "CONTENT_LENGTH": str(len(raw)),
+            "wsgi.input": io.BytesIO(raw),
+            "HTTP_AUTHORIZATION": "Bearer legacy",
+        },
+        lambda status, headers: result.update(status=status),
+    )
+    handle = provider.calls[0][2].split("handle=", 1)[1].splitlines()[0]
+    adopter = policy("user-engine", adopt_legacy_verifications=True)
+    adopted = ScopedTransactionalApplication(
+        store, provider, "https://issuer.example.test", [adopter]
+    )
+    assert (
+        invoke(adopted, payload, caller="user-engine", path="/v1/registration-verifications")[0]
+        == 202
+    )
+    assert len(provider.calls) == 1
+    assert (
+        invoke(
+            adopted,
+            {"handle": handle},
+            caller="user-engine",
+            path="/v1/registration-verifications/consume",
+        )[0]
+        == 200
+    )
+    with pytest.raises(ValueError, match="original_caller"):
+        policy(adopt_legacy_verifications=True)
+
+
+def test_payload_limit_unknown_operation_and_provider_status_have_no_implicit_grant(tmp_path):
+    app, _, provider = application(tmp_path, [policy(max_payload_bytes=100)])
+    assert invoke(app, event(data={"primary_email": "x" * 1000}))[0] == 413
+    assert invoke(app, {}, path="/v1/status")[0] == 403
+    assert invoke(app, {}, path="/unknown")[0] == 404
+    assert not provider.calls
+
+
+@pytest.mark.parametrize(
+    "change",
+    [
+        {"burst": 0},
+        {"rate_per_minute": True},
+        {"max_payload_bytes": 200000},
+        {"token": ""},
+        {"bindings": frozenset()},
+        {"templates": frozenset({"arbitrary-body"})},
+    ],
+)
+def test_policy_refuses_unbounded_or_unauthorized_configuration(change):
+    with pytest.raises(ValueError):
+        policy(**change)
+
+
+def test_policy_loader_uses_independent_files_and_never_echoes_credentials(tmp_path):
+    token_file = tmp_path / "token"
+    token_file.write_text("synthetic-token-caller-a\n")
+    entry = dict(
+        caller_id="caller-a",
+        source="caller-a",
+        bindings=[["tenant:a", "client:a"]],
+        operations=["send"],
+        templates=["family_member.invited"],
+        rate_per_minute=10,
+        burst=3,
+        max_payload_bytes=8192,
+        token_file=str(token_file),
+    )
+    config = tmp_path / "policy.json"
+    config.write_text(json.dumps({"schema_version": 1, "callers": [entry]}))
+    loaded = load_callers(config)
+    assert loaded[0].token == "synthetic-token-caller-a"
+    assert "synthetic-token" not in repr(loaded)
+    config.write_text(
+        json.dumps({"schema_version": 1, "callers": [entry, dict(entry, caller_id="caller-b")]})
+    )
+    with pytest.raises(ValueError, match="caller_configuration_invalid") as error:
+        load_callers(config)
+    assert "synthetic-token" not in str(error.value)
+    entry["token"] = "synthetic-embedded-secret"
+    config.write_text(json.dumps({"schema_version": 1, "callers": [entry]}))
+    with pytest.raises(ValueError, match="caller_configuration_invalid"):
+        load_callers(config)
+
+
+@pytest.mark.parametrize("retryable", [False, True])
+def test_definitive_failure_retains_retry_policy_after_restart(tmp_path, retryable):
+    app, store, provider = application(tmp_path)
+
+    def reject(*args):
+        provider.calls.append(args)
+        raise ProviderError(
+            "temporary_deferral" if retryable else "permanent_rejection", retryable=retryable
+        )
+
+    provider.send = reject
+    first = invoke(app, event())
+    assert first[0] == (503 if retryable else 422)
+    store.db.close()
+    restarted, _, next_provider = application(tmp_path)
+    result = invoke(restarted, event())
+    assert result[0] == (202 if retryable else 422)
+    assert len(next_provider.calls) == int(retryable)
+
+
+def test_legacy_invitation_adoption_preserves_accepted_receipt(tmp_path):
+    store = SQLiteDeliveryStore(str(tmp_path / "ledger.db"))
+    provider = Provider()
+    legacy = TransactionalApplication(
+        store, provider, "legacy-token", "https://issuer.example.test"
+    )
+    old = event("user-engine")
+    old.pop("tenant")
+    old.pop("client_id")
+    assert invoke(legacy, old, token="legacy-token")[0] == 202
+    admitted = policy("user-engine", legacy_invitation_binding=("tenant:a", "client:a"))
+    scoped = ScopedTransactionalApplication(
+        store, provider, "https://issuer.example.test", [admitted]
+    )
+    assert invoke(scoped, old, caller="user-engine")[1]["status"] == "duplicate"
+    assert len(provider.calls) == 1
+    assert invoke(scoped, dict(old, id="ec2_reserved"), caller="user-engine")[0] == 400
+
+
+def test_historical_failed_reservation_is_conservatively_terminal(tmp_path):
+    import sqlite3
+
+    path = tmp_path / "old.db"
+    with sqlite3.connect(path) as db:
+        db.execute(
+            "CREATE TABLE send_attempts (event_id TEXT PRIMARY KEY, "
+            "fingerprint TEXT NOT NULL, state TEXT NOT NULL)"
+        )
+        db.execute("INSERT INTO send_attempts VALUES ('old-event','old-fingerprint','failed')")
+    store = SQLiteDeliveryStore(str(path))
+    with pytest.raises(ProviderError, match="delivery_outcome_unknown"):
+        store.reserve_send("old-event", "old-fingerprint")
+
+
+def test_invalid_policy_refuses_startup_before_database_mutation(tmp_path, monkeypatch):
+    from email_connect.transactional import main
+
+    monkeypatch.setenv("EMAIL_CONNECT_CALLERS_FILE", str(tmp_path / "missing.json"))
+    monkeypatch.setenv("EMAIL_CONNECT_DATABASE_PATH", str(tmp_path / "not-created.db"))
+    with pytest.raises(ValueError, match="caller_configuration_invalid"):
+        main()
+    assert not (tmp_path / "not-created.db").exists()

tests/test_ledger_backup.py

@@ -433,3 +433,45 @@ def test_helper_has_no_secret_or_token_mount_and_requires_no_egress(modules):
         },
         labels,
     )
+
+
+def test_scoped_ledger_rehearsal_preserves_ownership_and_budgets(modules, tmp_path, monkeypatch, capsys):
+    archive, _, _ = modules
+    path = tmp_path / "source.sqlite3"
+    synthetic_store(path)
+    store = SQLiteDeliveryStore(str(path))
+    payload = {
+        "registration_id": "scoped", "normalized_email": "scoped@example.test",
+        "preferred_username": "synthetic", "client_id": "synthetic-client",
+        "tenant": "synthetic-tenant", "correlation_id": "trace-scoped",
+    }
+    expires = (datetime.now(UTC) + timedelta(days=1)).isoformat()
+    request, _ = store.reserve_verification(
+        payload, hashlib.sha256(b"scoped-handle").hexdigest(), expires, caller_id="caller-a"
+    )
+    store.verification_accepted(request, "synthetic-scoped-reference")
+    store.reserve_verification(
+        dict(payload, registration_id="scoped-unknown"),
+        hashlib.sha256(b"scoped-unknown-handle").hexdigest(), expires, caller_id="caller-b"
+    )
+    assert store.request_budget("caller-a", rate=2, burst=1, now=100)[0]
+    store.db.close()
+    before = archive.fingerprint(path)
+    assert before["tables"]["caller_buckets"]["rows"] == 1
+    script = Path(__file__).resolve().parents[1] / "scripts/rehearse-ledger.py"
+    spec = importlib.util.spec_from_file_location("rehearsal", script)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    recovered = tmp_path / "recovered.sqlite3"
+    code = module.CODE.replace("Path('/tmp/recovered.sqlite3')", "Path(" + repr(str(recovered)) + ")")
+    monkeypatch.setattr(sys, "stdin", io.TextIOWrapper(io.BytesIO(path.read_bytes())))
+    monkeypatch.setattr(__import__("os"), "getuid", lambda: 10001)
+    exec(compile(code, str(script), "exec"), {})
+    receipt = json.loads(capsys.readouterr().out)
+    assert receipt["checks"]["accepted_verifications"] == 4
+    assert receipt["checks"]["unknown_verifications"] == 2
+    assert receipt["mail_sent"] is False
+    assert archive.fingerprint(recovered) == before
+    restarted = SQLiteDeliveryStore(str(recovered))
+    assert not restarted.request_budget("caller-a", rate=2, burst=1, now=100)[0]
+    restarted.db.close()

workplans/EMAIL-WP-0008-scoped-transactional-callers.md

@@ -0,0 +1,70 @@
+---
+id: EMAIL-WP-0008
+type: workplan
+title: "Isolate explicitly admitted transactional mail callers"
+domain: infotech
+repo: email-connect
+status: active
+owner: codex
+topic_slug: netkingdom
+created: "2026-10-08"
+updated: "2026-10-08"
+origin: residual
+origin_ref: CSOC-WP-0010
+---
+
+CSOC-WP-0010-T04 needs independent caller credentials, bounded template/tenant
+permissions, durable limits and scoped diagnostics. This record owns receiver
+source and its admission; current live single-caller SMTP remains unchanged.
+
+## T01 — Implement explicit caller policies and durable isolation
+
+```task
+id: EMAIL-WP-0008-T01
+status: done
+priority: high
+```
+
+Opt-in read-only policy/token-file loader, fixed operation/template/binding
+admission, exact credential/source identity, per-caller invitations/verifications,
+atomic persistent request budgets, bounded input, redacted scoped status and
+privileged suppression controls. Invalid selected policy refuses startup without
+legacy fallback. Permanent invitation failures now remain terminal across
+restart; confirmed temporary failures can retry. Unknown outcomes never resend.
+Explicit original-caller legacy adoption preserves accepted evidence.
+
+## T02 — Prove isolation, retry policy and recovered-ledger continuity
+
+```task
+id: EMAIL-WP-0008-T02
+status: done
+priority: high
+```
+
+Synthetic tests cover cross-caller handle denial without mutation, scope changes,
+independent retries, permanent/temporary/unknown outcome, restart, concurrent
+budgets, backward clock, size/config failure and safe status. App owner gate
+binds exact tests to modules with no network or inherited secrets. Ledger
+fingerprint includes rate budgets; rehearsal must preserve scoped accepted and
+unknown evidence and quota rows without sending. Final owner gate:148passed, zero skips, including the loopback-only GreenMail
+harness. Network-denied app owner gate:66passed, zero skips. Ruff passes for
+receiver and caller module/tests. T03 remains waiting for runtime admission.
+
+## T03 — Admit original caller transition, runtime custody and recovery
+
+```task
+id: EMAIL-WP-0008-T03
+status: wait
+priority: high
+needs_human: true
+intervention_note: "Actual callers/scopes/quotas and separate token custody, platform sender/reply/bounce ownership, schema rollout/rollback and independent ledger recovery admission remain unapproved; source tests do not activate them."
+```
+
+Retain this live admission record before any completion. Choose real binding and
+operation sets/quotas, independent credential delivery and explicit legacy
+adoption. Review exact immutable image and additive schema migration/rollback.
+Repin guarded capture/rehearsal helpers together. EMAIL-WP-0007-T03/T04 and
+RPF-WP-0053-T05/T06 retain offsite archive/atomic credential pair custody.
+CSOC-WP-0010-T04/T05/T06 retain platform operations and productive/browser review.
+Review: docs/caller-admission-2026-10-08.md. No sender, caller, token, image,
+mailbox, production ledger, network admission, upload or timer is activated.