diff --git a/tests/applications/abri/test_dispatch.py b/tests/applications/abri/test_dispatch.py
index b41aa4d4b..9eac4fed7 100644
--- a/tests/applications/abri/test_dispatch.py
+++ b/tests/applications/abri/test_dispatch.py
@@ -5,11 +5,19 @@ import pytest
 
 from applications.abri.abri_trigger_request import AbriTriggerRequest
 from applications.abri.dispatch import AbriFlowRejectedError, dispatch_abri_flows
-from domain.abri.models import AbriRequestRejected, AppointmentAmended, PlaceRef
+from domain.abri.models import (
+    AbriRequestRejected,
+    AppointmentAmended,
+    JobAbandoned,
+    PlaceRef,
+)
 from orchestration.abri_orchestrator import (
+    AbandonJobOrchestrationResult,
     AmendJobOrchestrationResult,
     AppointmentChange,
     ConfirmedSurveyBooking,
+    DealAbandonment,
+    JobNoNotYetRecordedError,
     LogJobOrchestrationResult,
     LogJobSummary,
     LogJobWriteBackError,
@@ -30,6 +38,7 @@ def _request(flows: List[str]) -> AbriTriggerRequest:
             "deal_name": "49 Admers Crescent",
             "confirmed_survey_date": "2026-06-24",
             "confirmed_survey_time": "14:30",
+            "outcome": "no answer",
         }
     )
 
@@ -53,6 +62,10 @@ class FakeOrchestrator:
             contact_ids=("101", "102"),
             vulnerable_contact_count=1,
         )
+        self.abandon_outcome: AbandonJobOrchestrationResult = JobAbandoned(
+            job_no="AC0439951"
+        )
+        self.abandon_error: Optional[Exception] = None
 
     def amend_job(self, change: AppointmentChange) -> AmendJobOrchestrationResult:
         self.calls.append(("amend_job", change))
@@ -70,6 +83,14 @@ class FakeOrchestrator:
         self.calls.append(("sync_tenant_data", place_ref, deal_id))
         return self.sync_outcome
 
+    def abandon_job(
+        self, abandonment: DealAbandonment
+    ) -> AbandonJobOrchestrationResult:
+        self.calls.append(("abandon_job", abandonment))
+        if self.abandon_error is not None:
+            raise self.abandon_error
+        return self.abandon_outcome
+
 
 class FakeDealDatabase:
     """In-memory stand-in for the deal-database gateway."""
@@ -233,6 +254,70 @@ def test_a_log_write_back_failure_raises_a_non_retriable_error(
     assert [call[0] for call in orchestrator.calls] == ["log_job"]
 
 
+# --- abandon runs last, after amend, log and tenant sync ---
+
+
+def test_abandon_runs_last_in_the_fixed_dispatch_order(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange
+    request = _request(["abandon_job", "sync_tenant_data", "log_job", "amend_job"])
+
+    # Act
+    dispatch_abri_flows(request, orchestrator, deal_database)
+
+    # Assert
+    assert [call[0] for call in orchestrator.calls] == [
+        "amend_job",
+        "log_job",
+        "sync_tenant_data",
+        "abandon_job",
+    ]
+
+
+def test_the_abandon_flow_receives_the_deal_id_and_outcome_from_the_message(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange
+    request = _request(["abandon_job"])
+
+    # Act
+    dispatch_abri_flows(request, orchestrator, deal_database)
+
+    # Assert
+    assert orchestrator.calls == [
+        ("abandon_job", DealAbandonment(deal_id=DEAL_ID, outcome="no answer"))
+    ]
+
+
+def test_an_abandon_rejection_raises_so_the_task_fails(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange
+    request = _request(["abandon_job"])
+    orchestrator.abandon_outcome = AbriRequestRejected(
+        code="RelayError", message="Job_no AC0439951 is invalid. Job not found."
+    )
+
+    # Act / Assert
+    with pytest.raises(AbriFlowRejectedError) as exc_info:
+        dispatch_abri_flows(request, orchestrator, deal_database)
+    assert "abandon_job" in str(exc_info.value)
+    assert "RelayError" in str(exc_info.value)
+
+
+def test_an_abandon_before_the_job_no_is_recorded_propagates_the_retriable_error(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange
+    request = _request(["abandon_job"])
+    orchestrator.abandon_error = JobNoNotYetRecordedError(deal_id=DEAL_ID)
+
+    # Act / Assert
+    with pytest.raises(JobNoNotYetRecordedError):
+        dispatch_abri_flows(request, orchestrator, deal_database)
+
+
 # --- the task output is a PII-free summary of what ran ---
 
 
@@ -252,3 +337,16 @@ def test_dispatch_returns_a_pii_free_summary_of_what_ran(
         "log_job": "skipped: job AC0439951 already logged",
         "sync_tenant_data": "2 contacts created (1 vulnerable)",
     }
+
+
+def test_the_summary_records_an_abandoned_job(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange
+    request = _request(["abandon_job"])
+
+    # Act
+    summary = dispatch_abri_flows(request, orchestrator, deal_database)
+
+    # Assert
+    assert summary == {"abandon_job": "job AC0439951 abandoned"}