Short answer: choose a messaging API by testing whether one app can keep 2FA login SMS OTP traffic separate from subscription renewal notices and an auditable logistics report email. For the report attachment, persist approved bytes before dispatch, give every resend a new attempt ID, and permit cancel only while that attempt is still queued.
The deciding constraint is compliance evidence. A delivery result alone cannot show which generated report was approved, which attachment bytes were used, or whether cancellation won a race with dispatch. Those questions need an application-owned record even when a Node.js builder, Python worker, or managed transport handles the final send.
A direct send_email(report) call is appealing in a notebook. It becomes ambiguous in production: after a timeout, a blind retry can create another message with the same attachment. The useful experiment is not “did one email arrive?” It is “can every outcome be reconstructed without guessing?”
What should an API prove for 2FA login SMS OTP retries?
A logistics report has three separate identities: the business report, its exact bytes, and a delivery attempt. Keep them separate. A corrected attachment can belong to the same shipment review, but it must have a new content digest. An intentional resend of unchanged bytes retains the report and digest while receiving a fresh attempt ID. Imagine that SHIP-1042 is approved at 09:00, queued at 09:01, and canceled while a worker is claiming it. If the only stored value is sent=true, the record cannot explain which operation won. If the report row is overwritten during a later correction, it cannot even prove which PDF the worker loaded. Stable report and attempt identities turn both questions into lookups. The same principle applies to an OTP resend: a new attempt may represent a deliberate user action, while a repeated command with the same idempotency key is just the same action arriving twice. The policies differ, but the identity problem does not.
That distinction is the test.
SMS creates a useful contrast. A 2FA login OTP is short-lived authentication traffic, while subscription renewal notices are lifecycle communication. They should not share retry policy merely because both use the same API. The email attachment has a third policy because its evidence must connect approval, recipient resolution, exact bytes, and dispatch.
DKIM addresses another layer. RFC 6376 defines a domain-level message authentication framework using cryptographic signatures. It also says a valid signature does not assert that content is safe or desirable. Record a DKIM result when the mail path provides one, but do not treat that result as proof that an operator approved the right PDF.
My notebook-to-production rule is to make the evidence model executable before connecting a transport. This Python example persists an immutable artifact and appends hash-linked events. The chain can reveal an edited or reordered local stream; it is not tamper-proof storage. That limitation matters.
from __future__ import annotations
import hashlib
import json
import os
import tempfile
import uuid
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
def digest(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
@dataclass(frozen=True)
class ReportArtifact:
report_id: str
shipment_id: str
sha256: str
path: str
class EvidenceLog:
def __init__(self, path: Path) -> None:
self.path = path
def append(self, event: dict[str, Any]) -> None:
previous = "0" * 64
if self.path.exists():
last = self.path.read_text(encoding="utf-8").splitlines()[-1]
previous = json.loads(last)["event_hash"]
record = {
"at": datetime.now(timezone.utc).isoformat(),
"previous_hash": previous,
**event,
}
encoded = json.dumps(record, sort_keys=True, separators=(",", ":"))
record["event_hash"] = digest(encoded.encode())
with self.path.open("a", encoding="utf-8") as stream:
stream.write(json.dumps(record, sort_keys=True) + "\n")
stream.flush()
os.fsync(stream.fileno())
def persist_report(root: Path, shipment_id: str, pdf: bytes) -> ReportArtifact:
report_id = str(uuid.uuid4())
artifact = ReportArtifact(
report_id=report_id,
shipment_id=shipment_id,
sha256=digest(pdf),
path=str(root / f"{report_id}.pdf"),
)
Path(artifact.path).write_bytes(pdf)
return artifact
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
evidence = EvidenceLog(root / "evidence.jsonl")
report = persist_report(root, "SHIP-1042", b"approved report bytes")
evidence.append({"type": "report.approved", "report": asdict(report)})
print(report.sha256)
Access control, retention, independent copies, and review remain deployment decisions. A digest detects changed bytes; it does not explain who was authorized to approve them.
Where do duplicate delivery attempts begin?
Cancellation is meaningful only before a worker crosses the dispatch boundary. Once a transport accepts a message, the application should record that fact instead of relabeling history as canceled. The recipient may already have it.
No retroactive cancel.
Use an idempotency key for the user command and a different identifier for each attempt. Repeating the same command returns the existing attempt. An explicit resend creates another attempt pointing to the same approved artifact. This separates an accidental client retry from an intentional second delivery.
from dataclasses import dataclass, field
from enum import Enum
from uuid import uuid4
class Status(str, Enum):
QUEUED = "queued"
CANCELED = "canceled"
DISPATCHED = "dispatched"
@dataclass
class Attempt:
attempt_id: str
report_id: str
command_key: str
status: Status = Status.QUEUED
transitions: list[str] = field(default_factory=lambda: ["queued"])
class Outbox:
def __init__(self) -> None:
self.by_command: dict[str, Attempt] = {}
def enqueue(self, report_id: str, command_key: str) -> Attempt:
if command_key in self.by_command:
return self.by_command[command_key]
attempt = Attempt(str(uuid4()), report_id, command_key)
self.by_command[command_key] = attempt
return attempt
def cancel(self, attempt: Attempt) -> bool:
if attempt.status is not Status.QUEUED:
return False
attempt.status = Status.CANCELED
attempt.transitions.append("canceled")
return True
def claim(self, attempt: Attempt) -> bool:
if attempt.status is not Status.QUEUED:
return False
attempt.status = Status.DISPATCHED
attempt.transitions.append("dispatched")
return True
outbox = Outbox()
first = outbox.enqueue("REPORT-7", "command-31")
same = outbox.enqueue("REPORT-7", "command-31")
assert first.attempt_id == same.attempt_id
assert outbox.cancel(first) is True
assert outbox.claim(first) is False
A production database must make cancel and claim conditional updates in one transaction. The in-memory version exposes the invariant, not a storage recipe. Two workers racing on a plain Python object are outside its scope.
Keep generated instructions away from approved bytes
The worker should receive identifiers, not free-form generated instructions. It loads the approved recipient and artifact, verifies the digest again, builds the message, and hands it to a generic interface. An AI workflow may generate the report, but the model should not choose an arbitrary file path, recipient, or MIME type during dispatch.
Tool definitions make this boundary concrete. The tool-use guide in References describes detailed tool descriptions and JSON Schema inputs. Expose a narrow request tool accepting a known report ID and recipient ID, then resolve both server-side. Tool descriptions consume prompt tokens, so keep that contract focused and put rejected IDs, unauthorized recipients, and stale approvals in the eval set.
from email.message import EmailMessage
from pathlib import Path
from typing import Protocol
class MailTransport(Protocol):
def submit(self, message: EmailMessage) -> str:
...
def build_message(
sender: str,
recipient: str,
artifact: ReportArtifact,
) -> EmailMessage:
payload = Path(artifact.path).read_bytes()
if digest(payload) != artifact.sha256:
raise ValueError("artifact digest mismatch")
message = EmailMessage()
message["From"] = sender
message["To"] = recipient
message["Subject"] = f"Approved logistics report {artifact.shipment_id}"
message.set_content(
f"The approved report for shipment {artifact.shipment_id} is attached."
)
message.add_attachment(
payload,
maintype="application",
subtype="pdf",
filename=f"{artifact.shipment_id}-report.pdf",
)
return message
Authentication signing can happen in the transport or mail infrastructure. The application still owns the evidence tying SHIP-1042, the approved digest, recipient, and transition together.
Evaluate the failure paths
The best API is the one whose boundary supports the evidence model. Before selection, run the same contract suite against every candidate transport and a deterministic fake.
| Experiment | Injected condition | Required evidence |
|---|---|---|
| Duplicate command | Same command key twice | One attempt and one queued item |
| Intentional resend | New key, same report | New attempt with the same digest |
| Early cancel | Cancel wins before claim | Canceled state and no submission |
| Late cancel | Claim wins first | Dispatched state; cancel rejected |
| Changed artifact | Bytes differ after approval | Digest failure before submission |
| Uncertain result | Submission times out | Named unresolved state for reconciliation |
Do not turn an uncertain submission into a fresh send automatically. Reconcile it using an observable result from the mail path, then record the retry decision. This trade-off can reduce throughput, but it makes duplicate risk and operator intent visible.
Measure race outcomes under concurrent workers, evidence completeness, duplicate attachments at the test recipient, and prompt tokens consumed by an agent-facing tool. Also test recovery: starting with one attempt ID, can an operator find the command, report digest, approval, recipient resolution, and transport result without searching unrelated logs?
Keep the state machine boring. Persist approved bytes outside the prompt, verify them at dispatch, and represent ambiguous results honestly. The experiment is ready when concurrency evals preserve those invariants and the evidence query works end to end. Latency and developer ergonomics can then choose among compatible transports; compliance evidence remains the gate.
References
- RFC 6376: DomainKeys Identified Mail (DKIM): https://datatracker.ietf.org/doc/html/rfc6376
- Anthropic, Tool use overview: https://platform.claude.com/docs/en/agents-and-tools/tool-use/overview
Top comments (0)