Merge pull request #413 from bilawalsidhu/refactor/live-source-contracts

Separate live sources from aircraft and vessel layers

CHANGELOG.md

@@ -13,6 +13,9 @@ of current runtime behavior, see [`docs/CURRENT-STATE.md`](docs/CURRENT-STATE.md
 
 ## [Unreleased]
 
+- Separate aircraft/vessel transport and normalization from layer rendering, preserving observation timestamps, altitude datums and optional history.
+- Retain absent aircraft during partially admitted snapshots and bound source error messages.
+
 - Drive share updates, Location feedback and Scene controls through immutable state snapshots and disposable subscriptions.
 - Keep stale lookup/load completions from publishing accepted results and retain shot rows during playback progress updates.
 - Export the existing Scene director with explicit playback and editing outcomes.

docs/CURRENT-STATE.md

@@ -1,5 +1,24 @@
 # God's Eye View Current State
 
+## Browser live-source observations
+
+Flights, Military Flights and AIS Vessels obtain snapshots and optional history
+through `gods-eye-view/sources/live`. The standalone adapters use the existing
+same-origin routes. Aircraft observations distinguish barometric metres from
+WGS84 ellipsoid metres and retain source position/contact epochs; vessel records
+retain separate heading/course and sea-surface datum. History is a best-effort
+addition to the locally accumulated trail, never a promise of complete coverage.
+
+Snapshot coverage, completeness and freshness are separate fields. A partially
+admitted aircraft snapshot retains absent contacts for up to five minutes before
+the usual missed-poll eviction. Unknown snapshot times remain unknown in stats. An invalid nonempty
+snapshot retains the previous display. Empty vessel refreshes retain the existing
+first-connect grace and warm-data behavior. Known source failures have bounded
+messages; arbitrary HTTP response bodies are not surfaced as diagnostics.
+Sources receive cancellation signals and check them after body parsing. The
+layer's existing lifecycle and selection guards continue to reject late work.
+
+
 ## State and action outcomes
 
 Share preferences, place lookups and Scene playback expose immutable snapshots

package.json

@@ -141,6 +141,7 @@
     "./ui/scenes": "./src/ui/scenes.js",
     "./ui/shell": "./src/ui/applicationShell.js",
     "./ui/styles": "./src/ui/styles.js",
-    "./scenes": "./src/scenes/director.js"
+    "./scenes": "./src/scenes/director.js",
+    "./sources/live": "./src/sources/live/index.js"
   }
 }

scripts/format-scope.json

@@ -248,5 +248,11 @@
   "src/ui/styles/cockpit.css",
   "src/ui/styles/command-dock-compact.css",
   "src/app/stateChannel.js",
-  "src/app/stateChannel.test.mjs"
+  "src/app/stateChannel.test.mjs",
+  "src/sources/live/aircraft.js",
+  "src/sources/live/contract.js",
+  "src/sources/live/contract.test.mjs",
+  "src/sources/live/index.js",
+  "src/sources/live/standalone.js",
+  "src/sources/live/vessels.js"
 ]

scripts/package-boundaries.json

@@ -565,5 +565,16 @@
       "src/contextModePolicy.js"
     ],
     "external": ["cesium"]
+  },
+  "live-observations": {
+    "exports": ["./sources/live"],
+    "modules": [
+      "src/sources/live/aircraft.js",
+      "src/sources/live/contract.js",
+      "src/sources/live/index.js",
+      "src/sources/live/standalone.js",
+      "src/sources/live/vessels.js"
+    ],
+    "external": []
   }
 }

scripts/qa-vessel-cards.mjs

@@ -253,6 +253,10 @@ async function main() {
 
       const page = await browser.newPage();
       await page.setViewport({ width: 1600, height: 900 });
+      await page.evaluateOnNewDocument(() => {
+        // This harness inspects cards, so keep first-run chrome out of the view.
+        localStorage.setItem('gev:first-run-mission:v1', 'suppressed');
+      });
       page.on('pageerror', (err) => console.error(`    [page-error] ${err.message}`));
 
       await page.goto(APP_URL, { waitUntil: 'domcontentloaded', timeout: 60000 });
@@ -319,7 +323,10 @@ async function main() {
           orientation: { heading: p.heading, pitch: p.pitch, roll: 0 },
         });
       }, port);
-      const resettled = settled ? true : await page
+      // A changed camera invalidates the previous settled result. Let the new
+      // view schedule its tiles before checking readiness again.
+      await new Promise((resolve) => setTimeout(resolve, 1500));
+      const resettled = await page
         .waitForFunction(() => {
           const gev = window.__godsEyeView;
           const ais = gev.dataManager.getAll().find((l) => l.id === 'ais-live-vessels');

src/data/aisLiveVessels.js

@@ -1,3 +1,4 @@
+import { createAisStreamSource } from '../sources/live/standalone.js';
 import * as Cesium from 'cesium';
 import {
   registerEntityContext,
@@ -46,13 +47,14 @@ import {
 import { requestWorldFocus } from '../worldFocus.js';
 import { holdContinuousRender, releaseContinuousRender } from '../renderGovernor.js';
 
+let _source = createAisStreamSource({ apiUrl: import.meta.env?.VITE_AIS_LIVE_API_URL || '/api/ais-live' });
+
 const FOCUS_EVIDENCE_DEV = import.meta.env?.DEV === true;
 
 /** Camera pose signature at the last vessel rotation pass. */
 let _lastCamPoseSig = '';
 const _scratchFocusScreen = new Cesium.Cartesian2();
 
-const DEFAULT_API_URL = '/api/ais-live';
 const DEFAULT_RENDER_ROWS = 12000;
 const DEFAULT_ACTIVE_LABELS = 900;
 const REFRESH_MS = 60000;
@@ -343,6 +345,14 @@ const aisLiveVesselsLayer = {
   updateInterval: REFRESH_MS,
   statsRefreshInterval: 1000,
 
+  /** Configure the source before initialization; an active layer keeps its owner. */
+  setSource(source) {
+    if (state.viewer) throw new Error('Configure the source before layer initialization');
+    if (typeof source?.getSnapshot !== 'function') throw new TypeError('A snapshot source is required');
+    _source = source;
+    this.source = source.label || this.source;
+  },
+
   init(viewer) {
     state.viewer = viewer;
     ensureCollections(viewer);
@@ -844,33 +854,21 @@ async function loadLivePositions(viewer) {
   state.abort = requestController;
 
   try {
-    const url = liveApiUrl();
     // Combine the layer's teardown-abort with a hard timeout so a hung upstream
     // can't wedge the poll indefinitely (parity with the track fetch + flights).
     const signal = typeof AbortSignal.any === 'function'
       ? AbortSignal.any([requestController.signal, AbortSignal.timeout(10000)])
       : requestController.signal;
-    const response = await fetch(url, {
-      signal,
-      cache: 'no-store',
+    const snapshot = await _source.getSnapshot({ maxRows: renderRowLimit() }, { signal });
+    if (!ownsAisRequest(requestController, requestSessionId)) return;
+    aisLiveVesselsLayer.source = snapshot.source;
+    // Map observations into the existing display store; source fields stop here.
+    applyAisFeedSnapshot(viewer, {
+      rows: snapshot.records.map(vesselDisplayRow), status: snapshot.transportStatus,
+      lastMessageAt: snapshot.lastMessageAt, nextAttemptAt: snapshot.nextAttemptAt,
+      refreshing: snapshot.stale, newestPositionAt: snapshot.observedAtMs == null ? null : new Date(snapshot.observedAtMs).toISOString(),
+      silentForMs: snapshot.silentForMs, reconnectAttempt: snapshot.reconnectAttempt,
     });
-    if (!ownsAisRequest(requestController, requestSessionId)) return;
-    if (!response.ok) {
-      // The 503 key-absent / 502 stream-error bodies still carry {status,error}.
-      // Prefer a clean surfaced reason over a cryptic "AIS live HTTP 503".
-      let reason = `AIS live HTTP ${response.status}`;
-      try {
-        const errPayload = await response.json();
-        if (!ownsAisRequest(requestController, requestSessionId)) return;
-        reason = deriveAisFeedError(errPayload, 0)
-          || (typeof errPayload?.error === 'string' && errPayload.error.trim()) || reason;
-      } catch { /* non-JSON body — keep the HTTP status reason */ }
-      throw new Error(reason);
-    }
-
-    const payload = await response.json();
-    if (!ownsAisRequest(requestController, requestSessionId)) return;
-    applyAisFeedSnapshot(viewer, payload);
   } catch (error) {
     if (ownsAisRequest(requestController, requestSessionId) && error?.name !== 'AbortError') {
       markAisUnavailable(error?.message || 'AIS live load failed');
@@ -942,11 +940,15 @@ function applyAisFeedSnapshot(viewer, payload) {
   return { reconciled: true, ...snapshot };
 }
 
-function liveApiUrl() {
-  const base = import.meta.env?.VITE_AIS_LIVE_API_URL || DEFAULT_API_URL;
-  const url = new URL(base, window.location.origin);
-  url.searchParams.set('maxRows', String(renderRowLimit()));
-  return url.toString();
+function vesselDisplayRow(record) {
+  return {
+    mmsi: record.id, reference: record.reference, lat: record.latitude, lon: record.longitude,
+    name: record.name, imo: record.imo, type: record.type, destination: record.destination,
+    speed: record.speedMps == null ? null : record.speedMps / 0.514444,
+    course: record.courseDeg, heading: record.headingDeg,
+    last_position_epoch: record.observedAtMs == null ? null : record.observedAtMs / 1000,
+    last_position_UTC: record.observedAtMs == null ? '' : new Date(record.observedAtMs).toISOString(),
+  };
 }
 
 function renderRowLimit() {
@@ -1145,6 +1147,7 @@ function normalizeVessel(row) {
     lon,
     name: String(row.name || row.input_name || row.mmsi || row.input_identifier || 'VESSEL'),
     mmsi: String(row.mmsi || row.input_identifier || '').trim(),
+    reference: row.reference ?? String(row.mmsi || row.input_identifier || '').trim(),
     imo: String(row.imo || ''),
     type: String(row.type_specific || row.type || ''),
     destination: String(row.destination || ''),
@@ -1638,7 +1641,7 @@ function startSelectedVesselTrail(record) {
     state.trail = createTrail(state.viewer, { color: TRAIL_COLOR, width: 2.5 });
   }
   if (state.trail) state.trail.setPositions(state.trailPositions);
-  backfillVesselTrail(record.mmsi, state.trailBackfillToken);
+  backfillVesselTrail(record.mmsi, state.trailBackfillToken, record.reference);
 }
 
 /**
@@ -1651,15 +1654,11 @@ function startSelectedVesselTrail(record) {
  * @param {number} token - Backfill token captured at request time.
  * @returns {Promise<void>}
  */
-async function backfillVesselTrail(mmsi, token) {
+async function backfillVesselTrail(mmsi, token, reference = mmsi) {
   let samples = null;
   try {
-    const response = await fetch('/api/ais-live/track?mmsi=' + encodeURIComponent(mmsi), {
-      signal: AbortSignal.timeout(8000),
-    });
-    if (!response.ok) return;
-    const payload = await response.json();
-    samples = Array.isArray(payload?.samples) ? payload.samples : null;
+    const track = await _source.getTrack?.(reference, { signal: AbortSignal.timeout(8000) });
+    samples = track?.records ?? null;
   } catch {
     return; // silent — keep the live-accumulated trail
   }
@@ -1668,8 +1667,8 @@ async function backfillVesselTrail(mmsi, token) {
 
   const older = [];
   for (const sample of samples) {
-    const lat = Number(sample?.lat);
-    const lon = Number(sample?.lon);
+    const lat = Number(sample?.latitude);
+    const lon = Number(sample?.longitude);
     if (!Number.isFinite(lat) || !Number.isFinite(lon)) continue;
     // Per-sample N (≤ TRAIL_MAX_POINTS lookups) — same sea-surface datum as
     // the live vertices so the spliced trail is height-continuous.
@@ -1735,7 +1734,7 @@ function registerSelectedContext(record) {
       id: `ais-${record.mmsi}`,
       layerId: 'ais-live-vessels',
       layerName: 'Live AIS Vessels',
-      source: 'AISStream',
+      source: aisLiveVesselsLayer.source,
       label: displayVesselName(record),
       latitude: record.lat,
       longitude: record.lon,

src/data/aisLiveVessels.test.mjs

@@ -1641,3 +1641,28 @@ test('a vessel analyst record carries the MMSI the tracker keys on', () => {
   assert.equal(nameless.id, '366999124');
   assert.equal(nameless.mmsi, '366999124');
 });
+
+test('vessel selection passes the opaque source reference to optional history', async () => {
+  const { createAisStreamSource } = await import('../sources/live/standalone.js');
+  _setVesselStateForTest({ enabled: false });
+  const requests = [];
+  aisLiveVesselsLayer.setSource({
+    label: 'Test vessel source',
+    getSnapshot: async () => ({ records: [] }),
+    getTrack: async (reference, options) => { requests.push({ reference, options }); return { records: [] }; },
+  });
+  const harness = installWireHarness(undefined, { selectedRecord: null, trailMmsi: null });
+  try {
+    harness.record.reference = 'opaque:test-reference';
+    assert.equal(aisLiveVesselsLayer.selectById(harness.record.mmsi), true);
+    await new Promise(resolve => setImmediate(resolve));
+    assert.equal(requests.length, 1);
+    assert.equal(requests[0].reference, 'opaque:test-reference');
+    assert.ok(requests[0].options.signal instanceof AbortSignal);
+    assert.equal(aisLiveVesselsLayer.source, 'Test vessel source');
+  } finally {
+    harness.cleanup();
+    _setVesselStateForTest({ enabled: false });
+    aisLiveVesselsLayer.setSource(createAisStreamSource());
+  }
+});

src/data/flights.js

@@ -1,3 +1,4 @@
+import { createOpenSkySource } from '../sources/live/standalone.js';
 /**
  * @module flights
  * @description Real-time flight tracking layer powered by the OpenSky Network API
@@ -95,6 +96,8 @@ import {
 import { CONTACT_MATCH_TIER, contactMatchWins, rankContactMatch } from './contactMatch.js';
 import { holdContinuousRender, releaseContinuousRender } from '../renderGovernor.js';
 
+let _source = createOpenSkySource();
+
 const FOCUS_EVIDENCE_DEV = import.meta.env?.DEV === true;
 
 /** Amber tint for known-military aircraft rendered by this layer (matches the military layer's icon color). */
@@ -269,11 +272,6 @@ const _scratchModelBS = new Cesium.BoundingSphere(new Cesium.Cartesian3(), 1.0);
 /** Last limb taper per billboard, retained across class/ground/cockpit repaints. */
 const _billboardLimbScale = new WeakMap();
 
-/** @constant {string} API_URL - Vite proxy endpoint for OpenSky /states/all */
-const API_URL = '/api/opensky';
-const SOURCE_STALE_MS = 120_000;
-/** @constant {number} BACKOFF_INTERVAL - Cooldown (ms) after 429 / auth errors */
-const BACKOFF_INTERVAL = 45000; // 45s on rate limit
 /** @constant {number} ERROR_BACKOFF_INTERVAL - Cooldown (ms) after transient errors */
 const ERROR_BACKOFF_INTERVAL = 20000; // transient error retry
 /** @constant {number} POSITION_HISTORY_LIMIT - Max position samples kept per aircraft for dead reckoning */
@@ -322,17 +320,12 @@ let _lastSource = 'OpenSky Network';
 /** @type {string} Completeness boundary for the latest successful snapshot. */
 let _lastCoverage = 'worldwide upstream snapshot';
 
-function _flightApiUrl(viewer) {
+function _flightQuery(viewer) {
   const cartographic = viewer?.camera?.positionCartographic;
-  if (!cartographic) return API_URL;
-  const latitude = Cesium.Math.toDegrees(cartographic.latitude);
-  const longitude = Cesium.Math.toDegrees(cartographic.longitude);
-  if (!Number.isFinite(latitude) || !Number.isFinite(longitude)) return API_URL;
-  const params = new URLSearchParams({
-    lat: latitude.toFixed(4),
-    lon: longitude.toFixed(4),
-  });
-  return `${API_URL}?${params}`;
+  return cartographic ? {
+    latitude: Cesium.Math.toDegrees(cartographic.latitude),
+    longitude: Cesium.Math.toDegrees(cartographic.longitude),
+  } : {};
 }
 
 // ---------------------------------------------------------------------------
@@ -348,7 +341,7 @@ let _lastTrackingRefreshOutcome = {
   epoch: 0,
   status: 'unavailable',
   ids: new Set(),
-  source: 'OpenSky Network',
+  source: _lastSource,
   coverage: null,
 };
 /** @type {Cesium.Entity|null} Entity used for camera tracking */
@@ -420,7 +413,7 @@ function _contextSubjectMetadata(icao24) {
     id: icao24,
     layerId: 'flights',
     layerName: 'Live Flights',
-    source: 'OpenSky Network',
+    source: _lastSource,
     label: _contactLabel(icao24, _flightData.get(icao24)),
     latitude: described.latitude,
     longitude: described.longitude,
@@ -785,10 +778,10 @@ let _enrichDripTimer = null;
 const _enrichQueue = [];
 const _enrichSeen = new Set();
 
-function _enqueueEnrich(key, url, onData, priority = false) {
+function _enqueueEnrich(key, query, onData, priority = false) {
   if (_enrichSeen.has(key)) return;
   _enrichSeen.add(key);
-  const job = { url, onData };
+  const job = { query, onData };
   // Priority (tracked / model-eligible) goes to the FRONT so a deep ambient
   // backlog can never delay the plane the user just clicked or zoomed into.
   if (priority) _enrichQueue.unshift(job); else _enrichQueue.push(job);
@@ -810,8 +803,7 @@ function _drainEnrich() {
     _enrichLastDispatchMs = Date.now();
     const job = _enrichQueue.shift();
     _enrichActive += 1;
-    fetch(job.url)
-      .then((r) => (r.ok ? r.json() : null))
+    Promise.resolve(_source.getEnrichment?.(job.query))
       .then((data) => { if (data && data.found) job.onData(data); })
       .catch(() => { /* enrichment never surfaces errors */ })
       .finally(() => { _enrichActive -= 1; _drainEnrich(); });
@@ -820,7 +812,7 @@ function _drainEnrich() {
 
 function _requestTypeEnrichment(icao24, priority = false) {
   if (!/^[0-9a-f]{6}$/i.test(icao24)) return;
-  _enqueueEnrich(`t:${icao24}`, `/api/adsbdb/type/${icao24.toLowerCase()}`, (data) => {
+  _enqueueEnrich(`t:${icao24}`, { kind: 'type', id: icao24.toLowerCase() }, (data) => {
     const meta = _flightData.get(icao24);
     if (!meta) return; // evicted while the lookup was in flight
     meta.typeCode = data.typeCode || meta.typeCode;
@@ -844,7 +836,7 @@ function _requestTypeEnrichment(icao24, priority = false) {
 function _requestRouteEnrichment(icao24) {
   const cs = String(_flightData.get(icao24)?.callsign || '').trim().toUpperCase();
   if (!/^[A-Z]{3}\d/.test(cs)) return; // airline-style callsigns only (LLL + digit); GA tails won't resolve
-  _enqueueEnrich(`r:${cs}`, `/api/adsbdb/route/${encodeURIComponent(cs)}`, (data) => {
+  _enqueueEnrich(`r:${cs}`, { kind: 'route', id: cs }, (data) => {
     const meta = _flightData.get(icao24);
     if (!meta) return;
     meta.airline = data.airline || meta.airline;
@@ -1005,15 +997,6 @@ let _drReconcileValid = false;
 /** @type {string|null} icao the reconciliation state currently belongs to */
 let _drReconcileIcao = null;
 
-/**
- * Normalize a value to a trimmed lowercase string.
- * @param {*} value - Any value (typically a header string or null).
- * @returns {string} Lowercase trimmed string, or '' if falsy.
- */
-function _toLowerText(value) {
-  return String(value || '').trim().toLowerCase();
-}
-
 /**
  * Normalize a value to a trimmed string. A whitespace-only field ("   ") is
  * truthy, so every label chain must trim FIRST and then fall through.
@@ -1044,45 +1027,6 @@ function _contactLabel(icao24, info) {
   return _toCleanText(info?.callsign) || _toCleanText(info?.registration) || icao24;
 }
 
-/**
- * Map OpenSky proxy response headers into a human-readable auth error string.
- * The Vite proxy forwards `x-opensky-auth-mode-used` and `x-opensky-auth-reason`
- * headers so the client can display a meaningful diagnostic.
- * @param {object} params
- * @param {string} params.detail  - Error body text from the proxy, if any.
- * @param {string} params.authMode - Normalized auth mode header value.
- * @param {string} params.authReason - Normalized auth reason header value.
- * @returns {string} Concise error description for UI display.
- */
-function _deriveOpenSkyAuthError({ detail, authMode, authReason }) {
-  const reason = _toLowerText(authReason);
-  const mode = _toLowerText(authMode);
-
-  if (reason === 'oauth_invalid_or_missing') {
-    return 'OpenSky OAuth client missing/invalid';
-  }
-  if (reason === 'oauth_invalid_credentials') {
-    return 'OpenSky OAuth rejected credentials';
-  }
-  if (reason === 'basic_invalid_credentials') {
-    return 'OpenSky username/password rejected';
-  }
-  if (reason === 'missing_basic_creds' || reason === 'missing_oauth_and_basic_creds') {
-    return 'OpenSky auth missing';
-  }
-  if (reason === 'auth_required') {
-    return 'OpenSky auth required';
-  }
-  if (reason.startsWith('oauth_') || reason.startsWith('basic_')) {
-    return 'OpenSky auth invalid';
-  }
-  if (reason === 'forced_anonymous' || mode === 'anon') {
-    return 'OpenSky auth required';
-  }
-  if (detail) return detail;
-  return 'OpenSky auth failed';
-}
-
 /**
  * Dead-reckon an aircraft's current position using ENU (East-North-Up) frame math.
  *
@@ -3041,12 +2985,8 @@ function _startTrail(icao24) {
 async function _backfillTrail(icao24, token, oldestFixEpochSec) {
   let path = null;
   try {
-    const response = await fetch('/api/opensky-track?icao24=' + encodeURIComponent(icao24), {
-      signal: AbortSignal.timeout(8000),
-    });
-    if (!response.ok) return;
-    const data = await response.json();
-    path = Array.isArray(data?.path) ? data.path : null;
+    const track = await _source.getTrack?.(_flightData.get(icao24)?.sourceReference ?? icao24, { signal: AbortSignal.timeout(8000) });
+    path = track?.records ?? null;
   } catch {
     return; // silent fallback to the accumulated trail
   }
@@ -3061,8 +3001,8 @@ async function _backfillTrail(icao24, token, oldestFixEpochSec) {
   await ensureGeoidReady();
   const parsed = [];
   for (const waypoint of path) {
-    if (!Array.isArray(waypoint)) continue;
-    const [time, lat, lon, baroAlt] = waypoint;
+    const { observedAtMs, latitude: lat, longitude: lon, baroAltitudeM: baroAlt } = waypoint;
+    const time = observedAtMs / 1000;
     if (!Number.isFinite(lat) || !Number.isFinite(lon)) continue;
     if (!Number.isFinite(time) || time >= oldestFixEpochSec) continue;
     parsed.push({ lat, lon, baroAlt });
@@ -3213,12 +3153,6 @@ function _normalizeTrackedIcao(candidate) {
   return normalized || null;
 }
 
-function _isUsableOpenSkyState(state) {
-  if (!Array.isArray(state) || typeof state[0] !== 'string' || !_normalizeTrackedIcao(state[0])) {
-    return false;
-  }
-  return Number.isFinite(state[5]) && Number.isFinite(state[6]);
-}
 
 /**
  * Whether the Military layer suppresses this civil duplicate right now.
@@ -3882,13 +3816,22 @@ const flightsLayer = {
   id: 'flights',
   name: 'Live Flights',
   icon: '✈️',
-  source: 'OpenSky Network',
+  source: _lastSource,
   // Browser-harness seam: isolates synthetic display-floor scenarios without
   // changing any production lifecycle or cache policy.
   _clearDisplayFloorStateForTest,
   /** @type {number} Polling interval (ms) between update() calls */
   updateInterval: 30000,
 
+  /** Configure the source before initialization; an active layer keeps its owner. */
+  setSource(source) {
+    if (_viewer) throw new Error('Configure the source before layer initialization');
+    if (typeof source?.getSnapshot !== 'function') throw new TypeError('A snapshot source is required');
+    _source = source;
+    _lastSource = source.label || _lastSource;
+    this.source = _lastSource;
+  },
+
   /**
    * Initialize the flights layer.
    * Creates the BillboardCollection, resets all state, and installs the
@@ -3928,7 +3871,7 @@ const flightsLayer = {
     _retryAt = 0;
     _lastError = null;
     _lastStatus = null;
-    _lastSource = 'OpenSky Network';
+    _lastSource = _source.label || 'OpenSky Network';
     _lastCoverage = 'worldwide upstream snapshot';
     _trackedIcao = null;
     _resetTrackedSelectionState();
@@ -4076,90 +4019,22 @@ const flightsLayer = {
       : resourceController.signal;
     try {
       updateSignal.throwIfAborted();
-      const response = await fetch(_flightApiUrl(viewer || _viewer), { signal: updateSignal });
-      _lastStatus = response.status;
-      const responseSource = response.headers.get('x-flight-source');
-      const responseCoverage = response.headers.get('x-flight-coverage');
-      const authMode = _toLowerText(
-        response.headers.get('x-opensky-auth-mode-used') || response.headers.get('x-opensky-auth')
-      );
-      const authReason = _toLowerText(response.headers.get('x-opensky-auth-reason'));
-
-      if (response.status === 429) {
-        console.warn('[Data:Flights] Rate limited, backing off to 30s');
-        _backoff = true;
-        _retryAt = nowMs + BACKOFF_INTERVAL;
-        _lastError = authMode && authMode !== 'anon'
-          ? 'OpenSky rate limited'
-          : 'OpenSky rate limited (anonymous)';
-        return;
-      }
-
-      if (response.status === 401 || response.status === 403) {
-        console.warn(`[Data:Flights] OpenSky unavailable (${response.status}), backing off`);
-        _backoff = true;
-        _retryAt = nowMs + BACKOFF_INTERVAL;
-        let detail = '';
-        try {
-          const body = await response.json();
-          updateSignal.throwIfAborted();
-          detail = typeof body?.error === 'string' ? body.error.trim() : '';
-        } catch {
-          detail = '';
-        }
-        _lastError = _deriveOpenSkyAuthError({
-          detail,
-          authMode,
-          authReason,
-        });
-        return;
-      }
-
-      if (!response.ok) {
-        console.warn(`[Data:Flights] API returned ${response.status}`);
-        _backoff = true;
-        _retryAt = nowMs + ERROR_BACKOFF_INTERVAL;
-        let detail = '';
-        try {
-          const body = await response.json();
-          updateSignal.throwIfAborted();
-          detail = typeof body?.error === 'string' ? body.error.trim() : '';
-        } catch {
-          detail = '';
-        }
-        _lastError = detail || `OpenSky HTTP ${response.status}`;
-        return;
-      }
-
-      const data = await response.json();
+      const snapshot = await _source.getSnapshot(_flightQuery(viewer || _viewer), { signal: updateSignal });
       updateSignal.throwIfAborted();
-      if (!data || !Array.isArray(data.states)) {
-        _backoff = true;
-        _retryAt = nowMs + ERROR_BACKOFF_INTERVAL;
-        _lastError = 'Malformed OpenSky response';
-        return;
-      }
-
-      const usableStates = data.states.filter(_isUsableOpenSkyState);
-      if (data.states.length > 0 && usableStates.length === 0) {
-        _backoff = true;
-        _retryAt = nowMs + ERROR_BACKOFF_INTERVAL;
-        _lastError = 'Malformed OpenSky aircraft rows';
-        return;
-      }
-
-      const sourceEpochMs = Number.isFinite(Number(data.time)) && Number(data.time) > 0
-        ? Number(data.time) * 1000
-        : null;
-      const sourceAgeMs = sourceEpochMs == null ? 0 : Math.max(0, Date.now() - sourceEpochMs);
-      const sourceStale = sourceAgeMs > SOURCE_STALE_MS;
+      _lastStatus = snapshot.status ?? 200;
+      const usableStates = snapshot.records;
+      const sourceEpochMs = snapshot.observedAtMs;
+      const sourceAgeMs = snapshot.ageMs;
+      const sourceStale = snapshot.stale || snapshot.freshness === 'unknown';
       _backoff = sourceStale;
       _retryAt = 0;
       _lastError = sourceStale
-        ? `Source snapshot ${Math.max(2, Math.round(sourceAgeMs / 60_000))} min old`
+        ? sourceAgeMs == null ? 'Source snapshot time unavailable'
+          : `Source snapshot ${Math.max(2, Math.round(sourceAgeMs / 60_000))} min old`
         : null;
-      _lastSource = responseSource || 'OpenSky Network';
-      _lastCoverage = responseCoverage || 'worldwide upstream snapshot';
+      _lastSource = snapshot.source;
+      _lastCoverage = snapshot.coverage;
+      flightsLayer.source = _lastSource;
       const currentIcaos = new Set();
       const acceptedSnapshotIcaos = new Set();
       const now = Cesium.JulianDate.now();
@@ -4171,20 +4046,13 @@ const flightsLayer = {
       const viewerLonDeg = viewerCarto ? Cesium.Math.toDegrees(viewerCarto.longitude) : null;
       const floorWarmPoints = [];
 
-      // Destructure OpenSky state vector array (indices per API spec):
-      // [0] icao24, [1] callsign, [2] origin_country, [3] time_position,
-      // [4] last_contact, [5] longitude, [6] latitude, [7] baro_altitude,
-      // [8] on_ground, [9] velocity, [10] true_track, [11] vertical_rate,
-      // [12] sensors, [13] geo_altitude (WGS84 ellipsoidal — the CORRECT
-      // globe-render height when present; height-datum fix Task 6).
-      // Keep military classification fresh while the military layer is off
+      // Classification and display policy consume source-independent observations.
       refreshMilitaryRegistryIfStale();
-
-      for (const state of usableStates) {
-        const [rawIcao24, callsign, origin_country, time_position, last_contact, lon, lat, baro_alt, on_ground, velocity, true_track, , , geo_alt] = state;
-        const icao24 = _normalizeTrackedIcao(rawIcao24);
-        const category = Number.isFinite(state[17]) ? state[17] : null; // extended=1 emitter category
-        const vertical_rate = Number.isFinite(state[11]) ? state[11] : null; // m/s, + = climbing
+      for (const observation of usableStates) {
+        const { id: icao24, callsign, originCountry: origin_country,
+          longitude: lon, latitude: lat, baroAltitudeM: baro_alt, onGround: on_ground,
+          speedMps: velocity, courseDeg: true_track, ellipsoidAltitudeM: geo_alt,
+          category, verticalRateMps: vertical_rate } = observation;
         acceptedSnapshotIcaos.add(icao24);
         const onGround = on_ground === true;
 
@@ -4347,6 +4215,8 @@ const flightsLayer = {
         // Store flight metadata for click-to-track labels
         const cat = stickyNumber(category, prevMeta?.category, null);
         const meta = {
+          sourceReference: observation.reference,
+          observedReceiptMs: Date.now(),
           callsign: stickyText(callsign, prevMeta?.callsign),
           altitude: alt,
           // geoAltitudeM/renderAltitudeM are ADDITIVE fields alongside the
@@ -4378,7 +4248,7 @@ const flightsLayer = {
           // contact time so a temporarily old position does not hard-freeze
           // while fresh velocity/track messages are still arriving.
           lastContactEpochMs: stickyNumber(
-            Number.isFinite(last_contact) ? last_contact * 1000 : null,
+            observation.contactTimeMs,
             prevMeta?.lastContactEpochMs,
             null,
           ),
@@ -4404,8 +4274,8 @@ const flightsLayer = {
         // 5-15s stale and receipt-time stamping is what caused the
         // back/forward oscillation. Only append when the fix actually
         // advances, so stale repeats don't create zero-dt segments.
-        const fixEpochMs = Number.isFinite(time_position) && time_position > 0
-          ? time_position * 1000
+        const fixEpochMs = Number.isFinite(observation.positionTimeMs) && observation.positionTimeMs > 0
+          ? observation.positionTimeMs
           : Date.now();
         const fixTime = Cesium.JulianDate.fromDate(new Date(fixEpochMs));
         if (!_positionHistory.has(icao24)) {
@@ -4527,6 +4397,8 @@ const flightsLayer = {
       // feed gap, and the full grace left phantom planes parked at airports.
       for (const [icao24, bb] of _billboards) {
         if (currentIcaos.has(icao24)) continue;
+        // Partial admissions do not prove absence, but stale retention is bounded.
+        if (!snapshot.complete && Date.now() - (_flightData.get(icao24)?.observedReceiptMs ?? 0) < 300000) continue;
         const misses = (_missingPolls.get(icao24) || 0) + 1;
         const limit = _likelyLanded(icao24) ? LANDED_MISSING_POLL_LIMIT : MISSING_POLL_LIMIT;
         if (misses < limit) {
@@ -4605,7 +4477,7 @@ const flightsLayer = {
       _count = _billboards.size;
       // Freshness belongs to the source snapshot, not the moment this browser
       // received a cached 200 response.
-      _lastUpdate = sourceEpochMs ?? Date.now();
+      _lastUpdate = sourceEpochMs;
       _lastTrackingRefreshOutcome = {
         epoch: trackingRefreshEpoch,
         status: 'accepted',
@@ -4622,8 +4494,10 @@ const flightsLayer = {
       }
       console.warn('[Data:Flights] Fetch error:', e);
       _backoff = true;
-      _retryAt = Date.now() + ERROR_BACKOFF_INTERVAL;
-      _lastError = 'OpenSky network error';
+      _retryAt = Date.now() + (e?.retryAfterMs ?? ERROR_BACKOFF_INTERVAL);
+      _lastStatus = e?.status ?? null;
+      if (e?.source) { _lastSource = e.source; this.source = _lastSource; }
+      _lastError = e?.name === 'LiveSourceError' ? e.message : 'Live data unavailable';
     } finally {
       _activeUpdateControllers.delete(resourceController);
     }

src/data/militaryFlights.js

@@ -1,3 +1,4 @@
+import { createAdsbLolSource } from '../sources/live/standalone.js';
 import * as Cesium from 'cesium';
 import { aircraftIncludedInNearby } from './aircraftNearbyPolicy.js';
 import { registerPickOwner, unregisterPickOwner, isOwnedByOtherLayer, resolvePickId } from './pickRegistry.js';
@@ -62,6 +63,8 @@ import {
 import { CONTACT_MATCH_TIER, contactMatchWins, rankContactMatch } from './contactMatch.js';
 import { holdContinuousRender, releaseContinuousRender } from '../renderGovernor.js';
 
+let _source = createAdsbLolSource();
+
 /**
  * @module militaryFlights
  * @description Real-time military flight tracking layer powered by the adsb.lol API.
@@ -79,12 +82,8 @@ import { holdContinuousRender, releaseContinuousRender } from '../renderGovernor
  * rotation (camera.heading - track) with hysteresis.
  */
 
-/** @constant {string} API endpoint proxied to adsb.lol military feed */
-const API_URL = '/api/adsblol/mil';
 /** @constant {number} Milliseconds to wait before retrying after a transient error */
 const ERROR_BACKOFF_INTERVAL = 20000;
-/** @constant {number} Longer cooldown (ms) after a 429 rate-limit, mirroring flights.js */
-const BACKOFF_INTERVAL = 45000;
 /** @constant {number} Max position samples retained per aircraft for dead reckoning */
 const POSITION_HISTORY_LIMIT = 5;
 /** @constant {number} Base billboard display scale */
@@ -269,6 +268,7 @@ function _abortActiveUpdates() {
 }
 /** @type {number|null} HTTP status code from the last API response */
 let _lastStatus = null;
+let _lastSource = 'adsb.lol';
 /** @type {boolean} True once ensureGeoidReady() has resolved (awaited once at enable()). Mirror of flights.js. */
 let _geoidReady = false;
 /** @type {Map<string, number>} icao24 -> geoid undulation N (m), cached (negligible drift per-aircraft). */
@@ -284,7 +284,7 @@ let _lastTrackingRefreshOutcome = {
   epoch: 0,
   status: 'unavailable',
   ids: new Set(),
-  source: 'adsb.lol',
+  source: _lastSource,
 };
 /** @type {Cesium.Entity|null} Entity created for the tracked aircraft (camera follows this) */
 let _trackedEntity = null;
@@ -346,7 +346,7 @@ function _contextSubjectMetadata(icao24) {
     id: icao24,
     layerId: 'military',
     layerName: 'Military Flights',
-    source: 'adsb.lol',
+    source: _lastSource,
     label,
     latitude: described.latitude,
     longitude: described.longitude,
@@ -841,7 +841,7 @@ export function _setTrackedMilitaryRefreshStateForTest({
 export function _setMilitaryTrackingRefreshOutcomeForTest({
   status = 'accepted',
   ids = [],
-  source = 'adsb.lol',
+  source = _lastSource,
 } = {}) {
   const epoch = ++_trackingRefreshEpoch;
   _lastTrackingRefreshOutcome = {
@@ -2194,20 +2194,14 @@ function _startTrail(icao24) {
  * @returns {Promise<void>}
  */
 async function _backfillTrail(icao24, token, oldestFixEpochSec) {
-  let baseEpochSec = null;
   let trace = null;
   try {
-    const response = await fetch('/api/adsblol/trace?hex=' + encodeURIComponent(icao24), {
-      signal: AbortSignal.timeout(8000),
-    });
-    if (!response.ok) return;
-    const data = await response.json();
-    baseEpochSec = Number(data?.timestamp);
-    trace = Array.isArray(data?.trace) ? data.trace : null;
+    const track = await _source.getTrack?.(_flightData.get(icao24)?.sourceReference ?? icao24, { signal: AbortSignal.timeout(8000) });
+    trace = track?.records ?? null;
   } catch {
     return; // silent fallback to the accumulated trail
   }
-  if (!trace || !Number.isFinite(baseEpochSec)) return;
+  if (!trace) return;
   if (token !== _trailBackfillToken || icao24 !== _trackedIcao) return;
 
   // Height-datum fix (Task 7, mirror of flights.js Task 6 _backfillTrail): a
@@ -2220,13 +2214,11 @@ async function _backfillTrail(icao24, token, oldestFixEpochSec) {
   // readsb trace points: [secondsAfterTimestamp, lat, lon, alt_ft|'ground'|null, gs_kt, track, flags, ...]
   const parsed = [];
   for (const point of trace) {
-    if (!Array.isArray(point)) continue;
-    const t = baseEpochSec + Number(point[0]);
-    const lat = Number(point[1]);
-    const lon = Number(point[2]);
+    const t = point.observedAtMs / 1000;
+    const lat = point.latitude, lon = point.longitude;
     if (!Number.isFinite(lat) || !Number.isFinite(lon)) continue;
     if (!Number.isFinite(t) || t >= oldestFixEpochSec) continue;
-    parsed.push({ lat, lon, altFt: point[3] });
+    parsed.push({ lat, lon, baroAltitudeM: point.baroAltitudeM });
   }
   if (!parsed.length) return;
 
@@ -2247,10 +2239,8 @@ async function _backfillTrail(icao24, token, oldestFixEpochSec) {
 
   let older = [];
   let lastAltM = null; // carry-forward for ground points whose cell isn't warm yet
-  for (const { lat, lon, altFt } of parsed) {
-    const baroM = (altFt === 'ground' || altFt == null || !Number.isFinite(Number(altFt)))
-      ? null
-      : Number(altFt) * 0.3048 + geoidHeight(lat, lon);
+  for (const { lat, lon, baroAltitudeM } of parsed) {
+    const baroM = baroAltitudeM == null ? null : baroAltitudeM + geoidHeight(lat, lon);
     let altM = floorAltitudeM(baroM, cachedGroundFloor(lat, lon));
     // Ground/no-alt point with an unresolved floor: hold the previous
     // waypoint's altitude (continuity — never a dive to a made-up depth).
@@ -2386,12 +2376,7 @@ function _normalizeTrackedIcao(candidate) {
   return normalized || null;
 }
 
-function _isUsableMilitaryAircraft(aircraft) {
-  if (!aircraft || Array.isArray(aircraft) || typeof aircraft !== 'object') return false;
-  if (typeof aircraft.hex !== 'string' || !_normalizeTrackedIcao(aircraft.hex)) return false;
-  return Number.isFinite(_toFiniteNumber(aircraft.lon))
-    && Number.isFinite(_toFiniteNumber(aircraft.lat));
-}
+
 
 function _applyPendingTrackingRestore() {
   const pending = _pendingTrackingRestore;
@@ -2603,10 +2588,19 @@ const militaryFlightsLayer = {
   id: 'military',
   name: 'Military Flights',
   icon: '🎖️',
-  source: 'adsb.lol',
+  source: _lastSource,
   /** @type {number} Polling interval in ms between API fetches */
   updateInterval: 15000,
 
+  /** Configure the source before initialization; an active layer keeps its owner. */
+  setSource(source) {
+    if (_viewer) throw new Error('Configure the source before layer initialization');
+    if (typeof source?.getSnapshot !== 'function') throw new TypeError('A snapshot source is required');
+    _source = source;
+    _lastSource = source.label || _lastSource;
+    this.source = _lastSource;
+  },
+
   /**
    * Initialize the layer: create the billboard collection, reset all state,
    * and install the click-to-track handler.
@@ -2771,7 +2765,7 @@ const militaryFlightsLayer = {
       epoch: trackingRefreshEpoch,
       status: 'source-unavailable',
       ids: new Set(),
-      source: 'adsb.lol',
+      source: _lastSource,
     };
     if (_retryAt && nowMs < _retryAt) {
       _backoff = true;
@@ -2785,73 +2779,24 @@ const militaryFlightsLayer = {
       : resourceController.signal;
     try {
       updateSignal.throwIfAborted();
-      const response = await fetch(API_URL, { signal: updateSignal });
-      _lastStatus = response.status;
-
-      if (!response.ok) {
-        _backoff = true;
-        // Parity with flights.js: a 429 gets the LONGER cooldown + a friendly
-        // rate-limit label instead of the generic transient backoff, so we don't
-        // hammer the upstream and the UI reads honestly.
-        if (response.status === 429) {
-          _retryAt = nowMs + BACKOFF_INTERVAL;
-          _lastError = 'adsb.lol rate limited';
-          return;
-        }
-        _retryAt = nowMs + ERROR_BACKOFF_INTERVAL;
-        let detail = '';
-        try {
-          const body = await response.json();
-          updateSignal.throwIfAborted();
-          detail = _toCleanText(body?.error || body?.message);
-        } catch {
-          detail = '';
-        }
-        _lastError = detail || `adsb.lol HTTP ${response.status}`;
-        return;
-      }
-
-      // adsb.lol returns { ac: [...aircraft], msg: "...", ... }
-      const data = await response.json();
+      const snapshot = await _source.getSnapshot({}, { signal: updateSignal });
       updateSignal.throwIfAborted();
-      if (!data || !Array.isArray(data.ac)) {
-        _backoff = true;
-        _retryAt = nowMs + ERROR_BACKOFF_INTERVAL;
-        _lastError = 'Malformed adsb.lol response';
-        return;
-      }
-
-      const usableAircraft = data.ac.filter(_isUsableMilitaryAircraft);
-      if (data.ac.length > 0 && usableAircraft.length === 0) {
-        _backoff = true;
-        _retryAt = nowMs + ERROR_BACKOFF_INTERVAL;
-        _lastError = 'Malformed adsb.lol aircraft rows';
-        return;
-      }
-
-      _backoff = false;
+      _lastStatus = snapshot.status ?? 200;
+      _lastSource = snapshot.source;
+      militaryFlightsLayer.source = _lastSource;
+      const usableAircraft = snapshot.records;
+      _backoff = snapshot.stale || snapshot.freshness === 'unknown';
       _retryAt = 0;
-      _lastError = null;
+      _lastError = snapshot.freshness === 'unknown' ? 'Source snapshot time unavailable' : null;
       const currentIcaos = new Set();
-      const cacheAgeMs = Number(response.headers?.get?.('x-ads-b-cache-age-ms'));
-      const receiptNowMs = Date.now() - (Number.isFinite(cacheAgeMs) && cacheAgeMs > 0 ? cacheAgeMs : 0);
-      _backoff = response.headers?.get?.('x-ads-b-cache') === 'STALE';
+      const receiptNowMs = snapshot.observedAtMs;
       // Field-test fix (RS46): coarse floor cells to warm for the below-ground
       // clamp — collected during the loop (low airborne contacts only),
       // batch-resolved once after it. Never a fetch inside the loop.
       const _floorWarmPoints = [];
 
-      // -- Parse each aircraft record from the adsb.lol response --
-      // Fields used: hex (ICAO), lon, lat, alt_baro (ft, barometric/MSL),
-      // alt_geom (ft, geometric/WGS84 ellipsoidal — probed live 2026-07-05:
-      // present on ~45% of records; readsb omits it when the aircraft hasn't
-      // reported a geometric altitude this cycle), track (deg true), gs
-      // (ground speed in knots), seen_pos (age of last position, seconds),
-      // flight (callsign), t (type), r (registration), ownOp (operator).
       for (const aircraft of usableAircraft) {
-        const icao24 = _toCleanText(aircraft?.hex).toLowerCase();
-        const lon = _toFiniteNumber(aircraft?.lon);
-        const lat = _toFiniteNumber(aircraft?.lat);
+        const { id: icao24, longitude: lon, latitude: lat } = aircraft;
 
         currentIcaos.add(icao24);
         _missingPolls.delete(icao24);
@@ -2861,40 +2806,19 @@ const militaryFlightsLayer = {
         // separate boolean). Grounded planes fall back to their last known
         // altitude (field elevation is unknowable here), else 0 m — never the
         // 3 km airborne default (a parked plane must not float).
-        const rawAltBaro = aircraft?.alt_baro;
-        const onGround = typeof rawAltBaro === 'string' && rawAltBaro.trim().toLowerCase() === 'ground';
-        const altitudeFt = _toFiniteNumber(rawAltBaro);
-        const altitudeM = Number.isFinite(altitudeFt)
-          ? altitudeFt * 0.3048
-          : (onGround
-            ? (Number.isFinite(prevMeta?.altitudeFt) ? prevMeta.altitudeFt * 0.3048 : 0)
-            : 3048);
-        const track = _toFiniteNumber(aircraft?.track) || 0;
-        // Convert ground speed from knots to meters/sec for dead reckoning
-        const speedKt = _toFiniteNumber(aircraft?.gs);
-        const speedMps = Number.isFinite(speedKt) ? speedKt * 0.514444 : 0;
-        // Analyst seam (additive): readsb baro_rate is ft/min; keep m/s.
-        const baroRateFtMin = _toFiniteNumber(aircraft?.baro_rate);
-        const verticalRateMps = Number.isFinite(baroRateFtMin) ? baroRateFtMin * 0.00508 : null;
-
-        const callsign = _toCleanText(aircraft?.flight);
-        const type = _toCleanText(aircraft?.t);
-        const registration = _toCleanText(aircraft?.r);
-        const operator = _toCleanText(aircraft?.ownOp);
-        const seenSec = _toFiniteNumber(aircraft?.seen);
-
-        // Height-datum fix (Task 7, mirror of flights.js Task 6): `altitudeM`
-        // (above) stays the AVIATION field — the sticky barometric/MSL altitude
-        // read by labels (FL/altitude readout) and the landed-fast-cull
-        // heuristic. It is NEVER overwritten or renamed. Where the aircraft
-        // actually RENDERS on the ellipsoidal globe (billboard path only — a
-        // ground-snapped 3D MODEL uses groundSnap.js's own tileset sample and
-        // ignores this value entirely) is a SEPARATE value, renderAltitudeM:
-        // alt_geom when readsb reports it (already WGS84 ellipsoidal), else
-        // alt_baro+geoid as a visual fallback, else ground surface when parked.
-        const altGeomFt = _toFiniteNumber(aircraft?.alt_geom);
-        const geoAltitudeM = Number.isFinite(altGeomFt) ? altGeomFt * 0.3048 : null;
-        const baroAltitudeM = Number.isFinite(altitudeFt) ? altitudeFt * 0.3048 : null;
+        const onGround = aircraft.onGround;
+        const altitudeFt = aircraft.baroAltitudeM == null ? null : aircraft.baroAltitudeM / 0.3048;
+        const altitudeM = aircraft.baroAltitudeM
+          ?? (onGround ? (Number.isFinite(prevMeta?.altitudeFt) ? prevMeta.altitudeFt * 0.3048 : 0) : 3048);
+        const track = aircraft.courseDeg || 0;
+        const speedMps = aircraft.speedMps ?? 0;
+        const verticalRateMps = aircraft.verticalRateMps;
+        const callsign = aircraft.callsign;
+        const type = aircraft.typeCode;
+        const registration = aircraft.registration;
+        const operator = aircraft.operator;
+        const geoAltitudeM = aircraft.ellipsoidAltitudeM;
+        const baroAltitudeM = aircraft.baroAltitudeM;
 
         // geoid undulation N: cached per-aircraft (negligible drift — see
         // task brief) once the geoid grid has loaded; unavailable pre-load
@@ -2979,6 +2903,8 @@ const militaryFlightsLayer = {
         // last-known-good (bounded by the layer's eviction, which deletes the entry).
         const stickyType = stickyText(type, prevMeta?.type);
         const meta = {
+          sourceReference: aircraft.reference,
+          observedReceiptMs: Date.now(),
           callsign: stickyText(callsign, prevMeta?.callsign),
           type: stickyType,
           // Type outranks category automatically inside classifyAircraft.
@@ -2997,7 +2923,7 @@ const militaryFlightsLayer = {
           // Analyst seam (additive): sticky like the other kinematics.
           verticalRateMps: stickyNumber(verticalRateMps, prevMeta?.verticalRateMps, null),
           lastContactEpochMs: stickyNumber(
-            Number.isFinite(seenSec) ? receiptNowMs - seenSec * 1000 : null,
+            aircraft.contactTimeMs,
             prevMeta?.lastContactEpochMs,
             null,
           ),
@@ -3022,8 +2948,7 @@ const militaryFlightsLayer = {
         // report, so the fix epoch is receipt time minus that age. Only
         // append when the fix actually advances, so stale repeats don't
         // create zero-dt segments.
-        const seenPos = _toFiniteNumber(aircraft?.seen_pos);
-        const fixEpochMs = receiptNowMs - (Number.isFinite(seenPos) ? seenPos * 1000 : 0);
+        const fixEpochMs = aircraft.positionTimeMs ?? receiptNowMs;
         const fixTime = Cesium.JulianDate.fromDate(new Date(fixEpochMs));
         if (!_positionHistory.has(icao24)) {
           _positionHistory.set(icao24, []);
@@ -3175,6 +3100,8 @@ const militaryFlightsLayer = {
       // feed gap, and the full grace left phantom planes parked at airports.
       for (const [icao24, bb] of _billboards) {
         if (currentIcaos.has(icao24)) continue;
+        // Partial admissions do not prove absence, but stale retention is bounded.
+        if (!snapshot.complete && Date.now() - (_flightData.get(icao24)?.observedReceiptMs ?? 0) < 300000) continue;
         const misses = (_missingPolls.get(icao24) || 0) + 1;
         const limit = _likelyLanded(icao24) ? LANDED_MISSING_POLL_LIMIT : MISSING_POLL_LIMIT;
         if (misses < limit) {
@@ -3222,7 +3149,7 @@ const militaryFlightsLayer = {
         epoch: trackingRefreshEpoch,
         status: 'accepted',
         ids: currentIcaos,
-        source: 'adsb.lol',
+        source: _lastSource,
       };
       console.log(`[Data:Military] Updated: ${_count} aircraft`);
       _applyPendingTrackingRestore();
@@ -3233,8 +3160,10 @@ const militaryFlightsLayer = {
       }
       console.warn('[Data:Military] Fetch error:', e);
       _backoff = true;
-      _retryAt = Date.now() + ERROR_BACKOFF_INTERVAL;
-      _lastError = 'adsb.lol network error';
+      _retryAt = Date.now() + (e?.retryAfterMs ?? ERROR_BACKOFF_INTERVAL);
+      _lastStatus = e?.status ?? null;
+      if (e?.source) { _lastSource = e.source; this.source = _lastSource; }
+      _lastError = e?.name === 'LiveSourceError' ? e.message : 'Live data unavailable';
     } finally {
       _activeUpdateControllers.delete(resourceController);
     }
@@ -3308,7 +3237,7 @@ const militaryFlightsLayer = {
       epoch: _trackingRefreshEpoch,
       status: 'destroyed',
       ids: new Set(),
-      source: 'adsb.lol',
+      source: _lastSource,
     };
     _resetTrackedSelectionState(); // next lifecycle re-evaluates against the ENTER ceiling
     _viewer = null;
@@ -3670,7 +3599,7 @@ const militaryFlightsLayer = {
     if (outcome.status !== 'accepted') {
       return {
         status: 'source-unavailable',
-        reason: 'adsb.lol snapshot unavailable',
+        reason: `${_lastSource} snapshot unavailable`,
         refreshEpoch: outcome.epoch,
         source: outcome.source,
       };
@@ -3772,7 +3701,7 @@ const militaryFlightsLayer = {
       error: _lastError,
       status: _lastStatus,
       retryInSec,
-      source: 'adsb.lol',
+      source: _lastSource,
       fallback: false,
     };
   },

src/sources/live/aircraft.js

@@ -0,0 +1,167 @@
+import {
+  admitRecords,
+  cleanText,
+  coordinates,
+  epoch,
+  finite,
+} from './contract.js';
+
+/** Normalize the OpenSky state-vector route, including its regional fallback. */
+export function normalizeOpenSkyAircraft(row) {
+  if (!Array.isArray(row)) return null;
+  const id = cleanText(row[0]).toLowerCase();
+  if (!id || !coordinates(row[6], row[5])) return null;
+  return {
+    id,
+    reference: id,
+    latitude: row[6],
+    longitude: row[5],
+    callsign: cleanText(row[1]),
+    originCountry: cleanText(row[2]),
+    positionTimeMs: epoch(row[3], 1000),
+    contactTimeMs: epoch(row[4], 1000),
+    baroAltitudeM: finite(row[7]),
+    ellipsoidAltitudeM: finite(row[13]),
+    onGround: row[8] === true,
+    speedMps: finite(row[9]),
+    courseDeg: finite(row[10]),
+    verticalRateMps: finite(row[11]),
+    category: finite(row[17]),
+    typeCode: null,
+    registration: null,
+    operator: null,
+  };
+}
+
+/** Normalize readsb observations. The caller supplies the source snapshot epoch. */
+export function normalizeReadsbAircraft(row, snapshotTimeMs) {
+  const id = cleanText(row?.hex).toLowerCase();
+  const latitude = finite(row?.lat),
+    longitude = finite(row?.lon);
+  if (!id || !coordinates(latitude, longitude)) return null;
+  const baroFt = finite(row.alt_baro),
+    geoFt = finite(row.alt_geom);
+  const speedKts = finite(row.gs),
+    rateFtMin = finite(row.baro_rate);
+  const seen = finite(row.seen),
+    seenPos = finite(row.seen_pos);
+  return {
+    id,
+    reference: id,
+    latitude,
+    longitude,
+    callsign: cleanText(row.flight),
+    originCountry: null,
+    positionTimeMs:
+      snapshotTimeMs == null ? null : snapshotTimeMs - (seenPos ?? 0) * 1000,
+    contactTimeMs:
+      snapshotTimeMs == null || seen == null
+        ? null
+        : snapshotTimeMs - seen * 1000,
+    baroAltitudeM: baroFt == null ? null : baroFt * 0.3048,
+    ellipsoidAltitudeM: geoFt == null ? null : geoFt * 0.3048,
+    onGround: cleanText(row.alt_baro).toLowerCase() === 'ground',
+    speedMps: speedKts == null ? null : speedKts * 0.514444,
+    courseDeg: finite(row.track),
+    verticalRateMps: rateFtMin == null ? null : rateFtMin * 0.00508,
+    category: row.category ?? null,
+    typeCode: cleanText(row.t),
+    registration: cleanText(row.r),
+    operator: cleanText(row.ownOp),
+  };
+}
+
+export function openSkySnapshot(
+  payload,
+  {
+    source = 'OpenSky Network',
+    coverage = 'worldwide upstream snapshot',
+    now = Date.now(),
+    stale = false,
+  } = {},
+) {
+  const admitted = admitRecords(
+    payload?.states,
+    normalizeOpenSkyAircraft,
+    'OpenSky',
+  );
+  const observedAtMs = epoch(payload?.time, 1000);
+  const ageMs = observedAtMs == null ? null : Math.max(0, now - observedAtMs);
+  return {
+    ...admitted,
+    source,
+    coverage,
+    observedAtMs,
+    ageMs,
+    stale: stale || (ageMs != null && ageMs > 120000),
+    freshness:
+      observedAtMs == null
+        ? 'unknown'
+        : stale || ageMs > 120000
+          ? 'stale'
+          : 'current',
+  };
+}
+
+export function readsbSnapshot(
+  payload,
+  {
+    observedAtMs,
+    source = 'adsb.lol',
+    coverage = 'military upstream snapshot',
+    now = Date.now(),
+    stale = false,
+  } = {},
+) {
+  const admitted = admitRecords(
+    payload?.ac,
+    (row) => normalizeReadsbAircraft(row, observedAtMs),
+    'adsb.lol',
+  );
+  const ageMs = observedAtMs == null ? null : Math.max(0, now - observedAtMs);
+  return {
+    ...admitted,
+    source,
+    coverage,
+    observedAtMs,
+    ageMs,
+    stale,
+    freshness: observedAtMs == null ? 'unknown' : stale ? 'stale' : 'current',
+  };
+}
+
+/** Track altitude remains barometric; renderers choose a visual ground fallback. */
+export function normalizeAircraftTrack(
+  rows,
+  { baseTimeMs = 0, readsb = false } = {},
+) {
+  if (!Array.isArray(rows)) return [];
+  return rows.flatMap((row) => {
+    if (!Array.isArray(row)) return [];
+    const latitude = finite(row[1]),
+      longitude = finite(row[2]);
+    const offset = finite(row[0]);
+    const observedAtMs =
+      offset == null ? null : epoch(baseTimeMs + offset * 1000);
+    if (
+      !coordinates(latitude, longitude) ||
+      observedAtMs == null ||
+      observedAtMs <= 0
+    )
+      return [];
+    const altitude = finite(row[3]);
+    return [
+      {
+        latitude,
+        longitude,
+        observedAtMs,
+        baroAltitudeM:
+          altitude == null ? null : altitude * (readsb ? 0.3048 : 1),
+        ellipsoidAltitudeM: null,
+        onGround: readsb
+          ? cleanText(row[3]).toLowerCase() === 'ground'
+          : row[5] === true,
+      },
+    ];
+  });
+}

src/sources/live/contract.js

@@ -0,0 +1,125 @@
+/**
+ * Browser live-source contract. Coordinates are WGS84 degrees; lengths and
+ * velocities are metres and metres/second; times are Unix milliseconds.
+ * Barometric altitude and ellipsoid altitude are separate, nullable observations.
+ * A missing timestamp is unknown, never the time the response was received.
+ *
+ * Sources expose getSnapshot(query, { signal }) and optionally
+ * getTrack(reference, { signal }). References are opaque to consumers.
+ * Snapshots describe coverage and completeness independently of freshness.
+ * Records contain observation data only, with no scene objects or transport data.
+ */
+export class LiveSourceError extends Error {
+  constructor(
+    code,
+    message,
+    { status = null, retryAfterMs = 20000, source = null } = {},
+  ) {
+    super(message);
+    this.name = 'LiveSourceError';
+    this.code = code;
+    this.status = status;
+    this.retryAfterMs = retryAfterMs;
+    this.source = source;
+  }
+}
+
+export function finite(value) {
+  if (value == null || value === '' || typeof value === 'boolean') return null;
+  const number = Number(value);
+  return Number.isFinite(number) ? number : null;
+}
+
+export function epoch(value, scale = 1) {
+  const number = finite(value);
+  const milliseconds = number == null ? null : number * scale;
+  return milliseconds != null &&
+    milliseconds > 0 &&
+    milliseconds <= 8640000000000000
+    ? milliseconds
+    : null;
+}
+
+export function cleanText(value) {
+  return typeof value === 'string' ? value.trim() : '';
+}
+
+export function coordinates(latitude, longitude) {
+  return (
+    Number.isFinite(latitude) &&
+    Math.abs(latitude) <= 90 &&
+    Number.isFinite(longitude) &&
+    Math.abs(longitude) <= 180
+  );
+}
+
+/** Admit a snapshot atomically; an all-invalid nonempty feed is unavailable. */
+export function admitRecords(rows, normalize, label) {
+  if (!Array.isArray(rows))
+    throw new LiveSourceError('malformed', `Malformed ${label} response`);
+  const records = [];
+  const ids = new Set();
+  for (const row of rows) {
+    const record = normalize(row);
+    if (record && !ids.has(record.id)) {
+      records.push(record);
+      ids.add(record.id);
+    }
+  }
+  if (rows.length && !records.length) {
+    throw new LiveSourceError('malformed', `Malformed ${label} aircraft rows`);
+  }
+  return {
+    records,
+    complete: records.length === rows.length,
+    rejectedCount: rows.length - records.length,
+  };
+}
+
+/** Cancellation is checked after body parsing even when a transport ignores it. */
+export async function readResponse(
+  fetchImpl,
+  url,
+  { signal, ...init } = {},
+  source = 'Live source',
+) {
+  signal?.throwIfAborted();
+  let response;
+  try {
+    response = await fetchImpl(url, { ...init, signal });
+    signal?.throwIfAborted();
+  } catch (error) {
+    signal?.throwIfAborted();
+    if (error?.name === 'AbortError') throw error;
+    throw new LiveSourceError('unavailable', `${source} network error`, {
+      source,
+    });
+  }
+  let payload = null;
+  try {
+    payload = await response.json();
+  } catch {
+    /* classified below */
+  }
+  signal?.throwIfAborted();
+  return { response, payload };
+}
+
+export function httpError(response, source) {
+  const status = response.status;
+  const code =
+    status === 429
+      ? 'limited'
+      : status === 401 || status === 403
+        ? 'denied'
+        : 'unavailable';
+  return new LiveSourceError(
+    code,
+    status === 429 ? `${source} rate limited` : `${source} HTTP ${status}`,
+    {
+      status,
+      source,
+      retryAfterMs: code === 'limited' || code === 'denied' ? 45000 : 20000,
+    },
+  );
+}

src/sources/live/contract.test.mjs

@@ -0,0 +1,273 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import {
+  createOpenSkySource,
+  createAdsbLolSource,
+  createAisStreamSource,
+  normalizeReadsbAircraft,
+  normalizeVesselObservation,
+  openSkySnapshot,
+  readsbSnapshot,
+} from './index.js';
+
+const now = 1800000000000;
+const aircraft = [
+  'AbC123',
+  'TEST',
+  'Example',
+  now / 1000 - 15,
+  now / 1000 - 2,
+  -97,
+  30,
+  1000,
+  false,
+  100,
+  250,
+  -2,
+  null,
+  1040,
+];
+const response = (payload, headers = {}, status = 200) =>
+  Response.json(payload, { headers, status });
+
+test('civil observations retain identity, separate altitude datums and source epochs', () => {
+  const snapshot = openSkySnapshot(
+    { time: now / 1000 - 130, states: [aircraft] },
+    { now, source: 'Test source', coverage: 'regional' },
+  );
+  const row = snapshot.records[0];
+  assert.equal(row.id, 'abc123');
+  assert.equal(row.baroAltitudeM, 1000);
+  assert.equal(row.ellipsoidAltitudeM, 1040);
+  assert.equal(row.positionTimeMs, now - 15000);
+  assert.equal(row.contactTimeMs, now - 2000);
+  assert.equal(snapshot.ageMs, 130000);
+  assert.equal(snapshot.freshness, 'stale');
+  assert.equal(snapshot.coverage, 'regional');
+  const missing = openSkySnapshot(
+    { states: [[...aircraft.slice(0, 3), null, null, ...aircraft.slice(5)]] },
+    { now },
+  );
+  assert.equal(missing.observedAtMs, null);
+  assert.equal(missing.freshness, 'unknown');
+  assert.equal(missing.records[0].positionTimeMs, null);
+});
+
+test('mixed-invalid and duplicate admissions are explicitly incomplete; all-invalid rejects', () => {
+  const invalid = [...aircraft];
+  invalid[5] = 181;
+  const partial = openSkySnapshot({ states: [aircraft, invalid, aircraft] });
+  assert.equal(partial.records.length, 1);
+  assert.equal(partial.complete, false);
+  assert.equal(partial.rejectedCount, 2);
+  assert.throws(() => openSkySnapshot({ states: [invalid] }), {
+    code: 'malformed',
+  });
+  assert.throws(() => openSkySnapshot({}), { code: 'malformed' });
+  assert.equal(openSkySnapshot({ states: [] }).complete, true);
+});
+
+test('readsb units and position/contact ages are normalized once against the source epoch', () => {
+  const row = normalizeReadsbAircraft(
+    {
+      hex: 'abc123',
+      lat: '30',
+      lon: '-97',
+      alt_baro: 10000,
+      alt_geom: 10100,
+      gs: 120,
+      baro_rate: -600,
+      seen: 2,
+      seen_pos: 12,
+    },
+    now,
+  );
+  assert.equal(row.baroAltitudeM, 3048);
+  assert.equal(row.ellipsoidAltitudeM, 3078.48);
+  assert.equal(row.speedMps, 120 * 0.514444);
+  assert.equal(row.verticalRateMps, -600 * 0.00508);
+  assert.equal(row.positionTimeMs, now - 12000);
+  assert.equal(row.contactTimeMs, now - 2000);
+  const ground = normalizeReadsbAircraft(
+    { hex: 'abc123', lat: 30, lon: -97, alt_baro: 'ground' },
+    null,
+  );
+  assert.equal(ground.onGround, true);
+  assert.equal(ground.baroAltitudeM, null);
+  assert.equal(ground.positionTimeMs, null);
+  assert.equal(
+    readsbSnapshot({ ac: [] }, { observedAtMs: null }).freshness,
+    'unknown',
+  );
+});
+
+test('vessel identity, opaque reference and heading/course remain distinct, unknowns stay null', () => {
+  const row = normalizeVesselObservation(
+    {
+      mmsi: '123456789',
+      lat: 30,
+      lon: -97,
+      speed: 10,
+      heading: 45,
+      course: 60,
+      last_position_epoch: now / 1000,
+      extra: 'discard',
+    },
+    'opaque-reference',
+  );
+  assert.equal(row.id, '123456789');
+  assert.equal(row.reference, 'opaque-reference');
+  assert.equal(row.headingDeg, 45);
+  assert.equal(row.courseDeg, 60);
+  assert.equal(row.speedMps, 10 * 0.514444);
+  assert.equal(row.altitudeDatum, 'sea-surface');
+  assert.equal(row.observedAtMs, now);
+  assert.equal('extra' in row, false);
+  assert.equal(
+    normalizeVesselObservation({ mmsi: 'x', lat: null, lon: null }),
+    null,
+  );
+});
+
+test('construction is inert; adapters preserve routes, viewport query, cache epoch and history datum', async () => {
+  const requests = [];
+  const fetchImpl = async (url, init) => {
+    requests.push({ url, init });
+    if (url.startsWith('/api/opensky?'))
+      return response(
+        { time: now / 1000, states: [aircraft] },
+        {
+          'x-flight-source': 'adsb.lol',
+          'x-flight-coverage': '250 nm regional',
+        },
+      );
+    if (url.startsWith('/api/opensky-track'))
+      return response({ path: [[now / 1000 - 30, 30, -97, 1000, 45, false]] });
+    if (url === '/api/adsblol/mil')
+      return response(
+        { ac: [{ hex: 'abc123', lat: 30, lon: -97, seen_pos: 15 }] },
+        { 'x-ads-b-cache-age-ms': '60000', 'x-ads-b-cache': 'STALE' },
+      );
+    return response({
+      timestamp: now / 1000 - 60,
+      trace: [[10, 30, -97, 1000]],
+    });
+  };
+  const civil = createOpenSkySource({ fetchImpl, now: () => now });
+  const military = createAdsbLolSource({ fetchImpl, now: () => now });
+  assert.equal(requests.length, 0);
+  const signal = new AbortController().signal;
+  const snapshot = await civil.getSnapshot(
+    { latitude: 30.123456, longitude: -97 },
+    { signal },
+  );
+  assert.equal(requests[0].url, '/api/opensky?lat=30.1235&lon=-97.0000');
+  assert.equal(requests[0].init.signal, signal);
+  assert.equal(snapshot.source, 'adsb.lol');
+  assert.equal(snapshot.coverage, '250 nm regional');
+  assert.equal((await civil.getTrack('abc123')).records[0].baroAltitudeM, 1000);
+  const mil = await military.getSnapshot();
+  assert.equal(mil.observedAtMs, now - 60000);
+  assert.equal(mil.records[0].positionTimeMs, now - 75000);
+  assert.equal(mil.stale, true);
+  const track = await military.getTrack('abc123');
+  assert.equal(track.complete, false);
+  assert.equal(track.records[0].observedAtMs, now - 50000);
+  assert.equal(track.records[0].baroAltitudeM, 304.8);
+  assert.equal(track.records[0].ellipsoidAltitudeM, null);
+});
+
+test('cancellation after slow body parsing rejects even with a transport that ignores abort', async () => {
+  let release;
+  const controller = new AbortController();
+  const source = createOpenSkySource({
+    fetchImpl: async () => ({
+      ok: true,
+      json: () =>
+        new Promise((resolve) => {
+          release = resolve;
+        }),
+    }),
+  });
+  const pending = source.getSnapshot({}, { signal: controller.signal });
+  await new Promise((resolve) => setImmediate(resolve));
+  controller.abort();
+  release({ states: [aircraft] });
+  await assert.rejects(pending, { name: 'AbortError' });
+});
+
+test('denials and outages never start another source and do not echo arbitrary response bodies', async () => {
+  for (const status of [401, 403, 429, 502]) {
+    let calls = 0;
+    const source = createOpenSkySource({
+      fetchImpl: async () => {
+        calls++;
+        return response({ error: 'sensitive upstream detail' }, {}, status);
+      },
+    });
+    await assert.rejects(source.getSnapshot(), (error) => {
+      assert.equal(error.status, status);
+      assert.equal(error.message.includes('sensitive'), false);
+      assert.ok(error.retryAfterMs >= 20000);
+      return true;
+    });
+    assert.equal(calls, 1);
+  }
+});
+
+test('AIS reports limited received coverage and keeps connection state separate from positions', async () => {
+  const source = createAisStreamSource({
+    origin: () => 'http://example.test',
+    fetchImpl: async (url) => {
+      if (url.includes('/track?'))
+        return response({ samples: [{ lat: 30, lon: -97, t: now / 1000 }] });
+      assert.equal(url, 'http://example.test/api/ais-live?maxRows=500');
+      return response({
+        status: 'reconnecting',
+        refreshing: true,
+        rows: [
+          {
+            mmsi: '123456789',
+            lat: 30,
+            lon: -97,
+            last_position_UTC: new Date(now).toISOString(),
+          },
+        ],
+        newestPositionAt: new Date(now).toISOString(),
+      });
+    },
+  });
+  const snapshot = await source.getSnapshot({ maxRows: 500 });
+  assert.equal(snapshot.complete, false);
+  assert.equal(snapshot.stale, true);
+  assert.equal(snapshot.observedAtMs, now);
+  assert.equal(snapshot.transportStatus, 'reconnecting');
+  assert.equal(
+    (await source.getTrack('123456789')).records[0].observedAtMs,
+    now,
+  );
+});
+
+test('a malformed vessel row cannot prevent admission of valid positions', async () => {
+  const source = createAisStreamSource({
+    fetchImpl: async () =>
+      response({
+        status: 'live',
+        rows: [null, { mmsi: '123456789', lat: 30, lon: -97 }],
+      }),
+  });
+  const snapshot = await source.getSnapshot();
+  assert.equal(snapshot.records.length, 1);
+  assert.equal(snapshot.rejectedCount, 1);
+  assert.equal(snapshot.complete, false);
+});
+
+test('out-of-range source epochs do not reach Date or globe time constructors', () => {
+  const row = [...aircraft];
+  row[3] = 1e100;
+  row[4] = 1e100;
+  const snapshot = openSkySnapshot({ time: 1e100, states: [row] });
+  assert.equal(snapshot.observedAtMs, null);
+  assert.equal(snapshot.records[0].positionTimeMs, null);
+  assert.equal(snapshot.records[0].contactTimeMs, null);
+});

src/sources/live/index.js

@@ -0,0 +1,18 @@
+export { LiveSourceError } from './contract.js';
+export {
+  normalizeOpenSkyAircraft,
+  normalizeReadsbAircraft,
+  normalizeAircraftTrack,
+  openSkySnapshot,
+  readsbSnapshot,
+} from './aircraft.js';
+export {
+  normalizeVesselObservation,
+  normalizeVesselTrack,
+  vesselSnapshot,
+} from './vessels.js';
+export {
+  createOpenSkySource,
+  createAdsbLolSource,
+  createAisStreamSource,
+} from './standalone.js';

src/sources/live/standalone.js

@@ -0,0 +1,201 @@
+import {
+  epoch,
+  finite,
+  httpError,
+  LiveSourceError,
+  readResponse,
+} from './contract.js';
+import {
+  normalizeAircraftTrack,
+  openSkySnapshot,
+  readsbSnapshot,
+} from './aircraft.js';
+import { normalizeVesselTrack, vesselSnapshot } from './vessels.js';
+
+const defaultFetch = (...args) => globalThis.fetch(...args);
+const header = (response, name) => response.headers?.get?.(name);
+
+function openSkyError(response) {
+  const error = httpError(response, 'OpenSky');
+  const mode = String(
+    header(response, 'x-opensky-auth-mode-used') ||
+      header(response, 'x-opensky-auth') ||
+      '',
+  ).toLowerCase();
+  const reason = String(
+    header(response, 'x-opensky-auth-reason') || '',
+  ).toLowerCase();
+  if (response.status === 429 && (!mode || mode === 'anon'))
+    error.message = 'OpenSky rate limited (anonymous)';
+  if (response.status === 401 || response.status === 403) {
+    const reasons = {
+      oauth_invalid_or_missing: 'OpenSky OAuth client missing/invalid',
+      oauth_invalid_credentials: 'OpenSky OAuth rejected credentials',
+      basic_invalid_credentials: 'OpenSky username/password rejected',
+      missing_basic_creds: 'OpenSky auth missing',
+      missing_oauth_and_basic_creds: 'OpenSky auth missing',
+      auth_required: 'OpenSky auth required',
+      forced_anonymous: 'OpenSky auth required',
+    };
+    error.message =
+      reasons[reason] ||
+      (/^(oauth_|basic_)/.test(reason)
+        ? 'OpenSky auth invalid'
+        : mode === 'anon'
+          ? 'OpenSky auth required'
+          : 'OpenSky auth failed');
+  }
+  return error;
+}
+
+/** Existing same-origin aircraft routes; no request starts during construction. */
+export function createOpenSkySource({
+  fetchImpl = defaultFetch,
+  now = () => Date.now(),
+} = {}) {
+  return {
+    label: 'OpenSky Network',
+    async getSnapshot(query = {}, { signal } = {}) {
+      const params = new URLSearchParams();
+      if (Number.isFinite(query.latitude) && Number.isFinite(query.longitude)) {
+        params.set('lat', query.latitude.toFixed(4));
+        params.set('lon', query.longitude.toFixed(4));
+      }
+      const { response, payload } = await readResponse(
+        fetchImpl,
+        `/api/opensky${params.size ? '?' + params : ''}`,
+        { signal },
+        'OpenSky',
+      );
+      if (!response.ok) throw openSkyError(response);
+      return {
+        ...openSkySnapshot(payload, {
+          source: header(response, 'x-flight-source') || 'OpenSky Network',
+          coverage:
+            header(response, 'x-flight-coverage') ||
+            'worldwide upstream snapshot',
+          now: now(),
+        }),
+        status: response.status,
+      };
+    },
+    async getTrack(reference, { signal } = {}) {
+      const { response, payload } = await readResponse(
+        fetchImpl,
+        '/api/opensky-track?icao24=' + encodeURIComponent(reference),
+        { signal },
+        'OpenSky',
+      );
+      if (!response.ok) throw httpError(response, 'OpenSky');
+      return {
+        records: normalizeAircraftTrack(payload?.path),
+        complete: false,
+      };
+    },
+    async getEnrichment(query, { signal } = {}) {
+      if (!['type', 'route'].includes(query.kind))
+        throw new LiveSourceError('unsupported', 'Enrichment unavailable');
+      const { response, payload } = await readResponse(
+        fetchImpl,
+        `/api/adsbdb/${query.kind}/${encodeURIComponent(query.id)}`,
+        { signal },
+        'adsbdb',
+      );
+      if (!response.ok) throw httpError(response, 'adsbdb');
+      return payload;
+    },
+  };
+}
+
+export function createAdsbLolSource({
+  fetchImpl = defaultFetch,
+  now = () => Date.now(),
+} = {}) {
+  return {
+    label: 'adsb.lol',
+    async getSnapshot(_query = {}, { signal } = {}) {
+      const { response, payload } = await readResponse(
+        fetchImpl,
+        '/api/adsblol/mil',
+        { signal },
+        'adsb.lol',
+      );
+      if (!response.ok) throw httpError(response, 'adsb.lol');
+      const age = finite(header(response, 'x-ads-b-cache-age-ms'));
+      return {
+        ...readsbSnapshot(payload, {
+          observedAtMs: now() - (age != null && age > 0 ? age : 0),
+          now: now(),
+          stale: header(response, 'x-ads-b-cache') === 'STALE',
+        }),
+        status: response.status,
+      };
+    },
+    async getTrack(reference, { signal } = {}) {
+      const { response, payload } = await readResponse(
+        fetchImpl,
+        '/api/adsblol/trace?hex=' + encodeURIComponent(reference),
+        { signal },
+        'adsb.lol',
+      );
+      if (!response.ok) throw httpError(response, 'adsb.lol');
+      const baseTimeMs = epoch(payload?.timestamp, 1000);
+      return {
+        records:
+          baseTimeMs == null
+            ? []
+            : normalizeAircraftTrack(payload?.trace, {
+                baseTimeMs,
+                readsb: true,
+              }),
+        complete: false,
+      };
+    },
+  };
+}
+
+export function createAisStreamSource({
+  fetchImpl = defaultFetch,
+  apiUrl = '/api/ais-live',
+  origin = () => globalThis.location?.origin || 'http://localhost',
+} = {}) {
+  return {
+    label: 'AISStream',
+    async getSnapshot({ maxRows = 12000 } = {}, { signal } = {}) {
+      const url = new URL(apiUrl, origin());
+      url.searchParams.set('maxRows', String(maxRows));
+      const { response, payload } = await readResponse(
+        fetchImpl,
+        url.toString(),
+        { signal, cache: 'no-store' },
+        'AIS live',
+      );
+      if (!response.ok) {
+        const error = httpError(response, 'AIS live');
+        const reasons = {
+          'missing-key': 'AISSTREAM_API_KEY not set',
+          'auth-failed': 'API key rejected — check AISSTREAM_API_KEY',
+          unsupported: 'live feed unsupported',
+          error: 'feed down',
+          closed: 'feed disconnected',
+        };
+        error.message = reasons[payload?.status] || error.message;
+        throw error;
+      }
+      return { ...vesselSnapshot(payload), status: response.status };
+    },
+    async getTrack(reference, { signal } = {}) {
+      const { response, payload } = await readResponse(
+        fetchImpl,
+        '/api/ais-live/track?mmsi=' + encodeURIComponent(reference),
+        { signal },
+        'AIS live',
+      );
+      if (!response.ok) throw httpError(response, 'AIS live');
+      return {
+        records: normalizeVesselTrack(payload?.samples),
+        complete: false,
+      };
+    },
+  };
+}

src/sources/live/vessels.js

@@ -0,0 +1,96 @@
+import { coordinates, epoch, finite, LiveSourceError } from './contract.js';
+
+/** AIS positions are sea-surface observations; heading and course are distinct. */
+export function normalizeVesselObservation(row, reference = null) {
+  const id = String(row?.mmsi || row?.input_identifier || '').trim();
+  const latitude = finite(row?.lat),
+    longitude = finite(row?.lon);
+  if (!id || !coordinates(latitude, longitude)) return null;
+  const speedKts = finite(row.speed);
+  return {
+    id,
+    reference: reference ?? id,
+    latitude,
+    longitude,
+    name: String(row.name || row.input_name || id),
+    imo: String(row.imo || ''),
+    type: String(row.type_specific || row.type || ''),
+    destination: String(row.destination || ''),
+    speedMps: speedKts == null ? null : speedKts * 0.514444,
+    courseDeg: finite(row.course),
+    headingDeg: finite(row.heading),
+    observedAtMs:
+      epoch(row.last_position_epoch, 1000) ??
+      epoch(Date.parse(row.last_position_UTC)),
+    altitudeDatum: 'sea-surface',
+  };
+}
+
+export function vesselSnapshot(
+  payload,
+  {
+    source = 'AISStream',
+    coverage = 'received AIS positions',
+    referenceFor = (row) => String(row.mmsi || row.input_identifier || ''),
+  } = {},
+) {
+  if (!Array.isArray(payload?.rows))
+    throw new LiveSourceError('malformed', 'Malformed vessel response');
+  const rows = payload.rows;
+  const records = [],
+    ids = new Set();
+  for (const row of rows) {
+    const record = normalizeVesselObservation(row);
+    if (record) record.reference = referenceFor(row);
+    if (record && !ids.has(record.id)) {
+      records.push(record);
+      ids.add(record.id);
+    }
+  }
+  const observedAtMs =
+    epoch(payload?.newestPositionAt) ??
+    epoch(Date.parse(payload?.newestPositionAt)) ??
+    records.reduce(
+      (newest, row) => Math.max(newest || 0, row.observedAtMs || 0) || null,
+      null,
+    );
+  return {
+    records,
+    source,
+    coverage,
+    complete: records.length === rows.length && !payload?.refreshing,
+    rejectedCount: rows.length - records.length,
+    observedAtMs,
+    freshness: payload?.refreshing
+      ? 'stale'
+      : observedAtMs == null
+        ? 'unknown'
+        : 'current',
+    stale: Boolean(payload?.refreshing),
+    // Connection state is distinct from receipt of a usable position.
+    transportStatus:
+      typeof payload?.status === 'string' ? payload.status : null,
+    lastMessageAt: payload?.lastMessageAt ?? null,
+    nextAttemptAt: finite(payload?.nextAttemptAt),
+    silentForMs: finite(payload?.silentForMs),
+    reconnectAttempt: finite(payload?.reconnectAttempt),
+    rawRowCount: rows.length,
+  };
+}
+
+export function normalizeVesselTrack(samples) {
+  if (!Array.isArray(samples)) return [];
+  return samples.flatMap((sample) => {
+    const latitude = finite(sample?.lat),
+      longitude = finite(sample?.lon);
+    if (!coordinates(latitude, longitude)) return [];
+    return [
+      {
+        latitude,
+        longitude,
+        observedAtMs: epoch(sample?.t, 1000),
+        altitudeDatum: 'sea-surface',
+      },
+    ];
+  });
+}