MAS-WP-0012 finished: one outbound path and one send cap

mas_modes.mailer.Mailer is the only outbound path (own SQLite + lock),
shared by the ticket service and both EmailMode roles. Over the cap mail
is held and reported (send-cap-reached, held); held mail is purged after
email-held-duration-max (7d) and not stored beyond emails-held-storage-max
(100MB). The ticket service's legacy outbox is imported once so the live
count carries over (checked on a copy of the live DB: 4 sent). Entrypoint
passes the cap and both limits.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 5787@bnt-lap001
Assistant-Session: 5db26b82-138b-40cc-9856-4d5eb0f752fd

mas_modes/email_mode.py

@@ -13,9 +13,12 @@ Takes normalized gateway events (POST /mas-events), mode statuses
 Every peer is authenticated (CONTRACT.md section 7): requests in and out
 are signed with pairwise keys, master commands are accepted only from the
 master configured for the resource type, and mode statuses only at the
-handoff's own tokenized URL, from the peer it was sent to. Every outbound
-mail goes through principle 10 (`mailer.refusal`), whoever asked for it. Outbound contract messages are queued while the lock is
-held and sent after it is released, so peers may call back at once.
+handoff's own tokenized URL, from the peer it was sent to.
+
+Every outbound mail goes through one shared `Mailer` (principle 10 and the
+send cap), whoever asked for it. Outbound contract messages are queued while
+the lock is held and sent after it is released, so peers may call back at
+once.
 """
 
 from __future__ import annotations
@@ -111,11 +114,11 @@ class EmailMode:
                  store: Optional[MasterStore] = None, peer_id: str = "email",
                  event_peer: Optional[str] = None, db_path: str = ":memory:",
                  statutes: Optional[dict[str, Statute]] = None,
-                 clock: Callable[[], float] = time.time, smtp=None, max_sent=None,
-                 require_dmarc_pass: bool = True):
+                 clock: Callable[[], float] = time.time, mailer: Optional[Mailer] = None):
         """keys: peer id -> key shared with that peer. event_peer: the gateway's
         peer id when events must be signed; None accepts unsigned events from
-        loopback only."""
+        loopback only. mailer: the deployment's one outbound path, shared with
+        the master store; by default a private in-memory one."""
         if any(c.role == "master" for c in roles.values()) and store is None:
             raise ValueError("a master role needs a MasterStore")
         for t, c in roles.items():
@@ -135,7 +138,7 @@ class EmailMode:
         self.db = sqlite3.connect(db_path, check_same_thread=False)
         self.db.row_factory = sqlite3.Row
         self.db.executescript(SCHEMA)
-        self.mailer = Mailer(self.db, smtp, max_sent, require_dmarc_pass)
+        self.mailer = mailer or Mailer()
         self.lock = threading.RLock()
         self.base_url = ""            # set by serve(); status URLs are base_url/mode-status/TOKEN
         self._queue: list = []
@@ -439,10 +442,12 @@ class EmailMode:
             if m is None:
                 return refuse("refused", msg["in_reply_to"], reason="unknown-message")
             orig = json.loads(m["message"])
-            why = self.mailer.reply(row["key"], row["address"], orig,
-                                    msg.get("subject") or orig.get("subject") or "Your request",
-                                    msg["body"])
-            return (200, {"ok": True, "sent": True}) if why is None else refuse("refused", why, reason=why)
+            why, held = self.mailer.send(row["key"], row["address"], orig,
+                                         msg.get("subject") or orig.get("subject") or "Your request",
+                                         msg["body"])
+            if why is None:
+                return 200, {"ok": True, "sent": True}
+            return refuse("refused", why, reason=why, **({"held": held} if why == "send-cap-reached" else {}))
         if t == "handoff.cancel":
             h = self._handoff(msg["handoff_id"])
             if h is None or h["case_key"] != row["key"]:

mas_modes/harness.py

@@ -19,6 +19,7 @@ from mas_gateway.core import Resolver, gateway, message_meta_from_rfc822
 
 from .contract import serve
 from .email_mode import EmailMode, RoleConfig
+from .mailer import Mailer
 from .stubs.callback import CallbackStub
 from .stubs.master import StubMaster
 
@@ -71,15 +72,21 @@ class Harness:
         defaults = {"master": {"modes": {"callback": {"url": "{callback}", "peer": "callback-1"}}},
                     "slave": {"master_url": "{master}", "master_peer": "cms-1"}}
         configs = {t: RoleConfig(**fill({**defaults[c["role"]], **c})) for t, c in roles.items()}
+        # One outbound path for both roles. SMTP is a sink when a cap is under test,
+        # since only mail actually sent counts against it.
+        self.sent: list = []
+        self.mailer = Mailer(smtp=("sink", 0) if max_sent is not None else None, max_sent=max_sent,
+                             clock=self.clock)
+        self.mailer._send = lambda *mail: self.sent.append(mail)
         self.tickets = None
         if any(c.role == "master" for c in configs.values()):
             kw = {"id_factory": (lambda it=iter(ticket_ids): next(it))} if ticket_ids else {}
-            self.tickets = reference_ticket_service()(require_dmarc_pass=True, max_sent=max_sent, **kw)
+            self.tickets = reference_ticket_service()(mailer=self.mailer, **kw)
         self.signer = ("gateway", self.keys["gateway"]) if sign_events else None
         self.email = EmailMode(configs, dict(self.keys), self.tickets,
                                event_peer="gateway" if sign_events else None,
                                statutes=dict(self.resolver.pinned), clock=self.clock,
-                               max_sent=max_sent)
+                               mailer=self.mailer)
         self.email.base_url = self._serve(self.email.server())
         self.master.email_url = self.email.base_url + "/master"
         self.app_url = self.email.base_url + "/mas-events"
@@ -111,11 +118,8 @@ class Harness:
         return self.email.tick()
 
     def outbox(self) -> list[dict]:
-        """Every mail email decided to send, in either role."""
-        mails = self.email.mailer.outbox()
-        if self.tickets is not None:
-            mails = self.tickets.outbox() + mails
-        return mails
+        """Every mail email decided to send, in either role (held mail excluded)."""
+        return [m for m in self.mailer.outbox() if m["held"] is None]
 
     def thread(self, ticket_id: str) -> list[str]:
         """Mode entries recorded in a master-role ticket thread."""

mas_modes/mailer.py

@@ -1,24 +1,56 @@
-"""Outbound mail under principle 10 (CONTRACT.md section 6).
+"""The one outbound path for MAS mail, under principle 10 (CONTRACT.md section 6).
 
 Every send is a reply to a recorded inbound message: the recipient is that
 message's sender, and the message's own sender authentication decides
-whether email may answer at all.
+whether email may answer at all. All senders in a deployment (the ticket
+service, both roles of the email side) share one Mailer, so one outbox and
+one lifetime cap bound everything sent.
+
+Over the cap a reply is *held*: recorded unsent, and reported as
+`send-cap-reached`. Held mail is purged after `held_max_age` and is not
+stored at all once held mail would exceed `held_max_bytes`.
 """
 
 from __future__ import annotations
 
+import re
 import smtplib
 import sqlite3
+import threading
+import time
 from email.message import EmailMessage
-from typing import Optional
+from typing import Callable, Optional
 
 OUTBOX = """
 CREATE TABLE IF NOT EXISTS mode_outbox (
   seq INTEGER PRIMARY KEY AUTOINCREMENT, case_ref TEXT, mail_from TEXT, mail_to TEXT,
-  subject TEXT, body TEXT, in_reply_to TEXT, sent INTEGER DEFAULT 0
+  subject TEXT, body TEXT, in_reply_to TEXT, sent INTEGER DEFAULT 0, held TEXT, created REAL
 );
+CREATE TABLE IF NOT EXISTS mailer_meta (key TEXT PRIMARY KEY, value TEXT);
 """
 
+DEFAULT_HELD_MAX_AGE = 7 * 86400          # email-held-duration-max: 7d
+DEFAULT_HELD_MAX_BYTES = 100 * 1024 ** 2  # emails-held-storage-max: 100MB
+
+_UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800}
+_SIZES = {"": 1, "b": 1, "kb": 1024, "mb": 1024 ** 2, "gb": 1024 ** 3}
+
+
+def parse_duration(text: str) -> float:
+    """'7d', '12h', '30m', '45s' or plain seconds."""
+    m = re.fullmatch(r"\s*(\d+(?:\.\d+)?)\s*([smhdw]?)\s*", str(text).lower())
+    if not m:
+        raise ValueError(f"not a duration: {text!r}")
+    return float(m.group(1)) * _UNITS[m.group(2) or "s"]
+
+
+def parse_size(text: str) -> int:
+    """'100MB', '512KB', '1GB' or plain bytes (binary units)."""
+    m = re.fullmatch(r"\s*(\d+)\s*([kmg]?b?)\s*", str(text).lower())
+    if not m:
+        raise ValueError(f"not a size: {text!r}")
+    return int(m.group(1)) * _SIZES[m.group(2)]
+
 
 def refusal(msg: dict, require_dmarc_pass: bool = True) -> Optional[str]:
     """Why email must not answer `msg`, or None if it may."""
@@ -31,45 +63,116 @@ def refusal(msg: dict, require_dmarc_pass: bool = True) -> Optional[str]:
     return None
 
 
-class Mailer:
-    """Gated replies, recorded in an outbox; sent over SMTP when configured,
-    up to a lifetime cap counted in the outbox."""
+def _size(*parts) -> int:
+    return sum(len((p or "").encode()) for p in parts)
 
-    def __init__(self, db: sqlite3.Connection, smtp: Optional[tuple[str, int]] = None,
-                 max_sent: Optional[int] = None, require_dmarc_pass: bool = True):
-        self.db = db
+
+class Mailer:
+    """Gated replies in one outbox (its own SQLite file and lock); sent over
+    SMTP when configured, up to a lifetime cap counted in the outbox.
+    Without SMTP, replies are only recorded (and never count)."""
+
+    def __init__(self, path: str = ":memory:", smtp: Optional[tuple[str, int]] = None,
+                 max_sent: Optional[int] = None, require_dmarc_pass: bool = True,
+                 held_max_age: float = DEFAULT_HELD_MAX_AGE,
+                 held_max_bytes: int = DEFAULT_HELD_MAX_BYTES,
+                 clock: Callable[[], float] = time.time):
+        self.db = sqlite3.connect(path, check_same_thread=False)
+        self.db.row_factory = sqlite3.Row
         self.db.executescript(OUTBOX)
+        self.lock = threading.RLock()
         self.smtp = smtp
         self.max_sent = max_sent
         self.require_dmarc_pass = require_dmarc_pass
+        self.held_max_age = held_max_age
+        self.held_max_bytes = held_max_bytes
+        self.clock = clock
 
     def reply(self, case_ref: str, mail_from: str, msg: dict, subject: str,
               body: str) -> Optional[str]:
-        """Answer `msg` from `mail_from`. Returns None when accepted, else the reason."""
+        """Answer `msg` from `mail_from`. None when sent (or recorded, without
+        SMTP); else the reason it was not sent."""
+        return self.send(case_ref, mail_from, msg, subject, body)[0]
+
+    def send(self, case_ref: str, mail_from: str, msg: dict, subject: str,
+             body: str) -> tuple[Optional[str], bool]:
+        """Like reply(), plus whether the mail was held: over the cap it is
+        recorded unsent if held storage allows."""
         why = refusal(msg, self.require_dmarc_pass)
         if why:
-            return why
-        if self.cap_reached():
-            return "send-cap-reached"
+            return why, False
         if not subject.lower().startswith("re:"):
             subject = "Re: " + subject
-        cur = self.db.execute(
-            "INSERT INTO mode_outbox (case_ref, mail_from, mail_to, subject, body, in_reply_to)"
-            " VALUES (?,?,?,?,?,?)", (case_ref, mail_from, msg["from"], subject, body,
-                                      msg.get("message_id")))
-        if self.smtp:
-            self._send(mail_from, msg["from"], subject, body, msg.get("message_id"))
-            self.db.execute("UPDATE mode_outbox SET sent=1 WHERE seq=?", (cur.lastrowid,))
-        return None
+        row = (case_ref, mail_from, msg["from"], subject, body, msg.get("message_id"))
+        with self.lock, self.db:
+            self._purge()
+            if self.smtp and self._cap_reached():
+                return "send-cap-reached", self._hold(row)
+            cur = self.db.execute(
+                "INSERT INTO mode_outbox (case_ref, mail_from, mail_to, subject, body, in_reply_to,"
+                " created) VALUES (?,?,?,?,?,?,?)", (*row, self.clock()))
+            if self.smtp:
+                self._send(mail_from, msg["from"], subject, body, msg.get("message_id"))
+                self.db.execute("UPDATE mode_outbox SET sent=1 WHERE seq=?", (cur.lastrowid,))
+        return None, False
 
-    def cap_reached(self) -> bool:
+    def _hold(self, row) -> bool:
+        if self.held_bytes() + _size(*row) > self.held_max_bytes:
+            return False
+        self.db.execute("INSERT INTO mode_outbox (case_ref, mail_from, mail_to, subject, body,"
+                        " in_reply_to, held, created) VALUES (?,?,?,?,?,?,?,?)",
+                        (*row, "send-cap-reached", self.clock()))
+        return True
+
+    def _cap_reached(self) -> bool:
         if self.max_sent is None:
             return False
-        n = self.db.execute("SELECT COUNT(*) FROM mode_outbox WHERE sent=1").fetchone()[0]
-        return n >= self.max_sent
+        return self.sent_count() >= self.max_sent
+
+    def cap_reached(self) -> bool:
+        with self.lock:
+            return bool(self.smtp) and self._cap_reached()
+
+    def sent_count(self) -> int:
+        return self.db.execute("SELECT COUNT(*) FROM mode_outbox WHERE sent=1").fetchone()[0]
+
+    def held_bytes(self) -> int:
+        row = self.db.execute(
+            "SELECT COALESCE(SUM(LENGTH(CAST(COALESCE(case_ref,'') || COALESCE(mail_from,'') ||"
+            " COALESCE(mail_to,'') || COALESCE(subject,'') || COALESCE(body,'') ||"
+            " COALESCE(in_reply_to,'') AS BLOB))), 0) FROM mode_outbox WHERE held IS NOT NULL").fetchone()
+        return row[0]
+
+    def purge_held(self) -> int:
+        """Delete held mail older than held_max_age. Returns how many."""
+        with self.lock, self.db:
+            return self._purge()
+
+    def _purge(self) -> int:
+        cur = self.db.execute("DELETE FROM mode_outbox WHERE held IS NOT NULL AND created < ?",
+                              (self.clock() - self.held_max_age,))
+        return cur.rowcount
 
     def outbox(self) -> list[dict]:
-        return [dict(r) for r in self.db.execute("SELECT * FROM mode_outbox ORDER BY seq")]
+        with self.lock:
+            return [dict(r) for r in self.db.execute("SELECT * FROM mode_outbox ORDER BY seq")]
+
+    def import_legacy(self, source: str, rows: list[dict]) -> int:
+        """Copy an older outbox once (marker `source`), so mail sent before keeps
+        counting against the cap. Rows need mail_from, mail_to, subject, body,
+        in_reply_to, sent and a case reference under `case_ref`."""
+        with self.lock, self.db:
+            key = f"imported:{source}"
+            if self.db.execute("SELECT 1 FROM mailer_meta WHERE key=?", (key,)).fetchone():
+                return 0
+            for r in rows:
+                self.db.execute(
+                    "INSERT INTO mode_outbox (case_ref, mail_from, mail_to, subject, body, in_reply_to,"
+                    " sent, created) VALUES (?,?,?,?,?,?,?,?)",
+                    (r["case_ref"], r["mail_from"], r["mail_to"], r["subject"], r["body"],
+                     r["in_reply_to"], r["sent"], self.clock()))
+            self.db.execute("INSERT INTO mailer_meta VALUES (?, ?)", (key, str(len(rows))))
+            return len(rows)
 
     def _send(self, mail_from, to, subject, body, in_reply_to):
         m = EmailMessage()