From 015be6a48e430eb64b3fd13e18e38f64aead7c88 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:20:32 -0800 Subject: [PATCH 1/5] fix(db): keep runtime grants for data gap tracking across deploys Every data gap write has failed in production with permission denied. v135 granted data_gap_events to regengine only in the migration, and the per-deploy runtime-role cutover rebuilds grants from v111's closed inventory, so the grant was revoked. The insert also reads supplier_traceability_lots, which regengine never had. Gap logging is fail-open, so the only trace was a data_gap_write_failed log line. Add both to the v111 inventory and apply them in v142. supplier_traceability_lots gets only the three columns the lookup reads, and data_gap_events gets no DELETE. Both tables have forced tenant RLS. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- ...3_b71d9e4c2a60_v111_runtime_role_grants.py | 8 +++ ...0bfe465816_v142_data_gap_runtime_grants.py | 66 +++++++++++++++++++ ...104120000_v142_data_gap_runtime_grants.sql | 21 ++++++ .../test_v142_data_gap_runtime_grants.py | 56 ++++++++++++++++ 4 files changed, 151 insertions(+) create mode 100644 alembic/versions/20261104_9c0bfe465816_v142_data_gap_runtime_grants.py create mode 100644 supabase/migrations/20261104120000_v142_data_gap_runtime_grants.sql create mode 100644 tests/migrations/test_v142_data_gap_runtime_grants.py diff --git a/alembic/versions/20260803_b71d9e4c2a60_v111_runtime_role_grants.py b/alembic/versions/20260803_b71d9e4c2a60_v111_runtime_role_grants.py index 8ae192e18..e1e433432 100644 --- a/alembic/versions/20260803_b71d9e4c2a60_v111_runtime_role_grants.py +++ b/alembic/versions/20260803_b71d9e4c2a60_v111_runtime_role_grants.py @@ -499,6 +499,7 @@ def _build_migration_owner_posture() -> str: "compliance_notifications", "compliance_status_log", "controls", + "data_gap_events", "food_traceability_list", "funnel_events", "graph_outbox", @@ -568,6 +569,9 @@ def _build_migration_owner_posture() -> str: "public.audit_logs": ("SELECT", "INSERT"), "public.compliance_alerts": ("SELECT", "INSERT"), "public.controls": ("SELECT",), + # v142 -- ingest opens and closes data gaps. No DELETE: gap history is + # evidence. + "public.data_gap_events": ("SELECT", "INSERT", "UPDATE"), "public.food_traceability_list": ("SELECT",), "public.funnel_events": ("SELECT", "INSERT"), "public.obligation_cte_rules": ("SELECT",), @@ -659,6 +663,9 @@ def _build_migration_owner_posture() -> str: # without exposing names/settings. v139 applies the same column grant to # databases already past v111; this inventory keeps it across cutover. "public.tenants": {"SELECT": ("id", "status")}, + # v142 -- opening a gap attributes it to the lot's facility. Only the + # columns that lookup reads; the rest stays control-plane. + "public.supplier_traceability_lots": {"SELECT": ("tenant_id", "tlc_code", "facility_id")}, # v126 -- POST /api/v1/ingest/csv (#3122) retains each upload verbatim # in the ingest transaction (webhook_compat._open_ingestion_run). # v138 extends INSERT with storage_url for partner-uploads pointer rows; @@ -727,6 +734,7 @@ def _build_migration_owner_posture() -> str: "public.audit_logs": ("SELECT", "INSERT"), "public.auth_challenges": ("SELECT", "INSERT", "UPDATE", "DELETE"), "public.compliance_alerts": ("SELECT", "INSERT", "UPDATE"), + "public.data_gap_events": ("SELECT",), "public.funnel_events": ("SELECT",), "public.graph_outbox": ("SELECT", "INSERT", "UPDATE"), "public.invites": ("SELECT", "INSERT", "UPDATE"), diff --git a/alembic/versions/20261104_9c0bfe465816_v142_data_gap_runtime_grants.py b/alembic/versions/20261104_9c0bfe465816_v142_data_gap_runtime_grants.py new file mode 100644 index 000000000..deb140484 --- /dev/null +++ b/alembic/versions/20261104_9c0bfe465816_v142_data_gap_runtime_grants.py @@ -0,0 +1,66 @@ +"""Grant the runtime role what data gap tracking needs (v142). + +v135 granted ``public.data_gap_events`` to ``regengine``, but only in that +migration. ``apply_runtime_role_cutover`` rebuilds runtime grants from v111's +closed inventory on every deploy, so the grant was revoked at once. The gap +insert also reads ``public.supplier_traceability_lots``, which ``regengine`` +never had. Every gap write failed with ``permission denied``, and because gap +logging is fail-open the only trace was a ``data_gap_write_failed`` log line. + +This revision applies the grants, and the matching v111 inventory entries keep +them across deploys. ``supplier_traceability_lots`` gets only the three columns +the lookup reads. Both tables are tenant-scoped by forced RLS. + +Mirrors ``supabase/migrations/20261104120000_v142_data_gap_runtime_grants.sql``. + +Revision ID: 9c0bfe465816 +Revises: 08c7cf36ddbe +Create Date: 2026-11-04 +""" + +from typing import Sequence, Union + +from alembic import op + +revision: str = "9c0bfe465816" # pragma: allowlist secret +down_revision: Union[str, Sequence[str], None] = "08c7cf36ddbe" # pragma: allowlist secret +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.execute( + """ + DO $grant$ + BEGIN + IF to_regclass('public.data_gap_events') IS NOT NULL + AND to_regclass('public.supplier_traceability_lots') IS NOT NULL + AND EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'regengine') + AND EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'regengine_sysadmin') THEN + GRANT SELECT, INSERT, UPDATE ON TABLE public.data_gap_events TO regengine; + GRANT SELECT ON TABLE public.data_gap_events TO regengine_sysadmin; + GRANT SELECT (tenant_id, tlc_code, facility_id) ON TABLE public.supplier_traceability_lots TO regengine; + END IF; + END + $grant$; + """ + ) + + +def downgrade() -> None: + op.execute( + """ + DO $revoke$ + BEGIN + IF to_regclass('public.data_gap_events') IS NOT NULL + AND to_regclass('public.supplier_traceability_lots') IS NOT NULL + AND EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'regengine') + AND EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'regengine_sysadmin') THEN + REVOKE SELECT, INSERT, UPDATE ON TABLE public.data_gap_events FROM regengine; + REVOKE SELECT ON TABLE public.data_gap_events FROM regengine_sysadmin; + REVOKE SELECT (tenant_id, tlc_code, facility_id) ON TABLE public.supplier_traceability_lots FROM regengine; + END IF; + END + $revoke$; + """ + ) diff --git a/supabase/migrations/20261104120000_v142_data_gap_runtime_grants.sql b/supabase/migrations/20261104120000_v142_data_gap_runtime_grants.sql new file mode 100644 index 000000000..9be62e91a --- /dev/null +++ b/supabase/migrations/20261104120000_v142_data_gap_runtime_grants.sql @@ -0,0 +1,21 @@ +-- v142: runtime grants for data gap tracking. +-- +-- Mirrors alembic/versions/20261104_9c0bfe465816_v142_data_gap_runtime_grants.py. +-- The v135 grant was revoked by the per-deploy runtime-role cutover, and the +-- gap insert reads supplier_traceability_lots, which regengine never had, so +-- every gap write failed. v111's inventory keeps these grants across deploys. +-- +-- Idempotent. + +DO $grant$ +BEGIN + IF to_regclass('public.data_gap_events') IS NOT NULL + AND to_regclass('public.supplier_traceability_lots') IS NOT NULL + AND EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'regengine') + AND EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'regengine_sysadmin') THEN + GRANT SELECT, INSERT, UPDATE ON TABLE public.data_gap_events TO regengine; + GRANT SELECT ON TABLE public.data_gap_events TO regengine_sysadmin; + GRANT SELECT (tenant_id, tlc_code, facility_id) ON TABLE public.supplier_traceability_lots TO regengine; + END IF; +END +$grant$; diff --git a/tests/migrations/test_v142_data_gap_runtime_grants.py b/tests/migrations/test_v142_data_gap_runtime_grants.py new file mode 100644 index 000000000..997a8cfee --- /dev/null +++ b/tests/migrations/test_v142_data_gap_runtime_grants.py @@ -0,0 +1,56 @@ +"""v142: data gap writes must keep working after the per-deploy cutover. + +A grant applied only by a migration is revoked by +``apply_runtime_role_cutover``, so these checks pin the v111 inventory, which +is what survives a deploy. +""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +V111 = REPO_ROOT / "alembic/versions/20260803_b71d9e4c2a60_v111_runtime_role_grants.py" +V142 = REPO_ROOT / "alembic/versions/20261104_9c0bfe465816_v142_data_gap_runtime_grants.py" +SUPABASE = REPO_ROOT / "supabase/migrations/20261104120000_v142_data_gap_runtime_grants.sql" +OPEN_GAP_COLUMNS = ("tenant_id", "tlc_code", "facility_id") + + +def _v111(): + spec = importlib.util.spec_from_file_location("regengine_v111_for_v142", V111) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_inventory_keeps_gap_grants_across_cutover() -> None: + v111 = _v111() + + assert "data_gap_events" in v111.APPLICATION_TABLES["public"] + assert set(v111.REGENGINE_TABLE_GRANTS["public.data_gap_events"]) == {"SELECT", "INSERT", "UPDATE"} + assert v111.REGENGINE_SYSADMIN_TABLE_GRANTS["public.data_gap_events"] == ("SELECT",) + + +def test_supplier_lots_stay_column_scoped_for_runtime() -> None: + v111 = _v111() + + assert "public.supplier_traceability_lots" not in v111.REGENGINE_TABLE_GRANTS + assert v111.REGENGINE_COLUMN_GRANTS["public.supplier_traceability_lots"] == {"SELECT": OPEN_GAP_COLUMNS} + + +def test_gap_history_cannot_be_deleted_by_runtime() -> None: + assert "DELETE" not in _v111().REGENGINE_TABLE_GRANTS["public.data_gap_events"] + + +def test_alembic_and_supabase_apply_the_same_grants() -> None: + expected = ( + "GRANT SELECT, INSERT, UPDATE ON TABLE public.data_gap_events TO regengine;", + "GRANT SELECT ON TABLE public.data_gap_events TO regengine_sysadmin;", + "GRANT SELECT (tenant_id, tlc_code, facility_id) ON TABLE public.supplier_traceability_lots TO regengine;", + ) + for path in (V142, SUPABASE): + sql = path.read_text(encoding="utf-8") + for statement in expected: + assert statement in sql, f"{path.name} is missing {statement}" From 85331955f0ef70ddff8df1b2cdd414ea64055bd5 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 4 Nov 2026 03:01:27 -0800 Subject: [PATCH 2/5] feat(fda-export): mark rows with an open data gap in the spreadsheet A shipment stored with an open gap exported with blank TLC source cells and nothing saying why. The sheet gains a last column, Open Data Gaps, naming the gaps open on the row's lot and CTE when the export was generated. Gaps are read as of the export, and the audit row keeps that moment, so /export/verify regenerates the same bytes after a gap closes. Exports logged before the column still verify against the older layout. A failed lookup stops the export with a 503, because a blank cell states that no gap was open. Co-Authored-By: Claude Opus 5.5 --- docs/runbooks/FDA_RECALL_RESPONSE.md | 4 +- services/ingestion/app/fda_export/queries.py | 61 +++- services/ingestion/app/fda_export/recall.py | 3 + services/ingestion/app/fda_export/router.py | 6 + .../ingestion/app/fda_export/verification.py | 22 +- services/ingestion/app/fda_export_service.py | 41 ++- .../tests/test_fda_export_open_gaps.py | 320 ++++++++++++++++++ .../tests/test_fda_export_open_gaps_pg.py | 99 ++++++ .../test_fda_export_v2_reproducibility.py | 2 + .../test_fda_export_verification_1342.py | 4 + services/shared/cte_persistence/exports.py | 12 +- services/shared/data_gap_tracking.py | 37 ++ 12 files changed, 597 insertions(+), 14 deletions(-) create mode 100644 services/ingestion/tests/test_fda_export_open_gaps.py create mode 100644 services/ingestion/tests/test_fda_export_open_gaps_pg.py diff --git a/docs/runbooks/FDA_RECALL_RESPONSE.md b/docs/runbooks/FDA_RECALL_RESPONSE.md index 1280b2f33..cb26bc1a1 100644 --- a/docs/runbooks/FDA_RECALL_RESPONSE.md +++ b/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 35 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: diff --git a/services/ingestion/app/fda_export/queries.py b/services/ingestion/app/fda_export/queries.py index 83d9584d5..89eade52d 100644 --- a/services/ingestion/app/fda_export/queries.py +++ b/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() diff --git a/services/ingestion/app/fda_export/recall.py b/services/ingestion/app/fda_export/recall.py index 97b290744..8179727ab 100644 --- a/services/ingestion/app/fda_export/recall.py +++ b/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() diff --git a/services/ingestion/app/fda_export/router.py b/services/ingestion/app/fda_export/router.py index 34947985a..39334bd46 100644 --- a/services/ingestion/app/fda_export/router.py +++ b/services/ingestion/app/fda_export/router.py @@ -71,6 +71,7 @@ _verify_chain_for_export, build_recall_where_clause, build_v2_where_clause, + attach_open_gaps, fetch_export_log_history, fetch_recall_events, fetch_trace_graph_data, @@ -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) @@ -1061,6 +1065,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 +1116,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() diff --git a/services/ingestion/app/fda_export/verification.py b/services/ingestion/app/fda_export/verification.py index 494513127..29a448f83 100644 --- a/services/ingestion/app/fda_export/verification.py +++ b/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 diff --git a/services/ingestion/app/fda_export_service.py b/services/ingestion/app/fda_export_service.py index 113283029..c20f5eb5d 100644 --- a/services/ingestion/app/fda_export_service.py +++ b/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,7 +171,7 @@ 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) @@ -170,6 +179,8 @@ def _redact_location_value(value: str, column_name: str, include_pii: bool) -> s "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,20 @@ 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 + to_row = _event_to_fda_row_v2 if v2 else _event_to_fda_row + output = io.StringIO() + writer = csv.DictWriter(output, fieldnames=columns, quoting=csv.QUOTE_ALL, extrasaction="ignore") + writer.writeheader() + + for event in events: + writer.writerow(to_row(event, include_pii=include_pii)) + + return output.getvalue() diff --git a/services/ingestion/tests/test_fda_export_open_gaps.py b/services/ingestion/tests/test_fda_export_open_gaps.py new file mode 100644 index 000000000..7e06c78bd --- /dev/null +++ b/services/ingestion/tests/test_fda_export_open_gaps.py @@ -0,0 +1,320 @@ +"""The FDA spreadsheet marks rows whose lot and CTE have an open data gap. + +Locks: +- Open Data Gaps is the last column in the v1 and v2 sheets, and no earlier + column moves. +- A row with a gap names it; a row without one is unchanged and its cell blank. +- Gaps match on lot and CTE, so another CTE of the same lot stays blank. +- A failed lookup stops the export: a blank cell states no gap was open. +- /export/verify reads gaps as of the export, so a gap that closed later is not + reported as tampering, and an export from before the column still verifies. +""" + +from __future__ import annotations + +import asyncio +import csv +import hashlib +import io +import json +import sys +import zipfile +from datetime import datetime, timezone +from types import ModuleType, SimpleNamespace + +import pytest +from app import fda_export_service as svc +from app.fda_export import verification as v +from app.fda_export.queries import attach_open_gaps +from fastapi import HTTPException +from shared.data_gap_tracking import fetch_gaps_open_at +from sqlalchemy.exc import ProgrammingError + +TENANT = "7b6f7f0e-6f0a-4a5e-9d0b-1f2f7c1f0a11" +EXPORTED_AT = datetime(2026, 11, 1, 12, 0, tzinfo=timezone.utc) +GAP_LABEL = "missing_kde: tlc_source_reference" + + +def _event(tlc="LOT-1", event_type="shipping") -> dict: + return { + "id": f"{tlc}-{event_type}", + "traceability_lot_code": tlc, + "product_description": "Romaine", + "quantity": 10, + "unit_of_measure": "cases", + "event_type": event_type, + "event_timestamp": "2026-10-30T10:00:00+00:00", + "sha256_hash": "a" * 64, + "chain_hash": "b" * 64, + "kdes": {}, + } + + +def _rows(csv_content: str) -> list[dict]: + return list(csv.DictReader(io.StringIO(csv_content))) + + +class _Result: + def __init__(self, rows=(), row=None): + self._rows = list(rows) + self._row = row + + def fetchall(self): + return self._rows + + def fetchone(self): + return self._row + + +class _GapSession: + """Answers the gap query from ``gap_rows`` and records its parameters.""" + + def __init__(self, gap_rows=(), audit_row=None): + self.gap_rows = gap_rows + self.audit_row = audit_row + self.gap_params: list[dict] = [] + self.gap_sql: list[str] = [] + + def execute(self, stmt, params=None): + sql = str(stmt) + if "public.data_gap_events" in sql: + self.gap_params.append(dict(params)) + self.gap_sql.append(sql) + return _Result(self.gap_rows) + if "fda_export_log" in sql: + return _Result(row=self.audit_row) + return _Result() + + def close(self): + pass + + +# --------------------------------------------------------------------------- +# Spreadsheet +# --------------------------------------------------------------------------- + + +def test_column_is_last_and_no_earlier_column_moves() -> None: + assert svc.FDA_COLUMNS[-1] == svc.OPEN_GAPS_COLUMN + assert svc.FDA_COLUMNS_V2[-1] == svc.OPEN_GAPS_COLUMN + assert svc.FDA_COLUMNS[:-1] == svc.FDA_COLUMNS_WITHOUT_OPEN_GAPS + assert svc.FDA_COLUMNS_V2[:-1] == svc.FDA_COLUMNS_V2_WITHOUT_OPEN_GAPS + assert svc.FDA_COLUMNS_V2_WITHOUT_OPEN_GAPS[: len(svc.FDA_COLUMNS_WITHOUT_OPEN_GAPS)] == ( + svc.FDA_COLUMNS_WITHOUT_OPEN_GAPS + ) + + +@pytest.mark.parametrize("generate", [svc._generate_csv, svc._generate_csv_v2]) +def test_row_with_a_gap_is_marked_and_row_without_is_blank(generate) -> None: + marked = {**_event("LOT-1"), "open_data_gaps": [GAP_LABEL, "missing_kde: other"]} + clean = {**_event("LOT-2"), "open_data_gaps": []} + + rows = _rows(generate([marked, clean])) + + assert rows[0][svc.OPEN_GAPS_COLUMN] == f"{GAP_LABEL}; missing_kde: other" + assert rows[1][svc.OPEN_GAPS_COLUMN] == "" + + +@pytest.mark.parametrize( + ("generate", "columns", "v2"), + [ + (svc._generate_csv, svc.FDA_COLUMNS_WITHOUT_OPEN_GAPS, False), + (svc._generate_csv_v2, svc.FDA_COLUMNS_V2_WITHOUT_OPEN_GAPS, True), + ], +) +def test_no_gap_row_keeps_every_other_cell(generate, columns, v2) -> None: + """Only the new cell differs from the layout before the column.""" + event = {**_event(), "open_data_gaps": []} + + current = _rows(generate([event]))[0] + older = _rows(svc._generate_csv_before_open_gaps([event], include_pii=False, v2=v2))[0] + + assert list(older) == columns + assert {column: current[column] for column in columns} == older + + +def test_cell_is_sanitized() -> None: + """A gap key can echo a validator message, so it gets the same formula guard.""" + event = {**_event(), "open_data_gaps": ['=HYPERLINK("http://x")']} + + assert not _rows(svc._generate_csv([event]))[0][svc.OPEN_GAPS_COLUMN].startswith("=") + + +def test_pii_redaction_leaves_the_marker() -> None: + event = {**_event(), "open_data_gaps": [GAP_LABEL]} + + assert _rows(svc._generate_csv([event], include_pii=False))[0][svc.OPEN_GAPS_COLUMN] == GAP_LABEL + + +# --------------------------------------------------------------------------- +# Lookup +# --------------------------------------------------------------------------- + + +def test_lookup_is_tenant_scoped_and_as_of_the_export() -> None: + session = _GapSession() + + fetch_gaps_open_at(session, TENANT, ["LOT-2", " LOT-1 ", "LOT-2", ""], EXPORTED_AT) + + assert session.gap_params == [{"tid": TENANT, "lots": ["LOT-1", "LOT-2"], "as_of": EXPORTED_AT}] + assert "tenant_id = CAST(:tid AS uuid)" in session.gap_sql[0] + + +def test_lookup_labels_are_sorted_so_bytes_are_stable() -> None: + session = _GapSession( + [ + ("LOT-1", "shipping", "missing_kde", "z_field"), + ("LOT-1", "shipping", "cte_validation_failed", "rule-7"), + ("LOT-1", "shipping", "missing_kde", "a_field"), + ] + ) + + gaps = fetch_gaps_open_at(session, TENANT, ["LOT-1"], EXPORTED_AT) + + assert gaps == { + ("LOT-1", "shipping"): ( + "cte_validation_failed: rule-7", + "missing_kde: a_field", + "missing_kde: z_field", + ) + } + + +def test_no_lots_runs_no_query() -> None: + session = _GapSession() + + assert fetch_gaps_open_at(session, TENANT, ["", " "], EXPORTED_AT) == {} + assert session.gap_params == [] + + +def test_attach_matches_lot_and_cte() -> None: + session = _GapSession([("LOT-1", "shipping", "missing_kde", "tlc_source_reference")]) + shipped = _event("LOT-1", "SHIPPING") + received = _event("LOT-1", "receiving") + other_lot = _event("LOT-2", "shipping") + + returned = attach_open_gaps(session, TENANT, [shipped, received, other_lot], as_of=EXPORTED_AT) + + assert returned == EXPORTED_AT + assert shipped["open_data_gaps"] == [GAP_LABEL] + assert received["open_data_gaps"] == [] + assert other_lot["open_data_gaps"] == [] + + +def test_attach_binds_the_tenant_for_rls() -> None: + calls: list[dict] = [] + + class Session(_GapSession): + def execute(self, stmt, params=None): + if "set_config" in str(stmt): + calls.append(dict(params)) + return super().execute(stmt, params) + + attach_open_gaps(Session(), TENANT, [_event()], as_of=EXPORTED_AT) + + assert any(TENANT in call.values() for call in calls) + + +def test_failed_lookup_stops_the_export() -> None: + class Denied(_GapSession): + def execute(self, stmt, params=None): + if "public.data_gap_events" in str(stmt): + raise ProgrammingError("SELECT", {}, Exception("permission denied")) + return super().execute(stmt, params) + + event = _event() + + with pytest.raises(HTTPException) as exc: + attach_open_gaps(Denied(), TENANT, [event], as_of=EXPORTED_AT) + + assert exc.value.status_code == 503 + assert "open_data_gaps" not in event + + +# --------------------------------------------------------------------------- +# Package +# --------------------------------------------------------------------------- + + +def test_package_counts_marked_rows_in_the_hashed_evidence() -> None: + events = [{**_event("LOT-1"), "open_data_gaps": [GAP_LABEL]}, {**_event("LOT-2"), "open_data_gaps": []}] + csv_content = svc._generate_csv(events) + csv_hash = hashlib.sha256(csv_content.encode("utf-8")).hexdigest() + chain = SimpleNamespace(valid=True, chain_length=2, errors=[], checked_at="now") + completeness = svc._build_completeness_summary(events) + chain_payload = svc._build_chain_verification_payload( + tenant_id=TENANT, + tlc=None, + events=events, + csv_hash=csv_hash, + chain_verification=chain, + completeness_summary=completeness, + ) + + package_bytes, meta = svc._build_fda_package( + events=events, + csv_content=csv_content, + csv_hash=csv_hash, + chain_payload=chain_payload, + completeness_summary=completeness, + tenant_id=TENANT, + tlc=None, + query_start_date=None, + query_end_date=None, + ) + + with zipfile.ZipFile(io.BytesIO(package_bytes)) as archive: + evidence = json.loads(archive.read(meta["evidence_context_name"])) + readme = archive.read("README.txt").decode("utf-8") + assert evidence["known_gaps"]["events_with_open_data_gaps"] == 1 + assert svc.OPEN_GAPS_COLUMN in readme + + +# --------------------------------------------------------------------------- +# /export/verify +# --------------------------------------------------------------------------- + + +def _audit_row(export_hash: str) -> tuple: + return ("fda_spreadsheet", "LOT-1", None, None, 1, export_hash, EXPORTED_AT, False, None) + + +def _verify(monkeypatch, session: _GapSession) -> dict: + db_module = ModuleType("shared.database") + db_module.SessionLocal = lambda: session + cte_module = ModuleType("shared.cte_persistence") + cte_module.CTEPersistence = lambda db: SimpleNamespace(query_events_by_tlc=lambda **_: [_event("LOT-1")]) + monkeypatch.setitem(sys.modules, "shared.database", db_module) + monkeypatch.setitem(sys.modules, "shared.cte_persistence", cte_module) + return asyncio.run(v.verify_export_handler("exp-1", TENANT)) + + +def test_verify_reads_gaps_as_of_the_export(monkeypatch) -> None: + """The stored moment is what a gap closed since then is judged against.""" + marked = svc._generate_csv([{**_event("LOT-1"), "open_data_gaps": [GAP_LABEL]}]) + session = _GapSession( + gap_rows=[("LOT-1", "shipping", "missing_kde", "tlc_source_reference")], + audit_row=_audit_row(hashlib.sha256(marked.encode("utf-8")).hexdigest()), + ) + + out = _verify(monkeypatch, session) + + assert session.gap_params[0]["as_of"] == EXPORTED_AT + assert out["data_integrity"] == "VERIFIED" + + +def test_verify_accepts_an_export_from_before_the_column(monkeypatch) -> None: + older = svc._generate_csv_before_open_gaps([_event("LOT-1")], include_pii=False, v2=False) + original_hash = hashlib.sha256(older.encode("utf-8")).hexdigest() + + out = _verify(monkeypatch, _GapSession(audit_row=_audit_row(original_hash))) + + assert out["hashes_match"] is True + assert out["regenerated_hash"] == original_hash + + +def test_verify_still_reports_a_real_mismatch(monkeypatch) -> None: + out = _verify(monkeypatch, _GapSession(audit_row=_audit_row("0" * 64))) + + assert out["data_integrity"] == "MISMATCH_DETECTED" + assert out["regenerated_hash"] != "0" * 64 diff --git a/services/ingestion/tests/test_fda_export_open_gaps_pg.py b/services/ingestion/tests/test_fda_export_open_gaps_pg.py new file mode 100644 index 000000000..79a9df15d --- /dev/null +++ b/services/ingestion/tests/test_fda_export_open_gaps_pg.py @@ -0,0 +1,99 @@ +"""The FDA export's gap lookup against real ``public.data_gap_events`` rows. + +The unit tests in ``test_fda_export_open_gaps.py`` run on a fake session that +never parses the SQL. This module runs the real statement at alembic head. + +Skips cleanly when no migrated Postgres is reachable, except under +``REGENGINE_REQUIRE_DB=1`` where an unreachable database fails rather than +skipping (#2530). +""" + +from __future__ import annotations + +import os +import uuid +from datetime import datetime, timedelta, timezone + +import pytest +from shared.data_gap_tracking import fetch_gaps_open_at +from shared.db_url import sync_postgres_url +from sqlalchemy import create_engine, text + +DSN = os.environ.get( + "REMEDIATION_TEST_SUPERUSER_DSN", + "postgresql://regengine:regengine@localhost:55432/regengine", # pragma: allowlist secret — throwaway scratch-DB role +) +RECEIPT = "a" * 64 + + +def _schema_ready(dsn: str) -> bool: + try: + engine = create_engine(sync_postgres_url(dsn), pool_pre_ping=True) + with engine.connect() as conn: + found = conn.execute(text("SELECT to_regclass('public.data_gap_events') IS NOT NULL")).scalar() + engine.dispose() + return bool(found) + except Exception: + return False + + +_DB_READY = _schema_ready(DSN) + +if not _DB_READY and os.environ.get("REGENGINE_REQUIRE_DB") == "1": + raise RuntimeError(f"REGENGINE_REQUIRE_DB=1 but no migrated Postgres is reachable at {DSN!r} — refusing to skip.") + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif(not _DB_READY, reason="migrated Postgres with fsma schema not available"), +] + + +@pytest.fixture() +def conn(): + """A connection whose transaction is rolled back, so seeds are never committed.""" + engine = create_engine(sync_postgres_url(DSN)) + with engine.connect() as connection: + outer = connection.begin() + try: + yield connection + finally: + outer.rollback() + engine.dispose() + + +def _seed(conn, tenant_id: str, gap_key: str, opened_at: datetime, closed_at: datetime | None = None) -> None: + conn.execute( + text( + """ + INSERT INTO public.data_gap_events + (tenant_id, supplier_key, lot_code, cte_type, gap_kind, gap_key, + opened_at, closed_at, closing_record_id, closing_record_hash) + VALUES (:tid, 'unassigned', 'LOT-1', 'shipping', 'missing_kde', :key, + :opened, :closed, :rid, :rhash) + """ + ), + { + "tid": tenant_id, + "key": gap_key, + "opened": opened_at, + "closed": closed_at, + "rid": str(uuid.uuid4()) if closed_at else None, + "rhash": RECEIPT if closed_at else None, + }, + ) + + +def test_gap_state_is_read_as_of_a_moment_for_one_tenant(conn) -> None: + now = datetime.now(timezone.utc) + tenant_id, other_tenant_id = str(uuid.uuid4()), str(uuid.uuid4()) + as_of = now - timedelta(days=1) + + _seed(conn, tenant_id, "open_then", now - timedelta(days=2)) + _seed(conn, tenant_id, "closed_since", now - timedelta(days=2), now - timedelta(hours=1)) + _seed(conn, tenant_id, "closed_before", now - timedelta(days=3), now - timedelta(days=2)) + _seed(conn, tenant_id, "opened_after", now - timedelta(hours=1)) + _seed(conn, other_tenant_id, "other_tenant", now - timedelta(days=2)) + + 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")} diff --git a/services/ingestion/tests/test_fda_export_v2_reproducibility.py b/services/ingestion/tests/test_fda_export_v2_reproducibility.py index a7652d7e3..5093ba5ee 100644 --- a/services/ingestion/tests/test_fda_export_v2_reproducibility.py +++ b/services/ingestion/tests/test_fda_export_v2_reproducibility.py @@ -243,6 +243,7 @@ def test_v2_tlc_export_verifies_with_v2_csv(self, monkeypatch): [ _Result(row=_audit_row(export_hash=original_hash)), _Result(rows=rows), + _Result(), # open-gap lookup ] ) _install(monkeypatch, session) @@ -273,6 +274,7 @@ def test_v2_full_export_applies_stored_event_type_filter(self, monkeypatch): ) ), _Result(rows=rows), + _Result(), # open-gap lookup ] ) _install(monkeypatch, session) diff --git a/services/ingestion/tests/test_fda_export_verification_1342.py b/services/ingestion/tests/test_fda_export_verification_1342.py index 592f5459e..8d4a6413e 100644 --- a/services/ingestion/tests/test_fda_export_verification_1342.py +++ b/services/ingestion/tests/test_fda_export_verification_1342.py @@ -53,6 +53,10 @@ def __init__(self, row): def fetchone(self): return self._row + def fetchall(self): + # The open-gap lookup: these fakes model an export with no gaps. + return [] + class _FakeSession: def __init__(self, *, row=None, execute_raises=None): diff --git a/services/shared/cte_persistence/exports.py b/services/shared/cte_persistence/exports.py index 15ee81e9e..5562c1da5 100644 --- a/services/shared/cte_persistence/exports.py +++ b/services/shared/cte_persistence/exports.py @@ -7,6 +7,7 @@ from __future__ import annotations import hashlib +from datetime import datetime from typing import Optional from uuid import uuid4 @@ -66,6 +67,7 @@ def log_export( export_type: str = "fda_spreadsheet", include_pii: bool = False, query_event_type: Optional[str] = None, + generated_at: Optional[datetime] = None, ) -> str: """Log an FDA export event and return the export log ID. @@ -86,6 +88,10 @@ def log_export( tampered — so both are recorded on the audit row. A schema that predates the v113 migration falls back to the legacy insert below, leaving them NULL, which the verifier reads as "not reproducible". + + ``generated_at`` is the moment the export's Open Data Gaps column + describes, so ``/export/verify`` can read gaps as of the same moment. + Omitted, the column default applies. """ export_id = str(uuid4()) fingerprint = self._compute_export_fingerprint( @@ -105,10 +111,11 @@ def log_export( id, tenant_id, export_type, query_tlc, query_start_date, query_end_date, record_count, export_hash, generated_by, export_fingerprint, - include_pii, query_event_type + include_pii, query_event_type, generated_at ) VALUES ( :id, :tid, :etype, :tlc, :start, :end, :count, - :hash, :by, :fp, :pii, :qetype + :hash, :by, :fp, :pii, :qetype, + COALESCE(CAST(:generated_at AS timestamptz), now()) ) ON CONFLICT (tenant_id, export_fingerprint) WHERE export_fingerprint IS NOT NULL @@ -129,6 +136,7 @@ def log_export( "fp": fingerprint, "pii": bool(include_pii), "qetype": query_event_type, + "generated_at": generated_at, }, ).fetchone() except Exception: diff --git a/services/shared/data_gap_tracking.py b/services/shared/data_gap_tracking.py index f9a66f986..311a79eee 100644 --- a/services/shared/data_gap_tracking.py +++ b/services/shared/data_gap_tracking.py @@ -320,6 +320,43 @@ def record_outcomes(db_session: Any, tenant_id: str, outcomes: Sequence[Validati ) +# Open as of a moment, not as of now. The same predicate is in is_open_at. An +# FDA export is regenerated later by /export/verify, and a gap that has closed +# since must still read as open at the original moment or the bytes differ. +_GAPS_OPEN_AT_SQL = text( + """ + SELECT lot_code, cte_type, gap_kind, gap_key + FROM public.data_gap_events + WHERE tenant_id = CAST(:tid AS uuid) + AND lot_code = ANY(:lots) + AND opened_at <= CAST(:as_of AS timestamptz) + AND (closed_at IS NULL OR closed_at > CAST(:as_of AS timestamptz)) + """ +) + + +def fetch_gaps_open_at( + db_session: Any, + tenant_id: str, + lot_codes: Iterable[str], + as_of: datetime, +) -> dict[tuple[str, str], tuple[str, ...]]: + """Gaps open at ``as_of``, by (lot code, lower-case CTE type). + + Values are sorted ``": "`` labels, so a caller writing + them into an export gets the same bytes on every run. Raises on a database + error: unlike the writer, a reader's caller decides what failure means. + """ + lots = sorted({lot.strip() for lot in lot_codes if lot and lot.strip()}) + if not lots: + return {} + rows = db_session.execute(_GAPS_OPEN_AT_SQL, {"tid": tenant_id, "lots": lots, "as_of": as_of}).fetchall() + labels: dict[tuple[str, str], set[str]] = {} + for lot_code, cte_type, gap_kind, gap_key in rows: + labels.setdefault((lot_code, cte_type), set()).add(f"{gap_kind}: {gap_key}") + return {key: tuple(sorted(values)) for key, values in labels.items()} + + def fetch_supplier_summaries( db_session: Any, tenant_id: str, From 003fdb89eeb15195901b92a51e086018a3408e70 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 4 Nov 2026 03:03:44 -0800 Subject: [PATCH 3/5] docs(fda-export): correct the runbook column count Co-Authored-By: Claude Opus 5.5 --- docs/runbooks/FDA_RECALL_RESPONSE.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/runbooks/FDA_RECALL_RESPONSE.md b/docs/runbooks/FDA_RECALL_RESPONSE.md index cb26bc1a1..9f8e518d3 100644 --- a/docs/runbooks/FDA_RECALL_RESPONSE.md +++ b/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 35 columns defined in `FDA_COLUMNS` (`fda_export_service.py`): +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 From b56e30874da44c742d48e9c4d2471f0ccfce2fa9 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 4 Nov 2026 03:07:59 -0800 Subject: [PATCH 4/5] fix(fda-export): store the export moment on /export/all audit rows /export/all read gaps as of one moment and left the audit row to take its own default, so verify could read gap state a moment later than the export did. A new test requires every audit write in the export paths to carry the moment. Co-Authored-By: Claude Opus 5.5 --- services/ingestion/app/fda_export/router.py | 5 +++-- .../tests/test_fda_export_open_gaps.py | 20 +++++++++++++++++++ 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/services/ingestion/app/fda_export/router.py b/services/ingestion/app/fda_export/router.py index 39334bd46..b68ac6d58 100644 --- a/services/ingestion/app/fda_export/router.py +++ b/services/ingestion/app/fda_export/router.py @@ -68,10 +68,11 @@ 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, - attach_open_gaps, fetch_export_log_history, fetch_recall_events, fetch_trace_graph_data, @@ -79,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, @@ -643,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: diff --git a/services/ingestion/tests/test_fda_export_open_gaps.py b/services/ingestion/tests/test_fda_export_open_gaps.py index 7e06c78bd..3815fd596 100644 --- a/services/ingestion/tests/test_fda_export_open_gaps.py +++ b/services/ingestion/tests/test_fda_export_open_gaps.py @@ -12,6 +12,7 @@ from __future__ import annotations +import ast import asyncio import csv import hashlib @@ -20,6 +21,7 @@ import sys import zipfile from datetime import datetime, timezone +from pathlib import Path from types import ModuleType, SimpleNamespace import pytest @@ -275,6 +277,24 @@ def test_package_counts_marked_rows_in_the_hashed_evidence() -> None: # --------------------------------------------------------------------------- +@pytest.mark.parametrize("module", ["router.py", "recall.py"]) +def test_every_export_stores_the_moment_its_gaps_describe(module) -> None: + """Verify reads gaps as of the audit row, so each audit write must carry that moment.""" + tree = ast.parse((Path(v.__file__).parent / module).read_text(encoding="utf-8")) + writers = [ + node + for node in ast.walk(tree) + if isinstance(node, ast.Call) + and (getattr(node.func, "attr", None) == "log_export" or getattr(node.func, "id", "").endswith("_export")) + and getattr(node.func, "id", "log_export").startswith("log_") + ] + + assert writers + for call in writers: + assert "generated_at" in {keyword.arg for keyword in call.keywords}, f"line {call.lineno}" + + + def _audit_row(export_hash: str) -> tuple: return ("fda_spreadsheet", "LOT-1", None, None, 1, export_hash, EXPORTED_AT, False, None) From 34ca566c26cb62a91f0aeb42a589a6f705b48ccb Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 4 Nov 2026 03:29:03 -0800 Subject: [PATCH 5/5] fix(fda-export): keep the older-layout writer inside the safe-cell rule Semgrep's FDA CSV guard only recognises the row builders by name, so the indirect call failed SAST. Both branches now call a builder directly. Output is unchanged. Co-Authored-By: Claude Opus 5.5 --- services/ingestion/app/fda_export_service.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/services/ingestion/app/fda_export_service.py b/services/ingestion/app/fda_export_service.py index c20f5eb5d..8f4cad36c 100644 --- a/services/ingestion/app/fda_export_service.py +++ b/services/ingestion/app/fda_export_service.py @@ -1047,12 +1047,14 @@ def _generate_csv_before_open_gaps(events: list[dict], *, include_pii: bool, v2: 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 - to_row = _event_to_fda_row_v2 if v2 else _event_to_fda_row output = io.StringIO() writer = csv.DictWriter(output, fieldnames=columns, quoting=csv.QUOTE_ALL, extrasaction="ignore") writer.writeheader() for event in events: - writer.writerow(to_row(event, include_pii=include_pii)) + 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()