One message dispatches the fired Abri flows in a fixed, retry-safe order 🟥

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

applications/abri/dispatch.py

@@ -0,0 +1,59 @@
+"""Dispatches the flows named in a trigger message through the orchestrator.
+
+The scraper decides which flows fire; this is a dumb dispatcher. Flows run
+in a fixed, retry-safe order — amend, then log, then tenant sync — so that
+redelivery re-runs idempotent flows before non-idempotent ones: amend is
+naturally idempotent, log is guarded by the job-number check, and tenant
+contact creation (not idempotent) runs last so its failure can never cause
+a duplicate job log, and a log failure can never duplicate contacts.
+"""
+
+from typing import Dict, Protocol
+
+from applications.abri.abri_trigger_request import AbriTriggerRequest
+from domain.abri.models import AbriRequestRejected, PlaceRef
+from orchestration.abri_orchestrator import (
+    AmendJobOrchestrationResult,
+    AppointmentChange,
+    ConfirmedSurveyBooking,
+    DealDatabaseGateway,
+    LogJobOrchestrationResult,
+    TenantDataSyncResult,
+)
+
+
+class AbriFlows(Protocol):
+    """The per-flow surface of the AbriOrchestrator that dispatch drives."""
+
+    def amend_job(self, change: AppointmentChange) -> AmendJobOrchestrationResult: ...
+
+    def log_job(self, booking: ConfirmedSurveyBooking) -> LogJobOrchestrationResult: ...
+
+    def sync_tenant_data(
+        self, place_ref: PlaceRef, deal_id: str
+    ) -> TenantDataSyncResult: ...
+
+
+class AbriFlowRejectedError(Exception):
+    """An OpenHousing rejection, surfaced as a failure so the task fails.
+
+    Retrying a permanent rejection is wasteful but harmless — every flow is
+    safe to re-run — and redelivery eventually parks the message in the
+    dead-letter queue, which is the ops inbox.
+    """
+
+    def __init__(self, flow: str, rejection: AbriRequestRejected) -> None:
+        self.flow = flow
+        self.rejection = rejection
+        super().__init__(
+            f"{flow} rejected by OpenHousing: {rejection.code}: {rejection.message}"
+        )
+
+
+def dispatch_abri_flows(
+    request: AbriTriggerRequest,
+    flows: AbriFlows,
+    deal_database: DealDatabaseGateway,
+) -> Dict[str, str]:
+    """Run the fired flows in the fixed order; returns a PII-free summary."""
+    raise NotImplementedError

tests/applications/abri/test_dispatch.py

@@ -0,0 +1,254 @@
+from datetime import date
+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 domain.abri.models import AbriRequestRejected, AppointmentAmended, PlaceRef
+from orchestration.abri_orchestrator import (
+    AmendJobOrchestrationResult,
+    AppointmentChange,
+    ConfirmedSurveyBooking,
+    LogJobOrchestrationResult,
+    LogJobSummary,
+    LogJobWriteBackError,
+    TenantDataSyncResult,
+    TenantDataSyncSummary,
+)
+from utilities.aws_lambda.task_handler import NonRetriableTaskError
+
+DEAL_ID = "9876543210"
+
+
+def _request(flows: List[str]) -> AbriTriggerRequest:
+    return AbriTriggerRequest.model_validate(
+        {
+            "hubspot_deal_id": DEAL_ID,
+            "flows": flows,
+            "place_ref": "1007165",
+            "deal_name": "49 Admers Crescent",
+            "confirmed_survey_date": "2026-06-24",
+            "confirmed_survey_time": "14:30",
+        }
+    )
+
+
+class FakeOrchestrator:
+    """Records flow invocations in order; outcomes are programmable."""
+
+    def __init__(self) -> None:
+        self.calls: List[Any] = []
+        self.amend_outcome: AmendJobOrchestrationResult = AppointmentAmended(
+            job_no="AC0439951",
+            appointment_date="24/06/2026",
+            appointment_time="PM",
+        )
+        self.log_outcome: LogJobOrchestrationResult = LogJobSummary(
+            deal_id=DEAL_ID, job_no="AD0226519"
+        )
+        self.log_error: Optional[Exception] = None
+        self.sync_outcome: TenantDataSyncResult = TenantDataSyncSummary(
+            tenancy_reference="TEN0001",
+            contact_ids=("101", "102"),
+            vulnerable_contact_count=1,
+        )
+
+    def amend_job(self, change: AppointmentChange) -> AmendJobOrchestrationResult:
+        self.calls.append(("amend_job", change))
+        return self.amend_outcome
+
+    def log_job(self, booking: ConfirmedSurveyBooking) -> LogJobOrchestrationResult:
+        self.calls.append(("log_job", booking))
+        if self.log_error is not None:
+            raise self.log_error
+        return self.log_outcome
+
+    def sync_tenant_data(
+        self, place_ref: PlaceRef, deal_id: str
+    ) -> TenantDataSyncResult:
+        self.calls.append(("sync_tenant_data", place_ref, deal_id))
+        return self.sync_outcome
+
+
+class FakeDealDatabase:
+    """In-memory stand-in for the deal-database gateway."""
+
+    def __init__(self) -> None:
+        self.job_nos: Dict[str, str] = {}
+
+    def record_job_no(self, deal_id: str, job_no: str) -> None:
+        self.job_nos[deal_id] = job_no
+
+    def job_no_for_deal(self, deal_id: str) -> Optional[str]:
+        return self.job_nos.get(deal_id)
+
+
+@pytest.fixture()
+def orchestrator() -> FakeOrchestrator:
+    return FakeOrchestrator()
+
+
+@pytest.fixture()
+def deal_database() -> FakeDealDatabase:
+    return FakeDealDatabase()
+
+
+# --- fixed dispatch order: amend -> log -> tenant sync ---
+
+
+def test_fired_flows_run_in_the_fixed_order_regardless_of_message_order(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange
+    request = _request(["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",
+    ]
+
+
+def test_flows_the_message_does_not_name_do_not_run(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange
+    request = _request(["sync_tenant_data"])
+
+    # Act
+    dispatch_abri_flows(request, orchestrator, deal_database)
+
+    # Assert
+    assert orchestrator.calls == [
+        ("sync_tenant_data", PlaceRef("1007165"), DEAL_ID)
+    ]
+
+
+# --- the flows receive the deal fields from the message ---
+
+
+def test_flows_receive_the_deal_fields_carried_by_the_message(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange
+    request = _request(["amend_job", "log_job"])
+
+    # Act
+    dispatch_abri_flows(request, orchestrator, deal_database)
+
+    # Assert
+    amend_call, log_call = orchestrator.calls
+    assert amend_call == (
+        "amend_job",
+        AppointmentChange(
+            deal_id=DEAL_ID,
+            confirmed_survey_date=date(2026, 6, 24),
+            confirmed_survey_time="14:30",
+        ),
+    )
+    assert log_call == (
+        "log_job",
+        ConfirmedSurveyBooking(
+            deal_id=DEAL_ID,
+            place_ref=PlaceRef("1007165"),
+            deal_name="49 Admers Crescent",
+            confirmed_survey_date=date(2026, 6, 24),
+            confirmed_survey_time="14:30",
+        ),
+    )
+
+
+# --- idempotency guard: a job is never logged twice for the same deal ---
+
+
+def test_the_log_flow_is_skipped_when_the_deal_already_has_a_job_no(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange
+    request = _request(["log_job", "sync_tenant_data"])
+    deal_database.job_nos[DEAL_ID] = "AD0226519"
+
+    # Act
+    dispatch_abri_flows(request, orchestrator, deal_database)
+
+    # Assert
+    assert [call[0] for call in orchestrator.calls] == ["sync_tenant_data"]
+
+
+def test_the_log_flow_runs_when_the_deal_has_no_job_no(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange
+    request = _request(["log_job"])
+
+    # Act
+    dispatch_abri_flows(request, orchestrator, deal_database)
+
+    # Assert
+    assert [call[0] for call in orchestrator.calls] == ["log_job"]
+
+
+# --- rejections raise, so the task fails and the DLQ is the ops inbox ---
+
+
+def test_an_openhousing_rejection_raises_and_stops_the_dispatch(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange
+    request = _request(["amend_job", "sync_tenant_data"])
+    orchestrator.amend_outcome = AbriRequestRejected(
+        code="RelayError", message="Job_no 1019905 is invalid. Job not found."
+    )
+
+    # Act / Assert
+    with pytest.raises(AbriFlowRejectedError) as exc_info:
+        dispatch_abri_flows(request, orchestrator, deal_database)
+    assert "RelayError" in str(exc_info.value)
+    assert "Job_no 1019905 is invalid" in str(exc_info.value)
+    assert [call[0] for call in orchestrator.calls] == ["amend_job"]
+
+
+# --- write-back failure: failed task, but the message is never requeued ---
+
+
+def test_a_log_write_back_failure_raises_a_non_retriable_error(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange
+    request = _request(["log_job", "sync_tenant_data"])
+    orchestrator.log_error = LogJobWriteBackError(
+        message="HubSpot update failed", deal_id=DEAL_ID, job_no="AD0226519"
+    )
+
+    # Act / Assert: the orphaned job_no stays visible in the failure record,
+    # and the tenant sync never runs on this delivery.
+    with pytest.raises(NonRetriableTaskError) as exc_info:
+        dispatch_abri_flows(request, orchestrator, deal_database)
+    assert "AD0226519" in str(exc_info.value)
+    assert [call[0] for call in orchestrator.calls] == ["log_job"]
+
+
+# --- the task output is a PII-free summary of what ran ---
+
+
+def test_dispatch_returns_a_pii_free_summary_of_what_ran(
+    orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase
+) -> None:
+    # Arrange
+    request = _request(["amend_job", "log_job", "sync_tenant_data"])
+    deal_database.job_nos[DEAL_ID] = "AC0439951"
+
+    # Act
+    summary = dispatch_abri_flows(request, orchestrator, deal_database)
+
+    # Assert
+    assert summary == {
+        "amend_job": "appointment amended to 24/06/2026 PM (job AC0439951)",
+        "log_job": "skipped: job AC0439951 already logged",
+        "sync_tenant_data": "2 contacts created (1 vulnerable)",
+    }