Merge pull request #1482 from Hestia-Homes/feature/abri-api-integration

Abri API Integration: lambda handler and plumb everything together

.github/workflows/_deploy_lambda.yml

@@ -100,6 +100,15 @@ on:
 
       TF_VAR_open_epc_api_token:
         required: false
+
+      TF_VAR_abri_relay_url:
+        required: false
+      TF_VAR_abri_relay_username:
+        required: false
+      TF_VAR_abri_relay_password:
+        required: false
+      TF_VAR_abri_relay_default_resource:
+        required: false
 jobs:
   deploy:
     runs-on: ubuntu-latest
@@ -175,6 +184,10 @@ jobs:
           TF_VAR_magicplan_api_key: ${{ secrets.TF_VAR_magicplan_api_key }}
           TF_VAR_openai_api_key: ${{ secrets.TF_VAR_openai_api_key }}
           TF_VAR_open_epc_api_token: ${{ secrets.TF_VAR_open_epc_api_token }}
+          TF_VAR_abri_relay_url: ${{ secrets.TF_VAR_abri_relay_url }}
+          TF_VAR_abri_relay_username: ${{ secrets.TF_VAR_abri_relay_username }}
+          TF_VAR_abri_relay_password: ${{ secrets.TF_VAR_abri_relay_password }}
+          TF_VAR_abri_relay_default_resource: ${{ secrets.TF_VAR_abri_relay_default_resource }}
         run: |
           ECR_REPO_URL_VAR=""
           if [[ -n "${{ inputs.ecr_repo }}" ]]; then
@@ -228,6 +241,10 @@ jobs:
           TF_VAR_magicplan_api_key: ${{ secrets.TF_VAR_magicplan_api_key }}
           TF_VAR_openai_api_key: ${{ secrets.TF_VAR_openai_api_key }}
           TF_VAR_open_epc_api_token: ${{ secrets.TF_VAR_open_epc_api_token }}
+          TF_VAR_abri_relay_url: ${{ secrets.TF_VAR_abri_relay_url }}
+          TF_VAR_abri_relay_username: ${{ secrets.TF_VAR_abri_relay_username }}
+          TF_VAR_abri_relay_password: ${{ secrets.TF_VAR_abri_relay_password }}
+          TF_VAR_abri_relay_default_resource: ${{ secrets.TF_VAR_abri_relay_default_resource }}
         run: |
           EXTRA_VARS=""
           if [[ -n "${{ inputs.ecr_repo }}" ]]; then

.github/workflows/deploy_terraform.yml

@@ -700,6 +700,48 @@ jobs:
       TF_VAR_magicplan_customer_id: ${{ secrets.MAGICPLAN_CUSTOMER_ID }}
       TF_VAR_magicplan_api_key: ${{ secrets.MAGICPLAN_API_KEY }}
 
+  # ============================================================
+  # Build Abri Lambda image
+  # ============================================================
+  abri_image:
+    needs: [determine_stage, shared_terraform]
+    uses: ./.github/workflows/_build_image.yml
+    with:
+      ecr_repo: abri-${{ needs.determine_stage.outputs.stage }}
+      dockerfile_path: applications/abri/handler/Dockerfile
+      build_context: .
+    secrets:
+      AWS_ACCESS_KEY_ID: ${{ secrets.DEV_AWS_ACCESS_KEY_ID }}
+      AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }}
+      AWS_REGION: ${{ secrets.DEV_AWS_REGION }}
+
+  # ============================================================
+  # Deploy Abri Lambda
+  # ============================================================
+  abri_api_lambda:
+    needs: [abri_image, determine_stage]
+    uses: ./.github/workflows/_deploy_lambda.yml
+    with:
+      lambda_name: abri
+      lambda_path: deployment/terraform/lambda/abri
+      stage: ${{ needs.determine_stage.outputs.stage }}
+      ecr_repo: abri-${{ needs.determine_stage.outputs.stage }}
+      image_digest: ${{ needs.abri_image.outputs.image_digest }}
+      terraform_apply: ${{ needs.determine_stage.outputs.terraform_apply }}
+    secrets:
+      AWS_ACCESS_KEY_ID: ${{ secrets.DEV_AWS_ACCESS_KEY_ID }}
+      AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }}
+      AWS_REGION: ${{ secrets.DEV_AWS_REGION }}
+      TF_VAR_db_host: ${{ secrets.DEV_DB_HOST }}
+      TF_VAR_db_name: ${{ secrets.DEV_DB_NAME }}
+      TF_VAR_db_port: ${{ secrets.DEV_DB_PORT }}
+      TF_VAR_hubspot_api_key: ${{ secrets.HUBSPOT_API_KEY }}
+      # Awaiting API Key from Abri
+      TF_VAR_abri_relay_url: ${{ secrets.ABRI_RELAY_URL }}
+      TF_VAR_abri_relay_username: ${{ secrets.ABRI_RELAY_USERNAME }}
+      TF_VAR_abri_relay_password: ${{ secrets.ABRI_RELAY_PASSWORD }}
+      TF_VAR_abri_relay_default_resource: ${{ secrets.ABRI_RELAY_DEFAULT_RESOURCE }}
+
   # ============================================================
   # Build Audit Generator image
   # ============================================================
@@ -778,7 +820,7 @@ jobs:
   # Deploy Hubspot ETL Lambda
   # ============================================================
   hubspot_etl_lambda:
-    needs: [hubspot_etl_image, determine_stage, pashub_to_ara_lambda, magic_plan_lambda]
+    needs: [hubspot_etl_image, determine_stage, pashub_to_ara_lambda, magic_plan_lambda, abri_lambda]
     uses: ./.github/workflows/_deploy_lambda.yml
     with:
       lambda_name: hubspot-etl-to-ara

applications/abri/abri_trigger_request.py

@@ -0,0 +1,44 @@
+"""The message contract between the HubSpot scraper and the abri lambda.
+
+One message per scrape, naming every Abri flow that fired plus the deal
+fields those flows need. A message missing a field its flow needs fails
+validation — the resulting failed task / DLQ entry is the agreed
+visibility surface.
+"""
+
+from datetime import date
+from typing import Dict, List, Literal, Optional, Tuple
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+AbriFlow = Literal["amend_job", "log_job", "sync_tenant_data"]
+
+# The deal fields each flow reads; hubspot_deal_id is always required.
+_FIELDS_BY_FLOW: Dict[AbriFlow, Tuple[str, ...]] = {
+    "amend_job": ("confirmed_survey_date",),
+    "log_job": ("place_ref", "deal_name", "confirmed_survey_date"),
+    "sync_tenant_data": ("place_ref",),
+}
+
+
+class AbriTriggerRequest(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    hubspot_deal_id: str
+    flows: List[AbriFlow] = Field(min_length=1)
+    place_ref: Optional[str] = None
+    deal_name: Optional[str] = None
+    confirmed_survey_date: Optional[date] = None
+    confirmed_survey_time: Optional[str] = None
+
+    @model_validator(mode="after")
+    def _each_fired_flow_has_its_fields(self) -> "AbriTriggerRequest":
+        missing = [
+            f"flow {flow} requires {field}"
+            for flow in self.flows
+            for field in _FIELDS_BY_FLOW[flow]
+            if getattr(self, field) is None
+        ]
+        if missing:
+            raise ValueError("; ".join(missing))
+        return self

applications/abri/dispatch.py

@@ -0,0 +1,137 @@
+"""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,
+    LogJobWriteBackError,
+    TenantDataSyncResult,
+)
+from utilities.aws_lambda.task_handler import NonRetriableTaskError
+
+
+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."""
+    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)"
+    )

applications/abri/handler.py

@@ -0,0 +1,59 @@
+"""Lambda entry point for the Abri OpenHousing flows.
+
+Consumes the abri SQS queue (one trigger message per scrape, sent by the
+HubSpot scraper) and dispatches the fired flows through the AbriOrchestrator.
+Env-to-clients glue only; behaviour lives behind the dispatch seam.
+"""
+
+import os
+from typing import Any, Dict
+
+from hubspot.client import Client  # type: ignore[reportMissingTypeStubs]
+from sqlalchemy.pool import NullPool
+from sqlmodel import Session
+
+from applications.abri.abri_trigger_request import AbriTriggerRequest
+from applications.abri.dispatch import dispatch_abri_flows
+from domain.tasks.tasks import Source
+from infrastructure.abri.abri_client import AbriClient
+from infrastructure.abri.config import AbriConfig
+from infrastructure.hubspot.deal_contacts_client import HubspotDealContactsClient
+from infrastructure.hubspot.deal_properties_client import HubspotDealPropertiesClient
+from infrastructure.postgres.config import PostgresConfig
+from infrastructure.postgres.engine import make_engine
+from orchestration.abri_orchestrator import AbriOrchestrator
+from repositories.hubspot_deals.deal_database_postgres_gateway import (
+    DealDatabasePostgresGateway,
+)
+from utilities.aws_lambda.task_handler import task_handler
+from utilities.logger import setup_logger
+
+logger = setup_logger()
+
+
+@task_handler(task_source="abri_api", source=Source.HUBSPOT_DEAL)
+def handler(body: dict[str, Any], context: Any) -> Dict[str, str]:
+    request = AbriTriggerRequest.model_validate(body)
+
+    abri_client = AbriClient(config=AbriConfig.from_env(os.environ))
+    sdk_client: Client = Client.create(  # type: ignore[reportUnknownMemberType]
+        access_token=os.environ["HUBSPOT_API_KEY"]
+    )
+
+    # NullPool: a fresh engine per invocation, so pooling would only
+    # accumulate idle connections across warm Lambda containers.
+    engine = make_engine(PostgresConfig.from_env(dict(os.environ)), poolclass=NullPool)
+    with Session(engine) as session:
+        deal_database = DealDatabasePostgresGateway(session=session)
+        orchestrator = AbriOrchestrator(
+            abri_client=abri_client,
+            deal_contacts=HubspotDealContactsClient(sdk_client=sdk_client),
+            deal_properties=HubspotDealPropertiesClient(sdk_client=sdk_client),
+            deal_database=deal_database,
+        )
+        summary = dispatch_abri_flows(request, orchestrator, deal_database)
+
+    logger.info(
+        "Abri flows completed for deal %s: %s", request.hubspot_deal_id, summary
+    )
+    return summary

applications/abri/handler/Dockerfile

@@ -0,0 +1,18 @@
+FROM public.ecr.aws/lambda/python:3.11
+
+WORKDIR /var/task
+
+COPY applications/abri/handler/requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY utilities/ utilities/
+COPY utils/ utils/
+COPY backend/ backend/
+COPY applications/ applications/
+COPY domain/ domain/
+COPY datatypes/ datatypes/
+COPY orchestration/ orchestration/
+COPY repositories/ repositories/
+COPY infrastructure/ infrastructure/
+
+CMD ["applications.abri.handler.handler"]

applications/abri/handler/requirements.txt

@@ -0,0 +1,8 @@
+awslambdaric
+requests
+sqlalchemy==2.0.36
+sqlmodel
+psycopg2-binary==2.9.10
+pydantic-settings==2.6.0
+boto3==1.35.44
+hubspot-api-client

backend/app/config.py

@@ -40,6 +40,7 @@ class Settings(BaseSettings):
     CATEGORISATION_SQS_URL: str = "changeme"
     PASHUB_TO_ARA_SQS_URL: str = "changeme"
     MAGICPLAN_SQS_URL: str = "changeme"
+    ABRI_SQS_URL: str = "changeme"
     POSTCODE_SPLITTER_SQS_URL: str = "changeme"
     COMBINER_SQS_URL: str = "changeme"
     LANDLORD_OVERRIDES_SQS_URL: str = "changeme"

backend/app/db/models/hubspot_deal_data.py

@@ -71,8 +71,10 @@ class HubspotDealData(SQLModel, table=True):
     confirmed_survey_time: Optional[str] = Field(default=None)
     surveyed_date: Optional[datetime] = Field(default=None)
     design_type: Optional[str] = Field(default=None)
-    # OpenHousing job number; HubSpot property ID is client_booking_reference.
-    job_no: Optional[str] = Field(default=None)
+    # The OpenHousing job number, named after the HubSpot property that
+    # carries it (matching the column the frontend repo's drizzle migration
+    # created). Only the domain calls it job_no.
+    client_booking_reference: Optional[str] = Field(default=None)
 
     survey_type: Optional[str] = Field(default=None)
     measures_for_pibi_ordered: Optional[str] = Field(default=None)

deployment/terraform/lambda/abri_api/main.tf

@@ -0,0 +1,40 @@
+data "aws_secretsmanager_secret_version" "db_credentials" {
+  secret_id = "${var.stage}/assessment_model/db_credentials"
+}
+
+locals {
+  db_credentials = jsondecode(data.aws_secretsmanager_secret_version.db_credentials.secret_string)
+}
+
+module "lambda" {
+  source = "../../modules/lambda_with_sqs"
+
+  name  = "abri_api"
+  stage = var.stage
+
+  image_uri = local.image_uri
+
+  maximum_concurrency            = var.maximum_concurrency
+  reserved_concurrent_executions = var.reserved_concurrent_executions
+  batch_size                     = var.batch_size
+
+  environment = {
+    STAGE     = var.stage
+    LOG_LEVEL = "info"
+
+    # Abri relay credentials are per-stage: the dev/staging stage points at
+    # Abri's UAT relay endpoint; production values are pending from Abri.
+    ABRI_RELAY_URL              = var.abri_relay_url
+    ABRI_RELAY_USERNAME         = var.abri_relay_username
+    ABRI_RELAY_PASSWORD         = var.abri_relay_password
+    ABRI_RELAY_DEFAULT_RESOURCE = var.abri_relay_default_resource
+
+    HUBSPOT_API_KEY = var.hubspot_api_key
+
+    POSTGRES_USERNAME = local.db_credentials.db_assessment_model_username
+    POSTGRES_PASSWORD = local.db_credentials.db_assessment_model_password
+    POSTGRES_HOST     = var.db_host
+    POSTGRES_DATABASE = var.db_name
+    POSTGRES_PORT     = var.db_port
+  }
+}

deployment/terraform/lambda/abri_api/outputs.tf

@@ -0,0 +1,9 @@
+output "abri_api_queue_url" {
+  value       = module.lambda.queue_url
+  description = "URL of the Abri SQS queue"
+}
+
+output "abri_api_queue_arn" {
+  value       = module.lambda.queue_arn
+  description = "ARN of the Abri SQS queue"
+}

deployment/terraform/lambda/abri_api/provider.tf

@@ -0,0 +1,16 @@
+terraform {
+  required_providers {
+    aws = {
+      source  = "hashicorp/aws"
+      version = ">= 5.0"
+    }
+  }
+
+  backend "s3" {
+    bucket = "abri-terraform-state"
+    key    = "terraform.tfstate"
+    region = "eu-west-2"
+  }
+
+  required_version = ">= 1.2.0"
+}

deployment/terraform/lambda/abri_api/variables.tf

@@ -0,0 +1,84 @@
+variable "lambda_name" {
+  type        = string
+  description = "Logical name of the lambda"
+  default     = "abri_api"
+}
+
+variable "stage" {
+  description = "Deployment stage (e.g. dev, prod)"
+  type        = string
+}
+
+variable "ecr_repo_url" {
+  type        = string
+  description = "ECR repository URL (no tag, no digest)"
+}
+
+variable "image_digest" {
+  type        = string
+  description = "Image digest (sha256:...)"
+}
+
+variable "maximum_concurrency" {
+  type        = number
+  default     = null
+  description = "Maximum number of concurrent Lambda invocations from SQS (2-1000). null = no limit."
+}
+
+variable "reserved_concurrent_executions" {
+  type    = number
+  default = 1
+}
+
+variable "batch_size" {
+  type    = number
+  default = 1
+}
+
+locals {
+  image_uri = "${var.ecr_repo_url}@${var.image_digest}"
+}
+
+output "resolved_image_uri" {
+  value = local.image_uri
+}
+
+variable "abri_relay_url" {
+  type      = string
+  sensitive = true
+}
+
+variable "abri_relay_username" {
+  type      = string
+  sensitive = true
+}
+
+variable "abri_relay_password" {
+  type      = string
+  sensitive = true
+}
+
+variable "abri_relay_default_resource" {
+  type      = string
+  sensitive = true
+}
+
+variable "hubspot_api_key" {
+  type      = string
+  sensitive = true
+}
+
+variable "db_host" {
+  type      = string
+  sensitive = true
+}
+
+variable "db_name" {
+  type      = string
+  sensitive = true
+}
+
+variable "db_port" {
+  type      = string
+  sensitive = true
+}

deployment/terraform/lambda/hubspot_deal_etl/main.tf

@@ -25,6 +25,15 @@ data "terraform_remote_state" "magic_plan" {
   }
 }
 
+data "terraform_remote_state" "abri_api" {
+  backend = "s3"
+  config = {
+    bucket = "abri-api-terraform-state"
+    key    = "env:/${var.stage}/terraform.tfstate"
+    region = "eu-west-2"
+  }
+}
+
 data "aws_secretsmanager_secret_version" "db_credentials" {
   secret_id = "${var.stage}/assessment_model/db_credentials"
 }
@@ -59,6 +68,7 @@ module "hubspot_deal_etl" {
 
     PASHUB_TO_ARA_SQS_URL = data.terraform_remote_state.pashub_to_ara.outputs.pashub_to_ara_queue_url
     MAGICPLAN_SQS_URL     = data.terraform_remote_state.magic_plan.outputs.magic_plan_queue_url
+    ABRI_SQS_URL          = data.terraform_remote_state.abri.outputs.abri_api_queue_url
   }
 }
 
@@ -100,4 +110,18 @@ module "hubspot_deal_etl_magicplan_sqs_policy" {
 resource "aws_iam_role_policy_attachment" "hubspot_deal_etl_magicplan_sqs_send" {
   role       = module.hubspot_deal_etl.role_name
   policy_arn = module.hubspot_deal_etl_magicplan_sqs_policy.policy_arn
+}
+
+module "hubspot_deal_etl_abri_sqs_policy" {
+  source = "../../modules/general_iam_policy"
+
+  policy_name        = "hubspot-deal-etl-abri-sqs-send-${var.stage}"
+  policy_description = "Allow HubSpot ETL Lambda to send messages to the Abri queue"
+  actions            = ["sqs:SendMessage"]
+  resources          = [data.terraform_remote_state.abri.outputs.abri_api_queue_arn]
+}
+
+resource "aws_iam_role_policy_attachment" "hubspot_deal_etl_abri_sqs_send" {
+  role       = module.hubspot_deal_etl.role_name
+  policy_arn = module.hubspot_deal_etl_abri_sqs_policy.policy_arn
 }

deployment/terraform/shared/main.tf

@@ -894,3 +894,17 @@ output "modelling_e2e_ecr_url" {
   value = module.modelling_e2e_registry.repository_url
 }
 
+
+################################################
+# Abri OpenHousing – Lambda
+################################################
+module "abri_api_state_bucket" {
+  source      = "../modules/tf_state_bucket"
+  bucket_name = "abri-terraform-state"
+}
+
+module "abri_api_registry" {
+  source = "../modules/container_registry"
+  name   = "abri_api"
+  stage  = var.stage
+}

etl/hubspot/hubspotDataTodB.py

@@ -274,7 +274,7 @@ class HubspotDataToDb:
                 deal_data.get("confirmed_survey_date")
             ),
             "confirmed_survey_time": deal_data.get("confirmed_survey_time"),
-            "job_no": deal_data.get("client_booking_reference"),
+            "client_booking_reference": deal_data.get("client_booking_reference"),
             "surveyed_date": parse_hs_date(deal_data.get("surveyed_date")),
             "design_type": deal_data.get("design_type"),
             "survey_type": deal_data.get("survey_type"),
@@ -383,7 +383,7 @@ class HubspotDataToDb:
             surveyor=deal_data.get("surveyor"),
             confirmed_survey_date=parse_hs_date(deal_data.get("confirmed_survey_date")),
             confirmed_survey_time=deal_data.get("confirmed_survey_time"),
-            job_no=deal_data.get("client_booking_reference"),
+            client_booking_reference=deal_data.get("client_booking_reference"),
             surveyed_date=parse_hs_date(deal_data.get("surveyed_date")),
             design_type=deal_data.get("design_type"),
             survey_type=deal_data.get("survey_type"),

etl/hubspot/hubspot_deal_differ.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional
 
 from backend.app.db.models.hubspot_deal_data import HubspotDealData
 from etl.hubspot.project_data import ProjectData
@@ -21,8 +21,14 @@ class HubspotDealDiffer:
         "tenant refusal",
         "not viable",
     ]
-    ABRI_CONDITION_ASSOCIATED_PROJECT_ID = "123456"
-    ABRI_CONDITION_PROJECT_CODE = "Abri Condition"
+    # Verified against the portal (2026-07-07): Abri stock-condition deals
+    # carry no project association, so the project_code match is the branch
+    # that fires in production; the id below is the project object id
+    # embedded at the end of that code, kept for when associations appear.
+    ABRI_CONDITION_ASSOCIATED_PROJECT_ID = "1247536318670"
+    ABRI_CONDITION_PROJECT_CODE = (
+        "[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]"
+    )
 
     @staticmethod
     def check_for_db_update_trigger(
@@ -97,7 +103,7 @@ class HubspotDealDiffer:
             "design_type": "design_type",
             "surveyor": "surveyor",
             "confirmed_survey_time": "confirmed_survey_time",
-            "client_booking_reference": "job_no",
+            "client_booking_reference": "client_booking_reference",
             "survey_type": "survey_type",
             "measures_for_pibi_ordered": "measures_for_pibi_ordered",
             "property_halted_reason": "property_halted_reason",
@@ -183,6 +189,46 @@ class HubspotDealDiffer:
 
         return False
 
+    @staticmethod
+    def check_abri_triggers_and_construct_message(
+        hubspot_deal_id: str,
+        new_deal: Dict[str, str],
+        new_project: Optional[ProjectData],
+        new_listing: Optional[Dict[str, str]],
+        old_deal: HubspotDealData,
+    ) -> Optional[Dict[str, Any]]:
+        flows: List[str] = []
+        if HubspotDealDiffer.check_for_abri_job_amendment(
+            new_deal=new_deal, new_project=new_project, old_deal=old_deal
+        ):
+            flows.append("amend_job")
+        if HubspotDealDiffer.check_for_abri_job_logging(
+            new_deal=new_deal, new_project=new_project, old_deal=old_deal
+        ):
+            flows.append("log_job")
+        if HubspotDealDiffer.check_for_abri_tenant_data_fetch(
+            new_deal=new_deal, new_project=new_project, old_deal=old_deal
+        ):
+            flows.append("sync_tenant_data")
+
+        if not flows:
+            return None
+
+        confirmed_survey_date = parse_hs_date(new_deal.get("confirmed_survey_date"))
+        listing = new_listing or {}
+        return {
+            "hubspot_deal_id": hubspot_deal_id,
+            "flows": flows,
+            "place_ref": listing.get("owner_property_id"),
+            "deal_name": new_deal.get("dealname"),
+            "confirmed_survey_date": (
+                confirmed_survey_date.date().isoformat()
+                if confirmed_survey_date is not None
+                else None
+            ),
+            "confirmed_survey_time": new_deal.get("confirmed_survey_time"),
+        }
+
     @staticmethod
     def check_for_abri_job_logging(
         new_deal: Dict[str, str],

etl/hubspot/scripts/scraper/main.py

@@ -110,21 +110,30 @@ def handler(body: dict[str, Any], context: Any) -> None:
                 f"Triggering Pas Hub file fetcher for HubSpot deal ID {hubspot_deal_id}"
             )
             _trigger_pashub_fetcher(sqs_client, hubspot_deal_id, hubspot_deal)
-        else:
-            logger.info(
-                f"Not Triggering PasHub file fetcher for HubSpot deal ID {hubspot_deal_id}"
-            )
 
         if HubspotDealDiffer.check_for_magicplan_trigger(
             new_deal=hubspot_deal, old_deal=db_deal
         ):
-            logger.info(
-                f"Triggering MagicPlan fetcher for HubSpot deal ID {hubspot_deal_id}"
-            )
             _trigger_magicplan_fetcher(
                 sqs_client, hubspot_deal, listing, hubspot_deal_id
             )
 
+        abri_message: Optional[Dict[str, Any]] = (
+            HubspotDealDiffer.check_abri_triggers_and_construct_message(
+                hubspot_deal_id=hubspot_deal_id,
+                new_deal=hubspot_deal,
+                new_project=project,
+                new_listing=listing,
+                old_deal=db_deal,
+            )
+        )
+        if abri_message is not None:
+            logger.info(
+                f"Triggering Abri flows {abri_message['flows']} for HubSpot "
+                f"deal ID {hubspot_deal_id}"
+            )
+            _trigger_abri_api_lambda(sqs_client, abri_message)
+
     print("done")
 
 
@@ -134,6 +143,8 @@ def _trigger_magicplan_fetcher(
     listing: Optional[dict[str, str]],
     hubspot_deal_id: str,
 ) -> None:
+    logger.info(f"Triggering MagicPlan fetcher for HubSpot deal ID {hubspot_deal_id}")
+
     message_body = {
         "address": hubspot_deal.get("dealname"),
         "hubspot_deal_id": hubspot_deal_id,
@@ -146,6 +157,14 @@ def _trigger_magicplan_fetcher(
     logger.info(f"Sent message to MagicPlan queue. MessageId: {response['MessageId']}")
 
 
+def _trigger_abri_api_lambda(sqs_client: Any, abri_message: Dict[str, Any]) -> None:
+    response = sqs_client.send_message(
+        QueueUrl=get_settings().ABRI_SQS_URL,
+        MessageBody=json.dumps(abri_message),
+    )
+    logger.info(f"Sent message to Abri queue. MessageId: {response['MessageId']}")
+
+
 def _trigger_pashub_fetcher(
     sqs_client: Any, deal_id: str, hubspot_deal: Dict[str, str]
 ) -> None:

etl/hubspot/tests/test_abri_flow_triggers.py

@@ -0,0 +1,232 @@
+from datetime import datetime, timezone
+from typing import Any, Dict
+import uuid
+
+from backend.app.db.models.hubspot_deal_data import HubspotDealData
+from etl.hubspot.abri_flow_triggers import check_for_abri_triggers_and_construct_message
+
+BASE_TIME = datetime(2025, 12, 1, 12, 0, 0)
+
+ABRI_PROJECT_CODE = "[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]"
+
+DEAL_ID = "9876543210"
+
+LISTING = {"listing_id": "55", "owner_property_id": "1007165", "national_uprn": "1"}
+
+
+def make_old_deal(**overrides: Any) -> HubspotDealData:
+    return HubspotDealData(
+        id=overrides.get("id", uuid.uuid4()),
+        deal_id=DEAL_ID,
+        created_at=BASE_TIME,
+        updated_at=BASE_TIME,
+        **{k: v for k, v in overrides.items() if k != "id"},
+    )
+
+
+def make_new_deal(**overrides: str) -> Dict[str, str]:
+    return {
+        "hs_object_id": DEAL_ID,
+        "dealname": "49 Admers Crescent",
+        "project_code": ABRI_PROJECT_CODE,
+        **overrides,
+    }
+
+
+# ==========================
+# SINGLE-FLOW FIRING
+# ==========================
+
+
+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)
+    new_deal = make_new_deal(
+        confirmed_survey_date="2026-06-24", confirmed_survey_time="14:30"
+    )
+
+    # Act
+    message = check_for_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 == {
+        "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",
+    }
+
+
+def test_a_changed_confirmed_survey_date_builds_an_amend_job_message() -> None:
+    # Arrange
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date=datetime(2026, 6, 18, tzinfo=timezone.utc),
+    )
+    new_deal = make_new_deal(
+        confirmed_survey_date="2026-06-24", confirmed_survey_time="14:30"
+    )
+
+    # Act
+    message = check_for_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["confirmed_survey_date"] == "2026-06-24"
+
+
+def test_a_first_expected_commencement_date_builds_a_tenant_sync_message() -> None:
+    # Arrange
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE, expected_commencement_date=None
+    )
+    new_deal = make_new_deal(expected_commencement_date="2026-07-01")
+
+    # Act
+    message = check_for_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"] == ["sync_tenant_data"]
+    assert message["place_ref"] == "1007165"
+
+
+# ==========================
+# ONE MESSAGE PER SCRAPE
+# ==========================
+
+
+def test_several_flows_firing_in_one_scrape_share_one_message() -> None:
+    # Arrange: date changed and commencement date first set together.
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date=datetime(2026, 6, 18, tzinfo=timezone.utc),
+        expected_commencement_date=None,
+    )
+    new_deal = make_new_deal(
+        confirmed_survey_date="2026-06-24",
+        confirmed_survey_time="14:30",
+        expected_commencement_date="2026-07-01",
+    )
+
+    # Act
+    message = check_for_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", "sync_tenant_data"]
+
+
+# ==========================
+# NOTHING FIRES
+# ==========================
+
+
+def test_no_message_when_no_abri_trigger_fires() -> None:
+    # Arrange: something changed, but nothing Abri cares about.
+    old_deal = make_old_deal(project_code=ABRI_PROJECT_CODE, outcome=None)
+    new_deal = make_new_deal(outcome="left voicemail")
+
+    # Act
+    message = check_for_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_no_message_for_a_non_abri_deal() -> None:
+    # Arrange: a survey date first set, but on another project's deal.
+    old_deal = make_old_deal(project_code="Other", confirmed_survey_date=None)
+    new_deal = make_new_deal(project_code="Other", confirmed_survey_date="2026-06-24")
+
+    # Act
+    message = check_for_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_abandonment_alone_stays_unwired_and_builds_no_message() -> None:
+    # Arrange: the abandonment predicate would fire, but CancelJob does not
+    # exist yet, so no flow may be dispatched for it.
+    old_deal = make_old_deal(
+        project_code=ABRI_PROJECT_CODE, number_of_attempts="2", outcome=None
+    )
+    new_deal = make_new_deal(number_of_attempts="3", outcome="no answer")
+
+    # Act
+    message = check_for_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
+
+
+# ==========================
+# PAYLOAD ROBUSTNESS
+# ==========================
+
+
+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")
+
+    # Act
+    message = check_for_abri_triggers_and_construct_message(
+        hubspot_deal_id=DEAL_ID,
+        new_deal=new_deal,
+        new_project=None,
+        new_listing=None,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert message is not None
+    assert message["place_ref"] is None
+    assert message["confirmed_survey_time"] is None

etl/hubspot/tests/test_hubspot_data_to_db.py

@@ -88,7 +88,7 @@ def test_update_existing_deal__no_project__clears_project_id() -> None:
 
 
 def test_update_existing_deal__client_booking_reference_maps_to_job_no() -> None:
-    existing = HubspotDealData(deal_id="MOCK_DEAL_ID", job_no=None)
+    existing = HubspotDealData(deal_id="MOCK_DEAL_ID", client_booking_reference=None)
     deal_data = {"client_booking_reference": "AD0226519"}
 
     _make_instance()._update_existing_deal(
@@ -98,11 +98,13 @@ def test_update_existing_deal__client_booking_reference_maps_to_job_no() -> None
         company=None,
     )
 
-    assert existing.job_no == "AD0226519"
+    assert existing.client_booking_reference == "AD0226519"
 
 
 def test_update_existing_deal__client_booking_reference_cleared__nulls_job_no() -> None:
-    existing = HubspotDealData(deal_id="MOCK_DEAL_ID", job_no="AD0226519")
+    existing = HubspotDealData(
+        deal_id="MOCK_DEAL_ID", client_booking_reference="AD0226519"
+    )
     deal_data = {}
 
     _make_instance()._update_existing_deal(
@@ -112,7 +114,7 @@ def test_update_existing_deal__client_booking_reference_cleared__nulls_job_no()
         company=None,
     )
 
-    assert existing.job_no is None
+    assert existing.client_booking_reference is None
 
 
 def test_build_new_deal__client_booking_reference_maps_to_job_no() -> None:
@@ -124,4 +126,4 @@ def test_build_new_deal__client_booking_reference_maps_to_job_no() -> None:
         project=None,
     )
 
-    assert new_deal.job_no == "AD0226519"
+    assert new_deal.client_booking_reference == "AD0226519"

etl/hubspot/tests/test_hubspot_deal_differ.py

@@ -11,6 +11,10 @@ from etl.hubspot.project_data import ProjectData
 
 BASE_TIME = datetime(2025, 12, 1, 12, 0, 0)
 
+ABRI_PROJECT_CODE = (
+    "[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]"
+)
+
 
 def make_old_deal(**overrides: Any) -> HubspotDealData:
     return HubspotDealData(
@@ -355,12 +359,12 @@ def test_abri_job_logging__date_first_set_on_abri_deal__returns_true() -> None:
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date=None,
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date="2026-07-15",
     )
 
@@ -381,12 +385,12 @@ def test_abri_job_logging__date_already_set__returns_false() -> None:
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date=datetime(2026, 7, 1, tzinfo=timezone.utc),
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date="2026-07-15",
     )
 
@@ -433,12 +437,41 @@ def test_abri_job_logging__project_code_cased_differently__returns_true() -> Non
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="ABRI CONDITION",
+        project_code=ABRI_PROJECT_CODE.upper(),
         confirmed_survey_date=None,
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="ABRI CONDITION",
+        project_code=ABRI_PROJECT_CODE.upper(),
+        confirmed_survey_date="2026-07-15",
+    )
+
+    # Act
+    result = HubspotDealDiffer.check_for_abri_job_logging(
+        new_deal=new_deal,
+        new_project=None,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert result is True
+
+
+def test_abri_job_logging__real_portal_project_code__returns_true() -> None:
+    # The code carried by every Abri stock-condition deal in the portal
+    # (verified 2026-07-07); these deals have no project association, so the
+    # project_code fallback is the branch that must fire in production.
+    deal_id = uuid.uuid4()
+
+    # Arrange
+    old_deal = make_old_deal(
+        id=deal_id,
+        project_code="[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]",
+        confirmed_survey_date=None,
+    )
+    new_deal = make_new_deal(
+        deal_id,
+        project_code="[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]",
         confirmed_survey_date="2026-07-15",
     )
 
@@ -469,12 +502,12 @@ def test_abri_job_logging__no_parseable_new_date__returns_false(
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date=None,
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         **new_overrides,
     )
 
@@ -527,12 +560,12 @@ def test_abri_job_logging__non_abri_project_id_with_abri_code__returns_false() -
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date=None,
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date="2026-07-15",
     )
     new_project = ProjectData(project_id="999999", name="Southern Retrofit")
@@ -559,12 +592,12 @@ def test_abri_job_amendment__date_changed_on_abri_deal__returns_true() -> None:
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date=datetime(2026, 7, 1, tzinfo=timezone.utc),
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date="2026-07-15",
     )
 
@@ -611,13 +644,13 @@ def test_abri_job_amendment__date_and_time_unchanged__returns_false() -> None:
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date=datetime(2026, 7, 15, tzinfo=timezone.utc),
         confirmed_survey_time="09:30",
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date="2026-07-15",
         confirmed_survey_time="09:30",
     )
@@ -639,13 +672,13 @@ def test_abri_job_amendment__time_changed_on_abri_deal__returns_true() -> None:
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date=datetime(2026, 7, 15, tzinfo=timezone.utc),
         confirmed_survey_time="09:30",
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date="2026-07-15",
         confirmed_survey_time="14:00",
     )
@@ -667,12 +700,12 @@ def test_abri_job_amendment__date_first_set_on_abri_deal__returns_false() -> Non
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date=None,
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         confirmed_survey_date="2026-07-15",
     )
 
@@ -700,12 +733,12 @@ def test_abri_tenant_data_fetch__commencement_date_first_set_on_abri_deal__retur
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         expected_commencement_date=None,
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         expected_commencement_date="2026-08-01",
     )
 
@@ -752,12 +785,12 @@ def test_abri_tenant_data_fetch__commencement_date_already_set__returns_false()
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         expected_commencement_date=datetime(2026, 7, 1, tzinfo=timezone.utc),
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         expected_commencement_date="2026-08-01",
     )
 
@@ -788,12 +821,12 @@ def test_abri_tenant_data_fetch__no_parseable_new_commencement_date__returns_fal
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         expected_commencement_date=None,
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         **new_overrides,
     )
 
@@ -846,12 +879,12 @@ def test_abri_tenant_data_fetch__non_abri_project_id_with_abri_code__returns_fal
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         expected_commencement_date=None,
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         expected_commencement_date="2026-08-01",
     )
     new_project = ProjectData(project_id="999999", name="Southern Retrofit")
@@ -880,12 +913,12 @@ def test_abri_deal_abandonment__third_attempt_with_negative_outcome__returns_tru
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         outcome=None,
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         number_of_attempts="3",
         outcome="No Answer",
     )
@@ -945,12 +978,12 @@ def test_abri_deal_abandonment__fewer_than_three_parseable_attempts__returns_fal
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         outcome=None,
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         outcome="No Answer",
         **new_overrides,
     )
@@ -982,12 +1015,12 @@ def test_abri_deal_abandonment__outcome_not_negative__returns_false(
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         outcome=None,
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         number_of_attempts="3",
         **new_overrides,
     )
@@ -1022,12 +1055,12 @@ def test_abri_deal_abandonment__each_negative_outcome__returns_true(
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         outcome=None,
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         number_of_attempts="3",
         outcome=outcome,
     )
@@ -1049,13 +1082,13 @@ def test_abri_deal_abandonment__deal_already_abandoned__returns_false() -> None:
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         number_of_attempts="3",
         outcome="No Answer",
     )
     new_deal = make_new_deal(
         deal_id,
-        project_code="Abri Condition",
+        project_code=ABRI_PROJECT_CODE,
         number_of_attempts="4",
         outcome="No Answer",
     )
@@ -1234,7 +1267,7 @@ def test_db_update_trigger__client_booking_reference_changed__returns_true() ->
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        job_no=None,
+        client_booking_reference=None,
     )
     new_deal = make_new_deal(
         deal_id,
@@ -1260,7 +1293,7 @@ def test_db_update_trigger__client_booking_reference_unchanged__returns_false()
     # Arrange
     old_deal = make_old_deal(
         id=deal_id,
-        job_no="AD0226519",
+        client_booking_reference="AD0226519",
     )
     new_deal = make_new_deal(
         deal_id,

etl/hubspot/tests/test_scraper_handler.py

@@ -12,6 +12,8 @@ DEAL_ID = "999"
 PASHUB_LINK = "https://pashub.example.com/deal/999"
 MAGICPLAN_QUEUE_URL = "https://sqs.eu-west-2.amazonaws.com/123/magic-plan-dev"
 PASHUB_QUEUE_URL = "https://sqs.test/pashub"
+ABRI_API_QUEUE_URL = "https://sqs.test/abri"
+ABRI_PROJECT_CODE = "[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]"
 PROJECT = {"project_id": "proj-1", "name": "Project One"}
 
 
@@ -59,6 +61,7 @@ def run_handler(
         mock_boto3.client.return_value = mock_sqs
         mock_settings.return_value.MAGICPLAN_SQS_URL = MAGICPLAN_QUEUE_URL
         mock_settings.return_value.PASHUB_TO_ARA_SQS_URL = PASHUB_QUEUE_URL
+        mock_settings.return_value.ABRI_SQS_URL = ABRI_API_QUEUE_URL
 
         handler.__wrapped__({"hubspot_deal_id": DEAL_ID}, "")
 
@@ -126,7 +129,9 @@ def test_existing_deal_surveyed_transition__sends_magicplan_sqs() -> None:
     listing = {"national_uprn": UPRN}
 
     # Act
-    mock_sqs, _ = run_handler(hubspot_deal=hubspot_deal, db_deal=db_deal, listing=listing)
+    mock_sqs, _ = run_handler(
+        hubspot_deal=hubspot_deal, db_deal=db_deal, listing=listing
+    )
 
     # Assert
     mock_sqs.send_message.assert_called_once_with(
@@ -229,6 +234,42 @@ def test_existing_deal_pashub_link_unchanged__no_pashub_sqs() -> None:
     mock_sqs.send_message.assert_not_called()
 
 
+# ==========================================
+# EXISTING DEAL PATH - Abri trigger
+# ==========================================
+
+
+def test_existing_abri_deal_first_survey_date__sends_abri_sqs() -> None:
+    # Arrange
+    db_deal = make_db_deal(project_code=ABRI_PROJECT_CODE, confirmed_survey_date=None)
+    hubspot_deal = make_hubspot_deal(
+        project_code=ABRI_PROJECT_CODE,
+        confirmed_survey_date="2026-06-24",
+        confirmed_survey_time="14:30",
+    )
+    listing = {"owner_property_id": "1007165", "national_uprn": UPRN}
+
+    # Act
+    mock_sqs, _ = run_handler(
+        hubspot_deal=hubspot_deal, db_deal=db_deal, listing=listing
+    )
+
+    # Assert
+    mock_sqs.send_message.assert_called_once_with(
+        QueueUrl=ABRI_API_QUEUE_URL,
+        MessageBody=json.dumps(
+            {
+                "hubspot_deal_id": DEAL_ID,
+                "flows": ["log_job"],
+                "place_ref": "1007165",
+                "deal_name": DEAL_NAME,
+                "confirmed_survey_date": "2026-06-24",
+                "confirmed_survey_time": "14:30",
+            }
+        ),
+    )
+
+
 # ====================================
 # PROJECT upsert
 # ====================================

orchestration/abri_orchestrator.py

@@ -5,6 +5,8 @@ from typing import Dict, List, Optional, Protocol, Tuple, Union
 from domain.abri.descriptions import build_job_descriptions
 from domain.abri.models import (
     AbriRequestRejected,
+    AmendJobRequest,
+    AppointmentAmended,
     LogJobRequest,
     PlaceRef,
     Tenant,
@@ -123,6 +125,35 @@ LogJobOrchestrationResult = Union[LogJobSummary, AbriRequestRejected]
 class DealDatabaseGateway(Protocol):
     def record_job_no(self, deal_id: str, job_no: str) -> None: ...
 
+    def job_no_for_deal(self, deal_id: str) -> Optional[str]: ...
+
+
+@dataclass(frozen=True)
+class AppointmentChange:
+    """The deal fields a changed survey booking contributes to an AmendJob."""
+
+    deal_id: str
+    confirmed_survey_date: date
+    confirmed_survey_time: Optional[str]
+
+
+AmendJobOrchestrationResult = Union[AppointmentAmended, AbriRequestRejected]
+
+
+class JobNoNotYetRecordedError(Exception):
+    """An amendment arrived before the deal's job_no landed in the database.
+
+    Retriable: the message should be redelivered until the log flow's
+    write-back lands (or it dead-letters), so a rapid log-then-amend
+    sequence eventually converges.
+    """
+
+    def __init__(self, deal_id: str) -> None:
+        self.deal_id = deal_id
+        super().__init__(
+            f"no job_no recorded yet for deal {deal_id}; cannot amend its appointment"
+        )
+
 
 class LogJobWriteBackError(Exception):
     """A job logged in OpenHousing whose job_no could not be written back.
@@ -208,6 +239,21 @@ class AbriOrchestrator:
             ),
         )
 
+    def amend_job(self, change: AppointmentChange) -> AmendJobOrchestrationResult:
+        job_no = self._deal_database.job_no_for_deal(change.deal_id)
+        if job_no is None:
+            raise JobNoNotYetRecordedError(deal_id=change.deal_id)
+
+        return self._abri.amend_job(
+            AmendJobRequest(
+                job_no=job_no,
+                appointment_date=change.confirmed_survey_date,
+                appointment_time=slot_for_confirmed_time(
+                    change.confirmed_survey_time
+                ),
+            )
+        )
+
     def log_job(self, booking: ConfirmedSurveyBooking) -> LogJobOrchestrationResult:
         descriptions = build_job_descriptions(booking.deal_name)
         outcome = self._abri.log_job(

repositories/hubspot_deals/deal_database_postgres_gateway.py

@@ -0,0 +1,34 @@
+"""Postgres implementation of the Abri deal-database gateway.
+
+First production implementation of the DealDatabaseGateway protocol used by
+the AbriOrchestrator (orchestration/abri_orchestrator.py): job numbers live
+on the hubspot_deal_data table, in the job_no column the scraper also
+upserts from HubSpot's client_booking_reference property.
+"""
+
+from typing import Optional
+
+from sqlmodel import Session, select
+
+from backend.app.db.models.hubspot_deal_data import HubspotDealData
+
+
+class DealDatabasePostgresGateway:
+    def __init__(self, session: Session) -> None:
+        self._session = session
+
+    def record_job_no(self, deal_id: str, job_no: str) -> None:
+        deal = self._find_deal(deal_id)
+        if deal is None:
+            raise ValueError(f"HubSpot deal {deal_id} not found")
+        deal.client_booking_reference = job_no
+        self._session.add(deal)
+        self._session.commit()
+
+    def job_no_for_deal(self, deal_id: str) -> Optional[str]:
+        deal = self._find_deal(deal_id)
+        return deal.client_booking_reference if deal is not None else None
+
+    def _find_deal(self, deal_id: str) -> Optional[HubspotDealData]:
+        statement = select(HubspotDealData).where(HubspotDealData.deal_id == deal_id)
+        return self._session.exec(statement).first()

scripts/smoke_test_abri_log_amend.py

@@ -0,0 +1,145 @@
+"""Manual smoke test for the Abri log -> amend wiring against UAT (issue #1480).
+
+Drives the real AbriOrchestrator log-job and amend-job flows end-to-end
+against Abri's UAT relay endpoint: a job is logged for a test place
+reference, its job_no is written back to a throwaway HubSpot deal, and the
+appointment is then amended to the following day's other slot — proving the
+full log -> write-back -> amend sequence converges.
+
+The deal-database gateway is in-memory: the write-back to the real deal-data
+table belongs to the deployed lambda, and this script must not touch a
+production database.
+
+Usage:
+  1. Ensure these are set in the environment or backend/.env:
+     HUBSPOT_API_KEY, ABRI_RELAY_URL (the UAT endpoint),
+     ABRI_RELAY_USERNAME, ABRI_RELAY_PASSWORD, ABRI_RELAY_DEFAULT_RESOURCE.
+  2. Create a throwaway test deal in the HubSpot UI and paste its id into
+     DEAL_ID below; paste Abri's agreed UAT test place reference into
+     PLACE_REF.
+  3. Run: python scripts/smoke_test_abri_log_amend.py
+  4. Check: the printed job_no matches the deal's client_booking_reference
+     in the HubSpot UI, and Abri can see the appointment on the amended
+     date/slot in their TM solution.
+  5. Clean up: delete the test deal in the UI and ask Abri to cancel the
+     UAT job (there is no CancelJob route yet).
+"""
+
+import os
+from datetime import date, timedelta
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+
+from dotenv import dotenv_values
+from hubspot.client import Client  # type: ignore[reportMissingTypeStubs]
+
+from domain.abri.models import AbriRequestRejected, PlaceRef
+from infrastructure.abri.abri_client import AbriClient
+from infrastructure.abri.config import AbriConfig
+from infrastructure.hubspot.deal_contacts_client import HubspotDealContactsClient
+from infrastructure.hubspot.deal_properties_client import HubspotDealPropertiesClient
+from orchestration.abri_orchestrator import (
+    AbriOrchestrator,
+    AppointmentChange,
+    ConfirmedSurveyBooking,
+)
+
+DEAL_ID = "EDIT-ME"
+PLACE_REF = PlaceRef("EDIT-ME")
+
+# Log for the first weekday at least a week out (a slot Abri UAT will take),
+# then amend to the following day's opposite slot.
+LOG_DATE = date.today() + timedelta(days=7)
+LOG_TIME = "10:00"  # AM slot
+AMEND_DATE = LOG_DATE + timedelta(days=1)
+AMEND_TIME = "14:30"  # PM slot
+
+
+def _env(name: str) -> str:
+    env_file = Path(__file__).parents[1] / "backend" / ".env"
+    value = os.getenv(name) or dotenv_values(env_file).get(name)
+    if not value:
+        raise SystemExit(f"Missing {name} (env var or backend/.env)")
+    return value
+
+
+def _hubspot_sdk_client() -> Client:
+    return Client.create(access_token=_env("HUBSPOT_API_KEY"))  # type: ignore[reportUnknownMemberType]
+
+
+def _uat_abri_client() -> AbriClient:
+    return AbriClient(
+        config=AbriConfig(
+            endpoint_url=_env("ABRI_RELAY_URL"),
+            username=_env("ABRI_RELAY_USERNAME"),
+            password=_env("ABRI_RELAY_PASSWORD"),
+            default_resource=_env("ABRI_RELAY_DEFAULT_RESOURCE"),
+        )
+    )
+
+
+class InMemoryDealDatabase:
+    """Stands in for the deal-data table; the real write-back is the lambda's."""
+
+    def __init__(self) -> None:
+        self.recorded: List[Tuple[str, str]] = []
+        self._job_nos: Dict[str, str] = {}
+
+    def record_job_no(self, deal_id: str, job_no: str) -> None:
+        self.recorded.append((deal_id, job_no))
+        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)
+
+
+def main() -> None:
+    if DEAL_ID == "EDIT-ME" or PLACE_REF == PlaceRef("EDIT-ME"):
+        raise SystemExit("Edit DEAL_ID and PLACE_REF before running")
+
+    sdk_client = _hubspot_sdk_client()
+    deal_database = InMemoryDealDatabase()
+    orchestrator = AbriOrchestrator(
+        abri_client=_uat_abri_client(),
+        deal_contacts=HubspotDealContactsClient(sdk_client=sdk_client),
+        deal_properties=HubspotDealPropertiesClient(sdk_client=sdk_client),
+        deal_database=deal_database,
+    )
+
+    print(f"Logging job for place_ref {PLACE_REF} on {LOG_DATE} at {LOG_TIME}...")
+    log_result = orchestrator.log_job(
+        ConfirmedSurveyBooking(
+            deal_id=DEAL_ID,
+            place_ref=PLACE_REF,
+            deal_name=f"Domna UAT smoke test {date.today().isoformat()}",
+            confirmed_survey_date=LOG_DATE,
+            confirmed_survey_time=LOG_TIME,
+        )
+    )
+    if isinstance(log_result, AbriRequestRejected):
+        raise SystemExit(
+            f"LogJob rejected: {log_result.code}: {log_result.message}"
+        )
+    print(f"Job logged: job_no={log_result.job_no} (written back to deal {DEAL_ID})")
+
+    print(f"Amending appointment to {AMEND_DATE} at {AMEND_TIME}...")
+    amend_result = orchestrator.amend_job(
+        AppointmentChange(
+            deal_id=DEAL_ID,
+            confirmed_survey_date=AMEND_DATE,
+            confirmed_survey_time=AMEND_TIME,
+        )
+    )
+    if isinstance(amend_result, AbriRequestRejected):
+        raise SystemExit(
+            f"AmendJob rejected: {amend_result.code}: {amend_result.message}"
+        )
+    print(
+        f"Appointment amended: job_no={amend_result.job_no} "
+        f"date={amend_result.appointment_date} slot={amend_result.appointment_time}"
+    )
+    print("Smoke test complete. Remember to clean up (see module docstring).")
+
+
+if __name__ == "__main__":
+    main()

scripts/smoke_test_abri_logjob_flow.py

@@ -0,0 +1,271 @@
+"""Manual smoke test for the scraper->dispatch LogJob wiring (issue #1480).
+
+The closest run to end-to-end available before Abri send API credentials:
+every seam is real except the Abri HTTP edge, which is stubbed to return a
+canned LogJob success carrying a fake job_no. Concretely, the script
+
+  1. fetches the deal, listing and project from the real HubSpot portal
+     (exactly what the scraper lambda sees),
+  2. reads the deal's row from the real (dev) deal database,
+  3. runs the real differ predicates via abri_message_if_triggered,
+  4. validates the real message contract (AbriTriggerRequest),
+  5. runs the real dispatch through the real AbriOrchestrator, so the fake
+     job_no is written to the deal's client_booking_reference property in
+     HubSpot first, then to the same-named column in the database,
+  6. reads both back and prints them.
+
+NOTE: the LogJob trigger fires when the CONFIRMED SURVEY DATE is first set
+on an Abri deal. (Setting the *expected commencement date* fires the
+tenant-data flow instead — that flow is skipped here so the stub never
+mints junk contacts; smoke_test_tenant_contacts.py covers it.)
+
+Usage:
+  1. Ensure HUBSPOT_API_KEY and the postgres_* connection values are in
+     backend/.env (they are read tolerantly; the strict Settings .env
+     parsing is bypassed).
+  2. Pick a test deal on the "Abri Condition" project that the scraper has
+     already ingested (its row must exist in hubspot_deal_data). Paste its
+     id into DEAL_ID below.
+  3. In the HubSpot UI, set the deal's confirmed survey date (and
+     optionally a time). If the dev scraper lambda syncs the change into
+     the database before you run this, the differ will see no change — set
+     RESET_DB_ROW_FIRST = True to null the row's survey date and job_no so
+     the trigger fires deterministically.
+  4. Run: python scripts/smoke_test_abri_logjob_flow.py
+  5. Check the printed read-backs, and the deal in the HubSpot UI: the
+     fake job_no (SMOKE...) should sit in client_booking_reference.
+  6. Run it again unchanged: the dispatch should print
+     "skipped: job SMOKE... already logged" — the idempotency guard.
+  7. Clean up: set CLEANUP = True and re-run to clear the fake job_no from
+     both HubSpot and the database, then unset the deal's survey date.
+"""
+
+import os
+import xml.etree.ElementTree as ET
+from datetime import datetime
+from pathlib import Path
+from typing import Any, List, Optional, cast
+
+from dotenv import dotenv_values
+
+DEAL_ID = "EDIT-ME"
+
+# Null the DB row's confirmed_survey_date + job_no before evaluating the
+# differ, so the trigger fires even if the dev scraper already synced it.
+RESET_DB_ROW_FIRST = False
+
+# Cleanup mode: clear the fake job_no from HubSpot and the database.
+CLEANUP = False
+
+FAKE_JOB_NO = "SMOKE" + datetime.now().strftime("%d%H%M")
+
+JOB_NO_DEAL_PROPERTY = "client_booking_reference"
+
+
+def _bootstrap_env() -> None:
+    """Export backend/.env tolerantly and bypass strict Settings parsing.
+
+    backend/.env carries keys the strict Settings model rejects; exporting
+    them ourselves and pointing ENVIRONMENT away from "local" lets the
+    legacy HubspotClient (and PostgresConfig.from_env) run unchanged.
+    """
+    env_file = Path(__file__).parents[1] / "backend" / ".env"
+    for key, value in dotenv_values(env_file).items():
+        if value is not None:
+            os.environ.setdefault(key.upper(), value)
+    os.environ["ENVIRONMENT"] = "smoke-test"
+
+
+class _CannedLogJobResponse:
+    content = (
+        f'<Root><Jobs><Job_logged job_no="{FAKE_JOB_NO}" '
+        f'logged_info="Job logged. Appointment booked." /></Jobs></Root>'
+    ).encode()
+
+    def raise_for_status(self) -> None:
+        pass
+
+
+class _StubAbriSession:
+    """Stands in for requests.Session; records envelopes, returns the canned
+    LogJob success. Anything but a logjob request is a wiring bug."""
+
+    def __init__(self) -> None:
+        self.sent_envelopes: List[bytes] = []
+
+    def post(self, url: str, data: bytes) -> _CannedLogJobResponse:
+        request_type = (
+            ET.fromstring(data).find("Body/Request").get("request_type")  # type: ignore[reportOptionalMemberAccess]
+        )
+        if request_type != "logjob":
+            raise AssertionError(
+                f"stub received a {request_type!r} request; only logjob is smoke-tested"
+            )
+        self.sent_envelopes.append(data)
+        return _CannedLogJobResponse()
+
+
+def _print_sent_parameters(envelope: bytes) -> None:
+    parameters = {
+        parameter.get("attribute"): parameter.get("attribute_value")
+        for parameter in ET.fromstring(envelope).findall("Body/Request/Parameters")
+    }
+    print("  Envelope sent to (stubbed) Abri relay:")
+    for name, value in parameters.items():
+        print(f"    {name} = {value}")
+
+
+def main() -> None:
+    if DEAL_ID == "EDIT-ME":
+        raise SystemExit("Edit DEAL_ID before running")
+
+    _bootstrap_env()
+
+    # Imports that read settings/env must come after the bootstrap.
+    from hubspot.client import Client  # type: ignore[reportMissingTypeStubs]
+    from sqlalchemy.pool import NullPool
+    from sqlmodel import Session, select
+
+    from applications.abri.abri_trigger_request import AbriTriggerRequest
+    from applications.abri.dispatch import dispatch_abri_flows
+    from backend.app.db.models.hubspot_deal_data import HubspotDealData
+    from etl.hubspot.abri_flow_triggers import (
+        check_for_abri_triggers_and_construct_message,
+    )
+    from etl.hubspot.hubspotClient import HubspotClient
+    from infrastructure.abri.abri_client import AbriClient
+    from infrastructure.abri.config import AbriConfig
+    from infrastructure.hubspot.deal_contacts_client import HubspotDealContactsClient
+    from infrastructure.hubspot.deal_properties_client import (
+        HubspotDealPropertiesClient,
+    )
+    from infrastructure.postgres.config import PostgresConfig
+    from infrastructure.postgres.engine import make_engine
+    from orchestration.abri_orchestrator import AbriOrchestrator
+    from repositories.hubspot_deals.deal_database_postgres_gateway import (
+        DealDatabasePostgresGateway,
+    )
+
+    engine = make_engine(PostgresConfig.from_env(dict(os.environ)), poolclass=NullPool)
+    sdk_client: Client = Client.create(  # type: ignore[reportUnknownMemberType]
+        access_token=os.environ["HUBSPOT_API_KEY"]
+    )
+    deal_properties = HubspotDealPropertiesClient(sdk_client=sdk_client)
+
+    with Session(engine) as session:
+        deal_row = session.exec(
+            select(HubspotDealData).where(HubspotDealData.deal_id == DEAL_ID)
+        ).first()
+        if deal_row is None:
+            raise SystemExit(
+                f"Deal {DEAL_ID} has no hubspot_deal_data row — let the scraper "
+                "ingest it first (the gateway write-back needs the row)."
+            )
+
+        if CLEANUP:
+            deal_properties.write_deal_property(
+                deal_id=DEAL_ID, property_name=JOB_NO_DEAL_PROPERTY, value=""
+            )
+            deal_row.client_booking_reference = None
+            session.add(deal_row)
+            session.commit()
+            print(
+                f"Cleaned up: cleared {JOB_NO_DEAL_PROPERTY} in HubSpot and "
+                "job_no in the database. Unset the survey date in the UI too."
+            )
+            return
+
+        if RESET_DB_ROW_FIRST:
+            deal_row.confirmed_survey_date = None
+            deal_row.client_booking_reference = None
+            session.add(deal_row)
+            session.commit()
+            session.refresh(deal_row)
+            print("Reset DB row: confirmed_survey_date and job_no nulled.")
+
+        print(f"Fetching deal {DEAL_ID} from HubSpot...")
+        hubspot_deal, _company, listing, project = (
+            HubspotClient().get_deal_and_company_and_listing_and_project(DEAL_ID)
+        )
+        print(
+            f"  dealname={hubspot_deal.get('dealname')!r} "
+            f"project_code={hubspot_deal.get('project_code')!r} "
+            f"confirmed_survey_date={hubspot_deal.get('confirmed_survey_date')!r} "
+            f"confirmed_survey_time={hubspot_deal.get('confirmed_survey_time')!r}"
+        )
+        print(
+            f"  DB row: confirmed_survey_date={deal_row.confirmed_survey_date} "
+            f"job_no={deal_row.client_booking_reference}"
+        )
+
+        message = check_for_abri_triggers_and_construct_message(
+            hubspot_deal_id=DEAL_ID,
+            new_deal=hubspot_deal,
+            new_project=project,
+            new_listing=listing,
+            old_deal=deal_row,
+        )
+        if message is None:
+            raise SystemExit(
+                "No Abri trigger fired. For LogJob, set the CONFIRMED SURVEY "
+                "DATE on the deal (expected commencement date fires the "
+                "tenant-data flow, not job logging). If the dev scraper "
+                "already synced your change, set RESET_DB_ROW_FIRST = True."
+            )
+        print(f"Trigger message built: flows={message['flows']}")
+
+        request = AbriTriggerRequest.model_validate(message)
+        if "log_job" not in request.flows:
+            raise SystemExit(
+                f"Fired flows {request.flows} do not include log_job — set the "
+                "confirmed survey date (not the expected commencement date)."
+            )
+        if request.flows != ["log_job"]:
+            skipped = [flow for flow in request.flows if flow != "log_job"]
+            print(f"  Restricting to log_job for this smoke; skipping {skipped}.")
+            request = request.model_copy(update={"flows": ["log_job"]})
+
+        stub_session = _StubAbriSession()
+        abri_client = AbriClient(
+            config=AbriConfig(
+                endpoint_url="https://stubbed.invalid/relay",
+                username="smoke-test",
+                password="",
+                default_resource="SMOKE",
+            )
+        )
+        abri_client._session = cast(  # pyright: ignore[reportPrivateUsage]
+            Any, stub_session
+        )
+        gateway = DealDatabasePostgresGateway(session=session)
+        orchestrator = AbriOrchestrator(
+            abri_client=abri_client,
+            deal_contacts=HubspotDealContactsClient(sdk_client=sdk_client),
+            deal_properties=deal_properties,
+            deal_database=gateway,
+        )
+
+        print("Dispatching...")
+        summary = dispatch_abri_flows(request, orchestrator, gateway)
+        for envelope in stub_session.sent_envelopes:
+            _print_sent_parameters(envelope)
+        print(f"Dispatch summary: {summary}")
+
+        # Read both write-backs back through the real interfaces.
+        deal = sdk_client.crm.deals.basic_api.get_by_id(  # type: ignore[reportUnknownMemberType]
+            DEAL_ID, properties=[JOB_NO_DEAL_PROPERTY]
+        )
+        hubspot_value = cast(
+            Optional[str],
+            cast(Any, deal).properties.get(JOB_NO_DEAL_PROPERTY),
+        )
+        print(f"HubSpot {JOB_NO_DEAL_PROPERTY} is now: {hubspot_value!r}")
+        print(f"Database job_no is now: {gateway.job_no_for_deal(DEAL_ID)!r}")
+        print(
+            "Re-run to see the idempotency guard skip the log; "
+            "set CLEANUP = True to undo."
+        )
+
+
+if __name__ == "__main__":
+    main()

scripts/smoke_test_tenant_contacts.py

@@ -22,7 +22,7 @@ All tenant details below are fictional (Ofcom-reserved number ranges).
 
 import os
 from pathlib import Path
-from typing import Any, List, cast
+from typing import Any, List, Optional, cast
 
 from dotenv import dotenv_values
 from hubspot.client import Client  # type: ignore[reportMissingTypeStubs]
@@ -101,11 +101,14 @@ def _stubbed_abri_client() -> AbriClient:
 
 
 class _UnusedDealDatabase:
-    """Satisfies the gateway dependency; the tenant-data flow never writes it."""
+    """Satisfies the gateway dependency; the tenant-data flow never touches it."""
 
     def record_job_no(self, deal_id: str, job_no: str) -> None:
         raise AssertionError("tenant-data sync must not touch the deal database")
 
+    def job_no_for_deal(self, deal_id: str) -> Optional[str]:
+        raise AssertionError("tenant-data sync must not touch the deal database")
+
 
 def _archive_contacts(sdk_client: Client) -> None:
     if not CONTACT_IDS_TO_ARCHIVE:

tests/applications/abri/test_abri_trigger_request.py

@@ -0,0 +1,103 @@
+from datetime import date
+from typing import Any, Dict
+
+import pytest
+from pydantic import ValidationError
+
+from applications.abri.abri_trigger_request import AbriTriggerRequest
+
+
+def _full_message() -> Dict[str, Any]:
+    return {
+        "hubspot_deal_id": "9876543210",
+        "flows": ["amend_job", "log_job", "sync_tenant_data"],
+        "place_ref": "1007165",
+        "deal_name": "49 Admers Crescent",
+        "confirmed_survey_date": "2026-06-24",
+        "confirmed_survey_time": "14:30",
+    }
+
+
+def test_a_full_message_parses_into_a_typed_request() -> None:
+    # Act
+    request = AbriTriggerRequest.model_validate(_full_message())
+
+    # Assert
+    assert request.hubspot_deal_id == "9876543210"
+    assert request.flows == ["amend_job", "log_job", "sync_tenant_data"]
+    assert request.place_ref == "1007165"
+    assert request.deal_name == "49 Admers Crescent"
+    assert request.confirmed_survey_date == date(2026, 6, 24)
+    assert request.confirmed_survey_time == "14:30"
+
+
+@pytest.mark.parametrize(
+    ("flow", "missing_field"),
+    [
+        ("log_job", "place_ref"),
+        ("log_job", "deal_name"),
+        ("log_job", "confirmed_survey_date"),
+        ("amend_job", "confirmed_survey_date"),
+        ("sync_tenant_data", "place_ref"),
+    ],
+)
+def test_a_message_missing_a_field_its_flow_needs_fails_validation(
+    flow: str, missing_field: str
+) -> None:
+    # Arrange
+    message = _full_message()
+    message["flows"] = [flow]
+    del message[missing_field]
+
+    # Act / Assert
+    with pytest.raises(ValidationError, match=missing_field):
+        AbriTriggerRequest.model_validate(message)
+
+
+def test_a_message_missing_only_fields_of_unfired_flows_parses() -> None:
+    # Arrange: tenant sync needs only the place_ref (and the deal id).
+    message = {
+        "hubspot_deal_id": "9876543210",
+        "flows": ["sync_tenant_data"],
+        "place_ref": "1007165",
+    }
+
+    # Act
+    request = AbriTriggerRequest.model_validate(message)
+
+    # Assert
+    assert request.flows == ["sync_tenant_data"]
+    assert request.confirmed_survey_date is None
+
+
+def test_a_message_without_a_confirmed_time_parses_for_logging() -> None:
+    # Arrange: the time slot is optional — no time means an all-day slot.
+    message = _full_message()
+    message["flows"] = ["log_job"]
+    del message["confirmed_survey_time"]
+
+    # Act
+    request = AbriTriggerRequest.model_validate(message)
+
+    # Assert
+    assert request.confirmed_survey_time is None
+
+
+def test_a_message_naming_an_unknown_flow_fails_validation() -> None:
+    # Arrange
+    message = _full_message()
+    message["flows"] = ["cancel_job"]
+
+    # Act / Assert
+    with pytest.raises(ValidationError):
+        AbriTriggerRequest.model_validate(message)
+
+
+def test_a_message_naming_no_flows_fails_validation() -> None:
+    # Arrange
+    message = _full_message()
+    message["flows"] = []
+
+    # Act / Assert
+    with pytest.raises(ValidationError):
+        AbriTriggerRequest.model_validate(message)

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)",
+    }

tests/orchestration/test_abri_orchestrator_amend_job.py

@@ -0,0 +1,178 @@
+import xml.etree.ElementTree as ET
+from datetime import date
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from domain.abri.models import AbriRequestRejected, AppointmentAmended
+from infrastructure.abri.abri_client import AbriClient
+from infrastructure.abri.config import AbriConfig
+from infrastructure.hubspot.deal_contacts_client import HubspotDealContactsClient
+from infrastructure.hubspot.deal_properties_client import HubspotDealPropertiesClient
+from orchestration.abri_orchestrator import (
+    AbriOrchestrator,
+    AppointmentChange,
+    JobNoNotYetRecordedError,
+)
+
+FIXTURE_DIR = Path(__file__).parents[1] / "abri"
+ENDPOINT_URL = "https://relay.example.test/api/DomnaRelay?code=test-function-key"
+DEAL_ID = "9876543210"
+JOB_NO = "AC0439951"
+
+CONFIG = AbriConfig(
+    endpoint_url=ENDPOINT_URL,
+    username="DomnaWeb",
+    password="",
+    default_resource="NAULKH",
+)
+
+CHANGE = AppointmentChange(
+    deal_id=DEAL_ID,
+    confirmed_survey_date=date(2026, 6, 24),
+    confirmed_survey_time="14:30",
+)
+
+
+def _load_fixture(name: str) -> bytes:
+    return (FIXTURE_DIR / name).read_bytes()
+
+
+class FakeDealDatabase:
+    """In-memory stand-in for the deal-database gateway."""
+
+    def __init__(self) -> None:
+        self.recorded_job_nos: List[Tuple[str, str]] = []
+        self.job_nos: Dict[str, str] = {}
+
+    def record_job_no(self, deal_id: str, job_no: str) -> None:
+        self.recorded_job_nos.append((deal_id, job_no))
+        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 mock_session() -> MagicMock:
+    return MagicMock()
+
+
+@pytest.fixture()
+def deal_database() -> FakeDealDatabase:
+    return FakeDealDatabase()
+
+
+@pytest.fixture()
+def orchestrator(
+    mock_session: MagicMock,
+    deal_database: FakeDealDatabase,
+) -> AbriOrchestrator:
+    sdk_client = MagicMock()
+    with patch(
+        "infrastructure.abri.abri_client.requests.Session",
+        return_value=mock_session,
+    ):
+        abri_client = AbriClient(config=CONFIG)
+    return AbriOrchestrator(
+        abri_client=abri_client,
+        deal_contacts=HubspotDealContactsClient(sdk_client=sdk_client),
+        deal_properties=HubspotDealPropertiesClient(sdk_client=sdk_client),
+        deal_database=deal_database,
+    )
+
+
+# --- outbound request: the AmendOptiAppt envelope targets the recorded job ---
+
+
+def test_amend_job_sends_an_amendoptiappt_envelope_for_the_deals_recorded_job(
+    orchestrator: AbriOrchestrator,
+    mock_session: MagicMock,
+    deal_database: FakeDealDatabase,
+) -> None:
+    # Arrange
+    deal_database.job_nos[DEAL_ID] = JOB_NO
+    mock_session.post.return_value.content = _load_fixture(
+        "abri_relay_amendoptiappt_success_response.xml"
+    )
+
+    # Act
+    orchestrator.amend_job(CHANGE)
+
+    # Assert
+    (url,) = mock_session.post.call_args.args
+    sent_body: bytes = mock_session.post.call_args.kwargs["data"]
+    sent_parameters = {
+        parameter.get("attribute"): parameter.get("attribute_value")
+        for parameter in ET.fromstring(sent_body).findall("Body/Request/Parameters")
+    }
+    assert url == ENDPOINT_URL
+    assert sent_parameters == {
+        "job_no": JOB_NO,
+        "action": "amend",
+        "appointment_date": "24/06/2026",
+        "appointment_time": "PM",
+    }
+
+
+# --- happy path: the amended appointment is confirmed back ---
+
+
+def test_amend_job_returns_the_amended_appointment(
+    orchestrator: AbriOrchestrator,
+    mock_session: MagicMock,
+    deal_database: FakeDealDatabase,
+) -> None:
+    # Arrange
+    deal_database.job_nos[DEAL_ID] = JOB_NO
+    mock_session.post.return_value.content = _load_fixture(
+        "abri_relay_amendoptiappt_success_response.xml"
+    )
+
+    # Act
+    result = orchestrator.amend_job(CHANGE)
+
+    # Assert
+    assert result == AppointmentAmended(
+        job_no=JOB_NO,
+        appointment_date="24/06/2026",
+        appointment_time="PM",
+    )
+
+
+# --- ordering race: amendment before the job_no write-back has landed ---
+
+
+def test_amend_job_raises_a_retriable_error_when_no_job_no_has_been_recorded(
+    orchestrator: AbriOrchestrator,
+    mock_session: MagicMock,
+) -> None:
+    # Act / Assert
+    with pytest.raises(JobNoNotYetRecordedError):
+        orchestrator.amend_job(CHANGE)
+    assert mock_session.post.call_count == 0
+
+
+# --- rejection passthrough: OpenHousing's actual reason, verbatim ---
+
+
+def test_amend_job_returns_the_openhousing_rejection_verbatim(
+    orchestrator: AbriOrchestrator,
+    mock_session: MagicMock,
+    deal_database: FakeDealDatabase,
+) -> None:
+    # Arrange
+    deal_database.job_nos[DEAL_ID] = JOB_NO
+    mock_session.post.return_value.content = _load_fixture(
+        "abri_relay_amendoptiappt_relayerror_response.xml"
+    )
+
+    # Act
+    result = orchestrator.amend_job(CHANGE)
+
+    # Assert
+    assert isinstance(result, AbriRequestRejected)
+    assert result.code == "RelayError"
+    assert result.message.startswith("Job_no 1019905 is invalid")

tests/orchestration/test_abri_orchestrator_log_job.py

@@ -3,7 +3,7 @@ import traceback
 import xml.etree.ElementTree as ET
 from datetime import date
 from pathlib import Path
-from typing import List, Tuple
+from typing import List, Optional, Tuple
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -57,6 +57,16 @@ class FakeDealDatabase:
     def record_job_no(self, deal_id: str, job_no: str) -> None:
         self.recorded_job_nos.append((deal_id, job_no))
 
+    def job_no_for_deal(self, deal_id: str) -> Optional[str]:
+        return next(
+            (
+                job_no
+                for recorded_deal_id, job_no in self.recorded_job_nos
+                if recorded_deal_id == deal_id
+            ),
+            None,
+        )
+
 
 @pytest.fixture()
 def mock_session() -> MagicMock:

tests/orchestration/test_abri_orchestrator_tenant_data_sync.py

@@ -2,6 +2,7 @@ import json
 import traceback
 import xml.etree.ElementTree as ET
 from pathlib import Path
+from typing import Optional
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -66,6 +67,9 @@ class FakeDealDatabase:
     def record_job_no(self, deal_id: str, job_no: str) -> None:
         raise AssertionError("tenant-data sync must not touch the deal database")
 
+    def job_no_for_deal(self, deal_id: str) -> Optional[str]:
+        raise AssertionError("tenant-data sync must not touch the deal database")
+
 
 @pytest.fixture()
 def orchestrator(

tests/repositories/hubspot_deals/test_deal_database_postgres_gateway.py

@@ -0,0 +1,115 @@
+from collections.abc import Iterator
+
+import pytest
+from sqlalchemy import Engine
+from sqlmodel import Session, text
+
+from backend.app.db.models.hubspot_deal_data import HubspotDealData
+from repositories.hubspot_deals.deal_database_postgres_gateway import (
+    DealDatabasePostgresGateway,
+)
+
+DEAL_ID = "9876543210"
+
+
+@pytest.fixture
+def session(db_engine: Engine) -> Iterator[Session]:
+    with Session(db_engine) as s:
+        yield s
+
+
+def _insert_deal(session: Session, deal_id: str) -> HubspotDealData:
+    deal = HubspotDealData(deal_id=deal_id, dealname="49 Admers Crescent")
+    session.add(deal)
+    session.commit()
+    session.refresh(deal)
+    return deal
+
+
+def test_record_job_no_persists_the_job_no_on_the_deal_row(session: Session) -> None:
+    # Arrange
+    deal = _insert_deal(session, DEAL_ID)
+    gateway = DealDatabasePostgresGateway(session=session)
+
+    # Act
+    gateway.record_job_no(deal_id=DEAL_ID, job_no="AD0226519")
+
+    # Assert
+    session.refresh(deal)
+    assert deal.client_booking_reference == "AD0226519"
+
+
+def test_record_job_no_for_an_unknown_deal_raises(session: Session) -> None:
+    # Arrange
+    gateway = DealDatabasePostgresGateway(session=session)
+
+    # Act / Assert
+    with pytest.raises(ValueError):
+        gateway.record_job_no(deal_id="0000000000", job_no="AD0226519")
+
+
+def test_the_job_no_is_stored_in_the_client_booking_reference_column(
+    session: Session,
+) -> None:
+    # The physical column matches the HubSpot property the job number is
+    # stored under (client_booking_reference); only the domain calls it
+    # job_no. The deployed table is migrated by the frontend repo, so the
+    # column name here must match what that migration created.
+    # Arrange
+    _insert_deal(session, DEAL_ID)
+    gateway = DealDatabasePostgresGateway(session=session)
+
+    # Act
+    gateway.record_job_no(deal_id=DEAL_ID, job_no="AD0226519")
+
+    # Assert
+    stored = (
+        session.connection()
+        .execute(
+            text(
+                "select client_booking_reference from hubspot_deal_data "
+                "where deal_id = :deal_id"
+            ),
+            {"deal_id": DEAL_ID},
+        )
+        .one()
+    )
+    assert stored[0] == "AD0226519"
+
+
+def test_job_no_for_deal_returns_the_recorded_job_no(session: Session) -> None:
+    # Arrange
+    _insert_deal(session, DEAL_ID)
+    gateway = DealDatabasePostgresGateway(session=session)
+    gateway.record_job_no(deal_id=DEAL_ID, job_no="AD0226519")
+
+    # Act
+    job_no = gateway.job_no_for_deal(DEAL_ID)
+
+    # Assert
+    assert job_no == "AD0226519"
+
+
+def test_job_no_for_deal_returns_none_before_any_job_is_recorded(
+    session: Session,
+) -> None:
+    # Arrange
+    _insert_deal(session, DEAL_ID)
+    gateway = DealDatabasePostgresGateway(session=session)
+
+    # Act
+    job_no = gateway.job_no_for_deal(DEAL_ID)
+
+    # Assert
+    assert job_no is None
+
+
+def test_job_no_for_deal_returns_none_for_an_unknown_deal(session: Session) -> None:
+    # Arrange
+    gateway = DealDatabasePostgresGateway(session=session)
+
+    # Act
+    job_no = gateway.job_no_for_deal("0000000000")
+
+    # Assert
+    assert job_no is None

tests/utilities/aws_lambda/test_task_handler.py

@@ -8,12 +8,13 @@ import pytest
 from sqlalchemy import Engine, text
 from sqlmodel import Session
 
+from domain.tasks.subtasks import SubTaskStatus
 from domain.tasks.subtasks import SubTaskFailure
 from domain.tasks.tasks import Source
 from orchestration.task_orchestrator import TaskOrchestrator
 from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository
 from repositories.tasks.task_postgres_repository import TaskPostgresRepository
-from utilities.aws_lambda.task_handler import task_handler
+from utilities.aws_lambda.task_handler import NonRetriableTaskError, task_handler
 
 
 @dataclass
@@ -48,12 +49,8 @@ def test_task_handler_records_cloudwatch_url_on_subtask(
 ) -> None:
     # arrange
     monkeypatch.setenv("AWS_REGION", "eu-west-2")
-    monkeypatch.setenv(
-        "AWS_LAMBDA_LOG_GROUP_NAME", "/aws/lambda/modelling-e2e"
-    )
-    monkeypatch.setenv(
-        "AWS_LAMBDA_LOG_STREAM_NAME", "2026/05/20/[$LATEST]abc123"
-    )
+    monkeypatch.setenv("AWS_LAMBDA_LOG_GROUP_NAME", "/aws/lambda/modelling-e2e")
+    monkeypatch.setenv("AWS_LAMBDA_LOG_STREAM_NAME", "2026/05/20/[$LATEST]abc123")
 
     @task_handler(
         task_source="modelling_e2e",
@@ -91,7 +88,10 @@ def test_task_handler_passes_orchestrator_and_task_id_when_flag_is_true(
         pass_task_orchestrator=True,
     )
     def _handler(
-        body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, task_id: UUID
+        body: dict[str, Any],
+        context: Any,
+        orchestrator: TaskOrchestrator,
+        task_id: UUID,
     ) -> None:
         received.append((orchestrator, task_id))
 
@@ -105,6 +105,54 @@ def test_task_handler_passes_orchestrator_and_task_id_when_flag_is_true(
     assert recv_task_id == UUID(result[0]["task_id"])
 
 
+def test_task_handler_reports_an_ordinarily_failing_record_for_redelivery(
+    harness: Harness,
+) -> None:
+    # Arrange
+    @task_handler(
+        task_source="abri_api",
+        source=Source.HUBSPOT_DEAL,
+        orchestrator_cm=harness.factory,
+    )
+    def handler(body: dict[str, Any], context: Any) -> None:
+        raise RuntimeError("transient failure")
+
+    event = {"Records": [{"messageId": "msg-1", "body": '{"hubspot_deal_id": "123"}'}]}
+
+    # Act
+    result = handler(event, context=None)
+
+    # Assert
+    assert result["batchItemFailures"] == [{"itemIdentifier": "msg-1"}]
+    subtask_id = result["tasks"][0]["subtask_id"]
+    assert harness.subtasks.get(UUID(subtask_id)).status is SubTaskStatus.FAILED
+
+
+def test_task_handler_does_not_requeue_a_record_failing_non_retriably(
+    harness: Harness,
+) -> None:
+    # Arrange
+    @task_handler(
+        task_source="abri_api",
+        source=Source.HUBSPOT_DEAL,
+        orchestrator_cm=harness.factory,
+    )
+    def handler(body: dict[str, Any], context: Any) -> None:
+        raise NonRetriableTaskError("job logged but write-back failed")
+
+    event = {"Records": [{"messageId": "msg-1", "body": '{"hubspot_deal_id": "123"}'}]}
+
+    # Act
+    result = handler(event, context=None)
+
+    # Assert: the failure is recorded, but the message is not requeued.
+    assert result["batchItemFailures"] == []
+    subtask_id = result["tasks"][0]["subtask_id"]
+    failed = harness.subtasks.get(UUID(subtask_id))
+    assert failed.status is SubTaskStatus.FAILED
+    assert failed.outputs == {"error": "job logged but write-back failed"}
+
+
 def test_task_handler_attaches_to_a_supplied_task_and_subtask(
     harness: Harness, db_engine: Engine
 ) -> None:
@@ -124,7 +172,10 @@ def test_task_handler_attaches_to_a_supplied_task_and_subtask(
         pass_task_orchestrator=True,
     )
     def handler(
-        body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, task_id: UUID
+        body: dict[str, Any],
+        context: Any,
+        orchestrator: TaskOrchestrator,
+        task_id: UUID,
     ) -> None:
         received.append(task_id)
 
@@ -145,9 +196,7 @@ def test_task_handler_attaches_to_a_supplied_task_and_subtask(
 
     # assert — ran under the supplied ids, sub_task completed, nothing created
     assert received == [task.id]
-    assert result["tasks"] == [
-        {"task_id": str(task.id), "subtask_id": str(subtask.id)}
-    ]
+    assert result["tasks"] == [{"task_id": str(task.id), "subtask_id": str(subtask.id)}]
     assert harness.subtasks.get(subtask.id).status.value == "complete"
     with db_engine.connect() as conn:
         assert conn.execute(text("SELECT count(*) FROM tasks")).scalar_one() == 1

utilities/aws_lambda/task_handler.py

@@ -22,6 +22,17 @@ logger = logging.getLogger(__name__)
 OrchestratorCM = Callable[[], AbstractContextManager[TaskOrchestrator]]
 
 
+class NonRetriableTaskError(Exception):
+    """A failure that must not be retried by SQS redelivery.
+
+    The SubTask is marked failed (the failure record is the visibility
+    surface) but the message is not returned to the queue, so the work is
+    never re-run. Raise it for failures where a retry could duplicate an
+    irreversible side effect — e.g. a job already logged in OpenHousing
+    whose HubSpot write-back failed.
+    """
+
+
 def task_handler(
     *,
     task_source: str,
@@ -57,6 +68,7 @@ def task_handler(
                     # app-owned task_id and its distributor-pre-created
                     # subtask_id — run under those; create nothing.
                     supplied = _supplied_ids(body)
+                    source_id: Optional[str] = None
                     if supplied is not None:
                         task_id, subtask_id = supplied
                     else:
@@ -88,6 +100,17 @@ def task_handler(
                                 work=lambda: func(body, context),
                                 cloud_logs_url=cloud_logs_url,
                             )
+                    except NonRetriableTaskError:
+                        # Failed, but never redelivered: the SubTask's failure
+                        # record is the only follow-up surface.
+                        logger.exception(
+                            "subtask failed non-retriably "
+                            "(task_source=%s source_id=%s)",
+                            task_source,
+                            source_id,
+                        )
+                        if "Records" not in event:
+                            raise
                     except SubTaskFailure as recorded:
                         # A recorded failure (ADR-0055): run_subtask has already
                         # failed the SubTask with the structured details — that