River Intelligence: Collector Data Freshness Audit (Codex Mini)

{
  "code": "## River Intelligence: Collector Data Freshness Audit\n\n### Implementation Approach\n\nThe `data_empire_pulse.py` has `check_bigquery()` (line 174) for connectivity but not per-table freshness. Here's the audit query:\n\n### BigQuery SQL for Freshness Audit\n```sql\nWITH table_stats AS (\n  SELECT\n    table_id,\n    row_count,\n    TIMESTAMP_MILLIS(last_modified_time) AS last_modified,\n    TIMESTAMP_DIFF(CURRENT_TIMESTAMP(), TIMESTAMP_MILLIS(last_modified_time), DAY) AS staleness_days,\n    ROUND(size_bytes / 1073741824, 2) AS size_gb\n  FROM `duty-pay.oneworld_trade.__TABLES__`\n  WHERE table_id IN (\n    'csms_messages', 'federal_register', 'cross_rulings', 'hts_tariffs',\n    'fda_import_alerts', 'fda_warning_letters', 'fda_recalls',\n    'bis_entity_list', 'bis_denied_persons', 'bis_unverified',\n    'bts_trade_statistics', 'bts_commodity_flow',\n    'census_trade_data', 'census_naics',\n    'un_comtrade', 'wto_tariff_profiles', 'wto_disputes',\n    'usitc_dataweb', 'informed_compliance',\n    'cbp_newsroom', 'cbp_documents'\n  )\n)\nSELECT *\nFROM table_stats\nORDER BY staleness_days DESC\n```\n\n### Expected Output Format\n```json\n[\n  {\n    \"table\": \"informed_compliance\",\n    \"row_count\": 4521,\n    \"last_modified\": \"2025-11-15T08:30:00Z\",\n    \"staleness_days\": 102,\n    \"size_gb\": 0.03,\n    \"status\": \"stale\"\n  },\n  {\n    \"table\": \"csms_messages\",\n    \"row_count\": 28934,\n    \"last_modified\": \"2026-02-24T14:22:00Z\",\n    \"staleness_days\": 1,\n    \"status\": \"fresh\"\n  }\n]\n```\n\n### Status Classification\n- `fresh`: staleness_days ≤ 7\n- `aging`: staleness_days 8-30\n- `stale`: staleness_days 31-90\n- `dead`: staleness_days > 90\n\n### Existing Infrastructure\n- `data_empire_worker.py`: Has `_run_bq()` helper for executing BQ queries via subprocess\n- `data_empire_pulse.py`: Has `check_bigquery()` for connectivity — extend with table-level stats\n- `river_scheduler.py`: `stage_pulse()` already gathers pipeline health — add BQ freshness section\n\n### CLI Execution\n```bash\nbq query --format=json --use_legacy_sql=false < freshness_audit.sql\n```\n\n### Integration Point\nAdd to `data_empire_pulse.py`:\n```python\ndef check_bigquery_freshness() -> list:\n    \"\"\"Per-table freshness audit for oneworld_trade dataset.\"\"\"\n    sql = \"SELECT table_id, row_count, ...\"\n    result = subprocess.run([\"bq\", \"query\", \"--format=json\", \"--use_legacy_sql=false\", sql], ...)\n    return sorted(json.loads(result.stdout), key=lambda x: x[\"staleness_days\"], reverse=True)\n```\n\nThis slots into the existing `data_empire_pulse()` async gather alongside `check_bigquery()`.",
  "content": "## River Intelligence: Collector Data Freshness Audit\n\n### Implementation Approach\n\nThe `data_empire_pulse.py` has `check_bigquery()` (line 174) for connectivity but not per-table freshness. Here's the audit query:\n\n### BigQuery SQL for Freshness Audit\n```sql\nWITH table_stats AS (\n  SELECT\n    table_id,\n    row_count,\n    TIMESTAMP_MILLIS(last_modified_time) AS last_modified,\n    TIMESTAMP_DIFF(CURRENT_TIMESTAMP(), TIMESTAMP_MILLIS(last_modified_time), DAY) AS staleness_days,\n    ROUND(size_bytes / 1073741824, 2) AS size_gb\n  FROM `duty-pay.oneworld_trade.__TABLES__`\n  WHERE table_id IN (\n    'csms_messages', 'federal_register', 'cross_rulings', 'hts_tariffs',\n    'fda_import_alerts', 'fda_warning_letters', 'fda_recalls',\n    'bis_entity_list', 'bis_denied_persons', 'bis_unverified',\n    'bts_trade_statistics', 'bts_commodity_flow',\n    'census_trade_data', 'census_naics',\n    'un_comtrade', 'wto_tariff_profiles', 'wto_disputes',\n    'usitc_dataweb', 'informed_compliance',\n    'cbp_newsroom', 'cbp_documents'\n  )\n)\nSELECT *\nFROM table_stats\nORDER BY staleness_days DESC\n```\n\n### Expected Output Format\n```json\n[\n  {\n    \"table\": \"informed_compliance\",\n    \"row_count\": 4521,\n    \"last_modified\": \"2025-11-15T08:30:00Z\",\n    \"staleness_days\": 102,\n    \"size_gb\": 0.03,\n    \"status\": \"stale\"\n  },\n  {\n    \"table\": \"csms_messages\",\n    \"row_count\": 28934,\n    \"last_modified\": \"2026-02-24T14:22:00Z\",\n    \"staleness_days\": 1,\n    \"status\": \"fresh\"\n  }\n]\n```\n\n### Status Classification\n- `fresh`: staleness_days ≤ 7\n- `aging`: staleness_days 8-30\n- `stale`: staleness_days 31-90\n- `dead`: staleness_days > 90\n\n### Existing Infrastructure\n- `data_empire_worker.py`: Has `_run_bq()` helper for executing BQ queries via subprocess\n- `data_empire_pulse.py`: Has `check_bigquery()` for connectivity — extend with table-level stats\n- `river_scheduler.py`: `stage_pulse()` already gathers pipeline health — add BQ freshness section\n\n### CLI Execution\n```bash\nbq query --format=json --use_legacy_sql=false < freshness_audit.sql\n```\n\n### Integration Point\nAdd to `data_empire_pulse.py`:\n```python\ndef check_bigquery_freshness() -> list:\n    \"\"\"Per-table freshness audit for oneworld_trade dataset.\"\"\"\n    sql = \"SELECT table_id, row_count, ...\"\n    result = subprocess.run([\"bq\", \"query\", \"--format=json\", \"--use_legacy_sql=false\", sql], ...)\n    return sorted(json.loads(result.stdout), key=lambda x: x[\"staleness_days\"], reverse=True)\n```\n\nThis slots into the existing `data_empire_pulse()` async gather alongside `check_bigquery()`."
}