From 8de8fddf37caff5e21804ef26dc6eb6e336343bc Mon Sep 17 00:00:00 2001 From: tegwick Date: Mon, 19 Oct 2026 12:44:16 +0200 Subject: [PATCH] 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 Assistant: claude-code Assistant-Model: opus Assistant-Process: 5787@bnt-lap001 Assistant-Session: 5db26b82-138b-40cc-9856-4d5eb0f752fd --- deploy/railiance/entrypoint.sh | 7 +- docs/reuse-surface.md | 3 +- examples/ticket_service/ticket_service.py | 83 +++++----- mas-modes-v0.1-spec/CONTRACT.md | 12 +- mas_modes/email_mode.py | 27 +-- mas_modes/harness.py | 18 +- mas_modes/mailer.py | 155 +++++++++++++++--- tests/test_end_to_end.py | 18 +- tests/test_outbound.py | 114 +++++++++++++ workplans/MAS-WP-0012-single-outbound-path.md | 35 +++- 10 files changed, 371 insertions(+), 101 deletions(-) create mode 100644 tests/test_outbound.py diff --git a/deploy/railiance/entrypoint.sh b/deploy/railiance/entrypoint.sh index 33f6025..bba6c73 100644 --- a/deploy/railiance/entrypoint.sh +++ b/deploy/railiance/entrypoint.sh @@ -40,10 +40,13 @@ postconf -e \ # container already isolates Postfix, so run the daemons unchrooted. postconf -F '*/*/chroot = n' -# The ticket service sends replies only with a relay, capped at MAS_MAX_SENT. +# The ticket service sends replies only with a relay, capped at MAS_MAX_SENT in +# total; over the cap they are held, purged after 7d, at most 100MB (MAS-WP-0012). SMTP_ARG="" if [ -n "$MAS_RELAY" ] && [ -s /etc/mas-relay/username ]; then - SMTP_ARG="--smtp 127.0.0.1:25 --max-sent ${MAS_MAX_SENT:-300} --require-dmarc-pass" + SMTP_ARG="--smtp 127.0.0.1:25 --max-sent ${MAS_MAX_SENT:-300} --require-dmarc-pass \ + --email-held-duration-max ${MAS_EMAIL_HELD_DURATION_MAX:-7d} \ + --emails-held-storage-max ${MAS_EMAILS_HELD_STORAGE_MAX:-100MB}" fi # Milters in the foreground: daemonized, OpenDKIM died on its first diff --git a/docs/reuse-surface.md b/docs/reuse-surface.md index 228c4fc..91034d1 100644 --- a/docs/reuse-surface.md +++ b/docs/reuse-surface.md @@ -29,7 +29,8 @@ You then: - answer `case.assign` with your own case id; email adopts it as `TYPE=ID@domain`; - send replies with `mail.send` (`in_reply_to` a recorded message); - principle 10 can refuse them; + principle 10 can refuse them, and over the deployment's one send cap + they are held (`held: true`) instead of sent; - decide per case on `autonomous_send` and `transfers`: `block`, or a mode webhook with that mode's `peer`. The peer must be one the email operator has a key for; diff --git a/examples/ticket_service/ticket_service.py b/examples/ticket_service/ticket_service.py index 1988084..47f01d8 100644 --- a/examples/ticket_service/ticket_service.py +++ b/examples/ticket_service/ticket_service.py @@ -23,15 +23,14 @@ from __future__ import annotations import argparse import json import secrets -import smtplib import sqlite3 import threading -from email.message import EmailMessage from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path from typing import Callable, Optional from mas_core.generator import resource_local_part -from mas_modes.mailer import refusal +from mas_modes.mailer import Mailer, parse_duration, parse_size RESOURCE_TYPE = "ticket" INTAKE_TARGET = "ticket" @@ -56,27 +55,37 @@ CREATE TABLE IF NOT EXISTS messages ( seq INTEGER PRIMARY KEY AUTOINCREMENT, ticket_id TEXT, disposition TEXT NOT NULL, recipient TEXT, facet TEXT, sender TEXT, subject TEXT, message_id TEXT, body TEXT ); -CREATE TABLE IF NOT EXISTS outbox ( - seq INTEGER PRIMARY KEY AUTOINCREMENT, ticket_id TEXT, mail_from TEXT, mail_to TEXT, - subject TEXT, body TEXT, in_reply_to TEXT, sent INTEGER DEFAULT 0 -); """ +# Before the shared Mailer (MAS-WP-0012) replies lived in this table. It is +# read once at start so mail sent before keeps counting against the cap. +LEGACY_OUTBOX = "outbox" class TicketService: def __init__(self, db_path: str = ":memory:", id_factory: Callable[[], str] = default_ticket_id, smtp: Optional[tuple[str, int]] = None, closed_notice: bool = True, - max_sent: Optional[int] = None, require_dmarc_pass: bool = False): + max_sent: Optional[int] = None, require_dmarc_pass: bool = False, + mailer: Optional[Mailer] = None): + """Replies go through `mailer`, the deployment's one outbound path (principle + 10, one lifetime cap). Without one, the service makes its own from smtp, + max_sent and require_dmarc_pass.""" self.db = sqlite3.connect(db_path, check_same_thread=False) self.db.row_factory = sqlite3.Row self.db.executescript(SCHEMA) self.lock = threading.Lock() self.new_id = id_factory - self.smtp = smtp self.closed_notice = closed_notice - self.max_sent = max_sent # lifetime cap on mail actually sent via smtp - # spec 14 item 10: auto-reply only to authenticated senders - self.require_dmarc_pass = require_dmarc_pass + self.mailer = mailer or Mailer(smtp=smtp, max_sent=max_sent, + require_dmarc_pass=require_dmarc_pass) + self._import_legacy_outbox() + + def _import_legacy_outbox(self): + exists = self.db.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", + (LEGACY_OUTBOX,)).fetchone() + if exists: + rows = [{**dict(r), "case_ref": r["ticket_id"]} + for r in self.db.execute(f"SELECT * FROM {LEGACY_OUTBOX} ORDER BY seq")] + self.mailer.import_legacy(f"ticket-service:{LEGACY_OUTBOX}", rows) # -- event handling ------------------------------------------------------------- @@ -147,34 +156,12 @@ class TicketService: msg.get("subject"), msg.get("message_id"), msg.get("text"))) def _reply(self, tid, mail_from, msg, subject, body) -> bool: - to = msg.get("from") - if refusal(msg, self.require_dmarc_pass): - return False # automated or unauthenticated sender: keep the ticket, send nothing - if not subject.lower().startswith("re:"): - subject = "Re: " + subject - self.db.execute("INSERT INTO outbox (ticket_id, mail_from, mail_to, subject, body, in_reply_to)" - " VALUES (?,?,?,?,?,?)", (tid, mail_from, to, subject, body, msg.get("message_id"))) - seq = self.db.execute("SELECT last_insert_rowid()").fetchone()[0] - if self.smtp and not self.send_cap_reached(): - self._send(mail_from, to, subject, body, msg.get("message_id")) - self.db.execute("UPDATE outbox SET sent=1 WHERE seq=?", (seq,)) - return True + """True when the reply went out (or was recorded, without SMTP). False for + automated or unauthenticated senders, and over the cap (then it is held).""" + return self.mailer.reply(tid, mail_from, msg, subject, body) is None def send_cap_reached(self) -> bool: - """Over the cap, replies stay in the outbox unsent.""" - if self.max_sent is None: - return False - return self.db.execute("SELECT COUNT(*) FROM outbox WHERE sent=1").fetchone()[0] >= self.max_sent - - def _send(self, mail_from, to, subject, body, in_reply_to): - m = EmailMessage() - m["From"], m["To"], m["Subject"] = mail_from, to, subject - m["Auto-Submitted"] = "auto-replied" - if in_reply_to: - m["In-Reply-To"] = m["References"] = in_reply_to - m.set_content(body) - with smtplib.SMTP(*self.smtp) as s: - s.send_message(m) + return self.mailer.cap_reached() # -- MasterStore (mas_modes.email_mode): email as master for tickets ------------- @@ -212,7 +199,8 @@ class TicketService: return {**dict(row), "messages": [dict(m) for m in msgs]} def outbox(self) -> list[dict]: - return [dict(r) for r in self.db.execute("SELECT * FROM outbox ORDER BY seq")] + """This service's replies in the shared outbox (`ticket_id` = case ref).""" + return [{**r, "ticket_id": r["case_ref"]} for r in self.mailer.outbox()] def quarantine(self) -> list[dict]: return [dict(r) for r in self.db.execute( @@ -265,9 +253,20 @@ if __name__ == "__main__": ap.add_argument("--require-dmarc-pass", action="store_true", help="auto-reply only when the event says the sender passed DMARC") ap.add_argument("--max-sent", type=int, metavar="N", - help="stop sending once N replies have been sent in total (kept in the db)") + help="stop sending once N replies have been sent in total (kept in the outbox)") + ap.add_argument("--outbox-db", metavar="FILE", + help="the shared outbox (default: DB with '-outbox' added before the suffix)") + ap.add_argument("--email-held-duration-max", default="7d", metavar="DURATION", + help="purge mail held over the cap after this long (default 7d)") + ap.add_argument("--emails-held-storage-max", default="100MB", metavar="SIZE", + help="hold no more mail once held mail would exceed this (default 100MB)") a = ap.parse_args() smtp = (a.smtp.rsplit(":", 1)[0], int(a.smtp.rsplit(":", 1)[1])) if a.smtp else None + outbox_db = a.outbox_db or (a.db if a.db == ":memory:" else + str(Path(a.db).with_name(Path(a.db).stem + "-outbox" + Path(a.db).suffix))) + mailer = Mailer(outbox_db, smtp, a.max_sent, a.require_dmarc_pass, + held_max_age=parse_duration(a.email_held_duration_max), + held_max_bytes=parse_size(a.emails_held_storage_max)) h, p = a.listen.rsplit(":", 1) - print(f"ticket service on http://{a.listen}/mas-events") - make_server(TicketService(a.db, smtp=smtp, max_sent=a.max_sent, require_dmarc_pass=a.require_dmarc_pass), h, int(p)).serve_forever() + print(f"ticket service on http://{a.listen}/mas-events (outbox {outbox_db})") + make_server(TicketService(a.db, mailer=mailer), h, int(p)).serve_forever() diff --git a/mas-modes-v0.1-spec/CONTRACT.md b/mas-modes-v0.1-spec/CONTRACT.md index 20cde98..1dd9fc4 100644 --- a/mas-modes-v0.1-spec/CONTRACT.md +++ b/mas-modes-v0.1-spec/CONTRACT.md @@ -147,7 +147,8 @@ A receiver MUST refuse a transition that the table does not allow `mail.send` responses: `{"ok": true, "sent": true}`, or `{"ok": false, "error": "refused", "reason": R}` with R one of `unauthenticated-sender`, -`automated-sender`, `send-cap-reached`, `unknown-message`. +`automated-sender`, `send-cap-reached`, `unknown-message`. With +`send-cap-reached` the response also carries `held` (bool, §6). ## 6. Principle 10 in both roles @@ -158,6 +159,15 @@ sender. It MUST NOT send once its send cap is reached. These rules hold whoever asks for the send: email itself as master, the case's autonomous policy, or a master's `mail.send`. +The send cap is **one cap per deployment**. Every mail email sends, in +either role and from any application behind it, counts against the same +total. Over the cap, a mail is **held**: recorded unsent, and reported to +whoever asked as `send-cap-reached`. Held mail MUST be deleted after a +configured maximum age (default 7 days). It MUST NOT be stored once held +mail would exceed a configured maximum size (default 100 MB); the mail is +then reported as not held. Mail actually sent is never purged, because it +is what the cap counts. + ## 7. Peer authentication and transport security Contract messages steer real mail (`mail.send`, `case.assign`, diff --git a/mas_modes/email_mode.py b/mas_modes/email_mode.py index 26c0c85..86a8e81 100644 --- a/mas_modes/email_mode.py +++ b/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"]: diff --git a/mas_modes/harness.py b/mas_modes/harness.py index f2acab0..ef19b81 100644 --- a/mas_modes/harness.py +++ b/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.""" diff --git a/mas_modes/mailer.py b/mas_modes/mailer.py index 185fa08..1cf8ab5 100644 --- a/mas_modes/mailer.py +++ b/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() diff --git a/tests/test_end_to_end.py b/tests/test_end_to_end.py index 918cca2..074a3de 100644 --- a/tests/test_end_to_end.py +++ b/tests/test_end_to_end.py @@ -102,15 +102,17 @@ def test_lower_cased_id_correlates(env): assert svc.outbox()[-1]["mail_from"] == f"ticket=GS-41872@{D}" # replies use the issued id -def test_send_cap_keeps_replies_in_outbox(env): +def test_send_cap_holds_replies(env): svc, send, *_ = env sent = [] - svc.smtp, svc.max_sent = ("127.0.0.1", 25), 2 - svc._send = lambda *a: sent.append(a) + svc.mailer.smtp, svc.mailer.max_sent = ("127.0.0.1", 25), 2 + svc.mailer._send = lambda *a: sent.append(a) for n in range(3): send(f"ticket!callback@{D}", f"request {n}") assert len(sent) == 2 and svc.send_cap_reached() - assert [r["sent"] for r in svc.outbox()] == [1, 1, 0] # third recorded, not sent + out = svc.outbox() + assert [(r["sent"], r["held"]) for r in out] == [(1, None), (1, None), (0, "send-cap-reached")] + assert out[2]["ticket_id"] == out[2]["case_ref"] def test_only_declared_intake_creates(env): @@ -232,18 +234,18 @@ def test_forged_or_foreign_authentication_is_ignored(): def test_reply_only_to_authenticated_sender(env): svc, send, *_ = env - svc.require_dmarc_pass = True + svc.mailer.require_dmarc_pass = True sent = [] - svc.smtp, svc._send = ("127.0.0.1", 25), lambda *a: sent.append(a) + svc.mailer.smtp, svc.mailer._send = ("127.0.0.1", 25), lambda *a: sent.append(a) send(f"ticket!callback@{D}", "spoofed") # no verdict assert sent == [] and svc.outbox() == [] and svc.ticket("GS-41872") is not None def test_authenticated_sender_gets_reply(env): svc, _, _, resolver = env - svc.require_dmarc_pass = True + svc.mailer.require_dmarc_pass = True sent = [] - svc.smtp, svc._send = ("127.0.0.1", 25), lambda *a: sent.append(a) + svc.mailer.smtp, svc.mailer._send = ("127.0.0.1", 25), lambda *a: sent.append(a) event = build_event(f"ticket!callback@{D}", _meta(AR_OURS + RECEIVED), resolver) status, body = svc.handle(event) assert status == 201 and len(sent) == 1 and sent[0][1] == CUSTOMER diff --git a/tests/test_outbound.py b/tests/test_outbound.py new file mode 100644 index 0000000..0f25486 --- /dev/null +++ b/tests/test_outbound.py @@ -0,0 +1,114 @@ +"""One outbound path and one send cap (MAS-WP-0012).""" +import sqlite3 + +import pytest + +from mas_modes.harness import ROOT, Harness, reference_ticket_service +from mas_modes.mailer import Mailer, parse_duration, parse_size + +STATUTE = ROOT / "mas-modes-v0.1-spec/conformance/statute.json" +ENTRYPOINT = ROOT / "deploy/railiance/entrypoint.sh" +OK = {"from": "c@example.net", "authentication": {"dmarc": "pass"}, "subject": "s", "message_id": ""} + + +def mailer(**kw): + m = Mailer(smtp=("sink", 0), **kw) + m.sent = [] + m._send = lambda *mail: m.sent.append(mail) + return m + + +def test_both_roles_share_one_cap(): + h = Harness(STATUTE, {"ticket": {"role": "master"}, "case": {"role": "slave"}}, max_sent=3) + try: + for n in range(2): + assert h.mail("ticket", f"ticket {n}").status == 201 # 2 creation replies + h.mail("case", "case intake") + [(_, body)] = h.master.pump() + code, resp = h.master.send(body["case"], body["message_ref"], "on it") + assert (code, resp["sent"]) == (200, True) # third send: the cap + code, resp = h.master.send(body["case"], body["message_ref"], "again") + assert code == 409 and resp["reason"] == "send-cap-reached" and resp["held"] is True + assert h.mail("ticket", "ticket 3").status == 201 # ticket created, reply held + assert len(h.sent) == 3 # exactly N, across both roles + held = [m for m in h.mailer.outbox() if m["held"]] + assert [m["body"] for m in held][0] == "again" and len(held) == 2 + finally: + h.close() + + +def test_held_mail_is_purged_after_max_age(): + clock = [0.0] + m = mailer(max_sent=0, held_max_age=parse_duration("7d"), clock=lambda: clock[0]) + assert m.send("t1", "a@d", OK, "s", "held") == ("send-cap-reached", True) + clock[0] += 7 * 86400 - 1 + assert m.purge_held() == 0 + clock[0] += 2 + assert m.purge_held() == 1 and m.outbox() == [] + m.send("t1", "a@d", OK, "s", "new") # purging also runs on every send + clock[0] += 8 * 86400 + m.send("t1", "a@d", OK, "s", "newer") + assert [r["body"] for r in m.outbox()] == ["newer"] + + +def test_held_storage_limit(): + m = mailer(max_sent=0, held_max_bytes=parse_size("1KB")) + assert m.send("t", "a@d", OK, "s", "x" * 600)[1] is True + assert m.send("t", "a@d", OK, "s", "y" * 600) == ("send-cap-reached", False) # would exceed 1 KB + assert len(m.outbox()) == 1 and m.held_bytes() < 1024 + + +def test_sent_mail_is_never_purged(): + clock = [0.0] + m = mailer(max_sent=1, clock=lambda: clock[0]) + assert m.reply("t", "a@d", OK, "s", "sent") is None + clock[0] += 365 * 86400 + m.purge_held() + assert [(r["body"], r["sent"]) for r in m.outbox()] == [("sent", 1)] + assert m.cap_reached() + + +def test_legacy_outbox_counts_once(tmp_path): + db = tmp_path / "tickets.sqlite3" + con = sqlite3.connect(db) + con.execute("CREATE TABLE outbox (seq INTEGER PRIMARY KEY AUTOINCREMENT, ticket_id TEXT," + " mail_from TEXT, mail_to TEXT, subject TEXT, body TEXT, in_reply_to TEXT," + " sent INTEGER DEFAULT 0)") + con.executemany("INSERT INTO outbox (ticket_id, mail_from, mail_to, subject, body, in_reply_to, sent)" + " VALUES (?,?,?,?,?,?,?)", [("gs-1", "t@d", "c@x", "s", "b", None, 1)] * 2 + + [("gs-2", "t@d", "c@x", "s", "b", None, 0)]) + con.commit() + con.close() + out = tmp_path / "tickets-outbox.sqlite3" + TicketService = reference_ticket_service() + svc = TicketService(str(db), mailer=mailer(path=str(out), max_sent=3)) + assert svc.mailer.sent_count() == 2 and len(svc.outbox()) == 3 + svc = TicketService(str(db), mailer=mailer(path=str(out), max_sent=3)) # restart: no re-import + assert len(svc.outbox()) == 3 + assert svc.mailer.reply("gs-1", "t@d", OK, "s", "third") is None + assert svc.send_cap_reached() # 2 legacy + 1 new = the cap of 3 + + +@pytest.mark.parametrize("text, seconds", [("7d", 604800), ("12h", 43200), ("30m", 1800), ("45", 45)]) +def test_parse_duration(text, seconds): + assert parse_duration(text) == seconds + + +@pytest.mark.parametrize("text, size", [("100MB", 100 * 1024 ** 2), ("512KB", 512 * 1024), ("1gb", 1024 ** 3), + ("2048", 2048)]) +def test_parse_size(text, size): + assert parse_size(text) == size + + +def test_bad_config_values(): + with pytest.raises(ValueError): + parse_duration("seven days") + with pytest.raises(ValueError): + parse_size("100 MiB") + + +def test_entrypoint_passes_one_cap_and_hold_limits(): + text = ENTRYPOINT.read_text() + assert "--max-sent ${MAS_MAX_SENT:-300}" in text + assert "--email-held-duration-max ${MAS_EMAIL_HELD_DURATION_MAX:-7d}" in text + assert "--emails-held-storage-max ${MAS_EMAILS_HELD_STORAGE_MAX:-100MB}" in text diff --git a/workplans/MAS-WP-0012-single-outbound-path.md b/workplans/MAS-WP-0012-single-outbound-path.md index 23f0f11..3e0986a 100644 --- a/workplans/MAS-WP-0012-single-outbound-path.md +++ b/workplans/MAS-WP-0012-single-outbound-path.md @@ -4,7 +4,7 @@ type: workplan title: "One outbound path and one send cap" domain: communication repo: mail-address-semantics -status: proposed +status: finished owner: claude topic_slug: coulomb-social created: "2026-10-19" @@ -28,7 +28,7 @@ total". ```task id: MAS-WP-0012-T01 -status: todo +status: done priority: medium state_hub_task_id: "5d9fefac-684d-5fe0-bead-f664952047e0" ``` @@ -44,7 +44,7 @@ semantics. ```task id: MAS-WP-0012-T02 -status: todo +status: done priority: medium state_hub_task_id: "431cc528-cf9f-5991-8044-7cbab2ac0021" ``` @@ -52,3 +52,32 @@ state_hub_task_id: "431cc528-cf9f-5991-8044-7cbab2ac0021" Add a test that a deployment with both roles stops at exactly N sends in total, and that the live entrypoint still passes `MAS_MAX_SENT` to the single cap. + +## Outcome + +- **Founder interview (2026-10-19).** Over the cap, mail is **held and + reported**. Held mail is purged after `email-held-duration-max` + (default 7d) and not stored beyond `emails-held-storage-max` (default + 100MB). The live counter is **migrated once at start**. +- **`mas_modes.mailer.Mailer`** is the one outbound path: its own SQLite + file and lock, principle 10, one lifetime cap, holding with a purge on + every send and `purge_held()`, a storage limit, `import_legacy()` with a + once-only marker, and `parse_duration` / `parse_size`. +- **Ticket service.** It sends through a Mailer it is given, or builds one. + `/outbox` shows the shared outbox (`ticket_id` = case ref). Its legacy + `outbox` table is imported once and then left alone. New CLI options: + `--outbox-db` (default `-outbox.sqlite3`), + `--email-held-duration-max` and `--emails-held-storage-max`. Over the + cap the closed notice reports `notified: false`. +- **EmailMode** takes the shared Mailer. `mail.send` over the cap answers + `send-cap-reached` with `held`. CONTRACT.md §5 and §6 are updated. +- **Deploy.** The entrypoint passes `MAS_MAX_SENT` (300) and the two + hold limits (`MAS_EMAIL_HELD_DURATION_MAX`, + `MAS_EMAILS_HELD_STORAGE_MAX`). The migration was checked on a copy of + the live DB: 9 legacy rows, 4 sent, carried over; no second import on + restart. +- **Tests.** `tests/test_outbound.py` (15): exactly N sends across both + roles, purge by age, storage limit, sent mail never purged, the legacy + migration, parsers, entrypoint flags. +- **Not deployed.** The live MX still runs `v0.2-a95f848`; this takes + effect on the next redeploy.