fix: pipeline reliability — WS, summarizer, ingest, health

Stop the live HUD reconnect storm (nginx WS snippet + backoff), copy intel/nous_client into the summarizer image, and make event ingest idempotent on URL. GDELT uses the DOC API; NWS no longer sends bbox; FIRMS is one ON CONFLICT batch; GET /api/aircraft serves last-known. Health reports freshness without 503ing docker. EONET + CISA KEV added.
2026-08-28 21:49:05 -04:00 · 2026-08-28 21:49:05 -04:00 · d81ea68dec
commit d81ea68dec
parent c6bfc50705
26 changed files with 1169 additions and 115 deletions
--- a/alembic/versions/007_event_dedup.py
+++ b/alembic/versions/007_event_dedup.py
@ -0,0 +1,115 @@
+"""event_dedup + Timescale compression/retention
+
+Revision ID: 007_event_dedup
+Revises: 006_merge_heads
+Create Date: 2026-08-28
+"""
+
+from alembic import op
+
+revision = "007_event_dedup"
+down_revision = "006_merge_heads"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute("""
+        CREATE TABLE IF NOT EXISTS event_dedup (
+            url TEXT PRIMARY KEY,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+        )
+    """)
+
+    # Keep the earliest row per URL, drop the 10× USGS/camera dupes.
+    op.execute("""
+        DELETE FROM events a
+        USING events b
+        WHERE a.url IS NOT NULL AND a.url <> ''
+          AND a.url = b.url
+          AND (a.ingested_at, a.id) > (b.ingested_at, b.id)
+    """)
+    op.execute("""
+        INSERT INTO event_dedup (url)
+        SELECT DISTINCT url FROM events
+        WHERE url IS NOT NULL AND url <> ''
+        ON CONFLICT (url) DO NOTHING
+    """)
+
+    # Compression + retention. Policies no-op if Timescale rejects (fresh PG).
+    op.execute("""
+        DO $$
+        BEGIN
+            PERFORM add_compression_policy('events', INTERVAL '7 days', if_not_exists => TRUE);
+        EXCEPTION WHEN OTHERS THEN
+            BEGIN
+                ALTER TABLE events SET (
+                    timescaledb.compress,
+                    timescaledb.compress_segmentby = 'source_type',
+                    timescaledb.compress_orderby = 'ingested_at DESC'
+                );
+                PERFORM add_compression_policy('events', INTERVAL '7 days', if_not_exists => TRUE);
+            EXCEPTION WHEN OTHERS THEN
+                NULL;
+            END;
+        END
+        $$;
+    """)
+    op.execute("""
+        DO $$
+        BEGIN
+            PERFORM add_retention_policy('events', INTERVAL '180 days', if_not_exists => TRUE);
+        EXCEPTION WHEN OTHERS THEN
+            NULL;
+        END
+        $$;
+    """)
+    op.execute("""
+        DO $$
+        BEGIN
+            ALTER TABLE fires SET (
+                timescaledb.compress,
+                timescaledb.compress_segmentby = 'satellite',
+                timescaledb.compress_orderby = 'acq_time DESC'
+            );
+            PERFORM add_compression_policy('fires', INTERVAL '7 days', if_not_exists => TRUE);
+            PERFORM add_retention_policy('fires', INTERVAL '90 days', if_not_exists => TRUE);
+        EXCEPTION WHEN OTHERS THEN
+            NULL;
+        END
+        $$;
+    """)
+    op.execute("""
+        DO $$
+        BEGIN
+            ALTER TABLE aircraft_positions SET (
+                timescaledb.compress,
+                timescaledb.compress_segmentby = 'hex',
+                timescaledb.compress_orderby = 'ts DESC'
+            );
+            PERFORM add_compression_policy('aircraft_positions', INTERVAL '1 day', if_not_exists => TRUE);
+            PERFORM add_retention_policy('aircraft_positions', INTERVAL '14 days', if_not_exists => TRUE);
+        EXCEPTION WHEN OTHERS THEN
+            NULL;
+        END
+        $$;
+    """)
+    op.execute("""
+        DO $$
+        BEGIN
+            ALTER TABLE vessel_positions SET (
+                timescaledb.compress,
+                timescaledb.compress_segmentby = 'mmsi',
+                timescaledb.compress_orderby = 'ts DESC'
+            );
+            PERFORM add_compression_policy('vessel_positions', INTERVAL '1 day', if_not_exists => TRUE);
+            PERFORM add_retention_policy('vessel_positions', INTERVAL '14 days', if_not_exists => TRUE);
+        EXCEPTION WHEN OTHERS THEN
+            NULL;
+        END
+        $$;
+    """)
+
+
+def downgrade() -> None:
+    op.execute("DROP TABLE IF EXISTS event_dedup")
--- a/app/fire_sources.py
+++ b/app/fire_sources.py
@ -162,6 +162,16 @@ async def publish_fire_batch(points: list[dict]) -> int:
    return len(points)


+async def persist_hotspots(points: list[dict]) -> int:
+    """Write a FIRMS poll to Postgres in one ON CONFLICT batch.
+
+    NATS-per-row was 93k commits + geofence/correlation per hotspot.
+    """
+    from ingestor import ingest_fire_rows
+
+    return await ingest_fire_rows(points)
+
+
 async def ingest_fires(bbox: str | None = None) -> int:
    """Fetch the FIRMS hotspot CSV for an area and publish it to NATS.

@ -206,7 +216,7 @@ async def ingest_fires(bbox: str | None = None) -> int:
                )
                continue
            points = parse_firms_csv(text)
-            published = await publish_fire_batch(points)
+            published = await persist_hotspots(points)
            total_published += published
            logger.info(
                "FIRMS: fetched %d hotspot(s) for bbox=%s (%s), published %d",
--- a/app/ingestor.py
+++ b/app/ingestor.py
@ -14,7 +14,9 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert
 from database import async_session
 from models import events as events_table
 from models import fires as fires_table
+from models import event_dedup as event_dedup_table
 from config import NATS_URL
+from sources import event_dedup_key

 logger = logging.getLogger("osint.ingestor")

@ -115,6 +117,42 @@ async def ingest_fire_row(msg: dict) -> bool:
    return inserted


+async def ingest_fire_rows(msgs: list[dict]) -> int:
+    """Bulk-insert FIRMS hotspots: one INSERT, one ON CONFLICT, one commit."""
+    rows = []
+    for msg in msgs:
+        row = _fire_row_from_msg(msg)
+        if row is not None:
+            rows.append(row)
+    if not rows:
+        return 0
+    async with async_session() as session:
+        stmt = (
+            pg_insert(fires_table)
+            .values(rows)
+            .on_conflict_do_nothing(constraint="pk_fires_natural_key")
+        )
+        result = await session.execute(stmt)
+        await session.commit()
+    inserted = int(result.rowcount or 0)
+    if inserted:
+        logger.info("bulk ingested %d/%d FIRMS hotspots", inserted, len(rows))
+        from live_layers import aircraft_last_known
+        from fire_aircraft import correlate_and_notify
+        from tracks import recent_markers
+        acs = list(aircraft_last_known.values()) or await recent_markers("aircraft")
+        if acs:
+            fires = [
+                {
+                    "id": f"firms:{r['latitude']:.4f},{r['longitude']:.4f}",
+                    "lat": r["latitude"], "lon": r["longitude"], "label": "FIRMS",
+                }
+                for r in rows[:500]
+            ]
+            await correlate_and_notify(fires, acs)
+    return inserted
+
+
 async def ingest_event(msg: dict):
    """Ingest a single event from NATS into PostgreSQL."""
    # Active fire/hotspot messages carry a dedicated schema and land in the
@ -149,10 +187,26 @@ async def ingest_event(msg: dict):
    }

    # Parse timestamp if string
-    if isinstance(event_row["source_timestamp"], str):
-        event_row["source_timestamp"] = datetime.fromisoformat(event_row["source_timestamp"])
+    ts = event_row["source_timestamp"]
+    if isinstance(ts, str):
+        ts = datetime.fromisoformat(ts.replace("Z", "+00:00"))
+    if isinstance(ts, datetime) and ts.tzinfo is None:
+        ts = ts.replace(tzinfo=timezone.utc)
+    event_row["source_timestamp"] = ts

+    key = event_dedup_key(event_row)
    async with async_session() as session:
+        if key:
+            dedup = (
+                pg_insert(event_dedup_table)
+                .values(url=key)
+                .on_conflict_do_nothing(index_elements=["url"])
+            )
+            claimed = await session.execute(dedup)
+            if not claimed.rowcount:
+                await session.commit()
+                logger.info("skip duplicate event url=%s", key)
+                return None
        result = await session.execute(events_table.insert().values(**event_row))
        await session.commit()
        event_id = result.inserted_primary_key[0]  # type: ignore[union-attr]
--- a/app/live_layers.py
+++ b/app/live_layers.py
@ -12,6 +12,7 @@ or AIS poll.
 from __future__ import annotations

 import asyncio
+import logging
 import math
 import time
 from datetime import datetime, timezone
@ -21,6 +22,8 @@ import httpx

 from config import OSINT_USER_AGENT

+logger = logging.getLogger("osint.live_layers")
+
 MARKER_FIELDS = ("id", "lat", "lon", "heading", "speed", "label", "extra")

 ADSB_LOL_BASE = "https://api.adsb.lol"
@ -800,8 +803,24 @@ async def _get_json(url: str, params: dict | None = None) -> Any:
    return resp.json()


-async def fetch_aircraft(bbox: str, limit: int = DEFAULT_LIMIT) -> list[dict]:
+async def fetch_aircraft(
+    bbox: str, limit: int = DEFAULT_LIMIT, persist: bool = False,
+) -> list[dict]:
+    """Viewport ADS-B.
+
+    GET path (persist=False) serves in-memory last-known and never writes
+    tracks/geofences. Background refresh (persist=True) hits ADSB.lol and
+    then persist_aircraft_snapshot.
+    """
    minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
+    if not persist:
+        cached = [
+            dict(v) for v in aircraft_last_known.values()
+            if v.get("lat") is not None and v.get("lon") is not None
+        ]
+        filtered = filter_points_bbox(cached, minlon, minlat, maxlon, maxlat, limit)
+        if filtered:
+            return filtered
    qminlon, qminlat, qmaxlon, qmaxlat = quantize_bbox(minlon, minlat, maxlon, maxlat)
    lat, lon, radius = bbox_center_radius_nm(qminlon, qminlat, qmaxlon, qmaxlat)
    cache_key = f"adsb:{lat:.2f}:{lon:.2f}:{radius}"
@ -811,12 +830,20 @@ async def fetch_aircraft(bbox: str, limit: int = DEFAULT_LIMIT) -> list[dict]:
        return transform_adsb_lol(await _get_json(url))

    rows = await _ttl_get(cache_key, 8.0, _load)
+    for m in rows:
+        aircraft_last_known[str(m.get("id"))] = m
+    if persist:
+        await persist_aircraft_snapshot(rows)
+    return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit)
+
+
+async def persist_aircraft_snapshot(rows: list[dict]) -> None:
+    """Track / geofence / WS / fire correlation — never on the GET path."""
    from ws_manager import manager
    from tracks import record_position
    from geofence import record_and_notify
-    aircraft_last_known.clear()
+
    for m in rows:
-        aircraft_last_known[str(m.get("id"))] = m
        mlat, mlon = m.get("lat"), m.get("lon")
        if mlat is None or mlon is None:
            continue
@ -830,7 +857,6 @@ async def fetch_aircraft(bbox: str, limit: int = DEFAULT_LIMIT) -> list[dict]:
    if fire_last_known:
        from fire_aircraft import correlate_and_notify
        await correlate_and_notify(fire_last_known, rows)
-    return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit)


 async def fetch_trains(bbox: str | None, limit: int = DEFAULT_LIMIT) -> list[dict]:
@ -964,13 +990,16 @@ async def fetch_weather_alerts(area: str | None, bbox: str | None) -> dict:
        clip_box = None
        if area:
            nws_params["area"] = area.upper()
-        elif bbox:
+        if bbox:
            clip_box = quantize_bbox(*parse_bbox(bbox))
-            nws_params["bbox"] = f"{clip_box[0]},{clip_box[1]},{clip_box[2]},{clip_box[3]}"
+        # NWS /alerts/active 400s on bbox= — never send it; clip locally.
        nws_fc: dict = {"features": []}
+        nws_ok = True
        try:
            nws_fc = await _get_json(NWS_ALERTS, nws_params)
-        except Exception:
+        except Exception as exc:
+            logger.warning("NWS alerts fetch failed: %s", exc)
+            nws_ok = False
            nws_fc = {"features": []}
        sbw_fc = await _ttl_get("iem:sbw", 45.0, _load_iem)
        features = []
@ -988,11 +1017,15 @@ async def fetch_weather_alerts(area: str | None, bbox: str | None) -> dict:
            if "event" not in props:
                props["event"] = props.get("ps") or "Storm-based warning"
            features.append({**feat, "properties": slim_alert_properties(props)})
-        merged = {"type": "FeatureCollection", "features": features}
+        merged = {"type": "FeatureCollection", "features": features, "nws_ok": nws_ok}
        if clip_box:
-            merged = clip_fc_to_bbox(merged, *clip_box)
+            clipped = clip_fc_to_bbox(merged, *clip_box)
+            clipped["nws_ok"] = nws_ok
+            merged = clipped
        elif bbox:
-            merged = clip_fc_to_bbox(merged, *parse_bbox(bbox))
+            clipped = clip_fc_to_bbox(merged, *parse_bbox(bbox))
+            clipped["nws_ok"] = nws_ok
+            merged = clipped
        return merged

    key = f"alerts:{area or ''}:{bbox_cell_key(bbox) if bbox else ''}"
--- a/app/main.py
+++ b/app/main.py
@ -72,12 +72,16 @@ async def _lifespan(app: FastAPI):
        pass
    from config import AISSTREAM_IN_APP
    ais_task = None
+    adsb_task = None
    if AISSTREAM_IN_APP:
        from ais_stream import run_ais_worker
        ais_task = asyncio.create_task(run_ais_worker())
+    adsb_task = asyncio.create_task(_adsb_refresh_loop())
    yield
    if ais_task is not None:
        ais_task.cancel()
+    if adsb_task is not None:
+        adsb_task.cancel()
    await close_http()


@ -92,6 +96,35 @@ app.add_middleware(GZipMiddleware, minimum_size=1024)
 STATIC_DIR = Path(__file__).parent / "static"


+async def _adsb_refresh_loop() -> None:
+    """Poll ADSB.lol for connected viewports — never from GET /api/aircraft."""
+    from ws_manager import manager
+    from live_layers import fetch_aircraft
+
+    while True:
+        try:
+            boxes = manager.viewports()
+            for box in boxes:
+                bbox = f"{box[0]},{box[1]},{box[2]},{box[3]}"
+                try:
+                    await fetch_aircraft(bbox, persist=True)
+                except Exception as exc:  # noqa: BLE001
+                    logger.warning("adsb_refresh_failed", error=str(exc))
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("adsb_refresh_loop", error=str(exc))
+        await asyncio.sleep(8)
+
+
+class CachedStaticFiles(StaticFiles):
+    """Long-cache hashed/vendor Leaflet assets; index.html is served separately."""
+
+    async def get_response(self, path: str, scope):
+        resp = await super().get_response(path, scope)
+        if path.startswith("vendor/") or path.endswith((".js", ".css", ".woff2", ".png", ".svg")):
+            resp.headers["Cache-Control"] = "public, max-age=31536000, immutable"
+        return resp
+
+
 # ── Helpers ───────────────────────────────────────────────────────────────

 def event_to_out(row: dict) -> EventOut:
@ -154,11 +187,65 @@ def alert_to_out(row: dict) -> AlertOut:

@app.get("/api/health")
 async def health():
-    """Health check with database connectivity."""
+    """Liveness: process + DB. Pipeline freshness is in ``checks`` (HTTP 200
+    unless DB is down, so docker healthcheck does not restart a working HUD).
+    """
+    try:
        async with async_session() as session:
            result = await session.execute(select(func.now()))
            db_time = result.scalar()
-    return {"status": "ok", "db_time": db_time.isoformat() if db_time else None}
+            checks = await _pipeline_checks(session)
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("health_db_failed", error=str(exc))
+        return JSONResponse(
+            {"status": "down", "db_time": None, "checks": {"db": False}},
+            status_code=503,
+        )
+    checks["db"] = True
+    stale = [
+        name for name, val in checks.items()
+        if isinstance(val, dict) and val.get("ok") is False
+    ]
+    status = "degraded" if stale else "ok"
+    return {
+        "status": status,
+        "db_time": db_time.isoformat() if db_time else None,
+        "checks": checks,
+    }
+
+
+@app.get("/api/ready")
+async def ready():
+    """Readiness: 503 only when the database is unreachable."""
+    payload = await health()
+    if isinstance(payload, JSONResponse):
+        return payload
+    return JSONResponse(payload, status_code=200)
+
+
+async def _pipeline_checks(session) -> dict:
+    now = datetime.now(timezone.utc)
+    checks: dict = {}
+
+    async def _age(sql: str, name: str, max_age_s: int) -> None:
+        try:
+            ts = (await session.execute(text(sql))).scalar()
+        except Exception:
+            checks[name] = {"ok": False, "age_s": None}
+            return
+        if ts is None:
+            checks[name] = {"ok": False, "age_s": None}
+            return
+        if getattr(ts, "tzinfo", None) is None:
+            ts = ts.replace(tzinfo=timezone.utc)
+        age = (now - ts).total_seconds()
+        checks[name] = {"ok": age <= max_age_s, "age_s": int(age)}
+
+    await _age("SELECT max(ingested_at) FROM events", "events", 20 * 60)
+    await _age("SELECT max(ingested_at) FROM fires", "fires", 30 * 60)
+    await _age("SELECT max(timestamp) FROM articles", "articles", 2 * 3600)
+    await _age("SELECT max(batch_timestamp) FROM article_summaries", "summaries", 2 * 3600)
+    return checks


 def overlay_json(data, max_age: int) -> JSONResponse:
@ -821,17 +908,19 @@ async def get_timeline(
    """Event timeline: counts and avg sentiment per time bucket."""
    async with async_session() as session:
        cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)
-        # Use date_trunc for bucketing
-        buckets = await session.execute(text(f"""
+        bucket_s = bucket_hours * 3600
+        buckets = await session.execute(text("""
            SELECT
-                date_trunc('hour', source_timestamp) AS ts,
+                to_timestamp(
+                    floor(extract(epoch FROM source_timestamp) / :bucket_s) * :bucket_s
+                ) AT TIME ZONE 'UTC' AS ts,
                COUNT(*) AS event_count,
                COALESCE(AVG(sentiment_score), 0) AS avg_sentiment
            FROM events
            WHERE source_timestamp >= :cutoff
            GROUP BY ts
            ORDER BY ts
-        """), {"cutoff": cutoff})
+        """), {"cutoff": cutoff, "bucket_s": bucket_s})
        rows = buckets.mappings().all()

    return [TimelinePoint(timestamp=r["ts"], event_count=r["event_count"],
@ -1319,7 +1408,7 @@ async def list_aircraft(
        ts = parse_timestamp(timestamp)
        if ts is not None:
            return overlay_json(await fetch_positions_at("aircraft", ts, bbox, limit), 5)
-        return overlay_json(await fetch_aircraft(bbox, limit), 5)
+        return overlay_json(await fetch_aircraft(bbox, limit, persist=False), 5)
    except ValueError as exc:
        raise HTTPException(422, str(exc)) from exc
    except Exception as exc:
@ -1550,7 +1639,7 @@ async def map_layer_times(
    return {"layer": layer, **domain}


-app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
+app.mount("/static", CachedStaticFiles(directory=str(STATIC_DIR)), name="static")


 if __name__ == "__main__":
--- a/app/models.py
+++ b/app/models.py
@ -68,6 +68,15 @@ Index("ix_events_search_vector", events.c.search_vector, postgresql_using="gin")
 # Spatial index on location
 Index("ix_events_location", events.c.location_lat, events.c.location_lon)

+# Timescale unique indexes must include the partition column, so URL
+# idempotency lives on a regular table — not the events hypertable.
+event_dedup = Table(
+    "event_dedup",
+    metadata,
+    Column("url", Text, primary_key=True),
+    Column("created_at", DateTime(timezone=True), server_default=func.now(), nullable=False),
+)
+

 # ── Entities (people, organizations, locations of interest) ──────────────

--- a/app/run_ingester.py
+++ b/app/run_ingester.py
@ -25,7 +25,7 @@ import sys
 sys.path.insert(0, sys_path)

 from config import NATS_URL, FIRMS_INTERVAL, FIRMS_DATASET, AISSTREAM_IN_INGEST  # noqa: E402
-from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes  # noqa: E402
+from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_eonet, ingest_cisa_kev  # noqa: E402
 from fire_sources import ingest_fires  # noqa: E402
 from ingestor import ingest_event, start_nats_consumer  # noqa: E402

@ -37,6 +37,8 @@ INTERVAL = int(os.getenv("INGEST_INTERVAL", "300"))
 GDELT_QUERY = os.getenv("GDELT_QUERY", "")
 ENABLE_QUAKES = os.getenv("INGEST_EARTHQUAKES", "1").lower() in ("1", "true", "yes")
 ENABLE_FIRES = os.getenv("INGEST_FIRES", "1").lower() in ("1", "true", "yes")
+ENABLE_EONET = os.getenv("INGEST_EONET", "1").lower() in ("1", "true", "yes")
+ENABLE_KEV = os.getenv("INGEST_KEV", "1").lower() in ("1", "true", "yes")

 NATS_STREAM = "events"

@ -62,6 +64,18 @@ async def producer_loop() -> None:
                    logger.info("USGS -> %d events", q)
                except Exception:  # noqa: BLE001
                    logger.exception("USGS fetch failed")
+            if ENABLE_EONET:
+                try:
+                    n = await ingest_eonet()
+                    logger.info("EONET -> %d events", n)
+                except Exception:  # noqa: BLE001
+                    logger.exception("EONET fetch failed")
+            if ENABLE_KEV:
+                try:
+                    k = await ingest_cisa_kev()
+                    logger.info("CISA KEV -> %d events", k)
+                except Exception:  # noqa: BLE001
+                    logger.exception("CISA KEV fetch failed")
        except Exception:  # noqa: BLE001
            logger.exception("producer cycle error")
        await asyncio.sleep(INTERVAL)
--- a/app/sources.py
+++ b/app/sources.py
@ -11,7 +11,7 @@ import httpx
 import feedparser
 import nats

-from config import NATS_URL
+from config import NATS_URL, OSINT_USER_AGENT
 from upstream_cache import rss_cache

 logger = logging.getLogger("osint.sources")
@ -26,26 +26,46 @@ def _parse_rfc822(date_str: object) -> str | None:
    except (ValueError, TypeError):
        return None

-# NATS connection
+
 NATS_URLS = NATS_URL
+_nc = None
+
+
+async def _jetstream():
+    """Reuse one NATS connection across publishes (no connect/close per event)."""
+    global _nc
+    if _nc is None or _nc.is_closed:
+        _nc = await nats.connect(NATS_URLS)
+    return _nc.jetstream()


 async def publish_event(subject: str, event: dict):
    """Publish an event to NATS JetStream."""
-    nc = await nats.connect(NATS_URLS)
-    js = nc.jetstream()
+    js = await _jetstream()
    await js.publish(subject, json.dumps(event).encode())
-    await nc.close()
    logger.debug("Published event to %s", subject)


+def event_dedup_key(msg: dict) -> str | None:
+    """Natural key for generic events. URL when present; else None (always insert)."""
+    url = msg.get("url")
+    if not isinstance(url, str):
+        return None
+    url = url.strip()
+    return url or None
+
+
+def _ua_headers() -> dict[str, str]:
+    return {"User-Agent": OSINT_USER_AGENT}
+
+
 # ─── RSS Feed Ingestor ──────────────────────────────────────────────────

 async def ingest_rss_feed(feed_url: str, source_id: str | None = None):
    """Fetch and parse an RSS feed, publish items to NATS."""
    text = rss_cache.get(feed_url)
    if text is None:
-        async with httpx.AsyncClient(timeout=30) as client:
+        async with httpx.AsyncClient(timeout=30, headers=_ua_headers()) as client:
            resp = await client.get(feed_url)
            resp.raise_for_status()
            text = resp.text
@ -76,51 +96,80 @@ async def ingest_rss_feed(feed_url: str, source_id: str | None = None):
    return count


-# ─── GDELT 2.0 Ingestor ─────────────────────────────────────────────────
+# ─── GDELT 2.0 DOC API ──────────────────────────────────────────────────

-GDELT_API = "https://api.gdeltproject.org/gdeltv2"
+GDELT_API = "https://api.gdeltproject.org/api/v2/doc/doc"
+GDELT_DEFAULT_QUERY = '(unrest OR protest OR outage OR cyber OR "power outage")'
+
+
+def gdelt_params(query: str = "", max_articles: int = 50) -> dict[str, str]:
+    """DOC 2.0 query string (not the retired gdeltv2 ``search`` param)."""
+    q = (query or "").strip() or GDELT_DEFAULT_QUERY
+    return {
+        "query": q,
+        "mode": "ArtList",
+        "format": "json",
+        "maxrecords": str(int(max_articles)),
+        "timespan": "1d",
+    }
+
+
+def _parse_gdelt_seendate(value: object) -> str:
+    if isinstance(value, str) and len(value) >= 15:
+        try:
+            return datetime.strptime(value[:15], "%Y%m%dT%H%M%S").replace(
+                tzinfo=timezone.utc
+            ).isoformat()
+        except ValueError:
+            pass
+    return datetime.now(timezone.utc).isoformat()
+
+
+def parse_gdelt_articles(data: dict) -> list[dict]:
+    events = []
+    for article in data.get("articles") or []:
+        if not isinstance(article, dict):
+            continue
+        url = article.get("url")
+        if not url:
+            continue
+        events.append({
+            "source_type": "gdel-t2",
+            "title": article.get("title"),
+            "body": article.get("domain") or article.get("language"),
+            "url": url,
+            "location_name": article.get("sourcecountry"),
+            "source_timestamp": _parse_gdelt_seendate(article.get("seendate")),
+            "tags": [t for t in (article.get("language"), article.get("sourcecountry")) if t],
+            "raw": article,
+        })
+    return events


 async def ingest_gdelt(query: str = "", max_articles: int = 50):
-    """Fetch articles from GDELT 2.0 API."""
-    params = {
-        "mode": "artlist",
-        "format": "json",
-        "maxrecords": max_articles,
-        "mode": "artlist",
-    }
-    if query:
-        params["search"] = query
-
-    async with httpx.AsyncClient(timeout=60) as client:
+    """Fetch articles from the GDELT DOC 2.0 API."""
+    params = gdelt_params(query=query, max_articles=max_articles)
+    data: dict = {"articles": []}
+    async with httpx.AsyncClient(
+        timeout=60, headers=_ua_headers(), follow_redirects=True,
+    ) as client:
+        try:
            resp = await client.get(GDELT_API, params=params)
            resp.raise_for_status()
            data = resp.json()
+        except (httpx.TransportError, httpx.HTTPStatusError) as exc:
+            # gdeltproject.org certs have expired in the wild; HTTP fallback.
+            logger.warning("GDELT HTTPS failed (%s); retrying HTTP", exc)
+            http_url = GDELT_API.replace("https://", "http://", 1)
+            resp = await client.get(http_url, params=params)
+            resp.raise_for_status()
+            data = resp.json()

-    count = 0
-    for article in data.get("articles", []):
-        event = {
-            "source_type": "gdel-t2",
-            "title": article.get("title"),
-            "body": article.get("articleBody"),
-            "url": article.get("url"),
-            "sentiment_score": _parse_gdelt_tone(article.get("Tone", "0")),
-            "location_lat": article.get("Latitude"),
-            "location_lon": article.get("Longitude"),
-            "location_name": article.get("Location"),
-            "source_timestamp": article.get("FirstCreated"),
-            "entities": [
-                {"name": e.get("Topic"), "type": "topic"}
-                for e in article.get("Mentions", [])
-                if e.get("Topic")
-            ],
-            "raw": article,
-        }
+    events = parse_gdelt_articles(data if isinstance(data, dict) else {})
+    for event in events:
        await publish_event("events.gdelt", event)
-        count += 1
-
-    logger.info("Ingested %d articles from GDELT", count)
-    return count
+    logger.info("Ingested %d articles from GDELT", len(events))
+    return len(events)


 def _parse_gdelt_tone(tone: str) -> float | None:
@ -137,22 +186,21 @@ def _parse_gdelt_tone(tone: str) -> float | None:
 USGS_API = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_hour.geojson"


-async def ingest_earthquakes():
-    """Fetch recent earthquakes from USGS."""
-    async with httpx.AsyncClient(timeout=30) as client:
-        resp = await client.get(USGS_API)
-        resp.raise_for_status()
-        data = resp.json()
-
-    count = 0
-    for feature in data.get("features", []):
-        props = feature.get("properties", {})
-        geometry = feature.get("geometry", {}).get("coordinates", [])
-        event = {
+def parse_usgs_feature(feature: dict) -> dict:
+    """Map one USGS GeoJSON feature, keeping the stable event id."""
+    props = feature.get("properties") or {}
+    geometry = (feature.get("geometry") or {}).get("coordinates") or []
+    usgs_id = feature.get("id")
+    url = props.get("url") or (
+        f"https://earthquake.usgs.gov/earthquakes/eventpage/{usgs_id}" if usgs_id else None
+    )
+    raw = dict(props)
+    raw["usgs_id"] = usgs_id
+    return {
        "source_type": "earthquake",
        "title": props.get("title"),
        "body": props.get("description"),
-            "url": props.get("url"),
+        "url": url,
        "location_lat": geometry[1] if len(geometry) > 1 else None,
        "location_lon": geometry[0] if len(geometry) > 0 else None,
        "location_name": props.get("place"),
@ -163,8 +211,20 @@ async def ingest_earthquakes():
            .replace(tzinfo=timezone.utc)
            .isoformat()
        ),
-            "raw": props,
+        "raw": raw,
    }
+
+
+async def ingest_earthquakes():
+    """Fetch recent earthquakes from USGS."""
+    async with httpx.AsyncClient(timeout=30, headers=_ua_headers()) as client:
+        resp = await client.get(USGS_API)
+        resp.raise_for_status()
+        data = resp.json()
+
+    count = 0
+    for feature in data.get("features", []):
+        event = parse_usgs_feature(feature)
        await publish_event("events.earthquake", event)
        count += 1

@ -172,6 +232,110 @@ async def ingest_earthquakes():
    return count


+# ─── NASA EONET v3 ──────────────────────────────────────────────────────
+
+EONET_API = "https://eonet.gsfc.nasa.gov/api/v3/events"
+
+
+def parse_eonet_events(payload: dict) -> list[dict]:
+    events = []
+    for item in payload.get("events") or []:
+        if not isinstance(item, dict):
+            continue
+        eid = item.get("id")
+        geoms = item.get("geometry") or []
+        point = None
+        for g in geoms:
+            if isinstance(g, dict) and g.get("type") == "Point":
+                point = g
+        if point is None:
+            continue
+        coords = point.get("coordinates") or []
+        if len(coords) < 2:
+            continue
+        lon, lat = float(coords[0]), float(coords[1])
+        cats = item.get("categories") or []
+        tags = []
+        for c in cats:
+            if isinstance(c, dict) and c.get("id"):
+                tags.append(str(c["id"]))
+        url = item.get("link") or (f"https://eonet.gsfc.nasa.gov/api/v3/events/{eid}" if eid else None)
+        ts = point.get("date") or datetime.now(timezone.utc).isoformat()
+        events.append({
+            "source_type": "disaster",
+            "title": item.get("title"),
+            "body": ", ".join(tags) if tags else None,
+            "url": url,
+            "location_lat": lat,
+            "location_lon": lon,
+            "location_name": item.get("title"),
+            "tags": tags,
+            "source_timestamp": ts,
+            "raw": {**item, "eonet_id": eid},
+        })
+    return events
+
+
+async def ingest_eonet():
+    """Volcanoes, storms, floods, drought — gaps USGS/FIRMS don't cover."""
+    async with httpx.AsyncClient(timeout=30, headers=_ua_headers()) as client:
+        resp = await client.get(EONET_API, params={"status": "open", "limit": 100})
+        resp.raise_for_status()
+        data = resp.json()
+    events = parse_eonet_events(data if isinstance(data, dict) else {})
+    for event in events:
+        await publish_event("events.disaster", event)
+    logger.info("Ingested %d EONET events", len(events))
+    return len(events)
+
+
+# ─── CISA KEV ───────────────────────────────────────────────────────────
+
+CISA_KEV_API = (
+    "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
+)
+
+
+def parse_cisa_kev(payload: dict) -> list[dict]:
+    events = []
+    for row in payload.get("vulnerabilities") or []:
+        if not isinstance(row, dict):
+            continue
+        cve = row.get("cveID")
+        if not cve:
+            continue
+        title = row.get("vulnerabilityName") or cve
+        vendor = row.get("vendorProject") or ""
+        product = row.get("product") or ""
+        events.append({
+            "source_type": "disaster",
+            "title": f"{cve}: {title}",
+            "body": row.get("shortDescription") or f"{vendor} {product}".strip(),
+            "url": f"https://nvd.nist.gov/vuln/detail/{cve}",
+            "location_lat": None,
+            "location_lon": None,
+            "location_name": None,
+            "tags": ["cisa-kev", cve, "ransomware" if row.get("knownRansomwareCampaignUse") == "Known" else None],
+            "source_timestamp": row.get("dateAdded") or datetime.now(timezone.utc).isoformat(),
+            "raw": {**row, "cveID": cve},
+        })
+        events[-1]["tags"] = [t for t in events[-1]["tags"] if t]
+    return events
+
+
+async def ingest_cisa_kev():
+    """Exploited-in-the-wild CVEs. No fake map coords — ticker/events only."""
+    async with httpx.AsyncClient(timeout=30, headers=_ua_headers(), follow_redirects=True) as client:
+        resp = await client.get(CISA_KEV_API)
+        resp.raise_for_status()
+        data = resp.json()
+    events = parse_cisa_kev(data if isinstance(data, dict) else {})
+    for event in events:
+        await publish_event("events.disaster", event)
+    logger.info("Ingested %d CISA KEV rows", len(events))
+    return len(events)
+
+
 # ─── Social Signals (Twitter/X-like placeholder) ────────────────────────

 async def ingest_social_signals(query: str = "", max_items: int = 50):
--- a/app/static/index.html
+++ b/app/static/index.html
@ -1149,9 +1149,12 @@ async function checkHealth() {
    try {
        const r = await fetch(`${API}/api/health`);
        const d = await r.json();
-        if (r.ok) {
+        if (r.ok && d && d.status === 'ok') {
            dot.className = 'dot ok';
            txt.textContent = 'SYSTEMS OK';
+        } else if (r.ok) {
+            dot.className = 'dot bad';
+            txt.textContent = 'DEGRADED';
        } else {
            dot.className = 'dot bad';
            txt.textContent = 'DEGRADED';
@ -1307,8 +1310,7 @@ function initMarketTicker() {
    loadMarketSymbols();
    buildMarketTrack(true);
    setMarketTag('standby');
-    probeMarket();
-    setInterval(probeMarket, MARKET_POLL_MS);
+    // GET /api/market is not wired — do not 404-poll every 15s.
 }

 /* ═══════════════ NEWS FEED + TICKER ═══════════════ */
@ -1813,6 +1815,7 @@ function bboxCell() {
    return currentBBox().split(',').map(n => Number(n).toFixed(2)).join(',') + '@' + map.getZoom();
 }
 let liveWs = null;
+let wsRetryMs = 1000;
 const FF_ICAO = new Set(['AT802','AT8T','AT8P','C130','C30J','C130J','DC10','MD10','MD87','S64','UH60','S70','CL415','CL215','B350','OV10','C208']);
 let firefighterHex = new Set();
 let dvrTs = null;
@ -1876,15 +1879,25 @@ function acColor(p) {
 }
 function connectLiveWs() {
    if (liveWs && (liveWs.readyState === 0 || liveWs.readyState === 1)) return;
-    try { liveWs = new WebSocket(liveWsUrl()); } catch (e) { return; }
-    liveWs.onopen = () => sendLiveViewport();
+    try { liveWs = new WebSocket(liveWsUrl()); } catch (e) {
+        const delay = wsRetryMs;
+        wsRetryMs = Math.min(wsRetryMs * 2, 60000);
+        setTimeout(connectLiveWs, delay);
+        return;
+    }
+    liveWs.onopen = () => { wsRetryMs = 1000; sendLiveViewport(); };
    liveWs.onmessage = (ev) => {
        try {
            const msg = JSON.parse(ev.data);
            applyLiveMarker(msg.type, msg.payload);
        } catch (e) { /* ignore malformed */ }
    };
-    liveWs.onclose = () => { liveWs = null; setTimeout(connectLiveWs, 4000); };
+    liveWs.onclose = () => {
+        liveWs = null;
+        const delay = wsRetryMs;
+        wsRetryMs = Math.min(wsRetryMs * 2, 60000);
+        setTimeout(connectLiveWs, delay);
+    };
 }
 function overlayFetch(url) {
    return fetch(url, overlayAbort ? { signal: overlayAbort.signal } : {});
@ -2022,12 +2035,16 @@ async function initMap() {
        trainsOn = document.getElementById('lp-trains-on').checked;
        vesselsOn = document.getElementById('lp-vessels-on').checked;
        stormsOn = document.getElementById('lp-storms-on').checked;
+        connectLiveWs();
+        requestAnimationFrame(() => {
            if (firesOn) loadFires();
+            setTimeout(() => {
                if (camsOn) loadCams();
                if (blipsOn) loadBlips();
                if (newsOn) loadNewsPins();
                refreshLiveOverlays();
-        connectLiveWs();
+            }, 250);
+        });
    } catch(e) {
        hint.textContent = `Failed to load map layers: ${e.message || e}`;
        console.error('Map init failed', e);
--- a/app/ws_manager.py
+++ b/app/ws_manager.py
@ -43,6 +43,9 @@ class ConnectionManager:
    def viewport_of(self, client_id: str) -> BBox | None:
        return self._viewports.get(client_id)

+    def viewports(self) -> list[BBox]:
+        return list(self._viewports.values())
+
    def has_clients(self) -> bool:
        return bool(self._queues)

--- a/deploy/osint-ws.nginx.conf
+++ b/deploy/osint-ws.nginx.conf
@ -0,0 +1,23 @@
+# osint.rpi.local — WebSocket upgrade for /ws/live
+#
+# GitOps: this file is the source of truth. On the Pi:
+#   sudo cp deploy/osint-ws.nginx.conf /etc/nginx/snippets/osint-ws.conf
+# then `include snippets/osint-ws.conf;` inside the osint.rpi.local server
+# block (before `location /`), `nginx -t && systemctl reload nginx`.
+#
+# Without these headers nginx proxies GET /ws/live as HTTP/1.0 → FastAPI 404
+# and the HUD reconnects every few seconds.
+
+location /ws/ {
+    proxy_pass http://127.0.0.1:8000;
+    proxy_http_version 1.1;
+    proxy_set_header Upgrade $http_upgrade;
+    proxy_set_header Connection "upgrade";
+    proxy_set_header Host $host;
+    proxy_set_header X-Real-IP $remote_addr;
+    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+    proxy_set_header X-Forwarded-Proto $scheme;
+    proxy_read_timeout 3600s;
+    proxy_send_timeout 3600s;
+    proxy_buffering off;
+}
--- a/news/scraper/newsScraper/feed_util.py
+++ b/news/scraper/newsScraper/feed_util.py
@ -0,0 +1,32 @@
+"""Feed helpers shared by the news spider (no Scrapy import)."""
+
+from __future__ import annotations
+
+import datetime
+from email.utils import parsedate_to_datetime
+from urllib.parse import urlparse
+
+AUDIO_EXT = (".mp3", ".m4a", ".ogg", ".wav", ".aac", ".flac", ".opus")
+
+
+def is_audio_url(url: str) -> bool:
+    path = urlparse(url or "").path.lower()
+    return any(path.endswith(ext) for ext in AUDIO_EXT)
+
+
+def article_timestamp(pub_date: str | None) -> datetime.datetime:
+    """Prefer the feed's pubDate/published; fall back to now (UTC)."""
+    if pub_date:
+        try:
+            return parsedate_to_datetime(pub_date).astimezone(datetime.timezone.utc)
+        except (TypeError, ValueError, IndexError):
+            pass
+        try:
+            raw = pub_date.replace("Z", "+00:00")
+            ts = datetime.datetime.fromisoformat(raw)
+            if ts.tzinfo is None:
+                ts = ts.replace(tzinfo=datetime.timezone.utc)
+            return ts
+        except ValueError:
+            pass
+    return datetime.datetime.now(datetime.timezone.utc)
--- a/news/scraper/newsScraper/pipelines.py
+++ b/news/scraper/newsScraper/pipelines.py
@ -53,8 +53,10 @@ class PostgresPipeline:
        self.connection.commit()

    def process_item(self, item, spider):
-        if item ['url'] in self.seen_urls:
-            raise DropItem()
+        url = item['url']
+        if url in self.seen_urls:
+            raise DropItem(f"Duplicate URL (in-memory): {url}")
+        self.seen_urls.add(url)
        try:
            self.cur.execute("""
                INSERT INTO articles (title, url, content, domain, timestamp)
@ -68,11 +70,12 @@ class PostgresPipeline:
                item['timestamp']
            ))
            if self.cur.rowcount == 0:
-                e = DropItem("Duplicate URL (database conflict)")
-                e.log_level = logging.DEBUG
-                raise e 
+                raise DropItem(f"Duplicate URL (database): {url}")
            self.connection.commit()
            return item
+        except DropItem:
+            self.connection.rollback()
+            raise
        except Exception as e:
            spider.logger.error(f"Error saving to Postgres: {e}")
            self.connection.rollback()
@ -88,5 +91,3 @@ from itemadapter import ItemAdapter
 class NewsscraperPipeline:
    def process_item(self, item, spider):
        return item
-
-
--- a/news/scraper/newsScraper/spiders/news_spider.py
+++ b/news/scraper/newsScraper/spiders/news_spider.py
@ -4,6 +4,8 @@ from urllib.parse import urljoin, urlparse
 import datetime
 import re

+from newsScraper.feed_util import article_timestamp, is_audio_url
+

 class NewsRSSSpider(Spider):
    """Crawl the curated news sources in urls.txt and extract articles.
@ -87,6 +89,8 @@ class NewsRSSSpider(Spider):
                or node.xpath('updated/text()').get()
            )
            if link:
+                if is_audio_url(link):
+                    continue
                yield scrapy.Request(
                    link,
                    callback=self.parse_article,
@ -112,5 +116,5 @@ class NewsRSSSpider(Spider):
                'url': response.url,
                'text': pure_text,
                'domain': urlparse(response.url).netloc,
-                'timestamp': datetime.datetime.now().isoformat()
+                'timestamp': article_timestamp(response.meta.get('date')).isoformat()
            }
--- a/news/summerizer/Dockerfile
+++ b/news/summerizer/Dockerfile
@ -13,7 +13,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
 COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt

-COPY summarizer.py run_news_summarizer.py ./
+COPY summarizer.py run_news_summarizer.py intel.py nous_client.py ./

 # Security: run as a non-privileged user.
 RUN useradd -m summarizer_user
--- a/pytest.ini
+++ b/pytest.ini
@ -0,0 +1,3 @@
+[pytest]
+testpaths = tests
+python_files = test_*.py
--- a/tests/test_cache_and_timeline.py
+++ b/tests/test_cache_and_timeline.py
@ -0,0 +1,21 @@
+"""Timeline bucket_hours + static Cache-Control."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+def test_timeline_uses_bucket_hours():
+    src = (ROOT / "app/main.py").read_text()
+    fn = src.split("async def get_timeline")[1].split("async def sentiment_by_source")[0]
+    assert "bucket_hours" in fn
+    assert "date_trunc('hour'" not in fn or "bucket" in fn.lower()
+    # Must not ignore the query param.
+    assert ":bucket" in fn or "bucket_hours" in fn.split("text(")[1][:800]
+
+
+def test_static_vendor_cache_control():
+    src = (ROOT / "app/main.py").read_text()
+    assert "max-age=31536000" in src or "immutable" in src.lower()
--- a/tests/test_event_ingest.py
+++ b/tests/test_event_ingest.py
@ -0,0 +1,133 @@
+"""Generic event ingest: idempotency, USGS ids, GDELT DOC URL."""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timezone
+
+
+def test_event_dedup_key_prefers_url():
+    from sources import event_dedup_key
+
+    assert event_dedup_key({"url": "https://earthquake.usgs.gov/earthquakes/eventpage/ci1"}) == (
+        "https://earthquake.usgs.gov/earthquakes/eventpage/ci1"
+    )
+    assert event_dedup_key({"url": "  "}) is None
+    assert event_dedup_key({}) is None
+
+
+def test_usgs_feature_keeps_id_and_url():
+    from sources import parse_usgs_feature
+
+    feature = {
+        "id": "ci39818991",
+        "properties": {
+            "title": "M 2.1 - 5 km W of",
+            "url": "https://earthquake.usgs.gov/earthquakes/eventpage/ci39818991",
+            "place": "5 km W of",
+            "mag": 2.1,
+            "time": 1_700_000_000_000,
+        },
+        "geometry": {"coordinates": [-118.5, 34.1, 10.0]},
+    }
+    event = parse_usgs_feature(feature)
+    assert event["url"] == "https://earthquake.usgs.gov/earthquakes/eventpage/ci39818991"
+    assert event["raw"]["usgs_id"] == "ci39818991"
+    assert event["location_lat"] == 34.1
+    assert event["location_lon"] == -118.5
+    assert event["source_type"] == "earthquake"
+
+
+def test_gdelt_uses_doc_api_and_query_param():
+    from sources import GDELT_API, gdelt_params
+
+    assert GDELT_API == "https://api.gdeltproject.org/api/v2/doc/doc"
+    params = gdelt_params(query="unrest", max_articles=50)
+    assert params["query"] == "unrest"
+    assert "search" not in params
+    assert params["mode"] == "ArtList"
+    assert params["format"] == "json"
+    assert int(params["maxrecords"]) == 50
+
+
+def test_gdelt_default_query_when_empty():
+    from sources import gdelt_params
+
+    params = gdelt_params(query="", max_articles=25)
+    assert params["query"]
+    assert "unrest" in params["query"].lower() or "cyber" in params["query"].lower()
+
+
+def test_parse_gdelt_articles_maps_doc_payload():
+    from sources import parse_gdelt_articles
+
+    payload = {
+        "articles": [
+            {
+                "url": "https://example.com/a",
+                "title": "Outage",
+                "seendate": "20240101T120000Z",
+                "domain": "example.com",
+                "language": "English",
+                "sourcecountry": "US",
+            }
+        ]
+    }
+    events = parse_gdelt_articles(payload)
+    assert len(events) == 1
+    assert events[0]["source_type"] == "gdel-t2"
+    assert events[0]["url"] == "https://example.com/a"
+    assert events[0]["title"] == "Outage"
+
+
+def test_ingest_event_skips_duplicate_url(monkeypatch):
+    """Second insert with the same url must not hit events_table.insert."""
+    from ingestor import ingest_event
+
+    calls = {"insert": 0, "dedup": 0}
+
+    class _Result:
+        rowcount = 1
+        inserted_primary_key = ["evt-1"]
+
+    class _Session:
+        async def execute(self, stmt):
+            sql = str(stmt).lower()
+            if "event_dedup" in sql or "on conflict" in sql:
+                calls["dedup"] += 1
+                self_result = _Result()
+                if calls["dedup"] > 1:
+                    self_result.rowcount = 0
+                return self_result
+            calls["insert"] += 1
+            return _Result()
+
+        async def commit(self):
+            return None
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *a):
+            return False
+
+    import ingestor
+
+    monkeypatch.setattr(ingestor, "async_session", lambda: _Session())
+
+    msg = {
+        "source_type": "earthquake",
+        "title": "M 2.1",
+        "url": "https://earthquake.usgs.gov/earthquakes/eventpage/ci1",
+        "source_timestamp": datetime(2026, 1, 1, tzinfo=timezone.utc).isoformat(),
+    }
+
+    async def run():
+        first = await ingest_event(msg)
+        second = await ingest_event(msg)
+        return first, second
+
+    first, second = asyncio.run(run())
+    assert first is not None
+    assert second is None
+    assert calls["insert"] == 1
--- a/tests/test_fire_ingest.py
+++ b/tests/test_fire_ingest.py
@ -169,3 +169,47 @@ def test_ingest_fire_row_correlates_from_hypertable_when_last_known_empty(monkey
    assert len(correlated) == 1
    assert correlated[0][1][0]["id"] == "acf001"
    assert correlated[0][0][0]["lat"] == 39.45678
+
+
+def test_ingest_fire_rows_one_execute_one_commit(monkeypatch):
+    """93k FIRMS points must not be 93k commits."""
+    from ingestor import ingest_fire_rows
+
+    class _Session:
+        def __init__(self):
+            self.executes = 0
+            self.commits = 0
+            self.rowcount = 3
+
+        async def execute(self, *a, **k):
+            self.executes += 1
+            return self
+
+        async def commit(self):
+            self.commits += 1
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *a):
+            return False
+
+    session = _Session()
+    import ingestor
+    monkeypatch.setattr(ingestor, "async_session", lambda: session)
+
+    msgs = [
+        make_fire_msg(latitude=39.1 + i * 0.01, longitude=-121.1)
+        for i in range(3)
+    ]
+
+    async def no_corr(*a, **k):
+        return []
+
+    monkeypatch.setattr("fire_aircraft.correlate_and_notify", no_corr)
+    monkeypatch.setattr("geofence.record_and_notify", no_corr)
+
+    inserted = asyncio.run(ingest_fire_rows(msgs))
+    assert inserted == 3
+    assert session.executes == 1
+    assert session.commits == 1
--- a/tests/test_fire_sources.py
+++ b/tests/test_fire_sources.py
@ -123,7 +123,7 @@ def test_ingest_fires_uses_keystore_key(monkeypatch):
        published.extend(points)
        return len(points)

-    monkeypatch.setattr("fire_sources.publish_fire_batch", fake_publish)
+    monkeypatch.setattr("fire_sources.persist_hotspots", fake_publish)

    assert asyncio.run(ingest_fires()) == 10  # NOAA-20 + NOAA-21 dual-write
    assert "a" * 32 in captured["url"]
--- a/tests/test_frontend_reliability.py
+++ b/tests/test_frontend_reliability.py
@ -0,0 +1,44 @@
+"""HUD load-time: no market 404 poll, deferred overlays, WS backoff, nginx snippet."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+HTML = (ROOT / "app/static/index.html").read_text()
+
+
+def test_summarizer_dockerfile_copies_intel_modules():
+    df = (ROOT / "news/summerizer/Dockerfile").read_text()
+    assert "intel.py" in df
+    assert "nous_client.py" in df
+
+
+def test_nginx_ws_snippet_has_upgrade_headers():
+    conf = (ROOT / "deploy/osint-ws.nginx.conf").read_text()
+    assert "proxy_http_version 1.1" in conf
+    assert "Upgrade" in conf
+    assert "Connection" in conf
+    assert "/ws/" in conf
+
+
+def test_market_ticker_does_not_poll_unwired_endpoint():
+    assert "setInterval(probeMarket" not in HTML
+    assert "initMarketTicker()" not in HTML or "probeMarket();" not in HTML.split("function initMarketTicker")[1][:400]
+
+
+def test_startup_defers_nonessential_overlays():
+    init = HTML.split("function initMap")[1].split("function readMapPrefs")[0]
+    # Must not fire all four DB loads + live overlays in the same tick.
+    assert "setTimeout" in init or "requestAnimationFrame" in init
+
+
+def test_ws_reconnect_uses_backoff():
+    assert "setTimeout(connectLiveWs, 4000)" not in HTML
+    ws = HTML.split("function connectLiveWs")[1][:1200]
+    assert "backoff" in ws.lower() or "wsRetry" in ws or "wsDelay" in ws
+
+
+def test_check_health_treats_degraded_status():
+    fn = HTML.split("async function checkHealth")[1].split("/* ═══════════════ NAV")[0]
+    assert "degraded" in fn.lower() or "d.status" in fn
--- a/tests/test_health_ready.py
+++ b/tests/test_health_ready.py
@ -0,0 +1,33 @@
+"""Liveness stays up; readiness/freshness is explicit."""
+
+from __future__ import annotations
+
+import asyncio
+
+import httpx
+
+from main import app
+
+BASE = "http://test"
+
+
+async def _get(path: str) -> httpx.Response:
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
+        return await client.get(path)
+
+
+def test_health_includes_checks_even_when_db_ok(monkeypatch):
+    """HUD must be able to show degraded without docker killing the container."""
+    body = asyncio.run(_get("/api/health")).json()
+    assert "status" in body
+    assert "checks" in body
+    assert "db" in body["checks"]
+
+
+def test_ready_endpoint_exists():
+    resp = asyncio.run(_get("/api/ready"))
+    assert resp.status_code in (200, 503)
+    body = resp.json()
+    assert "checks" in body
+    assert "status" in body
--- a/tests/test_live_layers.py
+++ b/tests/test_live_layers.py
@ -472,3 +472,103 @@ def test_transform_ais_position_decodes_navstat():
    assert row is not None
    assert row["extra"]["nav"] == "moored"
    assert row["extra"]["navstat"] == 5
+
+
+def test_nws_alerts_does_not_send_bbox_param(monkeypatch):
+    """api.weather.gov/alerts/active 400s on bbox — clip locally instead."""
+    import asyncio
+
+    from live_layers import fetch_weather_alerts, _cache
+
+    seen = []
+
+    async def fake_get(url, params=None):
+        seen.append((url, dict(params or {})))
+        if "weather.gov" in url:
+            return {
+                "type": "FeatureCollection",
+                "features": [{
+                    "type": "Feature",
+                    "properties": {"event": "Tornado Warning", "severity": "Extreme"},
+                    "geometry": {"type": "Point", "coordinates": [-78.7, 35.8]},
+                }],
+            }
+        return {"type": "FeatureCollection", "features": []}
+
+    monkeypatch.setattr("live_layers._get_json", fake_get)
+    _cache.clear()
+    fc = asyncio.run(fetch_weather_alerts(None, "-79.0,35.5,-78.0,36.0"))
+    nws_calls = [p for u, p in seen if "weather.gov" in u]
+    assert nws_calls, "NWS should still be fetched"
+    assert "bbox" not in nws_calls[0]
+    assert fc.get("nws_ok") is True
+    assert len(fc["features"]) == 1
+
+
+def test_nws_alerts_failure_is_flagged(monkeypatch):
+    import asyncio
+
+    from live_layers import fetch_weather_alerts, _cache
+
+    async def fake_get(url, params=None):
+        if "weather.gov" in url:
+            raise RuntimeError("400 Bad Request")
+        return {"type": "FeatureCollection", "features": []}
+
+    monkeypatch.setattr("live_layers._get_json", fake_get)
+    _cache.clear()
+    fc = asyncio.run(fetch_weather_alerts(None, None))
+    assert fc.get("nws_ok") is False
+
+
+def test_fetch_aircraft_get_path_does_not_persist(monkeypatch):
+    """GET /api/aircraft must serve last-known without track/geofence writes."""
+    import asyncio
+
+    from live_layers import (
+        aircraft_last_known, fetch_aircraft, persist_aircraft_snapshot, _cache,
+    )
+
+    aircraft_last_known.clear()
+    aircraft_last_known["abc"] = {
+        "id": "abc", "lat": 35.8, "lon": -78.7, "heading": 90, "speed": 400,
+        "label": "ABC", "extra": {},
+    }
+    writes = {"n": 0}
+
+    async def boom(*a, **k):
+        writes["n"] += 1
+        raise AssertionError("GET path must not persist")
+
+    monkeypatch.setattr("tracks.record_position", boom)
+    monkeypatch.setattr("geofence.record_and_notify", boom)
+    _cache.clear()
+    rows = asyncio.run(fetch_aircraft("-79,35,-78,36", persist=False))
+    assert writes["n"] == 0
+    assert any(r["id"] == "abc" for r in rows)
+
+
+def test_persist_aircraft_snapshot_writes_tracks(monkeypatch):
+    import asyncio
+
+    from live_layers import persist_aircraft_snapshot
+
+    recorded = []
+
+    async def fake_record(kind, marker):
+        recorded.append((kind, marker["id"]))
+        return True
+
+    async def fake_gf(**kw):
+        return 0
+
+    monkeypatch.setattr("tracks.record_position", fake_record)
+    monkeypatch.setattr("geofence.record_and_notify", fake_gf)
+    monkeypatch.setattr("ws_manager.manager.has_clients", lambda: False)
+
+    rows = [{
+        "id": "abc", "lat": 35.8, "lon": -78.7, "heading": 90, "speed": 400,
+        "label": "ABC", "extra": {},
+    }]
+    asyncio.run(persist_aircraft_snapshot(rows))
+    assert recorded == [("aircraft", "abc")]
--- a/tests/test_new_sources.py
+++ b/tests/test_new_sources.py
@ -0,0 +1,66 @@
+"""NASA EONET + CISA KEV parsers (no network)."""
+
+from __future__ import annotations
+
+
+def test_parse_eonet_keeps_stable_ids_and_points():
+    from sources import parse_eonet_events
+
+    payload = {
+        "events": [
+            {
+                "id": "EONET_6363",
+                "title": "Etna Volcano",
+                "categories": [{"id": "volcanoes", "title": "Volcanoes"}],
+                "geometry": [
+                    {"date": "2024-01-01T00:00:00Z", "type": "Point", "coordinates": [15.0, 37.7]},
+                ],
+                "link": "https://eonet.gsfc.nasa.gov/api/v3/events/EONET_6363",
+            },
+            {
+                "id": "EONET_skip",
+                "title": "No geometry",
+                "categories": [],
+                "geometry": [],
+                "link": "https://eonet.gsfc.nasa.gov/api/v3/events/EONET_skip",
+            },
+        ]
+    }
+    events = parse_eonet_events(payload)
+    assert len(events) == 1
+    ev = events[0]
+    assert ev["url"] == "https://eonet.gsfc.nasa.gov/api/v3/events/EONET_6363"
+    assert ev["source_type"] == "disaster"
+    assert ev["location_lat"] == 37.7
+    assert ev["location_lon"] == 15.0
+    assert ev["raw"]["eonet_id"] == "EONET_6363"
+    assert "volcanoes" in ev["tags"]
+
+
+def test_parse_cisa_kev_emits_cve_url_no_coords():
+    from sources import parse_cisa_kev
+
+    payload = {
+        "vulnerabilities": [
+            {
+                "cveID": "CVE-2024-1234",
+                "vendorProject": "Acme",
+                "product": "Widget",
+                "vulnerabilityName": "RCE",
+                "dateAdded": "2024-06-01",
+                "shortDescription": "Remote code execution",
+                "requiredAction": "Apply updates",
+                "dueDate": "2024-06-22",
+                "knownRansomwareCampaignUse": "Known",
+            }
+        ]
+    }
+    events = parse_cisa_kev(payload)
+    assert len(events) == 1
+    ev = events[0]
+    assert ev["url"] == "https://nvd.nist.gov/vuln/detail/CVE-2024-1234"
+    assert ev["location_lat"] is None
+    assert ev["location_lon"] is None
+    assert "cisa-kev" in ev["tags"]
+    assert "CVE-2024-1234" in ev["tags"]
+    assert ev["raw"]["cveID"] == "CVE-2024-1234"
--- a/tests/test_news_scraper.py
+++ b/tests/test_news_scraper.py
@ -0,0 +1,42 @@
+"""News spider/pipeline: skip audio, use pubDate, don't log dupes as errors."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+_SCRAPER = ROOT / "news/scraper"
+if str(_SCRAPER) not in sys.path:
+    sys.path.insert(0, str(_SCRAPER))
+
+
+def test_is_audio_enclosure():
+    from newsScraper.feed_util import is_audio_url
+
+    assert is_audio_url("https://cdn.example/podcast.mp3") is True
+    assert is_audio_url("https://cdn.example/show.m4a?x=1") is True
+    assert is_audio_url("https://www.example.com/world/story") is False
+
+
+def test_article_timestamp_prefers_pubdate():
+    from newsScraper.feed_util import article_timestamp
+
+    ts = article_timestamp("Tue, 01 Apr 2025 12:00:00 GMT")
+    assert ts.tzinfo is not None
+    assert ts.year == 2025
+    assert ts.month == 4
+    assert ts.day == 1
+
+
+def test_pipeline_does_not_wrap_dropitem_as_error():
+    src = (ROOT / "news/scraper/newsScraper/pipelines.py").read_text()
+    assert "except DropItem" in src
+    assert "seen_urls.add" in src or "self.seen_urls.add" in src
+
+
+def test_spider_skips_audio_before_request():
+    src = (ROOT / "news/scraper/newsScraper/spiders/news_spider.py").read_text()
+    assert "is_audio_url" in src
+    assert "article_timestamp" in src
+    assert "datetime.datetime.now()" not in src
--- a/tests/test_upstream_cache.py
+++ b/tests/test_upstream_cache.py
@ -55,7 +55,7 @@ def test_ingest_fires_hits_http_once_within_ttl(monkeypatch):
    async def fake_publish(points):
        return len(points)

-    monkeypatch.setattr("fire_sources.publish_fire_batch", fake_publish)
+    monkeypatch.setattr("fire_sources.persist_hotspots", fake_publish)

    assert asyncio.run(ingest_fires()) == 5
    assert asyncio.run(ingest_fires()) == 5