perf: optimize reconciliation candidate lookup

app/commercial_service.py

@@ -529,7 +529,6 @@ def add_document_lines(document_id: str, lines: Iterable[Dict[str, Any]]) -> Non
 
 
 def list_commercial_documents(opportunity_id: Optional[str] = None, customer_id: Optional[str] = None, limit: int = 100) -> List[Dict[str, Any]]:
-    ensure_commercial_schema()
     if opportunity_id:
         from app.document_reconciliation_service import resolve_document_links
         # Canonical resolver owns the only group-complete rollout fallback.
@@ -832,7 +831,6 @@ def unlink_customer_from_opportunity(opportunity_id: str) -> None:
 
 
 def get_customer_for_opportunity(opportunity_id: str) -> Optional[Dict[str, Any]]:
-    ensure_commercial_schema()
     with engine.begin() as conn:
         row = conn.execute(text("""
             SELECT c.id::text, c.name, c.tax_id, c.email, c.phone, c.street_name,

app/jasmin_backfill_service.py

@@ -12,6 +12,7 @@ import json
 
 from sqlalchemy import text
 
+from app.company_opportunity_linking import is_public_email_domain
 from app.db import engine
 from app.reconciliation_service import (
     _apply_jasmin_documents_to_opportunity,  # noqa: PLC2701 - deliberate operator maintenance helper
@@ -466,7 +467,8 @@ def find_jasmin_document_candidates_for_opportunity(opportunity_id: str, *, limi
         fiscal_customer_id = _as_text(opp.get("local_customer_id"))
         fiscal_name = _as_text(opp.get("fiscal_customer_name"))
         fiscal_email = _as_text(opp.get("fiscal_customer_email") or opp.get("customer_email")).lower()
-        domain = _email_domain(fiscal_email)
+        raw_domain = _email_domain(fiscal_email)
+        domain = "" if is_public_email_domain(raw_domain) else raw_domain
 
         current_docs = conn.execute(text("""
             SELECT
@@ -479,8 +481,7 @@ def find_jasmin_document_candidates_for_opportunity(opportunity_id: str, *, limi
             ORDER BY document_date DESC NULLS LAST, created_at DESC
         """), {"opportunity_id": opportunity_id}).mappings().all()
 
-        rows = conn.execute(text("""
+        candidate_columns = """
-            SELECT
                 ri.id::text,
                 ri.source_system,
                 ri.external_type,
@@ -502,52 +503,159 @@ def find_jasmin_document_candidates_for_opportunity(opportunity_id: str, *, limi
                 CASE
                     WHEN CAST(:fiscal_customer_id AS TEXT) <> '' AND ri.customer_id::text = CAST(:fiscal_customer_id AS TEXT) THEN 100
                     WHEN CAST(:fiscal_tax_id AS TEXT) <> '' AND ri.customer_tax_id = CAST(:fiscal_tax_id AS TEXT) THEN 98
-                    WHEN CAST(:fiscal_tax_id AS TEXT) <> '' AND ri.payload::text ILIKE '%' || CAST(:fiscal_tax_id AS TEXT) || '%' THEN 96
+                    {payload_nif_score}
                     WHEN CAST(:fiscal_email AS TEXT) <> '' AND LOWER(COALESCE(ri.customer_email, '')) = CAST(:fiscal_email AS TEXT) THEN 90
-                    WHEN CAST(:domain AS TEXT) <> '' AND ri.payload::text ILIKE '%' || CAST(:domain AS TEXT) || '%' THEN 78
+                    {payload_domain_score}
                     WHEN CAST(:fiscal_name AS TEXT) <> '' AND LOWER(COALESCE(ri.customer_name, '')) = LOWER(CAST(:fiscal_name AS TEXT)) THEN 75
                     ELSE 0
                 END AS match_score,
                 CASE
                     WHEN CAST(:fiscal_customer_id AS TEXT) <> '' AND ri.customer_id::text = CAST(:fiscal_customer_id AS TEXT) THEN 'customer_id'
                     WHEN CAST(:fiscal_tax_id AS TEXT) <> '' AND ri.customer_tax_id = CAST(:fiscal_tax_id AS TEXT) THEN 'nif'
-                    WHEN CAST(:fiscal_tax_id AS TEXT) <> '' AND ri.payload::text ILIKE '%' || CAST(:fiscal_tax_id AS TEXT) || '%' THEN 'payload_nif'
+                    {payload_nif_reason}
                     WHEN CAST(:fiscal_email AS TEXT) <> '' AND LOWER(COALESCE(ri.customer_email, '')) = CAST(:fiscal_email AS TEXT) THEN 'email'
-                    WHEN CAST(:domain AS TEXT) <> '' AND ri.payload::text ILIKE '%' || CAST(:domain AS TEXT) || '%' THEN 'domain_payload'
+                    {payload_domain_reason}
                     WHEN CAST(:fiscal_name AS TEXT) <> '' AND LOWER(COALESCE(ri.customer_name, '')) = LOWER(CAST(:fiscal_name AS TEXT)) THEN 'exact_name'
                     ELSE 'none'
                 END AS match_reason
+        """
+        common_from = """
             FROM reconciliation_items ri
             WHERE ri.source_system = 'jasmin'
               AND ri.external_type IN ('jasmin_quotation', 'jasmin_proforma', 'jasmin_invoice')
               AND (ri.opportunity_id IS NULL OR ri.opportunity_id = CAST(:opportunity_id AS UUID))
+              AND NOT (
+                    CAST(:fiscal_tax_id AS TEXT) <> ''
+                AND COALESCE(ri.customer_tax_id, '') <> ''
+                AND ri.customer_tax_id <> CAST(:fiscal_tax_id AS TEXT)
+              )
               AND NOT EXISTS (
                     SELECT 1 FROM commercial_documents cd
                     WHERE cd.opportunity_id = CAST(:opportunity_id AS UUID)
-                      AND (
+                      AND ri.document_number IS NOT NULL
-                            (ri.document_number IS NOT NULL AND cd.document_number = ri.document_number)
+                      AND cd.document_number = ri.document_number
-                         OR (ri.external_id IS NOT NULL AND cd.external_id = ri.external_id)
               )
+              AND NOT EXISTS (
+                    SELECT 1 FROM commercial_documents cd
+                    WHERE cd.opportunity_id = CAST(:opportunity_id AS UUID)
+                      AND ri.external_id IS NOT NULL
+                      AND cd.external_id = ri.external_id
               )
-              AND (
+        """
-                    (CAST(:fiscal_customer_id AS TEXT) <> '' AND ri.customer_id::text = CAST(:fiscal_customer_id AS TEXT))
+        params = {
-                 OR (CAST(:fiscal_tax_id AS TEXT) <> '' AND ri.customer_tax_id = CAST(:fiscal_tax_id AS TEXT))
-                 OR (CAST(:fiscal_tax_id AS TEXT) <> '' AND ri.payload::text ILIKE '%' || CAST(:fiscal_tax_id AS TEXT) || '%')
-                 OR (CAST(:fiscal_email AS TEXT) <> '' AND LOWER(COALESCE(ri.customer_email, '')) = CAST(:fiscal_email AS TEXT))
-                 OR (CAST(:domain AS TEXT) <> '' AND ri.payload::text ILIKE '%' || CAST(:domain AS TEXT) || '%')
-                 OR (CAST(:fiscal_name AS TEXT) <> '' AND LOWER(COALESCE(ri.customer_name, '')) = LOWER(CAST(:fiscal_name AS TEXT)))
-              )
-            ORDER BY match_score DESC, ri.document_date DESC NULLS LAST, ri.updated_at DESC
-            LIMIT :limit
-        """), {
             "opportunity_id": opportunity_id,
             "fiscal_customer_id": fiscal_customer_id,
             "fiscal_tax_id": fiscal_tax_id,
             "fiscal_email": fiscal_email,
             "domain": domain,
             "fiscal_name": fiscal_name,
-            "limit": int(limit),
+            "limit": max(1, int(limit)),
-        }).mappings().all()
+        }
+        structured_columns = candidate_columns.format(
+            payload_nif_score="",
+            payload_domain_score="",
+            payload_nif_reason="",
+            payload_domain_reason="",
+        )
+        structured_sql = """
+            WITH candidate_ids AS (
+                SELECT ri.id FROM reconciliation_items ri
+                WHERE CAST(:fiscal_customer_id AS TEXT) <> ''
+                  AND ri.customer_id IS NOT NULL
+                  AND ri.customer_id = CAST(NULLIF(:fiscal_customer_id, '') AS UUID)
+                UNION
+                SELECT ri.id FROM reconciliation_items ri
+                WHERE CAST(:fiscal_tax_id AS TEXT) <> ''
+                  AND ri.customer_tax_id IS NOT NULL AND ri.customer_tax_id <> ''
+                  AND ri.customer_tax_id = CAST(:fiscal_tax_id AS TEXT)
+                UNION
+                SELECT ri.id FROM reconciliation_items ri
+                WHERE CAST(:fiscal_email AS TEXT) <> ''
+                  AND ri.customer_email IS NOT NULL AND ri.customer_email <> ''
+                  AND lower(ri.customer_email) = CAST(:fiscal_email AS TEXT)
+                UNION
+                SELECT ri.id FROM reconciliation_items ri
+                WHERE CAST(:fiscal_name AS TEXT) <> ''
+                  AND ri.customer_name IS NOT NULL AND ri.customer_name <> ''
+                  AND lower(ri.customer_name) = lower(CAST(:fiscal_name AS TEXT))
+            )
+            SELECT {columns}
+            {common_from}
+              AND ri.id IN (SELECT id FROM candidate_ids)
+            ORDER BY match_score DESC, ri.document_date DESC NULLS LAST, ri.updated_at DESC
+            LIMIT :limit
+        """.format(columns=structured_columns, common_from=common_from)
+        structured_rows = [dict(row) for row in conn.execute(text(structured_sql), params).mappings().all()]
+
+        # Payload NIF has score 96. It cannot displace a full page made only of
+        # customer-id/NIF matches (100/98), so that is the semantics-preserving
+        # fast path. Email/name matches still require the legacy pass because a
+        # payload NIF candidate may outrank them.
+        strong_page_complete = (
+            len(structured_rows) >= params["limit"]
+            and int(structured_rows[params["limit"] - 1].get("match_score") or 0) >= 96
+        )
+        if strong_page_complete:
+            rows = structured_rows[:params["limit"]]
+        else:
+            payload_predicates: list[str] = []
+            payload_nif_score = payload_nif_reason = ""
+            payload_domain_score = payload_domain_reason = ""
+            if fiscal_tax_id:
+                payload_predicates.append(
+                    "ri.payload::text ILIKE '%' || CAST(:fiscal_tax_id AS TEXT) || '%'"
+                )
+                payload_nif_score = "WHEN ri.payload::text ILIKE '%' || CAST(:fiscal_tax_id AS TEXT) || '%' THEN 96"
+                payload_nif_reason = "WHEN ri.payload::text ILIKE '%' || CAST(:fiscal_tax_id AS TEXT) || '%' THEN 'payload_nif'"
+            if domain:
+                payload_predicates.append(
+                    "ri.payload::text ILIKE '%' || CAST(:domain AS TEXT) || '%'"
+                )
+                payload_domain_score = "WHEN ri.payload::text ILIKE '%' || CAST(:domain AS TEXT) || '%' THEN 78"
+                payload_domain_reason = "WHEN ri.payload::text ILIKE '%' || CAST(:domain AS TEXT) || '%' THEN 'domain_payload'"
+
+            if not payload_predicates:
+                rows = structured_rows[:params["limit"]]
+                payload_rows = []
+            else:
+                payload_columns = candidate_columns.format(
+                    payload_nif_score=payload_nif_score,
+                    payload_domain_score=payload_domain_score,
+                    payload_nif_reason=payload_nif_reason,
+                    payload_domain_reason=payload_domain_reason,
+                )
+                payload_sql = """
+                    SELECT {columns}
+                    {common_from}
+                      AND ({payload_where})
+                    ORDER BY match_score DESC, ri.document_date DESC NULLS LAST, ri.updated_at DESC
+                    LIMIT :limit
+                """.format(
+                    columns=payload_columns,
+                    common_from=common_from,
+                    payload_where=" OR ".join(f"({predicate})" for predicate in payload_predicates),
+                )
+                payload_rows = [dict(row) for row in conn.execute(text(payload_sql), params).mappings().all()]
+
+            if payload_predicates:
+                merged = {str(row.get("id")): row for row in structured_rows}
+                merged.update({str(row.get("id")): row for row in payload_rows})
+                rows = sorted(
+                    merged.values(),
+                    key=lambda row: (
+                        int(row.get("match_score") or 0),
+                        str(row.get("document_date") or ""),
+                        str(row.get("updated_at") or ""),
+                    ),
+                    reverse=True,
+                )[:params["limit"]]
+
+    if fiscal_tax_id:
+        rows = [
+            row for row in rows
+            if not _as_text(row.get("customer_tax_id"))
+            or _as_text(row.get("customer_tax_id")) == fiscal_tax_id
+        ]
 
     current_doc_items = [_commercial_doc_as_jasmin_item(dict(row)) for row in current_docs]
     current_latest_key = max((_jasmin_item_recency_tuple(item) for item in current_doc_items), default=None)

app/operation_service.py

@@ -77,7 +77,6 @@ def integration_settings_summary() -> Dict[str, Dict[str, Any]]:
     }
 
 def get_operation_links(opportunity_id: str) -> List[Dict[str, Any]]:
-    ensure_operation_schema()
     with engine.begin() as conn:
         rows = conn.execute(text("""
             SELECT id::text, opportunity_id::text, system, external_type, external_id, external_name, external_url, status, payload, last_synced_at, created_at, updated_at

app/opportunity_service.py

@@ -1600,7 +1600,6 @@ def list_opportunities(
 
 
 def get_opportunity(opportunity_id: str) -> Optional[Dict[str, Any]]:
-    ensure_opportunity_schema()
     sql = text("""
         SELECT
             o.*,
@@ -1664,7 +1663,6 @@ def get_opportunity(opportunity_id: str) -> Optional[Dict[str, Any]]:
 
 
 def list_opportunity_tasks(opportunity_id: str, *, limit: int = 100) -> List[Dict[str, Any]]:
-    ensure_opportunity_schema()
     sql = text("""
         SELECT
             id::text,
@@ -1693,7 +1691,6 @@ def list_opportunity_tasks(opportunity_id: str, *, limit: int = 100) -> List[Dic
 
 
 def list_opportunity_events(opportunity_id: str, *, limit: int = 100) -> List[Dict[str, Any]]:
-    ensure_opportunity_schema()
     sql = text("""
         SELECT
             id::text,

app/product_service.py

@@ -248,7 +248,6 @@ def list_products(
     active: Optional[str] = "true",
     limit: int = 300,
 ) -> List[Dict[str, Any]]:
-    ensure_product_schema()
     filters = []
     params: Dict[str, Any] = {"limit": int(limit)}
     if q:
@@ -394,7 +393,6 @@ def set_product_active(product_id: str, active: bool) -> bool:
 
 
 def list_opportunity_items(opportunity_id: str) -> List[Dict[str, Any]]:
-    ensure_product_schema()
     sql = text("""
         SELECT
             oi.id::text,

app/reconciliation_service.py

@@ -269,6 +269,11 @@ def ensure_reconciliation_schema() -> None:
         conn.execute(text("CREATE INDEX IF NOT EXISTS idx_reconciliation_type ON reconciliation_items(external_type, status)"))
         conn.execute(text("CREATE INDEX IF NOT EXISTS idx_reconciliation_opportunity ON reconciliation_items(opportunity_id)"))
         conn.execute(text("CREATE INDEX IF NOT EXISTS idx_reconciliation_customer_tax_id ON reconciliation_items(customer_tax_id) WHERE customer_tax_id IS NOT NULL AND customer_tax_id <> ''"))
+        conn.execute(text("CREATE INDEX IF NOT EXISTS idx_reconciliation_customer_id ON reconciliation_items(customer_id) WHERE customer_id IS NOT NULL"))
+        conn.execute(text("CREATE INDEX IF NOT EXISTS idx_reconciliation_customer_email_lower ON reconciliation_items(lower(customer_email)) WHERE customer_email IS NOT NULL AND customer_email <> ''"))
+        conn.execute(text("CREATE INDEX IF NOT EXISTS idx_reconciliation_customer_name_lower ON reconciliation_items(lower(customer_name)) WHERE customer_name IS NOT NULL AND customer_name <> ''"))
+        conn.execute(text("CREATE INDEX IF NOT EXISTS idx_commercial_documents_opportunity_number ON commercial_documents(opportunity_id, document_number) WHERE document_number IS NOT NULL"))
+        conn.execute(text("CREATE INDEX IF NOT EXISTS idx_commercial_documents_opportunity_external ON commercial_documents(opportunity_id, external_id) WHERE external_id IS NOT NULL"))
 
         conn.execute(text("""
             CREATE TABLE IF NOT EXISTS operation_links (

migrations/009_reconciliation_candidate_lookup_indexes.sql

@@ -0,0 +1,21 @@
+-- Structured lookup paths used by opportunity-detail reconciliation candidates.
+-- Existing opportunity, NIF and source/external indexes are intentionally reused.
+CREATE INDEX IF NOT EXISTS idx_reconciliation_customer_id
+ON reconciliation_items(customer_id)
+WHERE customer_id IS NOT NULL;
+
+CREATE INDEX IF NOT EXISTS idx_reconciliation_customer_email_lower
+ON reconciliation_items(lower(customer_email))
+WHERE customer_email IS NOT NULL AND customer_email <> '';
+
+CREATE INDEX IF NOT EXISTS idx_reconciliation_customer_name_lower
+ON reconciliation_items(lower(customer_name))
+WHERE customer_name IS NOT NULL AND customer_name <> '';
+
+CREATE INDEX IF NOT EXISTS idx_commercial_documents_opportunity_number
+ON commercial_documents(opportunity_id, document_number)
+WHERE document_number IS NOT NULL;
+
+CREATE INDEX IF NOT EXISTS idx_commercial_documents_opportunity_external
+ON commercial_documents(opportunity_id, external_id)
+WHERE external_id IS NOT NULL;

tests/test_reconciliation_candidate_lookup_performance.py

@@ -0,0 +1,216 @@
+from contextlib import nullcontext
+from inspect import getsource
+from pathlib import Path
+
+import app.commercial_service as commercial_service
+import app.jasmin_backfill_service as service
+import app.operation_service as operation_service
+import app.opportunity_service as opportunity_service
+import app.product_service as product_service
+
+
+class _Result:
+    def __init__(self, rows=()):
+        self.rows = list(rows)
+
+    def mappings(self):
+        return self
+
+    def all(self):
+        return self.rows
+
+    def first(self):
+        return self.rows[0] if self.rows else None
+
+
+class _Connection:
+    def __init__(self, structured_rows, payload_rows=(), *, fiscal_email="buyer@example.test", fiscal_tax_id="501000000"):
+        self.structured_rows = structured_rows
+        self.payload_rows = payload_rows
+        self.fiscal_email = fiscal_email
+        self.fiscal_tax_id = fiscal_tax_id
+        self.statements = []
+
+    def execute(self, statement, params=None):
+        sql = str(statement)
+        self.statements.append(sql)
+        if "FROM opportunities o" in sql:
+            return _Result([{
+                "id": "11111111-1111-1111-1111-111111111111",
+                "customer_name": "Example",
+                "customer_email": "buyer@example.test",
+                "local_customer_id": "22222222-2222-2222-2222-222222222222",
+                "fiscal_customer_name": "Example Lda",
+                "fiscal_customer_tax_id": self.fiscal_tax_id,
+                "fiscal_customer_email": self.fiscal_email,
+            }])
+        if "FROM commercial_documents" in sql and "SELECT\n                id::text" in sql:
+            return _Result()
+        if "WITH candidate_ids AS" in sql:
+            return _Result(self.structured_rows)
+        if "ri.payload::text ILIKE" in sql:
+            return _Result(self.payload_rows)
+        raise AssertionError(sql)
+
+
+class _Engine:
+    def __init__(self, connection):
+        self.connection = connection
+
+    def begin(self):
+        return nullcontext(self.connection)
+
+
+def _candidate(candidate_id, score, reason, *, document_date="2026-08-01"):
+    return {
+        "id": candidate_id,
+        "source_system": "jasmin",
+        "external_type": "jasmin_invoice",
+        "external_id": candidate_id,
+        "title": candidate_id,
+        "status": "open",
+        "opportunity_id": None,
+        "customer_id": None,
+        "customer_name": "Example Lda",
+        "customer_email": "buyer@example.test",
+        "customer_tax_id": "501000000",
+        "document_number": candidate_id,
+        "document_date": document_date,
+        "amount": 10,
+        "currency": "EUR",
+        "payload": {},
+        "created_at": "2026-08-01",
+        "updated_at": "2026-08-02",
+        "match_score": score,
+        "match_reason": reason,
+    }
+
+
+def test_structured_full_page_skips_payload_statement(monkeypatch):
+    structured = [_candidate("customer", 100, "customer_id"), _candidate("nif", 98, "nif")]
+    connection = _Connection(structured)
+    monkeypatch.setattr(service, "engine", _Engine(connection))
+
+    result = service.find_jasmin_document_candidates_for_opportunity(
+        "11111111-1111-1111-1111-111111111111", limit=2,
+    )
+
+    assert [row["match_reason"] for row in result] == ["customer_id", "nif"]
+    assert len(connection.statements) == 3  # opportunity, current docs, structured
+    assert sum("payload::text ILIKE" in sql for sql in connection.statements) == 0
+
+
+def test_payload_legacy_fallback_preserves_precedence_and_order(monkeypatch):
+    structured = [_candidate("email", 90, "email", document_date="2026-08-10")]
+    payload = [
+        _candidate("payload-nif", 96, "payload_nif", document_date="2026-07-01"),
+        _candidate("payload-domain", 78, "domain_payload", document_date="2026-08-12"),
+    ]
+    connection = _Connection(structured, payload)
+    monkeypatch.setattr(service, "engine", _Engine(connection))
+
+    result = service.find_jasmin_document_candidates_for_opportunity(
+        "11111111-1111-1111-1111-111111111111", limit=5,
+    )
+
+    assert {row["match_reason"] for row in result} == {"payload_nif", "email", "domain_payload"}
+    assert len(connection.statements) == 4
+    assert sum("payload::text ILIKE" in sql for sql in connection.statements) == 1
+
+
+def test_public_domain_never_adds_domain_payload_predicate(monkeypatch):
+    connection = _Connection([], [_candidate("payload-nif", 96, "payload_nif")], fiscal_email="geral.am.energy@gmail.com")
+    monkeypatch.setattr(service, "engine", _Engine(connection))
+
+    result = service.find_jasmin_document_candidates_for_opportunity(
+        "11111111-1111-1111-1111-111111111111", limit=8,
+    )
+
+    assert [row["match_reason"] for row in result] == ["payload_nif"]
+    payload_sql = connection.statements[-1]
+    assert "CAST(:domain AS TEXT)" not in payload_sql
+    assert "domain_payload" not in payload_sql
+
+
+def test_public_domain_without_tax_has_no_payload_statement(monkeypatch):
+    for domain in ("gmail.com", "outlook.com", "hotmail.com", "yahoo.com"):
+        connection = _Connection([], fiscal_email=f"person@{domain}", fiscal_tax_id="")
+        monkeypatch.setattr(service, "engine", _Engine(connection))
+
+        assert service.find_jasmin_document_candidates_for_opportunity(
+            "11111111-1111-1111-1111-111111111111", limit=8,
+        ) == []
+        assert len(connection.statements) == 3
+        assert all("payload::text ILIKE" not in sql for sql in connection.statements)
+
+
+def test_private_business_domain_keeps_domain_payload(monkeypatch):
+    candidate = _candidate("private-domain", 78, "domain_payload")
+    candidate["customer_tax_id"] = ""
+    connection = _Connection([], [candidate], fiscal_email="sales@private-business.pt", fiscal_tax_id="")
+    monkeypatch.setattr(service, "engine", _Engine(connection))
+
+    result = service.find_jasmin_document_candidates_for_opportunity(
+        "11111111-1111-1111-1111-111111111111", limit=8,
+    )
+
+    assert [row["match_reason"] for row in result] == ["domain_payload"]
+    assert "CAST(:domain AS TEXT)" in connection.statements[-1]
+
+
+def test_explicit_conflicting_candidate_nif_is_not_actionable(monkeypatch):
+    conflict = _candidate("conflict", 90, "email")
+    conflict["customer_tax_id"] = "999999999"
+    connection = _Connection([conflict])
+    monkeypatch.setattr(service, "engine", _Engine(connection))
+
+    assert service.find_jasmin_document_candidates_for_opportunity(
+        "11111111-1111-1111-1111-111111111111", limit=8,
+    ) == []
+
+
+def test_required_public_domain_provider_policy():
+    for domain in (
+        "gmail.com", "googlemail.com", "hotmail.com", "outlook.com", "live.com",
+        "yahoo.com", "icloud.com", "me.com", "sapo.pt",
+    ):
+        assert service.is_public_email_domain(domain)
+
+
+def test_candidate_sql_preserves_exclusion_schema_and_ordering():
+    source = getsource(service.find_jasmin_document_candidates_for_opportunity)
+    assert source.count("NOT EXISTS") >= 2
+    assert "cd.document_number = ri.document_number" in source
+    assert "cd.external_id = ri.external_id" in source
+    assert "ORDER BY match_score DESC, ri.document_date DESC NULLS LAST, ri.updated_at DESC" in source
+    for reason in ("customer_id", "nif", "payload_nif", "email", "domain_payload", "exact_name"):
+        assert reason in source
+
+
+def test_opportunity_detail_read_helpers_do_not_initialize_schema():
+    readers = (
+        opportunity_service.get_opportunity,
+        opportunity_service.list_opportunity_tasks,
+        opportunity_service.list_opportunity_events,
+        product_service.list_products,
+        product_service.list_opportunity_items,
+        commercial_service.list_commercial_documents,
+        commercial_service.get_customer_for_opportunity,
+        operation_service.get_operation_links,
+    )
+    for reader in readers:
+        assert "ensure_" not in getsource(reader), reader.__name__
+
+
+def test_only_plan_specific_indexes_are_migrated():
+    migration = Path("migrations/009_reconciliation_candidate_lookup_indexes.sql").read_text()
+    for expression in (
+        "reconciliation_items(customer_id)",
+        "reconciliation_items(lower(customer_email))",
+        "reconciliation_items(lower(customer_name))",
+        "commercial_documents(opportunity_id, document_number)",
+        "commercial_documents(opportunity_id, external_id)",
+    ):
+        assert expression in migration
+    assert "customer_tax_id" not in migration
+    assert "reconciliation_items(opportunity_id)" not in migration