Merge pull request #1706 from Hestia-Homes/feature/renamer-task-handler

SharePoint Renamer: put the run on the app-owned-task lane

.github/workflows/deploy_terraform.yml

@@ -532,6 +532,9 @@ jobs:
       TF_VAR_sharepoint_client_secret: ${{ secrets.SHAREPOINT_CLIENT_SECRET }}
       TF_VAR_sharepoint_tenant_id: ${{ secrets.SHAREPOINT_TENANT_ID }}
       TF_VAR_social_housing_wave_3_sharepoint_id: ${{ secrets.SOCIAL_HOUSING_WAVE_3_SHAREPOINT_ID }}
+      TF_VAR_db_host: ${{ secrets.DEV_DB_HOST }}
+      TF_VAR_db_name: ${{ secrets.DEV_DB_NAME }}
+      TF_VAR_db_port: ${{ secrets.DEV_DB_PORT }}
 
 
   # ============================================================

applications/sharepoint_renamer/handler.py

@@ -1,26 +1,131 @@
+"""SQS-triggered Lambda: rename Sero SharePoint files to their canonical names.
+
+Runs on the app-owned-task lane. Each invocation creates its own ``task`` plus
+an initial ``sub_task`` and runs inside the ``TaskOrchestrator`` lifecycle, so a
+button press in the admin portal leaves a row with a status, a CloudWatch link,
+and — on completion — a summary of what the run did. The portal reads
+``tasks`` / ``sub_task`` directly, so the trigger stays fire-and-forget SQS and
+nothing is handed back to the caller.
+
+Partial failures complete rather than fail: a run where SharePoint rejected some
+files ends **complete** with those failures listed in ``outputs``, matching the
+best-effort Download Package convention (ADR-0060). One locked file should not
+turn a productive run red. Missing folders likewise do not fail a run.
+"""
+
 import os
+from functools import partial
 from typing import Any
+from uuid import UUID
 
 from applications.sharepoint_renamer.sharepoint_renamer_request import (
     SharepointRenamerRequest,
 )
-from orchestration.sharepoint_renamer_orchestrator import SharepointRenamerOrchestrator
+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 DomnaSites
+from utils.sharepoint.domna_sites import resolve_site
 
 CSV_PATH = os.path.join(
     os.path.dirname(os.path.abspath(__file__)), "sero_address_list.csv"
 )
 
 
-def handler(event: dict[str, Any], context: Any) -> None:
-    request = SharepointRenamerRequest.model_validate(event)
-    sp_client = DomnaSharepointClient(DomnaSites.SOCIAL_HOUSING_WAVE_3)
-    orchestrator = SharepointRenamerOrchestrator(
+def _summary_json(summary: RenameSummary) -> dict[str, Any]:
+    """The portal's view of a run, mapped field by field rather than derived
+    from ``RenameSummary``. The explicit mapping is the point: it keeps the
+    portal's contract stable when the internal type is renamed or reshaped, and
+    ``outputs`` is serialised to text so it must be plain JSON either way.
+
+    Counts are per-**file**; ``missing_folders`` is per-**property**."""
+    return {
+        "renamed": summary.renamed,
+        "would_rename": summary.would_rename,
+        "skipped_already_canonical": summary.skipped_already_canonical,
+        "skipped_images": summary.skipped_images,
+        "missing_folders": summary.missing_folders,
+        "failed": [
+            {
+                "uprn": failure.uprn,
+                "original_name": failure.original_name,
+                "new_name": failure.new_name,
+                "error": failure.error,
+            }
+            for failure in summary.failed
+        ],
+    }
+
+
+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))
+    renamer = SharepointRenamerOrchestrator(
         sp_client, CSV_PATH, dry_run=request.dry_run
     )
-    orchestrator.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=partial(_record, renamer, plan, 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__":
-    handler({"dry_run": False}, None)
+    # 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.
+    handler({"sharepoint_site": "SOCIAL_HOUSING_WAVE_3", "dry_run": True}, None)

applications/sharepoint_renamer/handler/Dockerfile

@@ -10,6 +10,19 @@ COPY utilities/ utilities/
 COPY backend/__init__.py backend/__init__.py
 # SharepointSubfolders only — the rest of the PasHub service is not needed here.
 COPY domain/pashub_fetcher/ domain/pashub_fetcher/
+# The app-owned-task lane (@task_handler -> TaskOrchestrator -> Postgres): the
+# task domain types, both task repositories, and the Postgres modules they
+# reach. Copied file by file rather than whole-package to keep the image
+# minimal — tests/test_lambda_packaging.py is what says whether it is complete.
+COPY domain/tasks/ domain/tasks/
+COPY repositories/__init__.py repositories/__init__.py
+COPY repositories/tasks/ repositories/tasks/
+COPY infrastructure/__init__.py infrastructure/__init__.py
+COPY infrastructure/postgres/__init__.py infrastructure/postgres/__init__.py
+COPY infrastructure/postgres/config.py infrastructure/postgres/config.py
+COPY infrastructure/postgres/engine.py infrastructure/postgres/engine.py
+COPY infrastructure/postgres/task_table.py infrastructure/postgres/task_table.py
+COPY infrastructure/postgres/subtask_table.py infrastructure/postgres/subtask_table.py
 COPY orchestration/ orchestration/
 COPY applications/sharepoint_renamer/ applications/sharepoint_renamer/
 CMD ["applications.sharepoint_renamer.handler.handler"]

applications/sharepoint_renamer/handler/requirements.txt

@@ -1,3 +1,10 @@
 msal
 requests
 pydantic-settings==2.6.0
+
+# The app-owned-task lane: the handler creates its own task + sub_task through
+# TaskOrchestrator's Postgres repositories. psycopg2 is the driver
+# PostgresConfig defaults to (tests use a different one).
+sqlalchemy==2.0.36
+sqlmodel
+psycopg2-binary==2.9.10

applications/sharepoint_renamer/local_handler/invoke_local_lambda.py

@@ -1,4 +1,19 @@
 #!/usr/bin/env python3
+"""Invoke the locally-running Renamer image with a realistic SQS payload.
+
+The event is hand-built SQS-shaped on purpose. A direct invoke (a bare body,
+no ``Records``) is the one path where the request is read the same way whether
+or not the record body is unwrapped — so it is the one path that would NOT have
+caught ``dry_run`` being ignored on queue-triggered runs. Exercise the shape
+production actually delivers.
+
+This writes a real task + sub_task row to whatever database ``.env`` points at
+(dev), and is the only place the production Postgres driver is exercised — the
+tests use a different one. ``dry_run`` is True so nothing is renamed for real.
+"""
+
+import json
+
 import requests
 
 HOST = "localhost"
@@ -6,7 +21,16 @@ PORT = "9003"
 
 LAMBDA_URL = f"http://{HOST}:{PORT}/2015-03-31/functions/function/invocations"
 
-payload = {"dry_run": True}
+payload = {
+    "Records": [
+        {
+            "messageId": "local-test-1",
+            "body": json.dumps(
+                {"sharepoint_site": "SOCIAL_HOUSING_WAVE_3", "dry_run": True}
+            ),
+        }
+    ]
+}
 
 response = requests.post(LAMBDA_URL, json=payload)
 

applications/sharepoint_renamer/sharepoint_renamer_request.py

@@ -2,4 +2,10 @@ from pydantic import BaseModel
 
 
 class SharepointRenamerRequest(BaseModel):
+    # The DomnaSites member name the run targets, e.g. "SOCIAL_HOUSING_WAVE_3".
+    # Also the @task_handler `source_id` (the body key matches
+    # Source.SHAREPOINT_SITE's value), so a run is attributable after the fact.
+    # The base path is still hardcoded and Sero-specific — this records what
+    # ran; it is not yet a multi-site lever.
+    sharepoint_site: str = "SOCIAL_HOUSING_WAVE_3"
     dry_run: bool = False

backend/app/db/models/tasks.py

@@ -8,8 +8,13 @@ from sqlmodel import SQLModel, Field, Relationship
 
 
 class SourceEnum(enum.Enum):  # TODO: move to domain?
+    # Legacy mirror of the same `source` pgEnum as domain.tasks.tasks.Source.
+    # Already drifted — PROPERTY is missing here — so this is not a reliable
+    # lockstep copy. SHAREPOINT_SITE is added because the backend maps rows
+    # through this model and an unknown value raises LookupError on read.
     PORTFOLIO = "portfolio_id"
     HUBSPOT_DEAL = "hubspot_deal_id"
+    SHAREPOINT_SITE = "sharepoint_site"
 
 
 class Task(SQLModel, table=True):

deployment/terraform/lambda/sharepoint_renamer/main.tf

@@ -1,3 +1,11 @@
+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"
 
@@ -18,5 +26,13 @@ module "lambda" {
     SHAREPOINT_CLIENT_SECRET            = var.sharepoint_client_secret
     SHAREPOINT_TENANT_ID                = var.sharepoint_tenant_id
     SOCIAL_HOUSING_WAVE_3_SHAREPOINT_ID = var.social_housing_wave_3_sharepoint_id
+
+    # The run creates its own task + sub_task so the admin portal can see it.
+    # No VPC needed — the same public path every other task-lane Lambda uses.
+    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/sharepoint_renamer/variables.tf

@@ -27,7 +27,11 @@ variable "timeout" {
 variable "reserved_concurrent_executions" {
   type        = number
   default     = 1
-  description = "Prevent parallel renames causing race conditions on SharePoint."
+  description = <<-EOT
+    Prevent parallel renames causing race conditions on SharePoint. Since the
+    Renamer joined the app-owned-task lane this also caps its DB connections at
+    one — a second lever hanging off the same number.
+  EOT
 }
 
 variable "batch_size" {
@@ -55,6 +59,21 @@ variable "social_housing_wave_3_sharepoint_id" {
   sensitive = true
 }
 
+variable "db_host" {
+  type      = string
+  sensitive = true
+}
+
+variable "db_name" {
+  type      = string
+  sensitive = true
+}
+
+variable "db_port" {
+  type      = string
+  sensitive = true
+}
+
 locals {
   image_uri = "${var.ecr_repo_url}@${var.image_digest}"
 }

domain/tasks/tasks.py

@@ -18,6 +18,13 @@ class Source(str, Enum):
     PORTFOLIO = "portfolio_id"
     HUBSPOT_DEAL = "hubspot_deal_id"
     PROPERTY = "property_id"
+    # A run anchored on a SharePoint site rather than anything in our own
+    # schema — the SharePoint Renamer. The pgEnum is FE-owned via Drizzle
+    # (ADR-0003), so the Drizzle migration adding 'sharepoint_site' MUST apply
+    # before this deploys or the first run fails on an invalid enum value.
+    # Tests won't catch that: ephemeral DBs build from the SQLModel mirrors
+    # below and emit the member automatically.
+    SHAREPOINT_SITE = "sharepoint_site"
 
 
 @dataclass

orchestration/sharepoint_renamer_orchestrator.py

@@ -1,5 +1,7 @@
 import csv
 import os
+from collections.abc import Iterator
+from dataclasses import dataclass, field
 from typing import Optional
 
 from domain.pashub_fetcher.sharepoint_subfolders import SharepointSubfolders
@@ -16,6 +18,74 @@ BATCH_FOLDER_PREFIX = "_Sero Batch"
 logger = setup_logger()
 
 
+@dataclass
+class RenameFailure:
+    """One file SharePoint refused to rename, and why."""
+
+    uprn: str
+    original_name: str
+    new_name: str
+    error: str
+
+
+@dataclass
+class RenameIntent:
+    """One file the walk found needing a rename, and what to rename it to."""
+
+    uprn: str
+    item_id: str
+    original_name: str
+    new_name: str
+
+
+@dataclass
+class PropertyPlan:
+    """What one address-list row needs, found by walking its folder tree.
+
+    Yielded for *every* row, including rows with nothing to do — a plan with no
+    intents still carries the skips it observed, so run-level totals stay
+    complete even though only rows with intents are worth a sub_task.
+
+    The plan is what makes "when is this property finished?" answerable: its
+    size is known before the first rename is attempted."""
+
+    uprn: str
+    address: str
+    postcode: str
+    intents: list[RenameIntent] = field(default_factory=list[RenameIntent])
+    skipped_already_canonical: int = 0
+    skipped_images: int = 0
+    missing_folder: bool = False
+
+    def merge_in(self, other: "PropertyPlan") -> None:
+        self.intents.extend(other.intents)
+        self.skipped_already_canonical += other.skipped_already_canonical
+        self.skipped_images += other.skipped_images
+
+
+@dataclass
+class RenameSummary:
+    """What a run did, threaded back up through the traversal.
+
+    Counts are per-**file**; ``missing_folders`` is per-**property** (a UPRN with
+    no assessment folder under any root, so no file was even looked at)."""
+
+    renamed: int = 0
+    would_rename: int = 0
+    skipped_already_canonical: int = 0
+    skipped_images: int = 0
+    missing_folders: list[str] = field(default_factory=list[str])
+    failed: list[RenameFailure] = field(default_factory=list[RenameFailure])
+
+    def merge_in(self, other: "RenameSummary") -> None:
+        self.renamed += other.renamed
+        self.would_rename += other.would_rename
+        self.skipped_already_canonical += other.skipped_already_canonical
+        self.skipped_images += other.skipped_images
+        self.missing_folders.extend(other.missing_folders)
+        self.failed.extend(other.failed)
+
+
 def build_canonical_filename(
     uprn: str, address: str, postcode: str, original_name: str
 ) -> Optional[str]:
@@ -76,7 +146,13 @@ class SharepointRenamerOrchestrator:
         )
         return [BASE_PATH, *batch_roots]
 
-    def run(self) -> None:
+    def plan(self) -> Iterator[PropertyPlan]:
+        """Walk the address list, yielding what each property needs.
+
+        Lazy on purpose: the caller renames one property before the next is
+        walked, so a run that is killed part-way still leaves everything it
+        reached already renamed — and, on the task lane, a completed sub_task
+        recording it."""
         roots = self._discover_roots()
         with open(self._csv_path, newline="", encoding="utf-8-sig") as f:
             reader = csv.DictReader(f)
@@ -87,16 +163,65 @@ class SharepointRenamerOrchestrator:
                 )
 
             for row in reader:
-                self._process_row(
+                yield self._plan_row(
                     roots,
                     uprn=row["UPRN"].strip(),
                     address=row["Address"].strip(),
                     postcode=row["Postcode"].strip(),
                 )
 
-    def _process_row(
+    def rename(self, plan: PropertyPlan) -> RenameSummary:
+        """Apply one property's plan, returning what actually happened.
+
+        Never raises for a file SharePoint rejects: the rejection is recorded
+        in the summary so one locked file cannot fail the property, let alone
+        the run."""
+        summary = RenameSummary(
+            skipped_already_canonical=plan.skipped_already_canonical,
+            skipped_images=plan.skipped_images,
+            missing_folders=[plan.uprn] if plan.missing_folder else [],
+        )
+        for intent in plan.intents:
+            if self._dry_run:
+                summary.would_rename += 1
+                logger.info(
+                    f'Would rename: "{intent.original_name}" → '
+                    f'"{intent.new_name}" (UPRN: {intent.uprn})'
+                )
+                continue
+            try:
+                self._sp_client.rename_file(intent.item_id, intent.new_name)
+                summary.renamed += 1
+                logger.info(
+                    f'Renamed: "{intent.original_name}" → '
+                    f'"{intent.new_name}" (UPRN: {intent.uprn})'
+                )
+            except Exception as e:
+                summary.failed.append(
+                    RenameFailure(
+                        uprn=intent.uprn,
+                        original_name=intent.original_name,
+                        new_name=intent.new_name,
+                        error=str(e),
+                    )
+                )
+                logger.error(
+                    f'Failed to rename "{intent.original_name}" → '
+                    f'"{intent.new_name}" (UPRN: {intent.uprn}): {e}'
+                )
+        return summary
+
+    def run(self) -> RenameSummary:
+        """Plan and rename every property in one pass — the entry point for
+        callers that want no per-property bookkeeping."""
+        summary = RenameSummary()
+        for plan in self.plan():
+            summary.merge_in(self.rename(plan))
+        return summary
+
+    def _plan_row(
         self, roots: list[str], uprn: str, address: str, postcode: str
-    ) -> None:
+    ) -> PropertyPlan:
         # Batch folders name properties "{uprn}_{address}"; filenames must not
         # repeat the UPRN, so the folder lookup and the canonical name diverge.
         display_address = address.removeprefix(f"{uprn}_")
@@ -105,48 +230,54 @@ class SharepointRenamerOrchestrator:
                 f"{root}/{address}, {postcode}"
                 f"/{SharepointSubfolders.ASSESSMENT.value}/{ASSESSMENT_SUBFOLDER}"
             )
-            if self._process_folder(folder_path, uprn, display_address, postcode):
-                return
+            found = self._plan_folder(folder_path, uprn, display_address, postcode)
+            if found is not None:
+                return found
         logger.warning(
             f"Missing folder for UPRN {uprn} in any root: {address}, {postcode}"
         )
+        return PropertyPlan(
+            uprn=uprn,
+            address=display_address,
+            postcode=postcode,
+            missing_folder=True,
+        )
 
-    def _process_folder(
+    def _plan_folder(
         self, folder_path: str, uprn: str, address: str, postcode: str
-    ) -> bool:
+    ) -> Optional[PropertyPlan]:
+        """What this folder (and its subfolders) needs renaming, or None when
+        the folder does not exist under this root."""
         try:
             contents = self._sp_client.get_folders_in_path(folder_path)
         except ValueError:
-            return False
+            return None
 
+        plan = PropertyPlan(uprn=uprn, address=address, postcode=postcode)
         for item in contents.get("value", []):
             if "folder" in item:
-                self._process_folder(
+                child = self._plan_folder(
                     f"{folder_path}/{item['name']}", uprn, address, postcode
                 )
+                if child is not None:
+                    plan.merge_in(child)
             elif "file" in item:
                 original_name: str = item["name"]
                 if os.path.splitext(original_name)[1].lower() in {".jpg", ".heic"}:
+                    plan.skipped_images += 1
                     continue
                 new_name = build_canonical_filename(
                     uprn, address, postcode, original_name
                 )
-
                 if new_name is None:
+                    plan.skipped_already_canonical += 1
                     continue
-
-                if self._dry_run:
-                    logger.info(
-                        f'Would rename: "{original_name}" → "{new_name}" (UPRN: {uprn})'
+                plan.intents.append(
+                    RenameIntent(
+                        uprn=uprn,
+                        item_id=item["id"],
+                        original_name=original_name,
+                        new_name=new_name,
                     )
-                else:
-                    try:
-                        self._sp_client.rename_file(item["id"], new_name)
-                        logger.info(
-                            f'Renamed: "{original_name}" → "{new_name}" (UPRN: {uprn})'
-                        )
-                    except Exception as e:
-                        logger.error(
-                            f'Failed to rename "{original_name}" → "{new_name}" (UPRN: {uprn}): {e}'
-                        )
-        return True
+                )
+        return plan

tests/applications/sharepoint_renamer/test_handler.py

@@ -0,0 +1,457 @@
+"""Tests for the SharePoint Renamer Lambda handler on the app-owned-task lane.
+
+A real Lambda event goes through the real ``@task_handler`` decorator against a
+real ephemeral PostgreSQL database; only SharePoint is faked, and it is faked at
+the **client** boundary with canned Graph responses so the real rename
+orchestrator runs. What the tests assert is therefore what is visible outside
+the module: the rows that land in Postgres, and the calls made to SharePoint.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Generator, Iterator
+from contextlib import contextmanager
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Callable, Optional, cast
+from uuid import UUID
+
+import pytest
+from sqlalchemy import Engine
+from sqlmodel import Session
+
+from domain.pashub_fetcher.sharepoint_subfolders import SharepointSubfolders
+from domain.tasks.subtasks import SubTask, SubTaskStatus
+from domain.tasks.tasks import Source, TaskStatus
+from orchestration.sharepoint_renamer_orchestrator import (
+    ASSESSMENT_SUBFOLDER,
+    BASE_PATH,
+)
+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 utils.sharepoint.domna_sites import DomnaSites
+
+SITE = "SOCIAL_HOUSING_WAVE_3"
+UPRN = "100"
+ADDRESS = "1 High St"
+POSTCODE = "AB1 2CD"
+
+ASSESSMENT_PATH = (
+    f"{BASE_PATH}/{ADDRESS}, {POSTCODE}"
+    f"/{SharepointSubfolders.ASSESSMENT.value}/{ASSESSMENT_SUBFOLDER}"
+)
+
+
+# ---------------------------------------------------------------------------
+# Harness: a TaskOrchestrator on a real ephemeral PostgreSQL database
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class Harness:
+    orchestrator: TaskOrchestrator
+    tasks: TaskPostgresRepository
+    subtasks: SubTaskPostgresRepository
+
+    @contextmanager
+    def factory(self) -> Generator[TaskOrchestrator, None, None]:
+        yield self.orchestrator
+
+
+@pytest.fixture
+def harness(db_engine: Engine) -> Iterator[Harness]:
+    with Session(db_engine) as session:
+        tasks = TaskPostgresRepository(session=session)
+        subtasks = SubTaskPostgresRepository(session=session)
+        yield Harness(
+            orchestrator=TaskOrchestrator(task_repo=tasks, subtask_repo=subtasks),
+            tasks=tasks,
+            subtasks=subtasks,
+        )
+
+
+# ---------------------------------------------------------------------------
+# Fake SharePoint: canned Graph responses keyed by path
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class FakeSharepoint:
+    """Stands in for DomnaSharepointClient with canned Graph listings. Paths not
+    in ``contents`` raise ValueError, exactly as the real client does for a
+    folder that does not exist."""
+
+    contents: dict[str, dict[str, Any]]
+    reject: dict[str, Exception] = field(default_factory=dict[str, Exception])
+    renamed: list[tuple[str, str]] = field(default_factory=list[tuple[str, str]])
+
+    def get_folders_in_path(self, path: str) -> dict[str, Any]:
+        if path not in self.contents:
+            raise ValueError(f"not found: {path}")
+        return self.contents[path]
+
+    def rename_file(self, item_id: str, new_name: str) -> None:
+        if item_id in self.reject:
+            raise self.reject[item_id]
+        self.renamed.append((item_id, new_name))
+
+
+def _file(name: str, item_id: str) -> dict[str, Any]:
+    return {"name": name, "id": item_id, "file": {}}
+
+
+def _one_property_site(*files: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    """A site holding exactly one property folder, with `files` in its
+    assessment folder."""
+    return {
+        BASE_PATH: {"value": []},
+        ASSESSMENT_PATH: {"value": list(files)},
+    }
+
+
+# ---------------------------------------------------------------------------
+# Wiring: the real handler body, re-decorated onto the test orchestrator
+# ---------------------------------------------------------------------------
+
+
+def _install(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    sharepoint: FakeSharepoint,
+    *,
+    rows: Optional[list[tuple[str, str, str]]] = None,
+) -> None:
+    """Point the handler at the fake SharePoint and a small address list, and
+    configure the site the tests target."""
+    import applications.sharepoint_renamer.handler as module
+
+    rows = rows if rows is not None else [(UPRN, ADDRESS, POSTCODE)]
+
+    monkeypatch.setenv(f"{SITE}_SHAREPOINT_ID", "test-site-id")
+
+    def client_for(_site: DomnaSites) -> FakeSharepoint:
+        return sharepoint
+
+    monkeypatch.setattr(module, "DomnaSharepointClient", client_for)
+
+    csv_file = tmp_path / "addresses.csv"
+    csv_file.write_text(
+        "\n".join(["UPRN,Address,Postcode"] + [",".join(r) for r in rows]) + "\n",
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(module, "CSV_PATH", str(csv_file))
+
+
+def _handler(harness: Harness) -> Callable[..., Any]:
+    """The real handler body under the real decorator, wired to the test DB."""
+    import applications.sharepoint_renamer.handler as module
+
+    undecorated = cast(Callable[..., Any], getattr(module.handler, "__wrapped__"))
+    return task_handler(
+        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)}]}
+
+
+def _subtask_id(result: Any) -> UUID:
+    return UUID(result["tasks"][0]["subtask_id"])
+
+
+def _task_id(result: Any) -> UUID:
+    return UUID(result["tasks"][0]["task_id"])
+
+
+# ---------------------------------------------------------------------------
+# A completed run records what it did
+# ---------------------------------------------------------------------------
+
+
+def test_completed_subtask_carries_the_run_summary(
+    harness: Harness, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # Arrange
+    sharepoint = FakeSharepoint(
+        _one_property_site(
+            _file("Survey.pdf", "id-1"),
+            _file(f"{UPRN}_High St {POSTCODE}_Done.pdf", "id-2"),
+            _file("Front.jpg", "id-3"),
+        )
+    )
+    _install(monkeypatch, tmp_path, sharepoint)
+
+    # Act
+    result = _handler(harness)(_sqs_event({"sharepoint_site": SITE}), None)
+
+    # Assert
+    subtask = harness.subtasks.get(_subtask_id(result))
+    assert subtask.status is SubTaskStatus.COMPLETE
+    assert subtask.outputs == {
+        "result": {
+            "renamed": 1,
+            "would_rename": 0,
+            "skipped_already_canonical": 1,
+            "skipped_images": 1,
+            "missing_folders": [],
+            "failed": [],
+        }
+    }
+
+
+# ---------------------------------------------------------------------------
+# 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
+# ---------------------------------------------------------------------------
+
+
+def test_run_records_its_request_and_target_site(
+    harness: Harness, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """``sharepoint_site`` round-trips through the real ``source`` pgEnum — the
+    member the FE's Drizzle migration must add before this deploys."""
+    # Arrange
+    sharepoint = FakeSharepoint(_one_property_site())
+    _install(monkeypatch, tmp_path, sharepoint)
+    body = {"sharepoint_site": SITE, "dry_run": True}
+
+    # Act
+    result = _handler(harness)(_sqs_event(body), None)
+
+    # Assert
+    task = harness.tasks.get(_task_id(result))
+    assert (task.task_source, task.source, task.source_id) == (
+        "sharepoint_renamer",
+        Source.SHAREPOINT_SITE,
+        SITE,
+    )
+    assert harness.subtasks.get(_subtask_id(result)).inputs == body
+
+
+# ---------------------------------------------------------------------------
+# A queue-delivered dry run is actually a dry run
+# ---------------------------------------------------------------------------
+
+
+def test_dry_run_in_an_sqs_record_body_renames_nothing(
+    harness: Harness, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The request must come from the SQS *record body*, not the raw Lambda
+    event. Validating the raw event leaves every unknown key ignored, so
+    ``dry_run`` fell back to False and every queue-triggered run renamed live
+    files — the flag only ever worked on a direct invoke."""
+    # Arrange
+    sharepoint = FakeSharepoint(_one_property_site(_file("Survey.pdf", "id-1")))
+    _install(monkeypatch, tmp_path, sharepoint)
+
+    # Act
+    result = _handler(harness)(
+        _sqs_event({"sharepoint_site": SITE, "dry_run": True}), None
+    )
+
+    # Assert
+    assert sharepoint.renamed == []
+    assert harness.subtasks.get(_subtask_id(result)).outputs == {
+        "result": {
+            "renamed": 0,
+            "would_rename": 1,
+            "skipped_already_canonical": 0,
+            "skipped_images": 0,
+            "missing_folders": [],
+            "failed": [],
+        }
+    }
+
+
+# ---------------------------------------------------------------------------
+# Partial failures complete, they do not fail the run
+# ---------------------------------------------------------------------------
+
+
+def test_run_completes_with_rejected_files_and_missing_folders_listed(
+    harness: Harness, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """One locked file, or a property with no folder, must not turn a
+    productive run red — the failures are the record (ADR-0060)."""
+    # Arrange
+    sharepoint = FakeSharepoint(
+        _one_property_site(_file("Locked.pdf", "id-1"), _file("Survey.pdf", "id-2")),
+        reject={"id-1": PermissionError("locked by another user")},
+    )
+    _install(
+        monkeypatch,
+        tmp_path,
+        sharepoint,
+        rows=[(UPRN, ADDRESS, POSTCODE), ("999", "9 Nowhere Rd", "ZZ9 9ZZ")],
+    )
+
+    # Act
+    result = _handler(harness)(_sqs_event({"sharepoint_site": SITE}), None)
+
+    # Assert
+    assert harness.tasks.get(_task_id(result)).status is TaskStatus.COMPLETE
+    subtask = harness.subtasks.get(_subtask_id(result))
+    assert subtask.status is SubTaskStatus.COMPLETE
+    assert subtask.outputs == {
+        "result": {
+            "renamed": 1,
+            "would_rename": 0,
+            "skipped_already_canonical": 0,
+            "skipped_images": 0,
+            "missing_folders": ["999"],
+            "failed": [
+                {
+                    "uprn": UPRN,
+                    "original_name": "Locked.pdf",
+                    "new_name": f"{UPRN}_{ADDRESS} {POSTCODE}_Locked.pdf",
+                    "error": "locked by another user",
+                }
+            ],
+        }
+    }
+
+
+def test_property_whose_file_was_rejected_completes_rather_than_failing(
+    harness: Harness, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A failed child would turn the whole Task red — any failed sub_task does.
+    A locked file must be recorded on a *completed* property instead."""
+    # Arrange
+    sharepoint = FakeSharepoint(
+        _one_property_site(_file("Locked.pdf", "id-1"), _file("Survey.pdf", "id-2")),
+        reject={"id-1": PermissionError("locked by another user")},
+    )
+    _install(monkeypatch, tmp_path, sharepoint)
+
+    # Act
+    result = _handler(harness)(_sqs_event({"sharepoint_site": SITE}), None)
+
+    # Assert
+    child = _property_subtasks(harness, result)[0]
+    assert child.status is SubTaskStatus.COMPLETE
+    assert (child.outputs or {})["result"]["failed"] == [
+        {
+            "uprn": UPRN,
+            "original_name": "Locked.pdf",
+            "new_name": f"{UPRN}_{ADDRESS} {POSTCODE}_Locked.pdf",
+            "error": "locked by another user",
+        }
+    ]
+    assert harness.tasks.get(_task_id(result)).status is TaskStatus.COMPLETE
+
+
+# ---------------------------------------------------------------------------
+# An unusable site is rejected before anything is touched
+# ---------------------------------------------------------------------------
+
+
+def test_unconfigured_site_fails_the_run_without_touching_sharepoint(
+    harness: Harness, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Sites take their id from the environment at import, so every site this
+    Lambda does not configure collapses into an alias of the first one. An
+    unguarded lookup would return a valid member for the *wrong* site."""
+    # Arrange
+    sharepoint = FakeSharepoint(_one_property_site(_file("Survey.pdf", "id-1")))
+    _install(monkeypatch, tmp_path, sharepoint)
+    monkeypatch.delenv("ECO_SHAREPOINT_ID", raising=False)
+
+    # Act
+    result = _handler(harness)(_sqs_event({"sharepoint_site": "ECO"}), None)
+
+    # Assert
+    subtask = harness.subtasks.get(_subtask_id(result))
+    assert subtask.status is SubTaskStatus.FAILED
+    assert "ECO_SHAREPOINT_ID is unset" in (subtask.outputs or {})["error"]
+    assert sharepoint.renamed == []
+
+
+def test_unrecognised_site_fails_the_run(
+    harness: Harness, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # Arrange
+    sharepoint = FakeSharepoint(_one_property_site(_file("Survey.pdf", "id-1")))
+    _install(monkeypatch, tmp_path, sharepoint)
+
+    # Act
+    result = _handler(harness)(_sqs_event({"sharepoint_site": "NOT_A_SITE"}), None)
+
+    # Assert
+    subtask = harness.subtasks.get(_subtask_id(result))
+    assert subtask.status is SubTaskStatus.FAILED
+    assert (
+        "Unrecognised SharePoint site 'NOT_A_SITE'" in (subtask.outputs or {})["error"]
+    )
+    assert sharepoint.renamed == []

tests/scripts/test_rename_sharepoint_files.py

@@ -8,6 +8,10 @@ from domain.pashub_fetcher.sharepoint_subfolders import SharepointSubfolders
 from orchestration.sharepoint_renamer_orchestrator import (
     ASSESSMENT_SUBFOLDER,
     BASE_PATH,
+    PropertyPlan,
+    RenameFailure,
+    RenameIntent,
+    RenameSummary,
     SharepointRenamerOrchestrator,
     build_canonical_filename,
 )
@@ -39,24 +43,69 @@ def _make_package(name: str) -> dict[str, Any]:
     return {"name": name, "package": {}}
 
 
-def _make_orchestrator(sp: MagicMock, dry_run: bool = False) -> SharepointRenamerOrchestrator:
+def _make_orchestrator(
+    sp: MagicMock, dry_run: bool = False
+) -> SharepointRenamerOrchestrator:
     orchestrator = SharepointRenamerOrchestrator.__new__(SharepointRenamerOrchestrator)
-    orchestrator._sp_client = sp
-    orchestrator._dry_run = dry_run
+    orchestrator._sp_client = sp  # pyright: ignore[reportPrivateUsage]
+    orchestrator._dry_run = dry_run  # pyright: ignore[reportPrivateUsage]
     return orchestrator
 
 
+def _discover_roots(orchestrator: SharepointRenamerOrchestrator) -> list[str]:
+    # Root discovery is protected but is its own unit; the access is declared
+    # once here rather than at the call site.
+    return orchestrator._discover_roots()  # pyright: ignore[reportPrivateUsage]
+
+
+def _one_property_site(
+    tmp_path: Path,
+    folders: dict[str, list[dict[str, Any]]],
+    uprn: str = "100",
+    address: str = "1 High St",
+    postcode: str = "AB1 2CD",
+) -> tuple[MagicMock, str]:
+    """A SharePoint site holding exactly one property, plus its address-list CSV.
+
+    ``folders`` maps a path relative to the property's assessment folder ("" for
+    the folder itself, "SubA" for a subfolder) to its Graph listing. Anything
+    else raises ValueError, as the real client does for a folder that is not
+    there."""
+    assessment = _assessment_path(BASE_PATH, address, postcode)
+    listings: dict[str, dict[str, Any]] = {BASE_PATH: {"value": []}}
+    for relative, items in folders.items():
+        path = assessment if relative == "" else f"{assessment}/{relative}"
+        listings[path] = {"value": items}
+
+    sp = MagicMock()
+
+    def fake_get(path: str) -> dict[str, Any]:
+        if path not in listings:
+            raise ValueError(f"not found: {path}")
+        return listings[path]
+
+    sp.get_folders_in_path.side_effect = fake_get
+    return sp, _write_csv(tmp_path, [(uprn, address, postcode)])
+
+
 # ---------------------------------------------------------------------------
 # build_canonical_filename
 # ---------------------------------------------------------------------------
 
 
 def test_already_canonical_returns_none() -> None:
-    assert build_canonical_filename("100", "1 High St", "AB1 2CD", "100_High St AB1 2CD_Report.pdf") is None
+    assert (
+        build_canonical_filename(
+            "100", "1 High St", "AB1 2CD", "100_High St AB1 2CD_Report.pdf"
+        )
+        is None
+    )
 
 
 def test_strips_address_prefix_and_adds_uprn() -> None:
-    result = build_canonical_filename("100", "1 High St", "AB1 2CD", "1 High St AB1 2CD - Survey.pdf")
+    result = build_canonical_filename(
+        "100", "1 High St", "AB1 2CD", "1 High St AB1 2CD - Survey.pdf"
+    )
     assert result == "100_1 High St AB1 2CD_Survey.pdf"
 
 
@@ -83,7 +132,7 @@ def test_discover_roots_includes_sero_batch_folders() -> None:
     }
 
     # Act
-    roots = _make_orchestrator(sp)._discover_roots()
+    roots = _discover_roots(_make_orchestrator(sp))
 
     # Assert
     assert roots == [
@@ -124,6 +173,83 @@ def test_run_renames_files_for_property_under_batch_root(tmp_path: Path) -> None
     sp.rename_file.assert_called_once_with("id-1", "100_1 High St AB1 2CD_Survey.pdf")
 
 
+def test_plan_lists_each_file_a_property_needs_renamed(tmp_path: Path) -> None:
+    # Arrange
+    csv_path = _write_csv(tmp_path, [("100", "1 High St", "AB1 2CD")])
+    assessment = _assessment_path(BASE_PATH, "1 High St", "AB1 2CD")
+
+    sp = MagicMock()
+
+    def fake_get(path: str) -> dict[str, Any]:
+        if path == BASE_PATH:
+            return {"value": []}
+        if path == assessment:
+            return {
+                "value": [
+                    _make_file("Survey.pdf", "id-1"),
+                    _make_file("Report.docx", "id-2"),
+                ]
+            }
+        raise ValueError("not found")
+
+    sp.get_folders_in_path.side_effect = fake_get
+
+    # Act
+    plans = list(SharepointRenamerOrchestrator(sp, csv_path).plan())
+
+    # Assert
+    assert plans == [
+        PropertyPlan(
+            uprn="100",
+            address="1 High St",
+            postcode="AB1 2CD",
+            intents=[
+                RenameIntent(
+                    uprn="100",
+                    item_id="id-1",
+                    original_name="Survey.pdf",
+                    new_name="100_1 High St AB1 2CD_Survey.pdf",
+                ),
+                RenameIntent(
+                    uprn="100",
+                    item_id="id-2",
+                    original_name="Report.docx",
+                    new_name="100_1 High St AB1 2CD_Report.docx",
+                ),
+            ],
+        )
+    ]
+    sp.rename_file.assert_not_called()
+
+
+def test_run_summary_counts_each_renamed_file(tmp_path: Path) -> None:
+    # Arrange
+    csv_path = _write_csv(tmp_path, [("100", "1 High St", "AB1 2CD")])
+    assessment = _assessment_path(BASE_PATH, "1 High St", "AB1 2CD")
+
+    sp = MagicMock()
+
+    def fake_get(path: str) -> dict[str, Any]:
+        if path == BASE_PATH:
+            return {"value": []}
+        if path == assessment:
+            return {
+                "value": [
+                    _make_file("Survey.pdf", "id-1"),
+                    _make_file("Report.docx", "id-2"),
+                ]
+            }
+        raise ValueError("not found")
+
+    sp.get_folders_in_path.side_effect = fake_get
+
+    # Act
+    summary = SharepointRenamerOrchestrator(sp, csv_path).run()
+
+    # Assert
+    assert summary.renamed == 2
+
+
 def test_run_uprn_prefixed_address_yields_single_uprn_filename(
     tmp_path: Path,
 ) -> None:
@@ -152,94 +278,101 @@ def test_run_uprn_prefixed_address_yields_single_uprn_filename(
 
 
 # ---------------------------------------------------------------------------
-# _process_folder — files only at root level
+# Walking one property's folder tree — driven through the public run()
 # ---------------------------------------------------------------------------
 
 
-def test_renames_top_level_files(caplog: pytest.LogCaptureFixture) -> None:
-    sp = MagicMock()
-    sp.get_folders_in_path.return_value = {
-        "value": [
-            _make_file("Survey.pdf", "id-1"),
-            _make_file("Report.docx", "id-2"),
-        ]
-    }
+def test_renames_every_file_in_a_property_folder(tmp_path: Path) -> None:
+    # Arrange
+    sp, csv_path = _one_property_site(
+        tmp_path,
+        {"": [_make_file("Survey.pdf", "id-1"), _make_file("Report.docx", "id-2")]},
+    )
 
-    _make_orchestrator(sp)._process_folder("some/path", "100", "1 High St", "AB1 2CD")
+    # Act
+    SharepointRenamerOrchestrator(sp, csv_path).run()
 
+    # Assert
     assert sp.rename_file.call_count == 2
     sp.rename_file.assert_any_call("id-1", "100_1 High St AB1 2CD_Survey.pdf")
     sp.rename_file.assert_any_call("id-2", "100_1 High St AB1 2CD_Report.docx")
 
 
-# ---------------------------------------------------------------------------
-# _process_folder — recursive two-level hierarchy
-# ---------------------------------------------------------------------------
-
-
-def test_recurses_into_subfolders_and_renames_all_files() -> None:
-    sp = MagicMock()
-
-    root_contents: dict[str, Any] = {
-        "value": [
-            _make_file("Root.pdf", "root-file"),
-            _make_folder("SubA"),
-        ]
-    }
-    suba_contents: dict[str, Any] = {
-        "value": [
-            _make_file("Sub.pdf", "sub-file"),
-        ]
-    }
-
-    sp.get_folders_in_path.side_effect = lambda path: (
-        root_contents if path == "base/path" else suba_contents
-    )
-
-    _make_orchestrator(sp)._process_folder("base/path", "200", "2 Main Rd", "XY9 8ZW")
-
-    assert sp.rename_file.call_count == 2
-    sp.rename_file.assert_any_call("root-file", "200_2 Main Rd XY9 8ZW_Root.pdf")
-    sp.rename_file.assert_any_call("sub-file", "200_2 Main Rd XY9 8ZW_Sub.pdf")
-
-    sp.get_folders_in_path.assert_any_call("base/path/SubA")
-
-
-# ---------------------------------------------------------------------------
-# _process_folder — non-file, non-folder items are skipped
-# ---------------------------------------------------------------------------
-
-
-def test_ignores_package_items() -> None:
-    sp = MagicMock()
-    sp.get_folders_in_path.return_value = {"value": [_make_package("Notebook")]}
-
-    _make_orchestrator(sp)._process_folder("some/path", "300", "3 Oak Ave", "ZZ1 1ZZ")
-
-    sp.rename_file.assert_not_called()
-    assert sp.get_folders_in_path.call_count == 1
-
-
-# ---------------------------------------------------------------------------
-# _process_folder — missing folder
-# ---------------------------------------------------------------------------
-
-
-def test_missing_folder_returns_false_without_warning(
-    caplog: pytest.LogCaptureFixture,
-) -> None:
+def test_recurses_into_subfolders_and_renames_all_files(tmp_path: Path) -> None:
     # Arrange
-    sp = MagicMock()
-    sp.get_folders_in_path.side_effect = ValueError("not found")
+    sp, csv_path = _one_property_site(
+        tmp_path,
+        {
+            "": [_make_file("Root.pdf", "root-file"), _make_folder("SubA")],
+            "SubA": [_make_file("Sub.pdf", "sub-file")],
+        },
+    )
 
     # Act
-    found = _make_orchestrator(sp)._process_folder(
-        "missing/path", "400", "4 Elm St", "AA2 2BB"
-    )
+    SharepointRenamerOrchestrator(sp, csv_path).run()
+
+    # Assert
+    assert sp.rename_file.call_count == 2
+    sp.rename_file.assert_any_call("root-file", "100_1 High St AB1 2CD_Root.pdf")
+    sp.rename_file.assert_any_call("sub-file", "100_1 High St AB1 2CD_Sub.pdf")
+
+
+def test_subfolder_results_merge_into_the_parent_summary(tmp_path: Path) -> None:
+    # Arrange
+    sp, csv_path = _one_property_site(
+        tmp_path,
+        {
+            "": [_make_file("Root.pdf", "root-file"), _make_folder("SubA")],
+            "SubA": [
+                _make_file("Sub.pdf", "sub-file"),
+                _make_file("100_1 High St AB1 2CD_Done.pdf", "canonical-file"),
+            ],
+        },
+    )
+
+    # Act
+    summary = SharepointRenamerOrchestrator(sp, csv_path).run()
+
+    # Assert
+    assert summary == RenameSummary(renamed=2, skipped_already_canonical=1)
+
+
+def test_ignores_package_items(tmp_path: Path) -> None:
+    # Arrange
+    sp, csv_path = _one_property_site(tmp_path, {"": [_make_package("Notebook")]})
+
+    # Act
+    SharepointRenamerOrchestrator(sp, csv_path).run()
 
     # Assert
-    assert found is False
     sp.rename_file.assert_not_called()
+
+
+def test_property_found_under_a_later_root_is_not_reported_missing(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A miss on an earlier root is silent — only exhausting every root warns."""
+    # Arrange
+    csv_path = _write_csv(tmp_path, [("100", "1 High St", "AB1 2CD")])
+    batch_assessment = _assessment_path(
+        f"{BASE_PATH}/_Sero Batch 2", "1 High St", "AB1 2CD"
+    )
+    sp = MagicMock()
+
+    def fake_get(path: str) -> dict[str, Any]:
+        if path == BASE_PATH:
+            return {"value": [_make_folder("_Sero Batch 2")]}
+        if path == batch_assessment:
+            return {"value": [_make_file("Survey.pdf", "id-1")]}
+        raise ValueError("not found")
+
+    sp.get_folders_in_path.side_effect = fake_get
+
+    # Act
+    summary = SharepointRenamerOrchestrator(sp, csv_path).run()
+
+    # Assert
+    assert summary.missing_folders == []
     assert not any("Missing folder" in r.message for r in caplog.records)
 
 
@@ -267,38 +400,104 @@ def test_run_warns_once_when_property_missing_from_all_roots(
     assert "400" in warnings[0].message
 
 
-# ---------------------------------------------------------------------------
-# _process_folder — already-canonical files are skipped
-# ---------------------------------------------------------------------------
-
-
-def test_skips_already_canonical_files() -> None:
-    sp = MagicMock()
-    sp.get_folders_in_path.return_value = {
-        "value": [_make_file("500_Pine Ln BB3 3CC_Doc.pdf", "id-y")]
-    }
-
-    _make_orchestrator(sp)._process_folder("some/path", "500", "5 Pine Ln", "BB3 3CC")
-
-    sp.rename_file.assert_not_called()
-
-
-# ---------------------------------------------------------------------------
-# _process_folder — dry_run=True logs intent but never calls rename_file
-# ---------------------------------------------------------------------------
-
-
-def test_dry_run_logs_would_rename_without_calling_api(
-    caplog: pytest.LogCaptureFixture,
+def test_run_summary_lists_uprns_with_no_folder_under_any_root(
+    tmp_path: Path,
 ) -> None:
-    sp = MagicMock()
-    sp.get_folders_in_path.return_value = {
-        "value": [_make_file("Survey.pdf", "id-1")]
-    }
+    # Arrange
+    csv_path = _write_csv(
+        tmp_path, [("400", "4 Elm St", "AA2 2BB"), ("500", "5 Pine Ln", "BB3 3CC")]
+    )
+    found_assessment = _assessment_path(BASE_PATH, "5 Pine Ln", "BB3 3CC")
 
-    _make_orchestrator(sp, dry_run=True)._process_folder(
-        "some/path", "100", "1 High St", "AB1 2CD"
+    sp = MagicMock()
+
+    def fake_get(path: str) -> dict[str, Any]:
+        if path == BASE_PATH:
+            return {"value": []}
+        if path == found_assessment:
+            return {"value": []}
+        raise ValueError("not found")
+
+    sp.get_folders_in_path.side_effect = fake_get
+
+    # Act
+    summary = SharepointRenamerOrchestrator(sp, csv_path).run()
+
+    # Assert
+    assert summary.missing_folders == ["400"]
+
+
+# ---------------------------------------------------------------------------
+# Skips, failures and dry runs — driven through the public run()
+# ---------------------------------------------------------------------------
+
+
+def test_rejected_file_is_recorded_with_its_error_and_siblings_still_rename(
+    tmp_path: Path,
+) -> None:
+    # Arrange
+    sp, csv_path = _one_property_site(
+        tmp_path,
+        {"": [_make_file("Locked.pdf", "id-1"), _make_file("Survey.pdf", "id-2")]},
     )
 
+    def fake_rename(item_id: str, _new_name: str) -> None:
+        if item_id == "id-1":
+            raise PermissionError("locked by another user")
+
+    sp.rename_file.side_effect = fake_rename
+
+    # Act
+    summary = SharepointRenamerOrchestrator(sp, csv_path).run()
+
+    # Assert
+    assert summary == RenameSummary(
+        renamed=1,
+        failed=[
+            RenameFailure(
+                uprn="100",
+                original_name="Locked.pdf",
+                new_name="100_1 High St AB1 2CD_Locked.pdf",
+                error="locked by another user",
+            )
+        ],
+    )
+
+
+def test_summary_counts_already_canonical_and_image_skips_apart(tmp_path: Path) -> None:
+    # Arrange
+    sp, csv_path = _one_property_site(
+        tmp_path,
+        {
+            "": [
+                _make_file("100_1 High St AB1 2CD_Doc.pdf", "id-1"),
+                _make_file("Front elevation.JPG", "id-2"),
+                _make_file("Rear elevation.heic", "id-3"),
+            ]
+        },
+    )
+
+    # Act
+    summary = SharepointRenamerOrchestrator(sp, csv_path).run()
+
+    # Assert
+    assert summary == RenameSummary(skipped_already_canonical=1, skipped_images=2)
+    sp.rename_file.assert_not_called()
+
+
+def test_dry_run_logs_intent_and_counts_it_without_calling_the_api(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    # Arrange
+    sp, csv_path = _one_property_site(
+        tmp_path,
+        {"": [_make_file("Survey.pdf", "id-1"), _make_file("Report.docx", "id-2")]},
+    )
+
+    # Act
+    summary = SharepointRenamerOrchestrator(sp, csv_path, dry_run=True).run()
+
+    # Assert
+    assert summary == RenameSummary(renamed=0, would_rename=2)
     sp.rename_file.assert_not_called()
     assert any("Would rename" in r.message for r in caplog.records)

utils/sharepoint/domna_sites.py

@@ -11,3 +11,32 @@ class DomnaSites(Enum):
     SOCIAL_HOUSING_WAVE_3 = os.getenv("SOCIAL_HOUSING_WAVE_3_SHAREPOINT_ID")
     SOCIAL_HOUSING = os.getenv("SOCIAL_HOUSING_SHAREPOINT_ID")
     ECO = os.getenv("ECO_SHAREPOINT_ID")
+
+
+def resolve_site(name: str) -> DomnaSites:
+    """The site `name` refers to, or ValueError if it is not a real, configured
+    site.
+
+    A bare ``DomnaSites[name]`` is not safe here. Members take their value from
+    the environment at import, so every site an app does not configure is None —
+    and equal values make Enum collapse them into *aliases of the first member*.
+    ``DomnaSites["ECO"]`` on a Lambda that only sets
+    ``SOCIAL_HOUSING_WAVE_3_SHAREPOINT_ID`` therefore returns a perfectly valid
+    member for a different site, and the run silently touches the wrong
+    SharePoint. Both failures are rejected here: an unrecognised name, and a
+    recognised name with no configured id.
+
+    Configuration is checked against the environment at call time rather than
+    against ``member.value``, which the alias collapse has already destroyed.
+    """
+    if name not in DomnaSites.__members__:
+        raise ValueError(
+            f"Unrecognised SharePoint site {name!r}. "
+            f"Expected one of: {sorted(DomnaSites.__members__)}"
+        )
+    if not os.environ.get(f"{name}_SHAREPOINT_ID"):
+        raise ValueError(
+            f"SharePoint site {name!r} has no configured site id "
+            f"({name}_SHAREPOINT_ID is unset), so it cannot be resolved."
+        )
+    return DomnaSites[name]