diff --git a/orchestration/tenant_data_sync_orchestrator.py b/orchestration/tenant_data_sync_orchestrator.py
index 6edff7156..aa8005bd8 100644
--- a/orchestration/tenant_data_sync_orchestrator.py
+++ b/orchestration/tenant_data_sync_orchestrator.py
@@ -49,6 +49,41 @@ class TenantDataSyncSummary:
 TenantDataSyncResult = Union[TenantDataSyncSummary, AbriRequestRejected]
 
 
+class TenantDataSyncError(Exception):
+    """A sync that stopped at the first failed HubSpot write.
+
+    Carries a PII-free progress report so the operator knows the deal's
+    half-state before cleaning up and re-running; contacts created but not
+    associated are orphaned in HubSpot and need manual cleanup.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        deal_id: str,
+        tenant_index: int,
+        contact_ids_created: Tuple[str, ...],
+        contact_ids_associated: Tuple[str, ...],
+    ) -> None:
+        self.deal_id = deal_id
+        self.tenant_index = tenant_index
+        self.contact_ids_created = contact_ids_created
+        self.contact_ids_associated = contact_ids_associated
+        super().__init__(
+            f"{message} (deal: {deal_id}, tenant index: {tenant_index}, "
+            f"contacts created: {list(contact_ids_created)}, "
+            f"associated: {list(contact_ids_associated)})"
+        )
+
+    @property
+    def orphaned_contact_ids(self) -> Tuple[str, ...]:
+        return tuple(
+            contact_id
+            for contact_id in self.contact_ids_created
+            if contact_id not in self.contact_ids_associated
+        )
+
+
 class TenantDataSyncOrchestrator:
     """Syncs an Abri tenancy's signatories to a HubSpot deal.
 
diff --git a/tests/orchestration/test_tenant_data_sync_orchestrator.py b/tests/orchestration/test_tenant_data_sync_orchestrator.py
index 40b4c34da..92dd4a885 100644
--- a/tests/orchestration/test_tenant_data_sync_orchestrator.py
+++ b/tests/orchestration/test_tenant_data_sync_orchestrator.py
@@ -16,6 +16,7 @@ from infrastructure.abri.abri_client import AbriClient
 from infrastructure.abri.config import AbriConfig
 from infrastructure.abri.errors import AbriResponseParseError, AbriTransportError
 from orchestration.tenant_data_sync_orchestrator import (
+    TenantDataSyncError,
     TenantDataSyncOrchestrator,
     TenantDataSyncSummary,
 )
@@ -467,3 +468,34 @@ def test_run_raises_an_error_scrubbed_of_the_hubspot_response_body(
     assert "Amanjeet" not in rendered_traceback
     assert "Okello" not in rendered_traceback
     assert "09853460810" not in rendered_traceback
+
+
+def test_run_fails_fast_when_association_fails_and_reports_the_orphaned_contact(
+    orchestrator: TenantDataSyncOrchestrator,
+    mock_session: MagicMock,
+    sdk_client: MagicMock,
+) -> None:
+    # Arrange
+    mock_session.post.return_value.content = _load_fixture(
+        "abri_relay_getscstenantdata_multitenant_response.xml"
+    )
+    sdk_client.crm.associations.v4.basic_api.create.side_effect = (
+        _pii_echoing_api_error()
+    )
+
+    # Act
+    with pytest.raises(TenantDataSyncError) as exc_info:
+        orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
+
+    # Assert: the flow stopped at the first failed write, with progress reported
+    error = exc_info.value
+    assert error.deal_id == DEAL_ID
+    assert error.tenant_index == 0
+    assert error.contact_ids_created == ("60123",)
+    assert error.contact_ids_associated == ()
+    assert error.orphaned_contact_ids == ("60123",)
+    assert sdk_client.crm.contacts.basic_api.create.call_count == 1  # fail fast
+    rendered_traceback = "".join(traceback.format_exception(error))
+    assert "corr-abc-123" in str(error)  # scrubbed diagnostics carried through
+    assert "Solecka" not in rendered_traceback
+    assert "09261195827" not in rendered_traceback