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/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/services/email_dedupe_service.py b/backend/services/email_dedupe_service.py
index 1c74d5d11..679d21a8a 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,96 @@ 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 strong_email_fingerprint(
     *,
     sender: str | None,
@@ -30,6 +140,11 @@ def strong_email_fingerprint(
     date: datetime.datetime | None,
     body: str | None,
 ) -> str | None:
+    """Return the strong (sender+subject+Date+body) auto-dedupe fingerprint.
+
+    Requires a body; ``None`` for an empty body so bodyless rows cannot collapse
+    to a shared hash. Callers gate this on genuine Date provenance.
+    """
     if not body:
         return None
     return generate_email_fingerprint(
@@ -43,6 +158,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 +166,7 @@ def candidate_message_lookup_values(candidate: EmailDedupeCandidate) -> set[str]
 
 
 def candidate_strong_fingerprint(candidate: EmailDedupeCandidate) -> str | None:
+    """Return the candidate's strong fingerprint (see strong_email_fingerprint)."""
     return strong_email_fingerprint(
         sender=candidate.sender,
         subject=candidate.subject,
@@ -59,9 +176,138 @@ 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 genuine Date provenance.
+
+    A stored row may seed a strong (auto-dedupe) fingerprint only when its date
+    is genuinely parsed sender metadata; rows with a synthetic or
+    unknown-provenance date are excluded so they cannot manufacture a strong
+    duplicate match (naruon#1086).
+    """
+    if getattr(email_row, "date_provenance", None) != "parsed":
+        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). 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 or unknown, 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..8906bd076 100644
--- a/backend/services/email_import_service.py
+++ b/backend/services/email_import_service.py
@@ -31,7 +31,11 @@
     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,
+    source_email_fingerprint,
+    strong_email_fingerprint,
+)
 from services.email_parser import EmailData, parse_eml_bytes
 from services.embedding import (
     STORAGE_EMBEDDING_DIMENSION,
@@ -50,7 +54,6 @@
 )
 from services.threading_service import (
     assign_thread_id,
-    generate_email_fingerprint,
     normalize_message_id,
 )
 
@@ -194,20 +197,34 @@ def _message_id_for(parsed: EmailData, content: bytes) -> str:
     )
 
 
-def _email_fingerprint(parsed: EmailData, persisted_date: datetime.datetime) -> str:
-    strong_fingerprint = strong_email_fingerprint(
-        sender=parsed.get("sender"),
-        subject=parsed.get("subject"),
-        date=persisted_date,
-        body=parsed.get("body"),
-    )
+def _email_fingerprint(
+    parsed: EmailData,
+    persisted_date: datetime.datetime,
+    source_content: bytes | None = None,
+) -> str:
+    """Return trusted-Date evidence or a source-bound fallback identity.
+
+    ``persisted_date`` remains the storage timestamp and participates in
+    duplicate evidence only when it came from a valid sender ``Date``.
+    """
+    strong_fingerprint = None
+    if parsed.get("date_provenance") == "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 +387,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 +880,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,
diff --git a/backend/services/email_parser.py b/backend/services/email_parser.py
index be8bee1c4..3c9f04a8e 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]
@@ -152,26 +163,41 @@ 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:
+        # RFC 5322 section 3.3: a ``-0000`` zone means the time zone is
+        # unknown. Normalize the naive parser result to UTC so every parsed
+        # value still satisfies the timezone-aware storage contract.
+        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 +219,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 +243,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..07ceb9f7f 100644
--- a/backend/services/imap_worker.py
+++ b/backend/services/imap_worker.py
@@ -10,14 +10,18 @@
 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,
+    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(
@@ -27,13 +31,11 @@ async def process_fetched_email(
     organization_id: str | None,
     owner_addresses: Iterable[str] | None = None,
     is_read: bool = True,
-):
+    source_content: bytes | None = None,
+) -> Email:
+    """Persist one fetched email with provenance-safe identity."""
     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)
@@ -50,12 +52,24 @@ async def process_fetched_email(
         else str(recipients_list or "")
     )
 
-    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)
+    # Seed strong duplicate evidence only from a genuinely parsed Date.
+    strong_fingerprint = None
+    if email_data.get("date_provenance") == "parsed":
+        strong_fingerprint = strong_email_fingerprint(
+            sender=sender,
+            subject=subject,
+            date=persisted_date,
+            body=email_data.get("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",
+    )
 
     # Check if duplicate
     stmt = select(Email).where(
@@ -87,6 +101,7 @@ async def process_fetched_email(
         recipients=recipients,
         subject=subject,
         date=persisted_date,
+        date_provenance=email_data.get("date_provenance", "unknown"),
         body=email_data.get("body", ""),
         is_read=is_read,
         embedding=[0.0] * 1536,
@@ -98,6 +113,7 @@ async def process_fetched_email(
         await extract_knowledge_from_self_sent(session, new_email, owner_addresses)
     return new_email
 
+
 logger = logging.getLogger(__name__)
 MAX_IMAP_FETCH_MESSAGES = 10
 
@@ -112,7 +128,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 +237,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 +272,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
@@ -334,6 +355,7 @@ async def _import_messages(
                         config.organization_id,
                         owner_addresses=owner_addresses,
                         is_read=is_read,
+                        source_content=raw_message,
                     )
                     imported_count += 1
                 await session.commit()
@@ -388,6 +410,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..be76acdef 100644
--- a/backend/services/pop3_worker.py
+++ b/backend/services/pop3_worker.py
@@ -57,16 +57,18 @@ 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)
 
@@ -126,6 +128,7 @@ async def _import_messages(
                         config.user_id,
                         config.organization_id,
                         owner_addresses=owner_addresses,
+                        source_content=raw_message,
                     )
                     imported_count += 1
                 await session.commit()
@@ -175,11 +178,19 @@ def _do_pop3_sync(
                 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))
+                messages.append(self._message_bytes(lines))
             return messages
         finally:
             pop3_client.quit()
 
+    def _message_bytes(self, lines: list[bytes | str]) -> bytes:
+        """Reconstruct one POP3 RETR message with protocol CRLF terminators.
+
+        ``poplib`` removes line terminators from the multiline response while
+        RFC 1939 defines each transferred message line as CRLF-terminated.
+        """
+        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 = (
             listing.decode("ascii", errors="ignore")
@@ -195,7 +206,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_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="<shared@x>",
+        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="<shared@x>",
+        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="<shared@x>", 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_parser_provenance.py b/backend/tests/test_email_parser_provenance.py
new file mode 100644
index 000000000..64b197006
--- /dev/null
+++ b/backend/tests/test_email_parser_provenance.py
@@ -0,0 +1,127 @@
+"""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: <parsed-date@test.com>
+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: <missing-date@test.com>
+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: <invalid-date@test.com>
+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: <whitespace-date@test.com>
+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: <minus-zero-zone@test.com>
+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_embedded_message_id_provenance() -> None:
+    """A non-empty embedded Message-ID is identified as sender evidence."""
+    parsed = parse_eml_bytes(
+        _eml_with(
+            """Message-ID: <embedded-id@test.com>
+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_worker.py b/backend/tests/test_imap_worker.py
index d2798719a..529b569a0 100644
--- a/backend/tests/test_imap_worker.py
+++ b/backend/tests/test_imap_worker.py
@@ -126,6 +126,7 @@ 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()
@@ -173,7 +174,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_worker.py b/backend/tests/test_pop3_worker.py
index d08e79f6f..9e2cec1cb 100644
--- a/backend/tests/test_pop3_worker.py
+++ b/backend/tests/test_pop3_worker.py
@@ -132,7 +132,12 @@ async def rollback(self):
     session = FakeSession()
 
     async def fake_process_fetched_email(
-        db_session, email_data, user_id, organization_id, owner_addresses=None
+        db_session,
+        email_data,
+        user_id,
+        organization_id,
+        owner_addresses=None,
+        source_content=None,
     ):
         imported.append(
             {
@@ -141,6 +146,7 @@ async def fake_process_fetched_email(
                 "user_id": user_id,
                 "organization_id": organization_id,
                 "owner_addresses": owner_addresses,
+                "source_content": source_content,
             }
         )
 
@@ -174,6 +180,7 @@ 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"] == "<pop3-1@example.com>"
     assert imported[0]["email_data"]["subject"] == "POP3 import"
     assert session.committed is True
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": "<message@example.com>",
+        "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": "<message@example.com>",
+        "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..e0b4d339e
--- /dev/null
+++ b/docs/doctoring/email-source-identity-provenance.md
@@ -0,0 +1,76 @@
+# 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.
+
+## 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.
+
+## Verification contract
+
+- A valid sender `Date` may seed the reviewed strong fingerprint.
+- Missing and invalid sender dates cannot promote collection time to strong
+  evidence.
+- 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 effective collection timestamps and
+  provenance flags.
+- Canonical fallback serialization accepts only deterministic JSON-native parsed
+  values and rejects bytes, unordered collections, custom objects, non-string
+  mapping keys, and non-finite numbers 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.
+- Existing rows remain conservatively classified when provenance is unknown.
+
+## 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. 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