""" Email service module for Rex AI Assistant. Provides email triage functionality with classification and summarization. Design goals: - Works in stub/mock mode (no real IMAP/SMTP yet) by default. - Supports both: 1) "fetch_unread(...)" returning structured EmailSummary objects 2) "triage_unread(...)" returning CLI-friendly dict summaries - Optional EventBus publishing if rex.event_bus is available and provided. - Backend-aware: accepts an optional ``EmailBackend`` for real IMAP/SMTP. - Multi-account: resolves backends per validated user and account via ``rex.email_accounts.EmailAccountResolver`` (issue #303). - ``send()`` method delegates to the requesting user's authorized backend. Per-user isolation (issue #303): - Every operation that reads or mutates email data requires an explicit, validated ``user_id``. Missing, blank, malformed, or traversal-style identities fail closed (``EmailIdentityError``) before any account or credential lookup. - Account selection is restricted to the requesting user's authorized accounts; explicit foreign or nonexistent accounts raise the generic ``EmailAccountAccessError`` (indistinguishable to the caller). - Backends are cached per ``(user_id, account_id)``; a backend resolved for one user is never reused for another. - Stub/mock mode keeps a separate mutable inbox per user so one user's mark-read never alters another user's view. """ from __future__ import annotations import copy import json import logging from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path from typing import Any from rex.email_accounts import ( DEFAULT_PROFILE, EmailAccountResolver, require_user_id, ) logger = logging.getLogger(__name__) # Optional EventBus import for compatibility with earlier implementation try: from rex.openclaw.event_bus import EventBus except Exception: # pragma: no cover EventBus = Any # type: ignore # Optional credentials manager import try: from rex.credentials import get_credential_manager except Exception: # pragma: no cover get_credential_manager = None # type: ignore try: from rex.assistant_errors import IntegrationNotConfiguredError except Exception: # pragma: no cover IntegrationNotConfiguredError = RuntimeError # type: ignore # Optional Pydantic import (nice to have, not required at runtime) try: from pydantic import BaseModel, ConfigDict, Field, field_serializer class EmailSummary(BaseModel): """Summary of an email message.""" model_config = ConfigDict() id: str = Field(..., description="Unique email identifier") from_addr: str = Field(..., description="Sender email address") subject: str = Field(..., description="Email subject") snippet: str = Field(..., description="Brief preview of email body") received_at: datetime = Field(..., description="When the email was received") labels: list[str] = Field(default_factory=list, description="Email labels/tags") importance_score: float = Field(default=0.5, description="Importance score (0.0-1.0)") category: str | None = Field( default=None, description="Email category (important, promo, social, etc.)" ) @field_serializer("received_at", when_used="json") @classmethod def _serialize_received_at(cls, v: datetime, _info: object) -> str: return v.isoformat() except Exception: # pragma: no cover @dataclass class EmailSummary: # type: ignore """Fallback EmailSummary when pydantic is not installed.""" id: str from_addr: str subject: str snippet: str received_at: datetime labels: list[str] importance_score: float = 0.5 category: str | None = None @dataclass(frozen=True) class EmailMessage: """Legacy/compat message type used by older CLI/service code.""" message_id: str sender: str subject: str body: str received_at: datetime def to_summary(self) -> dict[str, Any]: return { "message_id": self.message_id, "sender": self.sender, "subject": self.subject, "received_at": self.received_at.isoformat(), "body": self.body, } class EmailService: """ Email service for reading, categorizing, and sending emails. Supports two modes: - **Stub mode** (default): reads from a local JSON fixture; send is a logged no-op. Used for offline development and testing. Each user gets an isolated mutable copy of the fixture inbox. - **Backend mode**: resolves an ``EmailBackend`` per validated user and authorized account via ``EmailAccountResolver``. An explicitly injected backend (constructor ``backend=`` or :meth:`set_backend`) serves only the user it is bound to. Every read/mutate/send operation requires a validated ``user_id`` and fails closed on missing or invalid identity. """ def __init__( self, mock_data_file: Path | None = None, *, event_bus: EventBus | None = None, mock_messages: list[EmailMessage] | None = None, mock_data_path: Path | None = None, backend: object | None = None, backend_user_id: str = DEFAULT_PROFILE, user_email_accounts: dict[str, list[Any]] | None = None, user_default_accounts: dict[str, str] | None = None, account_resolver: EmailAccountResolver | None = None, ) -> None: if event_bus is None and mock_data_file is not None and hasattr(mock_data_file, "publish"): event_bus = mock_data_file # type: ignore[assignment] mock_data_file = None if mock_data_file is None and mock_data_path is not None: mock_data_file = mock_data_path self.mock_data_file = mock_data_file or Path("data/mock_emails.json") self.connected = False self._event_bus = event_bus self._mock_messages = mock_messages # If provided, overrides file loading self._mock_emails: list[EmailSummary] = [] # Per-user isolated copies of the stub inbox: {user_id: [EmailSummary]} self._user_mock_emails: dict[str, list[EmailSummary]] = {} # Explicitly injected backend, bound to exactly one user. self._backend = backend self._backend_owner = require_user_id(backend_user_id) if backend is not None else None # Per-user account map: {user_id: [UserEmailAccount, ...]} self._user_email_accounts: dict[str, list[Any]] = user_email_accounts or {} # Per-user default account selection: {user_id: account_id} self._user_default_accounts: dict[str, str] = user_default_accounts or {} # Injected authorization/routing resolver (tests/embedding only; the # production path loads lazily so config changes are picked up). self._account_resolver = account_resolver self._resolver_cache: EmailAccountResolver | None = None self._resolver_stamp: int | None = None # Backends resolved from config, cached per (user_id, account_id). # A backend resolved for one user is never served to another. self._user_backends: dict[tuple[str, str], object] = {} self.credential_manager = None if get_credential_manager is not None: try: self.credential_manager = get_credential_manager() except Exception: self.credential_manager = None # ------------------------------------------------------------------ # Backend management # ------------------------------------------------------------------ def set_backend(self, backend: object, *, user_id: str = DEFAULT_PROFILE) -> None: """Swap the explicitly injected email backend at runtime. The injected backend is bound to *user_id* and serves only that user's operations; other users resolve through the account resolver. Args: backend: An ``EmailBackend`` instance (or ``None`` for stub). user_id: The user this backend belongs to (default: the legacy ``default`` profile). """ self._backend = backend self._backend_owner = require_user_id(user_id) if backend is not None else None self.connected = False @property def active_backend(self) -> object | None: """The explicitly injected backend (``None`` in stub/resolver mode).""" return self._backend # ------------------------------------------------------------------ # User-scoped account access (US-ME-002 / issue #303) # ------------------------------------------------------------------ def _get_resolver(self) -> EmailAccountResolver: """Return the account resolver for this service instance. Config-backed resolvers are invalidated when the runtime config file changes, so revoking or reassigning a user's accounts takes effect in long-lived processes without a restart. Authorization is re-checked on every operation, so cached backends never outlive their owner's assignment. """ if self._account_resolver is not None: return self._account_resolver from rex import email_accounts as _email_accounts stamp = _email_accounts.config_stamp() if self._resolver_cache is None or stamp != self._resolver_stamp: if self._user_email_accounts or self._user_default_accounts: # Explicitly injected authorization map: authoritative. base = EmailAccountResolver.load() self._resolver_cache = EmailAccountResolver( base.email_config, self._user_email_accounts, self._user_default_accounts, ) else: self._resolver_cache = EmailAccountResolver.load() self._resolver_stamp = stamp return self._resolver_cache def get_accounts(self, user_id: str) -> list[Any]: """Return the email account entries authorized for *user_id*. Returns an empty list (not an error) when the user has no accounts configured. Never returns accounts belonging to other users. Raises: EmailIdentityError: On missing or invalid identity. """ return self._get_resolver().entries_for_user(user_id) def _check_account_access(self, user_id: str, account_id: str) -> None: """Raise ``EmailAccountAccessError`` unless *user_id* owns *account_id*. Unauthorized and nonexistent accounts are indistinguishable. An empty authorization map means "no access", never "allow all". """ self._get_resolver().check_account_access(user_id, account_id) def _get_user_backend( self, user_id: str, account_id: str | None = None ) -> tuple[object | None, Any]: """Resolve the backend serving *user_id*, after ownership validation. Credential lookup happens only here, only after the account has been authorized, and only with that account definition's own ``credential_ref``. Returns: ``(backend_or_None, account_definition_or_None)``. """ validated = require_user_id(user_id) resolver = self._get_resolver() resolved_id = resolver.resolve_account_id(validated, account_id) if resolved_id is None: return None, None definition = resolver.get_account_definition(resolved_id) if definition is None: return None, None key = (validated, resolved_id) backend = self._user_backends.get(key) if backend is None: credential_getter = None if self.credential_manager is not None and hasattr( self.credential_manager, "get_token" ): credential_getter = self.credential_manager.get_token from rex.email_backends.account_router import build_backend_for_account backend = build_backend_for_account(definition, credential_getter=credential_getter) if backend is None: return None, definition try: backend.connect() except Exception as exc: logger.warning( "Failed to connect email backend for account %r: %s", resolved_id, exc ) self._user_backends[key] = backend return backend, definition # ------------------------------------------------------------------ # Connection # ------------------------------------------------------------------ def connect(self, user_id: str | None = None) -> bool: """ Connect to email service. With an injected backend, delegates to ``backend.connect()``. When real accounts are configured, *user_id* is required and only that user's authorized backend is connected. Otherwise falls back to stub behaviour (loads mock data). """ try: if self._backend is not None: result = self._backend.connect() # type: ignore[attr-defined] self.connected = bool(result) if self.connected: logger.info("Email service connected via backend") return self.connected resolver = self._get_resolver() if resolver.has_configured_accounts(): if user_id is None: logger.warning( "Email accounts are configured; connect() requires a user identity" ) return False backend, _definition = self._get_user_backend(user_id) if backend is None: return False self.connected = True return True # Stub mode if self.credential_manager is not None: try: email_creds = self.credential_manager.get_credential("email") if not email_creds: logger.warning("No email credentials configured (continuing in stub mode)") except Exception as e: logger.warning("Credential manager error (continuing in stub mode): %s", e) self._load_mock_data() self.connected = True logger.info("Email service connected (stub mode)") return True except Exception as e: logger.error("Failed to connect email service: %s", e, exc_info=True) self.connected = False return False def _publish(self, topic: str, payload: dict[str, Any]) -> None: if self._event_bus is None: return try: self._event_bus.publish(topic, payload) except Exception as e: logger.debug("EventBus publish failed for %s: %s", topic, e) def _publish_user_event( self, topic: str, user_id: str, shared_payload: dict[str, Any], private_payload: dict[str, Any] | None = None, ) -> None: """Publish a safe envelope on the shared topic and the full payload on the owner's user-scoped topic. The shared topic must never carry private email fields (subjects, senders, snippets); those go only to ``{topic}.user.{user_id}``. """ self._publish(topic, shared_payload) if private_payload is not None: self._publish(f"{topic}.user.{user_id}", private_payload) def _load_mock_data(self) -> None: """Load mock email data from file or legacy mock messages.""" if self._mock_messages is not None: self._mock_emails = [self._email_summary_from_message(m) for m in self._mock_messages] return if not self.mock_data_file.exists(): logger.warning("No mock email data at %s", self.mock_data_file) self._mock_emails = [] return try: raw = self.mock_data_file.read_text(encoding="utf-8") data = json.loads(raw) except Exception as e: logger.error("Failed to read mock email data: %s", e, exc_info=True) self._mock_emails = [] return # Accept either: # 1) list[dict] # 2) {"messages": list[dict]} items: list[dict[str, Any]] if isinstance(data, list): items = data elif isinstance(data, dict) and isinstance(data.get("messages"), list): items = [x for x in data["messages"] if isinstance(x, dict)] else: logger.warning( "Mock email data format not recognized, expected list or {messages: [...]}" ) self._mock_emails = [] return parsed: list[EmailSummary] = [] for item in items: try: parsed.append(self._email_summary_from_dict(item)) except Exception as e: logger.warning("Skipping invalid mock email record: %s", e) self._mock_emails = parsed logger.info("Loaded %d mock emails", len(self._mock_emails)) def _user_inbox(self, user_id: str) -> list[EmailSummary]: """Return *user_id*'s isolated mutable copy of the stub inbox.""" inbox = self._user_mock_emails.get(user_id) if inbox is None: inbox = copy.deepcopy(self._mock_emails) self._user_mock_emails[user_id] = inbox return inbox def _email_summary_from_message(self, msg: EmailMessage) -> EmailSummary: return EmailSummary( id=msg.message_id, from_addr=msg.sender, subject=msg.subject, snippet=(msg.body or "")[:200], received_at=msg.received_at, labels=["unread"], importance_score=0.5, ) def _email_summary_from_dict(self, d: dict[str, Any]) -> EmailSummary: # Support both schema variants: # - legacy: message_id, sender, subject, body, received_at # - newer: id, from_addr, subject, snippet, received_at, labels, importance_score email_id = str(d.get("id") or d.get("message_id") or "") if not email_id: raise ValueError("missing id/message_id") from_addr = str(d.get("from_addr") or d.get("sender") or "unknown@example.com") subject = str(d.get("subject") or "") body = d.get("body") snippet = d.get("snippet") if snippet is None and body is not None: snippet = str(body)[:200] if snippet is None: snippet = "" received_at_raw = d.get("received_at") received_at = self._parse_datetime(received_at_raw) or datetime.now(UTC) labels = d.get("labels") if not isinstance(labels, list): labels = ["unread"] importance_score = d.get("importance_score") try: importance_score_f = float(importance_score) if importance_score is not None else 0.5 except Exception: importance_score_f = 0.5 category = d.get("category") category_str = str(category) if category is not None else None return EmailSummary( id=email_id, from_addr=from_addr, subject=subject, snippet=str(snippet), received_at=received_at, labels=[str(x) for x in labels], importance_score=importance_score_f, category=category_str, ) def _parse_datetime(self, value: Any) -> datetime | None: if value is None: return None if isinstance(value, datetime): return value if value.tzinfo else value.replace(tzinfo=UTC) if isinstance(value, str): try: dt = datetime.fromisoformat(value) return dt if dt.tzinfo else dt.replace(tzinfo=UTC) except Exception: return None return None # ------------------------------------------------------------------ # Read operations # ------------------------------------------------------------------ def fetch_unread( self, limit: int = 10, *, user_id: str | None = None, account_id: str | None = None, ) -> list[EmailSummary]: """ Fetch unread email summaries for the requesting user. Args: limit: Maximum number of emails to return. user_id: Requesting user (required). Only that user's authorized accounts are consulted. account_id: Optional explicit account, validated for ownership. Returns: List of EmailSummary objects. Returns ``[]`` when the user has no authorized/usable account (documented not-configured result). Raises: EmailIdentityError: On missing or invalid identity. EmailAccountAccessError: When *account_id* is not available to this user (unauthorized or nonexistent). """ validated = require_user_id(user_id) resolver = self._get_resolver() if account_id: resolver.check_account_access(validated, account_id) if self._backend is not None and self._backend_owner == validated: if not self.connected: if not self.connect(): logger.warning("Email service not connected") return [] envelopes = self._backend.fetch_unread(limit=limit) # type: ignore[attr-defined] result = [self._envelope_to_summary(env) for env in envelopes] self._publish_user_event( "email.unread", validated, {"count": len(result), "user_id": validated, "account_id": account_id}, { "count": len(result), "user_id": validated, "account_id": account_id, "messages": [self._summary_dict(e) for e in result], }, ) return result if resolver.has_configured_accounts(): backend, definition = self._get_user_backend(validated, account_id) if backend is None: logger.warning("No usable email account for user %r", validated) return [] resolved_id = getattr(definition, "id", None) envelopes = backend.fetch_unread(limit=limit) # type: ignore[attr-defined] result = [self._envelope_to_summary(env) for env in envelopes] self._publish_user_event( "email.unread", validated, {"count": len(result), "user_id": validated, "account_id": resolved_id}, { "count": len(result), "user_id": validated, "account_id": resolved_id, "messages": [self._summary_dict(e) for e in result], }, ) return result # Stub mode: per-user isolated inbox. if not self.connected: if not self.connect(): logger.warning("Email service not connected") return [] inbox = self._user_inbox(validated) unread = [email for email in inbox if "unread" in (email.labels or [])] result = unread[: max(0, int(limit))] self._publish_user_event( "email.unread", validated, {"count": len(result), "user_id": validated, "account_id": account_id}, { "count": len(result), "user_id": validated, "account_id": account_id, "messages": [self._summary_dict(e) for e in result], }, ) return result def triage_unread( self, limit: int = 10, *, user_id: str | None = None, account_id: str | None = None, ) -> list[dict[str, Any]]: """ Legacy-friendly triage output used by older CLI command. Requires a validated ``user_id``; triages only that user's unread mail. Returns list of dicts: { message_id, sender, subject, received_at (datetime), category, summary } """ validated = require_user_id(user_id) unread = self.fetch_unread(limit=limit, user_id=validated, account_id=account_id) triaged: list[dict[str, Any]] = [] for email in unread: category = self.categorize(email) summary = self._triage_summary(email) triaged.append( { "message_id": email.id, "sender": email.from_addr, "subject": email.subject, "received_at": email.received_at, "category": category, "summary": summary, } ) self._publish_user_event( "email.triaged", validated, {"count": len(triaged), "user_id": validated}, { "count": len(triaged), "user_id": validated, "triaged": [ {"message_id": t["message_id"], "category": t["category"]} for t in triaged ], }, ) return triaged def mark_as_read(self, email_id: str, *, user_id: str | None = None) -> bool: """Mark an email as read within the requesting user's inbox.""" validated = require_user_id(user_id) if self._backend is not None and self._backend_owner == validated: if not self.connected: logger.warning("Email service not connected") return False result = self._backend.mark_as_read(email_id) # type: ignore[attr-defined] if result: self._publish("email.read", {"id": email_id, "user_id": validated}) return result # type: ignore[no-any-return] resolver = self._get_resolver() if resolver.has_configured_accounts(): backend, _definition = self._get_user_backend(validated) if backend is None: logger.warning("No usable email account for user %r", validated) return False result = backend.mark_as_read(email_id) # type: ignore[attr-defined] if result: self._publish("email.read", {"id": email_id, "user_id": validated}) return bool(result) if not self.connected: logger.warning("Email service not connected") return False for email in self._user_inbox(validated): if email.id == email_id: if email.labels and "unread" in email.labels: email.labels.remove("unread") logger.info("Marked email %s as read for user %s", email_id, validated) self._publish("email.read", {"id": email_id, "user_id": validated}) return True logger.warning("Email not found: %s", email_id) return False # ------------------------------------------------------------------ # Send operations # ------------------------------------------------------------------ def send( self, *, to: str | list[str], subject: str, body: str, from_addr: str | None = None, account_id: str | None = None, user_id: str | None = None, ) -> dict[str, Any]: """Send an email through the requesting user's authorized account. Args: to: Recipient address(es). subject: Email subject line. body: Plain-text body. from_addr: Sender address override (defaults to account address). account_id: Explicit account to route through, validated for ownership. user_id: Requesting user (required). Returns: A dict with keys ``ok`` (bool), ``message_id`` (str|None), and ``error`` (str|None). When real accounts are configured but none is available to this user, ``ok`` is False (fail closed) — never a fallback through another user's account. In pure stub mode (no accounts configured anywhere) the send is logged and reported as ok. Raises: EmailIdentityError: On missing or invalid identity. EmailAccountAccessError: When *account_id* is not available to this user (unauthorized or nonexistent). """ validated = require_user_id(user_id) resolver = self._get_resolver() if account_id: resolver.check_account_access(validated, account_id) to_addrs = [to] if isinstance(to, str) else list(to) send_backend = None resolved_account = None if self._backend is not None and self._backend_owner == validated: send_backend = self._backend else: send_backend, resolved_account = self._get_user_backend(validated, account_id) if send_backend is not None and hasattr(send_backend, "send"): sender = from_addr or getattr(resolved_account, "address", "") or "" result = send_backend.send( from_addr=sender, to_addrs=to_addrs, subject=subject, body=body, ) return { "ok": result.ok, "message_id": result.message_id, "error": result.error, } if resolver.has_configured_accounts(): # Real accounts exist but none is available to this user (or its # credentials are missing): fail closed, never fall back. logger.warning("Email send refused: no authorized account for user %r", validated) return { "ok": False, "message_id": None, "error": "No email account is available for this user", } # Stub mode: log and return success. logger.info( "[STUB] Would send email to=%s subject=%r", to_addrs, subject, ) return {"ok": True, "message_id": None, "error": None} # ------------------------------------------------------------------ # Categorisation / summarisation # ------------------------------------------------------------------ def categorize(self, email: EmailSummary) -> str: """ Categorize an email based on heuristics. Returns: Category string: important, promo, social, newsletter, finance, calendar, general """ subject_lower = (email.subject or "").lower() from_lower = (email.from_addr or "").lower() snippet_lower = (email.snippet or "").lower() # Finance indicators finance_keywords = ["invoice", "payment", "receipt", "billing", "charged", "refund"] if any(kw in subject_lower or kw in snippet_lower for kw in finance_keywords): return "finance" # Calendar indicators calendar_keywords = ["meeting", "schedule", "invite", "calendar", "appointment"] if any(kw in subject_lower or kw in snippet_lower for kw in calendar_keywords): return "calendar" # Promo indicators promo_keywords = [ "sale", "discount", "offer", "deal", "promotion", "coupon", "free shipping", ] if any(kw in subject_lower or kw in snippet_lower for kw in promo_keywords): return "promo" # Social indicators social_keywords = [ "liked your", "commented on", "mentioned you", "friend request", "connection", ] social_domains = ["facebook.com", "twitter.com", "linkedin.com", "instagram.com"] if any(kw in subject_lower or kw in snippet_lower for kw in social_keywords): return "social" if any(domain in from_lower for domain in social_domains): return "social" # Newsletter indicators if "unsubscribe" in snippet_lower or "newsletter" in subject_lower: return "newsletter" # Important indicators important_keywords = ["urgent", "important", "asap", "action required", "deadline"] if any(kw in subject_lower for kw in important_keywords): return "important" if getattr(email, "importance_score", 0.5) >= 0.8: return "important" return "general" def summarize(self, email_id: str, *, user_id: str | None = None) -> str: """ Get a simple summary of an email (stub implementation). Currently returns a formatted snippet from the requesting user's inbox. """ validated = require_user_id(user_id) if not self.connected: return "Email service not connected" for email in self._user_inbox(validated): if email.id == email_id: return f"From: {email.from_addr}\nSubject: {email.subject}\n\n{email.snippet}" return f"Email not found: {email_id}" def _triage_summary(self, email: EmailSummary) -> str: return f"{email.subject} (from {email.from_addr})" def _summary_dict(self, email: EmailSummary) -> dict[str, Any]: return { "id": email.id, "from_addr": email.from_addr, "subject": email.subject, "received_at": ( email.received_at.isoformat() if isinstance(email.received_at, datetime) else str(email.received_at) ), "labels": list(email.labels or []), "importance_score": float(getattr(email, "importance_score", 0.5)), "category": getattr(email, "category", None), } def get_all_emails(self) -> list[EmailSummary]: """Get all emails from the loaded fixture (stub/testing).""" return self._mock_emails.copy() # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ @staticmethod def _envelope_to_summary(env: object) -> EmailSummary: """Convert an ``EmailEnvelope`` from a backend to ``EmailSummary``.""" return EmailSummary( id=getattr(env, "message_id", ""), from_addr=getattr(env, "from_addr", ""), subject=getattr(env, "subject", ""), snippet=getattr(env, "snippet", ""), received_at=getattr(env, "received_at", datetime.now(UTC)), labels=list(getattr(env, "labels", [])), importance_score=0.5, ) # Global email service instance _email_service: EmailService | None = None def _create_configured_email_service() -> EmailService: """Create an EmailService wired to the per-user account resolver. Backends are resolved lazily per validated user and authorized account; no backend or credential is touched at construction time. Raises: IntegrationNotConfiguredError: when no email accounts are configured. """ resolver = EmailAccountResolver.load() if not resolver.has_configured_accounts(): raise IntegrationNotConfiguredError("Email: not configured") # No resolver is injected: the service loads it lazily and refreshes it # when the config file changes, so account revocations take effect in # long-lived processes without a restart. return EmailService() def get_email_service() -> EmailService: """Get the global email service instance. The returned service enforces per-user account ownership internally; every operation requires a validated ``user_id``. Raises: IntegrationNotConfiguredError: when no email accounts are configured. """ global _email_service if _email_service is None: _email_service = _create_configured_email_service() return _email_service def set_email_service(service: EmailService) -> None: """Set the global email service instance (for testing).""" global _email_service _email_service = service __all__ = [ "EmailSummary", "EmailMessage", "EmailService", "get_email_service", "set_email_service", ]