Give each property that needs renaming its own sub_task 🟩

Properties are planned and renamed one at a time, so only those with work
get a row and a run killed by the timeout still leaves every property it
reached both renamed and recorded.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

applications/sharepoint_renamer/handler.py

@@ -15,15 +15,18 @@ turn a productive run red. Missing folders likewise do not fail a run.
 
 import os
 from typing import Any
+from uuid import UUID
 
 from applications.sharepoint_renamer.sharepoint_renamer_request import (
     SharepointRenamerRequest,
 )
 from domain.tasks.tasks import Source
 from orchestration.sharepoint_renamer_orchestrator import (
+    PropertyPlan,
     RenameSummary,
     SharepointRenamerOrchestrator,
 )
+from orchestration.task_orchestrator import TaskOrchestrator
 from utilities.aws_lambda.task_handler import task_handler
 from utils.sharepoint.domna_sharepoint_client import DomnaSharepointClient
 from utils.sharepoint.domna_sites import resolve_site
@@ -58,16 +61,69 @@ def _summary_json(summary: RenameSummary) -> dict[str, Any]:
     }
 
 
-@task_handler(task_source="sharepoint_renamer", source=Source.SHAREPOINT_SITE)
-def handler(body: dict[str, Any], context: Any) -> dict[str, Any]:
+def _plan_json(plan: PropertyPlan) -> dict[str, Any]:
+    """A property sub_task's inputs: enough to see what was intended, and to
+    re-run just this property by re-sending them."""
+    return {
+        "uprn": plan.uprn,
+        "address": plan.address,
+        "postcode": plan.postcode,
+        "renames": [
+            {
+                "item_id": intent.item_id,
+                "original_name": intent.original_name,
+                "new_name": intent.new_name,
+            }
+            for intent in plan.intents
+        ],
+    }
+
+
+@task_handler(
+    task_source="sharepoint_renamer",
+    source=Source.SHAREPOINT_SITE,
+    pass_task_orchestrator=True,
+)
+def handler(
+    body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, task_id: UUID
+) -> dict[str, Any]:
     request = SharepointRenamerRequest.model_validate(body)
     sp_client = DomnaSharepointClient(resolve_site(request.sharepoint_site))
-    summary = SharepointRenamerOrchestrator(
+    renamer = SharepointRenamerOrchestrator(
         sp_client, CSV_PATH, dry_run=request.dry_run
-    ).run()
+    )
+
+    summary = RenameSummary()
+    # Plan and rename one property at a time (the plan is a lazy generator), so
+    # a run killed by the timeout still leaves every property it reached both
+    # renamed and recorded. Only properties with something to do are worth a
+    # sub_task — a steady-state run over already-canonical folders creates none.
+    for plan in renamer.plan():
+        if not plan.intents:
+            summary.merge_in(renamer.rename(plan))
+            continue
+        subtask = orchestrator.create_child_subtask(task_id, inputs=_plan_json(plan))
+        orchestrator.run_subtask(
+            subtask.id, work=lambda p=plan: _record(renamer, p, summary)
+        )
     return _summary_json(summary)
 
 
+def _record(
+    renamer: SharepointRenamerOrchestrator, plan: PropertyPlan, summary: RenameSummary
+) -> dict[str, Any]:
+    """Rename one property, adding its result to the run total and returning it
+    for that property's own sub_task outputs.
+
+    ``rename`` records a rejected file rather than raising, so the sub_task
+    completes with the failure in its outputs. Failing it instead would turn
+    the whole Task red — any failed child does — and one locked file must not
+    do that (ADR-0060)."""
+    result = renamer.rename(plan)
+    summary.merge_in(result)
+    return _summary_json(result)
+
+
 if __name__ == "__main__":
     # Defaults to a dry run: this module is now decorated, so invoking it by
     # hand writes a real task row *and*, without the flag, renames live files.

tests/applications/sharepoint_renamer/test_handler.py

@@ -22,7 +22,7 @@ from sqlalchemy import Engine
 from sqlmodel import Session
 
 from domain.pashub_fetcher.sharepoint_subfolders import SharepointSubfolders
-from domain.tasks.subtasks import SubTaskStatus
+from domain.tasks.subtasks import SubTask, SubTaskStatus
 from domain.tasks.tasks import Source, TaskStatus
 from orchestration.sharepoint_renamer_orchestrator import (
     ASSESSMENT_SUBFOLDER,
@@ -154,9 +154,21 @@ def _handler(harness: Harness) -> Callable[..., Any]:
         task_source="sharepoint_renamer",
         source=Source.SHAREPOINT_SITE,
         orchestrator_cm=harness.factory,
+        pass_task_orchestrator=True,
     )(undecorated)
 
 
+def _property_subtasks(harness: Harness, result: Any) -> list[SubTask]:
+    """The per-property children, i.e. every sub_task except the wrapping one
+    the decorator creates for the run itself."""
+    wrapper = _subtask_id(result)
+    return [
+        s
+        for s in harness.subtasks.list_by_task(_task_id(result))
+        if s.id != wrapper
+    ]
+
+
 def _sqs_event(body: dict[str, Any]) -> dict[str, Any]:
     return {"Records": [{"messageId": "msg-1", "body": json.dumps(body)}]}
 
@@ -205,6 +217,63 @@ def test_completed_subtask_carries_the_run_summary(
     }
 
 
+# ---------------------------------------------------------------------------
+# One sub_task per property that actually needs work
+# ---------------------------------------------------------------------------
+
+
+def test_each_property_with_work_gets_its_own_subtask(
+    harness: Harness, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # Arrange: one property with two renameable files, one with nothing to do
+    assessment_2 = (
+        f"{BASE_PATH}/2 Low St, AB1 2CE"
+        f"/{SharepointSubfolders.ASSESSMENT.value}/{ASSESSMENT_SUBFOLDER}"
+    )
+    sharepoint = FakeSharepoint(
+        {
+            BASE_PATH: {"value": []},
+            ASSESSMENT_PATH: {
+                "value": [_file("Survey.pdf", "id-1"), _file("Report.docx", "id-2")]
+            },
+            assessment_2: {"value": [_file("200_2 Low St AB1 2CE_Done.pdf", "id-3")]},
+        }
+    )
+    _install(
+        monkeypatch,
+        tmp_path,
+        sharepoint,
+        rows=[(UPRN, ADDRESS, POSTCODE), ("200", "2 Low St", "AB1 2CE")],
+    )
+
+    # Act
+    result = _handler(harness)(_sqs_event({"sharepoint_site": SITE}), None)
+
+    # Assert
+    children = _property_subtasks(harness, result)
+    assert [c.inputs for c in children] == [
+        {
+            "uprn": UPRN,
+            "address": ADDRESS,
+            "postcode": POSTCODE,
+            "renames": [
+                {
+                    "item_id": "id-1",
+                    "original_name": "Survey.pdf",
+                    "new_name": f"{UPRN}_{ADDRESS} {POSTCODE}_Survey.pdf",
+                },
+                {
+                    "item_id": "id-2",
+                    "original_name": "Report.docx",
+                    "new_name": f"{UPRN}_{ADDRESS} {POSTCODE}_Report.docx",
+                },
+            ],
+        }
+    ]
+    assert children[0].status is SubTaskStatus.COMPLETE
+    assert (children[0].outputs or {})["result"]["renamed"] == 2
+
+
 # ---------------------------------------------------------------------------
 # A run is attributable: its request, and the site it targeted
 # ---------------------------------------------------------------------------