perf: UX-safe fetch trimming and instant layer-enable refresh.

Drop duplicate slow-tier weather/ukraine jobs, gate correlations when off, slim health probes, keyed layer-panel subscriptions, align backend layer defaults with the dashboard, and fetch CCTV/FIRMS/PSK/etc. synchronously on enable so toggles stay responsive without background prefetch waste.

backend/main.py

@@ -3927,7 +3927,9 @@ class LayerUpdate(BaseModel):
 async def update_layers(update: LayerUpdate, request: Request):
     """Receive frontend layer toggle state. Starts/stops streams accordingly."""
     from services.fetchers._store import active_layers, bump_active_layers_version, is_any_active
+    from services.layer_enable_refresh import refresh_newly_enabled_layers, snapshot_active_layers
 
+    layers_before = snapshot_active_layers()
     # Snapshot old stream states before applying changes
     old_ships = is_any_active(
         "ships_military", "ships_cargo", "ships_civilian", "ships_passenger", "ships_tracked_yachts"
@@ -3935,8 +3937,6 @@ async def update_layers(update: LayerUpdate, request: Request):
     old_mesh = is_any_active("sigint_meshtastic")
     old_aprs = is_any_active("sigint_aprs")
     old_viirs = is_any_active("viirs_nightlights")
-    old_datacenters = is_any_active("datacenters")
-    old_fishing = is_any_active("fishing_activity")
 
     # Update only known keys
     changed = False
@@ -3955,8 +3955,6 @@ async def update_layers(update: LayerUpdate, request: Request):
     new_mesh = is_any_active("sigint_meshtastic")
     new_aprs = is_any_active("sigint_aprs")
     new_viirs = is_any_active("viirs_nightlights")
-    new_datacenters = is_any_active("datacenters")
-    new_fishing = is_any_active("fishing_activity")
 
     # Start/stop AIS stream on transition
     if old_ships and not new_ships:
@@ -4012,17 +4010,7 @@ async def update_layers(update: LayerUpdate, request: Request):
         _queue_viirs_change_refresh()
         logger.info("VIIRS change refresh queued (layer enabled)")
 
-    if not old_datacenters and new_datacenters:
-        from services.fetchers.infrastructure import fetch_datacenters
-
-        fetch_datacenters()
-        logger.info("Datacenters loaded (layer enabled)")
-
-    if not old_fishing and new_fishing:
-        from services.fetchers.geo import fetch_fishing_activity
-
-        fetch_fishing_activity()
-        logger.info("Fishing activity refresh queued (layer enabled)")
+    refresh_newly_enabled_layers(layers_before)
 
     return {"status": "ok"}
 

backend/routers/data.py

@@ -498,12 +498,13 @@ def _run_prediction_markets_disable() -> None:
 async def update_layers(update: LayerUpdate, request: Request):
     """Receive frontend layer toggle state. Starts/stops streams accordingly."""
     from services.fetchers._store import active_layers, bump_active_layers_version, is_any_active
+    from services.layer_enable_refresh import refresh_newly_enabled_layers, snapshot_active_layers
+
+    layers_before = snapshot_active_layers()
     old_ships = is_any_active("ships_military", "ships_cargo", "ships_civilian", "ships_passenger", "ships_tracked_yachts")
     old_mesh = is_any_active("sigint_meshtastic")
     old_aprs = is_any_active("sigint_aprs")
     old_viirs = is_any_active("viirs_nightlights")
-    old_datacenters = is_any_active("datacenters")
-    old_fishing = is_any_active("fishing_activity")
     changed = False
     for key, value in update.layers.items():
         if key in active_layers:
@@ -516,8 +517,6 @@ async def update_layers(update: LayerUpdate, request: Request):
     new_mesh = is_any_active("sigint_meshtastic")
     new_aprs = is_any_active("sigint_aprs")
     new_viirs = is_any_active("viirs_nightlights")
-    new_datacenters = is_any_active("datacenters")
-    new_fishing = is_any_active("fishing_activity")
     if old_ships and not new_ships:
         from services.ais_stream import stop_ais_stream
         stop_ais_stream()
@@ -561,16 +560,7 @@ async def update_layers(update: LayerUpdate, request: Request):
     if not old_viirs and new_viirs:
         _queue_viirs_change_refresh()
         logger.info("VIIRS change refresh queued (layer enabled)")
-    if not old_datacenters and new_datacenters:
-        from services.fetchers.infrastructure import fetch_datacenters
-
-        fetch_datacenters()
-        logger.info("Datacenters loaded (layer enabled)")
-    if not old_fishing and new_fishing:
-        from services.fetchers.geo import fetch_fishing_activity
-
-        fetch_fishing_activity()
-        logger.info("Fishing activity refresh queued (layer enabled)")
+    refresh_newly_enabled_layers(layers_before)
     return {"status": "ok"}
 
 

backend/routers/health.py

@@ -4,10 +4,50 @@ from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 from limiter import limiter
 from auth import require_admin
-from services.data_fetcher import get_latest_data
 from services.schemas import HealthResponse
 import os
 
+# Health/SLO probes only need counts + freshness — not a full dashboard deepcopy.
+_HEALTH_DATA_KEYS: tuple[str, ...] = (
+    "last_updated",
+    "commercial_flights",
+    "military_flights",
+    "private_jets",
+    "ships",
+    "satellites",
+    "earthquakes",
+    "cctv",
+    "news",
+    "uavs",
+    "firms_fires",
+    "liveuamap",
+    "gdelt",
+    "uap_sightings",
+    "wastewater",
+    "fimi",
+    "space_weather",
+    "weather_alerts",
+    "volcanoes",
+    "prediction_markets",
+)
+
+
+def _health_data_snapshot() -> dict:
+    from services.fetchers._store import get_latest_data_subset_refs
+    from services.slo import SLO_REGISTRY
+
+    keys = tuple(dict.fromkeys((*_HEALTH_DATA_KEYS, *SLO_REGISTRY.keys())))
+    return get_latest_data_subset_refs(*keys)
+
+
+def _health_row_count(value) -> int:
+    if value is None:
+        return 0
+    try:
+        return len(value)
+    except TypeError:
+        return 0
+
 APP_VERSION = os.environ.get("_HEALTH_APP_VERSION", "0.9.82")
 
 router = APIRouter()
@@ -41,7 +81,7 @@ async def health_check(request: Request):
     from services.fetchers._store import get_source_timestamps_snapshot
     from services.slo import compute_all_statuses, summarise_statuses
 
-    d = get_latest_data()
+    d = _health_data_snapshot()
     last = d.get("last_updated")
     timestamps = get_source_timestamps_snapshot()
     slo_statuses = compute_all_statuses(d, timestamps)
@@ -102,18 +142,18 @@ async def health_check(request: Request):
         "version": _get_app_version(),
         "last_updated": last,
         "sources": {
-            "flights": len(d.get("commercial_flights", [])),
-            "military": len(d.get("military_flights", [])),
-            "ships": len(d.get("ships", [])),
-            "satellites": len(d.get("satellites", [])),
-            "earthquakes": len(d.get("earthquakes", [])),
-            "cctv": len(d.get("cctv", [])),
-            "news": len(d.get("news", [])),
-            "uavs": len(d.get("uavs", [])),
-            "firms_fires": len(d.get("firms_fires", [])),
-            "liveuamap": len(d.get("liveuamap", [])),
-            "gdelt": len(d.get("gdelt", [])),
-            "uap_sightings": len(d.get("uap_sightings", [])),
+            "flights": _health_row_count(d.get("commercial_flights")),
+            "military": _health_row_count(d.get("military_flights")),
+            "ships": _health_row_count(d.get("ships")),
+            "satellites": _health_row_count(d.get("satellites")),
+            "earthquakes": _health_row_count(d.get("earthquakes")),
+            "cctv": _health_row_count(d.get("cctv")),
+            "news": _health_row_count(d.get("news")),
+            "uavs": _health_row_count(d.get("uavs")),
+            "firms_fires": _health_row_count(d.get("firms_fires")),
+            "liveuamap": _health_row_count(d.get("liveuamap")),
+            "gdelt": _health_row_count(d.get("gdelt")),
+            "uap_sightings": _health_row_count(d.get("uap_sightings")),
         },
         "freshness": timestamps,
         "uptime_seconds": round(_time_mod.time() - _get_start_time()),
@@ -127,4 +167,7 @@ async def health_check(request: Request):
 @router.get("/api/debug-latest", dependencies=[Depends(require_admin)])
 @limiter.limit("30/minute")
 async def debug_latest_data(request: Request):
-    return list(get_latest_data().keys())
+    from services.fetchers._store import latest_data, _data_lock
+
+    with _data_lock:
+        return list(latest_data.keys())

backend/services/data_fetcher.py

@@ -479,24 +479,27 @@ def update_slow_data():
         fetch_military_bases,
         fetch_scanners,
         fetch_psk_reporter,
-        fetch_weather_alerts,
+        # weather_alerts + ukraine_alerts: owned by dedicated scheduler jobs
+        # (5 min and 2 min) — keep off slow tier to avoid duplicate upstream work.
         fetch_air_quality,
         fetch_fishing_activity,
         fetch_power_plants,
-        fetch_ukraine_air_raid_alerts,
         fetch_malware_threats,
         fetch_cyber_threats,
         fetch_scm_suppliers,
     ]
     _run_tasks("slow-tier", slow_funcs)
-    # Run correlation engine after all data is fresh
+    # Run correlation engine after all data is fresh (skip when overlay is off).
     try:
+        from services.fetchers._store import is_any_active
         from services.correlation_engine import compute_correlations
-        with _data_lock:
-            snapshot = dict(latest_data)
-        correlations = compute_correlations(snapshot)
-        with _data_lock:
-            latest_data["correlations"] = correlations
+
+        if is_any_active("correlations"):
+            with _data_lock:
+                snapshot = dict(latest_data)
+            correlations = compute_correlations(snapshot)
+            with _data_lock:
+                latest_data["correlations"] = correlations
     except Exception as e:
         logger.error("Correlation engine failed: %s", e)
     try:

backend/services/fetchers/_store.py

@@ -334,15 +334,15 @@ active_layers: dict[str, bool] = {
     "ships_passenger": True,
     "ships_tracked_yachts": True,
     "earthquakes": True,
-    "cctv": True,
+    "cctv": False,
     "ukraine_frontline": True,
     "global_incidents": True,
     "gps_jamming": True,
     "kiwisdr": True,
     "scanners": True,
-    "firms": True,
+    "firms": False,
     "internet_outages": True,
-    "datacenters": True,
+    "datacenters": False,
     "military_bases": True,
     "sigint_meshtastic": True,
     "sigint_aprs": True,
@@ -353,9 +353,9 @@ active_layers: dict[str, bool] = {
     "satnogs": True,
     "tinygs": True,
     "ukraine_alerts": True,
-    "power_plants": True,
+    "power_plants": False,
     "viirs_nightlights": False,
-    "psk_reporter": True,
+    "psk_reporter": False,
     "correlations": True,
     "contradictions": True,
     "uap_sightings": True,

backend/services/layer_enable_refresh.py

@@ -0,0 +1,78 @@
+"""Immediate data refresh when the operator enables a map layer.
+
+Runs synchronously inside POST /api/layers so the frontend's post-toggle
+live-data refetch sees populated payloads (T_toggle_visible guardrail).
+"""
+from __future__ import annotations
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+def snapshot_active_layers() -> dict[str, bool]:
+    from services.fetchers._store import active_layers
+
+    return dict(active_layers)
+
+
+def refresh_newly_enabled_layers(before: dict[str, bool]) -> None:
+    """Fetch any layers that transitioned off → on."""
+    from services.fetchers._store import active_layers, bump_data_version
+
+    refreshed = False
+
+    def _enabled(key: str) -> bool:
+        return bool(active_layers.get(key, False))
+
+    def _was_off_now_on(key: str) -> bool:
+        return not bool(before.get(key, False)) and _enabled(key)
+
+    if _was_off_now_on("cctv"):
+        from services.fetchers.infrastructure import fetch_cctv
+
+        fetch_cctv()
+        refreshed = True
+        logger.info("CCTV loaded (layer enabled)")
+
+    if _was_off_now_on("firms"):
+        from services.fetchers.earth_observation import (
+            fetch_firms_country_fires,
+            fetch_firms_fires,
+        )
+
+        fetch_firms_fires()
+        fetch_firms_country_fires()
+        refreshed = True
+        logger.info("FIRMS fires loaded (layer enabled)")
+
+    if _was_off_now_on("power_plants"):
+        from services.fetchers.infrastructure import fetch_power_plants
+
+        fetch_power_plants()
+        refreshed = True
+        logger.info("Power plants loaded (layer enabled)")
+
+    if _was_off_now_on("psk_reporter"):
+        from services.fetchers.infrastructure import fetch_psk_reporter
+
+        fetch_psk_reporter()
+        refreshed = True
+        logger.info("PSK Reporter loaded (layer enabled)")
+
+    if _was_off_now_on("datacenters"):
+        from services.fetchers.infrastructure import fetch_datacenters
+
+        fetch_datacenters()
+        refreshed = True
+        logger.info("Datacenters loaded (layer enabled)")
+
+    if _was_off_now_on("fishing_activity"):
+        from services.fetchers.geo import fetch_fishing_activity
+
+        fetch_fishing_activity()
+        refreshed = True
+        logger.info("Fishing activity loaded (layer enabled)")
+
+    if refreshed:
+        bump_data_version()

backend/tests/test_layer_enable_integration.py

@@ -0,0 +1,18 @@
+"""Integration: layer enable triggers immediate data availability."""
+from __future__ import annotations
+
+from services.fetchers._store import active_layers, latest_data, _data_lock
+
+
+def test_firms_enable_populates_slow_payload(client):
+    with _data_lock:
+        active_layers["firms"] = False
+        latest_data["firms_fires"] = []
+
+    r = client.post("/api/layers", json={"layers": {"firms": True}})
+    assert r.status_code == 200
+
+    slow = client.get("/api/live-data/slow")
+    assert slow.status_code == 200
+    fires = slow.json().get("firms_fires") or []
+    assert len(fires) > 0, "firms layer should populate on enable without waiting for scheduler"

backend/tests/test_layer_enable_refresh.py

@@ -0,0 +1,36 @@
+"""Tests for on-enable layer refresh (Phase 2 UX guardrail)."""
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from services.fetchers._store import active_layers, bump_active_layers_version
+from services.layer_enable_refresh import refresh_newly_enabled_layers, snapshot_active_layers
+
+
+def test_refresh_firms_on_enable_only():
+    before = snapshot_active_layers()
+    active_layers["firms"] = True
+    bump_active_layers_version()
+
+    with (
+        patch("services.fetchers.earth_observation.fetch_firms_fires") as firms,
+        patch("services.fetchers.earth_observation.fetch_firms_country_fires") as country,
+        patch("services.fetchers._store.bump_data_version") as bump,
+    ):
+        refresh_newly_enabled_layers({**before, "firms": False})
+
+    firms.assert_called_once()
+    country.assert_called_once()
+    bump.assert_called_once()
+
+    active_layers["firms"] = before.get("firms", False)
+
+
+def test_refresh_skips_when_layer_stays_off():
+    before = {**snapshot_active_layers(), "cctv": False}
+    active_layers["cctv"] = False
+
+    with patch("services.fetchers.infrastructure.fetch_cctv") as fetch_cctv:
+        refresh_newly_enabled_layers(before)
+
+    fetch_cctv.assert_not_called()

backend/tests/test_perf_safe_optimizations.py

@@ -0,0 +1,59 @@
+"""Regression tests for UX-safe performance optimizations."""
+from __future__ import annotations
+
+import inspect
+
+
+def test_slow_tier_skips_duplicate_time_critical_fetchers():
+    """Weather + Ukraine alerts have dedicated scheduler jobs — not slow tier."""
+    from services import data_fetcher
+
+    source = inspect.getsource(data_fetcher.update_slow_data)
+    slow_block = source.split("_run_tasks(\"slow-tier\"", 1)[0]
+    assert "fetch_weather_alerts" not in slow_block
+    assert "fetch_ukraine_air_raid_alerts" not in slow_block
+
+
+def test_slow_tier_gates_correlation_engine_on_active_layer():
+    from services import data_fetcher
+
+    source = inspect.getsource(data_fetcher.update_slow_data)
+    assert 'is_any_active("correlations")' in source
+
+
+def test_health_uses_subset_refs_not_full_deepcopy():
+    from routers import health as health_router
+
+    source = inspect.getsource(health_router.health_check)
+    assert "_health_data_snapshot()" in source
+    assert "get_latest_data()" not in source
+
+    snap_source = inspect.getsource(health_router._health_data_snapshot)
+    assert "get_latest_data_subset_refs" in snap_source
+    assert "deepcopy" not in snap_source
+
+
+def test_active_layers_defaults_match_dashboard_first_paint():
+    """Backend must not prefetch layers the dashboard starts with disabled."""
+    from services.fetchers import _store
+
+    off_by_default = {
+        "cctv": False,
+        "firms": False,
+        "datacenters": False,
+        "power_plants": False,
+        "psk_reporter": False,
+        "viirs_nightlights": False,
+        "crowdthreat": False,
+        "gt_risk": False,
+    }
+    for key, expected in off_by_default.items():
+        assert _store.active_layers.get(key) is expected, key
+
+
+def test_layer_enable_refresh_covers_cold_toggle_layers():
+    from services import layer_enable_refresh
+
+    source = inspect.getsource(layer_enable_refresh.refresh_newly_enabled_layers)
+    for key in ("cctv", "firms", "power_plants", "psk_reporter", "datacenters"):
+        assert f'"{key}"' in source or f"'{key}'" in source

docker-compose.participant.yml

@@ -17,6 +17,9 @@ services:
       WORMHOLE_STARTUP_DEADLINE_S: "90"
       GT_ANALYTICS_ENABLED: "false"
       GT_ANALYTICS_PROFILE: "lean"
+      # Lean 1-vCPU nodes: fewer fetch worker threads reduces scheduler contention.
+      SHADOWBROKER_FETCH_WORKERS: "4"
+      SHADOWBROKER_HEAVY_FETCH_WORKERS: "1"
     deploy:
       resources:
         limits:

frontend/src/components/WorldviewLeftPanel.tsx

@@ -143,8 +143,61 @@ import type {
   KiwiSDR,
   Scanner,
   TrackedFlight,
+  DashboardData,
 } from '@/types/dashboard';
-import { useDataSnapshot } from '@/hooks/useDataStore';
+import { useDataKeys } from '@/hooks/useDataStore';
+
+/** Keys the layer panel reads — avoids re-rendering on unrelated fast-poll keys. */
+const WORLDVIEW_PANEL_DATA_KEYS = [
+  'ships',
+  'sigint_totals',
+  'sigint',
+  'cctv_total',
+  'cctv',
+  'satnogs_total',
+  'satnogs_stations',
+  'tinygs_total',
+  'tinygs_satellites',
+  'tracked_flights',
+  'commercial_flights',
+  'private_flights',
+  'private_jets',
+  'military_flights',
+  'gps_jamming',
+  'fishing_activity',
+  'satellite_source',
+  'satellite_analysis',
+  'satellites',
+  'road_corridor_trends',
+  'earthquakes',
+  'firms_fires',
+  'ukraine_alerts',
+  'weather_alerts',
+  'volcanoes',
+  'air_quality',
+  'sar_anomalies',
+  'sar_scenes',
+  'uap_sightings',
+  'wastewater',
+  'datacenters',
+  'internet_outages',
+  'power_plants',
+  'military_bases',
+  'trains',
+  'malware_threats',
+  'scm_suppliers',
+  'cyber_threats',
+  'kiwisdr',
+  'psk_reporter',
+  'scanners',
+  'frontlines',
+  'gdelt',
+  'telegram_osint',
+  'crowdthreat',
+  'correlations',
+  'gt_risk',
+  'freshness',
+] as const satisfies readonly (keyof DashboardData)[];
 
 // ---------------------------------------------------------------------------
 // ScannerTracker — in-app audio player for tracked police scanner systems
@@ -699,7 +752,7 @@ const WorldviewLeftPanel = React.memo(function WorldviewLeftPanel({
   onOpenSarAoiEditor?: () => void;
   viewBoundsRef?: React.RefObject<{ south: number; west: number; north: number; east: number } | null>;
 }) {
-  const data = useDataSnapshot() as import('@/types/dashboard').DashboardData;
+  const data = useDataKeys(WORLDVIEW_PANEL_DATA_KEYS) as DashboardData;
   const { t } = useTranslation();
   const [internalMinimized, setInternalMinimized] = useState(true);
   const isMinimized = isMinimizedProp !== undefined ? isMinimizedProp : internalMinimized;

scripts/bench_layer_toggle_latency.py

@@ -0,0 +1,202 @@
+#!/usr/bin/env python3
+"""Measure layer-toggle → data-visible latency (UX guardrail for perf work).
+
+Simulates what the dashboard does on toggle:
+  1. POST /api/layers (layer off → on)
+  2. Poll GET /api/live-data/slow until the layer's payload is non-empty
+
+Also reports whether data was already warm in the backend store before toggle
+(via /api/health source counts while the layer is still filtered off in the API).
+
+Usage:
+  python scripts/bench_layer_toggle_latency.py
+  python scripts/bench_layer_toggle_latency.py --base http://127.0.0.1:8000
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import time
+import urllib.error
+import urllib.request
+from dataclasses import dataclass
+from typing import Any, Callable
+
+# layer_key → JSON field in /api/live-data/slow + how to count "visible"
+LAYER_PROBE: dict[str, tuple[str, Callable[[Any], int]]] = {
+    "cctv": ("cctv", lambda _v: 0),  # fast tier — see FAST_LAYER_PROBE
+    "firms": ("firms_fires", lambda v: len(v) if isinstance(v, list) else 0),
+    "datacenters": ("datacenters", lambda v: len(v) if isinstance(v, list) else 0),
+    "power_plants": ("power_plants", lambda v: len(v) if isinstance(v, list) else 0),
+    "psk_reporter": ("psk_reporter", lambda v: len(v) if isinstance(v, list) else 0),
+}
+
+FAST_LAYER_PROBE = {
+    "cctv": ("cctv", lambda v: len(v) if isinstance(v, list) else 0),
+}
+
+HEALTH_SOURCE_KEY = {
+    "cctv": "cctv",
+    "firms": "firms_fires",
+    "datacenters": "datacenters",
+    "power_plants": "power_plants",
+    "psk_reporter": "psk_reporter",
+}
+
+
+@dataclass
+class ToggleResult:
+    layer: str
+    warm_store_count: int | None
+    time_to_visible_ms: float | None
+    visible_count: int
+    timed_out: bool
+    on_enable_fetch: bool
+    notes: str
+
+
+def _request(method: str, url: str, body: dict | None = None, timeout: float = 30.0) -> tuple[int, Any]:
+    data = None
+    headers = {"Accept": "application/json"}
+    if body is not None:
+        data = json.dumps(body).encode()
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        raw = resp.read()
+        return resp.status, json.loads(raw) if raw else None
+
+
+def get_health(base: str) -> dict:
+    _, payload = _request("GET", f"{base}/api/health")
+    return payload or {}
+
+
+def get_slow(base: str) -> dict:
+    _, payload = _request("GET", f"{base}/api/live-data/slow")
+    return payload or {}
+
+
+def get_fast(base: str) -> dict:
+    _, payload = _request("GET", f"{base}/api/live-data/fast")
+    return payload or {}
+
+
+def set_layer(base: str, layers: dict[str, bool]) -> None:
+    _request("POST", f"{base}/api/layers", {"layers": layers})
+
+
+def count_visible(payload: dict, field: str, counter: Callable[[Any], int]) -> int:
+    return counter(payload.get(field))
+
+
+ON_ENABLE_IMMEDIATE = {"datacenters", "fishing_activity"}
+
+
+def measure_layer(base: str, layer: str, timeout_s: float = 120.0) -> ToggleResult:
+    health = get_health(base)
+    warm = None
+    hk = HEALTH_SOURCE_KEY.get(layer)
+    if hk and isinstance(health.get("sources"), dict):
+        warm = health["sources"].get(hk)
+
+    # Ensure layer is off (frontend default for these probes)
+    set_layer(base, {layer: False})
+    time.sleep(0.25)
+
+    # Confirm API filters it off while toggled off
+    if layer in FAST_LAYER_PROBE:
+        field, counter = FAST_LAYER_PROBE[layer]
+        off_payload = get_fast(base)
+    else:
+        field, counter = LAYER_PROBE[layer]
+        off_payload = get_slow(base)
+    off_count = count_visible(off_payload, field, counter)
+
+    # Toggle on — mirrors dashboard POST + immediate slow/fast refetch
+    t0 = time.perf_counter()
+    set_layer(base, {layer: True})
+
+    visible_count = 0
+    timed_out = True
+    while (time.perf_counter() - t0) < timeout_s:
+        if layer in FAST_LAYER_PROBE:
+            payload = get_fast(base)
+        else:
+            payload = get_slow(base)
+        visible_count = count_visible(payload, field, counter)
+        if visible_count > 0:
+            timed_out = False
+            break
+        time.sleep(0.25)
+
+    elapsed_ms = None if timed_out else (time.perf_counter() - t0) * 1000.0
+
+    notes_parts = []
+    if off_count > 0:
+        notes_parts.append(f"unexpected visible while off ({off_count})")
+    if warm and warm > 0 and (timed_out or (elapsed_ms or 0) < 500):
+        notes_parts.append("warm store — toggle likely instant from prefetch")
+    elif warm == 0 and timed_out:
+        notes_parts.append("cold store — would feel broken to user")
+    elif timed_out:
+        notes_parts.append(f"no warm store signal; waited {timeout_s:.0f}s")
+
+    return ToggleResult(
+        layer=layer,
+        warm_store_count=warm,
+        time_to_visible_ms=elapsed_ms,
+        visible_count=visible_count,
+        timed_out=timed_out,
+        on_enable_fetch=layer in ON_ENABLE_IMMEDIATE,
+        notes="; ".join(notes_parts) or "ok",
+    )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--base", default="http://127.0.0.1:8000")
+    parser.add_argument("--timeout", type=float, default=120.0)
+    parser.add_argument("--layers", nargs="*", default=list(LAYER_PROBE.keys()))
+    args = parser.parse_args()
+
+    print(f"Backend: {args.base}")
+    try:
+        health = get_health(args.base)
+    except urllib.error.URLError as exc:
+        print(f"Health check failed: {exc}", file=sys.stderr)
+        return 1
+
+    print(f"Version: {health.get('version')}  uptime: {health.get('uptime_seconds')}s")
+    print(f"Runtime profile: {(health.get('runtime') or {}).get('profile')}")
+    print()
+    print(f"{'layer':<14} {'warm_store':>10} {'visible_ms':>11} {'count':>8} {'on_enable':>10}  notes")
+    print("-" * 90)
+
+    results: list[ToggleResult] = []
+    for layer in args.layers:
+        try:
+            r = measure_layer(args.base, layer, timeout_s=args.timeout)
+        except urllib.error.URLError as exc:
+            print(f"{layer:<14} ERROR: {exc}")
+            continue
+        results.append(r)
+        ms = f"{r.time_to_visible_ms:.0f}" if r.time_to_visible_ms is not None else f">{args.timeout:.0f}s"
+        warm = str(r.warm_store_count) if r.warm_store_count is not None else "?"
+        on_en = "yes" if r.on_enable_fetch else "no"
+        print(f"{layer:<14} {warm:>10} {ms:>11} {r.visible_count:>8} {on_en:>10}  {r.notes}")
+
+    print()
+    instant = [r for r in results if r.time_to_visible_ms is not None and r.time_to_visible_ms < 500]
+    slow = [r for r in results if r.timed_out or (r.time_to_visible_ms or 0) >= 5000]
+    print(f"Summary: {len(instant)}/{len(results)} toggles visible in <500ms; {len(slow)} slow or timed out")
+    if slow:
+        print("Layers that need on-enable fetch or prefetch to avoid UX pain:")
+        for r in slow:
+            print(f"  - {r.layer}: {r.notes}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())