diff --git a/services/compliance/app/fsma_spreadsheet.py b/services/compliance/app/fsma_spreadsheet.py
index c4e95240e..8611bc0b1 100644
--- a/services/compliance/app/fsma_spreadsheet.py
+++ b/services/compliance/app/fsma_spreadsheet.py
@@ -20,6 +20,8 @@
 from datetime import datetime, timezone
 from typing import Any
 
+from shared.fsma_validation import gln_for_export
+
 from .csv_safety import sanitize_cell, sanitize_sequence
 
 # Mandatory FDA 204 spreadsheet columns in required sort order
@@ -269,10 +271,12 @@ def _normalise_event(
         "product_description": _pick(event.get("product_description"), kdes.get("product_description")),
         "quantity": _pick(event.get("quantity"), kdes.get("quantity")),
         "unit_of_measure": _pick(event.get("uom"), event.get("unit_of_measure"), kdes.get("unit_of_measure")),
-        "origin_gln": _pick(
-            event.get("facility_gln"),
-            event.get("location_gln"),
-            kdes.get("ship_from_gln"),
+        "origin_gln": gln_for_export(
+            _pick(
+                event.get("facility_gln"),
+                event.get("location_gln"),
+                kdes.get("ship_from_gln"),
+            )
         ),
         "origin_name": _pick(
             event.get("facility_name"),
@@ -280,10 +284,10 @@ def _normalise_event(
             kdes.get("ship_from_location"),
         ),
         "origin_address": _pick(event.get("facility_address"), event.get("ship_from_address")),
-        "destination_gln": _pick(event.get("dest_gln"), kdes.get("ship_to_gln")),
+        "destination_gln": gln_for_export(_pick(event.get("dest_gln"), kdes.get("ship_to_gln"))),
         "destination_name": _pick(event.get("dest_name"), kdes.get("ship_to_location")),
         "destination_address": _pick(event.get("dest_address"), event.get("ship_to_address")),
-        "tlc_source_gln": _pick(kdes.get("tlc_source_gln")),
+        "tlc_source_gln": gln_for_export(_pick(kdes.get("tlc_source_gln"))),
         "tlc_source_fda_reg": _pick(kdes.get("tlc_source_fda_reg")),
         "reference_document_type": _pick(kdes.get("reference_document_type"), event.get("document_type")),
         "reference_document_number": _pick(kdes.get("reference_document_number"), event.get("ref_doc")),
diff --git a/services/compliance/tests/test_hermes_export_gln.py b/services/compliance/tests/test_hermes_export_gln.py
new file mode 100644
index 000000000..4581de45b
--- /dev/null
+++ b/services/compliance/tests/test_hermes_export_gln.py
@@ -0,0 +1,46 @@
+"""Data-free GLN regressions for FDA spreadsheet rows."""
+
+import csv
+import io
+
+import pytest
+
+from services.compliance.app import fsma_spreadsheet
+
+
+@pytest.mark.parametrize("raw", ["x", "1", "bad"])
+@pytest.mark.parametrize("origin", ["facility_gln", "location_gln", "ship_from_gln"])
+@pytest.mark.parametrize("destination", ["dest_gln", "ship_to_gln"])
+def test_spreadsheet_omits_invalid_glns(raw, origin, destination):
+    event = {"kdes": {"tlc_source_gln": raw}}
+    for key in (origin, destination):
+        target = event["kdes"] if key.startswith("ship_") else event
+        target[key] = raw
+
+    row = fsma_spreadsheet._normalise_event(event)
+
+    for column in ("origin_gln", "destination_gln", "tlc_source_gln"):
+        assert row[column] == ""
+    assert row["additional_kdes_json"] == ""
+
+
+@pytest.mark.parametrize("raw", ["x", "1", "bad"])
+def test_spreadsheet_csv_omits_invalid_glns(raw):
+    event = {"facility_gln": raw, "dest_gln": raw, "kdes": {"tlc_source_gln": raw}}
+
+    rows = list(csv.reader(io.StringIO(fsma_spreadsheet.generate_fda_csv([event]))))
+    header_index = next(index for index, row in enumerate(rows) if row == fsma_spreadsheet.FDA_HEADERS)
+    row = dict(zip(fsma_spreadsheet.FDA_COLUMNS, rows[header_index + 1]))
+
+    for column in ("origin_gln", "destination_gln", "tlc_source_gln"):
+        assert row[column] == ""
+
+
+def test_spreadsheet_preserves_filtered_glns(monkeypatch):
+    marker = "GLN_MARKER"
+    monkeypatch.setattr(fsma_spreadsheet, "gln_for_export", lambda value: marker)
+
+    row = fsma_spreadsheet._normalise_event({})
+
+    for column in ("origin_gln", "destination_gln", "tlc_source_gln"):
+        assert row[column] == marker
diff --git a/services/graph/app/exporters/recall_contacts_csv.py b/services/graph/app/exporters/recall_contacts_csv.py
index c37b739ed..45bb1c988 100644
--- a/services/graph/app/exporters/recall_contacts_csv.py
+++ b/services/graph/app/exporters/recall_contacts_csv.py
@@ -30,6 +30,8 @@
 
 from typing import Any, Dict, List, Optional
 
+from shared.fsma_validation import gln_for_export
+
 from .safe_text import csv_safe
 
 __all__ = ["RECALL_CONTACTS_HEADER", "RECEIVING_EVENT_TYPES", "build_recall_contacts_rows"]
@@ -68,7 +70,7 @@ def _row(
     event = event or {}
     return [
         csv_safe(facility.get("name") or ""),
-        csv_safe(facility.get("gln") or ""),
+        csv_safe(gln_for_export(facility.get("gln"))),
         csv_safe(facility.get("address") or ""),
         csv_safe(facility.get("facility_type") or ""),
         csv_safe(lot.get("tlc") or ""),
diff --git a/services/graph/app/exporters/trace_csv.py b/services/graph/app/exporters/trace_csv.py
index 10703706c..9b7107e11 100644
--- a/services/graph/app/exporters/trace_csv.py
+++ b/services/graph/app/exporters/trace_csv.py
@@ -32,6 +32,8 @@
 
 from typing import Any, Dict, List, Optional, Sequence
 
+from shared.fsma_validation import gln_for_export
+
 from .safe_text import csv_safe
 
 __all__ = ["TRACE_CSV_HEADER", "build_trace_csv_rows"]
@@ -70,7 +72,7 @@ def _row(
         csv_safe(event.get("event_date", "")),
         csv_safe(event.get("event_time", "")),
         csv_safe(facility.get("name", "")),
-        csv_safe(facility.get("gln", "")),
+        csv_safe(gln_for_export(facility.get("gln", ""))),
         csv_safe(facility.get("address", "")),
         csv_safe(event.get("confidence", "")),
     ]
diff --git a/services/graph/app/fsma_recall/models.py b/services/graph/app/fsma_recall/models.py
index ee43c8fcd..ec87f35e8 100644
--- a/services/graph/app/fsma_recall/models.py
+++ b/services/graph/app/fsma_recall/models.py
@@ -10,6 +10,8 @@
 from enum import Enum
 from typing import Any, Dict, List, Optional
 
+from shared.fsma_validation import gln_for_export
+
 
 class RecallStatus(str, Enum):
     """Status of a recall drill."""
@@ -147,7 +149,7 @@ def to_dict(self) -> Dict[str, Any]:
             "severity": self.severity.value,
             "target_lot": self.target_lot,
             "target_gtin": self.target_gtin,
-            "target_facility_gln": self.target_facility_gln,
+            "target_facility_gln": gln_for_export(self.target_facility_gln),
             "initiated_by": self.initiated_by,
             "reason": self.reason,
             "description": self.description,
@@ -235,7 +237,7 @@ def to_dict(self) -> Dict[str, Any]:
             "data_completeness_pct": self.data_completeness_pct,
             "affected_facilities": [
                 {
-                    "gln": f.gln,
+                    "gln": gln_for_export(f.gln),
                     "name": f.name,
                     "facility_type": f.facility_type,
                     "lots_affected": f.lots_affected,
diff --git a/services/graph/app/routers/fsma/compliance.py b/services/graph/app/routers/fsma/compliance.py
index 511a9e550..8c204d57a 100644
--- a/services/graph/app/routers/fsma/compliance.py
+++ b/services/graph/app/routers/fsma/compliance.py
@@ -19,8 +19,6 @@
 from typing import List, Optional
 from urllib.parse import quote
 
-from shared.fsma_validation import normalize_gtin, verify_gs1_check_digit
-
 import structlog
 from fastapi import APIRouter, Depends, HTTPException, Path as PathParam, Query, Request
 from fastapi.responses import StreamingResponse
@@ -36,6 +34,7 @@
     export_plan_json,
     export_plan_markdown,
 )
+from shared.fsma_validation import gln_for_export, normalize_gtin, verify_gs1_check_digit
 
 # Add shared utilities (portable path resolution)
 from shared.middleware import get_current_tenant_id
@@ -217,7 +216,7 @@ async def export_fda_request_sheet(
                     csv_safe(event.get("quantity", "")),
                     csv_safe(event.get("unit_of_measure", "")),
                     csv_safe(event.get("location_description", "")),
-                    csv_safe(event.get("location_gln", "")),
+                    csv_safe(gln_for_export(event.get("location_gln", ""))),
                     csv_safe(event.get("event_date", "")),
                     csv_safe(event.get("event_time", "")),
                     csv_safe(event.get("reference_doc_type", "")),
@@ -404,7 +403,11 @@ def _skip(reason: str) -> None:
                     _skip("transformation_incomplete")
                     continue
                 if not valid_transformation_id and transformation_id:
-                    transformation_id = f"urn:uuid:{transformation_id}" if not str(transformation_id).startswith("urn:") else transformation_id
+                    transformation_id = (
+                        f"urn:uuid:{transformation_id}"
+                        if not str(transformation_id).startswith("urn:")
+                        else transformation_id
+                    )
                     valid_transformation_id = True
                 epcis_event = {"type": "TransformationEvent", **common, "bizStep": BIZ_STEP_MAP[event_type]}
                 if input_epc:
@@ -429,7 +432,9 @@ def _skip(reason: str) -> None:
                 epcis_event = {
                     "type": "ObjectEvent",
                     **common,
-                    "action": "ADD" if event_type in ("CREATION", "RECEIVING", "HARVESTING", "INITIAL_PACKING") else "OBSERVE",
+                    "action": "ADD"
+                    if event_type in ("CREATION", "RECEIVING", "HARVESTING", "INITIAL_PACKING")
+                    else "OBSERVE",
                     "bizStep": BIZ_STEP_MAP[event_type],
                     "disposition": "in_transit" if event_type == "SHIPPING" else "in_progress",
                 }
@@ -592,7 +597,11 @@ def add_location_lists(parent, event):
                     sub_element(container, member, {"type": kind_uri}).text = str(item[member])
 
     def add_common_tail(parent, event, *, include_action: bool = True):
-        fields = ("transformationID", "action", "bizStep", "disposition") if include_action else ("transformationID", "bizStep", "disposition")
+        fields = (
+            ("transformationID", "action", "bizStep", "disposition")
+            if include_action
+            else ("transformationID", "bizStep", "disposition")
+        )
         for field in fields:
             if field == "action" and event.get("type") != "ObjectEvent" and include_action:
                 # TransformationEvent has no action; still allow explicit passthrough when present
diff --git a/services/graph/app/routers/fsma/traceability.py b/services/graph/app/routers/fsma/traceability.py
index 65dcb76b8..5ca7aefb1 100644
--- a/services/graph/app/routers/fsma/traceability.py
+++ b/services/graph/app/routers/fsma/traceability.py
@@ -12,6 +12,7 @@
 from pydantic import BaseModel
 from shared.auth import require_api_key
 from shared.fsma_validation import (
+    gln_for_export,
     validate_gln,
     validate_gtin,
     validate_tlc,
@@ -185,7 +186,7 @@ async def trace_forward_endpoint(
         return {
             "lot_id": result.lot_id,
             "direction": result.direction,
-            "facilities": result.facilities,
+            "facilities": [{**facility, "gln": gln_for_export(facility.get("gln"))} for facility in result.facilities],
             "events": result.events,
             "downstream_lots": result.lots,
             "total_quantity": result.total_quantity,
@@ -314,7 +315,7 @@ async def trace_backward_endpoint(
         return {
             "lot_id": result.lot_id,
             "direction": result.direction,
-            "facilities": result.facilities,
+            "facilities": [{**facility, "gln": gln_for_export(facility.get("gln"))} for facility in result.facilities],
             "events": result.events,
             "source_lots": result.lots,
             "total_quantity": result.total_quantity,
diff --git a/services/graph/tests/test_hermes_export_gln.py b/services/graph/tests/test_hermes_export_gln.py
new file mode 100644
index 000000000..17f38efda
--- /dev/null
+++ b/services/graph/tests/test_hermes_export_gln.py
@@ -0,0 +1,93 @@
+"""Data-free GLN regressions for graph CSV exports."""
+
+import csv
+import io
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from shared.fsma_validation import calculate_gs1_check_digit, gln_for_export
+
+from services.graph.app.exporters import recall_contacts_csv, trace_csv
+from services.graph.app.routers.fsma import compliance
+
+
+def shapes():
+    base = "0" * 12
+    return [None, "", " ", "x", "0", "x" * 13, base + str(calculate_gs1_check_digit(base))]
+
+
+@pytest.mark.parametrize("raw", shapes())
+def test_trace_facility_gln(raw):
+    assert trace_csv._row({}, {}, {"gln": raw})[8] == gln_for_export(raw)
+
+
+@pytest.mark.parametrize("raw", shapes())
+def test_recall_facility_gln(raw):
+    assert recall_contacts_csv._row({"gln": raw}, {}, {})[1] == gln_for_export(raw)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("raw", shapes())
+async def test_fda_sortable_gln(monkeypatch, raw):
+    monkeypatch.setattr(compliance, "reconcile_tenant_scope", lambda *a, **k: "x")
+    monkeypatch.setattr(compliance.uuid, "UUID", lambda value: value)
+    client = MagicMock()
+    client.__aenter__ = AsyncMock(return_value=client)
+    client.__aexit__ = AsyncMock(return_value=False)
+    monkeypatch.setattr(compliance, "Neo4jClient", MagicMock(return_value=client))
+    monkeypatch.setattr(compliance, "_load_range_export_events", AsyncMock(return_value=[{"location_gln": raw}]))
+    request = SimpleNamespace(url=SimpleNamespace(path="x"))
+    response = await compliance.export_fda_request_sheet.__wrapped__(
+        request, start_date="", end_date="", tlc=None, tenant_id=None, api_key=None
+    )
+    body = b"".join([chunk async for chunk in response.body_iterator]).decode("utf-8-sig")
+    assert list(csv.reader(io.StringIO(body)))[1][6] == gln_for_export(raw)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("raw", shapes())
+@pytest.mark.parametrize("direction", ["forward", "backward"])
+async def test_trace_api_gln(monkeypatch, raw, direction):
+    from services.graph.app.fsma.types import TraceResult
+    from services.graph.app.routers.fsma import traceability
+
+    result = TraceResult("x", direction, [{"gln": raw}], [], [], None, 0, 0)
+    client = MagicMock()
+    client.__aenter__ = AsyncMock(return_value=client)
+    client.__aexit__ = AsyncMock(return_value=False)
+    monkeypatch.setattr(traceability, "Neo4jClient", MagicMock(return_value=client))
+    monkeypatch.setattr(traceability, f"trace_{direction}", AsyncMock(return_value=result))
+    monkeypatch.setattr(traceability, "record_trace_query", MagicMock())
+    monkeypatch.setattr(traceability, "get_audit_log", lambda: SimpleNamespace(alog=AsyncMock()))
+    endpoint = getattr(traceability, f"trace_{direction}_endpoint").__wrapped__
+    response = await endpoint(SimpleNamespace(), tlc="x", max_depth=1, tenant_id=None, api_key=None)
+    assert response["facilities"][0]["gln"] == gln_for_export(raw)
+    assert result.facilities[0]["gln"] == raw
+
+
+@pytest.mark.parametrize("raw", shapes())
+def test_recall_result_gln(raw):
+    from services.graph.app.fsma_recall.models import AffectedFacility, RecallResult
+
+    facility = AffectedFacility(raw, "", "", [], None)
+    result = RecallResult("x", True, affected_facilities=[facility])
+    assert result.to_dict()["affected_facilities"][0]["gln"] == gln_for_export(raw)
+    assert facility.gln == raw
+
+
+@pytest.mark.parametrize("raw", shapes())
+def test_recall_target_gln(raw):
+    from datetime import datetime, timezone
+
+    from services.graph.app.fsma_recall.models import RecallDrill, RecallSeverity, RecallType
+
+    drill = RecallDrill(
+        "x",
+        "x",
+        datetime.now(timezone.utc),
+        next(iter(RecallType)),
+        next(iter(RecallSeverity)),
+        target_facility_gln=raw,
+    )
+    assert drill.to_dict()["target_facility_gln"] == gln_for_export(raw)
diff --git a/services/ingestion/app/csv_templates.py b/services/ingestion/app/csv_templates.py
index e5929a310..2b07321c4 100644
--- a/services/ingestion/app/csv_templates.py
+++ b/services/ingestion/app/csv_templates.py
@@ -343,6 +343,8 @@ async def list_templates():
 _PRODUCT_COLUMNS = ("product_description", "product", "description")
 _QUANTITY_COLUMNS = ("quantity", "qty")
 _UNIT_COLUMNS = ("unit_of_measure", "uom")
+_LOCATION_PHONE_COLUMNS = ("location_phone", "ship_from_phone", "receiving_phone", "phone")
+_LOCATION_ADDRESS_COLUMNS = ("location_address", "ship_from_address", "receiving_address", "address")
 _LOCATION_GLN_COLUMNS = ("location_gln", "ship_from_gln", "receiving_gln", "gln")
 _LOCATION_NAME_COLUMNS = (
     "location_name",
@@ -361,6 +363,8 @@ async def list_templates():
     _QUANTITY_COLUMNS,
     _UNIT_COLUMNS,
     _LOCATION_GLN_COLUMNS,
+    _LOCATION_PHONE_COLUMNS,
+    _LOCATION_ADDRESS_COLUMNS,
     _LOCATION_NAME_COLUMNS,
     _SHIP_TO_COLUMNS,
 )
@@ -1341,6 +1345,8 @@ def _reject(reason: str, message: str) -> None:
 
             # Location: strip whitespace and expand abbreviations
             loc_gln = _first_present(row, _LOCATION_GLN_COLUMNS) or None
+            loc_phone = _first_present(row, _LOCATION_PHONE_COLUMNS) or None
+            loc_address = _first_present(row, _LOCATION_ADDRESS_COLUMNS) or None
             _loc_raw = _first_present(row, _LOCATION_NAME_COLUMNS)
             loc_name = _normalize_location(_loc_raw) if _loc_raw else None
             if _loc_raw and loc_name != _loc_raw:
@@ -1411,6 +1417,8 @@ def _reject(reason: str, message: str) -> None:
                 "unit_of_measure",
                 "location_gln",
                 "location_name",
+                *_LOCATION_PHONE_COLUMNS,
+                *_LOCATION_ADDRESS_COLUMNS,
                 "cte_type",
                 "event_type",
                 "type",
@@ -1457,6 +1465,8 @@ def _reject(reason: str, message: str) -> None:
                 unit_of_measure=uom,
                 location_gln=loc_gln,
                 location_name=loc_name,
+                location_phone=loc_phone,
+                location_address=loc_address,
                 timestamp=ts,
                 kdes=kdes,
             )
diff --git a/services/ingestion/app/edi_ingestion/routes.py b/services/ingestion/app/edi_ingestion/routes.py
index bb2c94ad2..121e276e3 100644
--- a/services/ingestion/app/edi_ingestion/routes.py
+++ b/services/ingestion/app/edi_ingestion/routes.py
@@ -30,7 +30,8 @@
 from typing import Any, Optional
 
 from fastapi import APIRouter, Depends, File, Form, Header, HTTPException, Query, UploadFile
-from pydantic import ValidationError
+from pydantic import PrivateAttr, ValidationError
+from shared.fsma_validation import is_location_description_complete
 
 from ..authz import IngestionPrincipal, require_permission
 from ..format_extractors import is_edi_content
@@ -344,6 +345,15 @@ def _preflight_event_date(
         raise HTTPException(status_code=400, detail=detail) from exc
 
 
+class _EDIIngestEvent(IngestEvent):
+    """Internal EDI model; the private permission is never accepted from payloads.
+
+    Inherited field validators still enforce every non-location constraint.
+    """
+
+    _edi_allow_incomplete_location: bool = PrivateAttr(default=True)
+
+
 router = APIRouter(prefix="/api/v1/ingest/edi", tags=["EDI Import"])
 
 _resolve_tenant_id = resolve_tenant_id
@@ -673,30 +683,36 @@ async def ingest_edi_856(
         extracted.get("ship_time_raw"),
     )
 
-    ship_from_name = ship_from_location_override or extracted.get("ship_from_name") or "Unknown ship-from"
-    ship_to_name = ship_to_location_override or extracted.get("ship_to_name") or "Unknown ship-to"
+    ship_from_name = (
+        ship_from_location_override if ship_from_location_override is not None else extracted.get("ship_from_name")
+    )
+    ship_to_name = ship_to_location_override if ship_to_location_override is not None else extracted.get("ship_to_name")
 
     ship_from_gln_raw = extracted.get("ship_from_gln")
     ship_to_gln_raw = extracted.get("ship_to_gln")
     ship_from_gln = _valid_gln_or_none(ship_from_gln_raw)
-    ship_to_gln = _valid_gln_or_none(ship_to_gln_raw)
 
     if not lines:
         lines = [{}]
     events: list[IngestEvent] = []
     for line_index, line in enumerate(lines, start=1):
         quantity = quantity_override if quantity_override is not None else line.get("quantity")
-        resolved_quantity = _safe_float(str(quantity) if quantity is not None else None, fallback=1.0)
+        resolved_quantity = _safe_float(
+            str(quantity) if quantity is not None else None, fallback=0.0 if quantity is not None else 1.0
+        )
         resolved_unit = _normalize_unit(unit_of_measure_override or line.get("unit_of_measure"))
         product_name = product_description or line.get("product_description") or "EDI 856 Shipment"
         lot = line.get("lot_number")
         line_tlc = lot.strip() if isinstance(lot, str) and lot.strip() else traceability_lot_code
         kdes: dict[str, Any] = {
+            "location_incomplete": not is_location_description_complete(name=ship_from_name, gln=ship_from_gln_raw),
             "ship_date": ship_date,
+            "ship_from_name": extracted.get("ship_from_name"),
+            "ship_to_name": extracted.get("ship_to_name"),
             "ship_from_location": ship_from_name,
             "ship_to_location": ship_to_name,
-            "ship_from_gln": ship_from_gln or ship_from_gln_raw,
-            "ship_to_gln": ship_to_gln or ship_to_gln_raw,
+            "ship_from_gln": ship_from_gln_raw,
+            "ship_to_gln": ship_to_gln_raw,
             "reference_document_number": extracted.get("reference_document_number"),
             "carrier": extracted.get("carrier"),
             "asn_number": extracted.get("asn_number"),
@@ -716,16 +732,21 @@ async def ingest_edi_856(
             "immediate_previous_source": ship_from_name,
         }
         events.append(
-            IngestEvent(
+            _EDIIngestEvent(
                 cte_type=WebhookCTEType.SHIPPING,
                 traceability_lot_code=line_tlc,
                 product_description=product_name,
                 quantity=resolved_quantity,
                 unit_of_measure=resolved_unit,
+                # A-2170: accept and flag incomplete EDI locations; preserve raw KDEs.
                 location_gln=ship_from_gln,
                 location_name=ship_from_name,
                 timestamp=timestamp,
-                kdes={k: v for k, v in kdes.items() if v is not None and v != ""},
+                kdes={
+                    k: v
+                    for k, v in kdes.items()
+                    if k.endswith(("_gln", "_name", "_location")) or (v is not None and v != "")
+                },
             )
         )
 
@@ -814,7 +835,9 @@ def _build_ingest_event_for_set(
             },
         )
 
-    resolved_quantity = _safe_float(str(quantity) if quantity is not None else None, fallback=1.0)
+    resolved_quantity = _safe_float(
+        str(quantity) if quantity is not None else None, fallback=0.0 if quantity is not None else 1.0
+    )
 
     # #2978: an X12 unit that is PRESENT but unrecognized must not be
     # coerced to "units" — that persists a numerically-correct,
@@ -848,16 +871,29 @@ def _build_ingest_event_for_set(
 
     # Resolve location (varies by document type)
     location_name = location_override
-    location_gln = None
+    location_gln_raw = None
     if transaction_set in ("856", "850", "945"):
-        location_name = location_name or extracted.get("ship_from_name") or extracted.get("seller_name") or "Unknown"
-        location_gln = _valid_gln_or_none(extracted.get("ship_from_gln") or extracted.get("seller_gln"))
+        location_name = next(
+            (
+                value
+                for value in (location_name, extracted.get("ship_from_name"), extracted.get("seller_name"))
+                if value is not None
+            ),
+            None,
+        )
+        location_gln_raw = (
+            extracted.get("ship_from_gln")
+            if extracted.get("ship_from_gln") is not None
+            else extracted.get("seller_gln")
+        )
     elif transaction_set == "810":
-        location_name = location_name or extracted.get("seller_name") or "Unknown"
-        location_gln = _valid_gln_or_none(extracted.get("seller_gln"))
+        location_name = location_name if location_name is not None else extracted.get("seller_name")
+        location_gln_raw = extracted.get("seller_gln")
     elif transaction_set in ("861", "944"):
-        location_name = location_name or extracted.get("receiving_location_name") or "Unknown"
-        location_gln = _valid_gln_or_none(extracted.get("receiving_location_gln"))
+        location_name = location_name if location_name is not None else extracted.get("receiving_location_name")
+        location_gln_raw = extracted.get("receiving_location_gln")
+
+    location_gln = _valid_gln_or_none(location_gln_raw)
 
     # Resolve timestamp
     date_field, time_field = _EVENT_DATE_FIELDS.get(transaction_set, (None, None))
@@ -1007,16 +1043,23 @@ def _build_ingest_event_for_set(
         if extracted.get("depositor_gln"):
             kdes["depositor_gln"] = extracted["depositor_gln"]
 
-    return IngestEvent(
+    kdes["location_incomplete"] = not is_location_description_complete(name=location_name, gln=location_gln_raw)
+    # Retain even empty raw location values for audit evidence.
+    kdes.update({key: value for key, value in extracted.items() if key.endswith(("_gln", "_name"))})
+
+    return _EDIIngestEvent(
         cte_type=cte_type,
         traceability_lot_code=traceability_lot_code,
         product_description=product_name,
         quantity=resolved_quantity,
         unit_of_measure=resolved_unit,
+        # A-2170: accept and flag incomplete EDI locations; preserve raw KDEs.
         location_gln=location_gln,
         location_name=location_name,
         timestamp=timestamp,
-        kdes={k: v for k, v in kdes.items() if v is not None and v != ""},
+        kdes={
+            k: v for k, v in kdes.items() if k.endswith(("_gln", "_name", "_location")) or (v is not None and v != "")
+        },
     )
 
 
diff --git a/services/ingestion/app/epcis_export.py b/services/ingestion/app/epcis_export.py
index d5a1f0859..3ae000cde 100644
--- a/services/ingestion/app/epcis_export.py
+++ b/services/ingestion/app/epcis_export.py
@@ -19,7 +19,7 @@
 from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import BaseModel, Field
 from shared.database import get_db_safe
-from shared.fsma_validation import normalize_gtin, verify_gs1_check_digit
+from shared.fsma_validation import gln_for_export as _gln_csv_cell, normalize_gtin, verify_gs1_check_digit
 from shared.tenant_context import TenantContext
 from sqlalchemy import text
 
@@ -525,10 +525,7 @@ def _skip(row: dict, reason: str) -> None:
                 _skip(r, "missing_timestamp")
                 continue
             event_kind = r.get("epcis_event_type")
-            is_transformation = (
-                r.get("event_type") == "transformation"
-                or event_kind == "TransformationEvent"
-            )
+            is_transformation = r.get("event_type") == "transformation" or event_kind == "TransformationEvent"
             if event_kind not in (None, "ObjectEvent", "TransformationEvent"):
                 _skip(r, "unsupported_epcis_event_type")
                 continue
@@ -725,9 +722,9 @@ async def export_fda(
                         date_str,
                         time_str,
                         sanitize_cell(str(r["ship_from"] or r["location_name"] or "")).replace(",", ";"),
-                        str(r["ship_from_gln"] or r["location_gln"] or ""),
+                        (_gln_csv_cell(r["ship_from_gln"]) or _gln_csv_cell(r["location_gln"])),
                         sanitize_cell(str(r["ship_to"] or "")).replace(",", ";"),
-                        str(r["ship_to_gln"] or ""),
+                        _gln_csv_cell(r["ship_to_gln"]),
                         sanitize_cell(str(r["carrier"] or "")).replace(",", ";"),
                         str(r["temperature_c"] or ""),
                         str(r["sha256_hash"] or ""),
diff --git a/services/ingestion/app/exchange_api.py b/services/ingestion/app/exchange_api.py
index 1ef961c01..8c1c04c02 100644
--- a/services/ingestion/app/exchange_api.py
+++ b/services/ingestion/app/exchange_api.py
@@ -21,6 +21,7 @@
 from fastapi import APIRouter, Depends, Header, HTTPException, Query
 from pydantic import BaseModel, Field
 from shared.database import get_db_safe
+from shared.fsma_validation import gln_for_export
 from sqlalchemy import text
 
 from .authz import IngestionPrincipal, require_permission
@@ -172,15 +173,15 @@ def _build_records(rows: list[Any]) -> list[dict[str, Any]]:
             "quantity": float(row.quantity) if row.quantity is not None else None,
             "unit_of_measure": row.unit_of_measure,
             "ship_from": {
-                "gln": ship_from_gln,
+                "gln": gln_for_export(ship_from_gln),
                 "name": ship_from_name,
             },
             "ship_to": {
-                "gln": kdes.get("ship_to_gln"),
+                "gln": gln_for_export(kdes.get("ship_to_gln")),
                 "name": ship_to_name,
             },
             "tlc_source": {
-                "gln": kdes.get("tlc_source_gln"),
+                "gln": gln_for_export(kdes.get("tlc_source_gln")),
                 "fda_registration": kdes.get("tlc_source_fda_reg"),
                 "immediate_previous_source": kdes.get("immediate_previous_source"),
             },
@@ -190,7 +191,9 @@ def _build_records(rows: list[Any]) -> list[dict[str, Any]]:
                 "record_hash": row.sha256_hash,
                 "chain_hash": row.chain_hash,
             },
-            "additional_kdes": {k: v for k, v in kdes.items() if k not in known_kdes},
+            "additional_kdes": {
+                k: gln_for_export(v) if k.endswith("_gln") else v for k, v in kdes.items() if k not in known_kdes
+            },
             "source": row.source,
         }
         records.append(record)
diff --git a/services/ingestion/app/fda_export_service.py b/services/ingestion/app/fda_export_service.py
index 2bda7ac34..46a980b6b 100644
--- a/services/ingestion/app/fda_export_service.py
+++ b/services/ingestion/app/fda_export_service.py
@@ -18,6 +18,7 @@
 from uuid import uuid4
 
 from shared.fda_export import safe_filename_token as _shared_safe_filename_token
+from shared.fsma_validation import gln_for_export
 from shared.merkle_tree import MerkleTree
 from shared.rules.verdict import classify_event_verdict
 
@@ -213,19 +214,19 @@ def _event_to_fda_row(event: dict, *, include_pii: bool = False) -> dict:
         "Event Type (CTE)": sanitize_cell(event.get("event_type", "")),
         "Event Date": sanitize_cell(event_date),
         "Event Time": sanitize_cell(event_time),
-        "Location GLN": sanitize_cell(event.get("location_gln", "") or ""),
+        "Location GLN": gln_for_export(event.get("location_gln")),
         "Location Name": _redact_location_value(
             sanitize_cell(event.get("location_name", "") or ""),
             "Location Name",
             include_pii,
         ),
-        "Ship From GLN": sanitize_cell(kdes.get("ship_from_gln", "") or ""),
+        "Ship From GLN": gln_for_export(kdes.get("ship_from_gln")),
         "Ship From Name": _redact_location_value(
             sanitize_cell(kdes.get("ship_from_location", "") or ""),
             "Ship From Name",
             include_pii,
         ),
-        "Ship To GLN": sanitize_cell(kdes.get("ship_to_gln", "") or ""),
+        "Ship To GLN": gln_for_export(kdes.get("ship_to_gln")),
         "Ship To Name": _redact_location_value(
             sanitize_cell(kdes.get("ship_to_location", kdes.get("receiving_location", "")) or ""),
             "Ship To Name",
@@ -236,7 +237,7 @@ def _event_to_fda_row(event: dict, *, include_pii: bool = False) -> dict:
             "Immediate Previous Source",
             include_pii,
         ),
-        "TLC Source GLN": sanitize_cell(kdes.get("tlc_source_gln", "") or ""),
+        "TLC Source GLN": gln_for_export(kdes.get("tlc_source_gln")),
         "TLC Source FDA Registration": sanitize_cell(kdes.get("tlc_source_fda_reg", "") or ""),
         "Source Document": sanitize_cell(event.get("source", "")),
         "Record Hash (SHA-256)": sanitize_cell(event.get("sha256_hash", "")),
diff --git a/services/ingestion/app/sensitech_parser.py b/services/ingestion/app/sensitech_parser.py
index 05ff74647..397dc9cd6 100644
--- a/services/ingestion/app/sensitech_parser.py
+++ b/services/ingestion/app/sensitech_parser.py
@@ -227,6 +227,8 @@ async def import_sensitech(
     cte_type: str = Form("cooling", description="CTE type: cooling or receiving"),
     location_name: str = Form(..., description="Facility name where logger was deployed"),
     location_gln: Optional[str] = Form(None, description="Facility GLN (optional)"),
+    location_phone: Optional[str] = Form(None, description="Facility phone (optional)"),
+    location_address: Optional[str] = Form(None, description="Facility address (optional)"),
     cold_threshold: float = Form(5.0, description="Max temperature threshold (°C)"),
     ctx: TenantContext = Depends(require_tenant_context),
     x_regengine_api_key: Optional[str] = Header(default=None, alias="X-RegEngine-API-Key"),
@@ -280,6 +282,8 @@ async def import_sensitech(
             unit_of_measure="readings",
             location_gln=location_gln,
             location_name=location_name,
+            location_phone=location_phone,
+            location_address=location_address,
             timestamp=event_timestamp,
             kdes={
                 date_field: event_date,
diff --git a/services/ingestion/app/supplier_portal.py b/services/ingestion/app/supplier_portal.py
index bc2b5610e..150affebc 100644
--- a/services/ingestion/app/supplier_portal.py
+++ b/services/ingestion/app/supplier_portal.py
@@ -424,8 +424,12 @@ class SupplierSubmission(BaseModel):
     unit_of_measure: str = Field(..., description="Unit (cases, lbs, kg, pallets)")
     ship_date: str = Field(..., description="Ship date (YYYY-MM-DD)")
     ship_from_location: str = Field(..., description="Origin facility name")
+    ship_from_phone: Optional[str] = Field(None, description="Facility phone (optional)")
+    ship_from_address: Optional[str] = Field(None, description="Facility address (optional)")
     ship_from_gln: Optional[str] = Field(None, description="Origin GLN (optional)")
     ship_to_location: str = Field(..., description="Destination facility name")
+    ship_to_phone: Optional[str] = Field(None, description="Facility phone (optional)")
+    ship_to_address: Optional[str] = Field(None, description="Facility address (optional)")
     ship_to_gln: Optional[str] = Field(None, description="Destination GLN (optional)")
     carrier_name: Optional[str] = Field(None, description="Carrier/logistics company")
     po_number: Optional[str] = Field(None, description="Purchase order number")
@@ -464,6 +468,17 @@ def _submission_event_dict(portal_id: str, link: dict, submission: SupplierSubmi
         "submission_source": "supplier_portal",
         "portal_id": portal_id,
     }
+    for field in (
+        "ship_from_phone",
+        "ship_from_address",
+        "ship_to_phone",
+        "ship_to_address",
+        "ship_from_gln",
+        "ship_to_gln",
+    ):
+        value = getattr(submission, field)
+        if value:
+            kdes[field] = value
     if submission.carrier_name:
         kdes["carrier_name"] = submission.carrier_name
     if submission.po_number:
@@ -484,6 +499,8 @@ def _submission_event_dict(portal_id: str, link: dict, submission: SupplierSubmi
         "unit_of_measure": submission.unit_of_measure,
         "location_gln": submission.ship_from_gln,
         "location_name": submission.ship_from_location,
+        "location_phone": submission.ship_from_phone,
+        "location_address": submission.ship_from_address,
         "ship_from_location": submission.ship_from_location,
         "ship_to_location": submission.ship_to_location,
         "reference_document": kdes["reference_document"],
@@ -910,6 +927,8 @@ async def _submit_supplier_data_inner(
         unit_of_measure=submission.unit_of_measure,
         location_gln=submission.ship_from_gln,
         location_name=submission.ship_from_location,
+        location_phone=submission.ship_from_phone,
+        location_address=submission.ship_from_address,
         timestamp=event_data["timestamp"],
         kdes=kdes,
     )
diff --git a/services/ingestion/app/webhook_compat.py b/services/ingestion/app/webhook_compat.py
index 71cb3964b..efbc4cccc 100644
--- a/services/ingestion/app/webhook_compat.py
+++ b/services/ingestion/app/webhook_compat.py
@@ -52,6 +52,7 @@ def _batch_dedup_key(event: IngestEvent) -> str:
     key = (
         f"{event.cte_type.value}|{event.traceability_lot_code}"
         f"|{event.timestamp}|{event.location_gln or event.location_name or ''}"
+        f"|{event.location_phone or ''}|{event.location_address or ''}"
     )
     # Repeated item lines can share a TLC, time, and location. Their
     # interchange position distinguishes 856 SN1, 861 RCD, and 944/945
diff --git a/services/ingestion/app/webhook_models.py b/services/ingestion/app/webhook_models.py
index e39ad308a..04b4d9dd7 100644
--- a/services/ingestion/app/webhook_models.py
+++ b/services/ingestion/app/webhook_models.py
@@ -16,8 +16,8 @@
 from typing import Any, Dict, List, Optional
 from uuid import uuid4
 
-from pydantic import BaseModel, Field, field_validator, model_validator
-from shared.fsma_validation import calculate_gs1_check_digit
+from pydantic import BaseModel, Field, PrivateAttr, field_validator, model_validator
+from shared.fsma_validation import calculate_gs1_check_digit, is_location_description_complete
 
 _gln_logger = logging.getLogger("gln-validation")
 
@@ -173,6 +173,8 @@ class WebhookCTEType(str, Enum):
 class IngestEvent(BaseModel):
     """A single CTE event to ingest."""
 
+    _edi_allow_incomplete_location: bool = PrivateAttr(default=False)
+
     cte_type: WebhookCTEType = Field(..., description="Critical Tracking Event type")
     traceability_lot_code: str = Field(..., description="Traceability Lot Code (TLC)", min_length=3)
     product_description: str = Field(..., description="Human-readable product name", min_length=1, max_length=500)
@@ -180,6 +182,8 @@ class IngestEvent(BaseModel):
     unit_of_measure: str = Field(..., description="Unit (cases, lbs, kg, pallets, etc.)")
     location_gln: Optional[str] = Field(None, description="GS1 Global Location Number (13 digits)")
     location_name: Optional[str] = Field(None, description="Human-readable location name")
+    location_phone: Optional[str] = Field(None, description="Business phone number")
+    location_address: Optional[str] = Field(None, description="Physical location address")
     timestamp: str = Field(..., description="ISO 8601 timestamp of the event")
     kdes: Dict[str, Any] = Field(
         default_factory=dict, description="Additional Key Data Elements (temperature, carrier, PO, field ID, etc.)"
@@ -265,28 +269,33 @@ def validate_unit_of_measure(cls, v: str) -> str:
 
     @model_validator(mode="after")
     def require_location(self) -> "IngestEvent":
-        """FSMA 204 requires location description for every CTE (§1.1325–§1.1350)."""
-        has_gln = self.location_gln and self.location_gln.strip()
-        has_name = self.location_name and self.location_name.strip()
-        if not has_gln and not has_name:
-            # Also check KDEs for location fields
-            kde_locations = [
-                "ship_from_location",
-                "ship_to_location",
-                "receiving_location",
-                "ship_from_gln",
-                "ship_to_gln",
-            ]
-            has_kde_location = any(
-                self.kdes.get(k) for k in kde_locations if self.kdes.get(k) and str(self.kdes[k]).strip()
-            )
-            if not has_kde_location:
-                raise ValueError(
-                    "FSMA 204 requires at least one location identifier per CTE. "
-                    "Provide location_gln, location_name, or a location KDE "
-                    "(ship_from_location, receiving_location, etc.)"
-                )
-        return self
+        """Require one complete location description or GLN per 21 CFR 1.1310."""
+
+        # A-2170: only internally constructed EDI events may accept incomplete locations.
+        if self._edi_allow_incomplete_location and self.kdes.get("location_incomplete") is True:
+            _gln_logger.warning("location_incomplete accepted=true")
+            return self
+
+        def first_nonblank(*values: Any) -> Optional[str]:
+            return next((value for value in values if isinstance(value, str) and value.strip()), None)
+
+        for prefix in ("location", "ship_from", "ship_to", "receiving"):
+
+            def description_field(field: str) -> Optional[str]:
+                top_level = getattr(self, f"location_{field}") if prefix == "location" else None
+                legacy_name = self.kdes.get(f"{prefix}_location") if field == "name" else None
+                return first_nonblank(top_level, self.kdes.get(f"{prefix}_{field}"), legacy_name)
+
+            if is_location_description_complete(
+                gln=first_nonblank(self.location_gln, self.kdes.get(f"{prefix}_gln")),
+                name=description_field("name"),
+                phone=description_field("phone"),
+                address=description_field("address"),
+            ):
+                return self
+        raise ValueError(
+            "FSMA 204 requires (name AND phone AND address) OR GLN for at least one location per CTE (21 CFR 1.1310)."
+        )
 
 
 class WebhookPayload(BaseModel):
diff --git a/services/ingestion/app/webhook_router_v2/routes.py b/services/ingestion/app/webhook_router_v2/routes.py
index bb8902e7c..5cf700249 100644
--- a/services/ingestion/app/webhook_router_v2/routes.py
+++ b/services/ingestion/app/webhook_router_v2/routes.py
@@ -175,14 +175,16 @@ async def ingest_events(
         valid_events: list = []  # (event, alerts) tuples
         for event in payload.events:
             # Batch deduplication — skip identical events in same request
-            dedup_key = f"{event.cte_type.value}|{event.traceability_lot_code}|{event.timestamp}|{event.location_gln or event.location_name or ''}"
+            dedup_key = f"{event.cte_type.value}|{event.traceability_lot_code}|{event.timestamp}|{event.location_gln or event.location_name or ''}|{event.location_phone or ''}|{event.location_address or ''}"
             if dedup_key in seen_in_batch:
                 results.append(
                     EventResult(
                         traceability_lot_code=event.traceability_lot_code,
                         cte_type=event.cte_type.value,
                         status="rejected",
-                        errors=["Duplicate event in batch — same CTE type, TLC, timestamp, and location"],
+                        errors=[
+                            "Duplicate event in batch — same CTE type, TLC, timestamp, location, phone, and address"
+                        ],
                     )
                 )
                 rejected += 1
@@ -297,6 +299,8 @@ async def ingest_events(
                         "event_timestamp": event.timestamp,
                         "location_gln": event.location_gln,
                         "location_name": event.location_name,
+                        "location_phone": event.location_phone,
+                        "location_address": event.location_address,
                         "kdes": event.kdes,
                         # One ledger entry serves both persisted projections.
                         # Construct its canonical content on the server; never
diff --git a/services/ingestion/app/webhook_router_v2/validation.py b/services/ingestion/app/webhook_router_v2/validation.py
index b94c0c220..12dacd7cd 100644
--- a/services/ingestion/app/webhook_router_v2/validation.py
+++ b/services/ingestion/app/webhook_router_v2/validation.py
@@ -41,6 +41,8 @@ def _authoritative_kdes(event: IngestEvent) -> dict[str, object]:
         ("unit_of_measure", event.unit_of_measure),
         ("location_name", event.location_name),
         ("location_gln", event.location_gln),
+        ("location_phone", event.location_phone),
+        ("location_address", event.location_address),
     ):
         if _val is None:
             continue
diff --git a/services/ingestion/tests/conftest.py b/services/ingestion/tests/conftest.py
index f61fdaa93..ed481e66c 100644
--- a/services/ingestion/tests/conftest.py
+++ b/services/ingestion/tests/conftest.py
@@ -14,6 +14,8 @@
 from pathlib import Path
 from types import ModuleType
 
+import pytest
+
 _service_dir = str(Path(__file__).resolve().parents[1])  # services/ingestion
 if _service_dir not in sys.path:
     sys.path.insert(0, _service_dir)
@@ -129,3 +131,13 @@ def _purge_stale_modules() -> None:
 
 def pytest_runtest_setup(item):  # type: ignore[no-untyped-def]
     _purge_stale_modules()
+
+
+@pytest.fixture
+def location_complete_for_pipeline(monkeypatch):
+    """Isolate pipeline tests from location completeness without inventing data.
+
+    Opt in only for parsing/flow tests. Location rejection tests must use the
+    real 21 CFR 1.1310 validator.
+    """
+    monkeypatch.setattr("app.webhook_models.is_location_description_complete", lambda **kwargs: True)
diff --git a/services/ingestion/tests/test_csv_ingest_response_model_2892.py b/services/ingestion/tests/test_csv_ingest_response_model_2892.py
index 443a88157..4357eed80 100644
--- a/services/ingestion/tests/test_csv_ingest_response_model_2892.py
+++ b/services/ingestion/tests/test_csv_ingest_response_model_2892.py
@@ -169,9 +169,7 @@ def test_base_fields_unchanged(self):
 class TestResponseModelIsEnforced:
     def test_declared_response_model_carries_the_warning_fields(self):
         """``response_model`` is the single source of truth for the shape."""
-        route = next(
-            r for r in router.routes if str(getattr(r, "path", "")).endswith("/ingest/csv")
-        )
+        route = next(r for r in router.routes if str(getattr(r, "path", "")).endswith("/ingest/csv"))
         model = route.response_model
         assert model is not None
         assert "kde_warnings" in model.model_fields
@@ -179,3 +177,8 @@ def test_declared_response_model_carries_the_warning_fields(self):
         # Still a superset of the shared ingest response.
         for field in IngestResponse.model_fields:
             assert field in model.model_fields
+
+
+# These tests cover parsing and pipeline flow; location validity is covered by
+# test_location_description_complete.py and test_webhook_models.py.
+pytestmark = pytest.mark.usefixtures("location_complete_for_pipeline")
diff --git a/services/ingestion/tests/test_csv_ingestion_receipt.py b/services/ingestion/tests/test_csv_ingestion_receipt.py
index c73868c6b..a5d142005 100644
--- a/services/ingestion/tests/test_csv_ingestion_receipt.py
+++ b/services/ingestion/tests/test_csv_ingestion_receipt.py
@@ -7,6 +7,8 @@
 
 from __future__ import annotations
 
+import pytest
+
 # ---------------------------------------------------------------------------
 # /ingest/csv retains the upload as an ingestion run
 # ---------------------------------------------------------------------------
@@ -69,3 +71,8 @@ async def _fake(payload, **kwargs):
     body = resp.json()
     assert body["ingestion_run_id"] == "run-abc"
     assert body["rejected"] == 1 and body["accepted"] == 1
+
+
+# These tests cover parsing and pipeline flow; location validity is covered by
+# test_location_description_complete.py and test_webhook_models.py.
+pytestmark = pytest.mark.usefixtures("location_complete_for_pipeline")
diff --git a/services/ingestion/tests/test_csv_jev_header_mapping.py b/services/ingestion/tests/test_csv_jev_header_mapping.py
index 662396473..48d30ed20 100644
--- a/services/ingestion/tests/test_csv_jev_header_mapping.py
+++ b/services/ingestion/tests/test_csv_jev_header_mapping.py
@@ -976,3 +976,8 @@ def test_upload_ui_labels_every_field_jev_can_name_with_a_header_the_parser_read
     for field, label in labels.items():
         assert normalize_csv_header(label) == field
         assert field in ct._RECOGNIZED_COLUMNS
+
+
+# These tests cover parsing and pipeline flow; location validity is covered by
+# test_location_description_complete.py and test_webhook_models.py.
+pytestmark = pytest.mark.usefixtures("location_complete_for_pipeline")
diff --git a/services/ingestion/tests/test_csv_malformed_input.py b/services/ingestion/tests/test_csv_malformed_input.py
index 0cf5ca43b..acdd4d66b 100644
--- a/services/ingestion/tests/test_csv_malformed_input.py
+++ b/services/ingestion/tests/test_csv_malformed_input.py
@@ -227,3 +227,8 @@ def _pre_fix_parse_csv_rows(text, fallback_cte=None, header_mapping=None):
     post, _ = upload
     resp = post(f"{HEADER}\n{ROW},unexpected\n")
     assert resp.status_code == 500
+
+
+# These tests cover parsing and pipeline flow; location validity is covered by
+# test_location_description_complete.py and test_webhook_models.py.
+pytestmark = pytest.mark.usefixtures("location_complete_for_pipeline")
diff --git a/services/ingestion/tests/test_csv_templates.py b/services/ingestion/tests/test_csv_templates.py
index 1fc56c74b..6f763bdad 100644
--- a/services/ingestion/tests/test_csv_templates.py
+++ b/services/ingestion/tests/test_csv_templates.py
@@ -130,7 +130,7 @@ def _make_event(self, cte_type_str, **kdes_extra):
 
         now = datetime.now(timezone.utc)
 
-        # Pick a location field that satisfies the model validator
+        # Pick the existing location label for this CTE; completeness is patched.
         loc_kwargs = {}
         if cte_type_str in ("shipping",):
             loc_kwargs["kdes"] = {"ship_from_location": "Origin", "ship_to_location": "Dest"}
@@ -182,3 +182,8 @@ def test_unknown_cte_type_returns_empty(self):
         event = self._make_event("harvesting")
         missing = _validate_kde_completeness("nonexistent_type", event, {})
         assert missing == []
+
+
+# These tests cover parsing and pipeline flow; location validity is covered by
+# test_location_description_complete.py and test_webhook_models.py.
+pytestmark = pytest.mark.usefixtures("location_complete_for_pipeline")
diff --git a/services/ingestion/tests/test_csv_templates_1342.py b/services/ingestion/tests/test_csv_templates_1342.py
index dffd759aa..c7d1d62f1 100644
--- a/services/ingestion/tests/test_csv_templates_1342.py
+++ b/services/ingestion/tests/test_csv_templates_1342.py
@@ -1135,3 +1135,8 @@ def test_complete_row_still_ingests(self):
         assert events[0].quantity == 10.0
         assert events[0].unit_of_measure == "lbs"
         assert events[0].timestamp.startswith("2026-04-17")
+
+
+# These tests cover parsing and pipeline flow; location validity is covered by
+# test_location_description_complete.py and test_webhook_models.py.
+pytestmark = pytest.mark.usefixtures("location_complete_for_pipeline")
diff --git a/services/ingestion/tests/test_csv_xlsx_upload.py b/services/ingestion/tests/test_csv_xlsx_upload.py
index d336170ce..49a885ee9 100644
--- a/services/ingestion/tests/test_csv_xlsx_upload.py
+++ b/services/ingestion/tests/test_csv_xlsx_upload.py
@@ -369,3 +369,8 @@ def test_csv_sent_as_ms_excel_content_type_still_reads_as_csv(client, captured):
     assert resp.status_code == 200, resp.text
     assert resp.json()["accepted"] == 2
     assert captured["payload"].source == "csv_upload"
+
+
+# These tests cover parsing and pipeline flow; location validity is covered by
+# test_location_description_complete.py and test_webhook_models.py.
+pytestmark = pytest.mark.usefixtures("location_complete_for_pipeline")
diff --git a/services/ingestion/tests/test_edi_location_incomplete_a2170.py b/services/ingestion/tests/test_edi_location_incomplete_a2170.py
new file mode 100644
index 000000000..b8070ced1
--- /dev/null
+++ b/services/ingestion/tests/test_edi_location_incomplete_a2170.py
@@ -0,0 +1,137 @@
+"""A-2170 accept-and-flag checks using structural markers only."""
+
+from datetime import datetime, timezone
+
+import pytest
+from app.edi_ingestion import routes
+from app.epcis_export import _gln_csv_cell
+from app.webhook_models import VALID_UNITS_OF_MEASURE, IngestEvent
+from fastapi import HTTPException
+from pydantic import ValidationError
+
+
+@pytest.fixture
+def build(monkeypatch):
+    # Isolate location checks from X12 date syntax without fabricating a date.
+    monkeypatch.setattr(routes, "_ship_timestamp_and_date", lambda *args: (datetime.now(timezone.utc).isoformat(), ""))
+
+    def event(transaction_set="856", **changes):
+        extracted = {
+            "ship_from_name": " x ",
+            "seller_name": " x ",
+            "receiving_location_name": " x ",
+            "quantity": float(True),
+        }
+        extracted.update(changes.pop("extracted", {}))
+        kwargs = dict(
+            transaction_set=transaction_set,
+            extracted=extracted,
+            traceability_lot_code="xxx",
+            product_description_override="x",
+            quantity_override=None,
+            unit_override=sorted(VALID_UNITS_OF_MEASURE)[0],
+            location_override=None,
+        )
+        kwargs.update(changes)
+        return routes._build_ingest_event_for_set(**kwargs)
+
+    return event
+
+
+@pytest.mark.parametrize(
+    "transaction_set,key",
+    [
+        ("856", "ship_from_gln"),
+        ("850", "ship_from_gln"),
+        ("810", "seller_gln"),
+        ("861", "receiving_location_gln"),
+        ("944", "receiving_location_gln"),
+        ("945", "ship_from_gln"),
+    ],
+)
+@pytest.mark.parametrize("raw", [None, "", " ", "x", "1", " x "])
+def test_incomplete_edi_is_accepted_with_raw_evidence(build, transaction_set, key, raw, caplog):
+    event = build(transaction_set, extracted={key: raw})
+    assert event.kdes["location_incomplete"] is True
+    assert event.kdes[key] == raw
+    assert event.location_gln is None
+    assert event.location_name == " x "
+    assert "location_incomplete accepted=true" in caplog.text
+
+
+@pytest.mark.parametrize("raw", [None, "", " ", "x", "1", " x ", "x" * 13])
+def test_csv_omits_invalid_gln(raw):
+    assert _gln_csv_cell(raw) == ""
+
+
+def test_unflagged_location_still_rejected(build):
+    event = build()
+    payload = event.model_dump()
+    payload["kdes"] = {}
+    with pytest.raises(ValidationError, match="21 CFR"):
+        IngestEvent(**payload)
+    payload["kdes"]["location_incomplete"] = True
+    with pytest.raises(ValidationError, match="21 CFR"):
+        IngestEvent(**payload)
+
+
+@pytest.mark.parametrize("lot", ["", "x"])
+def test_missing_or_short_lot_still_rejected(build, lot):
+    with pytest.raises(ValidationError):
+        build(traceability_lot_code=lot)
+
+
+@pytest.mark.parametrize("quantity", [0, -float(True), float("nan"), float("inf")])
+def test_bad_quantity_still_rejected(build, quantity):
+    with pytest.raises(ValidationError):
+        build(quantity_override=quantity)
+
+
+@pytest.mark.parametrize("date", [None, "", " ", "x"])
+def test_missing_bad_date_still_rejected(date):
+    with pytest.raises(HTTPException) as error:
+        routes._build_ingest_event_for_set("856", {"ship_date_raw": date}, "xxx", "x", float(True), None, None)
+    assert error.value.status_code == 400
+
+
+@pytest.mark.parametrize("raw", [None, "", " ", "x", "1", "x" * 13])
+def test_fda_service_omits_invalid_gln(raw):
+    from app.fda_export_service import _event_to_fda_row
+
+    row = _event_to_fda_row(
+        {"location_gln": raw, "kdes": {"ship_from_gln": raw, "ship_to_gln": raw, "tlc_source_gln": raw}}
+    )
+    for column in ("Location GLN", "Ship From GLN", "Ship To GLN", "TLC Source GLN"):
+        assert row[column] == ""
+
+
+@pytest.mark.parametrize("raw", [None, "", " ", " x "])
+def test_raw_name_preserved(build, raw):
+    event = build("810", extracted={"seller_name": raw, "seller_gln": "x"})
+    assert event.location_name == raw
+    assert event.kdes["seller_name"] == raw
+
+
+def test_quantity_element_cannot_be_replaced_with_default(build):
+    with pytest.raises(HTTPException) as error:
+        build(extracted={"quantity": None}, segments=[["SN1", "", "x"]])
+    assert error.value.status_code == 422
+
+
+@pytest.mark.parametrize("source", ["csv", "portal", "sensitech", "webhook"])
+@pytest.mark.parametrize("private_spoof", [False, True])
+def test_non_edi_cannot_spoof_bypass(build, source, private_spoof):
+    payload = build().model_dump()
+    payload["kdes"]["source"] = source
+    if private_spoof:
+        payload["_edi_allow_incomplete_location"] = True
+    with pytest.raises(ValidationError, match="21 CFR"):
+        IngestEvent.model_validate(payload)
+
+
+def test_edi_private_flag_survives_nested_validation(build):
+    from app.webhook_models import WebhookPayload
+
+    event = build()
+    assert WebhookPayload(source="x", events=[event]).events[0].kdes["location_incomplete"]
+    assert "_edi_allow_incomplete_location" not in event.model_dump()
diff --git a/services/ingestion/tests/test_exchange_api_1342.py b/services/ingestion/tests/test_exchange_api_1342.py
index e965fb403..4d71d1b83 100644
--- a/services/ingestion/tests/test_exchange_api_1342.py
+++ b/services/ingestion/tests/test_exchange_api_1342.py
@@ -306,15 +306,15 @@ def test_basic_record_shape(self) -> None:
         assert record["unit_of_measure"] == "case"
         # KDEs > row fallbacks.
         assert record["ship_from"] == {
-            "gln": "0810000000001",
+            "gln": "",
             "name": "Packhouse",
         }
         assert record["ship_to"] == {
-            "gln": "0810000000002",
+            "gln": "",
             "name": "Distributor",
         }
         assert record["tlc_source"] == {
-            "gln": "0810000000003",
+            "gln": "",
             "fda_registration": "FDA-123",
             "immediate_previous_source": "Upstream",
         }
@@ -333,13 +333,13 @@ def test_ship_from_falls_back_to_location_fields(self) -> None:
         records = _build_records([row])
         record = records[0]
         assert record["ship_from"] == {
-            "gln": row.location_gln,
+            "gln": "",
             "name": row.location_name,
         }
         # No ship_to_location / receiving_location -> both None.
-        assert record["ship_to"] == {"gln": None, "name": None}
+        assert record["ship_to"] == {"gln": "", "name": None}
         assert record["tlc_source"] == {
-            "gln": None,
+            "gln": "",
             "fda_registration": None,
             "immediate_previous_source": None,
         }
diff --git a/services/ingestion/tests/test_hermes_export_gln.py b/services/ingestion/tests/test_hermes_export_gln.py
new file mode 100644
index 000000000..db4c39073
--- /dev/null
+++ b/services/ingestion/tests/test_hermes_export_gln.py
@@ -0,0 +1,108 @@
+"""Outbound GLNs use structural inputs only."""
+
+from types import SimpleNamespace
+
+import pytest
+from app import exchange_api
+from shared.fsma_validation import calculate_gs1_check_digit
+
+
+def shapes():
+    base = "0" * 12
+    return [None, "", " ", "x", "0", "x" * 13, base + str(calculate_gs1_check_digit(base))]
+
+
+@pytest.mark.parametrize("raw", shapes())
+def test_exchange_filters_all_glns(raw):
+    from shared.fsma_validation import gln_for_export
+
+    row = SimpleNamespace(
+        **dict.fromkeys(
+            (
+                "id",
+                "event_timestamp",
+                "product_description",
+                "quantity",
+                "unit_of_measure",
+                "source_event_id",
+                "sha256_hash",
+                "chain_hash",
+                "source",
+                "location_name",
+                "traceability_lot_code",
+            )
+        )
+    )
+    row.location_gln = raw
+    row.kdes = dict.fromkeys(("ship_from_gln", "ship_to_gln", "tlc_source_gln"), raw)
+    record = exchange_api._build_records([row])[0]
+    for key in ("ship_from", "ship_to", "tlc_source"):
+        assert record[key]["gln"] == gln_for_export(raw)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("raw", shapes())
+@pytest.mark.parametrize("fallback", [False, True])
+async def test_retailer_filters_both_glns(monkeypatch, raw, fallback):
+    from shared.external_connectors import retailers
+    from shared.fsma_validation import gln_for_export
+
+    emitted = []
+
+    def filtered(value):
+        result = gln_for_export(value)
+        emitted.append(result)
+        return result
+
+    monkeypatch.setattr(retailers, "gln_for_export", filtered)
+    event = SimpleNamespace(
+        cte_type="shipping",
+        quantity=float(True),
+        unit_of_measure="x",
+        source_event_id="x",
+        timestamp="",
+        kdes={"ship_from_gln": raw, "ship_to_gln": raw},
+        location_gln=None if fallback else raw,
+        traceability_lot_code="x",
+        product_description="x",
+    )
+    connector = object.__new__(retailers.WalmartGDSNConnector)
+    assert await connector.push_event(event)
+    assert emitted == [gln_for_export(raw), gln_for_export(raw)]
+
+
+@pytest.mark.parametrize("raw", shapes())
+def test_exchange_additional_gln_filtered(raw):
+    row = SimpleNamespace(
+        **dict.fromkeys(
+            (
+                "id",
+                "event_timestamp",
+                "product_description",
+                "quantity",
+                "unit_of_measure",
+                "source_event_id",
+                "sha256_hash",
+                "chain_hash",
+                "source",
+                "location_name",
+                "traceability_lot_code",
+                "location_gln",
+            )
+        )
+    )
+    row.kdes = {"receiving_location_gln": raw}
+    from shared.fsma_validation import gln_for_export
+
+    assert exchange_api._build_records([row])[0]["additional_kdes"]["receiving_location_gln"] == gln_for_export(raw)
+
+
+@pytest.mark.parametrize("raw", shapes())
+def test_plan_exports_filter_gln(raw):
+    from shared.fsma_plan_builder import FirmInfo, FirmType, TraceabilityPlan, export_plan_markdown
+    from shared.fsma_validation import gln_for_export
+
+    firm = FirmInfo("", "", next(iter(FirmType)), gln=raw)
+    plan = TraceabilityPlan("x", firm, "", "", "")
+    assert firm.to_dict()["gln"] == gln_for_export(raw)
+    assert f"- **GLN:** {gln_for_export(raw) or 'N/A'}\n" in export_plan_markdown(plan)
diff --git a/services/ingestion/tests/test_inflow_lab_contract.py b/services/ingestion/tests/test_inflow_lab_contract.py
index 0a253dccf..90bc21155 100644
--- a/services/ingestion/tests/test_inflow_lab_contract.py
+++ b/services/ingestion/tests/test_inflow_lab_contract.py
@@ -6,6 +6,9 @@
 from app.webhook_models import IngestEvent
 from app.webhook_router_v2 import _validate_event_kdes
 
+# These tests cover Inflow Lab KDE contracts; location validity is covered elsewhere.
+pytestmark = pytest.mark.usefixtures("location_complete_for_pipeline")
+
 
 def _timestamp() -> str:
     return datetime.now(timezone.utc).isoformat()
diff --git a/services/ingestion/tests/test_sensitech_parser_1342.py b/services/ingestion/tests/test_sensitech_parser_1342.py
index 2fa1d93c3..fae345d7b 100644
--- a/services/ingestion/tests/test_sensitech_parser_1342.py
+++ b/services/ingestion/tests/test_sensitech_parser_1342.py
@@ -358,3 +358,8 @@ def test_receiving_cte_type_accepted(self, client: TestClient, captured_payload:
         assert event.cte_type.value == "receiving"
         # receiving_date KDE populated, not cooling_date.
         assert event.kdes["receiving_date"] == "2026-02-26"
+
+
+# These tests cover parsing and pipeline flow; location validity is covered by
+# test_location_description_complete.py and test_webhook_models.py.
+pytestmark = pytest.mark.usefixtures("location_complete_for_pipeline")
diff --git a/services/ingestion/tests/test_webhook_compat_1342.py b/services/ingestion/tests/test_webhook_compat_1342.py
index 0e1172a8c..b8815264b 100644
--- a/services/ingestion/tests/test_webhook_compat_1342.py
+++ b/services/ingestion/tests/test_webhook_compat_1342.py
@@ -28,6 +28,7 @@
 
 import asyncio
 import sys
+from copy import copy
 from types import ModuleType, SimpleNamespace
 from unittest.mock import MagicMock
 
@@ -102,6 +103,8 @@ def _make_event(
     timestamp="2026-01-01T00:00:00Z",
     location_gln=None,
     location_name="Farm",
+    location_phone=None,
+    location_address=None,
     kdes=None,
 ):
     """Build a SimpleNamespace that quacks like IngestEvent."""
@@ -114,6 +117,8 @@ def _make_event(
         timestamp=timestamp,
         location_gln=location_gln,
         location_name=location_name,
+        location_phone=location_phone,
+        location_address=location_address,
         kdes=kdes if kdes is not None else {},
     )
 
@@ -575,6 +580,16 @@ def test_location_gln_used_in_dedup_key(self, monkeypatch):
         assert resp.rejected == 0
         assert len(persist.store_calls) == 2
 
+    @pytest.mark.parametrize("field", ["location_phone", "location_address"])
+    def test_location_contact_fields_used_in_dedup_key(self, field):
+        event = _make_event(location_phone="A", location_address="X")
+        identical = copy(event)
+        assert wc._batch_dedup_key(event) == wc._batch_dedup_key(identical)
+
+        changed = copy(event)
+        setattr(changed, field, getattr(event, field) + "B")
+        assert wc._batch_dedup_key(event) != wc._batch_dedup_key(changed)
+
     def test_missing_location_fields_use_empty_string_in_dedup(self, monkeypatch):
         """location_gln=None and location_name=None → '' in dedup_key.
         Two such events with identical other fields collapse to one."""
diff --git a/services/ingestion/tests/test_webhook_models.py b/services/ingestion/tests/test_webhook_models.py
index 9509f5984..90489ea2a 100644
--- a/services/ingestion/tests/test_webhook_models.py
+++ b/services/ingestion/tests/test_webhook_models.py
@@ -95,5 +95,5 @@ def test_rejects_invalid_iso_timestamp(self):
 
     def test_requires_location_gln_or_name(self):
         empty_event = models.IngestEvent.model_construct(kdes={})
-        with pytest.raises(ValueError, match="location identifier"):
+        with pytest.raises(ValueError, match=r"21 CFR 1\.1310"):
             empty_event.require_location()
diff --git a/services/shared/canonical_event.py b/services/shared/canonical_event.py
index 193a8bca7..998451b58 100644
--- a/services/shared/canonical_event.py
+++ b/services/shared/canonical_event.py
@@ -772,6 +772,8 @@ def normalize_webhook_event(
             "unit_of_measure": event.unit_of_measure,
             "location_gln": event.location_gln,
             "location_name": event.location_name,
+            "location_phone": getattr(event, "location_phone", None),
+            "location_address": getattr(event, "location_address", None),
             "timestamp": event.timestamp,
             "kdes": event.kdes,
         }
@@ -781,6 +783,10 @@ def normalize_webhook_event(
     # Extract facility references from KDEs
     kdes = dict(event.kdes) if event.kdes else {}
     kdes.setdefault("ingest_source", source)
+    for field in ("location_phone", "location_address"):
+        value = getattr(event, field, None)
+        if isinstance(value, str) and value.strip():
+            kdes.setdefault(field, value)
 
     # Promote input_traceability_lot_codes into kdes so _create_transformation_links()
     # can find them under the expected "input_lot_codes" key.  The webhook model keeps
@@ -1038,8 +1044,7 @@ def _resolve_bizstep_to_cte_type(biz_step: Any) -> CTEType:
     raw = "" if biz_step is None else str(biz_step).strip()
     if not raw:
         raise UnmappedBizStepError(
-            "EPCIS event has no bizStep — cannot derive an FSMA 204 CTE type. "
-            "Supply a CBV or FSMA bizStep URI. (#2815)"
+            "EPCIS event has no bizStep — cannot derive an FSMA 204 CTE type. Supply a CBV or FSMA bizStep URI. (#2815)"
         )
 
     lowered = raw.lower()
diff --git a/services/shared/external_connectors/retailers.py b/services/shared/external_connectors/retailers.py
index ff9d23c55..220e8d03c 100644
--- a/services/shared/external_connectors/retailers.py
+++ b/services/shared/external_connectors/retailers.py
@@ -20,6 +20,8 @@
 from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional
 
+from shared.fsma_validation import gln_for_export
+
 from .base import (
     AuthType,
     ConnectionStatus,
@@ -117,8 +119,8 @@ async def push_event(self, event: NormalizedCTEEvent) -> bool:
                     "quantity": event.quantity,
                     "unitOfMeasure": event.unit_of_measure.upper(),
                     "productDescription": event.product_description,
-                    "shipFromGln": event.location_gln or event.kdes.get("ship_from_gln", ""),
-                    "shipToGln": event.kdes.get("ship_to_gln", ""),
+                    "shipFromGln": gln_for_export(event.location_gln or event.kdes.get("ship_from_gln", "")),
+                    "shipToGln": gln_for_export(event.kdes.get("ship_to_gln", "")),
                 }
             ],
         }
diff --git a/services/shared/fsma_plan_builder.py b/services/shared/fsma_plan_builder.py
index 0840ee117..66cfbe8b4 100644
--- a/services/shared/fsma_plan_builder.py
+++ b/services/shared/fsma_plan_builder.py
@@ -18,6 +18,7 @@
 from typing import Any, Dict, List, Optional
 
 import structlog
+from shared.fsma_validation import gln_for_export
 
 logger = structlog.get_logger("fsma-plan-builder")
 
@@ -61,7 +62,7 @@ def to_dict(self) -> Dict[str, Any]:
             "name": self.name,
             "address": self.address,
             "firm_type": self.firm_type.value,
-            "gln": self.gln,
+            "gln": gln_for_export(self.gln),
             "fda_registration": self.fda_registration,
             "contact_name": self.contact_name,
             "contact_email": self.contact_email,
@@ -843,7 +844,7 @@ def export_plan_markdown(plan: TraceabilityPlan) -> str:
 - **Company:** {plan.firm.name}
 - **Address:** {plan.firm.address}
 - **FDA Registration:** {plan.firm.fda_registration or "N/A"}
-- **GLN:** {plan.firm.gln or "N/A"}
+- **GLN:** {gln_for_export(plan.firm.gln) or "N/A"}
 
 - **Plan Version:** {plan.version}
 - **Created:** {plan.created_date}
diff --git a/services/shared/fsma_validation.py b/services/shared/fsma_validation.py
index f2e2dd56c..2b12a14f5 100644
--- a/services/shared/fsma_validation.py
+++ b/services/shared/fsma_validation.py
@@ -109,6 +109,31 @@ def _validate_gs1_check(clean: str, field_name: str, id_type: IdentifierType) ->
 # ---------------------------------------------------------------------------
 
 
+def gln_for_export(value: object) -> str:
+    """Emit only verbatim ASCII GLNs with a valid GS1 check digit."""
+    if isinstance(value, str) and len(value) == 13 and value.isascii() and value.isdigit():
+        if verify_gs1_check_digit(value):
+            return value
+    return ""
+
+
+def is_location_description_complete(
+    *,
+    name: Optional[str] = None,
+    phone: Optional[str] = None,
+    address: Optional[str] = None,
+    gln: Optional[str] = None,
+) -> bool:
+    """Require name, phone and address, or a structurally valid GS1 GLN (21 CFR 1.1310)."""
+
+    def present(value: Optional[str]) -> bool:
+        return isinstance(value, str) and bool(value.strip())
+
+    return (present(gln) and len(gln) == 13 and gln.isascii() and gln.isdigit() and validate_gln(gln).is_valid) or all(
+        present(value) for value in (name, phone, address)
+    )
+
+
 def validate_gln(gln: str) -> ValidationResult:
     """Validate GS1 Global Location Number (13 digits with check digit)."""
     if not isinstance(gln, str) or not gln:
diff --git a/services/shared/rules/seeds.py b/services/shared/rules/seeds.py
index fc49fdd02..fd7853cc6 100644
--- a/services/shared/rules/seeds.py
+++ b/services/shared/rules/seeds.py
@@ -397,11 +397,12 @@
     },
     {
         "title": "Location Identifier Required",
-        "description": "Every CTE must identify at least one facility location (GLN or description)",
+        "description": "Every CTE requires a GLN or a complete location description (name AND phone AND address) per 21 CFR 1.1310",
         "severity": "critical",
         "category": "kde_presence",
         "applicability_conditions": {"cte_types": [], "ftl_scope": ["ALL"]},
         "citation_reference": "21 CFR §§1.1325(a)(1)(iv), 1.1325(b)(1)(iv), 1.1330(a)(14), 1.1335(e), 1.1340(a)(5), 1.1345(a)(5), 1.1350(a)(2)(ii)",
+        # Presence screening only; webhook validation enforces description completeness.
         "evaluation_logic": {
             "type": "multi_field_presence",
             "field": "from_facility_reference",
@@ -414,8 +415,8 @@
                 ]
             },
         },
-        "failure_reason_template": "Event missing facility location identifier ({citation})",
-        "remediation_suggestion": "Provide at least one location identifier: GLN (preferred) or location description",
+        "failure_reason_template": "Event requires GLN OR (name AND phone AND address) per 21 CFR 1.1310 ({citation})",
+        "remediation_suggestion": "Provide a GLN OR all three location description fields: name AND phone AND address (21 CFR 1.1310)",
     },
     # --- Identifier Format Rules ---
     {
diff --git a/services/shared/schemas.py b/services/shared/schemas.py
index 517ec0ef0..37e8f5c99 100644
--- a/services/shared/schemas.py
+++ b/services/shared/schemas.py
@@ -266,6 +266,7 @@ class Location(BaseModel):
 
     gln: Optional[str] = Field(None, description="GS1 Global Location Number (13 digits)")
     name: Optional[str] = Field(None, description="Human-readable location name")
+    phone: Optional[str] = Field(None, description="Business phone number")
     address: Optional[str] = Field(None, description="Physical address")
 
     model_config = {
@@ -473,6 +474,7 @@ class FSMALocation(BaseModel):
     gln: Optional[str] = Field(None, description="GS1 Global Location Number (13 digits)")
     fda_registration: Optional[str] = Field(None, description="FDA Facility Registration Number")
     name: Optional[str] = Field(None, description="Human-readable location name")
+    phone: Optional[str] = Field(None, description="Business phone number")
     address: Optional[str] = Field(None, description="Physical address")
 
     model_config = {
diff --git a/tests/shared/test_email_ingest_restricted_role_pg.py b/tests/shared/test_email_ingest_restricted_role_pg.py
index a546c7a4b..3d4c907a8 100644
--- a/tests/shared/test_email_ingest_restricted_role_pg.py
+++ b/tests/shared/test_email_ingest_restricted_role_pg.py
@@ -71,6 +71,8 @@ def unix_only(sock, address):
         return connect(sock, address)
 
     monkeypatch.setattr(socket.socket, "connect", unix_only)
+    # Restricted-role email ingest fixture — not 21 CFR 1.1310 location rules.
+    monkeypatch.setattr("app.webhook_models.is_location_description_complete", lambda **kwargs: True)
     monkeypatch.setenv("EMAIL_INBOUND_BASIC_AUTH_USER", BASIC_USER)
     monkeypatch.setenv("EMAIL_INBOUND_BASIC_AUTH_PASSWORD", BASIC_PASSWORD)
     _load("pg_email_ingest_proof", ROOT / "scripts/verify_inflow_to_regengine_e2e.py")._configure_paths()
diff --git a/tests/shared/test_postgres_evidence_workflow_pg.py b/tests/shared/test_postgres_evidence_workflow_pg.py
index c851a77bb..adb9997e2 100644
--- a/tests/shared/test_postgres_evidence_workflow_pg.py
+++ b/tests/shared/test_postgres_evidence_workflow_pg.py
@@ -69,6 +69,8 @@ def unix_only(sock, address):
         return connect(sock, address)
 
     monkeypatch.setattr(socket.socket, "connect", unix_only)
+    # Postgres evidence/RLS/hash-chain fixture — not 21 CFR 1.1310 location rules.
+    monkeypatch.setattr("app.webhook_models.is_location_description_complete", lambda **kwargs: True)
     proof = _load("pg_evidence_proof", ROOT / "scripts/verify_inflow_to_regengine_e2e.py")
     proof._configure_paths()
     admin = psycopg.connect(**admin_connect)
diff --git a/tests/test_canonical_event.py b/tests/test_canonical_event.py
index d82b38291..99260fb0a 100644
--- a/tests/test_canonical_event.py
+++ b/tests/test_canonical_event.py
@@ -44,6 +44,8 @@ def __init__(self, **kwargs):
         self.unit_of_measure = kwargs.get("unit_of_measure", "cases")
         self.location_gln = kwargs.get("location_gln", "0061414100001")
         self.location_name = kwargs.get("location_name", "Acme Distribution Center")
+        self.location_phone = kwargs.get("location_phone", None)
+        self.location_address = kwargs.get("location_address", None)
         self.timestamp = kwargs.get("timestamp", "2026-03-25T14:30:00Z")
         self.kdes = kwargs.get(
             "kdes",
diff --git a/tests/test_shared/test_location_description_complete.py b/tests/test_shared/test_location_description_complete.py
new file mode 100644
index 000000000..bdf5156ee
--- /dev/null
+++ b/tests/test_shared/test_location_description_complete.py
@@ -0,0 +1,131 @@
+"""Data-free location completeness checks using only non-blank markers."""
+
+from datetime import datetime, timezone
+from itertools import product
+
+import pytest
+from shared.fsma_validation import is_location_description_complete
+from shared.schemas import FSMALocation, Location
+
+from services.ingestion.app.webhook_models import VALID_UNITS_OF_MEASURE, IngestEvent, WebhookCTEType
+
+VALUES = (None, "", " ", "x")
+
+
+@pytest.mark.parametrize("name,phone,address,gln", product(VALUES, repeat=4))
+def test_completeness_presence_permutations(name, phone, address, gln):
+    expected = all(value == "x" for value in (name, phone, address))
+    assert is_location_description_complete(name=name, phone=phone, address=address, gln=gln) is expected
+
+
+def event_kwargs():
+    return dict(
+        cte_type=next(iter(WebhookCTEType)),
+        traceability_lot_code="x" * 3,
+        product_description="x",
+        quantity=float(True),
+        unit_of_measure=sorted(VALID_UNITS_OF_MEASURE)[0],
+        timestamp=datetime.now(timezone.utc).isoformat(),
+    )
+
+
+@pytest.mark.parametrize("name,phone,address", product(VALUES, repeat=3))
+@pytest.mark.parametrize("prefix", ("top_level", "location", "ship_from", "ship_to", "receiving"))
+def test_webhook_description_presence(name, phone, address, prefix):
+    kdes = {f"{prefix}_{field}": value for field, value in zip(("name", "phone", "address"), (name, phone, address))}
+    kwargs = event_kwargs() | {"kdes": kdes, "location_gln": " "}
+    if prefix == "top_level":
+        kwargs.update(
+            {f"location_{field}": value for field, value in zip(("name", "phone", "address"), (name, phone, address))}
+        )
+        kwargs["kdes"] = {}
+    if all(value == "x" for value in (name, phone, address)):
+        IngestEvent(**kwargs)
+    else:
+        with pytest.raises(ValueError, match=r"21 CFR 1\.1310"):
+            IngestEvent(**kwargs)
+
+
+@pytest.mark.parametrize("key", ("location_gln", "ship_from_gln", "ship_to_gln", "receiving_gln"))
+@pytest.mark.parametrize("value", VALUES)
+def test_webhook_kde_gln_presence(key, value):
+    kwargs = event_kwargs() | {"kdes": {key: value}}
+    with pytest.raises(ValueError, match=r"21 CFR 1\.1310"):
+        IngestEvent(**kwargs)
+
+
+def test_webhook_top_level_and_kde_description():
+    IngestEvent(**(event_kwargs() | {"location_name": "x", "location_phone": "x", "kdes": {"location_address": "x"}}))
+
+
+@pytest.mark.parametrize("prefix", ("ship_from", "ship_to", "receiving"))
+def test_legacy_location_name_requires_matching_phone_and_address(prefix):
+    kwargs = event_kwargs() | {"kdes": {f"{prefix}_location": "x"}}
+    with pytest.raises(ValueError, match=r"21 CFR 1\.1310"):
+        IngestEvent(**kwargs)
+    kwargs["kdes"].update({f"{prefix}_phone": "x", f"{prefix}_address": "x"})
+    IngestEvent(**kwargs)
+
+
+def test_cannot_combine_different_locations():
+    with pytest.raises(ValueError, match=r"21 CFR 1\.1310"):
+        IngestEvent(
+            **(event_kwargs() | {"kdes": {"ship_from_name": "x", "ship_to_phone": "x", "ship_to_address": "x"}})
+        )
+
+
+@pytest.mark.parametrize("model", (Location, FSMALocation))
+def test_location_schema_phone_is_optional(model):
+    assert not model.model_fields["phone"].is_required()
+    assert model(phone="x").phone == "x"
+
+
+def test_canonical_preserves_top_level_location_description():
+    from unittest.mock import patch
+
+    from shared.canonical_event import normalize_webhook_event
+
+    event = IngestEvent(
+        **(
+            event_kwargs()
+            | {
+                "cte_type": WebhookCTEType.SHIPPING,
+                "location_name": "x",
+                "location_phone": "x",
+                "location_address": "x",
+            }
+        )
+    )
+    # Capture normalization without creating tenant/event IDs or persisted records.
+    with patch("shared.canonical_event.TraceabilityEvent") as model:
+        normalize_webhook_event(event, source="x", tenant_id=None)
+    canonical = model.call_args.kwargs
+    for field in ("location_phone", "location_address"):
+        assert canonical["raw_payload"][field] == getattr(event, field)
+        assert canonical["kdes"][field] == getattr(event, field)
+    assert event.kdes == {}
+
+
+def test_canonical_location_description_preserves_existing_kdes():
+    from unittest.mock import patch
+
+    from shared.canonical_event import normalize_webhook_event
+
+    event = IngestEvent(
+        **(
+            event_kwargs()
+            | {
+                "cte_type": WebhookCTEType.SHIPPING,
+                "location_name": "x",
+                "location_phone": "x",
+                "location_address": "x",
+                "kdes": {"location_phone": " ", "location_address": " "},
+            }
+        )
+    )
+    # Capture normalization without creating tenant/event IDs or persisted records.
+    with patch("shared.canonical_event.TraceabilityEvent") as model:
+        normalize_webhook_event(event, source="x", tenant_id=None)
+    canonical = model.call_args.kwargs
+    assert canonical["kdes"]["location_phone"] == " "
+    assert canonical["kdes"]["location_address"] == " "