fix(desktop-build): strip layout.tsx force-dynamic on CRLF checkouts too

build-frontend-export.cjs stages a desktop-only frontend export tree and
strips the ``force-dynamic`` + ``revalidate`` directives from
``frontend/src/app/layout.tsx`` so Next's ``output: "export"`` can
prerender every route.

The strip regexes only matched LF (``\n``). Any Windows checkout without
``core.autocrlf=input`` has CRLF line endings, the strip silently
no-op'd, and the desktop build failed at the static-export step:

    Error: Page with `dynamic = "force-dynamic"` couldn't be exported.
    `output: "export"` requires all pages be renderable statically
    because there is no runtime server to dynamically render routes
    in this output format.
    Export encountered an error on /_not-found/page: /_not-found

Reaches every Windows contributor who hasn't normalized line endings
locally. Replacing each ``\n`` in the strip regexes with ``\r?\n``
makes the strip CRLF-tolerant; LF behavior is unchanged.

Verified by running both regexes against the actual layout.tsx (302
bytes removed, force-dynamic + revalidate both gone) and against a
synthetic LF input (296 bytes removed, same outcome).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

backend/services/constants.py

@@ -11,8 +11,13 @@ DEFAULT_TRAIL_TTL_S = 300  # 5 min - trail TTL for non-tracked flights
 HOLD_PATTERN_DEGREES = 300  # Total heading change to flag holding pattern
 GPS_JAMMING_NACP_THRESHOLD = 8  # NACp below this = degraded GPS signal
 GPS_JAMMING_GRID_SIZE = 1.0  # 1 degree grid for aggregation
-GPS_JAMMING_MIN_RATIO = 0.30  # 30% degraded aircraft to flag zone
-GPS_JAMMING_MIN_AIRCRAFT = 5  # Min aircraft in grid cell for statistical significance
+# Tuned 2026-05: previously 0.30 / 5 aircraft which — combined with the
+# -1 noise cushion in the detector AND the pre-fix nac_p==0 filter that
+# discarded jamming victims — meant the layer almost never lit up.
+# Lowering the bar so genuine jamming zones with sparser ADS-B coverage
+# clear (eastern Med, Russia/Ukraine border, Iran/Iraq).
+GPS_JAMMING_MIN_RATIO = 0.20  # 20% degraded aircraft to flag zone
+GPS_JAMMING_MIN_AIRCRAFT = 3  # Min aircraft in grid cell for statistical significance
 
 # ─── Network & Circuit Breaker ──────────────────────────────────────────────
 CIRCUIT_BREAKER_TTL_S = 120  # Skip domain for 2 min after total failure

backend/services/data_fetcher.py

@@ -777,6 +777,19 @@ def start_scheduler():
         misfire_grace_time=60,
     )
 
+    # Flight observation pruning — drops icao24 → first_seen_at entries we
+    # haven't seen in an hour. Same cadence as AIS prune for symmetry; the
+    # per-tick scan is O(in-flight aircraft) so it's cheap.
+    from services.fetchers.flight_observations import prune as _prune_flight_observations
+    _scheduler.add_job(
+        lambda: _run_task_with_health(_prune_flight_observations, "prune_flight_observations"),
+        "interval",
+        minutes=5,
+        id="flight_observation_prune",
+        max_instances=1,
+        misfire_grace_time=60,
+    )
+
     # AISHub REST fallback — slow polling when the AISStream WebSocket
     # primary is offline. Configurable interval via
     # AISHUB_POLL_INTERVAL_MINUTES env (default 20 min). Operator must
@@ -980,16 +993,19 @@ def start_scheduler():
         misfire_grace_time=600,
     )
 
-    # UAP sightings (NUFORC) — daily at 12:00 UTC
+    # UAP sightings (NUFORC) — weekly on Mondays at 12:00 UTC. The layer is a
+    # rolling last-60-days digest; refreshing once a week is enough cadence
+    # for human-readable map exploration and keeps load on nuforc.org light.
     _scheduler.add_job(
         lambda: _run_task_with_health(
             lambda: fetch_uap_sightings(force_refresh=True),
             "fetch_uap_sightings",
         ),
         "cron",
+        day_of_week="mon",
         hour=12,
         minute=0,
-        id="uap_sightings_daily",
+        id="uap_sightings_weekly",
         max_instances=1,
         misfire_grace_time=3600,
     )

backend/services/fetchers/earth_observation.py

@@ -1383,10 +1383,21 @@ def _build_uap_sightings_from_hf_mirror() -> list[dict]:
     This is a resilience fallback for local/Windows runs where nuforc.org is
     Cloudflare-gated and the Mapbox token is not configured. It is not as fresh
     as the live NUFORC AJAX feed, but it keeps the layer visible and cached.
+
+    Date-cutoff guard: the kcimc/NUFORC HF dataset is a static snapshot whose
+    maintainer refreshes it sporadically. Without a cutoff, sorting by
+    occurred-desc and taking the top N rows returns whatever the mirror's
+    newest rows happen to be — which can be years old if the snapshot is
+    stale. We apply the same ``_NUFORC_RECENT_DAYS`` window the live path
+    uses (60 days). If the HF mirror has nothing inside the window we return
+    ``[]`` rather than silently serving 3-year-old "newest" rows.
     """
     from services.fetchers.nuforc_enrichment import _HF_CSV_URL, _parse_date
     from services.geocode_validate import coord_in_country
 
+    cutoff_dt = datetime.utcnow() - timedelta(days=_NUFORC_RECENT_DAYS)
+    cutoff_str = cutoff_dt.strftime("%Y-%m-%d")
+
     try:
         response = fetch_with_curl(_HF_CSV_URL, timeout=180, follow_redirects=True)
         if not response or response.status_code != 200:
@@ -1400,6 +1411,7 @@ def _build_uap_sightings_from_hf_mirror() -> list[dict]:
         return []
 
     candidates: list[dict] = []
+    stale_rows_dropped = 0
     try:
         reader = csv.DictReader(io.StringIO(response.text))
         for row in reader:
@@ -1410,6 +1422,9 @@ def _build_uap_sightings_from_hf_mirror() -> list[dict]:
             )
             if not occurred:
                 continue
+            if occurred < cutoff_str:
+                stale_rows_dropped += 1
+                continue
             raw_location = _normalize_uap_location(
                 row.get("Location", "")
                 or row.get("City", "")
@@ -1444,6 +1459,19 @@ def _build_uap_sightings_from_hf_mirror() -> list[dict]:
         logger.warning("UAP sightings: HF fallback parse failed: %s", e)
         return []
 
+    if not candidates:
+        # HF mirror returned rows, but none inside the rolling window. This is
+        # the smoking gun for "the public HF dataset hasn't been refreshed in
+        # years" — log loudly so the operator sees it instead of guessing.
+        logger.error(
+            "UAP sightings: HF fallback yielded 0 rows within last %d days "
+            "(dropped %d stale rows). HF mirror is likely stale; the layer "
+            "will be empty until the live NUFORC path recovers.",
+            _NUFORC_RECENT_DAYS,
+            stale_rows_dropped,
+        )
+        return []
+
     candidates.sort(key=lambda row: (row["occurred"], row["posted"], row["id"]), reverse=True)
     candidates = candidates[:_NUFORC_HF_FALLBACK_LIMIT]
 
@@ -1515,13 +1543,29 @@ def fetch_uap_sightings(*, force_refresh: bool = False):
 
     sightings = _load_nuforc_sightings_cache(force_refresh=force_refresh)
     if sightings is None:
+        live_error: Exception | None = None
         try:
             sightings = _build_recent_uap_sightings()
         except Exception as e:
+            live_error = e
             logger.warning("UAP sightings: live NUFORC rebuild failed, using fallback: %s", e)
             sightings = _build_uap_sightings_from_hf_mirror()
         if sightings:
             _save_nuforc_sightings_cache(sightings)
+        elif live_error is not None:
+            # Both paths failed: live raised AND HF fallback returned empty
+            # (either the HF mirror is stale beyond the cutoff or the network
+            # is gone entirely). The previous code silently set the layer to
+            # ``[]`` and kept marking it fresh; that masked the failure for
+            # days. Surface it via assert_canary so the health registry shows
+            # the layer as broken instead of "fresh and empty".
+            from services.slo import assert_canary
+            assert_canary("uap_sightings", 0)
+            logger.error(
+                "UAP sightings: both live NUFORC and HF fallback produced 0 "
+                "rows; layer is unavailable. Live error: %s",
+                live_error,
+            )
 
     with _data_lock:
         latest_data["uap_sightings"] = sightings or []

backend/services/fetchers/flight_observations.py

@@ -0,0 +1,148 @@
+"""Per-aircraft observation tracking for cumulative fuel/CO2 estimates.
+
+Background
+----------
+The pre-existing emissions enrichment attached a *rate* to each flight
+(GPH and kg/hr) based on aircraft model. Users — reasonably — wanted the
+running total: how much fuel HAS this plane burned since we started
+seeing it? Multiplying the rate by elapsed observation time gets us
+there, but it requires somewhere to remember "when did this icao24
+first appear on our radar?"
+
+Why this lives outside ``flight_trails``
+----------------------------------------
+``flight_trails`` is sized and pruned aggressively for map rendering
+(5-minute TTL for untracked aircraft, 200 trail points max). That's
+wrong for cumulative burn: if a plane has been airborne 2 hours but
+its trail was pruned 30 min in, the "first trail point" timestamp is
+30 min ago, not 2h ago. Worse, when the trail expires and re-creates,
+the cumulative counter would reset mid-flight.
+
+This module tracks observation lifecycle separately:
+
+* When a hex is first observed: start a new flight session.
+* While observed regularly (gap < ``REOPEN_GAP_S``): keep accumulating.
+* When unseen for longer than ``REOPEN_GAP_S``: treat next sighting as
+  a new session (the plane landed and took off again, or it's a
+  different leg). Reset ``first_seen_at``.
+* Stale sessions are pruned every ``PRUNE_INTERVAL_S`` so memory stays
+  bounded.
+
+The user explicitly asked for this counting semantic: "as soon as a
+plane appears there should be a counter that keeps a running count of
+the fuel being burned... If there is no estimate take off time then it
+can just be from the time the server starts to keep a log of whats in
+the air."
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+
+
+# Gap between sightings that resets the session. ADS-B refreshes the
+# whole aircraft list every minute or two, so anything over a few
+# minutes means the plane left our coverage window (landed, transit
+# through dead zone, etc). 15 minutes is conservative.
+REOPEN_GAP_S = 15 * 60
+
+# Don't accumulate runaway memory: drop entries unseen for an hour.
+PRUNE_AFTER_S = 60 * 60
+
+# Cap on accumulated airtime per session so a single bug elsewhere
+# (e.g. ts clock skew) can't produce comically large numbers.
+MAX_SESSION_SECONDS = 24 * 3600  # 24h — longest realistic civilian leg
+
+
+_observations: dict[str, dict[str, float]] = {}
+_lock = threading.Lock()
+_last_prune_at = 0.0
+
+
+def record_observation(icao_hex: str, *, now: float | None = None) -> int:
+    """Record a sighting of ``icao_hex`` and return airtime so far (seconds).
+
+    Returns 0 for the first-ever sighting (no elapsed time yet) or when
+    ``icao_hex`` is falsy. The caller can multiply the returned seconds
+    by ``rate_per_hour / 3600`` to get cumulative consumption.
+    """
+    if not icao_hex:
+        return 0
+    key = str(icao_hex).strip().lower()
+    if not key:
+        return 0
+    current = float(now if now is not None else time.time())
+
+    with _lock:
+        entry = _observations.get(key)
+        if entry is None:
+            _observations[key] = {"first_seen_at": current, "last_seen_at": current}
+            return 0
+        # Use explicit ``is None`` checks instead of ``or`` short-circuit:
+        # ``0.0`` is a legitimate timestamp value (e.g. test fixtures
+        # seeding a far-past first_seen_at to exercise the clamp) but
+        # ``0.0 or fallback`` collapses to ``fallback`` because 0.0 is
+        # falsy. Bit me on my own test — leaving the safer form here.
+        last_raw = entry.get("last_seen_at")
+        last_seen = float(last_raw) if last_raw is not None else current
+        gap = current - last_seen
+        if gap > REOPEN_GAP_S:
+            # Treat as a new flight session — the plane landed/disappeared
+            # long enough that the prior cumulative count is no longer
+            # the same flight.
+            _observations[key] = {"first_seen_at": current, "last_seen_at": current}
+            return 0
+        first_raw = entry.get("first_seen_at")
+        first = float(first_raw) if first_raw is not None else current
+        # Clamp absurd values from clock skew or bad input.
+        elapsed = max(0, min(int(current - first), MAX_SESSION_SECONDS))
+        entry["last_seen_at"] = current
+        return elapsed
+
+
+def prune(*, now: float | None = None) -> int:
+    """Drop entries we haven't seen in ``PRUNE_AFTER_S`` seconds.
+
+    Returns number of entries dropped. Safe to call from a scheduler tick;
+    cheap (single dict scan) so cadence doesn't matter much.
+    """
+    current = float(now if now is not None else time.time())
+    dropped = 0
+    with _lock:
+        stale_keys = []
+        for k, v in _observations.items():
+            last_raw = v.get("last_seen_at")
+            last = float(last_raw) if last_raw is not None else 0.0
+            if current - last > PRUNE_AFTER_S:
+                stale_keys.append(k)
+        for k in stale_keys:
+            del _observations[k]
+            dropped += 1
+    return dropped
+
+
+def get_session_seconds(icao_hex: str, *, now: float | None = None) -> int:
+    """Read-only accessor: airtime for a known icao without bumping last-seen.
+
+    Used by tests and external consumers (e.g. when rendering a snapshot
+    of all in-flight aircraft, you want the current value, not to update
+    last_seen_at as a side effect).
+    """
+    if not icao_hex:
+        return 0
+    key = str(icao_hex).strip().lower()
+    with _lock:
+        entry = _observations.get(key)
+        if entry is None:
+            return 0
+        current = float(now if now is not None else time.time())
+        first_raw = entry.get("first_seen_at")
+        first = float(first_raw) if first_raw is not None else current
+        return max(0, min(int(current - first), MAX_SESSION_SECONDS))
+
+
+def _reset_for_tests() -> None:
+    """Drop all observations. Test helper only."""
+    with _lock:
+        _observations.clear()

backend/services/fetchers/flights.py

@@ -17,6 +17,7 @@ from services.network_utils import fetch_with_curl
 from services.fetchers._store import latest_data, _data_lock, _mark_fresh
 from services.fetchers.plane_alert import enrich_with_plane_alert, enrich_with_tracked_names
 from services.fetchers.emissions import get_emissions_info
+from services.fetchers.flight_observations import record_observation as _record_flight_observation
 from services.fetchers.retry import with_retry
 from services.fetchers.route_database import lookup_route
 from services.fetchers.aircraft_database import lookup_aircraft_type
@@ -29,6 +30,88 @@ _RE_AIRLINE_CODE_1 = re.compile(r"^([A-Z]{3})\d")
 _RE_AIRLINE_CODE_2 = re.compile(r"^([A-Z]{3})[A-Z\d]")
 
 
+def detect_gps_jamming_zones(
+    raw_flights: list[dict],
+    *,
+    min_aircraft: int | None = None,
+    min_ratio: float | None = None,
+    nacp_threshold: int | None = None,
+) -> list[dict]:
+    """Detect GPS interference zones from a snapshot of raw ADS-B aircraft.
+
+    Methodology mirrors GPSJam.org / Flightradar24: bin aircraft into 1°x1°
+    grid cells, flag cells where the fraction of aircraft reporting degraded
+    NACp clears a threshold.
+
+    Inputs
+    ------
+    raw_flights:
+        Iterable of dicts. Each item is expected to carry ``lat``, ``lng``
+        (or ``lon``), and ``nac_p``. Records missing position OR missing
+        ``nac_p`` entirely (typical for OpenSky-sourced flights) are
+        skipped — absence-of-data isn't evidence of anything.
+
+    nac_p == 0 IS counted as degraded. Pre-fix code skipped it on the theory
+    that "0 = old transponder, never computed accuracy." That's only half
+    right: modern Mode-S Enhanced Surveillance transponders also fall back
+    to nac_p=0 when they lose GPS lock entirely — which is exactly the
+    jamming signature we're trying to detect. Filtering 0 out was discarding
+    the strongest evidence.
+
+    Denoising:
+        1. Require ``min_aircraft`` per grid cell for statistical validity.
+        2. Subtract 1 from degraded count per cell (GPSJam's technique) so
+           a single quirky transponder can't flag an entire zone.
+        3. Require ratio ``adjusted_degraded / total > min_ratio``.
+
+    All thresholds default to the module-level constants but can be
+    overridden for testing.
+    """
+    min_aircraft = GPS_JAMMING_MIN_AIRCRAFT if min_aircraft is None else int(min_aircraft)
+    min_ratio = GPS_JAMMING_MIN_RATIO if min_ratio is None else float(min_ratio)
+    nacp_threshold = (
+        GPS_JAMMING_NACP_THRESHOLD if nacp_threshold is None else int(nacp_threshold)
+    )
+
+    jamming_grid: dict[str, dict[str, int]] = {}
+    for rf in raw_flights or []:
+        rlat = rf.get("lat")
+        rlng = rf.get("lng") if rf.get("lng") is not None else rf.get("lon")
+        if rlat is None or rlng is None:
+            continue
+        nacp = rf.get("nac_p")
+        if nacp is None:
+            continue
+        grid_key = f"{int(rlat)},{int(rlng)}"
+        cell = jamming_grid.setdefault(grid_key, {"degraded": 0, "total": 0})
+        cell["total"] += 1
+        if nacp < nacp_threshold:
+            cell["degraded"] += 1
+
+    jamming_zones: list[dict] = []
+    for gk, counts in jamming_grid.items():
+        if counts["total"] < min_aircraft:
+            continue
+        adjusted_degraded = max(counts["degraded"] - 1, 0)
+        if adjusted_degraded == 0:
+            continue
+        ratio = adjusted_degraded / counts["total"]
+        if ratio > min_ratio:
+            lat_i, lng_i = gk.split(",")
+            severity = "low" if ratio < 0.5 else "medium" if ratio < 0.75 else "high"
+            jamming_zones.append(
+                {
+                    "lat": int(lat_i) + 0.5,
+                    "lng": int(lng_i) + 0.5,
+                    "severity": severity,
+                    "ratio": round(ratio, 2),
+                    "degraded": counts["degraded"],
+                    "total": counts["total"],
+                }
+            )
+    return jamming_zones
+
+
 # ---------------------------------------------------------------------------
 # OpenSky Network API Client (OAuth2)
 # ---------------------------------------------------------------------------
@@ -459,6 +542,18 @@ def _classify_and_publish(all_adsb_flights):
 
             ac_category = "heli" if model_upper in _HELI_TYPES_BACKEND else "plane"
 
+            # Source attribution: prefer the explicit ``source`` tag stamped
+            # at fetch time (adsb.lol, OpenSky). If absent, fall back to the
+            # legacy ``supplemental_source`` (airplanes.live, adsb.fi) so
+            # supplementals are still attributed without changing their
+            # tagger. Final fallback "adsb.lol" preserves prior behavior for
+            # any caller that synthesizes records without going through one
+            # of our fetchers (e.g. tests).
+            source = (
+                f.get("source")
+                or f.get("supplemental_source")
+                or "adsb.lol"
+            )
             flights.append(
                 {
                     "callsign": flight_str,
@@ -480,6 +575,7 @@ def _classify_and_publish(all_adsb_flights):
                     "airline_code": airline_code,
                     "aircraft_category": ac_category,
                     "nac_p": f.get("nac_p"),
+                    "source": source,
                 }
             )
         except (ValueError, TypeError, KeyError, AttributeError) as loop_e:
@@ -506,6 +602,22 @@ def _classify_and_publish(all_adsb_flights):
         if model:
             emi = get_emissions_info(model)
             if emi:
+                # Cumulative fuel/CO2: multiply the per-hour rate by how
+                # long we've been observing this airframe. Users want to
+                # see the *amount* burned, not just the rate. If we've
+                # never seen this hex before, observed_seconds is 0 and
+                # the cumulative values are 0 until the next refresh —
+                # the rate is still useful info on its own.
+                observed_seconds = _record_flight_observation(
+                    f.get("icao24") or ""
+                )
+                elapsed_h = observed_seconds / 3600.0
+                emi = {
+                    **emi,
+                    "observed_seconds": observed_seconds,
+                    "fuel_gallons_burned": round(emi["fuel_gph"] * elapsed_h, 1),
+                    "co2_kg_emitted": round(emi["co2_kg_per_hour"] * elapsed_h, 1),
+                }
                 f["emissions"] = emi
 
         callsign = f.get("callsign", "").strip().upper()
@@ -724,56 +836,8 @@ def _classify_and_publish(all_adsb_flights):
         latest_data["military_flights"] = military_snapshot
 
     # --- GPS Jamming Detection ---
-    # Uses NACp (Navigation Accuracy Category – Position) from ADS-B to infer
-    # GPS interference zones, similar to GPSJam.org / Flightradar24.
-    # NACp < 8 = position accuracy worse than the FAA-mandated 0.05 NM.
-    #
-    # Denoising (to suppress false positives from old GA transponders):
-    # 1. Skip nac_p == 0 ("unknown accuracy") — old transponders that never
-    #    computed accuracy, NOT evidence of jamming.  Real jamming shows 1-7.
-    # 2. Require minimum aircraft per grid cell for statistical validity.
-    # 3. Subtract 1 from degraded count per cell (GPSJam's technique) so a
-    #    single quirky transponder can't flag an entire zone.
-    # 4. Require the adjusted ratio to exceed the threshold.
     try:
-        jamming_grid = {}
-        raw_flights = raw_flights_snapshot
-        for rf in raw_flights:
-            rlat = rf.get("lat")
-            rlng = rf.get("lng") or rf.get("lon")
-            if rlat is None or rlng is None:
-                continue
-            nacp = rf.get("nac_p")
-            if nacp is None or nacp == 0:
-                continue
-            grid_key = f"{int(rlat)},{int(rlng)}"
-            if grid_key not in jamming_grid:
-                jamming_grid[grid_key] = {"degraded": 0, "total": 0}
-            jamming_grid[grid_key]["total"] += 1
-            if nacp < GPS_JAMMING_NACP_THRESHOLD:
-                jamming_grid[grid_key]["degraded"] += 1
-
-        jamming_zones = []
-        for gk, counts in jamming_grid.items():
-            if counts["total"] < GPS_JAMMING_MIN_AIRCRAFT:
-                continue
-            adjusted_degraded = max(counts["degraded"] - 1, 0)
-            if adjusted_degraded == 0:
-                continue
-            ratio = adjusted_degraded / counts["total"]
-            if ratio > GPS_JAMMING_MIN_RATIO:
-                lat_i, lng_i = gk.split(",")
-                severity = "low" if ratio < 0.5 else "medium" if ratio < 0.75 else "high"
-                jamming_zones.append(
-                    {
-                        "lat": int(lat_i) + 0.5,
-                        "lng": int(lng_i) + 0.5,
-                        "severity": severity,
-                        "ratio": round(ratio, 2),
-                        "degraded": counts["degraded"],
-                        "total": counts["total"],
-                    }
-                )
+        jamming_zones = detect_gps_jamming_zones(raw_flights_snapshot)
         with _data_lock:
             latest_data["gps_jamming"] = jamming_zones
         if jamming_zones:
@@ -849,7 +913,15 @@ def _fetch_adsb_lol_regions():
             res = fetch_with_curl(url, timeout=10)
             if res.status_code == 200:
                 data = res.json()
-                return data.get("ac", [])
+                aircraft = data.get("ac", [])
+                # Stamp the source at the fetch site so attribution survives
+                # the OpenSky/supplemental dedupe-by-hex merge downstream.
+                # Previously adsb.lol records carried no marker while OpenSky
+                # records got ``is_opensky: True`` — which made flight tooltips
+                # look like everything came from OpenSky.
+                for a in aircraft:
+                    a["source"] = "adsb.lol"
+                return aircraft
         except (
             requests.RequestException,
             ConnectionError,
@@ -932,6 +1004,7 @@ def _enrich_with_opensky_and_supplemental(adsb_flights):
                                     "gs": (s[9] * 1.94384) if s[9] else 0,
                                     "t": "Unknown",
                                     "is_opensky": True,
+                                    "source": "OpenSky",
                                 }
                             )
                     elif os_res.status_code == 429:

backend/services/fetchers/military.py

@@ -7,6 +7,7 @@ import requests
 from services.network_utils import fetch_with_curl
 from services.fetchers._store import latest_data, _data_lock, _mark_fresh
 from services.fetchers.emissions import get_emissions_info
+from services.fetchers.flight_observations import record_observation as _record_flight_observation
 from services.fetchers.plane_alert import enrich_with_plane_alert
 
 logger = logging.getLogger("services.data_fetcher")
@@ -171,6 +172,7 @@ def fetch_military_flights():
                 h = a.get("hex", "").lower()
                 if h and h not in seen_hex:
                     seen_hex.add(h)
+                    a["source"] = "adsb.lol"
                     all_mil_ac.append(a)
     except Exception as e:
         logger.warning(f"adsb.lol mil fetch failed: {e}")
@@ -182,6 +184,7 @@ def fetch_military_flights():
                 h = a.get("hex", "").lower()
                 if h and h not in seen_hex:
                     seen_hex.add(h)
+                    a["source"] = "airplanes.live"
                     all_mil_ac.append(a)
             logger.info(f"airplanes.live mil: +{len(resp2.json().get('ac', []))} raw, {len(all_mil_ac)} total unique")
     except Exception as e:
@@ -234,6 +237,7 @@ def fetch_military_flights():
                             "registration": f.get("r", "N/A"),
                             "icao24": icao_hex,
                             "squawk": f.get("squawk", ""),
+                            "source": f.get("source") or "adsb.lol",
                         })
                         continue
 
@@ -258,7 +262,8 @@ def fetch_military_flights():
                         "model": f.get("t", "Unknown"),
                         "icao24": icao_hex,
                         "speed_knots": speed_knots,
-                        "squawk": f.get("squawk", "")
+                        "squawk": f.get("squawk", ""),
+                        "source": f.get("source") or "adsb.lol",
                     })
                 except Exception as loop_e:
                     logger.error(f"Mil flight interpolation error: {loop_e}")
@@ -296,6 +301,18 @@ def fetch_military_flights():
         if model:
             emissions = get_emissions_info(model)
             if emissions:
+                # Cumulative fuel/CO2 since first observation — mirrors
+                # the civilian path in flights._classify_and_publish.
+                observed_seconds = _record_flight_observation(
+                    mf.get("icao24") or ""
+                )
+                elapsed_h = observed_seconds / 3600.0
+                emissions = {
+                    **emissions,
+                    "observed_seconds": observed_seconds,
+                    "fuel_gallons_burned": round(emissions["fuel_gph"] * elapsed_h, 1),
+                    "co2_kg_emitted": round(emissions["co2_kg_per_hour"] * elapsed_h, 1),
+                }
                 mf["emissions"] = emissions
         if mf.get("alert_category"):
             mf["type"] = "tracked_flight"

backend/tests/test_flight_observations.py

@@ -0,0 +1,258 @@
+"""Cumulative fuel/CO2 tracking via per-aircraft observation timestamps.
+
+Background
+----------
+Users want the running total of fuel burned per aircraft — not just the
+rate. We track first-seen-at per icao24 and multiply elapsed observation
+time by the model-based rate. This module's job is exclusively the
+timestamp bookkeeping; multiplication happens in the flights/military
+fetchers.
+
+These tests pin:
+
+  * First sighting returns 0 (no airtime yet).
+  * Repeated sightings within ``REOPEN_GAP_S`` accumulate elapsed time.
+  * Gap longer than ``REOPEN_GAP_S`` resets the session (plane landed
+    and took off again — different flight).
+  * ``MAX_SESSION_SECONDS`` clamp protects against clock skew bugs.
+  * ``prune()`` drops stale entries.
+  * ``get_session_seconds`` reads without bumping last_seen.
+  * Empty / None icao input is a defensive no-op.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _reset_observations():
+    from services.fetchers import flight_observations as obs
+    obs._reset_for_tests()
+    yield
+    obs._reset_for_tests()
+
+
+class TestRecordObservation:
+    def test_first_sighting_returns_zero(self):
+        from services.fetchers.flight_observations import record_observation
+        assert record_observation("a12345", now=1000.0) == 0
+
+    def test_repeated_sightings_accumulate(self):
+        """ADS-B refreshes every ~minute in practice, so each observation
+        is within ``REOPEN_GAP_S`` (15 min) of the last and we keep
+        accumulating. Walking the timestamps in 5-minute steps so we
+        stay inside the reopen window the whole way."""
+        from services.fetchers.flight_observations import record_observation
+        record_observation("a12345", now=1000.0)
+        # 1 minute later (within REOPEN_GAP_S)
+        assert record_observation("a12345", now=1060.0) == 60
+        # Step through 5-minute spaced refreshes — first_seen_at stays
+        # at 1000.0 the whole time, and we approach a 1-hour airtime.
+        assert record_observation("a12345", now=1360.0) == 360
+        assert record_observation("a12345", now=1660.0) == 660
+        assert record_observation("a12345", now=1960.0) == 960
+        assert record_observation("a12345", now=2260.0) == 1260
+        assert record_observation("a12345", now=2560.0) == 1560
+        assert record_observation("a12345", now=2860.0) == 1860
+        assert record_observation("a12345", now=3160.0) == 2160
+        assert record_observation("a12345", now=3460.0) == 2460
+        assert record_observation("a12345", now=3760.0) == 2760
+        assert record_observation("a12345", now=4060.0) == 3060
+        assert record_observation("a12345", now=4360.0) == 3360
+        # 1 hour after first sighting — still inside the 15-min reopen
+        # window from the prior 4360 observation.
+        assert record_observation("a12345", now=4600.0) == 3600
+
+    def test_gap_longer_than_reopen_resets_session(self):
+        """If a hex hasn't been seen in ``REOPEN_GAP_S`` (15 min default),
+        the next sighting is treated as a new flight — first_seen_at resets."""
+        from services.fetchers.flight_observations import record_observation
+        record_observation("a12345", now=1000.0)
+        record_observation("a12345", now=1500.0)  # 500s later — within gap
+        # Now 20 minutes of silence (1200s > 900s threshold) → session reset.
+        assert record_observation("a12345", now=2700.0) == 0
+        # And the next quick sighting starts accumulating from 2700 again.
+        assert record_observation("a12345", now=2760.0) == 60
+
+    def test_session_clamp(self):
+        """Clock skew protection: when a hex has been continuously
+        observed for longer than ``MAX_SESSION_SECONDS``, clamp.
+
+        Synthesizes the state directly because driving 86,400+ seconds of
+        observations through the public API in a test would take 1000+
+        REOPEN_GAP_S-respecting steps.
+        """
+        from services.fetchers import flight_observations as obs
+        from services.fetchers.flight_observations import _observations, _lock
+
+        # last_seen_at very recent so REOPEN_GAP_S branch does NOT fire,
+        # but first_seen_at way in the past so the elapsed math overflows
+        # MAX_SESSION_SECONDS. Clamp must kick in.
+        big_now = float(obs.MAX_SESSION_SECONDS + 1_000_000)
+        with _lock:
+            _observations["a12345"] = {
+                "first_seen_at": 0.0,
+                "last_seen_at": big_now - 60,  # 60s ago — well inside gap window
+            }
+        elapsed = obs.record_observation("a12345", now=big_now)
+        assert elapsed == obs.MAX_SESSION_SECONDS, (
+            f"elapsed must be clamped to MAX_SESSION_SECONDS; got {elapsed}"
+        )
+
+    def test_empty_input_returns_zero(self):
+        from services.fetchers.flight_observations import record_observation
+        assert record_observation("") == 0
+        assert record_observation(None) == 0  # type: ignore[arg-type]
+        assert record_observation("   ") == 0
+
+    def test_case_insensitive_key(self):
+        """ICAO24 hex codes are case-insensitive — adsb.lol lowercases
+        them, OpenSky may not. Normalize so both refer to the same airframe."""
+        from services.fetchers.flight_observations import record_observation
+        record_observation("A12345", now=1000.0)
+        # Different case must hit the same entry.
+        assert record_observation("a12345", now=1060.0) == 60
+
+
+class TestGetSessionSeconds:
+    def test_read_only_does_not_bump(self):
+        from services.fetchers.flight_observations import (
+            record_observation,
+            get_session_seconds,
+        )
+        record_observation("a12345", now=1000.0)
+        record_observation("a12345", now=1060.0)  # bumps last_seen
+
+        # Now read at t=2000. Without bumping, gap=2000-1060=940 > 900,
+        # so a recording call would reset. But the read should NOT reset.
+        seconds_at_2000 = get_session_seconds("a12345", now=2000.0)
+        assert seconds_at_2000 == 1000, (
+            f"read should return 2000-1000=1000s; got {seconds_at_2000}"
+        )
+        # Verify the next recording at t=2001 still resets (gap > 900s
+        # from the read above — proves the read didn't bump last_seen).
+        from services.fetchers.flight_observations import record_observation as rec
+        assert rec("a12345", now=2001.0) == 0  # session reset
+
+    def test_unknown_hex_returns_zero(self):
+        from services.fetchers.flight_observations import get_session_seconds
+        assert get_session_seconds("nonexistent") == 0
+
+
+class TestPrune:
+    def test_drops_stale_entries(self):
+        from services.fetchers import flight_observations as obs
+
+        obs.record_observation("active", now=10_000.0)
+        obs.record_observation("stale", now=1.0)
+
+        dropped = obs.prune(now=10_000.0)
+
+        assert dropped == 1
+        # Active entry survives:
+        assert obs.get_session_seconds("active", now=10_001.0) == 1
+        # Stale entry was dropped — next obs starts fresh:
+        assert obs.record_observation("stale", now=10_002.0) == 0
+
+    def test_no_op_when_nothing_stale(self):
+        from services.fetchers import flight_observations as obs
+        obs.record_observation("hex1", now=1000.0)
+        obs.record_observation("hex2", now=1000.0)
+
+        dropped = obs.prune(now=1500.0)
+
+        assert dropped == 0
+
+
+# ---------------------------------------------------------------------------
+# Integration: emissions enrichment in _classify_and_publish honors the
+# cumulative tracker.
+# ---------------------------------------------------------------------------
+
+
+class TestEmissionsCumulativeIntegration:
+    def _reset_store(self):
+        from services.fetchers._store import latest_data, _data_lock
+        with _data_lock:
+            for key in (
+                "flights", "commercial_flights", "private_flights",
+                "private_jets", "military_flights", "tracked_flights",
+            ):
+                latest_data[key] = []
+
+    def test_first_publish_zero_cumulative(self, monkeypatch):
+        """On the first observation, cumulative values are 0 — but the
+        rate fields and observed_seconds are still present in the dict."""
+        from services.fetchers import flights as flights_module
+        from services.fetchers._store import latest_data, _data_lock
+
+        self._reset_store()
+        monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
+        monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
+
+        flights_module._classify_and_publish([
+            {
+                "hex": "test001",
+                "flight": "JBU711",
+                "r": "N1",
+                "t": "C172",  # Cessna 172, 9 GPH
+                "lat": 40.0,
+                "lon": -100.0,
+                "alt_baro": 3000,
+                "gs": 100,
+            }
+        ])
+
+        with _data_lock:
+            published = list(latest_data.get("flights", []))
+        assert len(published) == 1
+        emi = published[0].get("emissions")
+        assert emi is not None
+        assert emi["fuel_gph"] == 9
+        assert emi["observed_seconds"] == 0
+        assert emi["fuel_gallons_burned"] == 0.0
+        assert emi["co2_kg_emitted"] == 0.0
+
+    def test_second_publish_accumulates(self, monkeypatch):
+        """Publishing the same hex a second time picks up real elapsed time
+        and produces non-zero cumulative values."""
+        import time as _time_real
+        from services.fetchers import flights as flights_module
+        from services.fetchers import flight_observations as obs
+        from services.fetchers._store import latest_data, _data_lock
+
+        self._reset_store()
+        monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
+        monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
+
+        # Manually seed an observation 1 hour in the past so the next
+        # publish picks up ~3600s elapsed.
+        with obs._lock:
+            obs._observations["test002"] = {
+                "first_seen_at": _time_real.time() - 3600,
+                "last_seen_at": _time_real.time() - 60,
+            }
+
+        flights_module._classify_and_publish([
+            {
+                "hex": "test002",
+                "flight": "JBU711",
+                "r": "N1",
+                "t": "C172",  # 9 GPH
+                "lat": 40.0,
+                "lon": -100.0,
+                "alt_baro": 3000,
+                "gs": 100,
+            }
+        ])
+
+        with _data_lock:
+            published = list(latest_data.get("flights", []))
+        assert len(published) == 1
+        emi = published[0].get("emissions")
+        # Roughly 1 hour observed → 9 gal burned.
+        assert 3500 <= emi["observed_seconds"] <= 3700
+        assert 8.7 <= emi["fuel_gallons_burned"] <= 9.3
+        # CO2 = 9 gph * 9.57 kg/gal = 86.1 kg/hr.
+        assert 84 <= emi["co2_kg_emitted"] <= 88

backend/tests/test_flight_source_attribution.py

@@ -0,0 +1,354 @@
+"""Per-flight source attribution.
+
+Background
+----------
+Pre-fix, adsb.lol records (the primary source for most flights) carried
+no source marker. OpenSky records got ``is_opensky: True`` and
+supplementals got ``supplemental_source``, so any UI that wanted to show
+which provider a flight came from saw OpenSky/airplanes.live records as
+explicitly tagged and adsb.lol records as "unlabeled" — making it look
+like adsb.lol wasn't even being used.
+
+This caused user confusion ("only military planes have adsb.lol
+telemetry") that was diagnostic noise, not a real bug. The actual fix:
+stamp ``source`` at every fetch site so the downstream consumer can
+attribute the provider with no guesswork.
+
+These tests pin:
+
+  * adsb.lol regional records get ``source: "adsb.lol"`` at fetch time
+    (synthesized via the published flight dict).
+  * OpenSky records get ``source: "OpenSky"`` (alongside the existing
+    ``is_opensky: True`` for backwards compat).
+  * Supplementals (airplanes.live, adsb.fi) flow through with their
+    ``supplemental_source`` honored.
+  * The military fetcher tags ``source`` on military_flights and uavs.
+  * The published flight dict carries ``source`` so downstream code
+    can render attribution.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# _classify_and_publish — source field flows into published flight dict
+# ---------------------------------------------------------------------------
+
+
+class TestClassifyAndPublishSource:
+    def _reset_store(self):
+        """Clear store before each test so we get deterministic state."""
+        from services.fetchers._store import latest_data, _data_lock
+        with _data_lock:
+            for key in (
+                "flights", "commercial_flights", "private_flights",
+                "private_jets", "military_flights", "tracked_flights",
+            ):
+                latest_data[key] = []
+        return latest_data
+
+    def test_adsb_lol_record_tagged_in_published_flight(self, monkeypatch):
+        """A raw adsb.lol record (carrying ``source: 'adsb.lol'`` from the
+        fetch site) flows through ``_classify_and_publish`` and the
+        published flight dict carries the same ``source`` field."""
+        from services.fetchers import flights as flights_module
+        from services.fetchers._store import latest_data, _data_lock
+
+        self._reset_store()
+
+        # Patch route + type lookups so they don't try to hit the network.
+        monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
+        monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
+
+        flights_module._classify_and_publish(
+            [
+                {
+                    "hex": "ad7701",
+                    "flight": "JBU711",
+                    "r": "N967JT",
+                    "t": "A321",
+                    "lat": 40.0,
+                    "lon": -100.0,
+                    "alt_baro": 36000,
+                    "gs": 401.6,
+                    "nac_p": 9,
+                    "source": "adsb.lol",  # stamped at fetch site
+                }
+            ]
+        )
+
+        with _data_lock:
+            published = list(latest_data.get("flights", []))
+        assert len(published) == 1
+        assert published[0]["source"] == "adsb.lol"
+        # nac_p still flows through too — sanity check that adding source
+        # didn't break the existing GPS jamming signal.
+        assert published[0]["nac_p"] == 9
+
+    def test_opensky_record_tagged_in_published_flight(self, monkeypatch):
+        """OpenSky-sourced records carry ``source: 'OpenSky'`` (plus the
+        existing ``is_opensky: True`` for back-compat)."""
+        from services.fetchers import flights as flights_module
+        from services.fetchers._store import latest_data, _data_lock
+
+        self._reset_store()
+        monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
+        monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
+
+        flights_module._classify_and_publish(
+            [
+                {
+                    "hex": "a12345",
+                    "flight": "UAL100",
+                    "r": "N100UA",
+                    "t": "Unknown",
+                    "lat": 41.0,
+                    "lon": -87.0,
+                    "alt_baro": 35000,
+                    "gs": 450,
+                    # No nac_p — OpenSky doesn't carry it.
+                    "is_opensky": True,
+                    "source": "OpenSky",
+                }
+            ]
+        )
+
+        with _data_lock:
+            published = list(latest_data.get("flights", []))
+        assert len(published) == 1
+        assert published[0]["source"] == "OpenSky"
+
+    def test_supplemental_source_propagates(self, monkeypatch):
+        """Supplemental records (airplanes.live, adsb.fi) have their
+        legacy ``supplemental_source`` field promoted to the unified
+        ``source`` field in the published dict — so consumers don't have
+        to inspect two different keys."""
+        from services.fetchers import flights as flights_module
+        from services.fetchers._store import latest_data, _data_lock
+
+        self._reset_store()
+        monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
+        monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
+
+        flights_module._classify_and_publish(
+            [
+                {
+                    "hex": "b22222",
+                    "flight": "DAL200",
+                    "r": "N200DL",
+                    "t": "B738",
+                    "lat": 42.0,
+                    "lon": -90.0,
+                    "alt_baro": 32000,
+                    "gs": 420,
+                    "supplemental_source": "airplanes.live",
+                    # No explicit "source" — should fall through to
+                    # supplemental_source.
+                }
+            ]
+        )
+
+        with _data_lock:
+            published = list(latest_data.get("flights", []))
+        assert len(published) == 1
+        assert published[0]["source"] == "airplanes.live"
+
+    def test_explicit_source_wins_over_supplemental_source(self, monkeypatch):
+        """If both fields are present, explicit ``source`` wins (it's the
+        newer canonical tag)."""
+        from services.fetchers import flights as flights_module
+        from services.fetchers._store import latest_data, _data_lock
+
+        self._reset_store()
+        monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
+        monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
+
+        flights_module._classify_and_publish(
+            [
+                {
+                    "hex": "c33333",
+                    "flight": "AAL300",
+                    "r": "N300AA",
+                    "t": "A321",
+                    "lat": 33.0,
+                    "lon": -97.0,
+                    "alt_baro": 34000,
+                    "gs": 430,
+                    "source": "adsb.lol",
+                    "supplemental_source": "adsb.fi",
+                }
+            ]
+        )
+
+        with _data_lock:
+            published = list(latest_data.get("flights", []))
+        assert published[0]["source"] == "adsb.lol"
+
+    def test_untagged_record_defaults_to_adsb_lol(self, monkeypatch):
+        """A record with neither ``source`` nor ``supplemental_source``
+        (e.g. synthesized by a test, or a fetcher that hasn't been
+        migrated yet) defaults to ``"adsb.lol"`` since that's been the
+        primary source historically. Defensive default — better than
+        empty string."""
+        from services.fetchers import flights as flights_module
+        from services.fetchers._store import latest_data, _data_lock
+
+        self._reset_store()
+        monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
+        monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
+
+        flights_module._classify_and_publish(
+            [
+                {
+                    "hex": "d44444",
+                    "flight": "SWA400",
+                    "r": "N400SW",
+                    "t": "B737",
+                    "lat": 32.0,
+                    "lon": -110.0,
+                    "alt_baro": 30000,
+                    "gs": 410,
+                }
+            ]
+        )
+
+        with _data_lock:
+            published = list(latest_data.get("flights", []))
+        assert published[0]["source"] == "adsb.lol"
+
+
+# ---------------------------------------------------------------------------
+# adsb.lol regional fetcher tags at fetch time
+# ---------------------------------------------------------------------------
+
+
+class TestAdsbLolRegionalTagging:
+    def test_fetch_region_stamps_source_on_each_aircraft(self, monkeypatch):
+        """The wrapper around the adsb.lol regional endpoint stamps
+        ``source: 'adsb.lol'`` on every record before returning, so the
+        downstream merge step sees attribution survive even when the
+        record gets reshuffled (e.g. dedupe-by-hex during OpenSky merge)."""
+        from services.fetchers import flights as flights_module
+
+        # Fake response — 3 aircraft, none have a source field originally.
+        class FakeResp:
+            status_code = 200
+
+            def json(self):
+                return {
+                    "ac": [
+                        {"hex": "a1", "lat": 40.0, "lon": -100.0, "nac_p": 8},
+                        {"hex": "a2", "lat": 40.1, "lon": -100.1, "nac_p": 9},
+                        {"hex": "a3", "lat": 40.2, "lon": -100.2, "nac_p": 10},
+                    ]
+                }
+
+        monkeypatch.setattr(
+            flights_module, "fetch_with_curl", lambda *a, **kw: FakeResp()
+        )
+
+        results = flights_module._fetch_adsb_lol_regions()
+
+        assert len(results) >= 3
+        # Every aircraft we got back must be tagged.
+        sources = {a.get("source") for a in results}
+        assert sources == {"adsb.lol"}, (
+            f"adsb.lol regional fetcher must stamp source on every record; "
+            f"got: {sources}"
+        )
+
+    def test_fetch_region_failure_returns_empty_without_crashing(self, monkeypatch):
+        """If adsb.lol returns non-200, the fetcher returns [] gracefully —
+        downstream code already handles this. Sanity check that the source
+        tagging doesn't introduce a new failure mode."""
+        from services.fetchers import flights as flights_module
+
+        class FakeResp:
+            status_code = 500
+            def json(self): return {}
+
+        monkeypatch.setattr(
+            flights_module, "fetch_with_curl", lambda *a, **kw: FakeResp()
+        )
+
+        results = flights_module._fetch_adsb_lol_regions()
+
+        assert results == []
+
+
+# ---------------------------------------------------------------------------
+# Military fetcher tags source on output dicts
+# ---------------------------------------------------------------------------
+
+
+class TestMilitarySourceTagging:
+    def test_military_output_carries_source_field(self, monkeypatch):
+        """Each entry in ``military_flights`` should carry a ``source``
+        field. Pre-fix the only military attribution was inferring from
+        which endpoint we hit; now it's explicit."""
+        from services.fetchers import military as mil_module
+        from services.fetchers._store import latest_data, _data_lock
+
+        # Reset relevant store state.
+        with _data_lock:
+            latest_data["military_flights"] = []
+            latest_data["uavs"] = []
+            latest_data["tracked_flights"] = []
+
+        # Stub _store.is_any_active so the fetch doesn't early-return.
+        # The military module imports the function inline at call time,
+        # so we have to patch it on the _store module itself rather than
+        # on the military module.
+        from services.fetchers import _store as store_module
+        monkeypatch.setattr(store_module, "is_any_active", lambda *_: True)
+
+        # Stub fetch_with_curl to return one synthetic military aircraft
+        # from adsb.lol, none from airplanes.live.
+        class _RespMil:
+            status_code = 200
+            def json(self):
+                return {
+                    "ac": [
+                        {
+                            "hex": "ae6c1d",
+                            "flight": "CRUSH52",
+                            "r": "170281",
+                            "t": "C30J",
+                            "lat": 47.594,
+                            "lon": -124.879,
+                            "alt_baro": 9025,
+                            "gs": 162.8,
+                            "track": 334.5,
+                            "nac_p": 10,
+                        }
+                    ]
+                }
+
+        class _RespEmpty:
+            status_code = 200
+            def json(self):
+                return {"ac": []}
+
+        def _fake_fetch(url, *a, **kw):
+            if "adsb.lol" in url:
+                return _RespMil()
+            return _RespEmpty()
+
+        monkeypatch.setattr(mil_module, "fetch_with_curl", _fake_fetch)
+        # Stubs for downstream enrichments that try to hit external state.
+        monkeypatch.setattr(mil_module, "enrich_with_plane_alert", lambda mf: None)
+        monkeypatch.setattr(mil_module, "_enrich_country", lambda hex_, flag: ("US", "USAF"))
+        monkeypatch.setattr(mil_module, "_classify_military_type", lambda t: "transport")
+        monkeypatch.setattr(mil_module, "_classify_uav", lambda m, c: (False, "", ""))
+        monkeypatch.setattr(mil_module, "get_emissions_info", lambda model: None)
+        monkeypatch.setattr(mil_module, "_mark_fresh", lambda *keys: None)
+
+        mil_module.fetch_military_flights()
+
+        with _data_lock:
+            mil_published = list(latest_data.get("military_flights", []))
+
+        assert len(mil_published) == 1
+        assert mil_published[0]["source"] == "adsb.lol"

backend/tests/test_gps_jamming_detector.py

@@ -0,0 +1,333 @@
+"""GPS jamming detection — nac_p=0 counted, lowered thresholds.
+
+Background
+----------
+Pre-fix, the detector had three stacked filters that together meant the
+``gps_jamming`` layer almost never lit up:
+
+  1. ``nac_p == 0`` aircraft were dropped on the theory that "0 = old
+     transponder." But modern Mode-S Enhanced Surveillance transponders
+     also fall back to ``nac_p == 0`` when they lose GPS lock entirely —
+     which is *exactly* the jamming signature we want to catch.
+  2. ``GPS_JAMMING_MIN_AIRCRAFT = 5`` per 1°x1° cell.
+  3. ``GPS_JAMMING_MIN_RATIO = 0.30`` adjusted ratio.
+
+Combined with the existing ``-1`` noise cushion (``adjusted = degraded - 1``)
+the bar to clear required dense, busy airspace — but jamming hotspots
+(eastern Med, eastern Ukraine, Iran/Iraq) tend to have sparser traffic
+precisely because pilots avoid them.
+
+These tests pin the new behavior:
+
+  * ``nac_p == 0`` is now counted as degraded.
+  * ``nac_p == None`` (no field — typical for OpenSky records) is still
+    skipped — absence isn't evidence.
+  * Thresholds lowered to 3 aircraft / 0.20 ratio.
+  * Public function signature accepts overrides so callers / future
+    operators can re-tune without code edits.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# nac_p == 0 inclusion (the headline fix)
+# ---------------------------------------------------------------------------
+
+
+class TestNacpZeroCounted:
+    def test_cell_dominated_by_nacp_zero_now_fires(self):
+        """Three aircraft all reporting nac_p=0 in one cell, plus two
+        with valid GPS. Pre-fix the three nac_p=0 records were skipped
+        entirely (cell would have total=2, degraded=0, no zone). Post-fix
+        they count as degraded — this IS the jamming signature."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        # All in 1°x1° cell at int(lat)=40, int(lng)=-100
+        feed = [
+            {"hex": "a1", "lat": 40.1, "lng": -100.1, "nac_p": 0},
+            {"hex": "a2", "lat": 40.5, "lng": -100.5, "nac_p": 0},
+            {"hex": "a3", "lat": 40.9, "lng": -100.9, "nac_p": 0},
+            {"hex": "b1", "lat": 40.2, "lng": -100.3, "nac_p": 9},
+            {"hex": "b2", "lat": 40.7, "lng": -100.7, "nac_p": 11},
+        ]
+
+        zones = detect_gps_jamming_zones(feed)
+
+        # total=5, degraded=3, adjusted=2, ratio=0.40 > 0.20 → zone fires.
+        assert len(zones) == 1
+        assert zones[0]["degraded"] == 3
+        assert zones[0]["total"] == 5
+        assert zones[0]["ratio"] == 0.40
+        # Grid-cell center coords.
+        assert zones[0]["lat"] == 40.5
+        assert zones[0]["lng"] == -99.5
+
+    def test_nacp_zero_alone_clears_min_aircraft(self):
+        """A cell with exactly 3 aircraft all reporting nac_p=0 must
+        fire under the new MIN_AIRCRAFT=3 + MIN_RATIO=0.20 regime."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        feed = [
+            {"hex": "a1", "lat": 50.1, "lng": 30.1, "nac_p": 0},
+            {"hex": "a2", "lat": 50.5, "lng": 30.5, "nac_p": 0},
+            {"hex": "a3", "lat": 50.9, "lng": 30.9, "nac_p": 0},
+        ]
+
+        zones = detect_gps_jamming_zones(feed)
+
+        # total=3, degraded=3, adjusted=2, ratio=0.667 > 0.20 → fires.
+        # severity is "medium" because 0.5 ≤ ratio < 0.75.
+        assert len(zones) == 1
+        assert zones[0]["severity"] == "medium"
+
+
+# ---------------------------------------------------------------------------
+# nac_p == None is still skipped (preserve OpenSky behavior)
+# ---------------------------------------------------------------------------
+
+
+class TestNoneStillSkipped:
+    def test_none_records_dont_add_to_grid(self):
+        """OpenSky's /states/all doesn't include nac_p, so its records
+        arrive with the field absent (``rf.get("nac_p") is None``). These
+        records must NOT count toward total — absence-of-data isn't
+        evidence of either jamming OR working GPS."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        # 3 jammed + 4 OpenSky-style (no nac_p). Pre-fix and post-fix
+        # behavior should be identical here: None always skipped.
+        feed = [
+            {"hex": "a1", "lat": 40.1, "lng": -100.1, "nac_p": 0},
+            {"hex": "a2", "lat": 40.2, "lng": -100.2, "nac_p": 0},
+            {"hex": "a3", "lat": 40.3, "lng": -100.3, "nac_p": 0},
+            # OpenSky-style: no nac_p at all
+            {"hex": "o1", "lat": 40.4, "lng": -100.4},
+            {"hex": "o2", "lat": 40.5, "lng": -100.5},
+            {"hex": "o3", "lat": 40.6, "lng": -100.6},
+            {"hex": "o4", "lat": 40.7, "lng": -100.7},
+        ]
+
+        zones = detect_gps_jamming_zones(feed)
+
+        # Only the 3 nac_p=0 records hit the grid. total=3, not 7.
+        assert len(zones) == 1
+        assert zones[0]["total"] == 3
+        assert zones[0]["degraded"] == 3
+
+    def test_explicit_none_skipped(self):
+        """Same behavior when ``nac_p`` is present but set to None
+        (defensive — adsb.lol shouldn't do this, but downstream
+        normalizers might)."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        feed = [
+            {"hex": "a1", "lat": 0.1, "lng": 0.1, "nac_p": None},
+            {"hex": "a2", "lat": 0.2, "lng": 0.2, "nac_p": None},
+            {"hex": "a3", "lat": 0.3, "lng": 0.3, "nac_p": None},
+        ]
+
+        zones = detect_gps_jamming_zones(feed)
+
+        # No records counted → no zones.
+        assert zones == []
+
+
+# ---------------------------------------------------------------------------
+# Lowered MIN_AIRCRAFT (5 → 3)
+# ---------------------------------------------------------------------------
+
+
+class TestMinAircraftLowered:
+    def test_three_aircraft_cell_now_qualifies(self):
+        """Pre-fix MIN_AIRCRAFT=5 blocked sparse cells entirely. Post-fix
+        the bar is 3 aircraft per cell, which is realistic for the actual
+        jamming hotspots where traffic is thinner."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        feed = [
+            {"hex": "a1", "lat": 33.1, "lng": 44.1, "nac_p": 3},
+            {"hex": "a2", "lat": 33.2, "lng": 44.2, "nac_p": 5},
+            {"hex": "a3", "lat": 33.3, "lng": 44.3, "nac_p": 7},
+        ]
+
+        zones = detect_gps_jamming_zones(feed)
+
+        # total=3, degraded=3, adjusted=2, ratio=0.667 — fires under new
+        # rules, would have been blocked by MIN_AIRCRAFT=5 pre-fix.
+        assert len(zones) == 1
+
+    def test_two_aircraft_cell_still_blocked(self):
+        """We didn't lower the bar to 2 — that would create too much
+        single-transponder noise. Two aircraft per cell still doesn't
+        qualify."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        feed = [
+            {"hex": "a1", "lat": 33.1, "lng": 44.1, "nac_p": 3},
+            {"hex": "a2", "lat": 33.2, "lng": 44.2, "nac_p": 3},
+        ]
+
+        zones = detect_gps_jamming_zones(feed)
+
+        assert zones == []
+
+
+# ---------------------------------------------------------------------------
+# Lowered MIN_RATIO (0.30 → 0.20)
+# ---------------------------------------------------------------------------
+
+
+class TestMinRatioLowered:
+    def test_ratio_between_old_and_new_threshold_fires(self):
+        """Construct a cell whose ratio sits in the (0.20, 0.30) window:
+        fires under the new bar, would have been blocked pre-fix."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        # 10 aircraft, 4 degraded → adjusted=3, ratio=3/10=0.30.
+        # Pre-fix threshold was > 0.30 strict — would NOT fire.
+        # Post-fix threshold is > 0.20 — fires.
+        feed = (
+            [{"hex": f"d{i}", "lat": 40.1, "lng": -100.1, "nac_p": 3} for i in range(4)]
+            + [{"hex": f"c{i}", "lat": 40.5, "lng": -100.5, "nac_p": 9} for i in range(6)]
+        )
+
+        zones = detect_gps_jamming_zones(feed)
+
+        assert len(zones) == 1
+        assert zones[0]["degraded"] == 4
+        assert zones[0]["total"] == 10
+        assert zones[0]["ratio"] == 0.30
+
+    def test_ratio_at_or_below_new_threshold_does_not_fire(self):
+        """Ratio of exactly 0.20 must NOT fire (strict ``>`` comparison)."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        # 15 aircraft, 4 degraded → adjusted=3, ratio=3/15=0.20. Strictly
+        # not greater than 0.20, so doesn't qualify.
+        feed = (
+            [{"hex": f"d{i}", "lat": 40.1, "lng": -100.1, "nac_p": 3} for i in range(4)]
+            + [{"hex": f"c{i}", "lat": 40.5, "lng": -100.5, "nac_p": 9} for i in range(11)]
+        )
+
+        zones = detect_gps_jamming_zones(feed)
+
+        assert zones == []
+
+
+# ---------------------------------------------------------------------------
+# Pre-existing noise cushion (-1) preserved
+# ---------------------------------------------------------------------------
+
+
+class TestNoiseCushionPreserved:
+    def test_single_quirky_transponder_doesnt_fire(self):
+        """One degraded aircraft in a healthy cell shouldn't fire even
+        under the relaxed thresholds. The ``-1`` adjustment in the
+        detector exists for this reason."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        feed = (
+            [{"hex": "d1", "lat": 40.1, "lng": -100.1, "nac_p": 3}]
+            + [{"hex": f"c{i}", "lat": 40.5, "lng": -100.5, "nac_p": 9} for i in range(10)]
+        )
+
+        zones = detect_gps_jamming_zones(feed)
+
+        # total=11, degraded=1, adjusted=0 → cell short-circuits.
+        assert zones == []
+
+
+# ---------------------------------------------------------------------------
+# Constants pinned (catches accidental rollback)
+# ---------------------------------------------------------------------------
+
+
+class TestConstantsPinned:
+    def test_min_aircraft_is_three(self):
+        from services.constants import GPS_JAMMING_MIN_AIRCRAFT
+        assert GPS_JAMMING_MIN_AIRCRAFT == 3, (
+            "MIN_AIRCRAFT must be 3; raising it back to 5 brings back the "
+            "'jamming never shows' bug."
+        )
+
+    def test_min_ratio_is_0_20(self):
+        from services.constants import GPS_JAMMING_MIN_RATIO
+        assert GPS_JAMMING_MIN_RATIO == 0.20, (
+            "MIN_RATIO must be 0.20; raising it back to 0.30 brings back "
+            "the 'jamming never shows' bug."
+        )
+
+
+# ---------------------------------------------------------------------------
+# Overrides honored
+# ---------------------------------------------------------------------------
+
+
+class TestOverridesHonored:
+    def test_overrides_supersede_constants(self):
+        """The public signature accepts overrides so an operator can
+        re-tune at the call site (e.g. for a more aggressive setup in
+        an active conflict zone) without editing the module constants."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        feed = [
+            {"hex": "a1", "lat": 40.1, "lng": -100.1, "nac_p": 3},
+            {"hex": "a2", "lat": 40.2, "lng": -100.2, "nac_p": 3},
+        ]
+
+        # With defaults (min_aircraft=3) this is blocked. With override=2 it fires.
+        assert detect_gps_jamming_zones(feed) == []
+        zones = detect_gps_jamming_zones(feed, min_aircraft=2)
+        assert len(zones) == 1
+
+
+# ---------------------------------------------------------------------------
+# lon vs lng compatibility
+# ---------------------------------------------------------------------------
+
+
+class TestLonLngCompat:
+    def test_lon_key_accepted(self):
+        """adsb.lol records arrive with ``lon`` (no g). The OpenSky merge
+        normalizes to ``lng`` but raw records flowing into the detector
+        may use either. Make sure both work."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        feed = [
+            {"hex": "a1", "lat": 40.1, "lon": -100.1, "nac_p": 0},
+            {"hex": "a2", "lat": 40.2, "lon": -100.2, "nac_p": 0},
+            {"hex": "a3", "lat": 40.3, "lon": -100.3, "nac_p": 0},
+        ]
+
+        zones = detect_gps_jamming_zones(feed)
+
+        assert len(zones) == 1
+
+
+# ---------------------------------------------------------------------------
+# Empty / malformed inputs don't crash
+# ---------------------------------------------------------------------------
+
+
+class TestRobustness:
+    def test_empty_feed(self):
+        from services.fetchers.flights import detect_gps_jamming_zones
+        assert detect_gps_jamming_zones([]) == []
+
+    def test_none_feed(self):
+        """The wrapper at the call site passes ``raw_flights_snapshot``
+        which could in principle be None on a startup race. Handle it."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+        assert detect_gps_jamming_zones(None) == []
+
+    def test_records_missing_position_skipped(self):
+        from services.fetchers.flights import detect_gps_jamming_zones
+        feed = [
+            {"hex": "noloc", "nac_p": 0},
+            {"hex": "nolat", "lng": -100.0, "nac_p": 0},
+            {"hex": "nolng", "lat": 40.0, "nac_p": 0},
+        ]
+        assert detect_gps_jamming_zones(feed) == []

backend/tests/test_uap_hf_fallback_cutoff.py

@@ -0,0 +1,252 @@
+"""HF NUFORC fallback honors the rolling cutoff window.
+
+Background
+----------
+The UAP sightings layer is sourced primarily from a live scrape of
+nuforc.org. When that fails (Cloudflare 403, curl disabled on Windows,
+wdtNonce regex stale, etc.) the code falls back to a static CSV mirror
+hosted on Hugging Face at ``kcimc/NUFORC/nuforc_str.csv``.
+
+The HF mirror is maintained by a third party and refreshed sporadically.
+Pre-fix, the fallback parsed every row, sorted by ``occurred`` descending,
+and took the top 250 — **with no date cutoff**. When the HF mirror is
+stale (its "newest" rows are ~2-3 years old), users saw a map full of
+2022-2023 sightings labeled as the "last 60 days" layer.
+
+These tests pin the new behavior:
+
+* Rows older than ``_NUFORC_RECENT_DAYS`` are dropped before the take-top-N.
+* If the HF mirror has nothing in the window, the fallback returns ``[]``
+  and logs ERROR (don't silently serve stale data).
+* ``fetch_uap_sightings`` records the failure when BOTH paths fail, so
+  the layer shows as broken in the health registry instead of "fresh".
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime as real_datetime
+
+
+class _FixedDateTime(real_datetime):
+    """A datetime whose utcnow() returns a pinned value, for deterministic
+    cutoff math. Subclasses real datetime so existing operations still work."""
+
+    @classmethod
+    def utcnow(cls):
+        return cls(2026, 5, 1, 12, 0, 0)
+
+
+class _StubResponse:
+    status_code = 200
+
+    def __init__(self, text: str):
+        self.text = text
+
+
+def _stub_geocode_cache(*_args, **_kwargs):
+    """Pre-populated location cache so the fallback doesn't try to hit
+    Photon during the test."""
+    return {
+        "Denver, CO, USA": [39.7392, -104.9903],
+        "Seattle, WA, USA": [47.6062, -122.3321],
+        "Phoenix, AZ, USA": [33.4484, -112.0740],
+    }
+
+
+def test_hf_fallback_drops_rows_older_than_60_days(monkeypatch):
+    """Pre-fix: a row from 2023 would make it into the layer if it was
+    among the newest 250 in the HF mirror. Post-fix: it's filtered out
+    before we even count to 250."""
+    from services.fetchers import earth_observation as eo
+
+    # 2026-05-01 - 60 days = 2026-03-02. So 2026-03-01 is one day too old.
+    csv_text = (
+        "Sighting,Occurred,Location,Shape,Duration,Posted,Summary\n"
+        '1,2026-04-15 21:00:00 Local,"Denver, CO, USA",Triangle,5 minutes,2026-04-16,"In-window sighting"\n'
+        '2,2023-06-01 21:00:00 Local,"Seattle, WA, USA",Light,30 seconds,2023-06-02,"Three years old"\n'
+        '3,2022-01-15 20:00:00 Local,"Phoenix, AZ, USA",Disk,2 minutes,2022-01-16,"Even older"\n'
+    )
+
+    monkeypatch.setattr(eo, "datetime", _FixedDateTime)
+    monkeypatch.setattr(eo, "fetch_with_curl", lambda *a, **kw: _StubResponse(csv_text))
+    monkeypatch.setattr(eo, "_load_nuforc_location_cache", _stub_geocode_cache)
+    monkeypatch.setattr(eo, "_save_nuforc_location_cache", lambda cache: None)
+    # If the cutoff is missing, the geocoder may still get called for the
+    # 2022/2023 rows. We assert geocoder is NEVER invoked for stale rows.
+    geocode_calls: list[str] = []
+
+    def _geocode_spy(location, city, state, country=""):
+        geocode_calls.append(location)
+        return None  # already in cache, shouldn't be hit anyway
+
+    monkeypatch.setattr(eo, "_geocode_uap_location", _geocode_spy)
+
+    sightings = eo._build_uap_sightings_from_hf_mirror()
+
+    ids = [s["id"] for s in sightings]
+    assert ids == ["NUFORC-1"], f"only the 2026 row should survive: got {ids}"
+    # Stale rows must not have been geocoded — they should be dropped
+    # before the geocoding loop is reached.
+    assert geocode_calls == []
+
+
+def test_hf_fallback_returns_empty_when_mirror_is_fully_stale(monkeypatch, caplog):
+    """The smoking-gun case: the HF mirror is so stale that NO rows are
+    within the rolling window. Pre-fix returned 250 ancient rows. Post-fix
+    returns ``[]`` and logs ERROR so the operator knows the layer is dead."""
+    from services.fetchers import earth_observation as eo
+
+    csv_text = (
+        "Sighting,Occurred,Location,Shape,Duration,Posted,Summary\n"
+        '1,2023-04-15 21:00:00 Local,"Denver, CO, USA",Triangle,5 minutes,2023-04-16,"Old"\n'
+        '2,2022-06-01 21:00:00 Local,"Seattle, WA, USA",Light,30 seconds,2022-06-02,"Older"\n'
+        '3,2021-01-15 20:00:00 Local,"Phoenix, AZ, USA",Disk,2 minutes,2021-01-16,"Ancient"\n'
+    )
+
+    monkeypatch.setattr(eo, "datetime", _FixedDateTime)
+    monkeypatch.setattr(eo, "fetch_with_curl", lambda *a, **kw: _StubResponse(csv_text))
+    monkeypatch.setattr(eo, "_load_nuforc_location_cache", _stub_geocode_cache)
+    monkeypatch.setattr(eo, "_save_nuforc_location_cache", lambda cache: None)
+    monkeypatch.setattr(eo, "_geocode_uap_location", lambda *a, **kw: None)
+
+    with caplog.at_level(logging.ERROR, logger="services.fetchers.earth_observation"):
+        sightings = eo._build_uap_sightings_from_hf_mirror()
+
+    assert sightings == []
+    # The error log should mention how many stale rows were dropped so the
+    # operator can tell the mirror is the problem (not "we got 0 rows" which
+    # could also mean the download failed).
+    relevant = [r for r in caplog.records if "HF fallback yielded 0 rows" in r.getMessage()]
+    assert relevant, "expected loud ERROR when HF mirror is fully stale"
+    # The message should report the count of dropped stale rows.
+    assert any("dropped 3" in r.getMessage() for r in relevant)
+
+
+def test_hf_fallback_still_returns_data_when_some_rows_are_in_window(monkeypatch):
+    """Mixed-age mirror: some rows in the window, some not. The fallback
+    should return only the in-window rows and not log the doomsday ERROR."""
+    from services.fetchers import earth_observation as eo
+
+    csv_text = (
+        "Sighting,Occurred,Location,Shape,Duration,Posted,Summary\n"
+        '1,2026-04-15 21:00:00 Local,"Denver, CO, USA",Triangle,5 minutes,2026-04-16,"Fresh"\n'
+        '2,2026-04-10 21:00:00 Local,"Seattle, WA, USA",Light,30 seconds,2026-04-10,"Also fresh"\n'
+        '3,2020-01-15 20:00:00 Local,"Phoenix, AZ, USA",Disk,2 minutes,2020-01-16,"Ancient"\n'
+    )
+
+    monkeypatch.setattr(eo, "datetime", _FixedDateTime)
+    monkeypatch.setattr(eo, "fetch_with_curl", lambda *a, **kw: _StubResponse(csv_text))
+    monkeypatch.setattr(eo, "_load_nuforc_location_cache", _stub_geocode_cache)
+    monkeypatch.setattr(eo, "_save_nuforc_location_cache", lambda cache: None)
+    monkeypatch.setattr(eo, "_geocode_uap_location", lambda *a, **kw: None)
+
+    sightings = eo._build_uap_sightings_from_hf_mirror()
+
+    ids = sorted(s["id"] for s in sightings)
+    assert ids == ["NUFORC-1", "NUFORC-2"], f"only in-window rows should appear: got {ids}"
+
+
+def test_fetch_uap_sightings_marks_failure_when_both_paths_empty(monkeypatch, caplog):
+    """When the live path raises AND the HF fallback returns empty,
+    ``fetch_uap_sightings`` must:
+      * NOT mark the layer fresh (pre-fix bug: it did, so the layer
+        showed as healthy-but-empty for days)
+      * call ``assert_canary("uap_sightings", 0)`` so the health
+        registry surfaces the broken layer
+      * log an ERROR with the live-path exception for debugging
+    """
+    from services.fetchers import earth_observation as eo
+    from services.fetchers import _store
+
+    monkeypatch.setattr(_store, "is_any_active", lambda layer: True)
+    monkeypatch.setattr(eo, "_load_nuforc_sightings_cache", lambda force_refresh=False: None)
+
+    def _boom():
+        raise RuntimeError("NUFORC live: zero rows pulled across 3 months")
+
+    monkeypatch.setattr(eo, "_build_recent_uap_sightings", _boom)
+    monkeypatch.setattr(eo, "_build_uap_sightings_from_hf_mirror", lambda: [])
+
+    marked: list[str] = []
+    monkeypatch.setattr(eo, "_mark_fresh", lambda *keys: marked.extend(keys))
+
+    canary_calls: list[tuple[str, int]] = []
+    import services.slo as slo
+    monkeypatch.setattr(
+        slo, "assert_canary", lambda key, value: canary_calls.append((key, int(value)))
+    )
+
+    with caplog.at_level(logging.ERROR, logger="services.fetchers.earth_observation"):
+        eo.fetch_uap_sightings()
+
+    assert marked == [], "broken layer must NOT be marked fresh"
+    assert canary_calls == [("uap_sightings", 0)], (
+        f"expected canary trip when both paths fail; got {canary_calls}"
+    )
+    # The live error message should propagate into the error log so the
+    # operator can tell live failed AND fallback was empty (not the other
+    # way around).
+    assert any(
+        "both live NUFORC and HF fallback" in r.getMessage()
+        for r in caplog.records
+    )
+
+
+def test_fetch_uap_sightings_succeeds_when_fallback_returns_data(monkeypatch):
+    """Positive path: live fails, fallback returns rows. The layer is
+    populated and marked fresh; assert_canary is NOT tripped (we only
+    trip the canary when the layer has zero data)."""
+    from services.fetchers import earth_observation as eo
+    from services.fetchers import _store
+
+    monkeypatch.setattr(_store, "is_any_active", lambda layer: True)
+    monkeypatch.setattr(eo, "_load_nuforc_sightings_cache", lambda force_refresh=False: None)
+    monkeypatch.setattr(
+        eo, "_build_recent_uap_sightings", lambda: (_ for _ in ()).throw(RuntimeError("live down"))
+    )
+
+    fallback_rows = [{"id": "NUFORC-fb-1", "date_time": "2026-04-20", "lat": 0.0, "lng": 0.0}]
+    monkeypatch.setattr(eo, "_build_uap_sightings_from_hf_mirror", lambda: fallback_rows)
+    monkeypatch.setattr(eo, "_save_nuforc_sightings_cache", lambda s: None)
+
+    marked: list[str] = []
+    monkeypatch.setattr(eo, "_mark_fresh", lambda *keys: marked.extend(keys))
+
+    canary_calls: list[tuple[str, int]] = []
+    import services.slo as slo
+    monkeypatch.setattr(
+        slo, "assert_canary", lambda key, value: canary_calls.append((key, int(value)))
+    )
+
+    eo.fetch_uap_sightings()
+
+    assert marked == ["uap_sightings"]
+    assert canary_calls == [], "canary should not trip when fallback supplies data"
+
+
+def test_uap_scheduler_runs_weekly_not_daily():
+    """The cron job for the UAP layer must be configured for Mondays at
+    12:00 UTC, not daily. Daily was the pre-fix default; weekly matches
+    the layer's stated cadence (a rolling 60-day digest) and keeps load
+    on nuforc.org light."""
+    from services import data_fetcher
+
+    src = data_fetcher.__file__
+    with open(src, "r", encoding="utf-8") as f:
+        text = f.read()
+
+    # Anchor on the scheduler block by id, then assert the cron triggers.
+    assert "uap_sightings_weekly" in text, (
+        "scheduler id should be uap_sightings_weekly (was uap_sightings_daily pre-fix)"
+    )
+    # The day_of_week directive is the difference between daily and weekly.
+    # If somebody flips it back to daily, this fires.
+    weekly_block = text.split("uap_sightings_weekly", 1)[0]
+    # Walk backwards for the matching add_job call.
+    add_job_idx = weekly_block.rfind("add_job(")
+    assert add_job_idx >= 0, "could not locate add_job block for UAP scheduler"
+    job_block = text[add_job_idx : text.find(")", text.index("uap_sightings_weekly")) + 1]
+    assert 'day_of_week="mon"' in job_block, (
+        f"expected day_of_week='mon' in UAP scheduler block:\n{job_block}"
+    )

desktop-shell/tauri-skeleton/scripts/build-frontend-export.cjs

@@ -46,12 +46,18 @@ function prepareBuildTree() {
   const stagedLayoutPath = path.join(buildFrontendDir, 'src', 'app', 'layout.tsx');
   if (fs.existsSync(stagedLayoutPath)) {
     const layoutSource = fs.readFileSync(stagedLayoutPath, 'utf8');
+    // CRLF compatibility: on Windows checkouts without ``core.autocrlf=input``
+    // (the default) layout.tsx has CRLF line endings, but the original regexes
+    // only matched LF. The strip silently no-op'd, ``force-dynamic`` stayed,
+    // and Next's static-export refused to render ``/_not-found`` ("Page with
+    // `dynamic = \"force-dynamic\"` couldn't be exported"). Use ``\r?\n`` so
+    // the strip works regardless of line-ending normalization.
     fs.writeFileSync(
       stagedLayoutPath,
       layoutSource
-        .replace(/\n\/\/ The dashboard is a live local runtime[\s\S]*?client polling ever hydrates\.\n/g, '\n')
-        .replace(/\nexport const dynamic = ['"]force-dynamic['"];\n/g, '\n')
-        .replace(/\nexport const revalidate = 0;\n/g, '\n'),
+        .replace(/\r?\n\/\/ The dashboard is a live local runtime[\s\S]*?client polling ever hydrates\.\r?\n/g, '\n')
+        .replace(/\r?\nexport const dynamic = ['"]force-dynamic['"];\r?\n/g, '\n')
+        .replace(/\r?\nexport const revalidate = 0;\r?\n/g, '\n'),
     );
   }
 

frontend/src/components/NewsFeed.tsx

@@ -249,34 +249,70 @@ const VESSEL_TYPE_WIKI: Record<string, string> = {
 
 type FlightTrailPoint = { lat?: number; lng?: number; alt?: number; ts?: number } | number[];
 
+function formatObservedDuration(seconds: number): string {
+    // Compact "1h 14m" / "23m" / "45s" — matches the density of the rest
+    // of the flight tooltip. < 60s is shown as "<1m" so the user knows
+    // we've JUST started observing this hex (cumulative will still be 0).
+    if (!Number.isFinite(seconds) || seconds <= 0) return '<1m';
+    if (seconds < 60) return '<1m';
+    const totalMinutes = Math.floor(seconds / 60);
+    const hours = Math.floor(totalMinutes / 60);
+    const minutes = totalMinutes % 60;
+    if (hours > 0) return `${hours}h ${minutes}m`;
+    return `${minutes}m`;
+}
+
 function EmissionsEstimateBlock({ flight }: { flight: any }) {
     const emissions = flight?.emissions;
-    const context = emissions ? 'Model-based cruise estimate' : null;
+    // Cumulative fuel/CO2 since the backend first saw this hex this
+    // flight session. Prefer these big numbers — the user explicitly
+    // wanted "the actual fuel that has been burned", not the rate.
+    // Rates are still shown below as smaller context.
+    const observedSec = Number(emissions?.observed_seconds ?? 0);
+    const fuelBurned = Number(emissions?.fuel_gallons_burned ?? 0);
+    const co2Emitted = Number(emissions?.co2_kg_emitted ?? 0);
+    const haveCumulative = emissions && observedSec > 0;
 
     return (
         <div className="border-b border-[var(--border-primary)] pb-2">
             <span className="text-[var(--text-muted)] text-[10px] block mb-1.5">EMISSIONS ESTIMATE</span>
             <div className="flex gap-3">
                 <div className="flex-1 bg-[var(--bg-primary)]/50 border border-[var(--border-primary)] px-2 py-1.5">
-                    <div className="text-[11px] text-[var(--text-muted)] tracking-widest">FUEL RATE</div>
-                    <div className="text-xs font-bold text-orange-400">
-                        {emissions ? (
-                            <>{emissions.fuel_gph} <span className="text-[11px] text-[var(--text-muted)] font-normal">GPH</span></>
+                    <div className="text-[11px] text-[var(--text-muted)] tracking-widest">FUEL BURNED</div>
+                    <div className="text-sm font-bold text-orange-400">
+                        {haveCumulative ? (
+                            <>{fuelBurned.toLocaleString(undefined, { maximumFractionDigits: 1 })} <span className="text-[11px] text-[var(--text-muted)] font-normal">gal</span></>
+                        ) : emissions ? (
+                            <span className="text-[var(--text-muted)] font-normal text-xs">—</span>
                         ) : 'UNKNOWN'}
                     </div>
+                    {emissions && (
+                        <div className="text-[10px] text-[var(--text-muted)] mt-0.5">
+                            @ {emissions.fuel_gph} gph
+                        </div>
+                    )}
                 </div>
                 <div className="flex-1 bg-[var(--bg-primary)]/50 border border-[var(--border-primary)] px-2 py-1.5">
-                    <div className="text-[11px] text-[var(--text-muted)] tracking-widest">CO2 RATE</div>
-                    <div className="text-xs font-bold text-red-400">
-                        {emissions ? (
-                            <>{emissions.co2_kg_per_hour.toLocaleString()} <span className="text-[11px] text-[var(--text-muted)] font-normal">KG/HR</span></>
+                    <div className="text-[11px] text-[var(--text-muted)] tracking-widest">CO2 EMITTED</div>
+                    <div className="text-sm font-bold text-red-400">
+                        {haveCumulative ? (
+                            <>{co2Emitted.toLocaleString(undefined, { maximumFractionDigits: 1 })} <span className="text-[11px] text-[var(--text-muted)] font-normal">kg</span></>
+                        ) : emissions ? (
+                            <span className="text-[var(--text-muted)] font-normal text-xs">—</span>
                         ) : 'UNKNOWN'}
                     </div>
+                    {emissions && (
+                        <div className="text-[10px] text-[var(--text-muted)] mt-0.5">
+                            @ {emissions.co2_kg_per_hour.toLocaleString()} kg/hr
+                        </div>
+                    )}
                 </div>
             </div>
-            {context && (
+            {emissions && (
                 <div className="mt-1.5 text-[10px] text-[var(--text-muted)] leading-relaxed">
-                    {context}
+                    {haveCumulative
+                        ? `Observed in flight for ${formatObservedDuration(observedSec)} · model-based cruise estimate`
+                        : 'Just observed · totals will appear on next refresh'}
                 </div>
             )}
         </div>