An attach-mode batch models under the supplied sub_task without children 🟩

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

applications/modelling_e2e/handler.py

@@ -536,7 +536,9 @@ def handler(
 
         def _work(subtask: SubTask) -> None:
             inputs = subtask.inputs or {}
-            pid = int(inputs["property_id"])
+            _model_property(int(inputs["property_id"]))
+
+        def _model_property(pid: int) -> None:
             uprn = uprns[pid]
             postcode = postcodes.get(pid, "")
             logger.info(f"property={pid} uprn={uprn} postcode={postcode!r}")
@@ -685,16 +687,23 @@ def handler(
             )
             logger.info(f"property={pid} queued for write")
 
-        # Fan the batch out into one child SubTask per property and run them in
-        # a single batched pass: create all children, model each (failures
-        # isolated per child), then persist all their statuses in two writes +
-        # one cascade — not ~5 writes and a full parent re-roll-up per property
-        # (see TaskOrchestrator.run_subtasks).
-        orchestrator.run_subtasks(
-            task_id,
-            [{"property_id": pid} for pid in property_ids],
-            work=_work,
-        )
+        if trigger.subtask_id is not None:
+            # Attach mode (ADR-0055): the whole batch runs under the
+            # distributor's pre-created sub_task — the @task_handler wrapper is
+            # already inside it — so no per-property children are created.
+            for pid in property_ids:
+                _model_property(pid)
+        else:
+            # Fan the batch out into one child SubTask per property and run
+            # them in a single batched pass: create all children, model each
+            # (failures isolated per child), then persist all their statuses in
+            # two writes + one cascade — not ~5 writes and a full parent
+            # re-roll-up per property (see TaskOrchestrator.run_subtasks).
+            orchestrator.run_subtasks(
+                task_id,
+                [{"property_id": pid} for pid in property_ids],
+                work=_work,
+            )
 
         # Persist the whole batch in one transaction, then re-establish every
         # written Property's Baseline (the orchestrator batches its own UoW). The

applications/modelling_e2e/modelling_e2e_trigger_body.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import BaseModel, ConfigDict
 
 
@@ -11,3 +13,8 @@ class ModellingE2ETriggerBody(BaseModel):
     refetch_epc: bool = True
     repredict_epc: bool = True
     dry_run: bool = False
+    # Attach mode (ADR-0055): a Modelling Run batch carries the app-owned task
+    # and the distributor's pre-created sub_task; the handler then creates no
+    # per-property child sub_tasks. Absent on classic (script) messages.
+    task_id: Optional[str] = None
+    subtask_id: Optional[str] = None