diff --git a/applications/abri/abri_trigger_request.py b/applications/abri/abri_trigger_request.py
index 003fe47bb..50deed371 100644
--- a/applications/abri/abri_trigger_request.py
+++ b/applications/abri/abri_trigger_request.py
@@ -45,12 +45,15 @@ class AbriTriggerRequest(BaseModel):
     # OpenHousing's bookable resource for the survey; required by the log and
     # amend flows. Empty string is also rejected by Abri, so not allowed here
     third_party_surveyor_identifier: Optional[str] = None
+    # The Job Number OpenHousing returns on a successful log. Present means the
+    # job exists (amend/abandon); absent means it was never logged (or the
+    # reference was lost). Passed through, not required by any flow. A blank
+    # value carries no Job Number, so it is normalised to missing.
+    client_booking_reference: Optional[str] = None
 
-    @field_validator("third_party_surveyor_identifier")
+    @field_validator("third_party_surveyor_identifier", "client_booking_reference")
     @classmethod
-    def _blank_surveyor_identifier_is_missing(
-        cls, value: Optional[str]
-    ) -> Optional[str]:
+    def _blank_is_missing(cls, value: Optional[str]) -> Optional[str]:
         if value is not None and not value.strip():
             return None
         return value
diff --git a/applications/abri/dispatch.py b/applications/abri/dispatch.py
index 7d03f1120..2362fa1d0 100644
--- a/applications/abri/dispatch.py
+++ b/applications/abri/dispatch.py
@@ -44,6 +44,24 @@ class AbriFlows(Protocol):
     ) -> AbandonJobOrchestrationResult: ...
 
 
+class DealDivergedFromAbriError(NonRetriableTaskError):
+    """An amend/abandon on a deal carrying no client_booking_reference.
+
+    An absent Job Number is ambiguous — the job was never logged, or the
+    reference OpenHousing returned was lost — so the change or cancellation
+    cannot be synced to Abri. Non-retriable: redelivery cannot conjure a
+    reference; a human must reconcile the deal against OpenHousing.
+    """
+
+    def __init__(self, flow: str, deal_id: str) -> None:
+        self.flow = flow
+        self.deal_id = deal_id
+        super().__init__(
+            f"{flow} cannot run for deal {deal_id}: no client_booking_reference "
+            "(Job Number) — the deal has diverged from Abri; check OpenHousing"
+        )
+
+
 class AbriFlowRejectedError(Exception):
     """An OpenHousing rejection, surfaced as a failure so the task fails.
 
@@ -81,6 +99,10 @@ def dispatch_abri_flows(
 
 
 def _run_amend(request: AbriTriggerRequest, flows: AbriFlows) -> str:
+    if request.client_booking_reference is None:
+        raise DealDivergedFromAbriError(
+            flow="amend_job", deal_id=request.hubspot_deal_id
+        )
     if (
         request.confirmed_survey_date is None
         or request.third_party_surveyor_identifier is None
@@ -147,8 +169,13 @@ def _run_log(
 
 def _run_abandon(request: AbriTriggerRequest, flows: AbriFlows) -> str:
     # Runs last so a cancellation can never interfere with a log, amend or
-    # tenant sync in the same batch. A missing job_no raises the retriable
-    # not-yet-recorded error, which propagates so the task fails and redelivers.
+    # tenant sync in the same batch. With a Job Number, a missing DB job_no
+    # raises the retriable not-yet-recorded error, which propagates so the task
+    # fails and redelivers; with no Job Number at all, the deal has diverged.
+    if request.client_booking_reference is None:
+        raise DealDivergedFromAbriError(
+            flow="abandon_job", deal_id=request.hubspot_deal_id
+        )
     outcome = flows.abandon_job(
         DealAbandonment(
             deal_id=request.hubspot_deal_id,
diff --git a/etl/hubspot/hubspot_deal_differ.py b/etl/hubspot/hubspot_deal_differ.py
index bb1e58674..e50e07dc8 100644
--- a/etl/hubspot/hubspot_deal_differ.py
+++ b/etl/hubspot/hubspot_deal_differ.py
@@ -268,6 +268,12 @@ class HubspotDealDiffer:
             "third_party_surveyor_identifier": new_deal.get(
                 "third_party_surveyor_identifier"
             ),
+            # The Job Number OpenHousing returned on a successful log, if any.
+            # The consumer reads it to tell "job exists" (amend/abandon) from
+            # "never logged" (log).
+            "client_booking_reference": HubspotDealDiffer._client_booking_reference(
+                new_deal
+            ),
             # Carried for the abandon flow: it dates the cancel to the confirmed
             # survey date, falling back to the last submission date.
             "last_submission_date": (
@@ -295,19 +301,37 @@ class HubspotDealDiffer:
         if not HubspotDealDiffer._is_abri_condition_deal(new_deal, new_project):
             return False
 
-        new_survey_date = parse_hs_date(new_deal.get("confirmed_survey_date"))
-        logger.info(
-            "Abri job-logging check: old_confirmed_survey_date=%s "
-            "new_confirmed_survey_date=%s",
-            old_deal.confirmed_survey_date,
-            new_survey_date,
-        )
-
-        # Only a first-time survey date (previously empty) logs a new job.
-        if old_deal.confirmed_survey_date is not None:
+        # A deal that already carries a Job Number is logged; never log it
+        # again (that would duplicate the job in OpenHousing).
+        if HubspotDealDiffer._client_booking_reference(new_deal) is not None:
             return False
 
-        return new_survey_date is not None
+        new_survey_date = parse_hs_date(new_deal.get("confirmed_survey_date"))
+        new_surveyor = HubspotDealDiffer._normalised_surveyor(
+            new_deal.get("third_party_surveyor_identifier")
+        )
+        old_surveyor = HubspotDealDiffer._normalised_surveyor(
+            old_deal.third_party_surveyor_identifier
+        )
+        pair_complete_now = new_survey_date is not None and new_surveyor is not None
+        pair_complete_before = (
+            old_deal.confirmed_survey_date is not None and old_surveyor is not None
+        )
+        logger.info(
+            "Abri job-logging check: old_confirmed_survey_date=%s "
+            "new_confirmed_survey_date=%s old_surveyor=%s new_surveyor=%s "
+            "pair_complete_before=%s pair_complete_now=%s",
+            old_deal.confirmed_survey_date,
+            new_survey_date,
+            old_surveyor,
+            new_surveyor,
+            pair_complete_before,
+            pair_complete_now,
+        )
+
+        # Fire exactly once, on the scrape that completes the pair — whichever
+        # of the survey date or surveyor lands second.
+        return pair_complete_now and not pair_complete_before
 
     @staticmethod
     def check_for_abri_job_amendment(
@@ -340,21 +364,31 @@ class HubspotDealDiffer:
             new_surveyor,
         )
 
-        # A first-time survey date logs a new job, not an amendment.
-        if old_deal.confirmed_survey_date is None:
+        # Only a complete booking (both date and surveyor set now) can be
+        # amended; a lone field is nothing to amend, and a pair still forming
+        # waits for the log flow.
+        # TODO: a cleared survey date (was a complete booking, now not) no
+        # longer fires this check; the amendoptiappt route cannot express a
+        # cancellation, so cancellation semantics (the canceljob route) need a
+        # deliberate decision (existing differ TODO, issue #1478 out of scope).
+        if new_survey_date is None or new_surveyor is None:
             return False
 
-        # TODO: a cleared survey date (was set, now empty) still fires this
-        # check, but the amendoptiappt route cannot express a cancellation.
-        # Cancellation semantics (the "delete" action / canceljob route) need
-        # a deliberate decision before this trigger is wired to a consumer.
-        if old_deal.confirmed_survey_date != new_survey_date:
-            return True
+        change_occurred = (
+            old_deal.confirmed_survey_date != new_survey_date
+            or old_deal.confirmed_survey_time != new_survey_time
+            or old_surveyor != new_surveyor
+        )
+        if not change_occurred:
+            return False
 
-        if old_deal.confirmed_survey_time != new_survey_time:
-            return True
+        # The scrape that completes the pair logs the job; it never amends one.
+        if HubspotDealDiffer.check_for_abri_job_logging(
+            new_deal=new_deal, new_project=new_project, old_deal=old_deal
+        ):
+            return False
 
-        return old_surveyor != new_surveyor
+        return True
 
     @staticmethod
     def check_for_abri_tenant_data_fetch(
@@ -421,13 +455,25 @@ class HubspotDealDiffer:
         return new_outcome == "surveyed" and old_outcome != "surveyed"
 
     @staticmethod
-    def _normalised_surveyor(value: Optional[str]) -> Optional[str]:
-        # HubSpot sends "" for an unset property and the DB stores whatever
-        # the scrape wrote, so blank and missing must compare as equal.
+    def _blank_to_none(value: Optional[str]) -> Optional[str]:
+        # HubSpot sends "" for an unset property and the DB stores whatever the
+        # scrape wrote, so blank and missing must always compare as equal.
         if value is None or not value.strip():
             return None
         return value
 
+    @staticmethod
+    def _normalised_surveyor(value: Optional[str]) -> Optional[str]:
+        return HubspotDealDiffer._blank_to_none(value)
+
+    @staticmethod
+    def _client_booking_reference(new_deal: Dict[str, str]) -> Optional[str]:
+        # The Job Number OpenHousing returns on a successful log; a blank value
+        # carries no Job Number, so blank and missing both read as "no job".
+        return HubspotDealDiffer._blank_to_none(
+            new_deal.get("client_booking_reference")
+        )
+
     @staticmethod
     def _is_abandoned(
         number_of_attempts: Optional[str], outcome: Optional[str]
diff --git a/etl/hubspot/tests/test_abri_flow_triggers.py b/etl/hubspot/tests/test_abri_flow_triggers.py
index 1e0ecdf52..77aa04daf 100644
--- a/etl/hubspot/tests/test_abri_flow_triggers.py
+++ b/etl/hubspot/tests/test_abri_flow_triggers.py
@@ -40,9 +40,72 @@ def make_new_deal(**overrides: str) -> Dict[str, str]:
 # ==========================
 
 
-def test_a_first_confirmed_survey_date_builds_a_log_job_message() -> None:
-    # Arrange
-    old_deal = make_old_deal(project_code=ABRI_PROJECT_CODE, confirmed_survey_date=None)
+def test_a_surveyor_landing_after_the_date_completes_the_pair_and_logs() -> None:
+    # Arrange: the date was already set; the surveyor lands second, completing
+    # the pair, so the job is logged (not amended) whichever field lands last.
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date=datetime(2026, 6, 24, tzinfo=timezone.utc),
+        confirmed_survey_time="14:30",
+        third_party_surveyor_identifier=None,
+    )
+    new_deal = make_new_deal(
+        confirmed_survey_date="2026-06-24",
+        confirmed_survey_time="14:30",
+        third_party_surveyor_identifier="THIRDPARTY",
+    )
+
+    # Act
+    message = HubspotDealDiffer.check_abri_triggers_and_construct_message(
+        hubspot_deal_id=DEAL_ID,
+        new_deal=new_deal,
+        new_project=None,
+        new_listing=LISTING,
+        old_deal=old_deal,
+    )
+
+    # Assert: the completing scrape logs (amend suppression is asserted
+    # separately); the point here is that a late surveyor logs, not amends.
+    assert message is not None
+    assert "log_job" in message["flows"]
+
+
+def test_a_date_landing_after_the_surveyor_completes_the_pair_and_logs() -> None:
+    # Arrange: the surveyor was already set; the date lands second, completing
+    # the pair, so the job is logged — field ordering does not matter.
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date=None,
+        third_party_surveyor_identifier="THIRDPARTY",
+    )
+    new_deal = make_new_deal(
+        confirmed_survey_date="2026-06-24",
+        confirmed_survey_time="14:30",
+        third_party_surveyor_identifier="THIRDPARTY",
+    )
+
+    # Act
+    message = HubspotDealDiffer.check_abri_triggers_and_construct_message(
+        hubspot_deal_id=DEAL_ID,
+        new_deal=new_deal,
+        new_project=None,
+        new_listing=LISTING,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert message is not None
+    assert message["flows"] == ["log_job"]
+
+
+def test_a_lone_survey_date_without_a_surveyor_fires_nothing() -> None:
+    # Arrange: only the date lands; the pair is incomplete, so the job is not
+    # logged (the resource requirement is kept) and nothing is amended.
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date=None,
+        third_party_surveyor_identifier=None,
+    )
     new_deal = make_new_deal(
         confirmed_survey_date="2026-06-24", confirmed_survey_time="14:30"
     )
@@ -57,17 +120,88 @@ def test_a_first_confirmed_survey_date_builds_a_log_job_message() -> None:
     )
 
     # Assert
-    assert message == {
-        "hubspot_deal_id": DEAL_ID,
-        "flows": ["log_job"],
-        "place_ref": "1007165",
-        "deal_name": "49 Admers Crescent",
-        "confirmed_survey_date": "2026-06-24",
-        "confirmed_survey_time": "14:30",
-        "last_submission_date": None,
-        "outcome": None,
-        "third_party_surveyor_identifier": None,
-    }
+    assert message is None
+
+
+def test_a_lone_surveyor_without_a_survey_date_fires_nothing() -> None:
+    # Arrange: only the surveyor lands; the pair is incomplete, so nothing fires.
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date=None,
+        third_party_surveyor_identifier=None,
+    )
+    new_deal = make_new_deal(third_party_surveyor_identifier="THIRDPARTY")
+
+    # Act
+    message = HubspotDealDiffer.check_abri_triggers_and_construct_message(
+        hubspot_deal_id=DEAL_ID,
+        new_deal=new_deal,
+        new_project=None,
+        new_listing=LISTING,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert message is None
+
+
+def test_a_pair_already_complete_before_does_not_re_log() -> None:
+    # Arrange: both fields were already set and nothing changed, so the log
+    # must not fire a second time.
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date=datetime(2026, 6, 24, tzinfo=timezone.utc),
+        confirmed_survey_time="14:30",
+        third_party_surveyor_identifier="THIRDPARTY",
+    )
+    new_deal = make_new_deal(
+        confirmed_survey_date="2026-06-24",
+        confirmed_survey_time="14:30",
+        third_party_surveyor_identifier="THIRDPARTY",
+    )
+
+    # Act
+    message = HubspotDealDiffer.check_abri_triggers_and_construct_message(
+        hubspot_deal_id=DEAL_ID,
+        new_deal=new_deal,
+        new_project=None,
+        new_listing=LISTING,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert message is None
+
+
+def test_a_deal_that_already_has_a_client_booking_reference_is_never_logged() -> None:
+    # Arrange: the pair completes this scrape, but the deal already carries a
+    # Job Number, so the job exists and must not be logged again.
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date=datetime(2026, 6, 24, tzinfo=timezone.utc),
+        confirmed_survey_time="14:30",
+        third_party_surveyor_identifier=None,
+        client_booking_reference="AD0226519",
+    )
+    new_deal = make_new_deal(
+        confirmed_survey_date="2026-06-24",
+        confirmed_survey_time="14:30",
+        third_party_surveyor_identifier="THIRDPARTY",
+        client_booking_reference="AD0226519",
+    )
+
+    # Act
+    message = HubspotDealDiffer.check_abri_triggers_and_construct_message(
+        hubspot_deal_id=DEAL_ID,
+        new_deal=new_deal,
+        new_project=None,
+        new_listing=LISTING,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert message is not None
+    assert "log_job" not in message["flows"]
 
 
 def test_the_message_carries_the_deals_third_party_surveyor_identifier() -> None:
@@ -93,14 +227,75 @@ def test_the_message_carries_the_deals_third_party_surveyor_identifier() -> None
     assert message["third_party_surveyor_identifier"] == "THIRDPARTY"
 
 
-def test_a_changed_confirmed_survey_date_builds_an_amend_job_message() -> None:
-    # Arrange
+def test_the_message_carries_the_deals_client_booking_reference() -> None:
+    # Arrange: an amend on a deal that already carries a Job Number; the
+    # consumer reads the reference off the message to decide the job exists.
     old_deal = make_old_deal(
         project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date=datetime(2026, 6, 18, tzinfo=timezone.utc),
+        third_party_surveyor_identifier="THIRDPARTY",
+        client_booking_reference="AD0226519",
     )
     new_deal = make_new_deal(
-        confirmed_survey_date="2026-06-24", confirmed_survey_time="14:30"
+        confirmed_survey_date="2026-06-24",
+        confirmed_survey_time="14:30",
+        third_party_surveyor_identifier="THIRDPARTY",
+        client_booking_reference="AD0226519",
+    )
+
+    # Act
+    message = HubspotDealDiffer.check_abri_triggers_and_construct_message(
+        hubspot_deal_id=DEAL_ID,
+        new_deal=new_deal,
+        new_project=None,
+        new_listing=LISTING,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert message is not None
+    assert message["flows"] == ["amend_job"]
+    assert message["client_booking_reference"] == "AD0226519"
+
+
+def test_the_message_omits_the_client_booking_reference_when_the_deal_has_none() -> None:
+    # Arrange: a pair-completing log on a deal with no Job Number yet.
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date=None,
+        third_party_surveyor_identifier=None,
+    )
+    new_deal = make_new_deal(
+        confirmed_survey_date="2026-06-24",
+        confirmed_survey_time="14:30",
+        third_party_surveyor_identifier="THIRDPARTY",
+    )
+
+    # Act
+    message = HubspotDealDiffer.check_abri_triggers_and_construct_message(
+        hubspot_deal_id=DEAL_ID,
+        new_deal=new_deal,
+        new_project=None,
+        new_listing=LISTING,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert message is not None
+    assert message["client_booking_reference"] is None
+
+
+def test_a_changed_survey_date_on_a_complete_booking_builds_an_amend_job_message() -> None:
+    # Arrange: a complete booking (date + surveyor) whose date moves.
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date=datetime(2026, 6, 18, tzinfo=timezone.utc),
+        third_party_surveyor_identifier="THIRDPARTY",
+    )
+    new_deal = make_new_deal(
+        confirmed_survey_date="2026-06-24",
+        confirmed_survey_time="14:30",
+        third_party_surveyor_identifier="THIRDPARTY",
     )
 
     # Act
@@ -118,6 +313,62 @@ def test_a_changed_confirmed_survey_date_builds_an_amend_job_message() -> None:
     assert message["confirmed_survey_date"] == "2026-06-24"
 
 
+def test_log_takes_precedence_over_amend_on_the_pair_completing_scrape() -> None:
+    # Arrange: the surveyor lands to complete the pair on a deal with no Job
+    # Number. The surveyor change would otherwise read as an amend, but the
+    # pair-completing scrape must log, not amend.
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date=datetime(2026, 6, 24, tzinfo=timezone.utc),
+        confirmed_survey_time="14:30",
+        third_party_surveyor_identifier=None,
+    )
+    new_deal = make_new_deal(
+        confirmed_survey_date="2026-06-24",
+        confirmed_survey_time="14:30",
+        third_party_surveyor_identifier="THIRDPARTY",
+    )
+
+    # Act
+    message = HubspotDealDiffer.check_abri_triggers_and_construct_message(
+        hubspot_deal_id=DEAL_ID,
+        new_deal=new_deal,
+        new_project=None,
+        new_listing=LISTING,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert message is not None
+    assert message["flows"] == ["log_job"]
+
+
+def test_a_date_change_on_an_incomplete_pair_does_not_amend() -> None:
+    # Arrange: the date moves while no surveyor is set — there is no complete
+    # booking to amend, so nothing fires until the pair completes.
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date=datetime(2026, 6, 18, tzinfo=timezone.utc),
+        third_party_surveyor_identifier=None,
+    )
+    new_deal = make_new_deal(
+        confirmed_survey_date="2026-06-24",
+        third_party_surveyor_identifier="",
+    )
+
+    # Act
+    message = HubspotDealDiffer.check_abri_triggers_and_construct_message(
+        hubspot_deal_id=DEAL_ID,
+        new_deal=new_deal,
+        new_project=None,
+        new_listing=LISTING,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert message is None
+
+
 def test_a_changed_surveyor_on_a_booked_deal_builds_an_amend_job_message() -> None:
     # Arrange: booking exists; only the surveyor changes.
     old_deal = make_old_deal(
@@ -255,15 +506,18 @@ def test_a_first_expected_commencement_date_builds_a_tenant_sync_message() -> No
 
 
 def test_several_flows_firing_in_one_scrape_share_one_message() -> None:
-    # Arrange: date changed and commencement date first set together.
+    # Arrange: on a complete booking, the date changed (amend) and the
+    # commencement date was first set (tenant sync) together.
     old_deal = make_old_deal(
         project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date=datetime(2026, 6, 18, tzinfo=timezone.utc),
+        third_party_surveyor_identifier="THIRDPARTY",
         expected_commencement_date=None,
     )
     new_deal = make_new_deal(
         confirmed_survey_date="2026-06-24",
         confirmed_survey_time="14:30",
+        third_party_surveyor_identifier="THIRDPARTY",
         expected_commencement_date="2026-07-01",
     )
 
@@ -328,9 +582,15 @@ def test_the_test_override_env_var_fires_a_deal_on_its_own_project_code(
     # Arrange: a sandbox deal on its own project code, with the test override
     # pointing the gate at that code so it fires like a production Abri deal.
     monkeypatch.setenv("TEST_ABRI_PROJECT_CODE_OVERRIDE", "TEST-SANDBOX-CODE")
-    old_deal = make_old_deal(project_code="TEST-SANDBOX-CODE", confirmed_survey_date=None)
+    old_deal = make_old_deal(
+        project_code="TEST-SANDBOX-CODE",
+        confirmed_survey_date=None,
+        third_party_surveyor_identifier=None,
+    )
     new_deal = make_new_deal(
-        project_code="TEST-SANDBOX-CODE", confirmed_survey_date="2026-06-24"
+        project_code="TEST-SANDBOX-CODE",
+        confirmed_survey_date="2026-06-24",
+        third_party_surveyor_identifier="THIRDPARTY",
     )
 
     # Act
@@ -384,8 +644,15 @@ def test_a_deal_crossing_into_abandonment_builds_an_abandon_job_message() -> Non
 def test_a_missing_listing_still_builds_the_message_without_a_place_ref() -> None:
     # Arrange: validation in the abri lambda is the visibility surface for
     # a fired flow that lacks its place_ref; the scraper still sends.
-    old_deal = make_old_deal(project_code=ABRI_PROJECT_CODE, confirmed_survey_date=None)
-    new_deal = make_new_deal(confirmed_survey_date="2026-06-24")
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date=None,
+        third_party_surveyor_identifier=None,
+    )
+    new_deal = make_new_deal(
+        confirmed_survey_date="2026-06-24",
+        third_party_surveyor_identifier="THIRDPARTY",
+    )
 
     # Act
     message = HubspotDealDiffer.check_abri_triggers_and_construct_message(
diff --git a/etl/hubspot/tests/test_hubspot_deal_differ.py b/etl/hubspot/tests/test_hubspot_deal_differ.py
index 9a9e8f05e..e851b206b 100644
--- a/etl/hubspot/tests/test_hubspot_deal_differ.py
+++ b/etl/hubspot/tests/test_hubspot_deal_differ.py
@@ -353,19 +353,21 @@ def test_magicplan_trigger__outcome_surveyed_uppercase__returns_true() -> None:
 # ==============================
 
 
-def test_abri_job_logging__date_first_set_on_abri_deal__returns_true() -> None:
+def test_abri_job_logging__pair_completed_on_abri_deal__returns_true() -> None:
     deal_id = uuid.uuid4()
 
-    # Arrange
+    # Arrange: the surveyor lands to complete the date+surveyor pair.
     old_deal = make_old_deal(
         id=deal_id,
         project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date=None,
+        third_party_surveyor_identifier=None,
     )
     new_deal = make_new_deal(
         deal_id,
         project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date="2026-07-15",
+        third_party_surveyor_identifier="THIRDPARTY",
     )
 
     # Act
@@ -439,11 +441,13 @@ def test_abri_job_logging__project_code_cased_differently__returns_true() -> Non
         id=deal_id,
         project_code=ABRI_PROJECT_CODE.upper(),
         confirmed_survey_date=None,
+        third_party_surveyor_identifier=None,
     )
     new_deal = make_new_deal(
         deal_id,
         project_code=ABRI_PROJECT_CODE.upper(),
         confirmed_survey_date="2026-07-15",
+        third_party_surveyor_identifier="THIRDPARTY",
     )
 
     # Act
@@ -468,11 +472,13 @@ def test_abri_job_logging__real_portal_project_code__returns_true() -> None:
         id=deal_id,
         project_code="[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]",
         confirmed_survey_date=None,
+        third_party_surveyor_identifier=None,
     )
     new_deal = make_new_deal(
         deal_id,
         project_code="[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]",
         confirmed_survey_date="2026-07-15",
+        third_party_surveyor_identifier="THIRDPARTY",
     )
 
     # Act
@@ -530,11 +536,13 @@ def test_abri_job_logging__associated_abri_project_id__returns_true() -> None:
         id=deal_id,
         project_code="Southern Retrofit",
         confirmed_survey_date=None,
+        third_party_surveyor_identifier=None,
     )
     new_deal = make_new_deal(
         deal_id,
         project_code="Southern Retrofit",
         confirmed_survey_date="2026-07-15",
+        third_party_surveyor_identifier="THIRDPARTY",
     )
     new_project = ProjectData(
         project_id=HubspotDealDiffer.ABRI_CONDITION_ASSOCIATED_PROJECT_ID,
@@ -581,6 +589,39 @@ def test_abri_job_logging__non_abri_project_id_with_abri_code__returns_false() -
     assert result is False
 
 
+def test_abri_job_logging__deal_already_has_client_booking_reference__returns_false() -> (
+    None
+):
+    deal_id = uuid.uuid4()
+
+    # Arrange: the pair completes this scrape, but the deal already carries a
+    # Job Number, so the job exists and must not be logged again.
+    old_deal = make_old_deal(
+        id=deal_id,
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date=None,
+        third_party_surveyor_identifier=None,
+        client_booking_reference="AD0226519",
+    )
+    new_deal = make_new_deal(
+        deal_id,
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date="2026-07-15",
+        third_party_surveyor_identifier="THIRDPARTY",
+        client_booking_reference="AD0226519",
+    )
+
+    # Act
+    result = HubspotDealDiffer.check_for_abri_job_logging(
+        new_deal=new_deal,
+        new_project=None,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert result is False
+
+
 # ================================
 # ABRI JOB AMENDMENT TRIGGER TESTS
 # ================================
@@ -589,16 +630,18 @@ def test_abri_job_logging__non_abri_project_id_with_abri_code__returns_false() -
 def test_abri_job_amendment__date_changed_on_abri_deal__returns_true() -> None:
     deal_id = uuid.uuid4()
 
-    # Arrange
+    # Arrange: a complete booking (date + surveyor) whose date moves.
     old_deal = make_old_deal(
         id=deal_id,
         project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date=datetime(2026, 7, 1, tzinfo=timezone.utc),
+        third_party_surveyor_identifier="THIRDPARTY",
     )
     new_deal = make_new_deal(
         deal_id,
         project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date="2026-07-15",
+        third_party_surveyor_identifier="THIRDPARTY",
     )
 
     # Act
@@ -669,18 +712,20 @@ def test_abri_job_amendment__date_and_time_unchanged__returns_false() -> None:
 def test_abri_job_amendment__time_changed_on_abri_deal__returns_true() -> None:
     deal_id = uuid.uuid4()
 
-    # Arrange
+    # Arrange: a complete booking (date + surveyor) whose time moves.
     old_deal = make_old_deal(
         id=deal_id,
         project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date=datetime(2026, 7, 15, tzinfo=timezone.utc),
         confirmed_survey_time="09:30",
+        third_party_surveyor_identifier="THIRDPARTY",
     )
     new_deal = make_new_deal(
         deal_id,
         project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date="2026-07-15",
         confirmed_survey_time="14:00",
+        third_party_surveyor_identifier="THIRDPARTY",
     )
 
     # Act
diff --git a/tests/applications/abri/test_abri_trigger_request.py b/tests/applications/abri/test_abri_trigger_request.py
index 928c6973a..134264458 100644
--- a/tests/applications/abri/test_abri_trigger_request.py
+++ b/tests/applications/abri/test_abri_trigger_request.py
@@ -115,6 +115,41 @@ def test_a_blank_surveyor_identifier_is_treated_as_missing(blank: str) -> None:
         AbriTriggerRequest.model_validate(message)
 
 
+def test_a_message_carries_the_client_booking_reference() -> None:
+    # The Job Number OpenHousing returns on a successful log; the consumer
+    # reads it to tell "job exists" (amend/abandon) from "never logged" (log).
+    message = _full_message()
+    message["client_booking_reference"] = "AD0226519"
+
+    # Act
+    request = AbriTriggerRequest.model_validate(message)
+
+    # Assert
+    assert request.client_booking_reference == "AD0226519"
+
+
+@pytest.mark.parametrize("blank", ["", "   "])
+def test_a_blank_client_booking_reference_is_treated_as_missing(blank: str) -> None:
+    # A blank reference means no Job Number: the consumer must treat it as
+    # "never logged / reference lost", not "job exists".
+    message = _full_message()
+    message["client_booking_reference"] = blank
+
+    # Act
+    request = AbriTriggerRequest.model_validate(message)
+
+    # Assert
+    assert request.client_booking_reference is None
+
+
+def test_a_message_without_a_client_booking_reference_defaults_to_none() -> None:
+    # Act
+    request = AbriTriggerRequest.model_validate(_full_message())
+
+    # Assert
+    assert request.client_booking_reference is None
+
+
 def test_a_message_carries_the_outcome_for_the_abandonment_reason_mapping() -> None:
     # Arrange
     message = _full_message()
diff --git a/tests/applications/abri/test_dispatch.py b/tests/applications/abri/test_dispatch.py
index 7d898b155..19a988ee3 100644
--- a/tests/applications/abri/test_dispatch.py
+++ b/tests/applications/abri/test_dispatch.py
@@ -4,7 +4,11 @@ from typing import Any, Dict, List, Optional
 import pytest
 
 from applications.abri.abri_trigger_request import AbriTriggerRequest
-from applications.abri.dispatch import AbriFlowRejectedError, dispatch_abri_flows
+from applications.abri.dispatch import (
+    AbriFlowRejectedError,
+    DealDivergedFromAbriError,
+    dispatch_abri_flows,
+)
 from domain.abri.models import (
     AbriRequestRejected,
     AppointmentAmended,
@@ -29,7 +33,9 @@ from utilities.aws_lambda.task_handler import NonRetriableTaskError
 DEAL_ID = "9876543210"
 
 
-def _request(flows: List[str]) -> AbriTriggerRequest:
+def _request(
+    flows: List[str], client_booking_reference: Optional[str] = "AD0226519"
+) -> AbriTriggerRequest:
     return AbriTriggerRequest.model_validate(
         {
             "hubspot_deal_id": DEAL_ID,
@@ -41,6 +47,7 @@ def _request(flows: List[str]) -> AbriTriggerRequest:
             "last_submission_date": "2026-07-01",
             "outcome": "no answer",
             "third_party_surveyor_identifier": "THIRDPARTY",
+            "client_booking_reference": client_booking_reference,
         }
     )
 
@@ -218,6 +225,63 @@ def test_the_log_flow_runs_when_the_deal_has_no_job_no(
     assert [call[0] for call in orchestrator.calls] == ["log_job"]
 
 
+# --- divergence: amend/abandon with no Job Number cannot sync to Abri ---
+
+
+def test_an_amend_with_no_client_booking_reference_raises_a_divergence_error(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange: an amend fires on a deal that was never logged (no Job Number).
+    request = _request(["amend_job"], client_booking_reference=None)
+
+    # Act / Assert: non-retriable, and the amend flow never runs.
+    with pytest.raises(DealDivergedFromAbriError) as exc_info:
+        dispatch_abri_flows(request, orchestrator, deal_database)
+    assert isinstance(exc_info.value, NonRetriableTaskError)
+    assert DEAL_ID in str(exc_info.value)
+    assert orchestrator.calls == []
+
+
+def test_an_abandon_with_no_client_booking_reference_raises_a_divergence_error(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange: an abandon fires on a deal with no Job Number — possibly a lost
+    # reference, so a live Abri job must not be silently left un-cancelled.
+    request = _request(["abandon_job"], client_booking_reference=None)
+
+    # Act / Assert
+    with pytest.raises(DealDivergedFromAbriError) as exc_info:
+        dispatch_abri_flows(request, orchestrator, deal_database)
+    assert isinstance(exc_info.value, NonRetriableTaskError)
+    assert orchestrator.calls == []
+
+
+def test_an_amend_with_a_client_booking_reference_proceeds(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange: the deal carries a Job Number, so the job exists and is amended.
+    request = _request(["amend_job"], client_booking_reference="AD0226519")
+
+    # Act
+    dispatch_abri_flows(request, orchestrator, deal_database)
+
+    # Assert
+    assert [call[0] for call in orchestrator.calls] == ["amend_job"]
+
+
+def test_an_abandon_with_a_client_booking_reference_proceeds(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange
+    request = _request(["abandon_job"], client_booking_reference="AD0226519")
+
+    # Act
+    dispatch_abri_flows(request, orchestrator, deal_database)
+
+    # Assert
+    assert [call[0] for call in orchestrator.calls] == ["abandon_job"]
+
+
 # --- rejections raise, so the task fails and the DLQ is the ops inbox ---