From 7539d189c0fde34eee98e3bbf05b538d33584621 Mon Sep 17 00:00:00 2001
From: CS <cs@Mac.lan>
Date: Fri, 6 Nov 2026 02:31:19 -0800
Subject: [PATCH 1/4] fix(ingestion): gap report row-level reasons, GLN
 check-digit errors, broken-link warnings, rule source labels

---
 services/ingestion/app/csv_templates.py       | 220 +++++++++++++-----
 services/ingestion/app/gap_reasons.py         |  55 +++++
 .../app/supplier_readiness/shipping.py        |   7 +-
 .../ingestion/tests/test_csv_gap_report.py    | 184 +++++++++++++++
 .../ingestion/tests/test_csv_jev_file_cte.py  |   5 +-
 .../tests/test_csv_templates_1342.py          |   6 +-
 .../ingestion/tests/test_csv_xlsx_upload.py   |   2 +-
 .../tests/test_fda_export_open_gaps_pg.py     |  42 ++++
 8 files changed, 453 insertions(+), 68 deletions(-)
 create mode 100644 services/ingestion/app/gap_reasons.py
 create mode 100644 services/ingestion/tests/test_csv_gap_report.py

diff --git a/services/ingestion/app/csv_templates.py b/services/ingestion/app/csv_templates.py
index d55d14077..5dadd56c7 100644
--- a/services/ingestion/app/csv_templates.py
+++ b/services/ingestion/app/csv_templates.py
@@ -26,6 +26,7 @@
 from fastapi import APIRouter, Depends, File, Form, Header, HTTPException, UploadFile
 from fastapi.concurrency import run_in_threadpool
 from fastapi.responses import StreamingResponse
+from pydantic import ValidationError
 from shared.agent_task_bridge import enqueue_agent_artifact_task
 from shared.database import SessionLocal
 from shared.jev import JevDecider, get_jev_decider
@@ -51,6 +52,7 @@
 
 from .authz import IngestionPrincipal, require_permission
 from .config import get_settings
+from .gap_reasons import missing_field_reason
 from .shared.csv_safety import sanitize_cell
 from .shared.tenant_resolution import resolve_principal_tenant_id
 from .shared.upload_limits import MAX_CSV_FILE_SIZE_BYTES, read_upload_with_limit
@@ -66,6 +68,7 @@
     IngestEvent,
     WebhookCTEType,
     WebhookPayload,
+    validate_gln,
 )
 
 logger = logging.getLogger("csv-templates")
@@ -681,12 +684,9 @@ def _normalize_location(raw: str) -> str:
 # ── Flexible date parsing ────────────────────────────────────────────────────
 
 
-
 # ── SEL-549 clearer reject reasons (delimiter / TLC / ambiguous date) ────────
 
-_AMBIGUOUS_NUMERIC_DATE = re.compile(
-    r"^\s*(\d{1,2})([/-])(\d{1,2})\2(\d{2}|\d{4})\s*$"
-)
+_AMBIGUOUS_NUMERIC_DATE = re.compile(r"^\s*(\d{1,2})([/-])(\d{1,2})\2(\d{2}|\d{4})\s*$")
 
 
 def _is_ambiguous_numeric_date(raw: str) -> bool:
@@ -733,6 +733,7 @@ def _sniff_non_comma_delimiter(text: str) -> Optional[str]:
 def _delimiter_display(sep: str) -> str:
     return {";": "semicolon (;)", "\t": "tab", "|": "pipe (|)"}.get(sep, repr(sep))
 
+
 def _parse_date_flexible(raw: str, event_time: str = "") -> tuple[str | None, str | None]:
     """Parse a date string in *any* common format into an ISO 8601 UTC timestamp.
 
@@ -1071,26 +1072,26 @@ class _PhysicalDataLineLayout(NamedTuple):
 
 def _physical_data_line_layout(text: str) -> _PhysicalDataLineLayout:
     """Map DictReader data records to physical lines; list blank lines it omits."""
-    lines = text.splitlines()
-    if not lines:
-        return _PhysicalDataLineLayout(0, (), ())
+    reader = csv.reader(io.StringIO(text), strict=True)
     header_line = 0
     record_lines: list[int] = []
     blank_lines: list[int] = []
-    for index, line in enumerate(lines):
-        physical = index + 1
-        if not line.strip():
-            # Leading blanks before the header are also omitted by DictReader,
-            # but partner files put the header on line 1; only blanks after the
-            # header are recovered here so disposition row numbers stay aligned
-            # with physical data lines (SEL-549 first-file dry run).
-            if header_line != 0:
-                blank_lines.append(physical)
-            continue
-        if header_line == 0:
-            header_line = physical
-            continue
-        record_lines.append(physical)
+    try:
+        while True:
+            start = reader.line_num + 1
+            record = next(reader)
+            if not record:
+                blank_lines.append(start)
+            elif not header_line:
+                header_line = start
+            else:
+                record_lines.append(start)
+    except StopIteration:
+        pass
+    except csv.Error:
+        # Preserve the start of the unreadable record, even if its quoted
+        # field consumes several more physical lines before the error.
+        record_lines.append(start)
     return _PhysicalDataLineLayout(header_line, tuple(record_lines), tuple(blank_lines))
 
 
@@ -1106,8 +1107,7 @@ def _log_blank_physical_lines(
             disposition="skipped",
             reason="blank_physical_line",
             detail=(
-                f"Row {physical}: completely blank physical line was omitted by "
-                "the CSV reader and was not ingested"
+                f"Row {physical}: completely blank physical line was omitted by the CSV reader and was not ingested"
             ),
         )
 
@@ -1210,6 +1210,29 @@ def finalize_row_dispositions_from_ingest(
         )
 
 
+def warn_broken_shipment_links(
+    events: Sequence[IngestEvent],
+    source_rows: Sequence[int],
+    rejected_shipments: Mapping[str, Sequence[int]],
+    row_log: Optional[list[CsvRowDispositionEntry]],
+) -> list[str]:
+    """Same-lot candidates are advisory; they never invalidate receiving evidence."""
+    warnings = []
+    flagged = {entry.row_number for entry in (row_log or []) if entry.reason == "broken_shipment_link"}
+    for event, number in zip(events, source_rows):
+        matches = rejected_shipments.get(event.traceability_lot_code, [])
+        if event.cte_type == WebhookCTEType.RECEIVING and matches and number not in flagged:
+            detail = (
+                f"Row {number}: broken shipment link: shipping row(s) {', '.join(map(str, matches))} "
+                "for the same lot were rejected; verify the shipment reference"
+            )
+            warnings.append(detail)
+            _append_row_log(
+                row_log, row_number=number, disposition="flagged", reason="broken_shipment_link", detail=detail
+            )
+    return warnings
+
+
 def parse_csv_rows(
     text: str,
     fallback_cte: Optional[str] = None,
@@ -1260,7 +1283,10 @@ def parse_csv_rows(
 
     Returns ``(events, parse_errors, kde_warnings)``.
     """
+    if row_log is None:
+        row_log = []  # Retain rejection associations even for parse-only callers.
     events: list[IngestEvent] = []
+    parsed_source_rows: list[int] = []
     parse_errors: list[str] = []
     kde_warnings: list[str] = []
 
@@ -1282,7 +1308,7 @@ def parse_csv_rows(
         )
         return ([], [detail], [])
 
-    reader = csv.DictReader(io.StringIO(text))
+    reader = csv.DictReader(io.StringIO(text), strict=True)
     line_layout = _physical_data_line_layout(text)
 
     # Read the header and every record the loop below can reach before
@@ -1337,19 +1363,39 @@ def parse_csv_rows(
                     overflow_exc,
                 )
     except csv.Error as exc:
-        line = reader.reader.line_num
-        logger.warning("csv_unreadable line=%s error=%s", line, exc)
-        detail = (
-            f"Line {line}: CSV could not be read. Check for carriage-return-only line endings, "
-            f"an unclosed quote, or a field longer than {csv.field_size_limit()} characters"
-        )
-        _append_row_log(
-            row_log,
-            row_number=line or 0,
-            disposition="rejected",
-            reason="unreadable_csv",
-            detail=detail,
+        line = (
+            line_layout.record_lines[len(records)]
+            if len(records) < len(line_layout.record_lines)
+            else reader.reader.line_num or 1
         )
+        logger.warning("csv_unreadable line=%s error_type=%s", line, type(exc).__name__)
+        # csv.Error messages are interpreter-controlled; never echo their text
+        # (some include the unexpected delimiter character).
+        error = str(exc)
+        if "field larger than field limit" in error:
+            why = f"field exceeds the CSV limit of {csv.field_size_limit()} characters"
+        elif "unexpected end of data" in error:
+            why = "quoted field is not closed before the end of the file"
+        elif "new-line character seen in unquoted field" in error:
+            why = "unsupported carriage-return-only line endings; export with LF or CRLF"
+        else:
+            why = "invalid CSV quoting; close quoted fields and separate them with commas"
+        detail = f"Line {line}: CSV could not be read: {why}"
+        _append_row_log(row_log, row_number=line, disposition="rejected", reason="unreadable_csv", detail=detail)
+        # This path rejects the file as a whole. Account for the other physical
+        # lines too, without pretending earlier parsed records were ingested.
+        for number, physical in enumerate(text.splitlines(), start=1):
+            if number == line or number == line_layout.header_line:
+                continue
+            _append_row_log(
+                row_log,
+                row_number=number,
+                disposition="skipped",
+                reason="unreadable_file" if physical.strip() else "blank_physical_line",
+                detail=f"Row {number}: not ingested because the CSV file is unreadable at row {line}"
+                if physical.strip()
+                else f"Row {number}: blank physical line was not ingested",
+            )
         return ([], [detail], [])
     if header is not None and not any(name.strip() for name in header):
         detail = "CSV has no header row: the first line must list the column names"
@@ -1377,6 +1423,7 @@ def parse_csv_rows(
     )
     if missing_columns:
         detail = _format_missing_columns_error(missing_columns, header)
+        detail += ". " + "; ".join(missing_field_reason(fallback_cte or "", name) for name in missing_columns)
         _append_row_log(
             row_log,
             row_number=0,
@@ -1391,16 +1438,13 @@ def parse_csv_rows(
                 disposition="rejected",
                 reason="missing_column",
                 field=name,
-                detail=f"Required column {name!r} is absent from the partner file header",
+                detail=missing_field_reason(fallback_cte or "", name)
+                + "; column is absent from the partner file header",
             )
         # Every physical data row still needs an exclusive outcome so
         # accepted + rejected + skipped equals rows sent (SEL-549).
         for idx, record in enumerate(records):
-            row_num = (
-                line_layout.record_lines[idx]
-                if idx < len(line_layout.record_lines)
-                else idx + 2
-            )
+            row_num = line_layout.record_lines[idx] if idx < len(line_layout.record_lines) else idx + 2
             if not _is_data_record(record):
                 values = [str(v or "").strip() for k, v in record.items() if k is not None]
                 if values and values[0].startswith("#"):
@@ -1490,11 +1534,7 @@ def parse_csv_rows(
         )
 
     for idx, row in enumerate(records):
-        row_num = (
-            line_layout.record_lines[idx]
-            if idx < len(line_layout.record_lines)
-            else idx + 2
-        )
+        row_num = line_layout.record_lines[idx] if idx < len(line_layout.record_lines) else idx + 2
         # _file_cte_type_for_jev reads exactly the rows before this cut-off.
         if idx >= MAX_CSV_ROWS:
             # First over-limit row is rejected once. Further unread rows are
@@ -1512,9 +1552,7 @@ def parse_csv_rows(
                 for offset in range(1, skipped_overflow + 1):
                     pos = idx + offset
                     overflow_row = (
-                        line_layout.record_lines[pos]
-                        if pos < len(line_layout.record_lines)
-                        else row_num + offset
+                        line_layout.record_lines[pos] if pos < len(line_layout.record_lines) else row_num + offset
                     )
                     _append_row_log(
                         row_log,
@@ -1613,13 +1651,14 @@ def parse_csv_rows(
             )
             continue
 
-        def _reject(reason: str, message: str) -> None:
+        def _reject(reason: str, message: str, field: Optional[str] = None) -> None:
             parse_errors.append(message)
             _append_row_log(
                 row_log,
                 row_number=row_num,
                 disposition="rejected",
                 reason=reason,
+                field=field,
                 detail=message,
             )
 
@@ -1674,10 +1713,12 @@ def _reject(reason: str, message: str) -> None:
             event_time_raw = (row.get("event_time") or "").strip()
             ts, date_warning = _parse_date_flexible(raw_date, event_time_raw)
             if ts is None:
+                if not raw_date:
+                    date_warning = missing_field_reason(
+                        row_cte, {"shipping": "ship_date", "receiving": "receive_date"}.get(row_cte, "date")
+                    )
                 date_reason = (
-                    "ambiguous_date"
-                    if date_warning and date_warning.startswith("AMBIGUOUS_DATE:")
-                    else "bad_date"
+                    "ambiguous_date" if date_warning and date_warning.startswith("AMBIGUOUS_DATE:") else "bad_date"
                 )
                 _reject(date_reason, f"Row {row_num}: {date_warning}")
                 continue
@@ -1705,11 +1746,19 @@ def _reject(reason: str, message: str) -> None:
                     detail="Location name was normalised (abbreviations expanded)",
                 )
 
+            if loc_gln:
+                valid_gln, gln_error = validate_gln(loc_gln)
+                if not valid_gln:
+                    _reject("invalid_gln", f"Row {row_num}: location_gln: {gln_error}", "location_gln")
+                    continue
+
             # Unit of measure: map abbreviations before Pydantic sees them.
             # No default: a unit the uploader did not record is not "units".
             uom_raw = _first_present(row, _UNIT_COLUMNS)
             if not uom_raw:
-                _reject("missing_unit_of_measure", f"Row {row_num}: unit_of_measure is required")
+                _reject(
+                    "missing_unit_of_measure", f"Row {row_num}: " + missing_field_reason(row_cte, "unit_of_measure")
+                )
                 continue
             uom = _normalize_uom(uom_raw)
             if uom != uom_raw:
@@ -1727,7 +1776,7 @@ def _reject(reason: str, message: str) -> None:
             # below would reject the row too, but with an opaque message.
             quantity_raw = _first_present(row, _QUANTITY_COLUMNS)
             if not quantity_raw:
-                _reject("missing_quantity", f"Row {row_num}: quantity is required")
+                _reject("missing_quantity", f"Row {row_num}: " + missing_field_reason(row_cte, "quantity"))
                 continue
             try:
                 quantity = float(quantity_raw)
@@ -1756,8 +1805,7 @@ def _reject(reason: str, message: str) -> None:
             if len((lot_code or "").strip()) < 3:
                 _reject(
                     "bad_tlc",
-                    f"Row {row_num}: traceability_lot_code must be at least 3 characters "
-                    f"(got {lot_code!r})",
+                    f"Row {row_num}: traceability_lot_code must be at least 3 characters (got {lot_code!r})",
                 )
                 continue
             for integrity_warn in _check_lot_code_integrity(lot_code, row_log=row_log, row_number=row_num):
@@ -1823,28 +1871,73 @@ def _reject(reason: str, message: str) -> None:
                 kdes=kdes,
             )
             events.append(event)
+            parsed_source_rows.append(row_num)
 
             # KDE completeness validation (warn, don't reject)
             missing_kdes = _validate_kde_completeness(row_cte, event, kdes)
             if missing_kdes:
-                kde_warnings.append(f"Row {row_num}: Missing KDEs for {row_cte}: {', '.join(missing_kdes)}")
+                kde_warnings.extend(
+                    f"Row {row_num}: Missing KDEs: {missing_field_reason(row_cte, field)}" for field in missing_kdes
+                )
                 # Do not log a flagged "Accepted with missing KDEs" here: ingest
                 # may still reject the row, and a row must have exactly one
                 # final exclusive disposition. Carry the observation as a note
                 # for finalize_row_dispositions_from_ingest to attach.
                 if row_notes is not None:
-                    row_notes.setdefault(row_num, []).append(f"missing KDEs: {', '.join(missing_kdes)}")
+                    row_notes.setdefault(row_num, []).extend(
+                        "missing KDEs: " + missing_field_reason(row_cte, field) for field in missing_kdes
+                    )
             # Do not log accepted/ingested here. Dedup, KDE checks and
             # persistence decide the final disposition in ingest_events;
             # finalize_row_dispositions_from_ingest records it once.
             if event_source_rows is not None:
                 event_source_rows.append(row_num)
+        except ValidationError as exc:
+            # Never render ValidationError text: it contains input values and
+            # arbitrary exception messages. Use fixed field/type descriptions.
+            for error in exc.errors(include_input=False, include_context=False):
+                field = str(error["loc"][0]) if error["loc"] else "record"
+                if not error["loc"]:
+                    field = "location_name"
+                    reason = (
+                        missing_field_reason(row_cte, "location_name")
+                        + "; provide a location description or location KDE"
+                    )
+                    _reject("missing_location", f"Row {row_num}: {reason}", field)
+                    continue
+                if field not in IngestEvent.model_fields:
+                    field = "record"
+                kind = error["type"]
+                reasons = {
+                    "string_too_short": "must contain the minimum number of characters",
+                    "greater_than": "must be greater than the minimum allowed value",
+                    "finite_number": "must be a finite number",
+                    "float_parsing": "must be a number",
+                    "enum": "must be one of the supported event types",
+                }
+                reason = reasons.get(kind, "value does not meet the field format or constraints")
+                if kind == "missing" or not row.get(field, "").strip():
+                    reason = missing_field_reason(row_cte, field)
+                _reject("invalid_field", f"Row {row_num}: {field}: {reason}", field)
         except Exception as e:
             # Validation exceptions can include uploaded cell values. Keep
             # only the exception type in logs and a generic per-row response.
             logger.warning("csv_row_parse_failed row=%s error_type=%s", row_num, type(e).__name__)
-            _reject("parse_exception", f"Row {row_num}: could not be parsed (see server logs)")
+            _reject(
+                "parse_exception",
+                f"Row {row_num}: unexpected row validation failure; contact support with this row number",
+            )
 
+    rejected_rows = {entry.row_number for entry in (row_log or []) if entry.disposition == "rejected"}
+    rejected_shipments: dict[str, list[int]] = {}
+    for index, record in enumerate(records):
+        number = line_layout.record_lines[index] if index < len(line_layout.record_lines) else index + 2
+        normalized = {normalize_csv_header(k): (v or "").strip() for k, v in record.items() if k}
+        if number in rejected_rows and _detect_row_cte_type(normalized, fallback_cte) == "shipping":
+            lot = _first_present(normalized, _LOT_CODE_COLUMNS)
+            if lot:
+                rejected_shipments.setdefault(lot, []).append(number)
+    kde_warnings.extend(warn_broken_shipment_links(events, parsed_source_rows, rejected_shipments, row_log))
     return events, parse_errors, kde_warnings
 
 
@@ -2427,6 +2520,11 @@ async def ingest_csv(
     )
     # Final dispositions only: parse never claimed these rows were ingested.
     finalize_row_dispositions_from_ingest(row_log, event_source_rows, response.events, row_notes=row_notes)
+    rejected_shipments: dict[str, list[int]] = {}
+    for event, number, result in zip(events, event_source_rows, response.events):
+        if event.cte_type == WebhookCTEType.SHIPPING and result.status == "rejected":
+            rejected_shipments.setdefault(event.traceability_lot_code, []).append(number)
+    kde_warnings.extend(warn_broken_shipment_links(events, event_source_rows, rejected_shipments, row_log))
 
     artifacts_recorded = 0
     if header_mapping.ok and header_mapping.suggestions:
diff --git a/services/ingestion/app/gap_reasons.py b/services/ingestion/app/gap_reasons.py
new file mode 100644
index 000000000..bfd0daa05
--- /dev/null
+++ b/services/ingestion/app/gap_reasons.py
@@ -0,0 +1,55 @@
+"""Buyer-facing requirement provenance, without retailer deadline claims.
+
+Legal sections verified against eCFR Part 1 Subpart S, 2026-11-06:
+https://www.ecfr.gov/current/title-21/part-1/subpart-S
+The retailer overlay is the existing Albertsons ASN list in
+supplier_readiness/shipping.py; it is never treated as a legal requirement.
+"""
+
+_SECTIONS = {
+    "harvesting": "1.1325(a)",
+    "cooling": "1.1325(b)",
+    "initial_packing": "1.1330",
+    "first_land_based_receiving": "1.1335",
+    "shipping": "1.1340(a)",
+    "receiving": "1.1345(a)",
+    "transformation": "1.1350(a)",
+}
+_SHIPPING_PARAGRAPHS = {
+    "traceability_lot_code": "1",
+    "quantity": "2",
+    "unit_of_measure": "2",
+    "product_description": "3",
+    "ship_to_location": "4",
+    "ship_from_location": "5",
+    "ship_date": "6",
+    "tlc_source_reference": "7",
+    "reference_document": "8",
+}
+_FIELD_LABELS = {
+    "traceability_lot_code": "traceability lot code",
+    "tlc_source_reference": "traceability lot code source reference",
+    "receive_date": "receiving date",
+    "input_lot_codes": "input traceability lot codes",
+    "sscc": "serial shipping container code (SSCC)",
+}
+_BUYER_LABELS = {
+    "gtin": "global trade item number (GTIN)",
+    "po_number": "purchase order number",
+    "sscc": "serial shipping container code (SSCC)",
+    "product_date": "product date",
+}
+
+
+def missing_field_reason(cte: str, field: str, *, retailer: bool = False) -> str:
+    label = _BUYER_LABELS.get(field, _FIELD_LABELS.get(field, field.replace("_", " ")))
+    if retailer:
+        if field not in _BUYER_LABELS:
+            raise ValueError("Field is not in the Albertsons ASN overlay")
+        return f"Missing {label}: asked for by the retailer (Albertsons ASN field list)"
+    section = _SECTIONS.get(cte)
+    if section is None:
+        return f"Missing {label}: required by the application intake schema"
+    if cte == "shipping" and field in _SHIPPING_PARAGRAPHS:
+        section += f"({_SHIPPING_PARAGRAPHS[field]})"
+    return f"Missing {label}: required by law (21 CFR {section})"
diff --git a/services/ingestion/app/supplier_readiness/shipping.py b/services/ingestion/app/supplier_readiness/shipping.py
index aa56969d6..3416581df 100644
--- a/services/ingestion/app/supplier_readiness/shipping.py
+++ b/services/ingestion/app/supplier_readiness/shipping.py
@@ -14,6 +14,7 @@
 from typing import Any, Iterable, Mapping, Optional, Sequence
 
 from ..edi_ingestion.utils import _normalize_unit, _resolve_unit_code
+from ..gap_reasons import missing_field_reason
 from ..webhook_models import REQUIRED_KDES_BY_CTE, VALID_UNITS_OF_MEASURE, WebhookCTEType
 
 # Segments the 856 extractor / envelope path already understands. Anything
@@ -85,6 +86,7 @@ class ShippingReadinessResult:
     buyer: dict[str, Any]
     missing_repo_kdes: list[str] = field(default_factory=list)
     missing_buyer_fields: list[str] = field(default_factory=list)
+    gap_reasons: list[str] = field(default_factory=list)
     failures: list[dict[str, str]] = field(default_factory=list)
     flags: list[FlagRecord] = field(default_factory=list)
     label: str = "SYNTHETIC"
@@ -343,6 +345,8 @@ def score_shipping_readiness(
         buyer=dict(buyer),
         missing_repo_kdes=missing_repo,
         missing_buyer_fields=missing_buyer,
+        gap_reasons=[missing_field_reason("shipping", key) for key in missing_repo]
+        + [missing_field_reason("shipping", key, retailer=True) for key in missing_buyer],
         failures=failures,
         flags=flag_list,
         source=source,
@@ -508,8 +512,7 @@ def dry_run_shipping_csv(
         # Attach header unknown-column flags to row 2 only so they appear on a result
         row_flags = list(header_flags) if idx == 2 else []
         row_flags.extend(
-            FlagRecord(reason=f.reason, field=f.field, detail=f.detail, row_number=f.row_number or idx)
-            for f in flags
+            FlagRecord(reason=f.reason, field=f.field, detail=f.detail, row_number=f.row_number or idx) for f in flags
         )
         for f in row_flags:
             if f.reason == "unrecognized_column" and f.row_number == 1 and idx != 2:
diff --git a/services/ingestion/tests/test_csv_gap_report.py b/services/ingestion/tests/test_csv_gap_report.py
new file mode 100644
index 000000000..33566be78
--- /dev/null
+++ b/services/ingestion/tests/test_csv_gap_report.py
@@ -0,0 +1,184 @@
+"""Synthetic CSV gap-report acceptance guards; no partner data."""
+
+import csv
+import io
+from types import SimpleNamespace
+
+from app import csv_templates as ct
+from app.supplier_readiness.shipping import score_shipping_readiness
+
+
+def _row(**updates):
+    return {
+        "cte_type": "shipping",
+        "traceability_lot_code": "SYNTH-001",
+        "product_description": "Synthetic greens",
+        "quantity": "12",
+        "unit_of_measure": "cases",
+        "location_name": "Synthetic warehouse",
+        "location_gln": "",
+        "ship_date": "2026-06-01",
+        "receive_date": "2026-06-02",
+        "ship_to_location": "Synthetic receiver",
+        "ship_from_location": "Synthetic sender",
+        "receiving_location": "Synthetic receiver",
+        "immediate_previous_source": "Synthetic sender",
+        "reference_document": "SYNTH-DOC-001",
+        "tlc_source_reference": "Synthetic farm",
+        **updates,
+    }
+
+
+def _csv(*rows):
+    out = io.StringIO()
+    writer = csv.DictWriter(out, fieldnames=list(_row()), lineterminator="\n")
+    writer.writeheader()
+    for row in rows:
+        if row is None:
+            out.write("\n")
+        else:
+            writer.writerow(row)
+    return out.getvalue()
+
+
+def _parse(text):
+    log, numbers, notes = [], [], {}
+    events, errors, warnings = ct.parse_csv_rows(text, row_log=log, event_source_rows=numbers, row_notes=notes)
+    ct.finalize_row_dispositions_from_ingest(
+        log, numbers, [SimpleNamespace(status="accepted", errors=[]) for _ in events], row_notes=notes
+    )
+    return events, errors, warnings, log
+
+
+def test_blank_bad_gln_valid_and_receiving_broken_link(monkeypatch):
+    monkeypatch.setenv("STRICT_GLN_VALIDATION", "true")
+    text = _csv(
+        None, _row(location_gln="3456789012341"), _row(cte_type="receiving"), _row(traceability_lot_code="SYNTH-002")
+    )
+    events, errors, warnings, log = _parse(text)
+    assert len(events) == 2
+    bad = next(e for e in log if e.reason == "invalid_gln")
+    assert (bad.row_number, bad.field) == (3, "location_gln")
+    assert "GLN check digit invalid: expected 0, got 1" in bad.detail
+    assert next(e for e in log if e.reason == "blank_physical_line").row_number == 2
+    assert {e.row_number for e in log if e.disposition == "accepted"} == {4, 5}
+    assert any("Row 4: broken shipment link: shipping row(s) 3" in w for w in warnings)
+    assert not any("could not be parsed" in e for e in errors)
+    assert "Synthetic greens" not in str(log) + str(errors)
+
+
+def test_multiline_cell_keeps_physical_rows():
+    text = _csv(
+        _row(product_description="Synthetic\n\ngreens"),
+        None,
+        _row(quantity=""),
+        _row(traceability_lot_code="SYNTH-002"),
+    )
+    events, errors, warnings, log = _parse(text)
+    assert len(events) == 2
+    assert {e.row_number for e in log if e.reason == "blank_physical_line"} == {5}
+    assert any("Row 6:" in e and "required by law (21 CFR 1.1340(a)(2))" in e for e in errors)
+    assert {e.row_number for e in log if e.disposition == "accepted"} == {2, 7}
+
+
+def test_missing_kdes_name_law_and_retailer_separately_without_dates():
+    _, _, warnings, _ = _parse(_csv(_row(ship_to_location="")))
+    assert any("ship to location: required by law (21 CFR 1.1340(a)(4))" in w for w in warnings)
+    result = score_shipping_readiness({})
+    assert any("purchase order number: asked for by the retailer" in line for line in result.gap_reasons)
+    assert any("reference document: required by law" in line for line in result.gap_reasons)
+    assert not any(date in " ".join(result.gap_reasons) for date in ("Jan 1", "January 1", "Dec 31", "December 31"))
+
+
+def test_broken_row_does_not_hide_valid_row():
+    text = _csv(_row(), _row(traceability_lot_code="SYNTH-002"))
+    lines = text.splitlines()
+    lines[1] += ",synthetic-overflow"
+    events, errors, _, log = _parse("\n".join(lines) + "\n")
+    assert len(events) == 1
+    assert any(e.row_number == 2 and e.disposition == "rejected" for e in log)
+    assert any(e.row_number == 3 and e.disposition == "accepted" for e in log)
+    assert "synthetic-overflow" not in str(errors) + str(log)
+
+
+def test_unreadable_quoted_row_accounts_for_other_rows_without_contents():
+    text = _csv(_row()) + '\n"synthetic-unclosed-cell'
+    events, errors, _, log = _parse(text)
+    assert not events
+    assert "quoted field is not closed" in errors[0]
+    assert next(e for e in log if e.reason == "unreadable_csv").row_number == 4
+    assert {e.row_number for e in log if e.disposition == "skipped"} == {2, 3}
+    assert "synthetic-unclosed-cell" not in str(errors) + str(log)
+    assert "Synthetic greens" not in str(errors) + str(log)
+
+
+def test_ingest_rejected_shipment_preserves_receiving_acceptance(monkeypatch):
+    from app.authz import IngestionPrincipal, get_ingestion_principal
+    from app.webhook_models import EventResult, IngestResponse
+    from fastapi import FastAPI
+    from fastapi.testclient import TestClient
+
+    async def ingest(payload, **kwargs):
+        return IngestResponse(
+            accepted=1,
+            rejected=1,
+            total=2,
+            events=[
+                EventResult(
+                    traceability_lot_code=event.traceability_lot_code,
+                    cte_type=event.cte_type.value,
+                    status="rejected" if index == 0 else "accepted",
+                    errors=["rule_violation"] if index == 0 else [],
+                )
+                for index, event in enumerate(payload.events)
+            ],
+        )
+
+    monkeypatch.setattr(ct, "ingest_events", ingest)
+    app = FastAPI()
+    app.include_router(ct.router)
+    app.dependency_overrides[get_ingestion_principal] = lambda: IngestionPrincipal(
+        key_id="synthetic-key-id", scopes=["*"], tenant_id="synthetic-tenant"
+    )
+    with TestClient(app) as client:
+        response = client.post(
+            "/api/v1/ingest/csv",
+            files={"file": ("synthetic.csv", _csv(_row(), _row(cte_type="receiving")).encode(), "text/csv")},
+        )
+    assert response.status_code == 200
+    body = response.json()
+    assert (body["accepted"], body["rejected"]) == (1, 1)
+    assert any("Row 3: broken shipment link: shipping row(s) 2" in w for w in body["kde_warnings"])
+    assert any(e["row_number"] == 3 and e["disposition"] == "accepted" for e in body["row_disposition_log"])
+
+
+def test_model_error_has_specific_missing_location_without_row_contents(caplog):
+    text = _csv(_row(location_name="", ship_to_location="", ship_from_location="", receiving_location=""))
+    events, errors, _, log = _parse(text)
+    assert not events
+    assert any("Missing location name: required by law (21 CFR 1.1340(a))" in error for error in errors)
+    entry = next(e for e in log if e.reason == "missing_location")
+    assert (entry.row_number, entry.field) == (2, "location_name")
+    assert "Synthetic greens" not in str(errors) + str(log) + caplog.text
+
+
+def test_parse_only_callers_also_get_broken_link_warning():
+    events, errors, warnings = ct.parse_csv_rows(_csv(_row(quantity=""), _row(cte_type="receiving")))
+    assert len(events) == 1 and len(errors) == 1
+    assert any("Row 3: broken shipment link: shipping row(s) 2" in warning for warning in warnings)
+
+
+def test_receiving_missing_source_cites_receiving_rule():
+    events, errors, warnings, _ = _parse(_csv(_row(cte_type="receiving", tlc_source_reference="")))
+    assert len(events) == 1 and not errors
+    assert any(
+        "traceability lot code source reference: required by law (21 CFR 1.1345(a))" in w
+        for w in warnings
+    )
+
+
+def test_unreadable_multiline_record_reports_starting_physical_row():
+    events, errors, _, log = _parse(_csv(_row()) + '"synthetic-unclosed\ncontinued')
+    assert not events
+    assert errors[0].startswith("Line 3:")
+    assert next(e for e in log if e.reason == "unreadable_csv").row_number == 3
diff --git a/services/ingestion/tests/test_csv_jev_file_cte.py b/services/ingestion/tests/test_csv_jev_file_cte.py
index 3f4b88d65..df64978f7 100644
--- a/services/ingestion/tests/test_csv_jev_file_cte.py
+++ b/services/ingestion/tests/test_csv_jev_file_cte.py
@@ -460,7 +460,10 @@ def test_a_header_row_that_swallows_rows_is_never_sent(harness):
     resp = _post(b'Lot,"Qty\nTLC-1,5,Farm\n')
 
     assert resp.status_code == 400
-    assert resp.json()["detail"] == "CSV contained no data rows"
+    detail = resp.json()["detail"]
+    assert detail["message"] == "No valid rows found"
+    assert "quoted field is not closed" in detail["errors"][0]
+    assert detail["cte_type_suggestion"]["status"] == "malformed_header"
     assert "bodies" not in seen
 
 
diff --git a/services/ingestion/tests/test_csv_templates_1342.py b/services/ingestion/tests/test_csv_templates_1342.py
index 9bdb04f6b..a6be328b3 100644
--- a/services/ingestion/tests/test_csv_templates_1342.py
+++ b/services/ingestion/tests/test_csv_templates_1342.py
@@ -1113,7 +1113,7 @@ def _parse(self, row: str):
     def test_missing_date_rejects_row(self):
         events, errors, _ = self._parse("TLC1,Kale,10,lbs,,Farm A")
         assert events == []
-        assert errors == ["Row 2: date field was empty"]
+        assert errors == ["Row 2: Missing date: required by law (21 CFR 1.1325(a))"]
 
     def test_unparseable_date_rejects_row(self):
         events, errors, _ = self._parse("TLC1,Kale,10,lbs,2026-99-99,Farm A")
@@ -1124,12 +1124,12 @@ def test_unparseable_date_rejects_row(self):
     def test_missing_unit_rejects_row(self):
         events, errors, _ = self._parse("TLC1,Kale,10,,2026-04-17,Farm A")
         assert events == []
-        assert errors == ["Row 2: unit_of_measure is required"]
+        assert errors == ["Row 2: Missing unit of measure: required by law (21 CFR 1.1325(a))"]
 
     def test_missing_quantity_rejects_row(self):
         events, errors, _ = self._parse("TLC1,Kale,,lbs,2026-04-17,Farm A")
         assert events == []
-        assert errors == ["Row 2: quantity is required"]
+        assert errors == ["Row 2: Missing quantity: required by law (21 CFR 1.1325(a))"]
 
     def test_non_numeric_quantity_rejects_row(self):
         events, errors, _ = self._parse("TLC1,Kale,ten,lbs,2026-04-17,Farm A")
diff --git a/services/ingestion/tests/test_csv_xlsx_upload.py b/services/ingestion/tests/test_csv_xlsx_upload.py
index d336170ce..2c6df8fa4 100644
--- a/services/ingestion/tests/test_csv_xlsx_upload.py
+++ b/services/ingestion/tests/test_csv_xlsx_upload.py
@@ -296,7 +296,7 @@ def test_required_formula_without_cache_is_reported_on_invalid_rows(self, client
         assert resp.status_code == 400
         detail = resp.json()["detail"]
         assert any("Log!D2" in warning and "quantity" in warning for warning in detail["kde_warnings"])
-        assert any("quantity is required" in error for error in detail["errors"])
+        assert any("Missing quantity: required by law (21 CFR 1.1325(a))" in error for error in detail["errors"])
         assert "payload" not in captured
 
     def test_formula_only_sheet_reports_warning_without_data_rows(self, client, captured):
diff --git a/services/ingestion/tests/test_fda_export_open_gaps_pg.py b/services/ingestion/tests/test_fda_export_open_gaps_pg.py
index 79a9df15d..ad2065ff6 100644
--- a/services/ingestion/tests/test_fda_export_open_gaps_pg.py
+++ b/services/ingestion/tests/test_fda_export_open_gaps_pg.py
@@ -97,3 +97,45 @@ def test_gap_state_is_read_as_of_a_moment_for_one_tenant(conn) -> None:
     gaps = fetch_gaps_open_at(conn, tenant_id, ["LOT-1"], as_of)
 
     assert gaps == {("LOT-1", "shipping"): ("missing_kde: closed_since", "missing_kde: open_then")}
+
+
+def test_writer_persists_idempotent_gaps_and_receipt_for_only_its_tenant(conn):
+    """Run the real writer SQL and read back durable state in a scratch DB.
+
+    Outer test transaction is rolled back. PR 3816's deployment grant
+    inventory is separately pinned by test_v142_data_gap_runtime_grants.py.
+    """
+    from shared.data_gap_tracking import Gap, ValidationOutcome, record_outcomes
+    from sqlalchemy.orm import Session
+
+    tenant, other = str(uuid.uuid4()), str(uuid.uuid4())
+    lot = "SYNTH-GAP-PERSIST-001"
+    gap = Gap("missing_kde", "reference_document")
+    session = Session(bind=conn, join_transaction_mode="create_savepoint")
+    try:
+        failed = ValidationOutcome(order=0, lot_code=lot, cte_type="shipping", gaps=(gap,))
+        record_outcomes(session, tenant, [failed, failed])
+        record_outcomes(session, other, [failed])
+        rows = conn.execute(
+            text("SELECT tenant_id::text, closed_at FROM public.data_gap_events WHERE lot_code=:lot"),
+            {"lot": lot},
+        ).fetchall()
+        assert sorted(rows) == sorted([(tenant, None), (other, None)])
+        receipt_id = str(uuid.uuid4())
+        fixed = ValidationOutcome(
+            order=1, lot_code=lot, cte_type="shipping", closing_record_id=receipt_id, closing_record_hash=RECEIPT
+        )
+        record_outcomes(session, tenant, [fixed])
+        rows = conn.execute(
+            text(
+                "SELECT tenant_id::text, closed_at, closing_record_id::text, closing_record_hash "
+                "FROM public.data_gap_events WHERE lot_code=:lot"
+            ),
+            {"lot": lot},
+        ).fetchall()
+        by_tenant = {row[0]: row[1:] for row in rows}
+        assert by_tenant[tenant][0] is not None
+        assert by_tenant[tenant][1:] == (receipt_id, RECEIPT)
+        assert by_tenant[other] == (None, None, None)
+    finally:
+        session.close()

From d3dbfeb70d94addd82f7953c2a873efe2ef364bc Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Fri, 6 Nov 2026 10:52:12 +0000
Subject: [PATCH 2/4] fix(ingestion): preserve CSV record indices in diagnostic
 preview

Co-authored-by: PetrefiedThunder <131207452+PetrefiedThunder@users.noreply.github.com>
---
 services/ingestion/app/pilot_diagnostic.py        | 12 +++++++++++-
 services/ingestion/tests/test_pilot_diagnostic.py |  2 +-
 2 files changed, 12 insertions(+), 2 deletions(-)

diff --git a/services/ingestion/app/pilot_diagnostic.py b/services/ingestion/app/pilot_diagnostic.py
index cc7ec1799..05d29edcf 100644
--- a/services/ingestion/app/pilot_diagnostic.py
+++ b/services/ingestion/app/pilot_diagnostic.py
@@ -35,6 +35,7 @@
         "duplicate_column",
         "extra_fields",
         "header_guess_applied",
+        "invalid_field",
         "location_normalized",
         "lot_code_possible_character_swap",
         "lot_code_no_digits",
@@ -113,6 +114,10 @@ def _parse_preview(text: str, fallback_cte: Optional[str]) -> tuple[dict[str, An
     # Human-readable warnings may contain uploaded values. Their deterministic
     # conditions are represented by row_log and the KDE checks below instead.
     events, errors, _warnings = ct.parse_csv_rows(text, fallback_cte, row_log=row_log, event_source_rows=source_rows)
+    line_layout = ct._physical_data_line_layout(text)
+    record_indices = {line: index + 2 for index, line in enumerate(line_layout.record_lines)}
+    if line_layout.header_line:
+        record_indices[line_layout.header_line] = 1
     grouped: dict[tuple[str, Optional[str], str, bool], set[int]] = {}
 
     def add(reason: str, field: Optional[str], disposition: str, row: int) -> None:
@@ -123,7 +128,12 @@ def add(reason: str, field: Optional[str], disposition: str, row: int) -> None:
 
     for entry in row_log:
         if entry.disposition in {"rejected", "flagged", "skipped"}:
-            add(entry.reason, entry.field, entry.disposition, entry.row_number)
+            add(
+                entry.reason,
+                entry.field,
+                entry.disposition,
+                record_indices.get(entry.row_number, entry.row_number),
+            )
     for row, event in zip(source_rows, events):
         for field in ct._validate_kde_completeness(event.cte_type.value, event, event.kdes):
             add("missing_kde", field, "flagged", row)
diff --git a/services/ingestion/tests/test_pilot_diagnostic.py b/services/ingestion/tests/test_pilot_diagnostic.py
index f67601061..0f3128d72 100644
--- a/services/ingestion/tests/test_pilot_diagnostic.py
+++ b/services/ingestion/tests/test_pilot_diagnostic.py
@@ -472,7 +472,7 @@ def test_validation_exception_does_not_log_raw_values(client, caplog):
     caplog.set_level(logging.INFO)
     response = post(client, csv_bytes(header, rows))
     assert response.status_code == 200
-    assert any(issue["reason"] == "parse_exception" for issue in response.json()["issues"])
+    assert any(issue["reason"] == "invalid_field" for issue in response.json()["issues"])
     assert private_value not in response.text + caplog.text
 
 

From ea1041363775b5509c7e94ea386322c5ae7a260f Mon Sep 17 00:00:00 2001
From: CS <cs@Mac.lan>
Date: Fri, 6 Nov 2026 19:35:31 -0800
Subject: [PATCH 3/4] fix(ingestion): tolerant CSV parsing, honor
 STRICT_GLN_VALIDATION, mapped-column broken-link scan

---
 services/ingestion/app/csv_templates.py       | 34 +++++++++++++++++--
 .../ingestion/tests/test_csv_gap_report.py    | 30 ++++++++++++++++
 2 files changed, 61 insertions(+), 3 deletions(-)

diff --git a/services/ingestion/app/csv_templates.py b/services/ingestion/app/csv_templates.py
index 5dadd56c7..4a774efdf 100644
--- a/services/ingestion/app/csv_templates.py
+++ b/services/ingestion/app/csv_templates.py
@@ -16,6 +16,7 @@
 import itertools
 import logging
 import math
+import os
 import re
 from collections.abc import Mapping, Sequence
 from datetime import datetime, timezone
@@ -1072,7 +1073,7 @@ class _PhysicalDataLineLayout(NamedTuple):
 
 def _physical_data_line_layout(text: str) -> _PhysicalDataLineLayout:
     """Map DictReader data records to physical lines; list blank lines it omits."""
-    reader = csv.reader(io.StringIO(text), strict=True)
+    reader = csv.reader(io.StringIO(text))
     header_line = 0
     record_lines: list[int] = []
     blank_lines: list[int] = []
@@ -1308,7 +1309,7 @@ def parse_csv_rows(
         )
         return ([], [detail], [])
 
-    reader = csv.DictReader(io.StringIO(text), strict=True)
+    reader = csv.DictReader(io.StringIO(text))
     line_layout = _physical_data_line_layout(text)
 
     # Read the header and every record the loop below can reach before
@@ -1316,11 +1317,17 @@ def parse_csv_rows(
     # endings, a field over csv.field_size_limit(), usually from an unclosed
     # quote) is rejected whole with its line named, never partially ingested.
     records: list[dict] = []
+    invalid_quote_rows: set[int] = set()
+    physical_lines = text.splitlines(keepends=True)
     skipped_overflow = 0
     overflow_tail_unreadable = False
     header = None
     try:
         header = reader.fieldnames
+        # Headers must not swallow data rows through an unterminated quote.
+        # Strictness here is confined to the header, never the whole file.
+        if header is not None:
+            list(csv.reader(io.StringIO("".join(physical_lines[: reader.line_num])), strict=True))
         if header and any(name.strip() for name in header):
             seen_columns: dict[str, int] = {}
             for position, name in enumerate(header, start=1):
@@ -1341,6 +1348,17 @@ def parse_csv_rows(
                     return ([], [detail], [])
                 seen_columns[normalized] = position
         for record in reader:
+            index = len(records)
+            start = line_layout.record_lines[index] if index < len(line_layout.record_lines) else reader.line_num
+            # The lenient reader retains recoverable record boundaries. Check
+            # each record separately so bad quoting cannot discard its neighbours.
+            raw_record = "".join(physical_lines[start - 1 : reader.line_num])
+            try:
+                list(csv.reader(io.StringIO(raw_record), strict=True))
+            except csv.Error as exc:
+                if "unexpected end of data" in str(exc):
+                    raise  # Boundaries are ambiguous: retain whole-file handling.
+                invalid_quote_rows.add(start)
             records.append(record)
             if len(records) > MAX_CSV_ROWS:
                 break
@@ -1593,6 +1611,11 @@ def parse_csv_rows(
                 detail=limit_detail,
             )
             break
+        if row_num in invalid_quote_rows:
+            detail = f"Row {row_num}: invalid CSV quoting; close quoted fields and separate them with commas"
+            parse_errors.append(detail)
+            _append_row_log(row_log, row_number=row_num, disposition="rejected", reason="unreadable_csv", detail=detail)
+            continue
         # DictReader files values past the last header column under the None
         # restkey. A non-blank one rejects the row below rather than being
         # dropped; blank ones (a trailing comma) carry nothing and are ignored.
@@ -1746,7 +1769,7 @@ def _reject(reason: str, message: str, field: Optional[str] = None) -> None:
                     detail="Location name was normalised (abbreviations expanded)",
                 )
 
-            if loc_gln:
+            if loc_gln and os.getenv("STRICT_GLN_VALIDATION", "true").lower() == "true":
                 valid_gln, gln_error = validate_gln(loc_gln)
                 if not valid_gln:
                     _reject("invalid_gln", f"Row {row_num}: location_gln: {gln_error}", "location_gln")
@@ -1933,6 +1956,11 @@ def _reject(reason: str, message: str, field: Optional[str] = None) -> None:
     for index, record in enumerate(records):
         number = line_layout.record_lines[index] if index < len(line_layout.record_lines) else index + 2
         normalized = {normalize_csv_header(k): (v or "").strip() for k, v in record.items() if k}
+        # Match parse-time mapping precedence when identifying rejected shipments.
+        for source_key, canonical_field in (header_mapping or {}).items():
+            value = normalized.get(source_key)
+            if value and not _row_provides(normalized, canonical_field):
+                normalized[canonical_field] = value
         if number in rejected_rows and _detect_row_cte_type(normalized, fallback_cte) == "shipping":
             lot = _first_present(normalized, _LOT_CODE_COLUMNS)
             if lot:
diff --git a/services/ingestion/tests/test_csv_gap_report.py b/services/ingestion/tests/test_csv_gap_report.py
index 33566be78..88dd443e2 100644
--- a/services/ingestion/tests/test_csv_gap_report.py
+++ b/services/ingestion/tests/test_csv_gap_report.py
@@ -4,6 +4,7 @@
 import io
 from types import SimpleNamespace
 
+import pytest
 from app import csv_templates as ct
 from app.supplier_readiness.shipping import score_shipping_readiness
 
@@ -182,3 +183,32 @@ def test_unreadable_multiline_record_reports_starting_physical_row():
     assert not events
     assert errors[0].startswith("Line 3:")
     assert next(e for e in log if e.reason == "unreadable_csv").row_number == 3
+
+
+def test_stray_quote_rejects_only_its_physical_row():
+    text = _csv(_row(product_description="Synthetic valid"), None, _row(), _row(traceability_lot_code="SYNTH-002"))
+    text = text.replace("Synthetic greens", '"Synthetic" greens', 1)
+    events, errors, _, log = _parse(text)
+    assert [event.traceability_lot_code for event in events] == ["SYNTH-001", "SYNTH-002"]
+    assert len(errors) == 1 and errors[0].startswith("Row 4:")
+    assert {e.row_number for e in log if e.disposition == "rejected"} == {4}
+    assert {e.row_number for e in log if e.disposition == "accepted"} == {2, 5}
+    assert "Synthetic" not in str(errors) + str(log)
+
+
+def test_csv_gln_validation_can_be_disabled(monkeypatch):
+    monkeypatch.setenv("STRICT_GLN_VALIDATION", "false")
+    events, errors, _, log = _parse(_csv(_row(location_gln="3456789012341")))
+    assert len(events) == 1 and not errors
+    assert not any(e.reason == "invalid_gln" for e in log)
+
+
+@pytest.mark.parametrize("fields", [("cte_type",), ("traceability_lot_code",), ("cte_type", "traceability_lot_code")])
+def test_broken_link_uses_mapped_lot_and_cte_columns(fields):
+    text = _csv(_row(quantity=""), _row(cte_type="receiving"))
+    mapping = {f"partner_{field}": field for field in fields}
+    for source, canonical in mapping.items():
+        text = text.replace(canonical, source, 1)
+    events, errors, warnings = ct.parse_csv_rows(text, header_mapping=mapping)
+    assert len(events) == 1 and len(errors) == 1
+    assert any("Row 3: broken shipment link: shipping row(s) 2" in w for w in warnings)

From 08a14b596f4cd8e113bb4ac449dbd1f376bd0441 Mon Sep 17 00:00:00 2001
From: CS <cs@Mac.lan>
Date: Fri, 6 Nov 2026 20:08:13 -0800
Subject: [PATCH 4/4] Fix CSV physical_lines split to match reader (\n only)
 for PR #3834

Use io.StringIO(text).readlines() instead of text.splitlines() so quoted
cells containing \r/\x0b (and other non-\n line breaks) are not treated as
extra physical rows. Aligns row re-checks and unreadable-file numbering with
the CSV reader. Adds regression tests for acceptance and source row numbers.
---
 services/ingestion/app/csv_templates.py       |  4 +--
 .../ingestion/tests/test_csv_gap_report.py    | 25 +++++++++++++++++++
 2 files changed, 27 insertions(+), 2 deletions(-)

diff --git a/services/ingestion/app/csv_templates.py b/services/ingestion/app/csv_templates.py
index 4a774efdf..ae17af0e4 100644
--- a/services/ingestion/app/csv_templates.py
+++ b/services/ingestion/app/csv_templates.py
@@ -1318,7 +1318,7 @@ def parse_csv_rows(
     # quote) is rejected whole with its line named, never partially ingested.
     records: list[dict] = []
     invalid_quote_rows: set[int] = set()
-    physical_lines = text.splitlines(keepends=True)
+    physical_lines = io.StringIO(text).readlines()
     skipped_overflow = 0
     overflow_tail_unreadable = False
     header = None
@@ -1402,7 +1402,7 @@ def parse_csv_rows(
         _append_row_log(row_log, row_number=line, disposition="rejected", reason="unreadable_csv", detail=detail)
         # This path rejects the file as a whole. Account for the other physical
         # lines too, without pretending earlier parsed records were ingested.
-        for number, physical in enumerate(text.splitlines(), start=1):
+        for number, physical in enumerate(io.StringIO(text).readlines(), start=1):
             if number == line or number == line_layout.header_line:
                 continue
             _append_row_log(
diff --git a/services/ingestion/tests/test_csv_gap_report.py b/services/ingestion/tests/test_csv_gap_report.py
index 88dd443e2..3da6ec09f 100644
--- a/services/ingestion/tests/test_csv_gap_report.py
+++ b/services/ingestion/tests/test_csv_gap_report.py
@@ -82,6 +82,31 @@ def test_multiline_cell_keeps_physical_rows():
     assert {e.row_number for e in log if e.disposition == "accepted"} == {2, 7}
 
 
+def test_quoted_control_characters_keep_all_rows_and_source_numbers():
+    text = _csv(
+        _row(product_description="Synthetic x\ry, greens"),
+        _row(product_description="Synthetic p\nq, greens"),
+        _row(product_description="Synthetic x\x0by, greens"),
+        _row(traceability_lot_code="SYNTH-002"),
+    )
+    events, errors, _, log = _parse(text)
+    assert len(events) == 4
+    assert not errors
+    assert [(e.row_number, e.disposition) for e in log] == [
+        (2, "accepted"), (3, "accepted"), (5, "accepted"), (6, "accepted")
+    ]
+
+
+def test_unreadable_file_control_characters_do_not_add_physical_rows():
+    text = _csv(_row(product_description="Synthetic x\ry\x0bz, greens"))
+    events, errors, _, log = _parse(text + '"synthetic-unclosed-cell')
+    assert not events
+    assert errors[0].startswith("Line 3:")
+    assert [(e.row_number, e.reason) for e in log] == [
+        (3, "unreadable_csv"), (2, "unreadable_file")
+    ]
+
+
 def test_missing_kdes_name_law_and_retailer_separately_without_dates():
     _, _, warnings, _ = _parse(_csv(_row(ship_to_location="")))
     assert any("ship to location: required by law (21 CFR 1.1340(a)(4))" in w for w in warnings)