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

@@ -18,8 +18,10 @@ from orchestration.abri_orchestrator import (
     ConfirmedSurveyBooking,
     DealDatabaseGateway,
     LogJobOrchestrationResult,
+    LogJobWriteBackError,
     TenantDataSyncResult,
 )
+from utilities.aws_lambda.task_handler import NonRetriableTaskError
 
 
 class AbriFlows(Protocol):
@@ -56,4 +58,80 @@ def dispatch_abri_flows(
     deal_database: DealDatabaseGateway,
 ) -> Dict[str, str]:
     """Run the fired flows in the fixed order; returns a PII-free summary."""
-    raise NotImplementedError
+    summary: Dict[str, str] = {}
+
+    if "amend_job" in request.flows:
+        summary["amend_job"] = _run_amend(request, flows)
+    if "log_job" in request.flows:
+        summary["log_job"] = _run_log(request, flows, deal_database)
+    if "sync_tenant_data" in request.flows:
+        summary["sync_tenant_data"] = _run_tenant_sync(request, flows)
+
+    return summary
+
+
+def _run_amend(request: AbriTriggerRequest, flows: AbriFlows) -> str:
+    if request.confirmed_survey_date is None:
+        raise ValueError("amend_job fired without a confirmed_survey_date")
+    outcome = flows.amend_job(
+        AppointmentChange(
+            deal_id=request.hubspot_deal_id,
+            confirmed_survey_date=request.confirmed_survey_date,
+            confirmed_survey_time=request.confirmed_survey_time,
+        )
+    )
+    if isinstance(outcome, AbriRequestRejected):
+        raise AbriFlowRejectedError(flow="amend_job", rejection=outcome)
+    return (
+        f"appointment amended to {outcome.appointment_date} "
+        f"{outcome.appointment_time} (job {outcome.job_no})"
+    )
+
+
+def _run_log(
+    request: AbriTriggerRequest,
+    flows: AbriFlows,
+    deal_database: DealDatabaseGateway,
+) -> str:
+    existing_job_no = deal_database.job_no_for_deal(request.hubspot_deal_id)
+    if existing_job_no is not None:
+        return f"skipped: job {existing_job_no} already logged"
+
+    if (
+        request.place_ref is None
+        or request.deal_name is None
+        or request.confirmed_survey_date is None
+    ):
+        raise ValueError("log_job fired without place_ref/deal_name/survey date")
+    try:
+        outcome = flows.log_job(
+            ConfirmedSurveyBooking(
+                deal_id=request.hubspot_deal_id,
+                place_ref=PlaceRef(request.place_ref),
+                deal_name=request.deal_name,
+                confirmed_survey_date=request.confirmed_survey_date,
+                confirmed_survey_time=request.confirmed_survey_time,
+            )
+        )
+    except LogJobWriteBackError as error:
+        # The job exists in OpenHousing; a redelivery would log a duplicate.
+        # The orphaned job_no stays visible in the failure record.
+        raise NonRetriableTaskError(str(error)) from error
+    if isinstance(outcome, AbriRequestRejected):
+        raise AbriFlowRejectedError(flow="log_job", rejection=outcome)
+    return f"job {outcome.job_no} logged"
+
+
+def _run_tenant_sync(request: AbriTriggerRequest, flows: AbriFlows) -> str:
+    if request.place_ref is None:
+        raise ValueError("sync_tenant_data fired without a place_ref")
+    outcome = flows.sync_tenant_data(
+        place_ref=PlaceRef(request.place_ref),
+        deal_id=request.hubspot_deal_id,
+    )
+    if isinstance(outcome, AbriRequestRejected):
+        raise AbriFlowRejectedError(flow="sync_tenant_data", rejection=outcome)
+    return (
+        f"{len(outcome.contact_ids)} contacts created "
+        f"({outcome.vulnerable_contact_count} vulnerable)"
+    )