feat: handle multiple orgaisations per resource

[tombrooks248]

What type of PR is this? (check all applicable)

• Refactor
• Feature
• Bug Fix
• Optimization
• Documentation Update

Description

Changing the way the pipeline generates entities from a resource in a situation where multiple endpoints from different organisations provide identical data and therefore produce the same resource.

Before - No entities are produced from resources with multiple orgs.
After - now generate entities per organisation per resource.

Why

Because we are currently missing entities in our data and this is causing a bunch of problems particularly for local-plan data.

How the bug works

A resource is identified only by the hash of its contents. When two organisations publish the same URL, they produce identical bytes, so the collection merges them into a single resource that carries both organisation codes.

To generate entities, the pipeline stamps a default organisation onto every row of a resource (the data files themselves don't say which authority they belong to). But it only does this when there is exactly one organisation:

if len(organisations) == 1:
    default_values["organisation"] = organisations[0]

Worked example — South & Vale Joint Local Plan 2045

South Oxfordshire and Vale of White Horse jointly publish one local plan timetable at a single URL:

https://www.southandvale.gov.uk/app/uploads/2026/06/Local-Plan-Timetable-CSV.csv

In source.csv that one endpoint is attributed to both authorities:

organisation	dataset
local-authority:SOX	plan-timetable
local-authority:VAL	plan-timetable

lookup.csv already defines separate entities for each authority against the same references, for example:

dataset	reference	SOX entity	VAL entity
plan-timetable	Joint-Local-Plan-2045-scoping-consultation-start	5110200	5110188
local-plan	south-and-vale-joint-local-plan-2045	4221014	4221015

Before: the resource carries two organisations, the guard is skipped, and neither authority's entities are created — both show zero records supplied.

After: the resource is processed once per organisation (each set as the default in turn), both lookups resolve, and both authorities get their entities.

Related Tickets & Documents

• Ticket Link
• Related Issue #
• Closes #

QA Instructions, Screenshots, Recordings

Downstream impacts of this change.

Processing a resource once per organisation means a single resource now produces rows for more than one organisation. I looked at every step that runs after transform to check nothing relies on the old "one organisation per resource" assumption.

Handled by this change

• Duplicate-reference check. This flags a reference appearing more than once in a resource. With two organisations, the same reference legitimately appears for each (e.g. the same joint-plan milestone), which would be falsely flagged as "not unique". We now run this check per organisation pass, where references are genuinely expected to be unique, so no false positives are introduced.

• Doubled logs. The org-independent side-logs (column-field, converted-resource) would otherwise be written once per pass. We write the converted/harmonised artefacts on the first pass only and de-duplicate those log rows, so a multi-organisation resource produces the same single set of log rows as any other.

• Issue → entity attribution. Issues are logged before the entity is known and back-filled afterwards. We apply and reset that mapping per pass so each organisation's issues attach to that organisation's entities rather than the other's.

Not affected (checked, no change needed)

• Dataset / entity assembly. Entities are assembled by grouping on the entity number, not on the resource. Each organisation resolves to its own entity, so the two organisations' rows become two distinct entities — exactly as if they'd come from separate resources. The organisation is carried through as a normal field.

• Priority. Priority is decided per entity from that entity's authoritative organisation. Each organisation owns its own entity, so both resolve correctly.

• Task generation. Tasks are keyed by dataset/resource/field/issue-type/organisation and already expand a resource's organisations into one task per organisation. They take the organisation from the resource, not from anything this change touches, so they are unaffected.

• Single / no-organisation resources. The vast majority of resources have one organisation (or none). Those follow the same path as before — this change only alters behaviour when a resource has two or more organisations.

Testing in Staging

I deployed this to staging and ran the local-plan collection through to the plan-timetable Postgres load. Before this change the South & Vale Joint Local Plan produced no entities; after it, querying the staging entity table returns all 12 plan-timetable milestones twice — once for South Oxfordshire and once for Vale of White Horse — with the exact entity numbers defined in lookup.csv (e.g. Joint-Local-Plan-2045-scoping-consultation-start → 5110200 for South Oxfordshire and 5110188 for Vale of White Horse). That confirms both authorities now get their own entities from the single shared resource, end to end through transform, dataset build and the Postgres loader.

Why this is safer than it first looks

Changing how resources map to organisations sounds like it touches a core assumption, but the pipeline is already entity-centric below the resource level — facts, dataset assembly and tasks are all keyed on the entity, not the resource. So "process a resource once per organisation" fits the existing model rather than fighting it: each organisation resolves to its own entity, and from that point on everything treats the two organisations exactly as if they'd come from separate resources.

The load-bearing detail is that a fact's identity is a hash of entity : field : value, so two organisations producing identical rows still create distinct facts (different entities → different hashes). That's why nothing collapses or overwrites. I also confirmed the resource-level logs are set, not accumulated, so nothing double-counts.

The change is also naturally bounded: resources with one organisation (or none) follow exactly the same path as before — only resources with two or more organisations hit the new behaviour. Which is a relatively small subset of the total data.

Added/updated tests?

We encourage you to keep the code coverage percentage at 80% and above. Please refer to the Digital Land Testing Guidance for more information.

• Yes
• No, and this is why: please replace this line with details on why tests
  have not been included
• I need help with writing tests

[optional] Are there any post deployment tasks we need to perform?

[optional] Are there any dependencies on other PRs or Work?

Summary by CodeRabbit

• New Features

  • Multi-organisation resource processing now combines results from each organisation into one final output.
  • Output handling is more consistent across single-, zero-, and multi-organisation runs.

• Bug Fixes

  • Duplicate log entries are now reduced in the final output.
  • Organisation-specific issues no longer leak between processing passes, improving accuracy in multi-organisation runs.

[coderabbitai]

Walkthrough

Pipeline.transform now runs once per organisation pass, writes per-pass outputs, then concatenates them into the final CSV and deduplicates related logs. Phase construction is split into a pure builder. Integration tests cover helper behaviour and single-, zero-, and multi-organisation runs.

Changes

Multi-org transform orchestration

Layer / File(s)	Summary
Concat and dedup helper functions digital_land/pipeline/main.py	Adds _concat_transformed to merge per-org CSVs under one header and remove intermediate inputs, plus _dedup_log_rows to collapse duplicate log rows while preserving order.
Phase builder refactor digital_land/pipeline/main.py	_build_transform_phases becomes a pure per-pass builder that computes defaults and inputs, wires PriorityPhase with providers=providers, and returns the phase list instead of running the pipeline.
Transform execution flow with per-org passes digital_land/pipeline/main.py	transform initialises logs once, iterates organisation passes, runs duplicate-reference checks per pass, resets issue-log entity mapping between passes, concatenates outputs, deduplicates side-logs, and finalises status and entity counts.
Integration tests for multi-org transform tests/integration/pipeline/test_transform_multi_org.py	Adds helper tests, a pipeline fixture, an _run_transform harness, and scenario tests for single-org, zero-org, and multi-org runs covering pass counts, output paths, providers, artefacts, and log resets.

Estimated code review effort: 4 (Complex) | ~60 minutes

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title is concise and clearly matches the main change: handling multiple organisations per resource.

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/handle-mult-orgs-per-resource

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

🧹 Nitpick comments (2)

tests/integration/pipeline/test_transform_multi_org.py (2)

197-198: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Clarify the "(Option C)" reference.

The comment refers to a design option not documented anywhere in this file, making the rationale opaque to future readers.

✏️ Suggested clarification

-    # the entry-number -> entity map is reset between/after passes (Option C)
+    # the entry-number -> entity map is reset after each pass so entity
+    # attribution stays unambiguous for the next organisation's pass
     assert pipeline.issue_log.entry_to_entity == {}

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/integration/pipeline/test_transform_multi_org.py` around lines 197 -
198, Clarify the ambiguous “(Option C)” note in the test by removing the
undocumented option label or replacing it with a concrete description of the
behavior being asserted. Update the nearby comment around the pipeline issue log
reset so it explains, in plain terms, that issue_log.entry_to_entity is expected
to be empty after each pass and why that matters, without referencing any
external design option.

---

86-133: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Consider adding coverage for the combine_fields gating branch.

_run_transform never exercises the case where combine_fields != {}, which skips duplicate_reference_check per Pipeline.transform (main.py:621-749). Current tests only implicitly cover the default (empty) path.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/integration/pipeline/test_transform_multi_org.py` around lines 86 -
133, The current multi-org transform helper only covers the empty combine_fields
path, so add a test case through _run_transform/Pipeline.transform that passes a
non-empty combine_fields value and verifies the combine_fields gating branch is
taken. In that scenario, assert duplicate_reference_check is not invoked while
the rest of the transform orchestration still runs, using the existing fake
_build_transform_phases and transform() setup to keep the coverage focused.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@tests/integration/pipeline/test_transform_multi_org.py`:
- Around line 197-198: Clarify the ambiguous “(Option C)” note in the test by
removing the undocumented option label or replacing it with a concrete
description of the behavior being asserted. Update the nearby comment around the
pipeline issue log reset so it explains, in plain terms, that
issue_log.entry_to_entity is expected to be empty after each pass and why that
matters, without referencing any external design option.
- Around line 86-133: The current multi-org transform helper only covers the
empty combine_fields path, so add a test case through
_run_transform/Pipeline.transform that passes a non-empty combine_fields value
and verifies the combine_fields gating branch is taken. In that scenario, assert
duplicate_reference_check is not invoked while the rest of the transform
orchestration still runs, using the existing fake _build_transform_phases and
transform() setup to keep the coverage focused.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: ce99e12e-024c-4f7c-a996-1a0bb22b93fe

📥 Commits

Reviewing files that changed from the base of the PR and between 992bfe8 and ea47e10.

📒 Files selected for processing (2)

• digital_land/pipeline/main.py
• tests/integration/pipeline/test_transform_multi_org.py