Bulk download: match uploaded_files via hubspot_deal_id bridge + add worker observability

A live test returned 'no documents could be packaged' for 91 correctly-resolved
properties. Root cause: no upload source populates uploaded_files.landlord_property_id
(pashub/magic-plan/audit set hubspot_deal_id; ECMK sets hubspot_listing_id), so
matching on landlord_property_id found zero rows.

Match fix: the worker now joins uploaded_files -> hubspot_deal_data (on deal_id) ->
landlord_property_id, taking the property identity from the bridge. The repository
returns a small PropertyDocument read-model instead of the infra ORM row, so the
orchestrator no longer names infrastructure.postgres.* (resolves the leak dancafc
flagged) and the s3_upload_timestamp cast is gone. Coverage is limited to
deal-id-linked sources; listing_id/uprn-only files are a noted follow-up.

Observability: the empty-selection failure now carries stage counts
(selected/matched/planned/skipped) in both the message (-> the worker WARNING log)
and details (-> sub_task.outputs), and the run logs those counts. The next failure
says 'matched 0 documents' instead of failing opaquely.

ADR-0060 + CONTEXT.md updated (matching decision + considered options).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

CONTEXT.md

@@ -108,7 +108,7 @@ The single ZIP archive a **Bulk Document Download** produces: one folder per pro
 _Avoid_: export (that is the plan/scenario XLSX export), bundle, archive (ambiguous), zip (the format, not the concept)
 
 **Bulk Document Download**:
-The feature/job that assembles a **Download Package** for a chosen set of properties (the union of one or more whole HubSpot projects, selected by `project_code`, and a hand-picked `landlord_property_id` list) and emails the requester a link. Runs on the app-owned-task + attach-mode lane (ADR-0055): the FastAPI route resolves the selection to a distinct `landlord_property_id` set — project codes are expanded via `hubspot_deal_data` (which holds the project↔property grain), hand-picked ids are taken as given — pins that set and the recipient email onto the `sub_task`, and the `applications/bulk_document_download` Lambda builds the package, writes the URL to `sub_task.outputs`, and emails it (ADR-0059/0060). `landlord_property_id` is the selection key throughout because `uploaded_files` is matched on it (it has no `property_id`). Files are matched by `landlord_property_id` (the missing `property_id` on `uploaded_files` is a known future gap). Selection is capped by property count at the trigger.
+The feature/job that assembles a **Download Package** for a chosen set of properties (the union of one or more whole HubSpot projects, selected by `project_code`, and a hand-picked `landlord_property_id` list) and emails the requester a link. Runs on the app-owned-task + attach-mode lane (ADR-0055): the FastAPI route resolves the selection to a distinct `landlord_property_id` set — project codes are expanded via `hubspot_deal_data` (which holds the project↔property grain), hand-picked ids are taken as given — pins that set and the recipient email onto the `sub_task`, and the `applications/bulk_document_download` Lambda builds the package, writes the URL to `sub_task.outputs`, and emails it (ADR-0059/0060). The Lambda matches files by `hubspot_deal_id`, bridging each selected `landlord_property_id` through `hubspot_deal_data` (`uploaded_files.landlord_property_id` is unpopulated in practice, so the property identity comes from the deal bridge); coverage is limited to deal-id-linked sources. Selection is capped by property count at the trigger.
 _Avoid_: document export, bulk export, file dump
 
 ### Source data

docs/adr/0060-bulk-document-download-package-model.md

@@ -55,11 +55,17 @@ lane (ADR-0055).**
   query — cheap, no S3 on the hot path). A `/tmp` size budget inside the Lambda
   is a backstop that fails the sub_task with a clear reason if a pathological
   selection still overflows. `N` starts at a conservative, tunable value.
-- **Matching on `landlord_property_id`.** Files are gathered by
-  `landlord_property_id`; the missing `property_id` on `uploaded_files` is a
-  **known future gap**, out of scope here. Property display info (the folder
-  name) is enriched from the hubspot deals data
-  (`repositories/hubspot_deals/`).
+- **Matching via `hubspot_deal_id`, bridged through `hubspot_deal_data`.** The
+  selection is a set of `landlord_property_id`s, but `uploaded_files` is matched
+  on **`hubspot_deal_id`**: in practice no upload source populates
+  `uploaded_files.landlord_property_id` (pashub, magic plan and the audit
+  generator set `hubspot_deal_id`; ECMK sets `hubspot_listing_id`), so the
+  worker joins `uploaded_files → hubspot_deal_data (on deal_id) → landlord_property_id`
+  and takes the property identity from the bridge, not the file row. **Coverage
+  limitation:** files linked *only* by `hubspot_listing_id` (ECMK) or `uprn` are
+  not reached; extending the bridge to those keys is a follow-up. Property
+  display info (the folder name) is enriched from the property record
+  (`repositories/property/`).
 - **Layout.** One folder per property named by human-readable **address**
   (hubspot-enriched), with `landlord_property_id` appended for uniqueness;
   inside, one file per Document Type = the newest by `s3_upload_timestamp`.
@@ -94,16 +100,21 @@ lane (ADR-0055).**
 - **Strict "all-or-nothing" packaging.** Rejected: one unreadable row or a
   property with no documents would deny the entire package; best-effort +
   reporting matches "include documents and properties where they exist".
-- **Match on `uprn`.** Deferred: `landlord_property_id` is the agreed key and
-  hubspot enrichment is keyed to the deal/property; revisit if/when the
-  `property_id` gap on `uploaded_files` is closed.
+- **Match on `landlord_property_id` directly.** Rejected once tested against
+  real data: `uploaded_files.landlord_property_id` is unpopulated, so a direct
+  match returns nothing. `hubspot_deal_id` (bridged via `hubspot_deal_data`) is
+  the key that upload sources actually set.
+- **Match on `uprn` / `hubspot_listing_id` as well.** Deferred: `hubspot_deal_id`
+  covers the bulk of sources; add `uprn`/`listing_id` to the bridge if a source
+  keyed only by those needs including.
 
 ## Consequences
 
 - New DDD pieces: `applications/bulk_document_download/` (thin handler +
   trigger body), `orchestration/bulk_document_download_orchestrator.py`,
-  packaging rules in `domain/`, a `landlord_property_id`-keyed
-  "latest-per-Document-Type" query on the uploaded-file repository, a
+  packaging rules in `domain/`, a `hubspot_deal_id`-bridged query on the
+  uploaded-file repository that returns a small `PropertyDocument` read-model
+  (so the orchestrator never names `infrastructure.postgres.*`), a
   `generate_presigned_url` + multipart upload on `S3Client`, and the email
   port/adapter from ADR-0059. The only `backend/` touch is the trigger route.
 - The stored, pinned `property_id` set makes a run reproducible even if the
@@ -115,5 +126,8 @@ lane (ADR-0055).**
 - A **new S3 bucket** (`DOCUMENT_EXPORTS_BUCKET`) is provisioned for the
   packages, with its own IAM (single-writer, presign-read). Its retention policy
   is a deliberate follow-up, not set here.
-- Closing the `uploaded_files.property_id` gap later would let matching move off
-  `landlord_property_id` without changing the package model.
+- Because matching bridges through `hubspot_deal_data`, a file is only reachable
+  once its deal has a `hubspot_deal_data` row (and the file carries that
+  `hubspot_deal_id`). Populating `uploaded_files.landlord_property_id` at upload
+  time, or closing the `uploaded_files.property_id` gap, would let matching drop
+  the bridge without changing the package model.

domain/bulk_document_download/property_document.py

@@ -0,0 +1,24 @@
+"""The repository's read-model for a property's uploaded document (ADR-0060).
+
+A small domain type the uploaded-file repository hands back, so the orchestrator
+never names ``infrastructure.postgres.*``. Carries the property's
+``landlord_property_id`` (resolved through the ``hubspot_deal_data`` bridge — see
+the repository) alongside just the fields the Download Package needs.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Optional
+
+
+@dataclass(frozen=True)
+class PropertyDocument:
+    """One uploaded file matched to a property, ready to resolve into the plan."""
+
+    landlord_property_id: str
+    document_type: Optional[str]
+    s3_bucket: str
+    s3_key: str
+    uploaded_at: datetime

orchestration/bulk_document_download_orchestrator.py

@@ -10,28 +10,30 @@ rows are reported, not fatal; a wholly-empty selection is a recorded failure
 
 from __future__ import annotations
 
+import logging
 import os
 import tempfile
 import zipfile
 from dataclasses import dataclass
-from datetime import datetime
-from typing import Optional, Protocol, Sequence, cast
+from typing import Protocol, Sequence
 
 from domain.bulk_document_download.package_plan import (
     ResolvedDocument,
     SkippedDocument,
     plan_download_package,
 )
+from domain.bulk_document_download.property_document import PropertyDocument
 from domain.tasks.subtasks import SubTaskFailure
-from infrastructure.postgres.uploaded_file_table import UploadedFile
 from infrastructure.s3.s3_client import S3Client
 from repositories.email.email_sender import EmailSender
 
+logger = logging.getLogger(__name__)
+
 
 class UploadedFileReader(Protocol):
     def by_landlord_property_ids(
         self, landlord_property_ids: Sequence[str]
-    ) -> list[UploadedFile]: ...
+    ) -> list[PropertyDocument]: ...
 
 
 class AddressResolver(Protocol):
@@ -79,14 +81,28 @@ class BulkDocumentDownloadOrchestrator:
         recipient_email: str,
         package_name: str,
     ) -> DownloadPackageResult:
-        files = self._uploaded_files.by_landlord_property_ids(landlord_property_ids)
-        resolved = [self._resolve(file) for file in files]
-        plan = plan_download_package([r for r in resolved if r is not None])
+        documents = self._uploaded_files.by_landlord_property_ids(
+            landlord_property_ids
+        )
+        plan = plan_download_package([self._resolve(doc) for doc in documents])
 
         skipped: list[SkippedDocument] = list(plan.skipped)
 
+        logger.info(
+            "bulk_document_download: selected=%d matched_documents=%d "
+            "planned_entries=%d skipped=%d",
+            len(landlord_property_ids),
+            len(documents),
+            len(plan.entries),
+            len(skipped),
+        )
+
         if not plan.entries:
-            raise self._empty_failure(skipped)
+            raise self._empty_failure(
+                selected=len(landlord_property_ids),
+                matched=len(documents),
+                skipped=skipped,
+            )
 
         # Stream to a /tmp file and multipart-upload it (ADR-0060): a Download
         # Package can be several GB, so it is never held whole in memory. Each
@@ -120,7 +136,11 @@ class BulkDocumentDownloadOrchestrator:
 
             if included == 0:
                 # Every file that resolved failed to read — no package to send.
-                raise self._empty_failure(skipped)
+                raise self._empty_failure(
+                    selected=len(landlord_property_ids),
+                    matched=len(documents),
+                    skipped=skipped,
+                )
             self._packages.upload_file(zip_path, key)
         url = self._packages.generate_presigned_url(key, self._url_ttl_seconds)
 
@@ -141,13 +161,21 @@ class BulkDocumentDownloadOrchestrator:
             skipped=tuple(skipped),
         )
 
-    def _empty_failure(self, skipped: list[SkippedDocument]) -> SubTaskFailure:
+    def _empty_failure(
+        self, *, selected: int, matched: int, skipped: list[SkippedDocument]
+    ) -> SubTaskFailure:
         """A recorded failure when the selection yields no readable documents
-        (ADR-0060) — never an empty ZIP emailed to the user. The skip detail
-        rides on the failure."""
+        (ADR-0060) — never an empty ZIP emailed to the user. The message carries
+        the stage counts (so they surface in the worker's WARNING log) and the
+        skip detail rides on the failure's ``details`` (→ ``sub_task.outputs``)."""
         return SubTaskFailure(
-            "no documents could be packaged for the selection",
+            f"no documents could be packaged for the selection "
+            f"(selected {selected} properties, matched {matched} documents, "
+            f"{len(skipped)} skipped)",
             details={
+                "selected_properties": selected,
+                "matched_documents": matched,
+                "planned_entries": 0,
                 "skipped": [
                     {
                         "landlord_property_id": s.landlord_property_id,
@@ -155,22 +183,16 @@ class BulkDocumentDownloadOrchestrator:
                         "reason": s.reason,
                     }
                     for s in skipped
-                ]
+                ],
             },
         )
 
-    def _resolve(self, file: UploadedFile) -> Optional[ResolvedDocument]:
-        if file.landlord_property_id is None:
-            return None
+    def _resolve(self, document: PropertyDocument) -> ResolvedDocument:
         return ResolvedDocument(
-            landlord_property_id=file.landlord_property_id,
-            address=self._addresses.address_for(file.landlord_property_id),
-            document_type=file.file_type,
-            s3_bucket=file.s3_file_bucket,
-            s3_key=file.s3_file_key,
-            # TODO: the cast is only needed because UploadedFile types
-            # s3_upload_timestamp as `object` (though it is NOT NULL datetime in
-            # the DB). Type that column `datetime` on the table so callers don't
-            # each cast — see PR #1498 review.
-            uploaded_at=cast(datetime, file.s3_upload_timestamp),
+            landlord_property_id=document.landlord_property_id,
+            address=self._addresses.address_for(document.landlord_property_id),
+            document_type=document.document_type,
+            s3_bucket=document.s3_bucket,
+            s3_key=document.s3_key,
+            uploaded_at=document.uploaded_at,
         )

repositories/uploaded_file/uploaded_file_postgres_repository.py

@@ -2,9 +2,10 @@ from __future__ import annotations
 
 from typing import Optional, Sequence
 
-from sqlalchemy import select
+from sqlalchemy import select, text
 from sqlmodel import Session, col
 
+from domain.bulk_document_download.property_document import PropertyDocument
 from infrastructure.postgres.uploaded_file_table import FileTypeEnum, UploadedFile
 
 
@@ -26,15 +27,40 @@ class UploadedFilePostgresRepository:
 
     def by_landlord_property_ids(
         self, landlord_property_ids: Sequence[str]
-    ) -> list[UploadedFile]:
-        """Every uploaded file held for the given properties, matched by
-        `landlord_property_id` (ADR-0060), across all Document Types. Returns
-        the raw rows — latest-per-Document-Type selection and null-type
-        skipping are the Download Package plan's job, not the query's."""
-        stmt = select(UploadedFile).where(
-            col(UploadedFile.landlord_property_id).in_(list(landlord_property_ids))
+    ) -> list[PropertyDocument]:
+        """Every uploaded file held for the given properties (ADR-0060), across
+        all Document Types, as a small domain read-model.
+
+        Files are matched by **`hubspot_deal_id`**, bridged to the property
+        through `hubspot_deal_data` — `uploaded_files.landlord_property_id` is
+        unpopulated in practice (no upload source sets it), so the property
+        identity comes from the deal bridge, not the file row. Coverage is
+        therefore limited to deal-id-linked sources (pashub, magic plan, audit
+        generator); files keyed only by `hubspot_listing_id` or `uprn` are not
+        reached — see the ADR-0060 matching decision.
+
+        Latest-per-Document-Type selection and null-type skipping are the
+        Download Package plan's job, not the query's."""
+        rows = self._session.connection().execute(
+            text(
+                "SELECT h.landlord_property_id, u.file_type,"
+                " u.s3_file_bucket, u.s3_file_key, u.s3_upload_timestamp"
+                " FROM uploaded_files u"
+                " JOIN hubspot_deal_data h ON h.deal_id = u.hubspot_deal_id"
+                " WHERE h.landlord_property_id = ANY(:lpids)"
+            ),
+            {"lpids": list(landlord_property_ids)},
         )
-        return list(self._session.execute(stmt).scalars().all())  # pyright: ignore[reportDeprecated]
+        return [
+            PropertyDocument(
+                landlord_property_id=row[0],
+                document_type=row[1],
+                s3_bucket=row[2],
+                s3_key=row[3],
+                uploaded_at=row[4],
+            )
+            for row in rows.all()
+        ]
 
     def insert(self, uploaded_file: UploadedFile) -> None:
         self._session.add(uploaded_file)

tests/orchestration/test_bulk_document_download_orchestrator.py

@@ -7,12 +7,14 @@ import io
 import zipfile
 from collections.abc import Iterator
 from datetime import datetime, timezone
+from typing import Optional
 
 import pytest
 from moto import mock_aws
 
+from domain.bulk_document_download.property_document import PropertyDocument
 from domain.tasks.subtasks import SubTaskFailure
-from infrastructure.postgres.uploaded_file_table import FileTypeEnum, UploadedFile
+from infrastructure.postgres.uploaded_file_table import FileTypeEnum
 from infrastructure.s3.s3_client import S3Client
 from orchestration.bulk_document_download_orchestrator import (
     BulkDocumentDownloadOrchestrator,
@@ -23,14 +25,14 @@ DATA_BUCKET = "data-bucket"
 
 
 class _FakeUploadedFiles:
-    def __init__(self, files: list[UploadedFile]) -> None:
-        self._files = files
+    def __init__(self, documents: list[PropertyDocument]) -> None:
+        self._documents = documents
 
     def by_landlord_property_ids(
         self, landlord_property_ids: object
-    ) -> list[UploadedFile]:
+    ) -> list[PropertyDocument]:
         wanted = set(landlord_property_ids)  # type: ignore[call-overload]
-        return [f for f in self._files if f.landlord_property_id in wanted]
+        return [d for d in self._documents if d.landlord_property_id in wanted]
 
 
 class _FakeAddresses:
@@ -60,18 +62,18 @@ class _RecordingEmail:
         self.sent.append((to, subject, body))
 
 
-def _uploaded_file(
+def _document(
     landlord_property_id: str,
-    s3_file_bucket: str,
-    s3_file_key: str,
-    file_type: FileTypeEnum = FileTypeEnum.SITE_NOTE,
-) -> UploadedFile:
-    return UploadedFile(
-        s3_file_bucket=s3_file_bucket,
-        s3_file_key=s3_file_key,
-        s3_upload_timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc),
+    s3_bucket: str,
+    s3_key: str,
+    file_type: Optional[FileTypeEnum] = FileTypeEnum.SITE_NOTE,
+) -> PropertyDocument:
+    return PropertyDocument(
         landlord_property_id=landlord_property_id,
-        file_type=file_type.value,
+        document_type=file_type.value if file_type is not None else None,
+        s3_bucket=s3_bucket,
+        s3_key=s3_key,
+        uploaded_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
     )
 
 
@@ -85,7 +87,7 @@ def packages() -> Iterator[S3Client]:
 
 def test_builds_uploads_and_emails_a_download_package(packages: S3Client) -> None:
     # Arrange — one property with one site note.
-    files = [_uploaded_file("LP1", "src-bucket", "surveys/a.pdf")]
+    files = [_document("LP1", "src-bucket", "surveys/a.pdf")]
     email = _RecordingEmail()
     orchestrator = BulkDocumentDownloadOrchestrator(
         uploaded_files=_FakeUploadedFiles(files),
@@ -133,27 +135,29 @@ def test_a_selection_with_no_documents_is_a_recorded_failure(
 
     # Act / Assert — a recorded SubTask failure, and nobody is emailed an empty
     # package.
-    with pytest.raises(SubTaskFailure):
+    with pytest.raises(SubTaskFailure) as exc_info:
         orchestrator.run(
-            landlord_property_ids=["LP1"],
+            landlord_property_ids=["LP1", "LP2"],
             recipient_email="user@example.com",
             package_name="empty",
         )
     assert email.sent == []
 
+    # ...and it carries the stage counts so the failure is debuggable: the
+    # message (→ the worker's WARNING log) and details (→ sub_task.outputs).
+    failure = exc_info.value
+    assert "matched 0 documents" in str(failure)
+    assert failure.details is not None
+    assert failure.details["selected_properties"] == 2
+    assert failure.details["matched_documents"] == 0
+
 
 def test_packages_the_good_documents_and_reports_the_skipped_ones(
     packages: S3Client,
 ) -> None:
     # Arrange — one packable site note and one row with no Document Type.
-    good = _uploaded_file("LP1", "src-bucket", "surveys/a.pdf")
-    untyped = UploadedFile(
-        s3_file_bucket="src-bucket",
-        s3_file_key="mystery.bin",
-        s3_upload_timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc),
-        landlord_property_id="LP1",
-        file_type=None,
-    )
+    good = _document("LP1", "src-bucket", "surveys/a.pdf")
+    untyped = _document("LP1", "src-bucket", "mystery.bin", file_type=None)
     email = _RecordingEmail()
     orchestrator = BulkDocumentDownloadOrchestrator(
         uploaded_files=_FakeUploadedFiles([good, untyped]),
@@ -185,8 +189,8 @@ def test_a_file_that_cannot_be_read_is_skipped_not_fatal(packages: S3Client) ->
     # Arrange — two documents for one property; the photo pack's object is
     # missing (download raises), the site note reads fine. Best-effort (ADR-0060):
     # the run must still deliver the readable file and report the other.
-    good = _uploaded_file("LP1", "src-bucket", "surveys/a.pdf")
-    missing = _uploaded_file(
+    good = _document("LP1", "src-bucket", "surveys/a.pdf")
+    missing = _document(
         "LP1", "src-bucket", "surveys/gone.pdf", file_type=FileTypeEnum.PHOTO_PACK
     )
     email = _RecordingEmail()
@@ -221,7 +225,7 @@ def test_the_address_fallback_string_becomes_the_folder_name(
     packages: S3Client,
 ) -> None:
     # Arrange — a property whose address resolved to the fallback string.
-    doc = _uploaded_file("LP9", "src-bucket", "surveys/a.pdf")
+    doc = _document("LP9", "src-bucket", "surveys/a.pdf")
     orchestrator = BulkDocumentDownloadOrchestrator(
         uploaded_files=_FakeUploadedFiles([doc]),
         addresses=_FakeAddresses({"LP9": "address unavailable"}),

tests/repositories/uploaded_file/test_uploaded_file_postgres_repository.py

@@ -5,6 +5,7 @@ from datetime import datetime, timedelta, timezone
 from sqlalchemy import Engine
 from sqlmodel import Session
 
+from backend.app.db.models.hubspot_deal_data import HubspotDealData
 from infrastructure.postgres.uploaded_file_table import FileTypeEnum, UploadedFile
 from repositories.uploaded_file.uploaded_file_postgres_repository import (
     UploadedFilePostgresRepository,
@@ -80,28 +81,38 @@ def test_does_not_return_row_with_different_file_type(db_engine: Engine) -> None
     assert result is None
 
 
-def _landlord_file(
-    landlord_property_id: str,
+def _deal_file(
+    hubspot_deal_id: str,
     s3_file_key: str,
     file_type: FileTypeEnum = FileTypeEnum.SITE_NOTE,
 ) -> UploadedFile:
+    # A real uploaded file: keyed by hubspot_deal_id, landlord_property_id NULL
+    # (no upload source populates it — see ADR-0060 matching decision).
     return UploadedFile(
         s3_file_bucket=_BUCKET,
         s3_file_key=s3_file_key,
         s3_upload_timestamp=datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc),
-        landlord_property_id=landlord_property_id,
+        hubspot_deal_id=hubspot_deal_id,
         file_type=file_type.value,
     )
 
 
-def test_by_landlord_property_ids_returns_only_requested_properties(
+def _deal(landlord_property_id: str, deal_id: str) -> HubspotDealData:
+    return HubspotDealData(deal_id=deal_id, landlord_property_id=landlord_property_id)
+
+
+def test_by_landlord_property_ids_bridges_deal_id_to_the_property(
     db_engine: Engine,
 ) -> None:
-    # Arrange — files for three properties; only two are requested.
+    # Arrange — files are keyed by hubspot_deal_id; hubspot_deal_data bridges each
+    # deal to a landlord_property_id. Three properties exist; two are requested.
     with Session(db_engine) as session:
-        session.add(_landlord_file("LP1", "one.pdf"))
-        session.add(_landlord_file("LP2", "two.pdf"))
-        session.add(_landlord_file("LP3", "three.pdf"))
+        session.add(_deal("LP1", "deal-1"))
+        session.add(_deal("LP2", "deal-2"))
+        session.add(_deal("LP3", "deal-3"))
+        session.add(_deal_file("deal-1", "one.pdf"))
+        session.add(_deal_file("deal-2", "two.pdf"))
+        session.add(_deal_file("deal-3", "three.pdf"))
         session.commit()
 
     # Act
@@ -109,7 +120,25 @@ def test_by_landlord_property_ids_returns_only_requested_properties(
         found = UploadedFilePostgresRepository(session).by_landlord_property_ids(
             ["LP1", "LP2"]
         )
-        result = {(f.landlord_property_id, f.s3_file_key) for f in found}
+        result = {(d.landlord_property_id, d.s3_key) for d in found}
+
+    # Assert — the file's property comes from the bridge, keyed by the deal.
+    assert result == {("LP1", "one.pdf"), ("LP2", "two.pdf")}
+
+
+def test_by_landlord_property_ids_matches_nothing_without_a_deal_bridge(
+    db_engine: Engine,
+) -> None:
+    # Arrange — a file whose deal has no hubspot_deal_data row can't be reached.
+    with Session(db_engine) as session:
+        session.add(_deal_file("orphan-deal", "orphan.pdf"))
+        session.commit()
+
+    # Act
+    with Session(db_engine) as session:
+        found = UploadedFilePostgresRepository(session).by_landlord_property_ids(
+            ["LP1"]
+        )
 
     # Assert
-    assert result == {("LP1", "one.pdf"), ("LP2", "two.pdf")}
+    assert found == []