From a477561bbc0b7bef41bee12b96df1efa64365261 Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 13:03:04 +0000
Subject: [PATCH 01/20] correct tfstate bucket name

---
 infrastructure/terraform/lambda/hubspot_deal_etl/main.tf | 2 +-
 infrastructure/terraform/lambda/magic_plan/provider.tf   | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/infrastructure/terraform/lambda/hubspot_deal_etl/main.tf b/infrastructure/terraform/lambda/hubspot_deal_etl/main.tf
index 800dc3b6..ffb5f6f5 100644
--- a/infrastructure/terraform/lambda/hubspot_deal_etl/main.tf
+++ b/infrastructure/terraform/lambda/hubspot_deal_etl/main.tf
@@ -19,7 +19,7 @@ data "terraform_remote_state" "pashub_to_ara" {
 data "terraform_remote_state" "magic_plan" {
   backend = "s3"
   config = {
-    bucket = "magic-plan-hubspot-trigger-terraform-state"
+    bucket = "magic-plan-client-terraform-state"
     key    = "env:/${var.stage}/terraform.tfstate"
     region = "eu-west-2"
   }
diff --git a/infrastructure/terraform/lambda/magic_plan/provider.tf b/infrastructure/terraform/lambda/magic_plan/provider.tf
index 9e7020ac..a3dd6a7d 100644
--- a/infrastructure/terraform/lambda/magic_plan/provider.tf
+++ b/infrastructure/terraform/lambda/magic_plan/provider.tf
@@ -7,7 +7,7 @@ terraform {
   }
 
   backend "s3" {
-    bucket = "magic-plan-hubspot-trigger-terraform-state"
+    bucket = "magic-plan-client-terraform-state"
     key    = "terraform.tfstate"
     region = "eu-west-2"
   }

From a672c0dea0dbfb54a35d4d5deeefea7303c93193 Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 13:51:46 +0000
Subject: [PATCH 02/20] add localhandler for testing and update requirements

---
 backend/magic_plan/handler/requirements.txt   |  4 +++
 .../local_handler/docker-compose.yml          | 11 ++++++++
 .../local_handler/invoke_local_lambda.py      | 28 +++++++++++++++++++
 3 files changed, 43 insertions(+)
 create mode 100644 backend/magic_plan/local_handler/docker-compose.yml
 create mode 100644 backend/magic_plan/local_handler/invoke_local_lambda.py

diff --git a/backend/magic_plan/handler/requirements.txt b/backend/magic_plan/handler/requirements.txt
index cfacf455..29123caa 100644
--- a/backend/magic_plan/handler/requirements.txt
+++ b/backend/magic_plan/handler/requirements.txt
@@ -5,3 +5,7 @@ sqlmodel
 psycopg2-binary==2.9.10
 pydantic-settings==2.6.0
 boto3==1.35.44
+
+pytz==2024.2
+pandas==2.2.2
+numpy==2.1.2
diff --git a/backend/magic_plan/local_handler/docker-compose.yml b/backend/magic_plan/local_handler/docker-compose.yml
new file mode 100644
index 00000000..5a42d259
--- /dev/null
+++ b/backend/magic_plan/local_handler/docker-compose.yml
@@ -0,0 +1,11 @@
+version: "3.9"
+
+services:
+  ecmk-fetcher-lambda:
+    build:
+      context: ../../../
+      dockerfile: backend/magic_plan/handler/Dockerfile
+    ports:
+      - "9000:8080"
+    env_file:
+      - ../../../.env
\ No newline at end of file
diff --git a/backend/magic_plan/local_handler/invoke_local_lambda.py b/backend/magic_plan/local_handler/invoke_local_lambda.py
new file mode 100644
index 00000000..7bb65806
--- /dev/null
+++ b/backend/magic_plan/local_handler/invoke_local_lambda.py
@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+import json
+import requests
+
+HOST = "localhost"
+PORT = "9000"
+
+LAMBDA_URL = f"http://{HOST}:{PORT}/2015-03-31/functions/function/invocations"
+
+payload = {
+    "Records": [
+        {
+            "body": json.dumps(
+                # {
+                #     "address": "2 Laburnum Way, Rombley, BR2 8BZ | Retrofit Assessment",
+                #     "hubspot_deal_id": "500262906061",
+                # }
+                {"address": "33 Wallaby Way, Sydney", "hubspot_deal_id": "123456789"}
+            )
+        }
+    ]
+}
+
+response = requests.post(LAMBDA_URL, json=payload)
+
+print("Status code:", response.status_code)
+print("Response:")
+print(response.text)

From da4f5f44c0ba871f05cf7da3cc2f070a5398b496 Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 13:58:16 +0000
Subject: [PATCH 03/20] =?UTF-8?q?Set=20API=20key=20as=20session=20header?=
 =?UTF-8?q?=20on=20MagicPlanClient=20construction=20=F0=9F=9F=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/magic_plan/tests/test_magic_plan_client.py | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/backend/magic_plan/tests/test_magic_plan_client.py b/backend/magic_plan/tests/test_magic_plan_client.py
index c96b9cdf..cb2385b1 100644
--- a/backend/magic_plan/tests/test_magic_plan_client.py
+++ b/backend/magic_plan/tests/test_magic_plan_client.py
@@ -20,6 +20,7 @@ def _load_fixture(name: str) -> dict[str, Any]:
 
 
 def _make_client(mock_session: MagicMock) -> MagicPlanClient:
+    mock_session.headers = {}
     with patch(
         "backend.magic_plan.magic_plan_client.requests.Session",
         return_value=mock_session,
@@ -44,7 +45,14 @@ def test_customer_header_set_on_session(mock_session: MagicMock) -> None:
     # Act
     _make_client(mock_session)
     # Assert
-    mock_session.headers.update.assert_called_once_with({"customer": CUSTOMER_ID})
+    assert mock_session.headers["customer"] == CUSTOMER_ID
+
+
+def test_api_key_header_set_on_session(mock_session: MagicMock) -> None:
+    # Act
+    _make_client(mock_session)
+    # Assert
+    assert mock_session.headers["key"] == API_KEY
 
 
 # --- get_plans ---

From d59bf2d7cbf35299128099345a4334d1dc372c94 Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 13:59:33 +0000
Subject: [PATCH 04/20] =?UTF-8?q?Set=20API=20key=20as=20session=20header?=
 =?UTF-8?q?=20on=20MagicPlanClient=20construction=20=F0=9F=9F=A9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/magic_plan/magic_plan_client.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/backend/magic_plan/magic_plan_client.py b/backend/magic_plan/magic_plan_client.py
index 06905e6a..4029c436 100644
--- a/backend/magic_plan/magic_plan_client.py
+++ b/backend/magic_plan/magic_plan_client.py
@@ -9,7 +9,7 @@ class MagicPlanClient:
     def __init__(self, customer_id: str, api_key: str) -> None:
         self._api_key = api_key
         self._session = requests.Session()
-        self._session.headers.update({"customer": customer_id})
+        self._session.headers.update({"customer": customer_id, "key": api_key})
 
     def get_plans(self) -> PlansListResponse:
         r = self._session.get(f"{_BASE_URL}/plans", params={"key": self._api_key})

From ffcff33dd4434efe98bd2ca04d683b726eb453ba Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 14:00:07 +0000
Subject: [PATCH 05/20] =?UTF-8?q?get=5Fplans()=20sends=20no=20API=20key=20?=
 =?UTF-8?q?query=20parameter=20=F0=9F=9F=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/magic_plan/tests/test_magic_plan_client.py | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/backend/magic_plan/tests/test_magic_plan_client.py b/backend/magic_plan/tests/test_magic_plan_client.py
index cb2385b1..8b0b3f71 100644
--- a/backend/magic_plan/tests/test_magic_plan_client.py
+++ b/backend/magic_plan/tests/test_magic_plan_client.py
@@ -70,9 +70,7 @@ def test_get_plans_calls_correct_url(
     # Act
     client.get_plans()
     # Assert
-    mock_session.get.assert_called_once_with(
-        f"{BASE_URL}/plans", params={"key": API_KEY}
-    )
+    mock_session.get.assert_called_once_with(f"{BASE_URL}/plans")
 
 
 def test_get_plans_calls_raise_for_status(

From 20b32bcda03679d7d0c28f82d53621e4fb092af2 Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 14:01:35 +0000
Subject: [PATCH 06/20] =?UTF-8?q?get=5Fplans()=20sends=20no=20API=20key=20?=
 =?UTF-8?q?query=20parameter=20=F0=9F=9F=A9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/magic_plan/magic_plan_client.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/backend/magic_plan/magic_plan_client.py b/backend/magic_plan/magic_plan_client.py
index 4029c436..bed3dc9c 100644
--- a/backend/magic_plan/magic_plan_client.py
+++ b/backend/magic_plan/magic_plan_client.py
@@ -12,7 +12,7 @@ class MagicPlanClient:
         self._session.headers.update({"customer": customer_id, "key": api_key})
 
     def get_plans(self) -> PlansListResponse:
-        r = self._session.get(f"{_BASE_URL}/plans", params={"key": self._api_key})
+        r = self._session.get(f"{_BASE_URL}/plans")
         r.raise_for_status()
         return PlansListResponse.model_validate(r.json()["data"])
 

From 7752039dbdb5244d12a974e2ce3a17728d25293d Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 14:01:40 +0000
Subject: [PATCH 07/20] =?UTF-8?q?=5Ffetch=5Fplan()=20sends=20no=20API=20ke?=
 =?UTF-8?q?y=20query=20parameter=20=F0=9F=9F=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/magic_plan/tests/test_magic_plan_client.py | 8 ++------
 1 file changed, 2 insertions(+), 6 deletions(-)

diff --git a/backend/magic_plan/tests/test_magic_plan_client.py b/backend/magic_plan/tests/test_magic_plan_client.py
index 8b0b3f71..a0827bee 100644
--- a/backend/magic_plan/tests/test_magic_plan_client.py
+++ b/backend/magic_plan/tests/test_magic_plan_client.py
@@ -132,9 +132,7 @@ def test_get_plan_calls_correct_url(
     # Act
     client.get_plan(plan_id)
     # Assert
-    mock_session.get.assert_called_once_with(
-        f"{BASE_URL}/plans/{plan_id}", params={"key": API_KEY}
-    )
+    mock_session.get.assert_called_once_with(f"{BASE_URL}/plans/{plan_id}")
 
 
 def test_get_plan_calls_raise_for_status(
@@ -204,9 +202,7 @@ def test_get_plan_raw_calls_correct_url(
     # Act
     client.get_plan_raw(plan_id)
     # Assert
-    mock_session.get.assert_called_once_with(
-        f"{BASE_URL}/plans/{plan_id}", params={"key": API_KEY}
-    )
+    mock_session.get.assert_called_once_with(f"{BASE_URL}/plans/{plan_id}")
 
 
 def test_get_plan_raw_calls_raise_for_status(

From eb381a778c2dc8c2ff2341ff14a4376463e4b5d8 Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 14:02:17 +0000
Subject: [PATCH 08/20] =?UTF-8?q?=5Ffetch=5Fplan()=20sends=20no=20API=20ke?=
 =?UTF-8?q?y=20query=20parameter=20=F0=9F=9F=A9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/magic_plan/magic_plan_client.py | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/backend/magic_plan/magic_plan_client.py b/backend/magic_plan/magic_plan_client.py
index bed3dc9c..f9ae030f 100644
--- a/backend/magic_plan/magic_plan_client.py
+++ b/backend/magic_plan/magic_plan_client.py
@@ -23,8 +23,6 @@ class MagicPlanClient:
         return self._fetch_plan(plan_id).content
 
     def _fetch_plan(self, plan_id: str) -> requests.Response:
-        r = self._session.get(
-            f"{_BASE_URL}/plans/{plan_id}", params={"key": self._api_key}
-        )
+        r = self._session.get(f"{_BASE_URL}/plans/{plan_id}")
         r.raise_for_status()
         return r

From 3df726937e43c3c2758266144418740ae7636238 Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 14:03:07 +0000
Subject: [PATCH 09/20] =?UTF-8?q?Remove=20unused=20=5Fapi=5Fkey=20instance?=
 =?UTF-8?q?=20variable=20now=20auth=20is=20fully=20header-based=20?=
 =?UTF-8?q?=F0=9F=9F=AA?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/magic_plan/magic_plan_client.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/backend/magic_plan/magic_plan_client.py b/backend/magic_plan/magic_plan_client.py
index f9ae030f..2880bf43 100644
--- a/backend/magic_plan/magic_plan_client.py
+++ b/backend/magic_plan/magic_plan_client.py
@@ -7,7 +7,6 @@ _BASE_URL = "https://cloud.magicplan.app/api/v2"
 
 class MagicPlanClient:
     def __init__(self, customer_id: str, api_key: str) -> None:
-        self._api_key = api_key
         self._session = requests.Session()
         self._session.headers.update({"customer": customer_id, "key": api_key})
 

From 04df924146e336688bd4110111625b3738e3ac21 Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 14:13:12 +0000
Subject: [PATCH 10/20] fix local invoker

---
 backend/magic_plan/handler.py                           | 5 +++--
 backend/magic_plan/local_handler/invoke_local_lambda.py | 3 ++-
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/backend/magic_plan/handler.py b/backend/magic_plan/handler.py
index f2c03b90..5fd90b7a 100644
--- a/backend/magic_plan/handler.py
+++ b/backend/magic_plan/handler.py
@@ -20,7 +20,9 @@ def handler(body: dict[str, Any], context: Any) -> str:
         api_key=settings.MAGICPLAN_API_KEY,
     )
     # TODO: read s3_bucket from env var so staging/prod use the correct bucket
-    plan: Plan = MagicPlanService(client, s3_bucket="retrofit-energy-assessments-dev").run(payload)
+    plan: Plan = MagicPlanService(
+        client, s3_bucket="retrofit-energy-assessments-dev"
+    ).run(payload)
     logger.info("Saved MagicPlan plan uid=%s", plan.uid)
     return plan.uid
 
@@ -30,7 +32,6 @@ if __name__ == "__main__":
         "Records": [
             {
                 "body": '{"address": "2 Laburnum Way Bromley BR2 8BZ", "hubspot_deal_id": "local-test-deal"}',
-                "messageId": "local-test",
             }
         ]
     }
diff --git a/backend/magic_plan/local_handler/invoke_local_lambda.py b/backend/magic_plan/local_handler/invoke_local_lambda.py
index 7bb65806..146951fe 100644
--- a/backend/magic_plan/local_handler/invoke_local_lambda.py
+++ b/backend/magic_plan/local_handler/invoke_local_lambda.py
@@ -10,13 +10,14 @@ LAMBDA_URL = f"http://{HOST}:{PORT}/2015-03-31/functions/function/invocations"
 payload = {
     "Records": [
         {
+            "messageId": "test-message-id",
             "body": json.dumps(
                 # {
                 #     "address": "2 Laburnum Way, Rombley, BR2 8BZ | Retrofit Assessment",
                 #     "hubspot_deal_id": "500262906061",
                 # }
                 {"address": "33 Wallaby Way, Sydney", "hubspot_deal_id": "123456789"}
-            )
+            ),
         }
     ]
 }

From 75d03139341444ab5694793634f014c903fa6e3a Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 14:14:37 +0000
Subject: [PATCH 11/20] fix broken magicplan handler tests

---
 backend/magic_plan/tests/test_handler.py | 18 ++++++++++++------
 1 file changed, 12 insertions(+), 6 deletions(-)

diff --git a/backend/magic_plan/tests/test_handler.py b/backend/magic_plan/tests/test_handler.py
index 366f3ded..b0365f5b 100644
--- a/backend/magic_plan/tests/test_handler.py
+++ b/backend/magic_plan/tests/test_handler.py
@@ -54,7 +54,7 @@ def test_handler_raises_on_missing_address(mock_plan: MagicMock) -> None:
 
 def test_handler_constructs_client_from_settings(mock_service: MagicMock) -> None:
     # Arrange
-    body = {"address": ADDRESS}
+    body = {"address": ADDRESS, "hubspot_deal_id": "deal-123"}
     with patch("backend.magic_plan.handler.get_settings", return_value=_make_settings(customer_id="cust-xyz", api_key="key-xyz")), \
          patch("backend.magic_plan.handler.MagicPlanClient") as MockClient, \
          patch("backend.magic_plan.handler.MagicPlanService", return_value=mock_service):
@@ -69,31 +69,37 @@ def test_handler_constructs_client_from_settings(mock_service: MagicMock) -> Non
 
 def test_handler_calls_service_run_with_address(mock_service: MagicMock) -> None:
     # Arrange
-    body = {"address": ADDRESS}
+    body = {"address": ADDRESS, "hubspot_deal_id": "deal-123"}
     with patch("backend.magic_plan.handler.get_settings", return_value=_make_settings()), \
          patch("backend.magic_plan.handler.MagicPlanClient"), \
          patch("backend.magic_plan.handler.MagicPlanService", return_value=mock_service):
         # Act
         _call_handler(body)
     # Assert
-    mock_service.run.assert_called_once_with(ADDRESS, None)
+    mock_service.run.assert_called_once()
+    request = mock_service.run.call_args.args[0]
+    assert request.address == ADDRESS
+    assert request.uprn is None
 
 
 def test_handler_passes_uprn_to_service(mock_service: MagicMock) -> None:
     # Arrange
-    body = {"address": ADDRESS, "uprn": "100023336956"}
+    body = {"address": ADDRESS, "uprn": "100023336956", "hubspot_deal_id": "deal-123"}
     with patch("backend.magic_plan.handler.get_settings", return_value=_make_settings()), \
          patch("backend.magic_plan.handler.MagicPlanClient"), \
          patch("backend.magic_plan.handler.MagicPlanService", return_value=mock_service):
         # Act
         _call_handler(body)
     # Assert
-    mock_service.run.assert_called_once_with(ADDRESS, "100023336956")
+    mock_service.run.assert_called_once()
+    request = mock_service.run.call_args.args[0]
+    assert request.address == ADDRESS
+    assert request.uprn == "100023336956"
 
 
 def test_handler_returns_plan_uid(mock_service: MagicMock) -> None:
     # Arrange
-    body = {"address": ADDRESS}
+    body = {"address": ADDRESS, "hubspot_deal_id": "deal-123"}
     with patch("backend.magic_plan.handler.get_settings", return_value=_make_settings()), \
          patch("backend.magic_plan.handler.MagicPlanClient"), \
          patch("backend.magic_plan.handler.MagicPlanService", return_value=mock_service):

From beaf21fdcc4786dcf0b500dd37605e102eaa543f Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 14:32:37 +0000
Subject: [PATCH 12/20] correct magic plan url paths

---
 backend/magic_plan/magic_plan_client.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/backend/magic_plan/magic_plan_client.py b/backend/magic_plan/magic_plan_client.py
index 2880bf43..34c40695 100644
--- a/backend/magic_plan/magic_plan_client.py
+++ b/backend/magic_plan/magic_plan_client.py
@@ -11,7 +11,7 @@ class MagicPlanClient:
         self._session.headers.update({"customer": customer_id, "key": api_key})
 
     def get_plans(self) -> PlansListResponse:
-        r = self._session.get(f"{_BASE_URL}/plans")
+        r = self._session.get(f"{_BASE_URL}/workgroups/plans")
         r.raise_for_status()
         return PlansListResponse.model_validate(r.json()["data"])
 
@@ -22,6 +22,6 @@ class MagicPlanClient:
         return self._fetch_plan(plan_id).content
 
     def _fetch_plan(self, plan_id: str) -> requests.Response:
-        r = self._session.get(f"{_BASE_URL}/plans/{plan_id}")
+        r = self._session.get(f"{_BASE_URL}/plans/get/{plan_id}")
         r.raise_for_status()
         return r

From 8727a78f8bb0719061cc075ba49dc264ea90f9cd Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 14:33:58 +0000
Subject: [PATCH 13/20] =?UTF-8?q?correct=20magic=20plan=20url=20paths=20?=
 =?UTF-8?q?=F0=9F=9F=A9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/magic_plan/tests/test_magic_plan_client.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/backend/magic_plan/tests/test_magic_plan_client.py b/backend/magic_plan/tests/test_magic_plan_client.py
index a0827bee..27d4ebad 100644
--- a/backend/magic_plan/tests/test_magic_plan_client.py
+++ b/backend/magic_plan/tests/test_magic_plan_client.py
@@ -70,7 +70,7 @@ def test_get_plans_calls_correct_url(
     # Act
     client.get_plans()
     # Assert
-    mock_session.get.assert_called_once_with(f"{BASE_URL}/plans")
+    mock_session.get.assert_called_once_with(f"{BASE_URL}/workgroups/plans")
 
 
 def test_get_plans_calls_raise_for_status(
@@ -132,7 +132,7 @@ def test_get_plan_calls_correct_url(
     # Act
     client.get_plan(plan_id)
     # Assert
-    mock_session.get.assert_called_once_with(f"{BASE_URL}/plans/{plan_id}")
+    mock_session.get.assert_called_once_with(f"{BASE_URL}/plans/get/{plan_id}")
 
 
 def test_get_plan_calls_raise_for_status(
@@ -202,7 +202,7 @@ def test_get_plan_raw_calls_correct_url(
     # Act
     client.get_plan_raw(plan_id)
     # Assert
-    mock_session.get.assert_called_once_with(f"{BASE_URL}/plans/{plan_id}")
+    mock_session.get.assert_called_once_with(f"{BASE_URL}/plans/get/{plan_id}")
 
 
 def test_get_plan_raw_calls_raise_for_status(

From 62acc3ce98cb717a4c65711d3028e526443bbdfc Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 14:45:09 +0000
Subject: [PATCH 14/20] =?UTF-8?q?Paginate=20get=5Fplans=20to=20return=20fl?=
 =?UTF-8?q?at=20list[PlanSummary]=20=F0=9F=9F=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/magic_plan/magic_plan_client.py            |  8 +++-----
 backend/magic_plan/tests/test_magic_plan_client.py | 13 ++++++++-----
 2 files changed, 11 insertions(+), 10 deletions(-)

diff --git a/backend/magic_plan/magic_plan_client.py b/backend/magic_plan/magic_plan_client.py
index 34c40695..ee52ffb0 100644
--- a/backend/magic_plan/magic_plan_client.py
+++ b/backend/magic_plan/magic_plan_client.py
@@ -1,6 +1,6 @@
 import requests
 
-from datatypes.magicplan.api.response import MagicPlanPlan, PlansListResponse
+from datatypes.magicplan.api.response import MagicPlanPlan, PlanSummary, PlansListResponse
 
 _BASE_URL = "https://cloud.magicplan.app/api/v2"
 
@@ -10,10 +10,8 @@ class MagicPlanClient:
         self._session = requests.Session()
         self._session.headers.update({"customer": customer_id, "key": api_key})
 
-    def get_plans(self) -> PlansListResponse:
-        r = self._session.get(f"{_BASE_URL}/workgroups/plans")
-        r.raise_for_status()
-        return PlansListResponse.model_validate(r.json()["data"])
+    def get_plans(self) -> list[PlanSummary]:
+        raise NotImplementedError
 
     def get_plan(self, plan_id: str) -> MagicPlanPlan:
         return MagicPlanPlan.model_validate(self._fetch_plan(plan_id).json()["data"])
diff --git a/backend/magic_plan/tests/test_magic_plan_client.py b/backend/magic_plan/tests/test_magic_plan_client.py
index 27d4ebad..bf078517 100644
--- a/backend/magic_plan/tests/test_magic_plan_client.py
+++ b/backend/magic_plan/tests/test_magic_plan_client.py
@@ -7,7 +7,7 @@ import pytest
 import requests
 
 from backend.magic_plan.magic_plan_client import MagicPlanClient
-from datatypes.magicplan.api.response import MagicPlanPlan, PlansListResponse
+from datatypes.magicplan.api.response import MagicPlanPlan, PlanSummary
 
 FIXTURE_DIR = Path(__file__).parents[2] / "magic_plan"
 BASE_URL = "https://cloud.magicplan.app/api/v2"
@@ -70,7 +70,9 @@ def test_get_plans_calls_correct_url(
     # Act
     client.get_plans()
     # Assert
-    mock_session.get.assert_called_once_with(f"{BASE_URL}/workgroups/plans")
+    mock_session.get.assert_called_once_with(
+        f"{BASE_URL}/workgroups/plans", params={"page": 1}
+    )
 
 
 def test_get_plans_calls_raise_for_status(
@@ -88,7 +90,7 @@ def test_get_plans_calls_raise_for_status(
     mock_session.get.return_value.raise_for_status.assert_called_once()
 
 
-def test_get_plans_returns_plans_list_response(
+def test_get_plans_returns_list_of_plan_summaries(
     client: MagicPlanClient, mock_session: MagicMock
 ) -> None:
     # Arrange
@@ -100,8 +102,9 @@ def test_get_plans_returns_plans_list_response(
     # Act
     result = client.get_plans()
     # Assert
-    assert isinstance(result, PlansListResponse)
-    assert len(result.plans) == 1
+    assert isinstance(result, list)
+    assert len(result) == 1
+    assert isinstance(result[0], PlanSummary)
 
 
 def test_get_plans_propagates_http_error(

From f83ddd05a8a6a8bace716e0e449c95bf040b1527 Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 14:46:00 +0000
Subject: [PATCH 15/20] =?UTF-8?q?Paginate=20get=5Fplans=20to=20return=20fl?=
 =?UTF-8?q?at=20list[PlanSummary]=20=F0=9F=9F=A9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/magic_plan/magic_plan_client.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/backend/magic_plan/magic_plan_client.py b/backend/magic_plan/magic_plan_client.py
index ee52ffb0..de2fe4f6 100644
--- a/backend/magic_plan/magic_plan_client.py
+++ b/backend/magic_plan/magic_plan_client.py
@@ -11,7 +11,9 @@ class MagicPlanClient:
         self._session.headers.update({"customer": customer_id, "key": api_key})
 
     def get_plans(self) -> list[PlanSummary]:
-        raise NotImplementedError
+        r = self._session.get(f"{_BASE_URL}/workgroups/plans", params={"page": 1})
+        r.raise_for_status()
+        return PlansListResponse.model_validate(r.json()["data"]).plans
 
     def get_plan(self, plan_id: str) -> MagicPlanPlan:
         return MagicPlanPlan.model_validate(self._fetch_plan(plan_id).json()["data"])

From 6dfca082f8e9619c9a9c31dfd1c3524dc03be530 Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 14:52:31 +0000
Subject: [PATCH 16/20] =?UTF-8?q?Fetch=20all=20pages=20in=20get=5Fplans=20?=
 =?UTF-8?q?pagination=20loop=20=F0=9F=9F=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../tests/test_magic_plan_client.py           | 28 +++++++++++++++++++
 1 file changed, 28 insertions(+)

diff --git a/backend/magic_plan/tests/test_magic_plan_client.py b/backend/magic_plan/tests/test_magic_plan_client.py
index bf078517..211a5d4d 100644
--- a/backend/magic_plan/tests/test_magic_plan_client.py
+++ b/backend/magic_plan/tests/test_magic_plan_client.py
@@ -119,6 +119,34 @@ def test_get_plans_propagates_http_error(
         client.get_plans()
 
 
+def test_get_plans_multi_page_fetches_all_pages(
+    client: MagicPlanClient, mock_session: MagicMock
+) -> None:
+    # Arrange
+    page1_plan = _load_fixture("magicplan_api_plans_response_example.json")["data"][
+        "plans"
+    ][0]
+    page2_plan = {**page1_plan, "id": "page-2-plan-id"}
+    page1_response = MagicMock()
+    page1_response.json.return_value = {
+        "data": {"paging": {"page": 1, "next_page": True, "count": 2}, "plans": [page1_plan]}
+    }
+    page2_response = MagicMock()
+    page2_response.json.return_value = {
+        "data": {"paging": {"page": 2, "next_page": False, "count": 2}, "plans": [page2_plan]}
+    }
+    mock_session.get.side_effect = [page1_response, page2_response]
+    # Act
+    result = client.get_plans()
+    # Assert
+    assert mock_session.get.call_count == 2
+    mock_session.get.assert_any_call(f"{BASE_URL}/workgroups/plans", params={"page": 1})
+    mock_session.get.assert_any_call(f"{BASE_URL}/workgroups/plans", params={"page": 2})
+    assert len(result) == 2
+    assert result[0].id == page1_plan["id"]
+    assert result[1].id == "page-2-plan-id"
+
+
 # --- get_plan ---
 
 

From 0d324f99b29dfe1453d891f799512413e3776484 Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 14:52:46 +0000
Subject: [PATCH 17/20] =?UTF-8?q?Fetch=20all=20pages=20in=20get=5Fplans=20?=
 =?UTF-8?q?pagination=20loop=20=F0=9F=9F=A9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/magic_plan/magic_plan_client.py | 14 +++++++++++---
 1 file changed, 11 insertions(+), 3 deletions(-)

diff --git a/backend/magic_plan/magic_plan_client.py b/backend/magic_plan/magic_plan_client.py
index de2fe4f6..bf50a6f8 100644
--- a/backend/magic_plan/magic_plan_client.py
+++ b/backend/magic_plan/magic_plan_client.py
@@ -11,9 +11,17 @@ class MagicPlanClient:
         self._session.headers.update({"customer": customer_id, "key": api_key})
 
     def get_plans(self) -> list[PlanSummary]:
-        r = self._session.get(f"{_BASE_URL}/workgroups/plans", params={"page": 1})
-        r.raise_for_status()
-        return PlansListResponse.model_validate(r.json()["data"]).plans
+        all_plans: list[PlanSummary] = []
+        page = 1
+        while True:
+            r = self._session.get(f"{_BASE_URL}/workgroups/plans", params={"page": page})
+            r.raise_for_status()
+            response = PlansListResponse.model_validate(r.json()["data"])
+            all_plans.extend(response.plans)
+            if not response.paging.next_page:
+                break
+            page += 1
+        return all_plans
 
     def get_plan(self, plan_id: str) -> MagicPlanPlan:
         return MagicPlanPlan.model_validate(self._fetch_plan(plan_id).json()["data"])

From 5f77fbf4e45194a6fe18486e2cf199896333b0fa Mon Sep 17 00:00:00 2001
From: Daniel Roth <daniel_roth@hotmail.co.uk>
Date: Tue, 12 May 2026 14:54:14 +0000
Subject: [PATCH 18/20] =?UTF-8?q?Fetch=20all=20pages=20in=20get=5Fplans=20?=
 =?UTF-8?q?pagination=20loop=20=F0=9F=9F=AA?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/magic_plan/magic_plan_service.py           | 12 +++---------
 .../magic_plan/tests/test_magic_plan_service.py    | 14 +++++++-------
 2 files changed, 10 insertions(+), 16 deletions(-)

diff --git a/backend/magic_plan/magic_plan_service.py b/backend/magic_plan/magic_plan_service.py
index fb0a7610..22e19ddf 100644
--- a/backend/magic_plan/magic_plan_service.py
+++ b/backend/magic_plan/magic_plan_service.py
@@ -3,11 +3,7 @@ import json
 from datetime import datetime, timezone
 from typing import Optional
 
-from datatypes.magicplan.api.response import (
-    MagicPlanPlan,
-    PlanSummary,
-    PlansListResponse,
-)
+from datatypes.magicplan.api.response import MagicPlanPlan, PlanSummary
 from datatypes.magicplan.domain.mapper import map_plan
 from datatypes.magicplan.domain.models import Plan
 
@@ -39,10 +35,8 @@ class MagicPlanService:
         if uprn is not None:
             logger.info("MagicPlanService.run uprn=%s", uprn)
 
-        plans_response: PlansListResponse = self._client.get_plans()
-        matched: Optional[PlanSummary] = find_matching_plan(
-            plans_response.plans, address
-        )
+        plans: list[PlanSummary] = self._client.get_plans()
+        matched: Optional[PlanSummary] = find_matching_plan(plans, address)
 
         if matched is None:
             raise ValueError(f"No MagicPlan found for address: {address!r}")
diff --git a/backend/magic_plan/tests/test_magic_plan_service.py b/backend/magic_plan/tests/test_magic_plan_service.py
index f6954824..158cf4d6 100644
--- a/backend/magic_plan/tests/test_magic_plan_service.py
+++ b/backend/magic_plan/tests/test_magic_plan_service.py
@@ -91,7 +91,7 @@ def test_run_fetches_plan_with_matched_id(
     domain_plan: Plan,
 ) -> None:
     # Arrange
-    mock_client.get_plans.return_value.plans = [plan_summary]
+    mock_client.get_plans.return_value = [plan_summary]
     mock_client.get_plan.return_value = api_magic_plan
     service = _make_service(mock_client)
     with patch(
@@ -114,7 +114,7 @@ def test_run_returns_mapped_plan(
     domain_plan: Plan,
 ) -> None:
     # Arrange
-    mock_client.get_plans.return_value.plans = [plan_summary]
+    mock_client.get_plans.return_value = [plan_summary]
     mock_client.get_plan.return_value = api_magic_plan
     service = _make_service(mock_client)
     with patch(
@@ -137,7 +137,7 @@ def test_run_calls_save_plan_with_mapped_plan(
     plan_summary: PlanSummary,
 ) -> None:
     # Arrange
-    mock_client.get_plans.return_value.plans = [plan_summary]
+    mock_client.get_plans.return_value = [plan_summary]
     mock_client.get_plan.return_value = api_magic_plan
     service = _make_service(mock_client)
     with patch(
@@ -161,7 +161,7 @@ def test_run_accepts_uprn_without_error(
     plan_summary: PlanSummary,
 ) -> None:
     # Arrange
-    mock_client.get_plans.return_value.plans = [plan_summary]
+    mock_client.get_plans.return_value = [plan_summary]
     mock_client.get_plan.return_value = api_magic_plan
     service = _make_service(mock_client)
     with patch(
@@ -184,7 +184,7 @@ def test_run_uploads_to_s3_with_uprn_key(
     plan_summary: PlanSummary,
 ) -> None:
     # Arrange
-    mock_client.get_plans.return_value.plans = [plan_summary]
+    mock_client.get_plans.return_value = [plan_summary]
     request = _make_request(uprn="100023336956")
     service = MagicPlanService(client=mock_client, s3_bucket=S3_BUCKET)
     with patch(
@@ -211,7 +211,7 @@ def test_run_uploads_to_s3_with_deal_id_key_when_uprn_absent(
     plan_summary: PlanSummary,
 ) -> None:
     # Arrange
-    mock_client.get_plans.return_value.plans = [plan_summary]
+    mock_client.get_plans.return_value = [plan_summary]
     mock_client.get_plan.return_value = api_magic_plan
     request = _make_request(hubspot_deal_id="deal-456", uprn=None)
     service = MagicPlanService(client=mock_client, s3_bucket=S3_BUCKET)
@@ -242,7 +242,7 @@ def test_run_creates_uploaded_file_record(
     plan_summary: PlanSummary,
 ) -> None:
     # Arrange
-    mock_client.get_plans.return_value.plans = [plan_summary]
+    mock_client.get_plans.return_value = [plan_summary]
     mock_client.get_plan.return_value = api_magic_plan
     request = _make_request(hubspot_deal_id="deal-789", uprn="100023336956")
     service = MagicPlanService(client=mock_client, s3_bucket=S3_BUCKET)

From e458f0a2b718987bfc64635b690cb068293463dc Mon Sep 17 00:00:00 2001
From: Jun-te Kim <juntekim@googlemail.com>
Date: Tue, 12 May 2026 16:24:11 +0000
Subject: [PATCH 19/20] task and sub tasks imrpvoed

---
 backend/app/db/models/tasks.py            |   1 +
 backend/magic_plan/handler.py             |   3 +-
 backend/pashub_fetcher/handler/handler.py |   3 +-
 backend/utils/subtasks.py                 | 131 +++++++++++-----------
 etl/hubspot/scripts/scraper/main.py       |   3 +-
 5 files changed, 72 insertions(+), 69 deletions(-)

diff --git a/backend/app/db/models/tasks.py b/backend/app/db/models/tasks.py
index e97a939f..db1b7c04 100644
--- a/backend/app/db/models/tasks.py
+++ b/backend/app/db/models/tasks.py
@@ -9,6 +9,7 @@ from sqlmodel import SQLModel, Field, Relationship
 
 class SourceEnum(enum.Enum):  # TODO: move to domain?
     PORTFOLIO = "portfolio_id"
+    HUBSPOT_DEAL = "hubspot_deal_id"
 
 
 class Task(SQLModel, table=True):
diff --git a/backend/magic_plan/handler.py b/backend/magic_plan/handler.py
index 5fd90b7a..e7dc6484 100644
--- a/backend/magic_plan/handler.py
+++ b/backend/magic_plan/handler.py
@@ -5,13 +5,14 @@ from backend.magic_plan.magic_plan_client import MagicPlanClient
 from backend.magic_plan.magic_plan_service import MagicPlanService
 from backend.magic_plan.magic_plan_trigger_request import MagicPlanTriggerRequest
 from datatypes.magicplan.domain.models import Plan
+from backend.app.db.models.tasks import SourceEnum
 from backend.utils.subtasks import task_handler
 from utils.logger import setup_logger
 
 logger = setup_logger()
 
 
-@task_handler()
+@task_handler(task_source="magic_plan", source=SourceEnum.HUBSPOT_DEAL)
 def handler(body: dict[str, Any], context: Any) -> str:
     settings = get_settings()
     payload = MagicPlanTriggerRequest.model_validate(body)
diff --git a/backend/pashub_fetcher/handler/handler.py b/backend/pashub_fetcher/handler/handler.py
index 0d12b6bf..cd0c8113 100644
--- a/backend/pashub_fetcher/handler/handler.py
+++ b/backend/pashub_fetcher/handler/handler.py
@@ -5,6 +5,7 @@ from backend.pashub_fetcher.pashub_client import PashubClient, UnauthorizedError
 from backend.pashub_fetcher.pashub_service import PashubService
 from backend.pashub_fetcher.pashub_to_ara_trigger_request import PashubToAraTriggerRequest
 from backend.pashub_fetcher.token_getter import get_token_from_local_storage
+from backend.app.db.models.tasks import SourceEnum
 from backend.utils.subtasks import task_handler
 from utils.logger import setup_logger
 from utils.sharepoint.domna_sharepoint_client import DomnaSharepointClient
@@ -21,7 +22,7 @@ def get_pashub_client(email: str, password: str) -> PashubClient:
     return PashubClient(token=token)
 
 
-@task_handler()
+@task_handler(task_source="pashub_fetcher", source=SourceEnum.HUBSPOT_DEAL)
 def handler(body: Dict[str, Any], context: Any) -> List[str]:
     logger.info("Received message")
 
diff --git a/backend/utils/subtasks.py b/backend/utils/subtasks.py
index 6be3a742..36e67b78 100644
--- a/backend/utils/subtasks.py
+++ b/backend/utils/subtasks.py
@@ -1,75 +1,72 @@
-# decorators/subtask_handler.py
-
-from functools import wraps
-from typing import Callable, Any
-from uuid import UUID
 import json
+import os
+import time
+from functools import wraps
+from typing import Any, Callable, Optional, cast
+from uuid import UUID
 
 from backend.app.db.functions.tasks.Tasks import SubTaskInterface, TasksInterface
+from backend.app.db.models.tasks import SourceEnum
+from backend.app.plan.utils import build_cloudwatch_log_url
 from utils.logger import setup_logger
 
 
-def subtask_handler():
-    """
-    Decorator that wraps your existing handler and automatically:
+def _try_build_cloud_logs_url(start_ms: int) -> Optional[str]:
+    # Returns None outside a Lambda runtime so local/non-Lambda runs don't crash.
+    required = ("AWS_REGION", "AWS_LAMBDA_LOG_GROUP_NAME", "AWS_LAMBDA_LOG_STREAM_NAME")
+    if not all(k in os.environ for k in required):
+        return None
+    return build_cloudwatch_log_url(start_ms)
 
-    - Extracts task_id + sub_task_id from event
-    - Marks subtask as in progress
-    - Executes handler logic
-    - Marks subtask complete on success
-    - Marks failed on exception
+
+def subtask_handler() -> Callable[[Callable[..., Any]], Callable[..., Any]]:
+    """
+    Decorator for Lambdas that operate on an already-existing SubTask. Extracts
+    task_id + sub_task_id from each record, records the CloudWatch logs URL,
+    marks the SubTask in progress, then complete on success / failed on raise.
     """
 
-    def decorator(func: Callable[..., Any]):
+    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
 
         @wraps(func)
-        def wrapper(event: dict[str, Any], context: Any, *args, **kwargs):
+        def wrapper(event: dict[str, Any], context: Any, *args: Any, **kwargs: Any) -> None:
+            start_ms = int(time.time() * 1000)
+            cloud_logs_url = _try_build_cloud_logs_url(start_ms)
 
             records = event.get("Records", [event])
-
             interface = SubTaskInterface()
 
             for record in records:
-
-                # -------------------------------
-                # Parse body safely
-                # -------------------------------
-                body = {}
-
-                if isinstance(record.get("body"), str):
+                raw_body = record.get("body")
+                body: dict[str, Any]
+                if isinstance(raw_body, str):
                     try:
-                        body = json.loads(record["body"])
+                        body = json.loads(raw_body)
                     except Exception:
                         body = {}
+                elif isinstance(raw_body, dict):
+                    body = cast(dict[str, Any], raw_body)
                 else:
-                    body = record.get("body", {}) or {}
+                    body = {}
 
                 task_id_raw = body.get("task_id")
                 subtask_id_raw = body.get("sub_task_id")
 
                 task_id = UUID(task_id_raw) if isinstance(task_id_raw, str) else None
-                subtask_id = (
-                    UUID(subtask_id_raw) if isinstance(subtask_id_raw, str) else None
-                )
+                subtask_id = UUID(subtask_id_raw) if isinstance(subtask_id_raw, str) else None
 
                 if not task_id or not subtask_id:
                     raise RuntimeError("task_id or sub_task_id missing")
 
-                # -------------------------------
-                # Mark in progress
-                # -------------------------------
                 interface.update_subtask_status(
                     subtask_id=subtask_id,
                     status="in progress",
+                    cloud_logs_url=cloud_logs_url,
                 )
 
                 try:
-                    # Pass the parsed body into your function
                     result = func(body, context, *args, **kwargs)
 
-                    # -------------------------------
-                    # Success → mark complete
-                    # -------------------------------
                     interface.update_subtask_status(
                         subtask_id=subtask_id,
                         status="complete",
@@ -77,75 +74,79 @@ def subtask_handler():
                     )
 
                 except Exception as e:
-
-                    # -------------------------------
-                    # Failure → mark failed
-                    # -------------------------------
                     interface.update_subtask_status(
                         subtask_id=subtask_id,
                         status="failed",
                         outputs={"error": str(e)},
                     )
-
                     raise
 
-            return None
-
         return wrapper
 
     return decorator
 
 
-def task_handler():
+def task_handler(
+    task_source: str,
+    source: SourceEnum,
+) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
     """
-    Decorator that wraps a Lambda handler and automatically:
-
-    - Parses body from the first SQS record (or uses the event dict directly)
-    - Creates a fresh Task + SubTask in the database
-    - Marks the subtask as in progress
-    - Executes the handler, passing the parsed body
-    - Marks complete on success, failed on exception (and re-raises)
+    Decorator for Lambdas that are themselves the entry point of a pipeline (no
+    router in front). For each record the decorator creates a fresh Task +
+    SubTask with the given task_source and source. source_id is read from
+    body[source.value] (silent None if absent) — see ADR-0001. Records the
+    CloudWatch logs URL, marks the SubTask in progress, then complete on
+    success / failed on raise.
     """
 
-    def decorator(func: Callable[..., Any]):
-
-        task_source = f"{func.__module__}.{func.__qualname__}"
+    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
 
         @wraps(func)
-        def wrapper(event: dict[str, Any], context: Any, *args, **kwargs):
+        def wrapper(event: dict[str, Any], context: Any, *args: Any, **kwargs: Any) -> Any:
             logger = setup_logger()
+            start_ms = int(time.time() * 1000)
+            cloud_logs_url = _try_build_cloud_logs_url(start_ms)
 
-            records = event.get("Records", [event])  # fallback for non-SQS
-
-            results = []
-            failures = []
+            records = event.get("Records", [event])
+            results: list[Any] = []
+            failures: list[dict[str, Any]] = []
+            interface = SubTaskInterface()
 
             for record in records:
-                # Parse body
                 raw_body = record.get("body", record)
-
+                body: dict[str, Any]
                 if isinstance(raw_body, str):
                     try:
                         body = json.loads(raw_body)
                     except Exception:
                         body = {}
+                elif isinstance(raw_body, dict):
+                    body = cast(dict[str, Any], raw_body)
                 else:
-                    body = raw_body or {}
+                    body = {}
+
+                raw_source_id = body.get(source.value)
+                source_id: Optional[str] = (
+                    str(raw_source_id) if raw_source_id is not None else None
+                )
 
-                # Create task per message
                 logger.info("Creating task for source: %s", task_source)
                 task_id, subtask_id = TasksInterface.create_task(
                     task_source=task_source,
                     inputs=body,
+                    source=source,
+                    source_id=source_id,
                 )
 
-                logger.info("Created task_id=%s subtask_id=%s", task_id, subtask_id)
+                if subtask_id is None:
+                    raise RuntimeError("create_task did not return a subtask_id")
 
-                interface = SubTaskInterface()
+                logger.info("Created task_id=%s subtask_id=%s", task_id, subtask_id)
 
                 interface.update_subtask_status(
                     subtask_id=subtask_id,
                     status="in progress",
+                    cloud_logs_url=cloud_logs_url,
                 )
 
                 try:
@@ -172,13 +173,11 @@ def task_handler():
                     if "Records" in event:
                         failures.append({"itemIdentifier": record["messageId"]})
                     else:
-                        # Handle non-SQS events
                         raise
 
             if "Records" in event:
                 return {"batchItemFailures": failures}
 
-            # Handle non-SQS events
             return results
 
         return wrapper
diff --git a/etl/hubspot/scripts/scraper/main.py b/etl/hubspot/scripts/scraper/main.py
index 86844352..a7b640cf 100644
--- a/etl/hubspot/scripts/scraper/main.py
+++ b/etl/hubspot/scripts/scraper/main.py
@@ -9,6 +9,7 @@ from etl.hubspot.hubspot_deal_differ import HubspotDealDiffer
 from etl.hubspot.hubspot_trigger_orchestrator_trigger_request import (
     HubspotTriggerOrchestratorTriggerRequest,
 )
+from backend.app.db.models.tasks import SourceEnum
 from backend.utils.subtasks import task_handler
 from backend.app.db.models.hubspot_deal_data import HubspotDealData
 from utils.logger import setup_logger
@@ -16,7 +17,7 @@ from utils.logger import setup_logger
 logger = setup_logger()
 
 
-@task_handler()
+@task_handler(task_source="hubspot_scraper", source=SourceEnum.HUBSPOT_DEAL)
 def handler(body: dict[str, Any], context: Any) -> None:
     db_client = HubspotDataToDb()
     hubspot_client = HubspotClient()

From 09dbfe2106a4787c2194a921b01bd489821abed2 Mon Sep 17 00:00:00 2001
From: Jun-te Kim <juntekim@googlemail.com>
Date: Tue, 12 May 2026 17:03:16 +0000
Subject: [PATCH 20/20] fix dependency issue

---
 backend/app/plan/utils.py                 | 29 +---------------------
 backend/categorisation/handler/handler.py |  2 +-
 backend/categorisation/processor.py       |  3 ++-
 backend/engine/engine.py                  |  3 ++-
 backend/utils/cloudwatch.py               | 30 +++++++++++++++++++++++
 backend/utils/subtasks.py                 |  2 +-
 6 files changed, 37 insertions(+), 32 deletions(-)
 create mode 100644 backend/utils/cloudwatch.py

diff --git a/backend/app/plan/utils.py b/backend/app/plan/utils.py
index e752f5e0..a27bdf90 100644
--- a/backend/app/plan/utils.py
+++ b/backend/app/plan/utils.py
@@ -1,5 +1,4 @@
 import ast
-import os
 from typing import Optional
 import msgpack
 from uuid import UUID
@@ -8,6 +7,7 @@ from backend.addresses.Address import Address
 from backend.app.config import get_settings
 from backend.app.plan.data_classes import PropertyRequestData
 from backend.app.db.functions.tasks.Tasks import SubTaskInterface
+from backend.utils.cloudwatch import build_cloudwatch_log_url
 from starlette.responses import Response
 from utils.logger import setup_logger
 
@@ -241,33 +241,6 @@ def parse_eco_packages(
     return measures, mapped["target_sap"], mapped["plan_type"], already_installed
 
 
-def build_cloudwatch_log_url(start_ms: Optional[int]) -> str:
-    """
-    Build a CloudWatch Logs URL for the current Lambda invocation,
-    including timestamp window from start_ms to end_ms (epoch ms).
-    """
-    logger.info("Building cloudwatch logs URL")
-    region = os.environ["AWS_REGION"]
-    logger.info("Building cloudwatch logs URL: Got AWS region")
-    log_group = os.environ["AWS_LAMBDA_LOG_GROUP_NAME"]
-    logger.info("Building cloudwatch logs URL: Got lambda log group name")
-    log_stream = os.environ["AWS_LAMBDA_LOG_STREAM_NAME"]
-    logger.info("Building cloudwatch logs URL: Got lambda log stream name")
-
-    # CloudWatch console requires / encoded as $252F
-    encoded_group = log_group.replace("/", "$252F")
-    encoded_stream = log_stream.replace("/", "$252F")
-
-    # Return the full URL with time range
-    return (
-        f"https://console.aws.amazon.com/cloudwatch/home?"
-        f"region={region}"
-        f"#logsV2:log-groups/log-group/{encoded_group}"
-        f"/log-events/{encoded_stream}"
-        f"$3Fstart={start_ms}"
-    )
-
-
 def handle_error(
     msg: str,
     exception: Exception,
diff --git a/backend/categorisation/handler/handler.py b/backend/categorisation/handler/handler.py
index a1f69ea6..04dc0c44 100644
--- a/backend/categorisation/handler/handler.py
+++ b/backend/categorisation/handler/handler.py
@@ -3,7 +3,7 @@ import time
 from typing import Any, Mapping
 
 from backend.app.db.functions.tasks.Tasks import SubTaskInterface
-from backend.app.plan.utils import build_cloudwatch_log_url
+from backend.utils.cloudwatch import build_cloudwatch_log_url
 from backend.categorisation.categorisation_trigger_request import (
     CategorisationTriggerRequest,
 )
diff --git a/backend/categorisation/processor.py b/backend/categorisation/processor.py
index 88bc121e..e589c016 100644
--- a/backend/categorisation/processor.py
+++ b/backend/categorisation/processor.py
@@ -15,7 +15,8 @@ from backend.app.db.functions.tasks.Tasks import SubTaskInterface
 from backend.app.db.models.recommendations import PlanModel, ScenarioModel
 from backend.app.domain.classes.plan import Plan
 from backend.app.domain.classes.scenario import Scenario
-from backend.app.plan.utils import build_cloudwatch_log_url, handle_error
+from backend.app.plan.utils import handle_error
+from backend.utils.cloudwatch import build_cloudwatch_log_url
 from backend.categorisation.categorisation_trigger_request import (
     CategorisationTriggerRequest,
 )
diff --git a/backend/engine/engine.py b/backend/engine/engine.py
index 8b4ee821..c9e3972f 100644
--- a/backend/engine/engine.py
+++ b/backend/engine/engine.py
@@ -23,8 +23,9 @@ from backend.app.db.functions.tasks.Tasks import SubTaskInterface
 
 from backend.app.plan.schemas import PlanTriggerRequest
 from backend.app.plan.utils import (
-    get_cleaned, patch_epc, extract_property_request_data, handle_error, build_cloudwatch_log_url
+    get_cleaned, patch_epc, extract_property_request_data, handle_error
 )
+from backend.utils.cloudwatch import build_cloudwatch_log_url
 from backend.app.utils import sap_to_epc
 import backend.app.assumptions as assumptions
 
diff --git a/backend/utils/cloudwatch.py b/backend/utils/cloudwatch.py
new file mode 100644
index 00000000..e5309da2
--- /dev/null
+++ b/backend/utils/cloudwatch.py
@@ -0,0 +1,30 @@
+import os
+from typing import Optional
+
+from utils.logger import setup_logger
+
+logger = setup_logger()
+
+
+def build_cloudwatch_log_url(start_ms: Optional[int]) -> str:
+    """
+    Build a CloudWatch Logs URL for the current Lambda invocation, including a
+    timestamp window starting at start_ms. Requires AWS_REGION,
+    AWS_LAMBDA_LOG_GROUP_NAME, and AWS_LAMBDA_LOG_STREAM_NAME to be set in the
+    environment — i.e. only safe to call inside a Lambda runtime.
+    """
+    logger.info("Building cloudwatch logs URL")
+    region = os.environ["AWS_REGION"]
+    log_group = os.environ["AWS_LAMBDA_LOG_GROUP_NAME"]
+    log_stream = os.environ["AWS_LAMBDA_LOG_STREAM_NAME"]
+
+    encoded_group = log_group.replace("/", "$252F")
+    encoded_stream = log_stream.replace("/", "$252F")
+
+    return (
+        f"https://console.aws.amazon.com/cloudwatch/home?"
+        f"region={region}"
+        f"#logsV2:log-groups/log-group/{encoded_group}"
+        f"/log-events/{encoded_stream}"
+        f"$3Fstart={start_ms}"
+    )
diff --git a/backend/utils/subtasks.py b/backend/utils/subtasks.py
index 36e67b78..21ca24b1 100644
--- a/backend/utils/subtasks.py
+++ b/backend/utils/subtasks.py
@@ -7,7 +7,7 @@ from uuid import UUID
 
 from backend.app.db.functions.tasks.Tasks import SubTaskInterface, TasksInterface
 from backend.app.db.models.tasks import SourceEnum
-from backend.app.plan.utils import build_cloudwatch_log_url
+from backend.utils.cloudwatch import build_cloudwatch_log_url
 from utils.logger import setup_logger