diff --git a/backend/alembic/versions/0018_email_date_provenance.py b/backend/alembic/versions/0018_email_date_provenance.py new file mode 100644 index 000000000..c33a7a2f4 --- /dev/null +++ b/backend/alembic/versions/0018_email_date_provenance.py @@ -0,0 +1,49 @@ +"""add date_provenance to email_records + +Revision ID: 0018_email_date_provenance +Revises: 0017_merge_newsdom_carddav_heads +Create Date: 2026-07-30 00:00:00.000000 + +Records the provenance of each stored email ``date`` so a synthetic +collection-time fallback (missing/invalid RFC822 Date header) is never treated +as original sender metadata when seeding a strong auto-dedupe fingerprint +(naruon#1086). Nullable-free with a ``"unknown"`` server default so existing +rows backfill safely: their date provenance is genuinely unknown, and only +``"parsed"`` rows are eligible to seed a strong fingerprint, so the backfill is +conservative (it can only widen review, never manufacture a duplicate). +""" + +from alembic import op +import sqlalchemy as sa + +revision = "0018_email_date_provenance" +down_revision = "0017_merge_newsdom_carddav_heads" + +_EMAIL_TABLE = "email_records" +_PROVENANCE_COLUMN = "date_provenance" + + +def upgrade() -> None: + """Add the ``date_provenance`` column, backfilling existing rows to unknown.""" + connection = op.get_bind() + inspector = sa.inspect(connection) + columns = {column["name"] for column in inspector.get_columns(_EMAIL_TABLE)} + if _PROVENANCE_COLUMN not in columns: + op.add_column( + _EMAIL_TABLE, + sa.Column( + _PROVENANCE_COLUMN, + sa.String(), + nullable=False, + server_default="unknown", + ), + ) + + +def downgrade() -> None: + """Drop the ``date_provenance`` column if present.""" + connection = op.get_bind() + inspector = sa.inspect(connection) + columns = {column["name"] for column in inspector.get_columns(_EMAIL_TABLE)} + if _PROVENANCE_COLUMN in columns: + op.drop_column(_EMAIL_TABLE, _PROVENANCE_COLUMN) diff --git a/backend/alembic/versions/0019_pop3_observed_uidl.py b/backend/alembic/versions/0019_pop3_observed_uidl.py new file mode 100644 index 000000000..126c445bb --- /dev/null +++ b/backend/alembic/versions/0019_pop3_observed_uidl.py @@ -0,0 +1,76 @@ +"""add durable POP3 UIDL collection progress + +Revision ID: 0019_pop3_observed_uidl +Revises: 0018_email_date_provenance +Create Date: 2026-09-17 00:00:00.000000 + +Persists RFC 1939 UIDL provider identity and bounded retry disposition per +mailbox configuration. Provider state remains collection progress only; Naruon +email Message-ID and source fingerprints remain canonical message/deduplication +evidence. + +This revision identifier and parent are branch-local until the canonical +workspace/Alembic owner is integrated; #1195 must rechain this schema after the +then-protected migration head before merge. +""" + +from alembic import op +import sqlalchemy as sa + +revision = "0019_pop3_observed_uidl" +down_revision = "0018_email_date_provenance" + +_TABLE = "pop3_observed_messages" + + +def upgrade() -> None: + """Create owner-scoped durable POP3 UIDL collection progress if absent.""" + connection = op.get_bind() + inspector = sa.inspect(connection) + if _TABLE in inspector.get_table_names(): + return + + op.create_table( + _TABLE, + sa.Column("observed_message_id", sa.Integer(), primary_key=True), + sa.Column( + "tenant_config_id", + sa.Integer(), + sa.ForeignKey("tenant_configs.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("provider_uidl", sa.String(length=70), nullable=False), + sa.Column( + "collection_disposition", + sa.String(length=16), + nullable=False, + server_default="observed", + ), + sa.Column("retry_after", sa.DateTime(timezone=True), nullable=True), + sa.Column("observed_at", sa.DateTime(timezone=True), nullable=True), + sa.UniqueConstraint( + "tenant_config_id", + "provider_uidl", + name="uq_pop3_observed_messages_account_uidl", + ), + ) + op.create_index( + "ix_pop3_observed_messages_account_observed", + _TABLE, + ["tenant_config_id", "observed_at"], + unique=False, + ) + op.create_index( + "ix_pop3_observed_messages_account_retry", + _TABLE, + ["tenant_config_id", "collection_disposition", "retry_after"], + unique=False, + ) + + +def downgrade() -> None: + """Drop durable POP3 UIDL collection progress if present.""" + connection = op.get_bind() + inspector = sa.inspect(connection) + if _TABLE in inspector.get_table_names(): + op.drop_table(_TABLE) diff --git a/backend/db/models.py b/backend/db/models.py index 98e17eef2..a798b5f8c 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -801,6 +801,13 @@ def owner_filters(cls, user_id: str, organization_id: str | None): in_reply_to: Mapped[str | None] = mapped_column(String, nullable=True) references: Mapped[str | None] = mapped_column(String, nullable=True) date: Mapped[datetime.datetime] = mapped_column(DateTime(timezone=True), index=True) + # Provenance of the stored ``date``: "parsed" (genuine RFC822 Date header), + # "missing"/"invalid" (a synthetic collection-time fallback), or "unknown" + # (rows stored before provenance tracking). Only "parsed" rows may seed a + # strong auto-dedupe fingerprint (naruon#1086). + date_provenance: Mapped[str] = mapped_column( + String, nullable=False, server_default="unknown", default="unknown" + ) body: Mapped[str] = mapped_column(Text) # IMAP \Seen read state; defaults read so historical/file imports don't nag. is_read: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) diff --git a/backend/db/pop3_collection_models.py b/backend/db/pop3_collection_models.py new file mode 100644 index 000000000..45766295c --- /dev/null +++ b/backend/db/pop3_collection_models.py @@ -0,0 +1,58 @@ +import datetime + +from sqlalchemy import DateTime, ForeignKey, Index, String, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from db.models import Base + + +class Pop3ObservedMessage(Base): + """Durable provider progress for one POP3 UIDL within an account. + + POP3 message numbers are session-local positions. ``provider_uidl`` stores + RFC 1939 provider identity so bounded polling can make progress across + reconnects and renumbering. ``collection_disposition`` distinguishes + successfully observed messages from retryable collection failures; neither + state redefines Naruon's Message-ID or source-fingerprint identity. + """ + + __tablename__ = "pop3_observed_messages" + __table_args__ = ( + UniqueConstraint( + "tenant_config_id", + "provider_uidl", + name="uq_pop3_observed_messages_account_uidl", + ), + Index( + "ix_pop3_observed_messages_account_observed", + "tenant_config_id", + "observed_at", + ), + Index( + "ix_pop3_observed_messages_account_retry", + "tenant_config_id", + "collection_disposition", + "retry_after", + ), + ) + + observed_message_id: Mapped[int] = mapped_column(primary_key=True) + tenant_config_id: Mapped[int] = mapped_column( + ForeignKey("tenant_configs.id", ondelete="CASCADE"), + nullable=False, + ) + provider_uidl: Mapped[str] = mapped_column(String(70), nullable=False) + collection_disposition: Mapped[str] = mapped_column( + String(16), + default="observed", + nullable=False, + ) + retry_after: Mapped[datetime.datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + observed_at: Mapped[datetime.datetime | None] = mapped_column( + DateTime(timezone=True), + default=lambda: datetime.datetime.now(datetime.timezone.utc), + nullable=True, + ) diff --git a/backend/services/email_dedupe_service.py b/backend/services/email_dedupe_service.py index 1c74d5d11..539b7dfda 100644 --- a/backend/services/email_dedupe_service.py +++ b/backend/services/email_dedupe_service.py @@ -1,13 +1,41 @@ +"""Email de-duplication fingerprints and the Fellegi-Sunter decision classifier. + +This module is the deterministic core the import/IMAP paths compose to decide +whether an incoming email is a duplicate of a stored one, keeping strong +(auto-merge) evidence gated on genuine Date provenance (naruon#1086). +""" + import datetime +import hashlib +import json +import math +from collections.abc import Iterable, Mapping from dataclasses import dataclass +from typing import Literal from db.models import Email from services.email_service import generate_email_fingerprint from services.threading_service import normalize_message_id +# Fellegi & Sunter (1969) partition each candidate/record pair into three +# decision zones: a positive link (A1), a non-link (A3), and an indeterminate +# "possible match" band (A2) reserved for clerical review. naruon#1086 maps that +# rule onto email de-duplication: a reliable identity link auto-merges, a +# probable duplicate that lacks a reliable link is held for review instead of +# being silently kept or silently merged, and everything else is distinct. +DedupeDecision = Literal["auto_link", "review_required", "distinct"] + @dataclass(frozen=True) class EmailDedupeCandidate: + """An incoming email reduced to the fields the dedupe decision needs. + + ``date_provenance`` mirrors the parser's classification of the ``date`` + field (``parsed`` for a genuine RFC822 Date, otherwise a synthetic + collection-time fallback); only ``parsed`` may seed a strong auto-dedupe + match (naruon#1086). + """ + candidate_key: str message_id: str | None = None sender: str | None = None @@ -15,14 +43,120 @@ class EmailDedupeCandidate: subject: str | None = None date: datetime.datetime | None = None body: str | None = None + date_provenance: str = "unknown" def _date_to_fingerprint_value(value: datetime.datetime | None) -> str: + """Render a datetime as its ISO-8601 fingerprint token (``""`` when None).""" if value is None: return "" return value.isoformat() +_CANONICAL_SOURCE_FIELDS = ( + "message_id", + "sender", + "recipients", + "subject", + "body", + "reply_to", + "in_reply_to", + "references", + "attachments", +) + + +def _validate_canonical_source_value(value: object, *, path: str) -> None: + """Reject values outside the deterministic JSON-native EmailData surface. + + Silent ``str()`` coercion is unsafe for identity material because distinct + runtime types can render to the same text. Parsed canonical-source fields + therefore accept only JSON-native scalars, lists, and string-keyed mappings; + every nested value is checked before serialization. + """ + if value is None or isinstance(value, (str, bool, int)): + return + if isinstance(value, float): + if not math.isfinite(value): + raise TypeError(f"{path}: non-finite floats are not canonical") + return + if isinstance(value, list): + for index, item in enumerate(value): + _validate_canonical_source_value(item, path=f"{path}[{index}]") + return + if isinstance(value, dict): + for key, item in value.items(): + if not isinstance(key, str): + raise TypeError( + f"{path}: canonical mapping keys must be strings, got " + f"{type(key).__name__}" + ) + _validate_canonical_source_value(item, path=f"{path}.{key}") + return + raise TypeError( + f"{path}: unsupported canonical email source value type " + f"{type(value).__name__}" + ) + + +def canonical_email_source_content(email_data: Mapping[str, object]) -> bytes: + """Serialize stable parsed fields when raw transport bytes are unavailable. + + Collection-time ``date`` values and their provenance are deliberately + excluded. Transport-backed paths should provide exact RFC822 bytes. Values + outside the parsed ``EmailData`` JSON surface fail closed rather than being + string-coerced into potentially colliding identities. + """ + payload = {field: email_data.get(field) for field in _CANONICAL_SOURCE_FIELDS} + for field, value in payload.items(): + _validate_canonical_source_value(value, path=field) + return json.dumps( + payload, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8", errors="surrogatepass") + + +def source_email_fingerprint( + source_content: bytes, + *, + source_kind: Literal["raw", "canonical"] = "raw", +) -> str: + """Return a domain-separated SHA-256 identity for stable source bytes.""" + digest = hashlib.sha256() + digest.update(b"naruon-email-source-v1\0") + digest.update(source_kind.encode("ascii")) + digest.update(b"\0") + digest.update(source_content) + return digest.hexdigest() + + +def _has_nonempty_text(value: object) -> bool: + """Return whether a metadata field supplies usable non-empty text.""" + return isinstance(value, str) and bool(value.strip()) + + +def has_complete_strong_email_metadata( + *, + sender: object, + recipients: object, + subject: object, + body: object, +) -> bool: + """Require the full sender/recipient/subject/body evidence set for auto-linking. + + Strong metadata evidence is intentionally stricter than source-bound identity: + incomplete messages keep their deterministic raw/canonical source fingerprint + but cannot enter the automatic metadata-link path merely because Date parsed. + """ + return all( + _has_nonempty_text(value) + for value in (sender, recipients, subject, body) + ) + + def strong_email_fingerprint( *, sender: str | None, @@ -30,12 +164,19 @@ def strong_email_fingerprint( date: datetime.datetime | None, body: str | None, ) -> str | None: - if not body: + """Return the strong (sender+subject+Date+body) auto-dedupe fingerprint. + + The fingerprint itself preserves the established four-field contract. It is + withheld when sender, subject, or body is missing; callers that possess + recipient evidence additionally apply ``has_complete_strong_email_metadata`` + before treating this hash as automatic-link evidence. + """ + if not all(_has_nonempty_text(value) for value in (sender, subject, body)): return None return generate_email_fingerprint( { - "sender": sender or "", - "subject": subject or "", + "sender": sender, + "subject": subject, "date": _date_to_fingerprint_value(date), "body": body, } @@ -43,6 +184,7 @@ def strong_email_fingerprint( def candidate_message_lookup_values(candidate: EmailDedupeCandidate) -> set[str]: + """Return the bracketed and bare Message-ID lookup forms (empty if none).""" normalized = normalize_message_id(candidate.message_id) if not normalized: return set() @@ -50,6 +192,14 @@ def candidate_message_lookup_values(candidate: EmailDedupeCandidate) -> set[str] def candidate_strong_fingerprint(candidate: EmailDedupeCandidate) -> str | None: + """Return strong evidence only when the candidate metadata set is complete.""" + if not has_complete_strong_email_metadata( + sender=candidate.sender, + recipients=candidate.recipients, + subject=candidate.subject, + body=candidate.body, + ): + return None return strong_email_fingerprint( sender=candidate.sender, subject=candidate.subject, @@ -59,9 +209,146 @@ def candidate_strong_fingerprint(candidate: EmailDedupeCandidate) -> str | None: def email_strong_fingerprint(email_row: Email) -> str | None: + """Return a stored row's strong fingerprint, gated on trustworthy metadata. + + A stored row may seed a strong (auto-dedupe) fingerprint only when its date + is genuinely parsed sender metadata and sender/recipient/subject/body evidence + is complete. Rows that fail either condition remain eligible for weaker + review/source-bound identity paths but cannot manufacture an automatic link. + """ + if getattr(email_row, "date_provenance", None) != "parsed": + return None + if not has_complete_strong_email_metadata( + sender=getattr(email_row, "sender", None), + recipients=getattr(email_row, "recipients", None), + subject=getattr(email_row, "subject", None), + body=getattr(email_row, "body", None), + ): + return None return strong_email_fingerprint( sender=email_row.sender, subject=email_row.subject, date=email_row.date, body=email_row.body, ) + + +def content_email_fingerprint( + *, + sender: str | None, + subject: str | None, + body: str | None, +) -> str | None: + """Return a Date-independent content fingerprint (sender+subject+body). + + Unlike the strong fingerprint it omits the Date, so it survives an + untrustworthy Date provenance (naruon#1086) and can flag a probable + duplicate that the strong path deliberately withholds. Requires a body so + empty-body rows cannot collapse to a shared hash. + """ + if not body: + return None + return generate_email_fingerprint( + { + "sender": sender or "", + "subject": subject or "", + "date": "", + "body": body, + } + ) + + +def candidate_content_fingerprint(candidate: EmailDedupeCandidate) -> str | None: + """Return the candidate's Date-independent content fingerprint.""" + return content_email_fingerprint( + sender=candidate.sender, + subject=candidate.subject, + body=candidate.body, + ) + + +def email_content_fingerprint(email_row: Email) -> str | None: + """Return a stored row's Date-independent content fingerprint.""" + return content_email_fingerprint( + sender=email_row.sender, + subject=email_row.subject, + body=email_row.body, + ) + + +def classify_dedupe_decision( + candidate: EmailDedupeCandidate, existing_row: Email +) -> DedupeDecision: + """Assign a Fellegi-Sunter (1969) decision zone to a candidate/existing pair. + + - ``auto_link`` (A1, positive link): the pair shares a reliable identity + link -- the same normalized Message-ID, or a genuine strong match + (identical sender/subject/Date/body with a trusted, parsed Date on *both* + sides and complete sender/recipient/subject/body metadata). These are safe + to merge automatically. + - ``review_required`` (A2, possible match): the pair shares a + provenance-independent content signal (same sender/subject/body) but has + no reliable identity link -- typically because at least one side's Date + provenance is synthetic/unknown or strong metadata is incomplete, so the + strong fingerprint was withheld (naruon#1086). This is the clerical-review + band: a probable duplicate that must not be silently merged or silently kept. + - ``distinct`` (A3, non-link): no shared identity or content signal. + """ + candidate_message = normalize_message_id(candidate.message_id) + existing_message = normalize_message_id(existing_row.message_id) + if candidate_message and existing_message and candidate_message == existing_message: + return "auto_link" + + candidate_strong = candidate_strong_fingerprint(candidate) + existing_strong = email_strong_fingerprint(existing_row) + if ( + candidate.date_provenance == "parsed" + and candidate_strong is not None + and existing_strong is not None + and candidate_strong == existing_strong + ): + return "auto_link" + + candidate_content = candidate_content_fingerprint(candidate) + existing_content = email_content_fingerprint(existing_row) + if ( + candidate_content is not None + and existing_content is not None + and candidate_content == existing_content + ): + return "review_required" + + return "distinct" + + +def resolve_candidate_disposition( + candidate: EmailDedupeCandidate, existing_rows: Iterable[Email] +) -> tuple[DedupeDecision, Email | None]: + """Resolve a candidate against many stored rows to one Fellegi-Sunter disposition. + + Real de-duplication compares one incoming email against the *set* of stored + rows it might duplicate, not a single row, so this collapses the per-pair + ``classify_dedupe_decision`` results by the Fellegi & Sunter (1969) zone + priority A1 > A2 > A3: + + - the first stored row that yields ``auto_link`` (A1, a reliable identity + link) wins immediately -- a positive link cannot be outranked; + - absent any link, the first ``review_required`` match (A2) is held for + clerical review rather than silently merged or silently kept; + - only when no stored row shares any identity or content signal is the + candidate ``distinct`` (A3). + + Returns the decision together with the stored row that drove a link or a + review hold (``None`` when distinct), so the import/IMAP paths know which + email the disposition targets without re-deriving the match. + """ + review_match: Email | None = None + for existing_row in existing_rows: + decision = classify_dedupe_decision(candidate, existing_row) + if decision == "auto_link": + return "auto_link", existing_row + if decision == "review_required" and review_match is None: + review_match = existing_row + if review_match is not None: + return "review_required", review_match + return "distinct", None diff --git a/backend/services/email_import_service.py b/backend/services/email_import_service.py index ddfa350fd..9685e9166 100644 --- a/backend/services/email_import_service.py +++ b/backend/services/email_import_service.py @@ -31,7 +31,12 @@ try_batch_import_embeddings, ) from services.content_graph import ParseResult, parse_content -from services.email_dedupe_service import strong_email_fingerprint +from services.email_dedupe_service import ( + canonical_email_source_content, + has_complete_strong_email_metadata, + source_email_fingerprint, + strong_email_fingerprint, +) from services.email_parser import EmailData, parse_eml_bytes from services.embedding import ( STORAGE_EMBEDDING_DIMENSION, @@ -50,7 +55,6 @@ ) from services.threading_service import ( assign_thread_id, - generate_email_fingerprint, normalize_message_id, ) @@ -194,20 +198,50 @@ def _message_id_for(parsed: EmailData, content: bytes) -> str: ) -def _email_fingerprint(parsed: EmailData, persisted_date: datetime.datetime) -> str: - strong_fingerprint = strong_email_fingerprint( +def _has_strong_email_metadata(parsed: EmailData) -> bool: + """Return whether parsed metadata may drive an automatic strong fingerprint.""" + return parsed.get("date_provenance") == "parsed" and has_complete_strong_email_metadata( sender=parsed.get("sender"), + recipients=parsed.get("recipients"), subject=parsed.get("subject"), - date=persisted_date, body=parsed.get("body"), ) + + +def _dedupe_review_reason(parsed: EmailData) -> str | None: + """Expose when an import lacks enough metadata for strong automatic linking.""" + return None if _has_strong_email_metadata(parsed) else "dedupe_review_required" + + +def _email_fingerprint( + parsed: EmailData, + persisted_date: datetime.datetime, + source_content: bytes | None = None, +) -> str: + """Return trusted complete-metadata evidence or source-bound fallback identity. + + ``persisted_date`` remains the storage timestamp and participates in + duplicate evidence only when it came from a valid sender ``Date`` and the + sender/recipient/subject/body metadata evidence set is complete. + """ + strong_fingerprint = None + if _has_strong_email_metadata(parsed): + strong_fingerprint = strong_email_fingerprint( + sender=parsed.get("sender"), + subject=parsed.get("subject"), + date=persisted_date, + body=parsed.get("body"), + ) if strong_fingerprint: return strong_fingerprint - return generate_email_fingerprint( - parsed.get("subject"), - persisted_date.isoformat(), - parsed.get("sender"), - parsed.get("recipients"), + source_identity = ( + source_content + if source_content is not None + else canonical_email_source_content(parsed) + ) + return source_email_fingerprint( + source_identity, + source_kind="raw" if source_content is not None else "canonical", ) @@ -370,6 +404,7 @@ def _build_email_object( in_reply_to=parsed.get("in_reply_to"), references=parsed.get("references"), date=persisted_date, + date_provenance=parsed.get("date_provenance", "unknown"), body=parsed.get("body", ""), embedding=fitted_embeddings[0] if fitted_embeddings else _zero_embedding(), ) @@ -862,7 +897,7 @@ async def _import_single_eml( message_id = _message_id_for(parsed, content) parsed["message_id"] = message_id persisted_date = _utc_datetime(parsed.get("date")) - fingerprint = _email_fingerprint(parsed, persisted_date) + fingerprint = _email_fingerprint(parsed, persisted_date, content) existing_email = await _find_existing_email( session, @@ -933,6 +968,7 @@ async def _import_single_eml( return EmailImportItemResult( filename=display_filename, status="imported", + reason_code=_dedupe_review_reason(parsed), attachment_count=attachment_count, ) diff --git a/backend/services/email_parser.py b/backend/services/email_parser.py index be8bee1c4..b129d3d4e 100644 --- a/backend/services/email_parser.py +++ b/backend/services/email_parser.py @@ -5,12 +5,20 @@ import re from email.utils import getaddresses from email.utils import parsedate_to_datetime -from typing import NotRequired, TypedDict +from typing import Literal, NotRequired, TypedDict from .attachment_parser import parse_email_attachment from .exceptions import EmailParseError from .text_safety import strip_html_markup +# Provenance of the RFC822 ``Date`` header, kept explicit so a synthetic +# collection-time fallback is never mistaken for original sender metadata. +DateProvenance = Literal["parsed", "missing", "invalid"] + +# Provenance of the RFC822 ``Message-ID`` header. +MessageIdProvenance = Literal["embedded", "missing"] + + class EmailData(TypedDict): """Parsed email data structure.""" @@ -23,6 +31,9 @@ class EmailData(TypedDict): in_reply_to: str | None references: str | None date: datetime.datetime + header_date: NotRequired[datetime.datetime | None] + date_provenance: NotRequired[DateProvenance] + message_id_provenance: NotRequired[MessageIdProvenance] body: str body_content_type: NotRequired[str] body_parse_content: NotRequired[str] @@ -44,6 +55,12 @@ def _sanitize_display_text(text: str) -> str: # that force a quoted-string, and the characters escaped inside one. _ADDRESS_SPECIALS_RE = re.compile(r'[()<>@,;:\\".\[\]]') _ADDRESS_QUOTED_ESCAPE_RE = re.compile(r'["\\]') +# ``parsedate_to_datetime`` also accepts some zone-less values and returns a +# naive datetime. Only a Date that actually carries an RFC 5322 zone may be +# promoted to sender-supplied provenance when the parser result is naive. +_RFC5322_TRAILING_ZONE_RE = re.compile( + r"(?:[+-]\d{4}|[A-Za-z]{1,5})(?:\s*\([^)]*\))?\s*$" +) def _format_display_address(display_name: str, address: str) -> str: @@ -152,26 +169,43 @@ def _extract_body_and_attachments(msg: Message) -> tuple[str, str, list[dict]]: return html_body, "text/html" if html_body else "text/plain", attachments -def _extract_date(msg: Message) -> datetime.datetime: +def _extract_date_with_provenance( + msg: Message, +) -> tuple[datetime.datetime, datetime.datetime | None, DateProvenance]: + """Return effective date, genuine header date, and header provenance. + + The effective value is always timezone-aware and safe for storage. A + missing or invalid header receives a UTC collection-time fallback, while + ``header_date`` remains ``None`` so deduplication cannot treat that fallback + as sender-supplied evidence. + """ date_header = msg.get("Date") - parsed_date = None - if date_header: - try: - parsed_date = parsedate_to_datetime(date_header) - except (TypeError, ValueError): - parsed_date = None - - if not parsed_date: - parsed_date = datetime.datetime.now(datetime.timezone.utc) - elif parsed_date.tzinfo is None: - # RFC 5322 section 3.3: a "-0000" zone means the time zone is unknown, - # for which parsedate_to_datetime returns a naive datetime. Every other - # branch here yields a timezone-aware datetime, and mixing naive with - # aware datetimes raises TypeError on comparison/sorting and misbinds the - # instant when stored in a timestamptz column. Treat the unknown zone as - # UTC so the returned value is always timezone-aware. - parsed_date = parsed_date.replace(tzinfo=datetime.timezone.utc) - return parsed_date + header_text = str(date_header).strip() if date_header is not None else "" + fallback = datetime.datetime.now(datetime.timezone.utc) + + if not header_text: + return fallback, None, "missing" + + try: + header_date = parsedate_to_datetime(date_header) + except (TypeError, ValueError): + header_date = None + + if header_date is None: + return fallback, None, "invalid" + if header_date.tzinfo is None: + # ``-0000`` and obsolete alphabetic zones can yield a naive datetime, + # but a parseable value with no zone is not complete RFC 5322 Date + # evidence and must never seed a strong duplicate identity. + if not _RFC5322_TRAILING_ZONE_RE.search(header_text): + return fallback, None, "invalid" + header_date = header_date.replace(tzinfo=datetime.timezone.utc) + return header_date, header_date, "parsed" + + +def _message_id_provenance(raw_message_id: str) -> MessageIdProvenance: + """Classify whether a genuine ``Message-ID`` header was embedded.""" + return "embedded" if raw_message_id.strip() else "missing" def _extract_thread_id(msg: Message, message_id: str) -> str | None: @@ -193,8 +227,9 @@ def _extract_thread_id(msg: Message, message_id: str) -> str | None: def _message_to_email_data(msg: Message) -> EmailData: body, body_content_type, attachments = _extract_body_and_attachments(msg) - parsed_date = _extract_date(msg) + effective_date, header_date, date_provenance = _extract_date_with_provenance(msg) message_id = _sanitize_nul(msg.get("Message-ID", "")) + message_id_provenance = _message_id_provenance(message_id) thread_id = _extract_thread_id(msg, message_id) return { @@ -216,7 +251,10 @@ def _message_to_email_data(msg: Message) -> EmailData: "references": ( _sanitize_nul(msg.get("References", "")) if msg.get("References") else None ), - "date": parsed_date, + "date": effective_date, + "header_date": header_date, + "date_provenance": date_provenance, + "message_id_provenance": message_id_provenance, "body": _sanitize_display_text(body), "body_content_type": body_content_type, "body_parse_content": _sanitize_nul(body), diff --git a/backend/services/imap_worker.py b/backend/services/imap_worker.py index d618f1d3c..eb800ae1e 100644 --- a/backend/services/imap_worker.py +++ b/backend/services/imap_worker.py @@ -10,30 +10,41 @@ from db.models import Email, TenantConfig from db.session import AsyncSessionLocal from services.email_client import validate_imap_destination -from services.email_dedupe_service import strong_email_fingerprint +from services.email_dedupe_service import ( + canonical_email_source_content, + has_complete_strong_email_metadata, + source_email_fingerprint, + strong_email_fingerprint, +) from services.email_parser import EmailData, parse_eml_bytes from services.exceptions import EmailParseError from services.knowledge_extractor import ( extract_knowledge_from_self_sent, is_self_sent_email, ) -from services.threading_service import assign_thread_id, generate_email_fingerprint +from services.threading_service import assign_thread_id -async def process_fetched_email( +@dataclass(frozen=True, slots=True) +class FetchedEmailPersistenceResult: + """Report the persisted email and whether this sync created it.""" + + email_record: Email + created_record: bool + + +async def persist_fetched_email( session, email_data: EmailData, user_id: str, organization_id: str | None, owner_addresses: Iterable[str] | None = None, is_read: bool = True, -): + source_content: bytes | None = None, +) -> FetchedEmailPersistenceResult: + """Persist one fetched email and retain duplicate disposition for sync counts.""" subject = email_data.get("subject", "") date_obj = email_data.get("date") - if hasattr(date_obj, "isoformat"): - date_str = date_obj.isoformat() - else: - date_str = str(date_obj) if date_obj else "" if isinstance(date_obj, datetime.datetime): persisted_date = ( date_obj.astimezone(datetime.timezone.utc) @@ -49,15 +60,37 @@ async def process_fetched_email( if isinstance(recipients_list, list) else str(recipients_list or "") ) + body = email_data.get("body", "") + + # Metadata-based auto-linking needs both genuine Date provenance and the full + # sender/recipient/subject/body evidence set. Incomplete messages keep their + # raw/canonical source identity instead of manufacturing a strong match. + strong_fingerprint = None + if ( + email_data.get("date_provenance") == "parsed" + and has_complete_strong_email_metadata( + sender=sender, + recipients=recipients, + subject=subject, + body=body, + ) + ): + strong_fingerprint = strong_email_fingerprint( + sender=sender, + subject=subject, + date=persisted_date, + body=body, + ) + source_identity = ( + source_content + if source_content is not None + else canonical_email_source_content(email_data) + ) + fingerprint = strong_fingerprint or source_email_fingerprint( + source_identity, + source_kind="raw" if source_content is not None else "canonical", + ) - fingerprint = strong_email_fingerprint( - sender=sender, - subject=subject, - date=persisted_date, - body=email_data.get("body", ""), - ) or generate_email_fingerprint(subject, date_str, sender, recipients) - - # Check if duplicate stmt = select(Email).where( Email.user_id == user_id, Email.organization_id == (organization_id if organization_id else None), @@ -71,7 +104,10 @@ async def process_fetched_email( "Email with fingerprint %s already exists. Skipping duplicate insertion.", fingerprint, ) - return existing_email + return FetchedEmailPersistenceResult( + email_record=existing_email, + created_record=False, + ) thread_id = await assign_thread_id( session, email_data, user_id=user_id, organization_id=organization_id @@ -87,7 +123,8 @@ async def process_fetched_email( recipients=recipients, subject=subject, date=persisted_date, - body=email_data.get("body", ""), + date_provenance=email_data.get("date_provenance", "unknown"), + body=body, is_read=is_read, embedding=[0.0] * 1536, ) @@ -96,7 +133,33 @@ async def process_fetched_email( if is_self_sent_email(new_email, owner_addresses): await session.flush() await extract_knowledge_from_self_sent(session, new_email, owner_addresses) - return new_email + return FetchedEmailPersistenceResult( + email_record=new_email, + created_record=True, + ) + + +async def process_fetched_email( + session, + email_data: EmailData, + user_id: str, + organization_id: str | None, + owner_addresses: Iterable[str] | None = None, + is_read: bool = True, + source_content: bytes | None = None, +) -> Email: + """Persist one fetched email while preserving the legacy Email return contract.""" + persistence_result = await persist_fetched_email( + session, + email_data, + user_id, + organization_id, + owner_addresses=owner_addresses, + is_read=is_read, + source_content=source_content, + ) + return persistence_result.email_record + logger = logging.getLogger(__name__) MAX_IMAP_FETCH_MESSAGES = 10 @@ -112,7 +175,11 @@ def flags_indicate_seen(fetch_data) -> bool: for item in fetch_data or []: parts = item if isinstance(item, (tuple, list)) else (item,) for part in parts: - raw = part if isinstance(part, bytes) else str(part).encode("utf-8", "replace") + raw = ( + part + if isinstance(part, bytes) + else str(part).encode("utf-8", "replace") + ) upper = raw.upper() if b"FLAGS" in upper and b"\\SEEN" in upper: return True @@ -217,7 +284,7 @@ async def _sync_tenant(self, config: TenantConfig | ImapSyncConfig): config.user_id, ) return 0 - + logger.info( "Connecting to IMAP server %s:%s for user %s", imap_server, @@ -252,6 +319,7 @@ async def _fetch_messages( if imap_server is None or imap_port is None: imap_server, imap_port = self._validated_destination(config) import ssl + ssl_context = ssl.create_default_context() imap_client = aioimaplib.IMAP4_SSL( imap_server, imap_port, ssl_context=ssl_context @@ -327,15 +395,17 @@ async def _import_messages( config.user_id, ) continue - await process_fetched_email( + persistence_result = await persist_fetched_email( session, email_data, config.user_id, config.organization_id, owner_addresses=owner_addresses, is_read=is_read, + source_content=raw_message, ) - imported_count += 1 + if persistence_result.created_record: + imported_count += 1 await session.commit() except Exception: await session.rollback() @@ -388,6 +458,4 @@ def _looks_like_rfc822_message(self, value: bytes) -> bool: header_block = value.split(b"\r\n\r\n", maxsplit=1)[0] if header_block == value: header_block = value.split(b"\n\n", maxsplit=1)[0] - return b":" in header_block and ( - b"\r\n\r\n" in value or b"\n\n" in value - ) + return b":" in header_block and (b"\r\n\r\n" in value or b"\n\n" in value) diff --git a/backend/services/pop3_worker.py b/backend/services/pop3_worker.py index 601180027..c1de39d34 100644 --- a/backend/services/pop3_worker.py +++ b/backend/services/pop3_worker.py @@ -1,16 +1,57 @@ import asyncio +from dataclasses import dataclass +import datetime import logging import poplib +from typing import Literal, Mapping + from sqlalchemy import select -from db.session import AsyncSessionLocal +from sqlalchemy.dialects.postgresql import insert as pg_insert + from db.models import TenantConfig +from db.pop3_collection_models import Pop3ObservedMessage +from db.session import AsyncSessionLocal from services.email_client import validate_pop3_destination from services.email_parser import parse_eml_bytes from services.exceptions import EmailParseError -from services.imap_worker import process_fetched_email +from services.imap_worker import persist_fetched_email logger = logging.getLogger(__name__) MAX_POP3_FETCH_MESSAGES = 10 +POP3_RETRY_DELAY = datetime.timedelta(seconds=60) +Pop3CollectionDisposition = Literal["observed", "retryable"] + + +@dataclass(frozen=True) +class Pop3MessageIdentity: + """Current-session POP3 number bound to its RFC 1939 durable unique-id.""" + + message_number: int + provider_uidl: str + + +@dataclass(frozen=True) +class Pop3RetrievedMessage: + """Retrieved POP3 source bytes plus optional provider progress identity.""" + + source_content: bytes + provider_uidl: str | None = None + + +@dataclass(frozen=True) +class Pop3CollectionProgressState: + """Durable collection disposition for one owner-scoped provider UIDL.""" + + disposition: Pop3CollectionDisposition + retry_after: datetime.datetime | None + + +@dataclass(frozen=True) +class Pop3SyncBatch: + """Network retrieval result plus retryable UIDLs awaiting durable disposition.""" + + messages: list[Pop3RetrievedMessage] + retryable_uidls: frozenset[str] class Pop3SyncWorker: @@ -22,7 +63,6 @@ async def start(self): if self._is_running: logger.warning("Pop3SyncWorker is already running.") return - self._is_running = True self._task = asyncio.create_task(self._run_loop()) logger.info("Pop3SyncWorker started.") @@ -30,7 +70,6 @@ async def start(self): async def stop(self): if not self._is_running: return - self._is_running = False if self._task: self._task.cancel() @@ -48,7 +87,6 @@ async def _run_loop(self): break except Exception as e: logger.error(f"Error in Pop3SyncWorker loop: {e}", exc_info=True) - if self._is_running: try: await asyncio.sleep(60) @@ -57,16 +95,17 @@ async def _run_loop(self): async def _sync(self): async with AsyncSessionLocal() as session: - result = await session.execute(select(TenantConfig).where(TenantConfig.pop3_server.isnot(None))) + result = await session.execute( + select(TenantConfig).where(TenantConfig.pop3_server.isnot(None)) + ) configs = result.scalars().all() - + semaphore = asyncio.Semaphore(10) tasks = [] for config in configs: if not config.pop3_server or not config.pop3_port: continue tasks.append(self._sync_tenant(config, semaphore)) - if tasks: await asyncio.gather(*tasks, return_exceptions=True) @@ -84,11 +123,19 @@ async def _sync_tenant(self, config: TenantConfig, semaphore: asyncio.Semaphore) f"Connecting to POP3 server {pop3_server}:{pop3_port} for user {config.user_id}" ) try: - # We use asyncio.to_thread for synchronous poplib - messages = await asyncio.to_thread( - self._do_pop3_sync, config, pop3_server, pop3_port + collection_progress = await self._load_collection_progress(config) + batch = await asyncio.to_thread( + self._do_pop3_sync_batch, + config, + pop3_server, + pop3_port, + collection_progress, + ) + imported_count = await self._import_messages( + config, + batch.messages, + retryable_uidls=batch.retryable_uidls, ) - imported_count = await self._import_messages(config, messages) logger.info( "Successfully synced POP3 server for user %s with %s imported messages.", config.user_id, @@ -101,17 +148,69 @@ async def _sync_tenant(self, config: TenantConfig, semaphore: asyncio.Semaphore) type(e).__name__, ) + async def _load_collection_progress( + self, config: TenantConfig + ) -> dict[str, Pop3CollectionProgressState]: + """Load durable UIDL collection state before provider network I/O begins.""" + if config.id is None: + return {} + async with AsyncSessionLocal() as session: + result = await session.execute( + select( + Pop3ObservedMessage.provider_uidl, + Pop3ObservedMessage.collection_disposition, + Pop3ObservedMessage.retry_after, + ).where(Pop3ObservedMessage.tenant_config_id == config.id) + ) + return { + provider_uidl: Pop3CollectionProgressState( + disposition=disposition, + retry_after=retry_after, + ) + for provider_uidl, disposition, retry_after in result.all() + } + + async def _load_observed_uidls(self, config: TenantConfig) -> set[str]: + """Compatibility view over durable collection state for observed UIDLs.""" + progress = await self._load_collection_progress(config) + return { + provider_uidl + for provider_uidl, state in progress.items() + if state.disposition == "observed" + } + async def _import_messages( - self, config: TenantConfig, messages: list[bytes] + self, + config: TenantConfig, + messages: list[Pop3RetrievedMessage | bytes], + *, + retryable_uidls: frozenset[str] | set[str] | None = None, ) -> int: - if not messages: + retryable_uidls = retryable_uidls or frozenset() + if not messages and not retryable_uidls: return 0 imported_count = 0 owner_addresses = [config.pop3_username] if config.pop3_username else None + state_time = datetime.datetime.now(datetime.timezone.utc) async with AsyncSessionLocal() as session: try: - for raw_message in messages: + if config.id is not None: + for provider_uidl in retryable_uidls: + await self._upsert_collection_progress( + session, + config.id, + provider_uidl, + disposition="retryable", + state_time=state_time, + ) + for message in messages: + if isinstance(message, bytes): + raw_message = message + provider_uidl = None + else: + raw_message = message.source_content + provider_uidl = message.provider_uidl try: email_data = parse_eml_bytes(raw_message) except EmailParseError: @@ -119,21 +218,72 @@ async def _import_messages( "Skipping unparsable POP3 message for user %s.", config.user_id, ) + if provider_uidl is not None and config.id is not None: + await self._upsert_collection_progress( + session, + config.id, + provider_uidl, + disposition="retryable", + state_time=state_time, + ) continue - await process_fetched_email( + persistence_result = await persist_fetched_email( session, email_data, config.user_id, config.organization_id, owner_addresses=owner_addresses, + source_content=raw_message, ) - imported_count += 1 + if persistence_result.created_record: + imported_count += 1 + if provider_uidl is not None and config.id is not None: + await self._upsert_collection_progress( + session, + config.id, + provider_uidl, + disposition="observed", + state_time=state_time, + ) await session.commit() except Exception: await session.rollback() raise return imported_count + async def _upsert_collection_progress( + self, + session, + tenant_config_id: int, + provider_uidl: str, + *, + disposition: Pop3CollectionDisposition, + state_time: datetime.datetime, + ) -> None: + retry_after = ( + state_time + POP3_RETRY_DELAY if disposition == "retryable" else None + ) + observed_at = state_time if disposition == "observed" else None + statement = ( + pg_insert(Pop3ObservedMessage) + .values( + tenant_config_id=tenant_config_id, + provider_uidl=provider_uidl, + collection_disposition=disposition, + retry_after=retry_after, + observed_at=observed_at, + ) + .on_conflict_do_update( + index_elements=["tenant_config_id", "provider_uidl"], + set_={ + "collection_disposition": disposition, + "retry_after": retry_after, + "observed_at": observed_at, + }, + ) + ) + await session.execute(statement) + def _validated_destination(self, config: TenantConfig) -> tuple[str, int]: return validate_pop3_destination( str(config.pop3_server), @@ -145,10 +295,33 @@ def _do_pop3_sync( config: TenantConfig, pop3_server: str | None = None, pop3_port: int | None = None, - ) -> list[bytes]: + observed_uidls: set[str] | None = None, + ) -> list[Pop3RetrievedMessage]: + collection_progress = { + provider_uidl: Pop3CollectionProgressState( + disposition="observed", + retry_after=None, + ) + for provider_uidl in (observed_uidls or set()) + } + return self._do_pop3_sync_batch( + config, + pop3_server, + pop3_port, + collection_progress, + ).messages + + def _do_pop3_sync_batch( + self, + config: TenantConfig, + pop3_server: str | None = None, + pop3_port: int | None = None, + collection_progress: Mapping[str, Pop3CollectionProgressState] | None = None, + ) -> Pop3SyncBatch: if pop3_server is None or pop3_port is None: pop3_server, pop3_port = self._validated_destination(config) pop3_client = poplib.POP3_SSL(pop3_server, pop3_port) + collection_progress = collection_progress or {} try: if not config.pop3_username: logger.error( @@ -168,17 +341,236 @@ def _do_pop3_sync( ) pop3_client.user(config.pop3_username) pop3_client.pass_(config.pop3_password) + + uidl_identities = self._current_uidl_identities(pop3_client, config) + if uidl_identities is not None: + selected = self._select_uidl_candidates( + uidl_identities, + collection_progress, + datetime.datetime.now(datetime.timezone.utc), + ) + return self._retrieve_uidl_batch(pop3_client, config, selected) + _response, listings, _octets = pop3_client.list() - messages: list[bytes] = [] - for listing in listings[:MAX_POP3_FETCH_MESSAGES]: - message_number = self._message_number_from_listing(listing) - if message_number is None: - continue - _retr_response, lines, _retr_octets = pop3_client.retr(message_number) - messages.append(b"\r\n".join(self._bytes_line(line) for line in lines)) - return messages + message_numbers = [ + message_number + for listing in listings + if (message_number := self._message_number_from_listing(listing)) + is not None + ] + logger.warning( + "POP3 UIDL unavailable for user %s; bounded fallback cannot prove durable backlog progress.", + config.user_id, + ) + return Pop3SyncBatch( + messages=self._retrieve_fallback_messages( + pop3_client, + config, + sorted(message_numbers)[-MAX_POP3_FETCH_MESSAGES:], + ), + retryable_uidls=frozenset(), + ) finally: + self._close_pop3_client(pop3_client, config) + + def _select_uidl_candidates( + self, + identities: list[Pop3MessageIdentity], + collection_progress: Mapping[str, Pop3CollectionProgressState], + now: datetime.datetime, + ) -> list[Pop3MessageIdentity]: + """Prefer never-attempted UIDLs, then retries whose retry window is due.""" + fresh: list[Pop3MessageIdentity] = [] + due_retries: list[Pop3MessageIdentity] = [] + for identity in sorted( + identities, + key=lambda candidate: candidate.message_number, + reverse=True, + ): + state = collection_progress.get(identity.provider_uidl) + if state is None: + fresh.append(identity) + continue + if state.disposition == "observed": + continue + if state.retry_after is None or state.retry_after <= now: + due_retries.append(identity) + return (fresh + due_retries)[:MAX_POP3_FETCH_MESSAGES] + + def _current_uidl_identities( + self, + pop3_client: poplib.POP3_SSL, + config: TenantConfig, + ) -> list[Pop3MessageIdentity] | None: + try: + _response, listings, _octets = pop3_client.uidl() + except poplib.error_proto: + return None + + identities: list[Pop3MessageIdentity] = [] + for listing in listings: + identity = self._uidl_identity_from_listing(listing) + if identity is None: + logger.warning( + "POP3 UIDL response contained an invalid entry for user %s; durable progress is unavailable for this poll.", + config.user_id, + ) + return None + identities.append(identity) + return identities + + def _uidl_identity_from_listing( + self, listing: bytes | str + ) -> Pop3MessageIdentity | None: + try: + raw_listing = listing.decode("ascii") if isinstance(listing, bytes) else listing + except UnicodeDecodeError: + return None + parts = raw_listing.strip().split() + if len(parts) != 2 or not parts[0].isdigit(): + return None + provider_uidl = parts[1] + if not 1 <= len(provider_uidl) <= 70: + return None + if any(not 0x21 <= ord(character) <= 0x7E for character in provider_uidl): + return None + return Pop3MessageIdentity( + message_number=int(parts[0]), + provider_uidl=provider_uidl, + ) + + def _retrieve_uidl_batch( + self, + pop3_client: poplib.POP3_SSL, + config: TenantConfig, + identities: list[Pop3MessageIdentity], + ) -> Pop3SyncBatch: + messages: list[Pop3RetrievedMessage] = [] + retryable_uidls: set[str] = set() + for identity in identities: + try: + source_content = self._retrieve_message( + pop3_client, identity.message_number + ) + except poplib.error_proto as exc: + if not self._is_negative_pop3_response(exc): + logger.warning( + "POP3 RETR stopped after malformed protocol response for user %s: %s", + config.user_id, + type(exc).__name__, + ) + break + logger.warning( + "POP3 RETR rejected one message for user %s; continuing bounded batch: %s", + config.user_id, + type(exc).__name__, + ) + retryable_uidls.add(identity.provider_uidl) + continue + except OSError as exc: + logger.warning( + "POP3 RETR stopped after transport failure for user %s: %s", + config.user_id, + type(exc).__name__, + ) + break + messages.append( + Pop3RetrievedMessage( + source_content=source_content, + provider_uidl=identity.provider_uidl, + ) + ) + return Pop3SyncBatch( + messages=messages, + retryable_uidls=frozenset(retryable_uidls), + ) + + def _retrieve_uidl_messages( + self, + pop3_client: poplib.POP3_SSL, + config: TenantConfig, + identities: list[Pop3MessageIdentity], + ) -> list[Pop3RetrievedMessage]: + """Compatibility wrapper retaining the historical direct-return contract.""" + return self._retrieve_uidl_batch(pop3_client, config, identities).messages + + def _retrieve_fallback_messages( + self, + pop3_client: poplib.POP3_SSL, + config: TenantConfig, + message_numbers: list[int], + ) -> list[Pop3RetrievedMessage]: + messages: list[Pop3RetrievedMessage] = [] + for message_number in message_numbers: + try: + source_content = self._retrieve_message(pop3_client, message_number) + except poplib.error_proto as exc: + if not self._is_negative_pop3_response(exc): + logger.warning( + "POP3 fallback RETR stopped after malformed protocol response for user %s: %s", + config.user_id, + type(exc).__name__, + ) + break + logger.warning( + "POP3 fallback RETR rejected one message for user %s; continuing bounded batch: %s", + config.user_id, + type(exc).__name__, + ) + continue + except OSError as exc: + logger.warning( + "POP3 fallback RETR stopped after transport failure for user %s: %s", + config.user_id, + type(exc).__name__, + ) + break + messages.append( + Pop3RetrievedMessage( + source_content=source_content, + provider_uidl=None, + ) + ) + return messages + + def _retrieve_message(self, pop3_client: poplib.POP3_SSL, message_number: int) -> bytes: + _retr_response, lines, _retr_octets = pop3_client.retr(message_number) + return self._message_bytes(lines) + + def _is_negative_pop3_response(self, error: poplib.error_proto) -> bool: + """Return whether a protocol exception carries an RFC 1939 -ERR reply.""" + if not error.args: + return False + response = error.args[0] + if isinstance(response, bytes): + return response.startswith(b"-ERR") + return str(response).startswith("-ERR") + + def _close_pop3_client( + self, + pop3_client: poplib.POP3_SSL, + config: TenantConfig, + ) -> None: + try: pop3_client.quit() + except (OSError, poplib.error_proto) as exc: + logger.warning( + "POP3 QUIT cleanup failed for user %s: %s", + config.user_id, + type(exc).__name__, + ) + try: + pop3_client.close() + except OSError as close_exc: + logger.warning( + "POP3 transport close failed for user %s: %s", + config.user_id, + type(close_exc).__name__, + ) + + def _message_bytes(self, lines: list[bytes | str]) -> bytes: + """Reconstruct one POP3 RETR message with protocol CRLF terminators.""" + return b"\r\n".join(self._bytes_line(line) for line in lines) + b"\r\n" def _message_number_from_listing(self, listing: bytes | str) -> int | None: raw_listing = ( @@ -195,7 +587,5 @@ def _message_number_from_listing(self, listing: bytes | str) -> int | None: def _bytes_line(self, line: bytes | str) -> bytes: return ( - line - if isinstance(line, bytes) - else line.encode("utf-8", errors="replace") + line if isinstance(line, bytes) else line.encode("utf-8", errors="replace") ) diff --git a/backend/tests/test_email_date_provenance_migration_contract.py b/backend/tests/test_email_date_provenance_migration_contract.py new file mode 100644 index 000000000..1c62caf60 --- /dev/null +++ b/backend/tests/test_email_date_provenance_migration_contract.py @@ -0,0 +1,37 @@ +from pathlib import Path + + +BACKEND_ROOT = Path(__file__).resolve().parents[1] + + +def test_canonical_email_date_provenance_migration_contract() -> None: + revision_path = ( + BACKEND_ROOT + / "alembic" + / "versions" + / "0018_email_date_provenance.py" + ) + revision_text = revision_path.read_text() + + assert 'revision = "0018_email_date_provenance"' in revision_text + assert 'down_revision = "0017_merge_newsdom_carddav_heads"' in revision_text + assert '_EMAIL_TABLE = "email_records"' in revision_text + assert '_PROVENANCE_COLUMN = "date_provenance"' in revision_text + assert "nullable=False" in revision_text + assert 'server_default="unknown"' in revision_text + + assert 'op.add_column(' in revision_text + assert 'op.drop_column(_EMAIL_TABLE, _PROVENANCE_COLUMN)' in revision_text + + +def test_canonical_email_provenance_does_not_fork_parallel_evidence_columns() -> None: + revision_path = ( + BACKEND_ROOT + / "alembic" + / "versions" + / "0018_email_date_provenance.py" + ) + revision_text = revision_path.read_text() + + assert "date_evidence" not in revision_text + assert "message_id_evidence" not in revision_text diff --git a/backend/tests/test_email_dedupe_complete_metadata.py b/backend/tests/test_email_dedupe_complete_metadata.py new file mode 100644 index 000000000..1f6dd0d2b --- /dev/null +++ b/backend/tests/test_email_dedupe_complete_metadata.py @@ -0,0 +1,53 @@ +import datetime +from types import SimpleNamespace + +import pytest + +from services.email_dedupe_service import ( + EmailDedupeCandidate, + candidate_strong_fingerprint, + email_strong_fingerprint, +) + + +_COMPLETE = { + "sender": "sender@example.com", + "recipients": "recipient@example.com", + "subject": "Quarterly plan", + "date": datetime.datetime(2026, 9, 17, 6, 0, tzinfo=datetime.timezone.utc), + "body": "Please review the attached plan.", +} + + +def test_candidate_strong_fingerprint_requires_complete_metadata() -> None: + candidate = EmailDedupeCandidate( + candidate_key="candidate-1", + date_provenance="parsed", + **_COMPLETE, + ) + + assert candidate_strong_fingerprint(candidate) is not None + + +@pytest.mark.parametrize("missing_field", ["sender", "recipients", "subject", "body"]) +def test_candidate_strong_fingerprint_withholds_incomplete_metadata( + missing_field: str, +) -> None: + values = dict(_COMPLETE) + values[missing_field] = "" + candidate = EmailDedupeCandidate( + candidate_key=f"missing-{missing_field}", + date_provenance="parsed", + **values, + ) + + assert candidate_strong_fingerprint(candidate) is None + + +def test_stored_email_strong_fingerprint_withholds_incomplete_metadata() -> None: + stored = SimpleNamespace( + date_provenance="parsed", + **{**_COMPLETE, "recipients": None}, + ) + + assert email_strong_fingerprint(stored) is None diff --git a/backend/tests/test_email_dedupe_service.py b/backend/tests/test_email_dedupe_service.py index 1afbad5ed..924301287 100644 --- a/backend/tests/test_email_dedupe_service.py +++ b/backend/tests/test_email_dedupe_service.py @@ -7,9 +7,30 @@ strong_email_fingerprint, candidate_strong_fingerprint, email_strong_fingerprint, + content_email_fingerprint, + candidate_content_fingerprint, + email_content_fingerprint, + classify_dedupe_decision, + resolve_candidate_disposition, ) from db.models import Email + +def _email_row(**overrides): + fields = dict( + id=100, + user_id="user-1", + organization_id="org-1", + message_id=None, + sender="sender@example.com", + subject="Subject", + date=datetime(2023, 1, 1, 12, 0, 0, tzinfo=timezone.utc), + date_provenance="parsed", + body="Hello world", + ) + fields.update(overrides) + return Email(**fields) + def test_candidate_message_lookup_values_basic(): candidate = EmailDedupeCandidate( candidate_key="key", @@ -97,6 +118,7 @@ def test_email_strong_fingerprint(): sender="sender@example.com", subject="Subject", date=datetime(2023, 1, 1, 12, 0, 0, tzinfo=timezone.utc), + date_provenance="parsed", body="Hello world" ) result1 = email_strong_fingerprint(email) @@ -109,3 +131,236 @@ def test_email_strong_fingerprint(): ) assert result1 == result2 assert result1 is not None + + +def test_email_strong_fingerprint_gated_to_parsed_date_provenance(): + """A stored row seeds a strong fingerprint only when its date is genuine. + + naruon#1086: rows whose ``date`` is a synthetic collection-time fallback + (missing/invalid) or unknown-provenance (stored before tracking) must not + seed a strong auto-dedupe fingerprint, even though sender/subject/body/date + are populated. + """ + fields = dict( + id=2, + user_id="user-1", + organization_id="org-1", + message_id="msg-2", + sender="sender@example.com", + subject="Subject", + date=datetime(2023, 1, 1, 12, 0, 0, tzinfo=timezone.utc), + body="Hello world", + ) + assert email_strong_fingerprint(Email(**fields, date_provenance="parsed")) is not None + for provenance in ("missing", "invalid", "unknown"): + assert email_strong_fingerprint(Email(**fields, date_provenance=provenance)) is None + + +# --- content fingerprint (date-independent identity signal, naruon#1086) --- + +def test_content_email_fingerprint_none_without_body(): + assert content_email_fingerprint(sender="a@x", subject="S", body=None) is None + assert content_email_fingerprint(sender="a@x", subject="S", body="") is None + + +def test_content_email_fingerprint_is_date_independent_and_not_the_strong_one(): + """The content fingerprint ignores the Date; the strong one includes it.""" + content = content_email_fingerprint( + sender="sender@example.com", subject="Subject", body="Hello world" + ) + strong = strong_email_fingerprint( + sender="sender@example.com", + subject="Subject", + date=datetime(2023, 1, 1, 12, 0, 0, tzinfo=timezone.utc), + body="Hello world", + ) + assert content is not None + assert content != strong + # Same content, different Date -> identical content fingerprint. + candidate_a = EmailDedupeCandidate( + candidate_key="a", + sender="sender@example.com", + subject="Subject", + date=datetime(2023, 1, 1, tzinfo=timezone.utc), + body="Hello world", + ) + candidate_b = EmailDedupeCandidate( + candidate_key="b", + sender="sender@example.com", + subject="Subject", + date=datetime(2024, 6, 6, tzinfo=timezone.utc), + body="Hello world", + ) + assert candidate_content_fingerprint(candidate_a) == candidate_content_fingerprint( + candidate_b + ) + assert email_content_fingerprint(_email_row()) == content + + +def test_email_content_fingerprint_none_without_body(): + assert email_content_fingerprint(_email_row(body=None)) is None + + +# --- Fellegi-Sunter (1969) three-zone classifier --- + +def test_auto_link_on_matching_normalized_message_id(): + # Bracketed vs bare Message-ID normalize equal; content is irrelevant here. + candidate = EmailDedupeCandidate( + candidate_key="c", + message_id="", + sender="other@example.com", + subject="Totally different", + body="unrelated body", + ) + existing = _email_row(message_id="shared@x") + assert classify_dedupe_decision(candidate, existing) == "auto_link" + + +def test_auto_link_on_genuine_strong_match_without_message_id(): + date = datetime(2023, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + candidate = EmailDedupeCandidate( + candidate_key="c", + sender="sender@example.com", + subject="Subject", + date=date, + body="Hello world", + date_provenance="parsed", + ) + existing = _email_row(date=date, date_provenance="parsed") + assert classify_dedupe_decision(candidate, existing) == "auto_link" + + +def test_review_required_when_candidate_date_provenance_is_untrusted(): + date = datetime(2023, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + candidate = EmailDedupeCandidate( + candidate_key="c", + sender="sender@example.com", + subject="Subject", + date=date, + body="Hello world", + date_provenance="missing", + ) + existing = _email_row(date=date, date_provenance="parsed") + # Same content, but the candidate's Date is synthetic -> no strong match, no + # Message-ID link -> clerical-review band, not a silent merge. + assert classify_dedupe_decision(candidate, existing) == "review_required" + + +def test_review_required_when_existing_date_provenance_is_untrusted(): + date = datetime(2023, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + candidate = EmailDedupeCandidate( + candidate_key="c", + sender="sender@example.com", + subject="Subject", + date=date, + body="Hello world", + date_provenance="parsed", + ) + existing = _email_row(date=date, date_provenance="unknown") + assert classify_dedupe_decision(candidate, existing) == "review_required" + + +def test_review_required_when_content_matches_but_dates_differ_untrusted(): + candidate = EmailDedupeCandidate( + candidate_key="c", + sender="sender@example.com", + subject="Subject", + date=datetime(2024, 6, 6, tzinfo=timezone.utc), + body="Hello world", + date_provenance="invalid", + ) + existing = _email_row( + date=datetime(2023, 1, 1, tzinfo=timezone.utc), date_provenance="unknown" + ) + assert classify_dedupe_decision(candidate, existing) == "review_required" + + +def test_distinct_on_different_content_and_no_identity_link(): + candidate = EmailDedupeCandidate( + candidate_key="c", + message_id="only-on-candidate@x", + sender="different@example.com", + subject="Different", + body="different body", + date_provenance="parsed", + ) + existing = _email_row(message_id="only-on-existing@x") + assert classify_dedupe_decision(candidate, existing) == "distinct" + + +def test_distinct_when_candidate_has_no_body(): + candidate = EmailDedupeCandidate( + candidate_key="c", + sender="sender@example.com", + subject="Subject", + body=None, + date_provenance="parsed", + ) + existing = _email_row() + assert classify_dedupe_decision(candidate, existing) == "distinct" + + +def test_resolve_disposition_empty_corpus_is_distinct(): + candidate = EmailDedupeCandidate(candidate_key="c", message_id="m@x", body="b") + assert resolve_candidate_disposition(candidate, []) == ("distinct", None) + + +def test_resolve_disposition_all_distinct_returns_distinct_none(): + candidate = EmailDedupeCandidate( + candidate_key="c", + message_id="only-on-candidate@x", + sender="different@example.com", + subject="Different", + body="different body", + date_provenance="parsed", + ) + rows = [_email_row(id=1, message_id="a@x"), _email_row(id=2, message_id="b@x")] + assert resolve_candidate_disposition(candidate, rows) == ("distinct", None) + + +def test_resolve_disposition_returns_review_row_when_only_content_matches(): + date = datetime(2023, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + candidate = EmailDedupeCandidate( + candidate_key="c", + sender="sender@example.com", + subject="Subject", + date=date, + body="Hello world", + date_provenance="missing", # synthetic Date -> no strong/A1 link + ) + unrelated = _email_row(id=1, message_id="unrelated@x", body="different body") + content_match = _email_row(id=2, date=date, date_provenance="parsed") + decision, matched = resolve_candidate_disposition(candidate, [unrelated, content_match]) + assert decision == "review_required" + assert matched is content_match + + +def test_resolve_disposition_a1_link_dominates_earlier_a2_review(): + # A review_required content match appears BEFORE the auto_link row in the + # corpus; Fellegi-Sunter A1 must still win regardless of iteration order. + date = datetime(2023, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + candidate = EmailDedupeCandidate( + candidate_key="c", + message_id="", + sender="sender@example.com", + subject="Subject", + date=date, + body="Hello world", + date_provenance="missing", + ) + content_only = _email_row(id=1, date=date, date_provenance="unknown") + id_link = _email_row(id=2, message_id="shared@x", body="totally different") + decision, matched = resolve_candidate_disposition(candidate, [content_only, id_link]) + assert decision == "auto_link" + assert matched is id_link + + +def test_resolve_disposition_first_auto_link_wins(): + candidate = EmailDedupeCandidate( + candidate_key="c", message_id="", body="b", date_provenance="parsed" + ) + first = _email_row(id=1, message_id="shared@x") + second = _email_row(id=2, message_id="shared@x") + decision, matched = resolve_candidate_disposition(candidate, [first, second]) + assert decision == "auto_link" + assert matched is first diff --git a/backend/tests/test_email_import_service.py b/backend/tests/test_email_import_service.py index 51d2a2633..c7deec02a 100644 --- a/backend/tests/test_email_import_service.py +++ b/backend/tests/test_email_import_service.py @@ -792,3 +792,42 @@ async def test_generate_import_embeddings_recovers_valid_items_after_batch_failu assert embeddings[2] == [0.75] * (EMBEDDING_DIMENSION // 2) + [0.0] * ( EMBEDDING_DIMENSION // 2 ) + + +def test_email_fingerprint_uses_strong_key_only_for_parsed_date(): + """A strong (auto-dedupe) fingerprint is seeded only from a genuine Date. + + naruon#1086: when the RFC822 Date was missing or invalid the persisted date + is a synthetic collection-time fallback, which must not seed the strong + duplicate key. Only ``date_provenance == "parsed"`` yields the strong + fingerprint; missing/invalid provenance falls through to the weak fallback. + """ + from services.email_dedupe_service import strong_email_fingerprint + from services.email_import_service import _email_fingerprint + + persisted_date = datetime.datetime( + 2026, 4, 27, 10, 0, 0, tzinfo=datetime.timezone.utc + ) + parsed_fields = { + "sender": "sender@test.com", + "subject": "Quarterly report", + "body": "The full report body.", + "recipients": "recipient@test.com", + } + strong = strong_email_fingerprint( + sender=parsed_fields["sender"], + subject=parsed_fields["subject"], + date=persisted_date, + body=parsed_fields["body"], + ) + assert strong is not None + + assert ( + _email_fingerprint({**parsed_fields, "date_provenance": "parsed"}, persisted_date) + == strong + ) + for provenance in ("missing", "invalid"): + weak = _email_fingerprint( + {**parsed_fields, "date_provenance": provenance}, persisted_date + ) + assert weak != strong diff --git a/backend/tests/test_email_import_strong_evidence_boundary.py b/backend/tests/test_email_import_strong_evidence_boundary.py new file mode 100644 index 000000000..743a4f18f --- /dev/null +++ b/backend/tests/test_email_import_strong_evidence_boundary.py @@ -0,0 +1,115 @@ +import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest + +import services.email_import_service as email_import_module +from services.email_dedupe_service import source_email_fingerprint, strong_email_fingerprint +from services.email_import_service import _dedupe_review_reason, _email_fingerprint + + +_DATE = datetime.datetime(2026, 9, 17, 6, 30, tzinfo=datetime.timezone.utc) +_SOURCE = b"From: sender@example.com\r\nTo: recipient@example.com\r\nSubject: Plan\r\n\r\nBody" +_COMPLETE = { + "sender": "sender@example.com", + "recipients": "recipient@example.com", + "subject": "Plan", + "body": "Body", + "date": _DATE, + "date_provenance": "parsed", +} + + +def test_import_metadata_strong_fingerprint_requires_recipient_evidence() -> None: + expected_strong = strong_email_fingerprint( + sender=_COMPLETE["sender"], + subject=_COMPLETE["subject"], + date=_DATE, + body=_COMPLETE["body"], + ) + assert expected_strong is not None + assert _email_fingerprint(dict(_COMPLETE), _DATE, _SOURCE) == expected_strong + + incomplete = {**_COMPLETE, "recipients": ""} + assert _email_fingerprint(incomplete, _DATE, _SOURCE) == source_email_fingerprint( + _SOURCE, + source_kind="raw", + ) + + +def test_import_review_reason_tracks_withheld_strong_metadata_evidence() -> None: + assert _dedupe_review_reason(dict(_COMPLETE)) is None + assert _dedupe_review_reason({**_COMPLETE, "recipients": ""}) == ( + "dedupe_review_required" + ) + assert _dedupe_review_reason({**_COMPLETE, "date_provenance": "invalid"}) == ( + "dedupe_review_required" + ) + + +@pytest.mark.asyncio +async def test_import_result_marks_incomplete_metadata_for_dedupe_review( + monkeypatch, + tmp_path, +) -> None: + eml_path = tmp_path / "message.eml" + eml_path.write_bytes(_SOURCE) + parsed = { + **_COMPLETE, + "message_id": "", + "recipients": "", + "attachments": [], + } + email_object = MagicMock() + session = MagicMock() + session.add = MagicMock() + session.commit = AsyncMock() + session.rollback = AsyncMock() + + monkeypatch.setattr( + email_import_module.settings, + "PROJECT_GRAPH_EXTRACTION_ENABLED", + False, + ) + monkeypatch.setattr( + email_import_module, + "_read_and_parse_eml", + lambda _: (_SOURCE, parsed), + ) + monkeypatch.setattr( + email_import_module, + "_find_existing_email", + AsyncMock(return_value=None), + ) + monkeypatch.setattr( + email_import_module, + "assign_thread_id", + AsyncMock(return_value="thread-1"), + ) + monkeypatch.setattr( + email_import_module, + "_extract_and_generate_embeddings", + AsyncMock(return_value=([], [[0.0] * email_import_module.EMBEDDING_DIMENSION])), + ) + monkeypatch.setattr( + email_import_module, + "_build_email_object", + lambda **_: (email_object, 0), + ) + monkeypatch.setattr( + email_import_module, + "_persist_project_graph_projection", + AsyncMock(), + ) + + result = await email_import_module._import_single_eml( + session, + eml_path=eml_path, + display_filename="message.eml", + user_id="user-1", + organization_id="org-1", + ) + + assert result.status == "imported" + assert result.reason_code == "dedupe_review_required" + session.commit.assert_awaited_once() diff --git a/backend/tests/test_email_parser_provenance.py b/backend/tests/test_email_parser_provenance.py new file mode 100644 index 000000000..1a3a41a1e --- /dev/null +++ b/backend/tests/test_email_parser_provenance.py @@ -0,0 +1,144 @@ +"""Focused contracts for email metadata provenance classification.""" + +import datetime + +from services.email_parser import parse_eml_bytes + + +def _eml_with(headers: str) -> bytes: + """Build minimal EML bytes with the given header block and a plain body.""" + return (headers.strip("\r\n") + "\n\nBody text.").encode("utf-8") + + +def test_parse_eml_marks_valid_date_as_parsed_with_original_header_date() -> None: + """A valid Date header remains the genuine timezone-aware storage value.""" + parsed = parse_eml_bytes( + _eml_with( + """Message-ID: +From: sender@test.com +To: recipient@test.com +Subject: Valid date +Date: Mon, 27 Apr 2026 10:00:00 +0000""" + ) + ) + + expected = datetime.datetime(2026, 4, 27, 10, 0, 0, tzinfo=datetime.timezone.utc) + assert parsed["date_provenance"] == "parsed" + assert parsed["header_date"] == expected + assert parsed["date"] == expected + + +def test_parse_eml_marks_missing_date_without_promoting_the_fallback() -> None: + """A missing Date uses storage fallback without inventing sender evidence.""" + parsed = parse_eml_bytes( + _eml_with( + """Message-ID: +From: sender@test.com +To: recipient@test.com +Subject: No date header""" + ) + ) + + assert parsed["date_provenance"] == "missing" + assert parsed["header_date"] is None + assert parsed["date"].tzinfo is not None + + +def test_parse_eml_marks_invalid_date_without_promoting_the_fallback() -> None: + """An invalid Date uses storage fallback without inventing sender evidence.""" + parsed = parse_eml_bytes( + _eml_with( + """Message-ID: +From: sender@test.com +To: recipient@test.com +Subject: Unparseable date +Date: not-a-real-date""" + ) + ) + + assert parsed["date_provenance"] == "invalid" + assert parsed["header_date"] is None + assert parsed["date"].tzinfo is not None + + +def test_parse_eml_marks_whitespace_only_date_as_missing() -> None: + """A whitespace-only Date is missing rather than malformed evidence.""" + parsed = parse_eml_bytes( + _eml_with( + """Message-ID: +From: sender@test.com +To: recipient@test.com +Subject: Whitespace-only date +Date: """ + ) + ) + + assert parsed["date_provenance"] == "missing" + assert parsed["header_date"] is None + assert parsed["date"].tzinfo is not None + + +def test_parse_eml_normalizes_minus_zero_zone_to_utc() -> None: + """RFC 5322 -0000 dates remain timezone-aware for storage and comparison.""" + parsed = parse_eml_bytes( + _eml_with( + """Message-ID: +From: sender@test.com +To: recipient@test.com +Subject: Minus-zero timezone +Date: Sun, 01 Jan 2023 12:00:00 -0000""" + ) + ) + + assert parsed["date_provenance"] == "parsed" + assert parsed["header_date"] is not None + assert parsed["header_date"].tzinfo is not None + assert parsed["header_date"].utcoffset() == datetime.timedelta(0) + assert parsed["date"].tzinfo is not None + + +def test_parse_eml_marks_zone_less_date_as_invalid_source_evidence() -> None: + """A parseable but zone-less Date cannot become strong sender evidence.""" + parsed = parse_eml_bytes( + _eml_with( + """Message-ID: +From: sender@test.com +To: recipient@test.com +Subject: Missing timezone +Date: Sun, 01 Jan 2023 12:00:00""" + ) + ) + + assert parsed["date_provenance"] == "invalid" + assert parsed["header_date"] is None + assert parsed["date"].tzinfo is not None + + +def test_parse_eml_marks_embedded_message_id_provenance() -> None: + """A non-empty embedded Message-ID is identified as sender evidence.""" + parsed = parse_eml_bytes( + _eml_with( + """Message-ID: +From: sender@test.com +To: recipient@test.com +Subject: Has message id +Date: Mon, 27 Apr 2026 10:00:00 +0000""" + ) + ) + + assert parsed["message_id_provenance"] == "embedded" + + +def test_parse_eml_marks_missing_message_id_provenance() -> None: + """A missing Message-ID is explicitly classified as absent evidence.""" + parsed = parse_eml_bytes( + _eml_with( + """From: sender@test.com +To: recipient@test.com +Subject: No message id +Date: Mon, 27 Apr 2026 10:00:00 +0000""" + ) + ) + + assert parsed["message_id"] == "" + assert parsed["message_id_provenance"] == "missing" diff --git a/backend/tests/test_imap_strong_evidence_boundary.py b/backend/tests/test_imap_strong_evidence_boundary.py new file mode 100644 index 000000000..b523fbe08 --- /dev/null +++ b/backend/tests/test_imap_strong_evidence_boundary.py @@ -0,0 +1,47 @@ +import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from services.email_dedupe_service import source_email_fingerprint +from services.imap_worker import process_fetched_email + + +@pytest.mark.asyncio +async def test_imap_withholds_metadata_strong_fingerprint_without_recipients( + monkeypatch, +) -> None: + raw_message = b"From: sender@example.com\r\nSubject: Plan\r\n\r\nBody" + email_data = { + "message_id": "", + "sender": "sender@example.com", + "recipients": "", + "subject": "Plan", + "date": datetime.datetime(2026, 9, 17, 6, 45, tzinfo=datetime.timezone.utc), + "date_provenance": "parsed", + "body": "Body", + } + + query_result = MagicMock() + query_result.scalar_one_or_none.return_value = None + session = MagicMock() + session.execute = AsyncMock(return_value=query_result) + session.flush = AsyncMock() + + monkeypatch.setattr( + "services.imap_worker.assign_thread_id", + AsyncMock(return_value="thread-boundary"), + ) + + imported = await process_fetched_email( + session, + email_data, + "user-1", + "org-1", + source_content=raw_message, + ) + + assert imported.fingerprint == source_email_fingerprint( + raw_message, + source_kind="raw", + ) diff --git a/backend/tests/test_imap_worker.py b/backend/tests/test_imap_worker.py index d2798719a..4ac5acd37 100644 --- a/backend/tests/test_imap_worker.py +++ b/backend/tests/test_imap_worker.py @@ -1,3 +1,4 @@ +from types import SimpleNamespace from unittest.mock import AsyncMock import pytest @@ -90,7 +91,9 @@ async def test_imap_worker_imports_fetched_rfc822_messages(monkeypatch): session.__aenter__.return_value = session session.__aexit__.return_value = False - process_fetched_email_mock = AsyncMock() + persist_fetched_email_mock = AsyncMock( + return_value=SimpleNamespace(created_record=True) + ) monkeypatch.setattr( "services.imap_worker.validate_imap_destination", @@ -102,8 +105,8 @@ async def test_imap_worker_imports_fetched_rfc822_messages(monkeypatch): ) monkeypatch.setattr("services.imap_worker.AsyncSessionLocal", lambda: session) monkeypatch.setattr( - "services.imap_worker.process_fetched_email", - process_fetched_email_mock, + "services.imap_worker.persist_fetched_email", + persist_fetched_email_mock, ) imported_count = await worker._sync_tenant(config) @@ -116,8 +119,8 @@ async def test_imap_worker_imports_fetched_rfc822_messages(monkeypatch): imap_client.fetch.assert_awaited_once_with("1", "(RFC822 FLAGS)") imap_client.logout.assert_awaited_once() - process_fetched_email_mock.assert_awaited_once() - args, kwargs = process_fetched_email_mock.await_args + persist_fetched_email_mock.assert_awaited_once() + args, kwargs = persist_fetched_email_mock.await_args assert args[0] is session assert args[1]["message_id"] == "" assert args[1]["subject"] == "IMAP import" @@ -126,11 +129,70 @@ async def test_imap_worker_imports_fetched_rfc822_messages(monkeypatch): assert kwargs["is_read"] is False assert args[3] == "org-imap" assert kwargs["owner_addresses"] == ["imap-user@example.com"] + assert kwargs["source_content"] == raw_message session.commit.assert_awaited_once() session.rollback.assert_not_awaited() +@pytest.mark.asyncio +async def test_imap_duplicate_does_not_inflate_imported_count(monkeypatch): + from db.models import Email + + worker = ImapSyncWorker() + config = TenantConfig( + user_id="imap-user", + organization_id="org-imap", + imap_server="imap.example.com", + imap_port=993, + imap_username="imap-user@example.com", + imap_password="imap-secret", + ) + raw_message = ( + b"Message-ID: \r\n" + b"From: Sender \r\n" + b"To: imap-user@example.com\r\n" + b"Subject: Existing message\r\n" + b"Date: Mon, 15 Jun 2026 10:00:00 +0000\r\n" + b"\r\n" + b"Already imported.\r\n" + ) + existing_email = Email(id=1) + + class ExistingResult: + def scalar_one_or_none(self): + return existing_email + + class FakeSession: + def __init__(self): + self.committed = False + self.rolled_back = False + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def execute(self, _statement): + return ExistingResult() + + async def commit(self): + self.committed = True + + async def rollback(self): + self.rolled_back = True + + session = FakeSession() + monkeypatch.setattr("services.imap_worker.AsyncSessionLocal", lambda: session) + + imported_count = await worker._import_messages(config, [(raw_message, False)]) + + assert imported_count == 0 + assert session.committed is True + assert session.rolled_back is False + + @pytest.mark.asyncio async def test_imap_worker_requires_credentials_without_sensitive_log_names( caplog, monkeypatch @@ -173,7 +235,7 @@ def test_flags_indicate_seen_parses_seen_flag(): no_flags = ("OK", [(b"1 (RFC822 {%d}" % len(raw), raw)]) assert flags_indicate_seen(seen[1]) is True - assert flags_indicate_seen(unseen[1]) is False # other flags, but not \Seen + assert flags_indicate_seen(unseen[1]) is False # other flags, but not \Seen assert flags_indicate_seen(no_flags[1]) is False # no FLAGS section -> unread assert flags_indicate_seen([]) is False assert flags_indicate_seen(None) is False diff --git a/backend/tests/test_pop3_partial_retrieval.py b/backend/tests/test_pop3_partial_retrieval.py new file mode 100644 index 000000000..f55bb4617 --- /dev/null +++ b/backend/tests/test_pop3_partial_retrieval.py @@ -0,0 +1,117 @@ +import poplib +from unittest.mock import MagicMock + +from db.models import TenantConfig +from services.pop3_worker import Pop3SyncWorker + + +def _config() -> TenantConfig: + return TenantConfig( + user_id="pop3-user", + pop3_server="pop3.example.com", + pop3_port=995, + pop3_username="pop3-user@example.com", + pop3_password="pop3-secret", + ) + + +def test_pop3_transport_failure_preserves_already_retrieved_uidl_messages(monkeypatch): + worker = Pop3SyncWorker() + config = _config() + client = MagicMock() + client.uidl.return_value = (b"+OK", [b"1 uid-1", b"2 uid-2"], 32) + client.retr.side_effect = [ + (b"+OK", [b"Message-ID: ", b"", b"Body one"], 128), + OSError("transport failed"), + ] + monkeypatch.setattr("services.pop3_worker.poplib.POP3_SSL", lambda host, port: client) + + messages = worker._do_pop3_sync( + config, + "pop3.example.com", + 995, + observed_uidls=set(), + ) + + assert len(messages) == 1 + assert messages[0].provider_uidl == "uid-1" + assert b"" in messages[0].source_content + assert [entry.args[0] for entry in client.retr.call_args_list] == [1, 2] + + +def test_pop3_negative_retr_skips_failed_uidl_and_continues_same_session(monkeypatch): + worker = Pop3SyncWorker() + config = _config() + client = MagicMock() + client.uidl.return_value = ( + b"+OK", + [b"1 uid-1", b"2 uid-2", b"3 uid-3"], + 48, + ) + client.retr.side_effect = [ + (b"+OK", [b"Message-ID: ", b"", b"Body one"], 128), + poplib.error_proto("-ERR no such message"), + (b"+OK", [b"Message-ID: ", b"", b"Body three"], 128), + ] + monkeypatch.setattr("services.pop3_worker.poplib.POP3_SSL", lambda host, port: client) + + messages = worker._do_pop3_sync( + config, + "pop3.example.com", + 995, + observed_uidls=set(), + ) + + assert [message.provider_uidl for message in messages] == ["uid-1", "uid-3"] + assert [entry.args[0] for entry in client.retr.call_args_list] == [1, 2, 3] + + +def test_pop3_fallback_negative_retr_continues_bounded_window(monkeypatch): + worker = Pop3SyncWorker() + config = _config() + client = MagicMock() + client.uidl.side_effect = poplib.error_proto("-ERR UIDL unsupported") + client.list.return_value = (b"+OK", [b"1 100", b"2 100", b"3 100"], 300) + client.retr.side_effect = [ + (b"+OK", [b"Message-ID: ", b"", b"Body one"], 128), + poplib.error_proto("-ERR no such message"), + (b"+OK", [b"Message-ID: ", b"", b"Body three"], 128), + ] + monkeypatch.setattr("services.pop3_worker.poplib.POP3_SSL", lambda host, port: client) + + messages = worker._do_pop3_sync( + config, + "pop3.example.com", + 995, + observed_uidls=set(), + ) + + assert [message.provider_uidl for message in messages] == [None, None] + assert [entry.args[0] for entry in client.retr.call_args_list] == [1, 2, 3] + + +def test_pop3_malformed_protocol_response_stops_remaining_retrievals(monkeypatch): + worker = Pop3SyncWorker() + config = _config() + client = MagicMock() + client.uidl.return_value = ( + b"+OK", + [b"1 uid-1", b"2 uid-2", b"3 uid-3"], + 48, + ) + client.retr.side_effect = [ + (b"+OK", [b"Message-ID: ", b"", b"Body one"], 128), + poplib.error_proto("unexpected response"), + (b"+OK", [b"Message-ID: ", b"", b"Body three"], 128), + ] + monkeypatch.setattr("services.pop3_worker.poplib.POP3_SSL", lambda host, port: client) + + messages = worker._do_pop3_sync( + config, + "pop3.example.com", + 995, + observed_uidls=set(), + ) + + assert [message.provider_uidl for message in messages] == ["uid-1"] + assert [entry.args[0] for entry in client.retr.call_args_list] == [1, 2] diff --git a/backend/tests/test_pop3_quit_resilience.py b/backend/tests/test_pop3_quit_resilience.py new file mode 100644 index 000000000..7652696a7 --- /dev/null +++ b/backend/tests/test_pop3_quit_resilience.py @@ -0,0 +1,37 @@ +import poplib +from unittest.mock import MagicMock + +from db.models import TenantConfig +from services.pop3_worker import Pop3SyncWorker + + +def test_pop3_quit_failure_does_not_discard_retrieved_messages(monkeypatch): + worker = Pop3SyncWorker() + config = TenantConfig( + user_id="pop3-user", + pop3_server="pop3.example.com", + pop3_port=995, + pop3_username="pop3-user@example.com", + pop3_password="pop3-secret", + ) + raw_lines = [b"Message-ID: ", b"", b"Body"] + pop3_client = MagicMock() + pop3_client.uidl.return_value = (b"+OK", [b"1 uid-1"], 16) + pop3_client.retr.return_value = (b"+OK", raw_lines, 128) + pop3_client.quit.side_effect = poplib.error_proto("-ERR connection already closed") + + monkeypatch.setattr( + "services.pop3_worker.validate_pop3_destination", + lambda host, port: (host, port), + ) + monkeypatch.setattr( + "services.pop3_worker.poplib.POP3_SSL", + lambda host, port: pop3_client, + ) + + messages = worker._do_pop3_sync(config) + + assert messages[0].source_content == b"Message-ID: \r\n\r\nBody\r\n" + assert messages[0].provider_uidl == "uid-1" + pop3_client.quit.assert_called_once() + pop3_client.close.assert_called_once() diff --git a/backend/tests/test_pop3_retry_progress.py b/backend/tests/test_pop3_retry_progress.py new file mode 100644 index 000000000..1732a6d81 --- /dev/null +++ b/backend/tests/test_pop3_retry_progress.py @@ -0,0 +1,117 @@ +import datetime +import poplib +from unittest.mock import MagicMock + +from db.models import TenantConfig +from services.pop3_worker import ( + Pop3CollectionProgressState, + Pop3MessageIdentity, + Pop3SyncWorker, +) + + +def _identities(count: int = 12) -> list[Pop3MessageIdentity]: + return [ + Pop3MessageIdentity(message_number=number, provider_uidl=f"uid-{number}") + for number in range(1, count + 1) + ] + + +def _config() -> TenantConfig: + return TenantConfig( + id=42, + user_id="pop3-user", + organization_id="org-pop3", + pop3_server="pop3.example.com", + pop3_port=995, + pop3_username="pop3-user@example.com", + pop3_password="pop3-secret", + ) + + +def test_uidl_selection_prioritizes_never_attempted_backlog_before_due_retries(): + worker = Pop3SyncWorker() + now = datetime.datetime(2026, 9, 17, 10, 40, tzinfo=datetime.timezone.utc) + progress = { + f"uid-{number}": Pop3CollectionProgressState( + disposition="retryable", + retry_after=now - datetime.timedelta(minutes=1), + ) + for number in range(3, 13) + } + + selected = worker._select_uidl_candidates(_identities(), progress, now) + + assert [identity.provider_uidl for identity in selected[:2]] == ["uid-2", "uid-1"] + assert len(selected) == 10 + + +def test_uidl_selection_defers_retryable_identity_until_retry_after(): + worker = Pop3SyncWorker() + now = datetime.datetime(2026, 9, 17, 10, 40, tzinfo=datetime.timezone.utc) + progress = { + "uid-12": Pop3CollectionProgressState( + disposition="retryable", + retry_after=now + datetime.timedelta(minutes=5), + ) + } + + selected = worker._select_uidl_candidates(_identities(), progress, now) + + assert "uid-12" not in {identity.provider_uidl for identity in selected} + assert [identity.provider_uidl for identity in selected[:2]] == ["uid-11", "uid-10"] + + +def test_uidl_selection_never_retries_observed_identity(): + worker = Pop3SyncWorker() + now = datetime.datetime(2026, 9, 17, 10, 40, tzinfo=datetime.timezone.utc) + progress = { + "uid-12": Pop3CollectionProgressState( + disposition="observed", + retry_after=None, + ) + } + + selected = worker._select_uidl_candidates(_identities(), progress, now) + + assert "uid-12" not in {identity.provider_uidl for identity in selected} + + +def test_persistent_negative_tail_becomes_retryable_without_hiding_lower_backlog(monkeypatch): + worker = Pop3SyncWorker() + client = MagicMock() + client.uidl.return_value = ( + b"+OK", + [f"{number} uid-{number}".encode() for number in range(1, 13)], + 192, + ) + client.retr.side_effect = poplib.error_proto("-ERR no such message") + monkeypatch.setattr("services.pop3_worker.poplib.POP3_SSL", lambda host, port: client) + + first_batch = worker._do_pop3_sync_batch( + _config(), + "pop3.example.com", + 995, + collection_progress={}, + ) + + assert first_batch.messages == [] + assert first_batch.retryable_uidls == frozenset( + f"uid-{number}" for number in range(3, 13) + ) + + now = datetime.datetime(2026, 9, 17, 10, 40, tzinfo=datetime.timezone.utc) + persisted_retry_state = { + provider_uidl: Pop3CollectionProgressState( + disposition="retryable", + retry_after=now + datetime.timedelta(minutes=1), + ) + for provider_uidl in first_batch.retryable_uidls + } + next_selected = worker._select_uidl_candidates( + _identities(), + persisted_retry_state, + now, + ) + + assert [identity.provider_uidl for identity in next_selected] == ["uid-2", "uid-1"] diff --git a/backend/tests/test_pop3_uidl_migration_contract.py b/backend/tests/test_pop3_uidl_migration_contract.py new file mode 100644 index 000000000..d249e19d7 --- /dev/null +++ b/backend/tests/test_pop3_uidl_migration_contract.py @@ -0,0 +1,46 @@ +from pathlib import Path + +from db.pop3_collection_models import Pop3ObservedMessage + + +MIGRATION_PATH = ( + Path(__file__).resolve().parents[1] + / "alembic" + / "versions" + / "0019_pop3_observed_uidl.py" +) + + +def test_pop3_uidl_model_is_owner_scoped_and_tracks_retry_disposition(): + table = Pop3ObservedMessage.__table__ + + assert table.name == "pop3_observed_messages" + assert table.c.tenant_config_id.nullable is False + assert table.c.provider_uidl.nullable is False + assert table.c.provider_uidl.type.length == 70 + assert table.c.collection_disposition.nullable is False + assert table.c.collection_disposition.type.length == 16 + assert table.c.retry_after.nullable is True + assert table.c.observed_at.nullable is True + assert any( + constraint.name == "uq_pop3_observed_messages_account_uidl" + for constraint in table.constraints + ) + assert any( + index.name == "ix_pop3_observed_messages_account_retry" + for index in table.indexes + ) + + +def test_pop3_uidl_migration_extends_branch_local_collection_progress_revision(): + source = MIGRATION_PATH.read_text(encoding="utf-8") + + assert 'revision = "0019_pop3_observed_uidl"' in source + assert 'down_revision = "0018_email_date_provenance"' in source + assert 'sa.ForeignKey("tenant_configs.id", ondelete="CASCADE")' in source + assert 'sa.String(length=70)' in source + assert '"collection_disposition"' in source + assert '"retry_after"' in source + assert '"uq_pop3_observed_messages_account_uidl"' in source + assert '"ix_pop3_observed_messages_account_retry"' in source + assert "branch-local" in source diff --git a/backend/tests/test_pop3_uidl_progress.py b/backend/tests/test_pop3_uidl_progress.py new file mode 100644 index 000000000..f88913b20 --- /dev/null +++ b/backend/tests/test_pop3_uidl_progress.py @@ -0,0 +1,149 @@ +import poplib +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from db.models import TenantConfig +from services.pop3_worker import Pop3RetrievedMessage, Pop3SyncWorker + + +def _config() -> TenantConfig: + return TenantConfig( + id=42, + user_id="pop3-user", + organization_id="org-pop3", + pop3_server="pop3.example.com", + pop3_port=995, + pop3_username="pop3-user@example.com", + pop3_password="pop3-secret", + ) + + +def _client_with_uidls(count: int = 12) -> MagicMock: + client = MagicMock() + client.uidl.return_value = ( + b"+OK", + [f"{message_number} uid-{message_number}".encode() for message_number in range(1, count + 1)], + count * 16, + ) + client.list.return_value = ( + b"+OK", + [f"{message_number} 128".encode() for message_number in range(1, count + 1)], + count * 128, + ) + client.retr.side_effect = lambda message_number: ( + b"+OK", + [f"Message-ID: ".encode(), b"", b"Body"], + 128, + ) + return client + + +def test_pop3_uidl_progress_selects_older_unseen_after_newer_window(monkeypatch): + worker = Pop3SyncWorker() + client = _client_with_uidls() + monkeypatch.setattr("services.pop3_worker.poplib.POP3_SSL", lambda host, port: client) + + observed_uidls = {f"uid-{message_number}" for message_number in range(3, 13)} + messages = worker._do_pop3_sync( + _config(), + "pop3.example.com", + 995, + observed_uidls=observed_uidls, + ) + + assert [entry.args[0] for entry in client.retr.call_args_list] == [1, 2] + assert [message.provider_uidl for message in messages] == ["uid-1", "uid-2"] + + +def test_pop3_uidl_progress_uses_current_session_number_after_renumbering(monkeypatch): + worker = Pop3SyncWorker() + client = MagicMock() + client.uidl.return_value = ( + b"+OK", + [b"1 uid-old-2", b"2 uid-old-3", b"3 uid-new-4"], + 48, + ) + client.list.return_value = (b"+OK", [b"1 128", b"2 128", b"3 128"], 384) + client.retr.side_effect = lambda message_number: ( + b"+OK", + [f"Message-ID: ".encode(), b"", b"Body"], + 128, + ) + monkeypatch.setattr("services.pop3_worker.poplib.POP3_SSL", lambda host, port: client) + + messages = worker._do_pop3_sync( + _config(), + "pop3.example.com", + 995, + observed_uidls={"uid-old-2", "uid-old-3"}, + ) + + client.retr.assert_called_once_with(3) + assert messages[0].provider_uidl == "uid-new-4" + + +def test_pop3_uidl_unsupported_falls_back_without_claiming_durable_identity(monkeypatch): + worker = Pop3SyncWorker() + client = _client_with_uidls() + client.uidl.side_effect = poplib.error_proto("-ERR UIDL unsupported") + monkeypatch.setattr("services.pop3_worker.poplib.POP3_SSL", lambda host, port: client) + + messages = worker._do_pop3_sync( + _config(), + "pop3.example.com", + 995, + observed_uidls={"uid-3"}, + ) + + assert [entry.args[0] for entry in client.retr.call_args_list] == list(range(3, 13)) + assert all(message.provider_uidl is None for message in messages) + + +@pytest.mark.asyncio +async def test_import_records_uidl_after_successful_persistence(monkeypatch): + worker = Pop3SyncWorker() + config = _config() + raw_message = ( + b"Message-ID: \r\n" + b"From: Sender \r\n" + b"To: pop3-user@example.com\r\n" + b"Subject: UIDL progress\r\n" + b"Date: Mon, 15 Jun 2026 10:00:00 +0000\r\n\r\nBody\r\n" + ) + executed = [] + + class FakeSession: + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def execute(self, statement): + executed.append(statement) + return SimpleNamespace() + + async def commit(self): + return None + + async def rollback(self): + return None + + async def fake_persist(*args, **kwargs): + return SimpleNamespace(created_record=True) + + monkeypatch.setattr("services.pop3_worker.AsyncSessionLocal", lambda: FakeSession()) + monkeypatch.setattr("services.pop3_worker.persist_fetched_email", fake_persist) + + imported_count = await worker._import_messages( + config, + [Pop3RetrievedMessage(source_content=raw_message, provider_uidl="uid-1")], + ) + + assert imported_count == 1 + assert len(executed) == 1 + compiled = str(executed[0].compile(compile_kwargs={"literal_binds": True})) + assert "pop3_observed_messages" in compiled + assert "uid-1" in compiled diff --git a/backend/tests/test_pop3_worker.py b/backend/tests/test_pop3_worker.py index d08e79f6f..9880d7973 100644 --- a/backend/tests/test_pop3_worker.py +++ b/backend/tests/test_pop3_worker.py @@ -1,7 +1,10 @@ import asyncio -import pytest +import poplib +from types import SimpleNamespace from unittest.mock import MagicMock, patch +import pytest + from db.models import TenantConfig from services.pop3_worker import Pop3SyncWorker @@ -72,6 +75,47 @@ def test_pop3_do_sync_validates_destination_before_connect(): pop3_ssl.assert_not_called() +def test_pop3_sync_fetches_newest_bounded_message_numbers(monkeypatch): + worker = Pop3SyncWorker() + config = TenantConfig( + user_id="pop3-user", + pop3_server="pop3.example.com", + pop3_port=995, + pop3_username="pop3-user@example.com", + pop3_password="pop3-secret", + ) + pop3_client = MagicMock() + pop3_client.uidl.side_effect = poplib.error_proto("-ERR UIDL unsupported") + pop3_client.list.return_value = ( + b"+OK", + [f"{message_number} 128".encode() for message_number in range(1, 13)], + 1536, + ) + pop3_client.retr.side_effect = lambda message_number: ( + b"+OK", + [f"Message-ID: ".encode(), b"", b"Body"], + 128, + ) + + monkeypatch.setattr( + "services.pop3_worker.validate_pop3_destination", + lambda host, port: (host, port), + ) + monkeypatch.setattr( + "services.pop3_worker.poplib.POP3_SSL", + lambda host, port: pop3_client, + ) + + messages = worker._do_pop3_sync(config) + + assert [entry.args[0] for entry in pop3_client.retr.call_args_list] == list( + range(3, 13) + ) + assert len(messages) == 10 + assert all(message.provider_uidl is None for message in messages) + pop3_client.quit.assert_called_once() + + @pytest.mark.asyncio async def test_pop3_worker_skips_disallowed_destination(): worker = Pop3SyncWorker() @@ -108,7 +152,7 @@ async def test_pop3_worker_imports_retrieved_messages(monkeypatch): b"Imported from POP3.\r\n" ) pop3_client = MagicMock() - pop3_client.list.return_value = (b"+OK", [b"1 128"], 128) + pop3_client.uidl.return_value = (b"+OK", [b"1 uid-1"], 16) pop3_client.retr.return_value = (b"+OK", raw_message.splitlines(), len(raw_message)) imported: list[dict[str, object]] = [] @@ -131,8 +175,13 @@ async def rollback(self): session = FakeSession() - async def fake_process_fetched_email( - db_session, email_data, user_id, organization_id, owner_addresses=None + async def fake_persist_fetched_email( + db_session, + email_data, + user_id, + organization_id, + owner_addresses=None, + source_content=None, ): imported.append( { @@ -141,8 +190,10 @@ async def fake_process_fetched_email( "user_id": user_id, "organization_id": organization_id, "owner_addresses": owner_addresses, + "source_content": source_content, } ) + return SimpleNamespace(created_record=True) monkeypatch.setattr( "services.pop3_worker.validate_pop3_destination", @@ -153,9 +204,8 @@ async def fake_process_fetched_email( lambda: session, ) monkeypatch.setattr( - "services.pop3_worker.process_fetched_email", - fake_process_fetched_email, - raising=False, + "services.pop3_worker.persist_fetched_email", + fake_persist_fetched_email, ) monkeypatch.setattr( "services.pop3_worker.poplib.POP3_SSL", @@ -166,7 +216,8 @@ async def fake_process_fetched_email( pop3_client.user.assert_called_once_with("pop3-user@example.com") pop3_client.pass_.assert_called_once_with("pop3-secret") - pop3_client.list.assert_called_once() + pop3_client.uidl.assert_called_once() + pop3_client.list.assert_not_called() pop3_client.retr.assert_called_once_with(1) pop3_client.quit.assert_called_once() assert len(imported) == 1 @@ -174,7 +225,66 @@ async def fake_process_fetched_email( assert imported[0]["user_id"] == "pop3-user" assert imported[0]["organization_id"] == "org-pop3" assert imported[0]["owner_addresses"] == ["pop3-user@example.com"] + assert imported[0]["source_content"] == raw_message assert imported[0]["email_data"]["message_id"] == "" assert imported[0]["email_data"]["subject"] == "POP3 import" assert session.committed is True assert session.rolled_back is False + + +@pytest.mark.asyncio +async def test_pop3_duplicate_does_not_inflate_imported_count(monkeypatch): + from db.models import Email + + worker = Pop3SyncWorker() + config = TenantConfig( + user_id="pop3-user", + organization_id="org-pop3", + pop3_server="pop3.example.com", + pop3_port=995, + pop3_username="pop3-user@example.com", + pop3_password="pop3-secret", + ) + raw_message = ( + b"Message-ID: \r\n" + b"From: Sender \r\n" + b"To: pop3-user@example.com\r\n" + b"Subject: Existing message\r\n" + b"Date: Mon, 15 Jun 2026 10:00:00 +0000\r\n" + b"\r\n" + b"Already imported.\r\n" + ) + existing_email = Email(id=1) + + class ExistingResult: + def scalar_one_or_none(self): + return existing_email + + class FakeSession: + def __init__(self): + self.committed = False + self.rolled_back = False + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def execute(self, _statement): + return ExistingResult() + + async def commit(self): + self.committed = True + + async def rollback(self): + self.rolled_back = True + + session = FakeSession() + monkeypatch.setattr("services.pop3_worker.AsyncSessionLocal", lambda: session) + + imported_count = await worker._import_messages(config, [raw_message]) + + assert imported_count == 0 + assert session.committed is True + assert session.rolled_back is False diff --git a/backend/tests/test_source_bound_email_dedupe.py b/backend/tests/test_source_bound_email_dedupe.py new file mode 100644 index 000000000..62c4ddd6b --- /dev/null +++ b/backend/tests/test_source_bound_email_dedupe.py @@ -0,0 +1,201 @@ +"""Regression tests for source-bound fallback email identities.""" + +import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from services.email_dedupe_service import ( + canonical_email_source_content, + source_email_fingerprint, + strong_email_fingerprint, +) +from services.email_import_service import _email_fingerprint +from services.imap_worker import process_fetched_email + + +class _UnsupportedCanonicalValue: + """Represent a value that the parsed EmailData contract never permits.""" + + +def test_source_email_fingerprint_is_stable_content_bound_and_domain_separated() -> ( + None +): + """Hash equal sources equally while separating content and source domains.""" + first = source_email_fingerprint(b"same source") + assert first == source_email_fingerprint(b"same source") + assert first != source_email_fingerprint(b"different source") + assert first != source_email_fingerprint(b"same source", source_kind="canonical") + assert len(first) == 64 + + +def test_canonical_source_content_excludes_collection_date() -> None: + """Keep synthetic collection time outside direct-caller identity.""" + base = { + "message_id": "", + "sender": "sender@example.com", + "recipients": ["one@example.com", "two@example.com"], + "subject": "Subject", + "body": "Body", + "attachments": [{"filename": "note.txt", "content": "note"}], + } + first = { + **base, + "date": datetime.datetime(2026, 8, 4, 6, 30, tzinfo=datetime.timezone.utc), + "date_provenance": "missing", + } + second = { + **base, + "date": datetime.datetime(2026, 8, 4, 7, 30, tzinfo=datetime.timezone.utc), + "date_provenance": "invalid", + } + assert canonical_email_source_content(first) == canonical_email_source_content( + second + ) + assert canonical_email_source_content(first) != canonical_email_source_content( + {**second, "body": "Different body"} + ) + + +@pytest.mark.parametrize( + ("field_name", "unsupported_value"), + [ + ("body", b"raw bytes"), + ("subject", _UnsupportedCanonicalValue()), + ("attachments", [{"content": b"nested raw bytes"}]), + ("attachments", [{"tags": {"unordered", "values"}}]), + ], +) +def test_canonical_source_content_rejects_unsupported_values( + field_name: str, + unsupported_value: object, +) -> None: + """Reject non-EmailData values instead of coercing colliding strings.""" + email_data: dict[str, object] = { + "message_id": "", + "sender": "sender@example.com", + "recipients": "recipient@example.com", + "subject": "Subject", + "body": "Body", + "attachments": [], + field_name: unsupported_value, + } + + with pytest.raises(TypeError, match=field_name): + canonical_email_source_content(email_data) + + +def test_import_fingerprint_uses_trusted_date_or_raw_source() -> None: + """Use strong Date evidence only when provenance is genuinely parsed.""" + persisted_date = datetime.datetime(2026, 8, 4, 6, 30, tzinfo=datetime.timezone.utc) + fields = { + "message_id": "", + "sender": "sender@example.com", + "subject": "Same subject", + "body": "Same parsed body", + "recipients": "recipient@example.com", + } + first_source = b"From: sender@example.com\r\n\r\nFirst raw body" + second_source = b"From: sender@example.com\r\n\r\nSecond raw body" + strong = strong_email_fingerprint( + sender=fields["sender"], + subject=fields["subject"], + date=persisted_date, + body=fields["body"], + ) + assert strong is not None + assert ( + _email_fingerprint( + {**fields, "date_provenance": "parsed"}, + persisted_date, + first_source, + ) + == strong + ) + for provenance in ("missing", "invalid"): + first = _email_fingerprint( + {**fields, "date_provenance": provenance}, + persisted_date, + first_source, + ) + second = _email_fingerprint( + {**fields, "date_provenance": provenance}, + persisted_date, + second_source, + ) + assert first == source_email_fingerprint(first_source) + assert second == source_email_fingerprint(second_source) + assert first != second + + +def test_direct_fallback_is_collection_time_independent() -> None: + """Give callers without raw bytes a stable non-date fallback identity.""" + parsed = { + "message_id": "", + "sender": "sender@example.com", + "recipients": "recipient@example.com", + "subject": "Direct", + "body": "Body", + "date_provenance": "missing", + } + first_time = datetime.datetime(2026, 8, 4, 6, 30, tzinfo=datetime.timezone.utc) + second_time = first_time + datetime.timedelta(hours=1) + assert _email_fingerprint(parsed, first_time) == _email_fingerprint( + parsed, second_time + ) + assert _email_fingerprint(parsed, first_time) != _email_fingerprint( + {**parsed, "body": "Different body"}, second_time + ) + + +@pytest.mark.asyncio +async def test_missing_date_messages_use_raw_source_not_collection_time( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Do not merge different raw messages collected at the same instant.""" + session = AsyncMock() + session.add = MagicMock() + query_result = MagicMock() + query_result.scalar_one_or_none.return_value = None + session.execute.return_value = query_result + monkeypatch.setattr( + "services.imap_worker.assign_thread_id", + AsyncMock(side_effect=("thread-first", "thread-second")), + ) + monkeypatch.setattr( + "services.imap_worker.is_self_sent_email", + lambda _email, _owners: False, + ) + + collected_at = datetime.datetime(2026, 8, 4, 6, 30, tzinfo=datetime.timezone.utc) + common = { + "subject": "Same subject", + "date": collected_at, + "date_provenance": "missing", + "sender": "sender@example.com", + "recipients": "recipient@example.com", + "message_id": "", + } + first_source = b"From: sender@example.com\r\n\r\nFirst raw body" + second_source = b"From: sender@example.com\r\n\r\nSecond raw body" + + first_email = await process_fetched_email( + session, + {**common, "body": "First body"}, + "owner@example.com", + "org-acme", + source_content=first_source, + ) + second_email = await process_fetched_email( + session, + {**common, "body": "Second body"}, + "owner@example.com", + "org-acme", + source_content=second_source, + ) + + assert first_email.date == collected_at + assert second_email.date == collected_at + assert first_email.fingerprint == source_email_fingerprint(first_source) + assert second_email.fingerprint == source_email_fingerprint(second_source) + assert first_email.fingerprint != second_email.fingerprint diff --git a/docs/doctoring/email-source-identity-provenance.md b/docs/doctoring/email-source-identity-provenance.md new file mode 100644 index 000000000..2d3b1805d --- /dev/null +++ b/docs/doctoring/email-source-identity-provenance.md @@ -0,0 +1,272 @@ +# Email source identity provenance + +## Decision + +Naruon treats the sender-authored message and the observation process as +different evidence channels. A collection timestamp can be stored as an +operational timestamp, but it cannot become strong duplicate evidence unless a +valid sender `Date` field was genuinely parsed. Messages without that evidence +use a domain-separated SHA-256 identity over immutable RFC 822 source octets. A +deterministic projection of stable parsed fields is used only when the caller +cannot provide transport bytes. + +This boundary prevents two distinct messages collected at the same instant from +being linked merely because their observation metadata is similar. It also keeps +a repeated import of the same source stable across collection times. + +## Date zone evidence reconciliation + +RFC 5322 requires a zone in `date-time`. Python's +`email.utils.parsedate_to_datetime()` can nevertheless return a naive +`datetime` both for a valid `-0000` Date and for a parseable but non-conforming +Date with no zone. Those cases cannot share provenance semantics: `-0000` +retains sender-supplied Date evidence while an omitted zone does not satisfy the +source grammar and must not seed a strong duplicate identity. + +The broad #1086 lineage therefore keeps one `date_provenance` vocabulary and +separates those cases before binding a naive value to UTC: + +- `-0000` and parser-supported obsolete alphabetic zones with an explicit + trailing zone remain `parsed` and are normalized to timezone-aware UTC for + storage/comparison; +- a parseable Date with no trailing zone is `invalid`; `header_date` remains + absent and the effective stored date is a collection-time UTC fallback; +- the fallback instant is operational data only and never sender metadata. + +Source-order RED `66273d51f142fc46f42bfbe64b330f347f80c8fc` adds the zone-less +parser regression. Causal fix `37429ccb805385621844e7625d5da9e5b11eb6ef` +checks for a trailing RFC 5322/obsolete zone before normalizing a naive parsed +value. This inherits the valid #1656 finding into the broad #1195 lineage +without introducing its competing `date_evidence` column or parallel Alembic +revision. + +## Complete metadata evidence reconciliation + +A parsed Date is necessary but not sufficient for metadata-based automatic +linking. A strong metadata decision also requires non-empty sender, recipients, +subject, and body evidence. Missing any one of those fields makes the metadata +comparison incomplete; the message keeps its raw/canonical source-bound +identity and may enter the review band, but that incomplete tuple cannot create +an automatic metadata link. + +The rule is intentionally a gate rather than a new fingerprint format. The +existing strong fingerprint payload remains sender/subject/Date/body so stored +fingerprint compatibility is not silently broken; recipient evidence determines +whether that fingerprint is authoritative enough to use. Message-ID equality +remains an independent identity signal. + +The source-order sequence for this repair is: + +- `9fdb1207447fe1e47565f92ef57ccd53f02b15f2`: candidate/stored-row RED for + missing sender/recipients/subject/body evidence; +- `d2ae9d6df3906f1018030ff299e0665a96b5127b`: shared complete-metadata gate in + the domain classifier; +- `5d68f08b0de6838a818d59636edf2c2f599138ee` → + `c39f51743598d817edc2f1dfb585a240e0dcc2d7`: direct-import RED then causal + repair, including deterministic `dedupe_review_required` result semantics; +- `fd4cd4e405b3af30b4458993a4673063350b49d9` → + `d8a08deb1ad009b2d90de2eafbab5da61ac0cc79`: IMAP RED then causal repair. + POP3 consumes the same fetched-email persistence boundary rather than + duplicating dedupe rules. + +The canonical migration contract is separately pinned by +`540f6e4ec57ed355263d56e7da78de7ff5310360`: `0018_email_date_provenance` +remains the single message-provenance revision, with non-null +`date_provenance` and an `unknown` server default. The parallel #1656 +`date_evidence` / `message_id_evidence` migration is not adopted. + +## Message-ID provenance decision + +The parser already carries transient `message_id_provenance` (`embedded` or +`missing`) at the ingestion boundary. A missing direct-import Message-ID is +replaced with a deterministic `import-@local.naruon` +identifier, so equality of two such fallback identifiers is already equality of +the same raw-source digest. Persisting a second `message_id_evidence` column has +no current decision consumer or invariant that cannot be represented by the +existing Message-ID plus source-fingerprint semantics. + +Accordingly, #1656's proposed `message_id_evidence` persistence is rejected for +this lineage unless a concrete future consumer proves a storage requirement. +Provider UIDL state is a separate collection-progress concern and must not be +promoted to Naruon's Message-ID or source-fingerprint truth. + +## POP3 reconstruction contract + +POP3 `RETR` is a multiline response. RFC 1939 requires every transmitted line to +end in CRLF and terminates the response with a separate dot line. Python's +`poplib.POP3.retr()` returns the message as a list of lines without those line +terminators. Naruon therefore reconstructs source bytes by joining returned +message lines with CRLF and adding the final message-line CRLF. The POP3 +terminator line is not part of the source message. + +The reconstructed bytes are a transport-normalized POP3 representation. They +are not claimed to reproduce server storage outside the protocol-visible +message. IMAP and direct-file ingestion retain their own exact received byte +streams. Duplicate classification remains deterministic because the source kind +is domain separated and because collection time is excluded from fallback +identity. + +## POP3 durable bounded progress + +RFC 1939 assigns message number `1` to the first message in the opened maildrop +and number `n` to the nth message, but those positions are not durable client +identity. The predecessor implementation first selected the oldest bounded +`LIST` window; repair `72c64b46d120ee8c2f3f12ad04114e6a896cb15b` switched to the +highest current message numbers and removed that particular newest-mail +starvation mode. It still could not guarantee full-maildrop progress because a +static maildrop larger than the cap would repeat the same tail window. + +Issue #1717 therefore requires RFC 1939 UIDL-based progress. RFC 1939 defines a +UIDL value as a one-to-70-character server-determined identifier in the range +0x21–0x7E that identifies a message within a maildrop and persists across +sessions. Naruon now keeps this provider progress identity separate from email +identity: + +- RED `77d7f23ff603821d3247ac48b88247aaa1c70e8e` defines bounded unseen-UIDL + selection, current-session renumbering, an explicit UIDL-unavailable fallback, + and durable observation after successful persistence; +- model `b5344c498ccacaa0b8900be222c6a61bda6e13fe` introduces owner-scoped + `Pop3ObservedMessage`; migration `970c6d3a124cc26dd50a84b0eee58fb951abeef3` + adds `0019_pop3_observed_uidl` after the canonical `0018` provenance revision; +- worker repair `5fe993c1363579e439ee4d74ea0190e964315877` asks the server for UIDL, + filters already observed provider identities, retrieves at most the newest ten + unseen identities for the current poll, and records the UIDL in the same + database transaction as successful email persistence using an idempotent + `(tenant_config_id, provider_uidl)` conflict boundary; +- migration/model contract tests are pinned by + `c5a997615ab8c4ad3e23463199b2ca98a7a5795d`, and model/index parity is repaired + by `99e2e7ea6a3f5d79c546afa11f115b253407319a`; +- legacy worker tests were adapted at `3776e5474ec4ae3b5d3d3cadc6513fedfdd0eadf` + without converting UIDL into message identity. + +With UIDL support, a first poll can process the newest bounded unseen set and a +later poll filters those durable observations, exposing the next unseen set even +if current-session message numbers were renumbered. The database session used to +load progress is closed before POP3 network retrieval starts; `_import_messages` +opens a separate short persistence transaction only after RETR/session cleanup. +No explicit DB lock is held across provider I/O. + +UIDL is optional in POP3. If the server rejects UIDL or returns a malformed UIDL +listing, Naruon falls back to the bounded highest-number `LIST` window and logs +that durable backlog progress is not proven for that poll. The fallback is a +resource-bounded compatibility policy, not a correctness claim. Naruon still +does not issue `DELE`. + +`pop3_observed_messages` intentionally scopes UIDL by `tenant_config_id` and +keeps historical observations because RFC 1939 requires persistence across +sessions. This makes per-account progress reads proportional to stored provider +history; it is a measurable operability/performance surface and must be profiled +before any claim about very large long-lived POP3 mailboxes. + +## POP3 partial retrieval and session teardown + +A successful `RETR` has already returned protocol-visible message bytes before +later message or session cleanup can fail. A later `RETR`/`QUIT` failure must not +retroactively discard earlier successful bytes. + +The teardown source-order repair is: + +- RED `64baa1e2b71e192d14743bd11da017b4fa33279f` requires successful RETR bytes + to survive a `poplib.error_proto` raised by `QUIT`; +- strengthened RED `a74e02b76e236482f2c634a0c54f38450a63b769` also requires an explicit + transport close when graceful `QUIT` fails; +- repairs `fa06c566dc29f350ac1e7ff2888ac59bbf5c7ef4` and + `e809575329ff9b9643d7ce93a28556951cdc797a` preserve retrieved bytes and close + the transport explicitly after failed `QUIT`. + +Earlier partial-RETR repair `37a390c59c915016fa2e412c6e77e85e69042510` +correctly preserved messages retrieved before a later failure, but treated every +`poplib.error_proto` from `RETR` as a reason to stop the remaining bounded batch. +RFC 1939 permits repeated commands in TRANSACTION state and defines `-ERR no +such message` as a valid negative `RETR` response. A single permanently rejected +message therefore must not starve later selected UIDLs when the session remains +usable. + +The refined source-order repair is: + +- RED `d9c87082355fd133a5f4ed9715c560eaee28b211` requires a standards-conforming + per-message `-ERR` to leave that message unobserved while continuing to later + messages in both UIDL and bounded `LIST` fallback paths; an `OSError` remains a + transport-stop condition; +- initial repair `38803cc59be9cee2f72be5f48a9c3c9f1aca402e` separates POP3 protocol + exceptions from transport exceptions; +- hardening RED `7606387c829e27845633d180642369547d23336b` proves that an unexpected + non-`-ERR` protocol exception must not be treated as a safe per-message + rejection; +- final repair `a5d7d21011cf96548cce18b8119a553123a25ab5` continues only when the + protocol exception carries an RFC-style `-ERR` response and stops the + remaining batch on transport loss or malformed/unexpected protocol response. + +Only successfully returned messages can be persisted and only their UIDLs can +become observed. A rejected UIDL therefore remains eligible for retry, while one +persistent message-level rejection no longer blocks later selected messages in +that poll. Naruon still does not issue `DELE`; preserving or continuing retrieval +does not authorize or imply server-side deletion. + +## Verification contract + +- A valid sender `Date` may seed the reviewed strong fingerprint only when + sender/recipients/subject/body evidence is complete. +- Missing and invalid sender dates cannot promote collection time to strong + evidence. +- A parseable but zone-less Date remains invalid source evidence, while an + explicit `-0000` Date remains parsed and UTC-comparable. +- Incomplete metadata cannot produce a strong metadata auto-link; import reports + `dedupe_review_required` while retaining source-bound identity. +- Direct import, IMAP, and POP3 apply the same source-bound persistence rule. +- Two different raw messages collected at the same instant remain distinct; the + same raw message collected at different instants has the same fallback + identity. +- Canonical fallback identity excludes collection timestamps/provenance flags and + rejects unsupported serialization types instead of coercing them with + `str()`. +- IMAP and POP3 pass source bytes through the persistence boundary; POP3 source + reconstruction restores CRLF after every `RETR` message line. +- With UIDL support, a bounded poll retrieves only unseen provider identities; + persisted observations survive reconnects and current-session renumbering. +- UIDL is transport progress identity only. It cannot replace Message-ID, + source-fingerprint, or dedupe evidence. +- UIDL-unavailable/malformed fallback is explicitly compatibility-only and does + not claim eventual backlog completion. +- A per-message RFC-style RETR `-ERR` leaves that identity retryable and does not + starve later selected messages; transport loss or malformed protocol stops the + remaining batch. +- A QUIT failure preserves retrieved bytes and explicitly closes the transport. +- `0018_email_date_provenance` remains the sole message-provenance migration; + `0019_pop3_observed_uidl` succeeds it for provider collection state rather than + introducing parallel message provenance. + +The source-order tests and implementation above are not themselves hosted GREEN +evidence. PostgreSQL migration execution, repository tests/security/coverage, +current-head independent review, restart acceptance, and large-mailbox +performance still require exact-head receipts before #1717 or #1195 can be +accepted complete. + +## Claim boundary + +Hash equality is evidence that the selected source representation is identical; +it is not proof that two independently authored real-world communications are +the same event. Automatic linkage, clerical review, and distinct-message +outcomes remain separate decisions. Provider UIDL proves only the server's +maildrop identity contract. No automatic deletion or irreversible provider +action is introduced. + +## References + +Fellegi, I. P., & Sunter, A. B. (1969). A theory for record linkage. *Journal +of the American Statistical Association, 64*(328), 1183–1210. +https://doi.org/10.1080/01621459.1969.10501049 + +Fellegi and Sunter formalize record linkage by comparing field-level evidence +under match and non-match hypotheses. The resulting evidence score is evaluated +against two decision thresholds: sufficiently strong evidence produces a link, +sufficiently weak evidence produces a non-link, and the intermediate region is +reserved for clerical review. Naruon maps those three outcomes to `auto_link`, +`distinct`, and `review_required` while keeping provenance-gated evidence out of +the automatic-link region. + +Myers, J., & Rose, M. (1996). *Post Office Protocol—Version 3* (RFC 1939; +STD 53). Internet Engineering Task Force. https://doi.org/10.17487/RFC1939 + +Resnick, P. (2008). *Internet message format* (RFC 5322). Internet Engineering +Task Force. https://doi.org/10.17487/RFC5322 diff --git a/docs/doctoring/pop3-durable-retry-progress.md b/docs/doctoring/pop3-durable-retry-progress.md new file mode 100644 index 000000000..8edce81b4 --- /dev/null +++ b/docs/doctoring/pop3-durable-retry-progress.md @@ -0,0 +1,40 @@ +# POP3 durable retry progress + +Issue #1717 exposed a cross-poll starvation case that the first UIDL implementation did not cover. A bounded poll selected the highest ten unseen UIDLs. If all ten returned a standards-conforming `-ERR` from `RETR`, they remained unobserved and were selected again on the next poll. Lower valid UIDLs could therefore remain unreachable indefinitely even though each individual poll correctly continued after the rejected message. + +## Decision + +Provider UIDL remains collection identity, not email identity. Naruon persists one owner-scoped collection disposition per `(tenant_config_id, provider_uidl)`: + +- `observed` means retrieval and email persistence completed; the UIDL is not selected again; +- `retryable` means a message-level collection attempt failed without proving permanent loss; `retry_after` controls when it may re-enter the candidate set. + +Candidate selection prioritizes UIDLs with no durable attempt state before due retries. This is the causal property that prevents a persistent set of retryable failures from monopolizing every bounded poll while fresh backlog exists. The one-minute retry delay matches the worker's existing polling cadence; it is not a terminal-failure threshold and does not discard the UIDL. + +POP3 network I/O still happens before the database session used for collection-state writes. A standards-conforming per-message `-ERR` is returned from the network phase as a retryable UIDL and is persisted only after network I/O completes. A successfully persisted message transitions the same UIDL to `observed`. Parse failures for a retrieved UIDL are also retryable rather than silently observed. Transport loss or malformed protocol response still ends the remaining network batch because the session itself is no longer trusted. + +No `DELE` is issued. The UIDL state does not replace `Message-ID`, sender-authored Date provenance, or source fingerprints. + +## Source trace + +- RED `5d40f83ff48e7dd9c3bf6deb93995130c5a1470a` introduces candidate-selection regressions for fresh backlog versus due/future retry state. +- Fix `7406cb63923003df2c3a5d21a0104203f78b90bc` adds retryable collection disposition, persists `retry_after`, keeps provider I/O outside the DB session, and makes never-attempted UIDLs outrank retries. +- The same fix extends the branch-local `0019_pop3_observed_uidl` schema rather than creating another Alembic revision. Its revision identifier and parent remain non-authoritative until #1503's canonical workspace migration lineage reaches protected ancestry and #1195 is rechained after the then-current protected head. + +## Acceptance boundary + +Source order is not execution evidence. Before merge, the final migration-reconciled exact head still needs: + +- one Alembic head after ordinary adoption of the protected workspace migration lineage; +- fresh PostgreSQL upgrade acceptance from an empty database and from the historical protected migration point; +- repeated-poll and restart acceptance with more than the per-poll cap of persistent negative `RETR` responses ahead of valid unseen mail; +- proof that failed UIDLs remain retryable while valid backlog advances and duplicate import counts remain stable; +- the exact-head repository test, security, coverage, and independent-review gates. + +UIDL-unavailable `LIST` fallback remains compatibility-only and does not claim durable eventual progress. + +## Reference + +Myers, J., & Rose, M. (1996). *Post Office Protocol—Version 3* (RFC 1939). Internet Engineering Task Force. https://doi.org/10.17487/RFC1939 + +RFC 1939 assigns message numbers within the opened maildrop, permits repeated commands in the TRANSACTION state, defines negative `-ERR` command responses, and provides UIDL as the stable provider identifier needed to distinguish messages across sessions. Those protocol properties justify keeping session message numbers out of durable identity and retaining rejected UIDLs as retryable collection state rather than treating them as observed or deleted.