feat(fda-export): mark rows with an open data gap in the spreadsheet

docs/runbooks/FDA_RECALL_RESPONSE.md

@@ -176,7 +176,7 @@ section included, fails both the browser verifier and `services/admin/scripts/ve
 
 ### 3.3 FDA CSV Column Specification
 
-The export uses 34 columns defined in `FDA_COLUMNS` (`fda_export_service.py:29-65`):
+The export uses 34 columns defined in `FDA_COLUMNS` (`fda_export_service.py`):
 
 **Core fields:** TLC, Product Description, Quantity, UOM, Event Type (CTE), Event Date/Time, Location GLN/Name, Ship From/To, Immediate Previous Source, TLC Source, Source Document
 
@@ -186,6 +186,8 @@ The export uses 34 columns defined in `FDA_COLUMNS` (`fda_export_service.py:29-6
 
 **FSMA 204 required:** System Entry Timestamp (when the record was entered into the system, maps to `ingested_at` column)
 
+**Open Data Gaps:** always the last column. It names the data gaps open on that row's lot and CTE when the export was generated, for example `missing_kde: tlc_source_reference`. Blank means none was open. If the gaps cannot be read, the export stops with a 503.
+
 ### 3.4 Check Response Headers
 
 The export endpoint returns compliance metadata in response headers:

services/ingestion/app/fda_export/queries.py

@@ -11,7 +11,11 @@
 from datetime import date as _date, datetime, time as _time, timedelta, timezone
 from typing import Any, Optional
 
+from fastapi import HTTPException
+from shared.data_gap_tracking import fetch_gaps_open_at
+from shared.tenant_context import set_tenant_guc_if_uuid
 from sqlalchemy import text
+from sqlalchemy.exc import SQLAlchemyError
 
 logger = logging.getLogger("fda-export")
 
@@ -325,6 +329,50 @@ def fetch_recall_events_canonical(db_session, where_clause: str, params: dict, l
     return rows
 
 
+def attach_open_gaps(
+    db_session,
+    tenant_id: str,
+    events: list[dict],
+    *,
+    as_of: Optional[datetime] = None,
+) -> datetime:
+    """Set ``open_data_gaps`` on each event and return the moment it describes.
+
+    A gap belongs to a lot and CTE, so every row for that pair carries it.
+    ``as_of`` defaults to now. Exporters record the returned moment as the
+    audit row's ``generated_at`` and ``/export/verify`` passes that value back,
+    so a gap that closes after the export does not change the regenerated
+    bytes.
+
+    A blank cell states that no gap was open, so a failed lookup stops the
+    export with a 503 rather than emit rows it cannot vouch for. Scoped twice:
+    the ``tenant_id`` predicate here and forced RLS on the table.
+    """
+    as_of = as_of or datetime.now(timezone.utc)
+    lot_codes = {str(event.get("traceability_lot_code") or "") for event in events}
+    try:
+        set_tenant_guc_if_uuid(db_session, tenant_id)
+        gaps = fetch_gaps_open_at(db_session, tenant_id, lot_codes, as_of)
+    except SQLAlchemyError as exc:
+        logger.error(
+            "fda_export_open_gap_lookup_failed",
+            extra={"tenant_id": tenant_id, "error_type": type(exc).__name__},
+            exc_info=True,
+        )
+        raise HTTPException(
+            status_code=503,
+            detail=(
+                "FDA export halted: open data gaps could not be read, so the "
+                "export cannot state which rows have an outstanding item."
+            ),
+        ) from exc
+    for event in events:
+        lot_code = str(event.get("traceability_lot_code") or "").strip()
+        cte_type = str(event.get("event_type") or "").strip().lower()
+        event["open_data_gaps"] = list(gaps.get((lot_code, cte_type), ()))
+    return as_of
+
+
 def _verify_chain_for_export(persistence, db_session, tenant_id: str):
     """Chain verification for export handlers, honoring the WS-1b flag.
 
@@ -491,6 +539,7 @@ def log_recall_export(
     location: Optional[str] = None,
     event_type: Optional[str] = None,
     include_pii: bool = False,
+    generated_at: Optional[datetime] = None,
 ) -> None:
     """Write the recall-export audit log entry.
 
@@ -525,9 +574,10 @@ def log_recall_export(
                 INSERT INTO fsma.fda_export_log
                 (tenant_id, export_type, record_count, export_hash,
                  generated_by, query_tlc, query_start_date, query_end_date,
-                 include_pii)
+                 include_pii, generated_at)
                 VALUES (:tid, :etype, :cnt, :hash, :generated_by,
-                        :tlc, :sd, :ed, :pii)
+                        :tlc, :sd, :ed, :pii,
+                        COALESCE(CAST(:generated_at AS timestamptz), now()))
             """),
             {
                 "tid": tenant_id,
@@ -539,6 +589,7 @@ def log_recall_export(
                 "sd": start_date,
                 "ed": end_date,
                 "pii": bool(include_pii),
+                "generated_at": generated_at,
             },
         )
         db_session.commit()
@@ -769,6 +820,7 @@ def log_v2_export(
     source_ip: Optional[str] = None,
     include_pii: bool = False,
     event_type: Optional[str] = None,
+    generated_at: Optional[datetime] = None,
 ) -> None:
     """Write the v2 export audit log entry.
 
@@ -781,9 +833,9 @@ def log_v2_export(
                 INSERT INTO fsma.fda_export_log
                 (tenant_id, export_type, record_count, export_hash, generated_by,
                  query_tlc, query_start_date, query_end_date,
-                 include_pii, query_event_type)
+                 include_pii, query_event_type, generated_at)
                 VALUES (:tid, :etype, :cnt, :hash, :generated_by, :tlc, :sd, :ed,
-                        :pii, :qetype)
+                        :pii, :qetype, COALESCE(CAST(:generated_at AS timestamptz), now()))
             """),
             {
                 "tid": tenant_id,
@@ -796,6 +848,7 @@ def log_v2_export(
                 "ed": end_date,
                 "pii": bool(include_pii),
                 "qetype": event_type,
+                "generated_at": generated_at,
             },
         )
         db_session.commit()

services/ingestion/app/fda_export/recall.py

@@ -30,6 +30,7 @@
 from .queries import (
     AuditLogWriteError,
     _verify_chain_for_export,
+    attach_open_gaps,
     build_recall_where_clause,
     build_recall_where_clause_canonical,
     export_v1_reads_canonical,
@@ -176,6 +177,7 @@ async def export_recall_filtered_handler(
         # Convert to FDA export format
         events = rows_to_event_dicts(rows)
 
+        generated_at = attach_open_gaps(db_session, tenant_id, events)
         csv_content, export_hash = generate_csv_and_hash(events, include_pii=include_pii)
         chain_verification = _verify_chain_for_export(persistence, db_session, tenant_id)
         completeness_summary = _build_completeness_summary(events)
@@ -212,6 +214,7 @@ async def export_recall_filtered_handler(
             location=location,
             event_type=event_type,
             include_pii=include_pii,
+            generated_at=generated_at,
         )
 
         timestamp = make_timestamp()

services/ingestion/app/fda_export/router.py

@@ -68,7 +68,9 @@
 from .merkle import get_merkle_proof_handler, get_merkle_root_handler
 from .queries import (
     AuditLogWriteError,
+    ExportOverflowError,
     _verify_chain_for_export,
+    attach_open_gaps,
     build_recall_where_clause,
     build_v2_where_clause,
     fetch_export_log_history,
@@ -78,7 +80,6 @@
     format_export_log_rows,
     log_recall_export,
     log_v2_export,
-    ExportOverflowError,
     reconstruct_full_export_events,
     rows_to_event_dicts,
     v2_rows_to_event_dicts,
@@ -305,6 +306,7 @@ def export_fda_spreadsheet(
         # flag flows from the query string → authorization gate → CSV
         # generator; the generated hash covers the redacted (or full)
         # content exactly, so ``verify_export`` can reproduce it.
+        generated_at = attach_open_gaps(db_session, tenant_id, events)
         csv_content = _generate_csv(events, include_pii=include_pii)
         export_hash = hashlib.sha256(csv_content.encode("utf-8")).hexdigest()
         chain_verification = _verify_chain_for_export(persistence, db_session, tenant_id)
@@ -339,6 +341,7 @@ def export_fda_spreadsheet(
                 generated_by=generated_by,
                 export_type="fda_spreadsheet_package" if format == "package" else "fda_spreadsheet",
                 include_pii=include_pii,
+                generated_at=generated_at,
             )
             db_session.commit()
         except Exception as exc:
@@ -611,6 +614,7 @@ def export_all_events(
                 ),
             )
 
+        generated_at = attach_open_gaps(db_session, tenant_id, deduped)
         csv_content = _generate_csv(deduped, include_pii=include_pii)
         export_hash = hashlib.sha256(csv_content.encode("utf-8")).hexdigest()
         chain_verification = _verify_chain_for_export(persistence, db_session, tenant_id)
@@ -639,6 +643,7 @@ def export_all_events(
                 export_type="fda_export_all_package" if format == "package" else "fda_export_all",
                 include_pii=include_pii,
                 query_event_type=event_type,
+                generated_at=generated_at,
             )
             db_session.commit()
         except Exception as exc:
@@ -1061,6 +1066,7 @@ def export_fda_spreadsheet_v2(
         events = v2_rows_to_event_dicts(rows)
 
         # ----- Generate CSV with compliance columns -----
+        generated_at = attach_open_gaps(db_session, tenant_id, events)
         csv_content, export_hash = generate_csv_v2_and_hash(events, include_pii=include_pii)
         chain_verification = _verify_chain_for_export(persistence, db_session, tenant_id)
         completeness_summary = _build_completeness_summary(events)
@@ -1111,6 +1117,7 @@ def export_fda_spreadsheet_v2(
             source_ip=identity["source_ip"],
             include_pii=include_pii,
             event_type=event_type,
+            generated_at=generated_at,
         )
 
         timestamp = make_timestamp()

services/ingestion/app/fda_export/verification.py

@@ -14,11 +14,12 @@
 
 from ..authz import require_permission
 from ..export_models import ExportVerifyResponse
-from ..fda_export_service import _generate_csv, _generate_csv_v2
+from ..fda_export_service import _generate_csv, _generate_csv_before_open_gaps, _generate_csv_v2
 from ..subscription_gate import require_active_subscription
 from .queries import (
     FULL_EXPORT_TYPES,
     ExportOverflowError,
+    attach_open_gaps,
     build_v2_where_clause,
     fetch_v2_events,
     reconstruct_full_export_events,
@@ -140,8 +141,6 @@ async def verify_export_handler(
                     ),
                 )
             events = v2_rows_to_event_dicts(v2_rows)
-            csv_content = _generate_csv_v2(events, include_pii=include_pii)
-            regenerated_hash = hashlib.sha256(csv_content.encode("utf-8")).hexdigest()
         elif reconstruction == "single_tlc":
             events = persistence.query_events_by_tlc(
                 tenant_id=tenant_id,
@@ -172,9 +171,20 @@ async def verify_export_handler(
                     ),
                 ) from exc
 
-        if reconstruction != "v2":
-            csv_content = _generate_csv(events, include_pii=include_pii)
-            regenerated_hash = hashlib.sha256(csv_content.encode("utf-8")).hexdigest()
+        # Gaps as of the export, not now: one that has closed since must not
+        # read as tampering.
+        attach_open_gaps(db_session, tenant_id, events, as_of=generated_at)
+        generate_csv = _generate_csv_v2 if reconstruction == "v2" else _generate_csv
+        csv_content = generate_csv(events, include_pii=include_pii)
+        regenerated_hash = hashlib.sha256(csv_content.encode("utf-8")).hexdigest()
+
+        # An export logged before the Open Data Gaps column existed has the
+        # older layout. Without this, every such export would read as tampered.
+        if regenerated_hash != original_hash:
+            older_layout = _generate_csv_before_open_gaps(events, include_pii=include_pii, v2=reconstruction == "v2")
+            older_hash = hashlib.sha256(older_layout.encode("utf-8")).hexdigest()
+            if older_hash == original_hash:
+                regenerated_hash = older_hash
 
         match = regenerated_hash == original_hash
 

services/ingestion/app/fda_export_service.py

@@ -69,8 +69,17 @@
     "Additional KDEs (JSON)",
     # FSMA 204 §1.1455(c): date/time event entered into system
     "System Entry Timestamp",
+    # Names the gaps open on this row's lot and CTE when the export was
+    # generated, so a blank required KDE reads as a known outstanding item.
+    # Always the last column, in v1 and v2, so no earlier column moves.
+    "Open Data Gaps",
 ]
 
+OPEN_GAPS_COLUMN = FDA_COLUMNS[-1]
+
+# Exports logged before the column existed are verified against this layout.
+FDA_COLUMNS_WITHOUT_OPEN_GAPS = FDA_COLUMNS[:-1]
+
 
 # KDE keys that get their own named columns in the FDA export
 _NAMED_KDE_COLUMNS = {
@@ -162,14 +171,16 @@ def _redact_location_value(value: str, column_name: str, include_pii: bool) -> s
 
 
 # Extended column spec: original FDA columns + compliance + traceability graph columns
-FDA_COLUMNS_V2 = FDA_COLUMNS + [
+FDA_COLUMNS_V2_WITHOUT_OPEN_GAPS = FDA_COLUMNS_WITHOUT_OPEN_GAPS + [
     "Compliance Status",
     "Rule Failures",
     # Transformation traceability (added Phase 2)
     "Trace Relationship",  # "queried" | "linked_via_transformation"
     "Trace Seed TLC",  # The TLC originally queried (useful when this row is a linked lot)
 ]
 
+FDA_COLUMNS_V2 = FDA_COLUMNS_V2_WITHOUT_OPEN_GAPS + [OPEN_GAPS_COLUMN]
+
 
 def _event_to_fda_row(event: dict, *, include_pii: bool = False) -> dict:
     """Convert a persisted CTE event to an FDA spreadsheet row.
@@ -284,6 +295,10 @@ def _event_to_fda_row(event: dict, *, include_pii: bool = False) -> dict:
     # This maps to the ingested_at column which defaults to NOW() on INSERT.
     row["System Entry Timestamp"] = sanitize_cell(event.get("ingested_at", "") or "")
 
+    # Set by ``attach_open_gaps``. Blank means no gap was open at export time,
+    # so an export whose gap lookup failed must not reach this point.
+    row[OPEN_GAPS_COLUMN] = sanitize_cell("; ".join(event.get("open_data_gaps") or ()))
+
     return row
 
 
@@ -697,6 +712,7 @@ def _build_known_gaps(
     chain_payload: dict,
     coverage_gate_bypassed: bool,
     rule_versions: Optional[dict],
+    events_with_open_data_gaps: int = 0,
 ) -> dict:
     """Collect everything this package knows it does NOT prove.
 
@@ -711,6 +727,7 @@ def _build_known_gaps(
     rules = rule_versions or {}
     gaps: dict[str, Any] = {
         "required_kde_coverage_ratio": completeness_summary.get("required_kde_coverage_ratio"),
+        "events_with_open_data_gaps": events_with_open_data_gaps,
         "events_with_missing_required_fields": missing_total,
         "missing_required_by_field": completeness_summary.get("missing_required_by_field") or {},
         # The per-event list in ``completeness`` is capped (250); say so
@@ -785,6 +802,7 @@ def _build_fda_package(
         chain_payload=chain_payload,
         coverage_gate_bypassed=coverage_gate_bypassed,
         rule_versions=rule_versions,
+        events_with_open_data_gaps=sum(1 for event in events if event.get("open_data_gaps")),
     )
     evidence_context = {
         "version": "1.0",
@@ -818,6 +836,10 @@ def _build_fda_package(
         + "\n"
         + pii_note
         + "\n"
+        "Open Data Gaps: the last spreadsheet column names the data gaps open on that row's "
+        "traceability lot and CTE when this export was generated, for example a required KDE "
+        "still outstanding. Blank means none was open.\n"
+        "\n"
         "Verification:\n"
         "- Recompute each manifest file hash and the Merkle root to confirm "
         "the package bytes are internally consistent.\n"
@@ -1017,3 +1039,22 @@ def _generate_csv_v2(events: list[dict], *, include_pii: bool = False) -> str:
         writer.writerow(_event_to_fda_row_v2(event, include_pii=include_pii))
 
     return output.getvalue()
+
+
+def _generate_csv_before_open_gaps(events: list[dict], *, include_pii: bool, v2: bool) -> str:
+    """Write the layout from before the Open Data Gaps column existed.
+
+    Only ``/export/verify`` uses this, to reproduce exports logged back then.
+    """
+    columns = FDA_COLUMNS_V2_WITHOUT_OPEN_GAPS if v2 else FDA_COLUMNS_WITHOUT_OPEN_GAPS
+    output = io.StringIO()
+    writer = csv.DictWriter(output, fieldnames=columns, quoting=csv.QUOTE_ALL, extrasaction="ignore")
+    writer.writeheader()
+
+    for event in events:
+        if v2:
+            writer.writerow(_event_to_fda_row_v2(event, include_pii=include_pii))
+        else:
+            writer.writerow(_event_to_fda_row(event, include_pii=include_pii))
+
+    return output.getvalue()