From 11c327b63d1d788ac90c6fb03412edbb42271068 Mon Sep 17 00:00:00 2001
From: Tom Brooks <tombrooks248@gmail.com>
Date: Mon, 13 Jul 2026 15:12:55 +0100
Subject: [PATCH 1/7] feat: handle multiple organisation pser resource

---
 digital_land/check.py           |   7 +-
 digital_land/phase/lookup.py    | 104 ++++++++++++++++++++++-
 digital_land/pipeline/main.py   |   1 +
 tests/unit/phase/test_lookup.py | 142 ++++++++++++++++++++++++++++++++
 tests/unit/test_check.py        |  19 +++++
 5 files changed, 268 insertions(+), 5 deletions(-)

diff --git a/digital_land/check.py b/digital_land/check.py
index 9f043e7ac..8b64119b3 100644
--- a/digital_land/check.py
+++ b/digital_land/check.py
@@ -1,6 +1,7 @@
 import duckdb
 import logging
 import dask.dataframe as dd
+import pandas as pd
 
 
 # TODO This might need to move into expectations as it is a form of data checking
@@ -22,10 +23,11 @@ def duplicate_reference_check(issues=None, csv_path=None):
             "field",
             "value",
             "entry_date",
+            "entity",
             COUNT(*) AS count,
             STRING_AGG("entry_number"::TEXT, ',') AS entry_numbers
         FROM filtered_table
-        GROUP BY "field", "value", "entry_date"
+        GROUP BY "field", "value", "entry_date", "entity"
         HAVING COUNT(*) > 1;
         """
 
@@ -34,11 +36,14 @@ def duplicate_reference_check(issues=None, csv_path=None):
         if len(count_table) >= 1:
             duplicate_references = count_table[count_table["count"] > 1]
             for idx, row in duplicate_references.iterrows():
+                entity = row["entity"]
+                entity = int(entity) if pd.notna(entity) else None
                 for entry_number in row["entry_numbers"].split(","):
                     issues.log_issue(
                         "reference",
                         "reference values are not unique",
                         row["value"],
+                        entity=entity,
                         entry_number=int(entry_number),
                         line_number=int(entry_number)
                         + 1,  # TODO Check this makes sense in all cases
diff --git a/digital_land/phase/lookup.py b/digital_land/phase/lookup.py
index c5dfb712d..fa962c2fc 100644
--- a/digital_land/phase/lookup.py
+++ b/digital_land/phase/lookup.py
@@ -84,14 +84,14 @@ def check_associated_organisation(self, entity):
                     return ""
         return entity
 
-    def get_entity(self, block):
+    def get_entity(self, block, organisation=None):
         row = block["row"]
         prefix = row.get("prefix", "")
         reference = row.get("reference", "")
         # Eventually won't be needed but leave in for now
-        organisation = row.get("organisation", "").replace(
-            "local-authority-eng", "local-authority"
-        )
+        if organisation is None:
+            organisation = row.get("organisation", "")
+        organisation = organisation.replace("local-authority-eng", "local-authority")
         entry_number = block["entry-number"]
 
         entity = (
@@ -143,6 +143,31 @@ def redirect_entity(self, entity):
                     return ""
         return entity
 
+    def _log_unknown_entity(self, reference, curie, line_number):
+        if not self.issues:
+            return
+        if not reference:
+            self.issues.log_issue(
+                "entity",
+                "unknown entity - missing reference",
+                curie,
+                line_number=line_number,
+            )
+        else:
+            self.issues.log_issue(
+                "entity",
+                "unknown entity",
+                curie,
+                line_number=line_number,
+            )
+            if self.operational_issues:
+                self.operational_issues.log_issue(
+                    "entity",
+                    "unknown entity",
+                    curie,
+                    line_number=line_number,
+                )
+
     def process(self, stream):
         for block in stream:
             row = block["row"]
@@ -189,7 +214,36 @@ def process(self, stream):
 class EntityLookupPhase(LookupPhase):
     entity_field = "entity"
 
+    def __init__(
+        self,
+        lookups={},
+        redirect_lookups={},
+        issue_log=None,
+        operational_issue_log=None,
+        entity_range=[],
+        lookup_rules=None,
+        organisations=None,
+    ):
+        super().__init__(
+            lookups=lookups,
+            redirect_lookups=redirect_lookups,
+            issue_log=issue_log,
+            operational_issue_log=operational_issue_log,
+            entity_range=entity_range,
+            lookup_rules=lookup_rules,
+        )
+        self.organisations = organisations or []
+
     def process(self, stream):
+        # A resource may be associated with more than one organisation (e.g. a
+        # joint local plan shared by two authorities). In that case the entity
+        # is resolved once per organisation and a row is emitted per distinct
+        # entity. With a single organisation (or none) each row resolves to at
+        # most one entity.
+        if len(self.organisations) > 1:
+            yield from self._process_multi_org(stream)
+            return
+
         for block in super().process(stream):
             if self.issues:
                 self.issues.record_entity_map(
@@ -197,6 +251,48 @@ def process(self, stream):
                 )
             yield block
 
+    def _process_multi_org(self, stream):
+        for block in stream:
+            row = block["row"]
+            prefix = row.get("prefix", "")
+            reference = row.get("reference", "")
+            curie = f"{prefix}:{reference}"
+            line_number = block["line-number"]
+
+            # Nothing to look up if there is no prefix, or an entity has
+            # already been assigned upstream.
+            if not prefix or row.get(self.entity_field, ""):
+                if self.issues:
+                    self.issues.record_entity_map(
+                        block["entry-number"], row.get(self.entity_field, "")
+                    )
+                yield block
+                continue
+
+            emitted_entities = set()
+            for organisation in self.organisations:
+                entity = self.get_entity(block, organisation=organisation)
+
+                if not entity:
+                    # no entity for this organisation: log an unknown-entity
+                    # issue and emit no row for it
+                    self._log_unknown_entity(reference, curie, line_number)
+                    continue
+
+                entity = self.redirect_entity(entity)
+                if not entity or entity in emitted_entities:
+                    continue
+                emitted_entities.add(entity)
+
+                new_block = dict(block)
+                new_block["row"] = dict(row)
+                new_block["row"][self.entity_field] = entity
+                new_block["row"]["organisation"] = organisation
+                new_block["organisation"] = organisation
+                if self.issues:
+                    self.issues.record_entity_map(block["entry-number"], entity)
+                yield new_block
+
 
 class FactLookupPhase(LookupPhase):
     def __init__(
diff --git a/digital_land/pipeline/main.py b/digital_land/pipeline/main.py
index c259ce248..9da53a6d8 100644
--- a/digital_land/pipeline/main.py
+++ b/digital_land/pipeline/main.py
@@ -713,6 +713,7 @@ def transform(
                         operational_issue_log=self.operational_issue_log,
                         entity_range=[entity_range_min, entity_range_max],
                         lookup_rules=lookup_rules,
+                        organisations=organisations,
                     ),
                 ]
             )
diff --git a/tests/unit/phase/test_lookup.py b/tests/unit/phase/test_lookup.py
index 7fe963fab..18fc2c828 100644
--- a/tests/unit/phase/test_lookup.py
+++ b/tests/unit/phase/test_lookup.py
@@ -315,6 +315,148 @@ def test_entity_lookup_phase_blank(self):
 
         assert len(output) == 0
 
+    def test_multi_org_fans_out_to_distinct_entities(self):
+        # One joint-plan row, two orgs, two different entities -> two rows out,
+        # each stamped with the org it was looked up with (row + block level).
+        input_stream = [
+            {
+                "row": {
+                    "prefix": "local-plan",
+                    "reference": "owen",
+                    "organisation": "",
+                },
+                "entry-number": 1,
+                "line-number": 2,
+            }
+        ]
+        lookups = {
+            ",local-plan,owen,local-authorityoe": "101",
+            ",local-plan,owen,local-authoritytb": "102",
+        }
+        phase = EntityLookupPhase(
+            lookups=lookups,
+            organisations=["local-authority:OE", "local-authority:TB"],
+        )
+        output = [block for block in phase.process(input_stream)]
+
+        assert len(output) == 2
+        entity_to_org = {b["row"]["entity"]: b["row"]["organisation"] for b in output}
+        assert entity_to_org == {
+            "101": "local-authority:OE",
+            "102": "local-authority:TB",
+        }
+        # org also stamped at block level (FactLookupPhase reads this post-pivot)
+        assert {b["organisation"] for b in output} == {
+            "local-authority:OE",
+            "local-authority:TB",
+        }
+
+    def test_multi_org_same_entity_produces_single_row(self):
+        # Both orgs resolve to the same entity -> one row, no dedupe-after needed.
+        input_stream = [
+            {
+                "row": {
+                    "prefix": "local-plan",
+                    "reference": "owen",
+                    "organisation": "",
+                },
+                "entry-number": 1,
+                "line-number": 2,
+            }
+        ]
+        lookups = {
+            ",local-plan,owen,local-authorityoe": "101",
+            ",local-plan,owen,local-authoritytb": "101",
+        }
+        phase = EntityLookupPhase(
+            lookups=lookups,
+            organisations=["local-authority:OE", "local-authority:TB"],
+        )
+        output = [block for block in phase.process(input_stream)]
+
+        assert len(output) == 1
+        assert output[0]["row"]["entity"] == "101"
+        # first org to resolve the entity wins
+        assert output[0]["row"]["organisation"] == "local-authority:OE"
+
+    def test_multi_org_one_missing_raises_issue_and_emits_found(self):
+        # OE resolves, TB has no lookup -> one row for OE + one unknown-entity issue.
+        input_stream = [
+            {
+                "row": {
+                    "prefix": "local-plan",
+                    "reference": "owen",
+                    "organisation": "",
+                },
+                "entry-number": 1,
+                "line-number": 2,
+            }
+        ]
+        lookups = {",local-plan,owen,local-authorityoe": "101"}
+        issues = IssueLog()
+        phase = EntityLookupPhase(
+            lookups=lookups,
+            issue_log=issues,
+            organisations=["local-authority:OE", "local-authority:TB"],
+        )
+        output = [block for block in phase.process(input_stream)]
+
+        assert len(output) == 1
+        assert output[0]["row"]["entity"] == "101"
+        assert output[0]["row"]["organisation"] == "local-authority:OE"
+        assert [i["issue-type"] for i in issues.rows] == ["unknown entity"]
+
+    def test_multi_org_neither_found_emits_nothing_and_raises_issues(self):
+        input_stream = [
+            {
+                "row": {
+                    "prefix": "local-plan",
+                    "reference": "owen",
+                    "organisation": "",
+                },
+                "entry-number": 1,
+                "line-number": 2,
+            }
+        ]
+        lookups = {}
+        issues = IssueLog()
+        phase = EntityLookupPhase(
+            lookups=lookups,
+            issue_log=issues,
+            organisations=["local-authority:OE", "local-authority:TB"],
+        )
+        output = [block for block in phase.process(input_stream)]
+
+        assert output == []
+        # one unknown-entity issue per organisation that missed
+        assert [i["issue-type"] for i in issues.rows] == [
+            "unknown entity",
+            "unknown entity",
+        ]
+
+    def test_single_org_uses_unchanged_path(self):
+        # len(organisations) <= 1 must not fan out; org read from the row as before.
+        input_stream = [
+            {
+                "row": {
+                    "prefix": "local-plan",
+                    "reference": "owen",
+                    "organisation": "local-authority:OE",
+                },
+                "entry-number": 1,
+                "line-number": 2,
+            }
+        ]
+        lookups = {",local-plan,owen,local-authorityoe": "101"}
+        phase = EntityLookupPhase(
+            lookups=lookups,
+            organisations=["local-authority:OE"],
+        )
+        output = [block for block in phase.process(input_stream)]
+
+        assert len(output) == 1
+        assert output[0]["row"]["entity"] == "101"
+
 
 class TestFactLookupPhase:
     def test_missing_associated_entity_issue_raised(
diff --git a/tests/unit/test_check.py b/tests/unit/test_check.py
index 383d0d75c..3bf91c81e 100644
--- a/tests/unit/test_check.py
+++ b/tests/unit/test_check.py
@@ -109,6 +109,25 @@ def test_duplicate_reference_check(tmp_path):
                 },
             ],
         ),
+        (
+            "different_entity.csv",
+            [
+                {
+                    "entity": 7010002598,
+                    "entry-date": "2024-07-19",
+                    "entry-number": 1,
+                    "field": "reference",
+                    "value": "ref1",
+                },
+                {
+                    "entity": 7010002599,
+                    "entry-date": "2024-07-19",
+                    "entry-number": 1,
+                    "field": "reference",
+                    "value": "ref1",
+                },
+            ],
+        ),
         (
             "no_references.csv",
             [

From 27ecdf94c3a6e0fce149bc483af36c6c84505bda Mon Sep 17 00:00:00 2001
From: Tom Brooks <tombrooks248@gmail.com>
Date: Wed, 15 Jul 2026 12:22:36 +0100
Subject: [PATCH 2/7] adding some docstrings

---
 digital_land/phase/lookup.py  | 47 +++++++++++++++++++++++++++++++++++
 digital_land/pipeline/main.py |  4 ++-
 2 files changed, 50 insertions(+), 1 deletion(-)

diff --git a/digital_land/phase/lookup.py b/digital_land/phase/lookup.py
index fa962c2fc..459e4f019 100644
--- a/digital_land/phase/lookup.py
+++ b/digital_land/phase/lookup.py
@@ -85,6 +85,17 @@ def check_associated_organisation(self, entity):
         return entity
 
     def get_entity(self, block, organisation=None):
+        """Look up the entity for a block's prefix and reference.
+
+        Args:
+            organisation (str, optional): Organisation to resolve the entity against.
+                Defaults to None, meaning use the organisation already on the row.
+                EntityLookupPhase passes this explicitly when fanning out a
+                multi-organisation resource, where the row's organisation is blank.
+
+        Returns:
+            str: The entity number, or "" if nothing matched.
+        """
         row = block["row"]
         prefix = row.get("prefix", "")
         reference = row.get("reference", "")
@@ -144,6 +155,7 @@ def redirect_entity(self, entity):
         return entity
 
     def _log_unknown_entity(self, reference, curie, line_number):
+        """Raise an unknown-entity issue, distinguishing a missing reference."""
         if not self.issues:
             return
         if not reference:
@@ -212,6 +224,21 @@ def process(self, stream):
 
 
 class EntityLookupPhase(LookupPhase):
+    """
+    lookup the entity for an entry's reference
+
+    A resource is identified by the hash of its contents, so when several
+    organisations publish identical data the collection merges them into one
+    resource carrying every organisation's code (e.g. a joint local plan, or a
+    brownfield register published by two authorities). Those rows arrive here
+    with a blank organisation, because the resource-level default is only
+    applied when the resource has exactly one organisation.
+
+    When given more than one organisation this phase resolves the entity once
+    per organisation and emits a row per distinct entity. With one organisation
+    (or none) it behaves exactly as it always has.
+    """
+
     entity_field = "entity"
 
     def __init__(
@@ -224,6 +251,13 @@ def __init__(
         lookup_rules=None,
         organisations=None,
     ):
+        """
+        Args:
+            organisations (list, optional): Organisation codes associated with the
+                resource. More than one means the resource is shared, and the entity
+                is resolved once per organisation. Defaults to None (no fan-out).
+        """
+
         super().__init__(
             lookups=lookups,
             redirect_lookups=redirect_lookups,
@@ -252,6 +286,19 @@ def process(self, stream):
             yield block
 
     def _process_multi_org(self, stream):
+        """Resolve the entity once per organisation and fan out the row.
+
+        For each entry the reference is looked up against every organisation on
+        the resource. Distinct entities produce distinct rows, each stamped with
+        the organisation it was looked up with; if two organisations resolve to
+        the same entity only one row is emitted. An organisation that resolves
+        nothing raises an unknown-entity issue and produces no row, so an entry
+        may yield fewer rows than there are organisations — including none.
+
+        The organisation is set on the block as well as the row because
+        FactLookupPhase reads it after the pivot, once row-level fields are gone.
+        """
+
         for block in stream:
             row = block["row"]
             prefix = row.get("prefix", "")
diff --git a/digital_land/pipeline/main.py b/digital_land/pipeline/main.py
index 9da53a6d8..a6f760018 100644
--- a/digital_land/pipeline/main.py
+++ b/digital_land/pipeline/main.py
@@ -605,7 +605,9 @@ def transform(
             resource (str): Resource file identifier (hash), TBD can be removed.
             valid_category_values (dict): Dictionary of valid category values per field from the API/specification.
             endpoints (list, optional): List of endpoint hashes/identifiers for this resource. Defaults to None.
-            organisations (list, optional): List of organisation codes/identifiers associated with the resource. Defaults to None, Note if one passed, org will become default value
+            organisations (list, optional): List of organisation codes/identifiers associated with the resource. Defaults to None.
+                If one is passed it becomes the default organisation for every row; if more than one, no default is applied and
+                EntityLookupPhase resolves the entity once per organisation.
             entry_date (str, optional): Default entry-date value to apply to all records. Defaults to "".
             converted_path (str, optional): Path to save converted (pre-normalised) resource. Defaults to None.
             harmonised_output_path (str, optional): Path to save the harmonised/intermediate output. Defaults to None.

From d4ff0fcc6f445b3fde58f32811bf2b46ac484aa2 Mon Sep 17 00:00:00 2001
From: Tom Brooks <tombrooks248@gmail.com>
Date: Wed, 15 Jul 2026 15:12:16 +0100
Subject: [PATCH 3/7] updating docs and renaming to provider

---
 digital_land/phase/lookup.py    | 14 ++++++++------
 digital_land/pipeline/main.py   |  2 +-
 tests/unit/phase/test_lookup.py | 12 ++++++------
 3 files changed, 15 insertions(+), 13 deletions(-)

diff --git a/digital_land/phase/lookup.py b/digital_land/phase/lookup.py
index 459e4f019..294507fd7 100644
--- a/digital_land/phase/lookup.py
+++ b/digital_land/phase/lookup.py
@@ -91,7 +91,9 @@ def get_entity(self, block, organisation=None):
             organisation (str, optional): Organisation to resolve the entity against.
                 Defaults to None, meaning use the organisation already on the row.
                 EntityLookupPhase passes this explicitly when fanning out a
-                multi-organisation resource, where the row's organisation is blank.
+                multi-organisation resource, where the row's organisation is blank
+                because the resource-level default is only applied when a resource
+                has exactly one organisation.
 
         Returns:
             str: The entity number, or "" if nothing matched.
@@ -249,11 +251,11 @@ def __init__(
         operational_issue_log=None,
         entity_range=[],
         lookup_rules=None,
-        organisations=None,
+        providers=None,
     ):
         """
         Args:
-            organisations (list, optional): Organisation codes associated with the
+            providers (list, optional): Organisation codes that provided the
                 resource. More than one means the resource is shared, and the entity
                 is resolved once per organisation. Defaults to None (no fan-out).
         """
@@ -266,7 +268,7 @@ def __init__(
             entity_range=entity_range,
             lookup_rules=lookup_rules,
         )
-        self.organisations = organisations or []
+        self.providers = providers or []
 
     def process(self, stream):
         # A resource may be associated with more than one organisation (e.g. a
@@ -274,7 +276,7 @@ def process(self, stream):
         # is resolved once per organisation and a row is emitted per distinct
         # entity. With a single organisation (or none) each row resolves to at
         # most one entity.
-        if len(self.organisations) > 1:
+        if len(self.providers) > 1:
             yield from self._process_multi_org(stream)
             return
 
@@ -317,7 +319,7 @@ def _process_multi_org(self, stream):
                 continue
 
             emitted_entities = set()
-            for organisation in self.organisations:
+            for organisation in self.providers:
                 entity = self.get_entity(block, organisation=organisation)
 
                 if not entity:
diff --git a/digital_land/pipeline/main.py b/digital_land/pipeline/main.py
index a6f760018..829333e2f 100644
--- a/digital_land/pipeline/main.py
+++ b/digital_land/pipeline/main.py
@@ -715,7 +715,7 @@ def transform(
                         operational_issue_log=self.operational_issue_log,
                         entity_range=[entity_range_min, entity_range_max],
                         lookup_rules=lookup_rules,
-                        organisations=organisations,
+                        providers=organisations,
                     ),
                 ]
             )
diff --git a/tests/unit/phase/test_lookup.py b/tests/unit/phase/test_lookup.py
index 18fc2c828..2728243e0 100644
--- a/tests/unit/phase/test_lookup.py
+++ b/tests/unit/phase/test_lookup.py
@@ -335,7 +335,7 @@ def test_multi_org_fans_out_to_distinct_entities(self):
         }
         phase = EntityLookupPhase(
             lookups=lookups,
-            organisations=["local-authority:OE", "local-authority:TB"],
+            providers=["local-authority:OE", "local-authority:TB"],
         )
         output = [block for block in phase.process(input_stream)]
 
@@ -370,7 +370,7 @@ def test_multi_org_same_entity_produces_single_row(self):
         }
         phase = EntityLookupPhase(
             lookups=lookups,
-            organisations=["local-authority:OE", "local-authority:TB"],
+            providers=["local-authority:OE", "local-authority:TB"],
         )
         output = [block for block in phase.process(input_stream)]
 
@@ -397,7 +397,7 @@ def test_multi_org_one_missing_raises_issue_and_emits_found(self):
         phase = EntityLookupPhase(
             lookups=lookups,
             issue_log=issues,
-            organisations=["local-authority:OE", "local-authority:TB"],
+            providers=["local-authority:OE", "local-authority:TB"],
         )
         output = [block for block in phase.process(input_stream)]
 
@@ -423,7 +423,7 @@ def test_multi_org_neither_found_emits_nothing_and_raises_issues(self):
         phase = EntityLookupPhase(
             lookups=lookups,
             issue_log=issues,
-            organisations=["local-authority:OE", "local-authority:TB"],
+            providers=["local-authority:OE", "local-authority:TB"],
         )
         output = [block for block in phase.process(input_stream)]
 
@@ -435,7 +435,7 @@ def test_multi_org_neither_found_emits_nothing_and_raises_issues(self):
         ]
 
     def test_single_org_uses_unchanged_path(self):
-        # len(organisations) <= 1 must not fan out; org read from the row as before.
+        # len(providers) <= 1 must not fan out; org read from the row as before.
         input_stream = [
             {
                 "row": {
@@ -450,7 +450,7 @@ def test_single_org_uses_unchanged_path(self):
         lookups = {",local-plan,owen,local-authorityoe": "101"}
         phase = EntityLookupPhase(
             lookups=lookups,
-            organisations=["local-authority:OE"],
+            providers=["local-authority:OE"],
         )
         output = [block for block in phase.process(input_stream)]
 

From acf9f7b3d5d8713adf119c6a66cbf59b827ff698 Mon Sep 17 00:00:00 2001
From: Tom Brooks <tombrooks248@gmail.com>
Date: Wed, 15 Jul 2026 15:36:02 +0100
Subject: [PATCH 4/7] moving the block loop and skip guard up into process

---
 digital_land/phase/lookup.py | 111 +++++++++++++++++------------------
 1 file changed, 55 insertions(+), 56 deletions(-)

diff --git a/digital_land/phase/lookup.py b/digital_land/phase/lookup.py
index 294507fd7..bed8917e2 100644
--- a/digital_land/phase/lookup.py
+++ b/digital_land/phase/lookup.py
@@ -271,46 +271,25 @@ def __init__(
         self.providers = providers or []
 
     def process(self, stream):
-        # A resource may be associated with more than one organisation (e.g. a
-        # joint local plan shared by two authorities). In that case the entity
-        # is resolved once per organisation and a row is emitted per distinct
-        # entity. With a single organisation (or none) each row resolves to at
-        # most one entity.
-        if len(self.providers) > 1:
-            yield from self._process_multi_org(stream)
+        # With a single provider (or none) the row already carries the
+        # organisation, so the inherited behaviour is used unchanged. Only a
+        # resource shared by several organisations (e.g. a joint local plan)
+        # needs the entity resolving once per organisation.
+        if len(self.providers) <= 1:
+            for block in super().process(stream):
+                if self.issues:
+                    self.issues.record_entity_map(
+                        block["entry-number"], block["row"]["entity"]
+                    )
+                yield block
             return
 
-        for block in super().process(stream):
-            if self.issues:
-                self.issues.record_entity_map(
-                    block["entry-number"], block["row"]["entity"]
-                )
-            yield block
-
-    def _process_multi_org(self, stream):
-        """Resolve the entity once per organisation and fan out the row.
-
-        For each entry the reference is looked up against every organisation on
-        the resource. Distinct entities produce distinct rows, each stamped with
-        the organisation it was looked up with; if two organisations resolve to
-        the same entity only one row is emitted. An organisation that resolves
-        nothing raises an unknown-entity issue and produces no row, so an entry
-        may yield fewer rows than there are organisations — including none.
-
-        The organisation is set on the block as well as the row because
-        FactLookupPhase reads it after the pivot, once row-level fields are gone.
-        """
-
         for block in stream:
             row = block["row"]
-            prefix = row.get("prefix", "")
-            reference = row.get("reference", "")
-            curie = f"{prefix}:{reference}"
-            line_number = block["line-number"]
 
             # Nothing to look up if there is no prefix, or an entity has
             # already been assigned upstream.
-            if not prefix or row.get(self.entity_field, ""):
+            if not row.get("prefix", "") or row.get(self.entity_field, ""):
                 if self.issues:
                     self.issues.record_entity_map(
                         block["entry-number"], row.get(self.entity_field, "")
@@ -318,29 +297,49 @@ def _process_multi_org(self, stream):
                 yield block
                 continue
 
-            emitted_entities = set()
-            for organisation in self.providers:
-                entity = self.get_entity(block, organisation=organisation)
-
-                if not entity:
-                    # no entity for this organisation: log an unknown-entity
-                    # issue and emit no row for it
-                    self._log_unknown_entity(reference, curie, line_number)
-                    continue
-
-                entity = self.redirect_entity(entity)
-                if not entity or entity in emitted_entities:
-                    continue
-                emitted_entities.add(entity)
-
-                new_block = dict(block)
-                new_block["row"] = dict(row)
-                new_block["row"][self.entity_field] = entity
-                new_block["row"]["organisation"] = organisation
-                new_block["organisation"] = organisation
-                if self.issues:
-                    self.issues.record_entity_map(block["entry-number"], entity)
-                yield new_block
+            yield from self._fan_out(block)
+
+    def _fan_out(self, block):
+        """Resolve the entity once per provider, yielding a row per entity.
+
+        Distinct entities produce distinct rows, each stamped with the
+        organisation it was looked up with; if two providers resolve to the same
+        entity only one row is emitted. A provider that resolves nothing raises
+        an unknown-entity issue and produces no row, so an entry may yield fewer
+        rows than there are providers — including none.
+
+        The organisation is set on the block as well as the row because
+        FactLookupPhase reads it after the pivot, once row-level fields are gone.
+        """
+        row = block["row"]
+        prefix = row.get("prefix", "")
+        reference = row.get("reference", "")
+        curie = f"{prefix}:{reference}"
+        line_number = block["line-number"]
+
+        emitted_entities = set()
+        for organisation in self.providers:
+            entity = self.get_entity(block, organisation=organisation)
+
+            if not entity:
+                # no entity for this organisation: log an unknown-entity
+                # issue and emit no row for it
+                self._log_unknown_entity(reference, curie, line_number)
+                continue
+
+            entity = self.redirect_entity(entity)
+            if not entity or entity in emitted_entities:
+                continue
+            emitted_entities.add(entity)
+
+            new_block = dict(block)
+            new_block["row"] = dict(row)
+            new_block["row"][self.entity_field] = entity
+            new_block["row"]["organisation"] = organisation
+            new_block["organisation"] = organisation
+            if self.issues:
+                self.issues.record_entity_map(block["entry-number"], entity)
+            yield new_block
 
 
 class FactLookupPhase(LookupPhase):

From 6f8671680cdae57c38258ebaafdad26740447c59 Mon Sep 17 00:00:00 2001
From: Tom Brooks <tombrooks248@gmail.com>
Date: Wed, 15 Jul 2026 16:02:51 +0100
Subject: [PATCH 5/7] trusting the row's own organisation over the providers

---
 digital_land/phase/lookup.py    | 28 ++++++++++++++++++-------
 tests/unit/phase/test_lookup.py | 36 +++++++++++++++++++++++++++++++++
 2 files changed, 57 insertions(+), 7 deletions(-)

diff --git a/digital_land/phase/lookup.py b/digital_land/phase/lookup.py
index bed8917e2..9dfb0e3f7 100644
--- a/digital_land/phase/lookup.py
+++ b/digital_land/phase/lookup.py
@@ -300,25 +300,39 @@ def process(self, stream):
             yield from self._fan_out(block)
 
     def _fan_out(self, block):
-        """Resolve the entity once per provider, yielding a row per entity.
+        """Resolve the entity for a row that has no organisation of its own.
 
-        Distinct entities produce distinct rows, each stamped with the
-        organisation it was looked up with; if two providers resolve to the same
-        entity only one row is emitted. A provider that resolves nothing raises
-        an unknown-entity issue and produces no row, so an entry may yield fewer
-        rows than there are providers — including none.
+        The reference is looked up once per provider, and distinct entities
+        produce distinct rows, each stamped with the organisation it was looked
+        up with; if two providers resolve to the same entity only one row is
+        emitted. A provider that resolves nothing raises an unknown-entity issue
+        and produces no row, so an entry may yield fewer rows than there are
+        providers — including none.
+
+        Some collections do map a source column to the organisation (e.g.
+        conservation-area's `borough`), and that organisation need not be one of
+        the providers — so when the row knows, it is resolved against that alone.
 
         The organisation is set on the block as well as the row because
         FactLookupPhase reads it after the pivot, once row-level fields are gone.
         """
+
         row = block["row"]
         prefix = row.get("prefix", "")
         reference = row.get("reference", "")
         curie = f"{prefix}:{reference}"
         line_number = block["line-number"]
 
+        # Some collections map a source column to the organisation (e.g.
+        # conservation-area's `borough`), so a row may already say which
+        # authority it belongs to — and it need not be one of the providers.
+        # Trust the row when it knows; only try every provider when it doesn't.
+        organisations = (
+            [row["organisation"]] if row.get("organisation", "") else self.providers
+        )
+
         emitted_entities = set()
-        for organisation in self.providers:
+        for organisation in organisations:
             entity = self.get_entity(block, organisation=organisation)
 
             if not entity:
diff --git a/tests/unit/phase/test_lookup.py b/tests/unit/phase/test_lookup.py
index 2728243e0..9e10bd26f 100644
--- a/tests/unit/phase/test_lookup.py
+++ b/tests/unit/phase/test_lookup.py
@@ -434,6 +434,42 @@ def test_multi_org_neither_found_emits_nothing_and_raises_issues(self):
             "unknown entity",
         ]
 
+    def test_multi_org_row_with_own_organisation_is_not_fanned_out(self):
+        # Some collections map a source column to the organisation (e.g.
+        # conservation-area's `borough`), and it need not be one of the
+        # providers. The row knows which authority it belongs to, so it is
+        # resolved against that alone and the providers are never tried.
+        input_stream = [
+            {
+                "row": {
+                    "prefix": "local-plan",
+                    "reference": "owen",
+                    "organisation": "local-authority:XY",
+                },
+                "entry-number": 1,
+                "line-number": 2,
+            }
+        ]
+        lookups = {
+            ",local-plan,owen,local-authorityoe": "101",
+            ",local-plan,owen,local-authoritytb": "102",
+            ",local-plan,owen,local-authorityxy": "103",
+        }
+        issues = IssueLog()
+        phase = EntityLookupPhase(
+            lookups=lookups,
+            issue_log=issues,
+            providers=["local-authority:OE", "local-authority:TB"],
+        )
+        output = [block for block in phase.process(input_stream)]
+
+        # one row for the row's own org, not one per provider
+        assert len(output) == 1
+        assert output[0]["row"]["entity"] == "103"
+        assert output[0]["row"]["organisation"] == "local-authority:XY"
+        # the providers were never looked up, so nothing "missed"
+        assert issues.rows == []
+
     def test_single_org_uses_unchanged_path(self):
         # len(providers) <= 1 must not fan out; org read from the row as before.
         input_stream = [

From 382974846b17313f7af1bae7905e8a65bb23ea3e Mon Sep 17 00:00:00 2001
From: Tom Brooks <tombrooks248@gmail.com>
Date: Wed, 15 Jul 2026 16:29:09 +0100
Subject: [PATCH 6/7]  only raising an unknown entity issue when no
 organisation is found

---
 digital_land/phase/lookup.py    | 23 +++++++++++++++++------
 tests/unit/phase/test_lookup.py | 21 ++++++++++++---------
 2 files changed, 29 insertions(+), 15 deletions(-)

diff --git a/digital_land/phase/lookup.py b/digital_land/phase/lookup.py
index 9dfb0e3f7..19fa6e2b4 100644
--- a/digital_land/phase/lookup.py
+++ b/digital_land/phase/lookup.py
@@ -305,14 +305,22 @@ def _fan_out(self, block):
         The reference is looked up once per provider, and distinct entities
         produce distinct rows, each stamped with the organisation it was looked
         up with; if two providers resolve to the same entity only one row is
-        emitted. A provider that resolves nothing raises an unknown-entity issue
-        and produces no row, so an entry may yield fewer rows than there are
-        providers — including none.
+        emitted. A provider that resolves nothing simply produces no row, so an
+        entry may yield fewer rows than there are providers — including none.
 
         Some collections do map a source column to the organisation (e.g.
         conservation-area's `borough`), and that organisation need not be one of
         the providers — so when the row knows, it is resolved against that alone.
 
+        An unknown-entity issue is raised only when NO organisation resolves an
+        entity. A resource shared by several organisations often holds one
+        organisation's rows alongside another's, so a reference missing for the
+        organisation that does not own it is normal, not a fault. Issues also
+        carry no organisation of their own and are fanned out to every
+        organisation on the resource when they become tasks, so a per-organisation
+        miss would tell both authorities their records were missing while the
+        entity sat correctly assigned.
+
         The organisation is set on the block as well as the row because
         FactLookupPhase reads it after the pivot, once row-level fields are gone.
         """
@@ -336,9 +344,7 @@ def _fan_out(self, block):
             entity = self.get_entity(block, organisation=organisation)
 
             if not entity:
-                # no entity for this organisation: log an unknown-entity
-                # issue and emit no row for it
-                self._log_unknown_entity(reference, curie, line_number)
+                # this organisation does not own the reference: no row for it
                 continue
 
             entity = self.redirect_entity(entity)
@@ -355,6 +361,11 @@ def _fan_out(self, block):
                 self.issues.record_entity_map(block["entry-number"], entity)
             yield new_block
 
+        # no organisation resolved an entity: raise a single unknown-entity
+        # issue, as the inherited single-organisation path does
+        if not emitted_entities:
+            self._log_unknown_entity(reference, curie, line_number)
+
 
 class FactLookupPhase(LookupPhase):
     def __init__(
diff --git a/tests/unit/phase/test_lookup.py b/tests/unit/phase/test_lookup.py
index 9e10bd26f..6d766a51f 100644
--- a/tests/unit/phase/test_lookup.py
+++ b/tests/unit/phase/test_lookup.py
@@ -379,8 +379,13 @@ def test_multi_org_same_entity_produces_single_row(self):
         # first org to resolve the entity wins
         assert output[0]["row"]["organisation"] == "local-authority:OE"
 
-    def test_multi_org_one_missing_raises_issue_and_emits_found(self):
-        # OE resolves, TB has no lookup -> one row for OE + one unknown-entity issue.
+    def test_multi_org_one_missing_emits_found_without_issue(self):
+        # OE resolves, TB has no lookup -> one row for OE and NO issue. A shared
+        # resource routinely holds one organisation's rows alongside another's,
+        # so a reference missing for the organisation that does not own it is
+        # normal. Issues carry no organisation and are fanned out to every
+        # organisation on the resource as tasks, so raising one here would tell
+        # both authorities their records were missing.
         input_stream = [
             {
                 "row": {
@@ -404,9 +409,9 @@ def test_multi_org_one_missing_raises_issue_and_emits_found(self):
         assert len(output) == 1
         assert output[0]["row"]["entity"] == "101"
         assert output[0]["row"]["organisation"] == "local-authority:OE"
-        assert [i["issue-type"] for i in issues.rows] == ["unknown entity"]
+        assert issues.rows == []
 
-    def test_multi_org_neither_found_emits_nothing_and_raises_issues(self):
+    def test_multi_org_neither_found_emits_nothing_and_raises_one_issue(self):
         input_stream = [
             {
                 "row": {
@@ -428,11 +433,9 @@ def test_multi_org_neither_found_emits_nothing_and_raises_issues(self):
         output = [block for block in phase.process(input_stream)]
 
         assert output == []
-        # one unknown-entity issue per organisation that missed
-        assert [i["issue-type"] for i in issues.rows] == [
-            "unknown entity",
-            "unknown entity",
-        ]
+        # nothing resolved at all, so one issue for the entry -- not one per
+        # organisation, which would fan out into a task for each of them
+        assert [i["issue-type"] for i in issues.rows] == ["unknown entity"]
 
     def test_multi_org_row_with_own_organisation_is_not_fanned_out(self):
         # Some collections map a source column to the organisation (e.g.

From 1620777132871a6344fec2fc2684aacffd34c465 Mon Sep 17 00:00:00 2001
From: Tom Brooks <tombrooks248@gmail.com>
Date: Thu, 16 Jul 2026 11:05:43 +0100
Subject: [PATCH 7/7] changing fan_out func name

---
 digital_land/phase/lookup.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/digital_land/phase/lookup.py b/digital_land/phase/lookup.py
index 19fa6e2b4..6f4161b09 100644
--- a/digital_land/phase/lookup.py
+++ b/digital_land/phase/lookup.py
@@ -297,9 +297,9 @@ def process(self, stream):
                 yield block
                 continue
 
-            yield from self._fan_out(block)
+            yield from self._get_provider_entity_numbers(block)
 
-    def _fan_out(self, block):
+    def _get_provider_entity_numbers(self, block):
         """Resolve the entity for a row that has no organisation of its own.
 
         The reference is looked up once per provider, and distinct entities