Prepare definitive SMTP rejection proof and builder mail handoff

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a198ed-70cc-7362-a569-d5dbef39b86a

scripts/mail_return_operation.py

@@ -233,6 +233,127 @@ def connect(path, *, writable=False):
     return db
 
 
+def correlate_rejection(db, recipient, username, document):
+    """Bind a definitive SMTP failure to its original durable reservation.
+
+    This is a separate authority from an accepted-message DSN. Unknown outcomes,
+    retryable failures and caller-supplied rejection claims never qualify.
+    """
+    caller = selected_policy(document, "user-engine")
+    rows = db.execute(
+        "SELECT v.request_id,v.registration_id,v.email,v.username,v.client_id,"
+        "v.tenant,v.display_name,v.caller_id,v.canceled_at,v.consumed_at,"
+        "a.operation_key,a.fingerprint,a.state,a.error,a.retryable,a.provider_ref "
+        "FROM verifications v JOIN verification_attempts a ON a.request_id=v.request_id "
+        "WHERE v.email=? AND v.username=?",
+        (recipient.lower(), username),
+    ).fetchall()
+    require(len(rows) == 1, "one_recorded_rejection_required")
+    (
+        request_id,
+        registration,
+        email,
+        recorded_username,
+        client,
+        tenant,
+        display,
+        recorded_caller,
+        canceled,
+        consumed,
+        key,
+        fingerprint,
+        state,
+        error,
+        retryable,
+        reference,
+    ) = rows[0]
+    require(
+        state == "failed"
+        and error == "permanent_rejection"
+        and retryable == 0
+        and reference is None
+        and canceled is not None
+        and consumed is None,
+        "definitive_permanent_rejection_required",
+    )
+    require(
+        recorded_caller == "user-engine"
+        or (recorded_caller == "" and caller.get("adopt_legacy_verifications") is True),
+        "recorded_rejection_caller_denied",
+    )
+    require(binding_allowed(caller, tenant, client), "caller_binding_denied")
+    payload = {
+        "registration_id": registration,
+        "normalized_email": email,
+        "preferred_username": recorded_username,
+        "client_id": client,
+        "tenant": tenant,
+        "display_name": display,
+    }
+    identity = [tenant, client, registration]
+    if recorded_caller:
+        identity.insert(0, recorded_caller)
+    require(
+        digest(payload) == fingerprint
+        and hashlib.sha256(json.dumps(identity).encode()).hexdigest() == key,
+        "rejection_reservation_binding_changed",
+    )
+    binding = {
+        "kind": "verification-rejection",
+        "id": request_id,
+        "caller_id": "user-engine",
+        "tenant": tenant,
+        "client_id": client,
+        "recipient": email.lower(),
+        "username": recorded_username,
+        "provider_ref": None,
+    }
+    evidence = {
+        "operation_key": key,
+        "fingerprint": fingerprint,
+        "state": state,
+        "error": error,
+        "retryable": False,
+        "provider_ref": None,
+        "canceled_at": canceled,
+    }
+    return binding, evidence
+
+
+def prepare_rejection(path, recipient, username, document, review, *, now=None):
+    require(
+        isinstance(recipient, str)
+        and 3 <= len(recipient) <= 320
+        and isinstance(username, str)
+        and 1 <= len(username) <= 128,
+        "selected_rejection_identity_required",
+    )
+    require(
+        isinstance(review, str) and 10 <= len(review.strip()) <= 500,
+        "explicit_operator_provenance_review_required",
+    )
+    with closing(connect(path)) as db:
+        db.execute("BEGIN")
+        binding, evidence = correlate_rejection(db, recipient, username, document)
+        before = suppression(db, binding["recipient"])
+    require(before is None, "existing_suppression_requires_review")
+    created = timestamp(now or datetime.now(UTC))
+    return {
+        "schema": SCHEMA,
+        "action": "suppress",
+        "created_at": created,
+        "binding": binding,
+        "policy_sha256": digest(document),
+        "evidence_kind": "permanent-smtp-rejection",
+        "failure_evidence": evidence,
+        "evidence_sha256": digest(evidence),
+        "operator_review": review,
+        "before": None,
+        "after": [binding["recipient"], "hard_bounce", created],
+        "scope": "recipient-global across callers and tenants",
+    }
+
+
 def suppression(db, recipient):
     row = db.execute(
         "SELECT email,reason,created_at FROM suppressions WHERE email=?", (recipient,)
@@ -319,9 +440,7 @@ def list_suppressions(path, audit_root=None):
     }
 
 
-def prepare_untracked_release(
-    path, audit_root, email, reason, created_at, review, *, now=None
-):
+def prepare_untracked_release(path, audit_root, email, reason, created_at, review, *, now=None):
     """Release one ledger row that no suppress operation accounts for.
 
     A row created by this operator keeps its linked release. This path exists
@@ -488,13 +607,22 @@ def apply(path, plan, raw, document, expected, audit_root):
     private_directory(audit_root)
     if plan["action"] == "suppress":
         require(digest(document) == plan["policy_sha256"], "caller_policy_changed")
-        require(
-            hashlib.sha256(raw).hexdigest() == plan["evidence_sha256"], "selected_evidence_changed"
-        )
-        require(
-            permanent_dsn(raw, plan["binding"]["recipient"]) == plan["binding"]["provider_ref"],
-            "dsn_correlation_changed",
-        )
+        if plan.get("evidence_kind") == "permanent-smtp-rejection":
+            require(raw is None, "rejection_has_no_message")
+            require(
+                digest(plan["failure_evidence"]) == plan["evidence_sha256"],
+                "rejection_evidence_changed",
+            )
+        else:
+            require(plan.get("evidence_kind") is None, "evidence_kind_unknown")
+            require(
+                hashlib.sha256(raw).hexdigest() == plan["evidence_sha256"],
+                "selected_evidence_changed",
+            )
+            require(
+                permanent_dsn(raw, plan["binding"]["recipient"]) == plan["binding"]["provider_ref"],
+                "dsn_correlation_changed",
+            )
     elif plan["action"] == "release":
         original = completed_operation(
             audit_root, plan["source_plan_sha256"], "suppression_applied"
@@ -529,21 +657,28 @@ def apply(path, plan, raw, document, expected, audit_root):
 
 def commit(path, plan, raw, document, expected, directory):
     durable(directory / "plan.json", canonical(plan) + b"\n")
-    if plan["action"] == "suppress":
+    if plan["action"] == "suppress" and plan.get("evidence_kind") is None:
         durable(directory / "message.eml", raw)
     with closing(connect(path, writable=True)) as db:
         try:
             db.execute("BEGIN IMMEDIATE")
             recipient = plan["binding"]["recipient"]
+            preserved = protected_ledger_digest(db, recipient)
             if plan["action"] == "suppress":
-                bound = correlate(
-                    db,
-                    plan["binding"]["provider_ref"],
-                    recipient,
-                    plan["binding"]["caller_id"],
-                    document,
-                    plan["invitation_payload"],
-                )
+                if plan.get("evidence_kind") == "permanent-smtp-rejection":
+                    bound, evidence = correlate_rejection(
+                        db, recipient, plan["binding"]["username"], document
+                    )
+                    require(evidence == plan["failure_evidence"], "rejection_record_changed")
+                else:
+                    bound = correlate(
+                        db,
+                        plan["binding"]["provider_ref"],
+                        recipient,
+                        plan["binding"]["caller_id"],
+                        document,
+                        plan["invitation_payload"],
+                    )
                 require(bound == plan["binding"], "accepted_outbound_changed")
                 require(
                     plan["before"] is None
@@ -573,6 +708,10 @@ def commit(path, plan, raw, document, expected, directory):
                 )
             else:
                 db.execute("DELETE FROM suppressions WHERE email=?", (recipient,))
+            require(
+                protected_ledger_digest(db, recipient) == preserved,
+                "unrelated_ledger_state_changed",
+            )
             db.commit()
         except BaseException:
             db.rollback()
@@ -585,10 +724,39 @@ def commit(path, plan, raw, document, expected, directory):
                 "plan_sha256": expected,
                 "status": status,
                 "completed_at": timestamp(datetime.now(UTC)),
+                "unrelated_ledger_rows_preserved": True,
+                "protected_ledger_sha256": preserved,
             }
         ),
     )
-    return {"status": status, "plan_sha256": expected, "scope": plan["scope"], "mail_sent": False}
+    return {
+        "status": status,
+        "plan_sha256": expected,
+        "scope": plan["scope"],
+        "mail_sent": False,
+        "unrelated_ledger_rows_preserved": True,
+        "protected_ledger_sha256": preserved,
+    }
+
+
+def protected_ledger_digest(db, selected_recipient):
+    state = {}
+    for table in (
+        "deliveries",
+        "send_attempts",
+        "verifications",
+        "verification_attempts",
+        "caller_buckets",
+        "suppressions",
+    ):
+        query = 'SELECT * FROM "' + table + '"'
+        rows = (
+            db.execute(query + " WHERE email<>?", (selected_recipient,))
+            if table == "suppressions"
+            else db.execute(query)
+        )
+        state[table] = sorted(canonical(row).hex() for row in rows)
+    return digest(state)
 
 
 def inspect(path, directory):