fix(ingestion): gap report skipped-row reasons and rule labels (review only)

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})"

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)

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:

services/ingestion/tests/test_csv_gap_report.py

@@ -0,0 +1,239 @@
+"""Synthetic CSV gap-report acceptance guards; no partner data."""
+
+import csv
+import io
+from types import SimpleNamespace
+
+import pytest
+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_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)
+    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
+
+
+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)

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
 
 

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")