fix(ingestion): align FDA CSV export KDE columns (C3)

services/ingestion/app/epcis_export.py

@@ -8,6 +8,8 @@
 
 from __future__ import annotations
 
+import csv
+import io
 import json
 import logging
 import re
@@ -31,7 +33,8 @@
 FDA_CSV_HEADER = (
     "CTE_Type,Traceability_Lot_Code,Product_Description,Quantity,Unit_of_Measure,"
     "Event_Date,Event_Time,Ship_From_Location,Ship_From_GLN,Ship_To_Location,"
-    "Ship_To_GLN,Carrier,Temperature_C,SHA256_Hash"
+    "Ship_To_GLN,Carrier,Temperature_C,Reference_Document,TLC_Source_Reference,"
+    "Immediate_Previous_Source,SHA256_Hash,Additional_KDEs_JSON"
 )
 
 
@@ -715,38 +718,53 @@ async def export_fda(
     rows = _load_export_rows(tenant_id, request)
     data_source = "tenant" if rows else "empty"
 
-    if rows:
-        # Build CSV from real data
-        lines = [FDA_CSV_HEADER]
-        for r in rows:
-            ts = r["event_timestamp"]
-            date_str = ts.strftime("%Y-%m-%d") if ts else ""
-            time_str = ts.strftime("%H:%M:%SZ") if ts else ""
-            lines.append(
-                ",".join(
-                    [
-                        str(r["event_type"] or "").upper(),
-                        str(r["traceability_lot_code"] or ""),
-                        sanitize_cell(str(r["product_description"] or "")).replace(",", ";"),
-                        str(r["quantity"] or ""),
-                        str(r["unit_of_measure"] or ""),
-                        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 ""),
-                        sanitize_cell(str(r["ship_to"] or "")).replace(",", ";"),
-                        str(r["ship_to_gln"] or ""),
-                        sanitize_cell(str(r["carrier"] or "")).replace(",", ";"),
-                        str(r["temperature_c"] or ""),
-                        str(r["sha256_hash"] or ""),
-                    ]
-                )
+    # Use csv.writer so fields with commas (esp. Additional_KDEs_JSON) stay
+    # valid JSON / text; the writer quotes cells as needed (Hermes #3722).
+    buf = io.StringIO()
+    writer = csv.writer(buf, lineterminator="\n")
+    writer.writerow(FDA_CSV_HEADER.split(","))
+    for r in rows:
+        ts = r["event_timestamp"]
+        date_str = ts.strftime("%Y-%m-%d") if ts else ""
+        time_str = ts.strftime("%H:%M:%SZ") if ts else ""
+        additional = {
+            key: r.get(key)
+            for key in (
+                "field_id",
+                "harvester_name",
+                "harvester_business_name",
+                "harvester_business_phone",
             )
-        csv_content = "\n".join(lines) + "\n"
-        event_count = len(rows)
-    else:
-        csv_content = FDA_CSV_HEADER + "\n"
-        event_count = 0
+            if r.get(key)
+        }
+        # Emit valid JSON unchanged — do not replace commas; csv.writer quotes.
+        additional_json = (
+            sanitize_cell(json.dumps(additional, sort_keys=True)) if additional else ""
+        )
+        writer.writerow(
+            [
+                str(r.get("event_type") or "").upper(),
+                str(r.get("traceability_lot_code") or ""),
+                sanitize_cell(str(r.get("product_description") or "")),
+                str(r.get("quantity") or ""),
+                str(r.get("unit_of_measure") or ""),
+                date_str,
+                time_str,
+                sanitize_cell(str(r.get("ship_from") or r.get("location_name") or "")),
+                str(r.get("ship_from_gln") or r.get("location_gln") or ""),
+                sanitize_cell(str(r.get("ship_to") or "")),
+                str(r.get("ship_to_gln") or ""),
+                sanitize_cell(str(r.get("carrier") or "")),
+                str(r.get("temperature_c") or ""),
+                sanitize_cell(str(r.get("reference_document") or "")),
+                sanitize_cell(str(r.get("tlc_source_reference") or "")),
+                sanitize_cell(str(r.get("immediate_previous_source") or "")),
+                str(r.get("sha256_hash") or ""),
+                additional_json,
+            ]
+        )
+    csv_content = buf.getvalue()
+    event_count = len(rows)
 
     # EPIC-L (#1655): route filename through the shared safe_filename
     # builder so tenant_id can't carry CRLF / quotes into the header value.

services/ingestion/tests/test_epcis_export_fda_csv_c3.py

@@ -0,0 +1,187 @@
+"""C3: ingestion FDA CSV aligned with compliance sortable KDE columns.
+
+Synthetic fixtures only. Uses harvest CSV lot strings and asn_856 GLNs.
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+from fastapi import FastAPI, Request
+from fastapi.testclient import TestClient
+
+_ingestion_dir = Path(__file__).resolve().parent.parent
+if str(_ingestion_dir) not in sys.path:
+    sys.path.insert(0, str(_ingestion_dir))
+_services_dir = _ingestion_dir.parent
+if str(_services_dir) not in sys.path:
+    sys.path.insert(0, str(_services_dir))
+
+pytest.importorskip("fastapi")
+
+import app.epcis_export as epcis_export
+from app.epcis_export import FDA_CSV_HEADER, router as epcis_export_router
+from app.webhook_compat import _verify_api_key, require_tenant_context
+from shared.tenant_context import TenantContext
+
+FIXTURES = Path(__file__).resolve().parent / "fixtures"
+HARVEST_CSV = FIXTURES / "01_harvesting.csv"
+ASN_EXPECTED = FIXTURES / "edi_corpus" / "asn_856.expected.json"
+
+REQUIRED_NEW = (
+    "Reference_Document",
+    "TLC_Source_Reference",
+    "Immediate_Previous_Source",
+    "Additional_KDEs_JSON",
+)
+
+
+@pytest.fixture()
+def client() -> TestClient:
+    app = FastAPI()
+    app.include_router(epcis_export_router)
+
+    async def _tenant_ctx(request: Request) -> TenantContext:
+        tenant_id = None
+        if request.method in {"POST", "PUT", "PATCH"}:
+            try:
+                body = await request.json()
+            except Exception:
+                body = {}
+            if isinstance(body, dict):
+                tenant_id = body.get("tenant_id")
+        return TenantContext(
+            tenant_id=tenant_id or str(uuid4()),
+            principal_kind="test",
+            principal_id=uuid4().hex,
+        )
+
+    app.dependency_overrides[_verify_api_key] = lambda: None
+    app.dependency_overrides[require_tenant_context] = _tenant_ctx
+    return TestClient(app)
+
+
+def test_fda_header_includes_compliance_kde_columns() -> None:
+    cols = FDA_CSV_HEADER.split(",")
+    for name in REQUIRED_NEW:
+        assert name in cols
+    # SHA256 still present; new cols sit before it or adjacent
+    assert "SHA256_Hash" in cols
+    assert cols.index("Reference_Document") < cols.index("SHA256_Hash")
+
+
+def test_fda_export_row_fills_reference_and_tlc_source(
+    client: TestClient, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    with HARVEST_CSV.open(newline="") as fh:
+        harvest = next(csv.DictReader(fh))
+    asn = json.loads(ASN_EXPECTED.read_text())
+    lot = harvest["traceability_lot_code"]
+
+    def _rows(*_a, **_k):
+        return [
+            {
+                "event_type": "shipping",
+                "traceability_lot_code": lot,
+                "product_description": asn["line_items"][0]["product_description"],
+                "quantity": asn["line_items"][0]["quantity"],
+                "unit_of_measure": asn["line_items"][0]["unit_of_measure"],
+                "event_timestamp": datetime(2026, 3, 10, 12, 0, tzinfo=timezone.utc),
+                "location_gln": None,
+                "location_name": None,
+                "ship_from": asn["ship_from_name"],
+                "ship_from_gln": asn["ship_from_gln"],
+                "ship_to": asn["ship_to_name"],
+                "ship_to_gln": asn["ship_to_gln"],
+                "carrier": "CarrierCo",
+                "temperature_c": None,
+                "reference_document": f"BM:{asn['reference_document_number']}",
+                "tlc_source_reference": "https://id.gs1.org/414/6141410000014",
+                "immediate_previous_source": "",
+                "field_id": harvest.get("field_id"),
+                "harvester_name": harvest.get("harvester_name"),
+                "sha256_hash": "abc",
+            }
+        ]
+
+    monkeypatch.setattr(epcis_export, "_load_export_rows", _rows)
+    response = client.post("/api/v1/export/fda", json={"tenant_id": str(uuid4())})
+    assert response.status_code == 200
+    lines = [ln for ln in response.text.splitlines() if ln]
+    assert lines[0] == FDA_CSV_HEADER
+    reader = csv.DictReader(lines)
+    row = next(reader)
+    assert row["Traceability_Lot_Code"] == lot
+    assert row["Reference_Document"] == "BM:BOL-9001"
+    assert "414/6141410000014" in row["TLC_Source_Reference"]
+    assert "FIELD-SVF-7" in row["Additional_KDEs_JSON"]
+    assert row["Ship_From_GLN"] == "6141410000014"
+
+
+def test_fda_empty_export_header_only(
+    client: TestClient, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(epcis_export, "_load_export_rows", lambda *_a, **_k: [])
+    response = client.post("/api/v1/export/fda", json={"tenant_id": str(uuid4())})
+    assert response.status_code == 200
+    assert [ln for ln in response.text.splitlines() if ln] == [FDA_CSV_HEADER]
+
+def test_fda_additional_kdes_json_csv_round_trip(
+    client: TestClient, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Additional_KDEs_JSON with 2+ keys must stay valid JSON through CSV quoting."""
+    lot = "LOT-ROM-2026-0312A"
+    expected_kdes = {
+        "field_id": "FIELD-SVF-7",
+        "harvester_business_name": "Salinas Valley Farms LLC",
+        "harvester_business_phone": "+1-831-555-0100",
+        "harvester_name": "Salinas Valley Farms LLC",
+    }
+
+    def _rows(*_a, **_k):
+        return [
+            {
+                "event_type": "harvesting",
+                "traceability_lot_code": lot,
+                "product_description": "Organic Romaine Hearts",
+                "quantity": 5000,
+                "unit_of_measure": "lbs",
+                "event_timestamp": datetime(2026, 3, 12, 8, 0, tzinfo=timezone.utc),
+                "location_gln": "0061414100014",
+                "location_name": "Salinas Valley Farms",
+                "ship_from": "Salinas Valley Farms",
+                "ship_from_gln": "0061414100014",
+                "ship_to": "",
+                "ship_to_gln": "",
+                "carrier": "",
+                "temperature_c": None,
+                "reference_document": "HARV-SVF-2026-0312-001",
+                "tlc_source_reference": "",
+                "immediate_previous_source": "",
+                "field_id": expected_kdes["field_id"],
+                "harvester_name": expected_kdes["harvester_name"],
+                "harvester_business_name": expected_kdes["harvester_business_name"],
+                "harvester_business_phone": expected_kdes["harvester_business_phone"],
+                "sha256_hash": "abc",
+            }
+        ]
+
+    monkeypatch.setattr(epcis_export, "_load_export_rows", _rows)
+    response = client.post("/api/v1/export/fda", json={"tenant_id": str(uuid4())})
+    assert response.status_code == 200
+
+    # Write CSV body, read back with the csv module (not naive split).
+    reader = csv.DictReader(response.text.splitlines())
+    row = next(reader)
+    cell = row["Additional_KDEs_JSON"]
+    assert cell.count(",") >= 1  # multi-key JSON keeps commas
+    parsed = json.loads(cell)
+    assert len(parsed) >= 2
+    assert parsed == expected_kdes
+