alembic/versions/009_vessels.py

@@ -0,0 +1,41 @@
+"""vessels — daily VesselAPI snapshots for DVR as-of
+
+Revision ID: 009_vessels
+Revises: 008_summary_kind
+Create Date: 2026-08-29
+"""
+
+from alembic import op
+
+revision = "009_vessels"
+down_revision = "008_summary_kind"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE IF NOT EXISTS vessels (
+            mmsi     TEXT NOT NULL,
+            poll_at  TIMESTAMPTZ NOT NULL,
+            lat      DOUBLE PRECISION NOT NULL,
+            lon      DOUBLE PRECISION NOT NULL,
+            heading  DOUBLE PRECISION,
+            speed    DOUBLE PRECISION,
+            label    TEXT,
+            extra    JSONB,
+            PRIMARY KEY (mmsi, poll_at)
+        )
+        """
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_vessels_poll_at ON vessels (poll_at DESC)"
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_vessels_bbox ON vessels (lon, lat)"
+    )
+
+
+def downgrade() -> None:
+    op.execute("DROP TABLE IF EXISTS vessels")

app/live_layers.py

@@ -51,7 +51,9 @@ PC_SAS_TOKEN = "https://planetarycomputer.microsoft.com/api/sas/v1/token/sentine
 # nginx vhost. Relative template — Leaflet resolves it against the page origin,
 # so the browser never touches a raw loopback port or titiler.xyz.
 TITILER_COG_TILES = f"{TITILER_PUBLIC_BASE}/cog/tiles/WebMercatorQuad/{{z}}/{{x}}/{{y}}"
-SENTINEL1_TTL = 20 * 60  # 15–30 min quota-friendly window
+SENTINEL1_TTL = 6 * 3600  # S-1 revisit is 6–12 days; cache the COG all afternoon
+SENTINEL1_CELL = 2.0  # degrees — pan/zoom inside a cell reuses the same scene
+SENTINEL1_STAC_LIMIT = 8
 SENTINEL1_ATTRIBUTION = "Copernicus Sentinel-1 / Microsoft Planetary Computer"
 
 # GPSJAM (John Wiseman / ADS-B Exchange): daily H3 hexes of aircraft nav
@@ -1244,25 +1246,71 @@ def sentinel1_tile_url(signed_cog: str) -> str:
     return f"{TITILER_COG_TILES}?{params}"
 
 
+def feature_bbox(feat: dict) -> list[float] | None:
+    """STAC Feature bbox as [minlon, minlat, maxlon, maxlat], or None."""
+    raw = feat.get("bbox") if isinstance(feat, dict) else None
+    if isinstance(raw, (list, tuple)) and len(raw) >= 4:
+        try:
+            return [float(raw[0]), float(raw[1]), float(raw[2]), float(raw[3])]
+        except (TypeError, ValueError):
+            pass
+    geom = (feat or {}).get("geometry") or {}
+    coords = geom.get("coordinates") if isinstance(geom, dict) else None
+    if not coords:
+        return None
+    lons: list[float] = []
+    lats: list[float] = []
+
+    def _walk(node: Any) -> None:
+        if isinstance(node, (list, tuple)) and node and isinstance(node[0], (int, float)):
+            lons.append(float(node[0]))
+            lats.append(float(node[1]))
+        elif isinstance(node, (list, tuple)):
+            for child in node:
+                _walk(child)
+
+    _walk(coords)
+    if not lons:
+        return None
+    return [min(lons), min(lats), max(lons), max(lats)]
+
+
+def pick_sentinel_feature(features: list, lon: float, lat: float) -> dict | None:
+    """Prefer the scene whose bbox covers the viewport center; else first."""
+    if not features:
+        return None
+    for feat in features:
+        bb = feature_bbox(feat)
+        if bb and bb[0] <= lon <= bb[2] and bb[1] <= lat <= bb[3]:
+            return feat
+    return features[0]
+
+
 async def fetch_sentinel1(bbox: str) -> dict | None:
     """Most recent Sentinel-1 GRD COG for a viewport, signed and TiTiler-ready.
 
     Returns the overlay tile-template dict, or ``None`` when no GRD imagery
     covers the bbox in the last 7 days (caller maps to 404). Queries Planetary
-    Computer only when called; cached per quantized bbox + UTC day.
+    Computer only when called; cached per 2° cell + UTC day so pan/zoom inside
+    the same region reuses the COG instead of picking a neighbouring swath.
     """
     minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
+    qminlon, qminlat, qmaxlon, qmaxlat = quantize_bbox(
+        minlon, minlat, maxlon, maxlat, step=SENTINEL1_CELL,
+    )
     day = datetime.now(timezone.utc).date().isoformat()
-    key = f"sentinel1:{day}:{bbox_cell_key(bbox)}"
+    key = f"sentinel1:{day}:{qminlon:.4f},{qminlat:.4f},{qmaxlon:.4f},{qmaxlat:.4f}"
+    clon = (qminlon + qmaxlon) / 2.0
+    clat = (qminlat + qmaxlat) / 2.0
 
     async def _load() -> dict | None:
         now = datetime.now(timezone.utc)
         week_ago = now - timedelta(days=7)
         payload = {
             "collections": ["sentinel-1-grd"],
-            "bbox": [minlon, minlat, maxlon, maxlat],
+            "bbox": [qminlon, qminlat, qmaxlon, qmaxlat],
             "datetime": f"{week_ago.isoformat()}/{now.isoformat()}",
-            "limit": 1,
+            "limit": SENTINEL1_STAC_LIMIT,
             "sortby": [{"field": "datetime", "direction": "desc"}],
         }
         data = await _pc_call(_post_json(PC_STAC_SEARCH, json=payload))
@@ -1270,7 +1318,9 @@ async def fetch_sentinel1(bbox: str) -> dict | None:
         if not features:
             return None
 
-        feat = features[0]
+        feat = pick_sentinel_feature(features, clon, clat)
+        if not feat:
+            return None
         assets = feat.get("assets") or {}
         chosen_href: str | None = None
         polarization: str | None = None
@@ -1299,6 +1349,7 @@ async def fetch_sentinel1(bbox: str) -> dict | None:
             "datetime": props.get("datetime") or feat.get("datetime"),
             "polarization": polarization,
             "attribution": SENTINEL1_ATTRIBUTION,
+            "bbox": feature_bbox(feat),
         }
 
     return await _ttl_get(key, float(SENTINEL1_TTL), _load)

app/main.py

@@ -74,6 +74,11 @@ async def _lifespan(app: FastAPI):
         await refresh_cache()
     except Exception:
         pass
+    try:
+        from vesselapi import hydrate_last_known
+        await hydrate_last_known()
+    except Exception:
+        pass
     from config import AISSTREAM_IN_APP, VESSELAPI_IN_APP
     ais_task = None
     vesselapi_task = None
@@ -1263,6 +1268,7 @@ async def list_news_summaries(
 
 
 _FLAGGED = ("critical", "high")
+_LESSER = ("medium", "low")
 
 
 @app.get("/api/news/ticker", response_model=list[NewsTickerItemOut])
@@ -1273,7 +1279,7 @@ async def list_news_ticker(
     ),
     limit: int = Query(20, ge=1, le=50),
 ):
-    """Flagged ticker rows (critical/high), newest first. No LLM required."""
+    """Ticker rows: critical/high first; medium/low if nothing is flagged."""
     async with async_session() as session:
         stmt = (
             select(news_items)
@@ -1287,6 +1293,19 @@ async def list_news_ticker(
             stmt = stmt.where(news_items.c.created_at >= since)
         stmt = stmt.limit(limit)
         rows = (await session.execute(stmt)).mappings().all()
+        if not rows:
+            stmt = (
+                select(news_items)
+                .where(
+                    news_items.c.kind == "ticker",
+                    news_items.c.importance.in_(_LESSER),
+                )
+                .order_by(news_items.c.created_at.desc())
+            )
+            if since:
+                stmt = stmt.where(news_items.c.created_at >= since)
+            stmt = stmt.limit(limit)
+            rows = (await session.execute(stmt)).mappings().all()
     return [
         NewsTickerItemOut(
             id=r["id"], headline=r["headline"], importance=r["importance"],
@@ -1530,7 +1549,17 @@ async def list_vessels(
         from tracks import fetch_positions_at, parse_timestamp
         ts = parse_timestamp(timestamp)
         if ts is not None:
-            return overlay_json(await fetch_positions_at("vessel", ts, bbox, limit), 5)
+            from vesselapi import fetch_vessels_as_of
+            if src == "vesselapi":
+                return overlay_json(await fetch_vessels_as_of(ts, bbox, limit), 5)
+            ais = await fetch_positions_at("vessel", ts, bbox, limit)
+            if src == "aisstream":
+                return overlay_json(ais, 5)
+            va = await fetch_vessels_as_of(ts, bbox, limit)
+            by_id = {m["id"]: m for m in ais}
+            for m in va:
+                by_id[m["id"]] = m
+            return overlay_json(list(by_id.values())[:limit], 5)
         return overlay_json(await fetch_vessels(bbox, limit, src=src), 5)
     except ValueError as exc:
         raise HTTPException(422, str(exc)) from exc

app/static/index.html

@@ -358,6 +358,11 @@
         .lp-future .lp-name { cursor: not-allowed; }
         .lp-error { color: var(--red); font-size: 0.66rem; }
         .lp-note { font-size: 0.6rem; color: var(--muted); opacity: 0.85; line-height: 1.35; }
+        .gf-list { display: flex; flex-direction: column; gap: 0.22rem; max-height: 8rem; overflow-y: auto; }
+        .gf-item { display: flex; justify-content: space-between; align-items: center; gap: 0.4rem; font-size: 0.64rem; }
+        .gf-item span { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; color: var(--text); }
+        .gf-del { background: transparent; border: 1px solid var(--magenta); color: var(--magenta); font-family: 'Share Tech Mono', monospace; font-size: 0.58rem; letter-spacing: 0.08em; text-transform: uppercase; padding: 0.12rem 0.35rem; border-radius: 3px; cursor: pointer; }
+        .gf-del:hover { background: rgba(255,46,151,0.16); }
 
         /* ── Camera / blip popup thumbnails ── */
         .cam-pop { min-width: 210px; max-width: 260px; }
@@ -583,6 +588,8 @@
         .tick-item .b-tag { font-family: 'Share Tech Mono', monospace; font-size: 0.6rem; color: #1c0311; background: var(--magenta); border-radius: 2px; padding: 0.08rem 0.4rem; letter-spacing: 0.1em; }
         .tick-item .b-tag.critical { background: var(--red); }
         .tick-item .b-tag.high { background: var(--amber); }
+        .tick-item .b-tag.medium { background: var(--cyan); color: #041018; }
+        .tick-item .b-tag.low { background: var(--muted); color: #041018; }
         .tick-item.brief:hover { color: var(--magenta); }
         .tick-item.standby { color: var(--muted); opacity: 0.75; }
         .tick-item.standby .price { color: var(--muted); }
@@ -906,7 +913,8 @@
                             <div class="lp-row">
                                 <button class="btn" id="gf-draw" type="button" onclick="toggleGeofenceDraw()">Draw geofence</button>
                             </div>
-                            <div class="lp-note">No Leaflet.Draw — click vertices, double-click to close. Saved to POST /api/geofences.</div>
+                            <div id="gf-list" class="gf-list"></div>
+                            <div class="lp-note">Click vertices, double-click to close. Delete a saved fence from the list or its popup.</div>
                         </div>
                         <div class="lp-layer">
                             <div class="lp-row">
@@ -1420,7 +1428,7 @@ function initMarketTicker() {
 
 /* ═══════════════ NEWS FEED + TICKER ═══════════════ */
 let newsInterval = null;
-const NEWS_REFRESH_MS = 900000; // 15-min cycle
+const NEWS_REFRESH_MS = 180000; // 3-min ticker/brief poll
 let newsTickerBuilt = false;
 
 function newsEsc(s) {
@@ -1527,10 +1535,11 @@ function renderNewsTicker(articles, summaries, tickerItems) {
     if (ticks.length) {
         ticks.forEach(t => {
             const imp = String(t.importance || '').toLowerCase();
-            const tag = imp === 'critical' ? 'CRITICAL' : 'HIGH';
+            const tag = imp === 'critical' ? 'CRITICAL' : imp === 'high' ? 'HIGH' : imp === 'medium' ? 'MED' : 'LOW';
+            const tagClass = (imp === 'critical' || imp === 'high' || imp === 'medium' || imp === 'low') ? imp : 'high';
             const loc = (t.location_name || '').trim();
             const inner =
-                `<span class="b-tag ${imp === 'critical' ? 'critical' : 'high'}">${tag}</span>` +
+                `<span class="b-tag ${tagClass}">${tag}</span>` +
                 `<span>${newsEsc((t.headline || '').trim())}</span>` +
                 (loc ? `<span class="dom">${newsEsc(loc)}</span>` : '') +
                 `<span class="tt">${newsTimeAgo(t.created_at)}</span><span class="sep">▸</span>`;
@@ -1906,6 +1915,7 @@ let extraAttribs = new Set();
 let radarLayer = null, radarOn = true, radarOpacity = 0.7, radarMeta = null, radarTimer = null;
 let thermalLayer = null, thermalOn = false;
 let sentinelLayer = null, sentinelOn = false, sentinelOpacity = 0.8;
+let sentinelItemId = null, sentinelBounds = null;
 let wxAlertsGroup = null, wxAlertsOn = true;
 let perimGroup = null, perimOn = true;
 let incidentsGroup = null, incidentsOn = false;
@@ -3071,6 +3081,8 @@ async function toggleSentinel1() {
     if (sentinelOn) await loadSentinel1();
     else {
         sentinelLayer = dropLayer(sentinelLayer);
+        sentinelItemId = null;
+        sentinelBounds = null;
         const n = document.getElementById('lp-sentinel-count');
         if (n) n.textContent = '7d';
     }
@@ -3080,8 +3092,15 @@ function setSentinelOpacity(v) {
     document.getElementById('lp-sentinel-val').textContent = `${Math.round(v)}%`;
     if (sentinelLayer) sentinelLayer.setOpacity(sentinelOpacity);
 }
+function sentinelStillCovers() {
+    if (!map || !sentinelBounds || sentinelBounds.length < 4) return false;
+    const c = map.getCenter();
+    return c.lng >= sentinelBounds[0] && c.lng <= sentinelBounds[2]
+        && c.lat >= sentinelBounds[1] && c.lat <= sentinelBounds[3];
+}
 async function loadSentinel1() {
     if (!map || !sentinelOn) return;
+    if (sentinelLayer && sentinelStillCovers()) return;
     const req = ++overlayReq.sar;
     const countEl = document.getElementById('lp-sentinel-count');
     const hint = document.getElementById('map-hint');
@@ -3093,6 +3112,8 @@ async function loadSentinel1() {
         if (req !== overlayReq.sar) return;
         if (r.status === 404 && body && body.error === 'no_imagery') {
             sentinelLayer = dropLayer(sentinelLayer);
+            sentinelItemId = null;
+            sentinelBounds = null;
             if (hint) hint.textContent = 'No Sentinel-1 imagery for this view in the last 7 days.';
             if (countEl) countEl.textContent = 'none';
             return;
@@ -3109,13 +3130,31 @@ async function loadSentinel1() {
             if (countEl) countEl.textContent = 'err';
             return;
         }
+        if (body.itemId && body.itemId === sentinelItemId && sentinelLayer) {
+            if (Array.isArray(body.bbox) && body.bbox.length >= 4) sentinelBounds = body.bbox;
+            sentinelLayer.setOpacity(sentinelOpacity);
+            if (countEl) countEl.textContent = String(body.polarization || 'SAR').toUpperCase();
+            return;
+        }
         const attrib = body.attribution || '';
-        sentinelLayer = dropLayer(sentinelLayer);
+        const layerOpts = {
-        sentinelLayer = L.tileLayer(body.tileUrl, {
             opacity: sentinelOpacity,
             maxZoom: 18,
             attribution: attrib,
-        }).addTo(map);
+            noWrap: true,
+        };
+        if (Array.isArray(body.bbox) && body.bbox.length >= 4) {
+            sentinelBounds = body.bbox;
+            layerOpts.bounds = L.latLngBounds(
+                [body.bbox[1], body.bbox[0]],
+                [body.bbox[3], body.bbox[2]],
+            );
+        } else {
+            sentinelBounds = null;
+        }
+        sentinelLayer = dropLayer(sentinelLayer);
+        sentinelLayer = L.tileLayer(body.tileUrl, layerOpts).addTo(map);
+        sentinelItemId = body.itemId || null;
         if (countEl) countEl.textContent = String(body.polarization || 'SAR').toUpperCase();
         if (hint) {
             const when = body.datetime ? ` · ${body.datetime}` : '';
@@ -3477,17 +3516,44 @@ async function onGfClose(e) {
     toggleGeofenceDraw();
     loadGeofences();
 }
+async function deleteGeofence(id) {
+    if (!id) return;
+    try {
+        const r = await fetch(`${API}/api/geofences/${encodeURIComponent(id)}`, { method: 'DELETE' });
+        if (!r.ok && r.status !== 204) throw new Error('delete failed');
+    } catch (err) { console.error('geofence delete failed', err); }
+    loadGeofences();
+}
 async function loadGeofences() {
     if (!map) return;
     try {
         const r = await overlayFetch(`${API}/api/geofences`);
         const rows = await r.json();
+        const list = Array.isArray(rows) ? rows : [];
+        const box = document.getElementById('gf-list');
+        if (box) {
+            box.innerHTML = list.map(f => {
+                const id = newsEsc(f.id || '');
+                const name = newsEsc(f.name || 'Fence');
+                return `<div class="gf-item"><span title="${name}">${name}</span>` +
+                    `<button class="gf-del" type="button" onclick="deleteGeofence('${id}')">Delete</button></div>`;
+            }).join('');
+        }
         if (gfSaved && map.hasLayer(gfSaved)) map.removeLayer(gfSaved);
-        const feats = (Array.isArray(rows) ? rows : []).map(f => ({
+        const feats = list.map(f => ({
             type: 'Feature', properties: { name: f.name, id: f.id }, geometry: f.geojson,
         }));
         gfSaved = L.geoJSON({ type: 'FeatureCollection', features: feats }, {
             style: { color: '#ff2e97', weight: 2, fillOpacity: 0.08 },
+            onEachFeature: (feat, layer) => {
+                const id = feat.properties && feat.properties.id;
+                const name = newsEsc((feat.properties && feat.properties.name) || 'Fence');
+                if (!id) return;
+                layer.bindPopup(
+                    `<div class="gf-pop"><b>${name}</b><br>` +
+                    `<button class="gf-del" type="button" onclick="deleteGeofence('${newsEsc(id)}')">Delete</button></div>`
+                );
+            },
         }).addTo(map);
     } catch (e) { if (!isAbort(e)) console.error('geofences load failed', e); }
 }

app/tracks.py

@@ -186,15 +186,14 @@ async def track_range() -> dict:
         async with async_session() as session:
             row = (await session.execute(text(
                 """
-                SELECT
+                SELECT min(t) AS tmin, max(t) AS tmax FROM (
-                  LEAST(
+                  SELECT min(bucket) AS t FROM vessel_tracks_1min
-                    (SELECT min(bucket) FROM vessel_tracks_1min),
+                  UNION ALL SELECT max(bucket) FROM vessel_tracks_1min
-                    (SELECT min(bucket) FROM aircraft_tracks_1min)
+                  UNION ALL SELECT min(bucket) FROM aircraft_tracks_1min
-                  ) AS tmin,
+                  UNION ALL SELECT max(bucket) FROM aircraft_tracks_1min
-                  GREATEST(
+                  UNION ALL SELECT min(poll_at) FROM vessels
-                    (SELECT max(bucket) FROM vessel_tracks_1min),
+                  UNION ALL SELECT max(poll_at) FROM vessels
-                    (SELECT max(bucket) FROM aircraft_tracks_1min)
+                ) s
-                  ) AS tmax
                 """
             ))).mappings().first()
         if not row or row["tmin"] is None:

app/vesselapi.py

@@ -17,6 +17,7 @@ from __future__ import annotations
 
 import asyncio
 import calendar
+import json
 import logging
 import os
 from datetime import date, datetime, timezone
@@ -32,7 +33,7 @@ from config import (
     VESSELAPI_MAX_CALLS_PER_DAY,
 )
 from database import async_session, engine, metadata
-from live_layers import to_marker, upsert_vessel
+from live_layers import parse_bbox, to_marker, upsert_vessel, vessel_last_known, vessel_lock
 
 logger = logging.getLogger("osint.vesselapi")
 
@@ -176,6 +177,49 @@ def transform_vesselapi_payload(payload: dict | None) -> list[dict]:
     return out
 
 
+def utc_day_start(now: datetime) -> datetime:
+    """Floor ``now`` to 00:00:00 UTC."""
+    if now.tzinfo is None:
+        now = now.replace(tzinfo=timezone.utc)
+    now = now.astimezone(timezone.utc)
+    return now.replace(hour=0, minute=0, second=0, microsecond=0)
+
+
+def pick_poll_at(poll_times: list[datetime], as_of: datetime) -> datetime | None:
+    """Latest poll timestamp at or before ``as_of`` (DVR as-of)."""
+    if as_of.tzinfo is None:
+        as_of = as_of.replace(tzinfo=timezone.utc)
+    else:
+        as_of = as_of.astimezone(timezone.utc)
+    eligible: list[datetime] = []
+    for raw in poll_times:
+        ts = raw if raw.tzinfo else raw.replace(tzinfo=timezone.utc)
+        ts = ts.astimezone(timezone.utc)
+        if ts <= as_of:
+            eligible.append(ts)
+    return max(eligible) if eligible else None
+
+
+def snapshot_as_of(rows: list[dict], as_of: datetime) -> list[dict]:
+    """Keep only rows from the latest poll_at ≤ ``as_of``."""
+    chosen = pick_poll_at(
+        [r["poll_at"] for r in rows if r.get("poll_at") is not None],
+        as_of,
+    )
+    if chosen is None:
+        return []
+    out = []
+    for row in rows:
+        ts = row.get("poll_at")
+        if ts is None:
+            continue
+        if ts.tzinfo is None:
+            ts = ts.replace(tzinfo=timezone.utc)
+        if ts.astimezone(timezone.utc) == chosen:
+            out.append(row)
+    return out
+
+
 # ── Durable daily quota (Postgres, survives restarts) ─────────────────────
 # Mirrors keystore.api_keys: lazy CREATE TABLE IF NOT EXISTS, no alembic fork.
 
@@ -261,6 +305,183 @@ class PgQuotaStore:
         return (int(existing) if existing else 0) + 1
 
 
+# ── Daily VesselAPI snapshots (DVR as-of + survive restarts) ──────────────
+# Cleared at the UTC day boundary so the table holds today's 5 polls only.
+
+_CREATE_VESSELS_SQL = text(
+    """
+    CREATE TABLE IF NOT EXISTS vessels (
+        mmsi     TEXT NOT NULL,
+        poll_at  TIMESTAMPTZ NOT NULL,
+        lat      DOUBLE PRECISION NOT NULL,
+        lon      DOUBLE PRECISION NOT NULL,
+        heading  DOUBLE PRECISION,
+        speed    DOUBLE PRECISION,
+        label    TEXT,
+        extra    JSONB,
+        PRIMARY KEY (mmsi, poll_at)
+    )
+    """
+)
+_CREATE_VESSELS_POLL_IDX = text(
+    "CREATE INDEX IF NOT EXISTS ix_vessels_poll_at ON vessels (poll_at DESC)"
+)
+_CREATE_VESSELS_BBOX_IDX = text(
+    "CREATE INDEX IF NOT EXISTS ix_vessels_bbox ON vessels (lon, lat)"
+)
+
+_vessels_lock = asyncio.Lock()
+_vessels_ensured = False
+
+
+async def ensure_vessels_table() -> None:
+    global _vessels_ensured
+    if _vessels_ensured:
+        return
+    async with _vessels_lock:
+        if _vessels_ensured:
+            return
+        async with engine.begin() as conn:
+            await conn.execute(_CREATE_VESSELS_SQL)
+            await conn.execute(_CREATE_VESSELS_POLL_IDX)
+            await conn.execute(_CREATE_VESSELS_BBOX_IDX)
+        _vessels_ensured = True
+
+
+def _marker_from_vessel_row(r) -> dict:
+    extra = r.get("extra") or {}
+    if isinstance(extra, str):
+        try:
+            extra = json.loads(extra)
+        except (TypeError, ValueError):
+            extra = {}
+    if not isinstance(extra, dict):
+        extra = {}
+    extra.setdefault("src", "vesselapi")
+    poll_at = r.get("poll_at")
+    if poll_at is not None and hasattr(poll_at, "isoformat"):
+        extra["poll_at"] = poll_at.isoformat()
+    marker = to_marker(
+        str(r["id"]), r["lat"], r["lon"],
+        heading=r.get("heading"), speed=r.get("speed"),
+        label=r.get("label") or str(r["id"]),
+        extra=extra,
+    )
+    marker["seen_at"] = extra.get("poll_at") or datetime.now(timezone.utc).isoformat()
+    return marker
+
+
+async def persist_vessel_snapshot(markers: list[dict], poll_at: datetime) -> None:
+    """Write one VesselAPI poll into ``vessels`` (today's snapshots)."""
+    await ensure_vessels_table()
+    if not markers:
+        return
+    async with async_session() as session:
+        for m in markers:
+            vid = str(m.get("id") or "")
+            lat, lon = m.get("lat"), m.get("lon")
+            if not vid or lat is None or lon is None:
+                continue
+            extra = dict(m.get("extra") or {})
+            extra.setdefault("src", "vesselapi")
+            await session.execute(
+                text(
+                    """
+                    INSERT INTO vessels
+                        (mmsi, poll_at, lat, lon, heading, speed, label, extra)
+                    VALUES
+                        (:mmsi, :poll_at, :lat, :lon, :heading, :speed, :label,
+                         CAST(:extra AS jsonb))
+                    ON CONFLICT (mmsi, poll_at) DO UPDATE SET
+                        lat = EXCLUDED.lat,
+                        lon = EXCLUDED.lon,
+                        heading = EXCLUDED.heading,
+                        speed = EXCLUDED.speed,
+                        label = EXCLUDED.label,
+                        extra = EXCLUDED.extra
+                    """
+                ),
+                {
+                    "mmsi": vid,
+                    "poll_at": poll_at,
+                    "lat": float(lat),
+                    "lon": float(lon),
+                    "heading": m.get("heading"),
+                    "speed": m.get("speed"),
+                    "label": m.get("label") or vid,
+                    "extra": json.dumps(extra),
+                },
+            )
+        await session.commit()
+
+
+async def purge_old_vessels(before: datetime | None = None) -> None:
+    """Drop snapshots from before the current UTC day (or ``before``)."""
+    await ensure_vessels_table()
+    cutoff = before or utc_day_start(datetime.now(timezone.utc))
+    async with async_session() as session:
+        await session.execute(
+            text("DELETE FROM vessels WHERE poll_at < :cutoff"),
+            {"cutoff": cutoff},
+        )
+        await session.commit()
+
+
+async def fetch_vessels_as_of(
+    ts: datetime,
+    bbox: str | None = None,
+    limit: int = 2000,
+) -> list[dict]:
+    """Latest VesselAPI poll at or before ``ts`` (DVR as-of, not exact minute)."""
+    try:
+        await ensure_vessels_table()
+        async with async_session() as session:
+            poll = (await session.execute(
+                text("SELECT max(poll_at) FROM vessels WHERE poll_at <= :ts"),
+                {"ts": ts},
+            )).scalar()
+            if poll is None:
+                return []
+            sql = """
+                SELECT mmsi AS id, lat, lon, heading, speed, label, extra, poll_at
+                FROM vessels
+                WHERE poll_at = :poll
+            """
+            params: dict = {"poll": poll, "limit": limit}
+            if bbox:
+                minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
+                sql += (
+                    " AND lon BETWEEN :minlon AND :maxlon"
+                    " AND lat BETWEEN :minlat AND :maxlat"
+                )
+                params.update(
+                    minlon=minlon, minlat=minlat, maxlon=maxlon, maxlat=maxlat,
+                )
+            sql += " LIMIT :limit"
+            rows = (await session.execute(text(sql), params)).mappings().all()
+        return [_marker_from_vessel_row(r) for r in rows]
+    except Exception:
+        logger.exception("VesselAPI snapshot fetch failed")
+        return []
+
+
+async def hydrate_last_known() -> int:
+    """Seed in-memory last-known from today's latest poll (app boot)."""
+    try:
+        rows = await fetch_vessels_as_of(datetime.now(timezone.utc))
+    except Exception:
+        logger.exception("VesselAPI hydrate failed")
+        return 0
+    if not rows:
+        return 0
+    async with vessel_lock:
+        for m in rows:
+            vid = str(m.get("id") or "")
+            if vid:
+                vessel_last_known[vid] = m
+    return len(rows)
+
+
 # ── Budget / scheduling (pure, unit-testable) ─────────────────────────────
 
 def days_left_in_month(now: datetime) -> int:
@@ -406,6 +627,11 @@ async def poll_once(store, boxes: list[tuple[float, float, float, float]], key:
     markers = transform_vesselapi_payload(data)
     for m in markers:
         await upsert_vessel(m)
+    try:
+        await persist_vessel_snapshot(markers, now)
+        await purge_old_vessels(utc_day_start(now))
+    except Exception:  # noqa: BLE001 — live overlay must not die on persist
+        logger.exception("VesselAPI snapshot persist failed")
     logger.info(
         "VesselAPI poll OK: %d vessels (remaining=%s, calls_today=%d)",
         len(markers), remaining, calls,

docs/news.md

@@ -128,11 +128,12 @@ Key set **unchanged** (no `lat`/`lon` on articles; geo lives on `/api/news/map`)
 `?kind=daily_recap` pins the nightly 24h recap. Empty DB → `[]` (no crash).
 Malformed `kind` → `422`.
 
-### GET /api/news/ticker — flagged HUD headlines
+### GET /api/news/ticker — HUD headlines
 
-Critical/high `news_items` with `kind=ticker` only. Do **not** reuse
+Critical/high `news_items` with `kind=ticker` first. If none are flagged,
-`GET /api/alerts`. Bottom HUD `#nt-track` scrolls these rows, not a dump of
+medium/low ticker rows fill the tape so the dock is not blank. Do **not**
-the whole brief.
+reuse `GET /api/alerts`. Bottom HUD `#nt-track` scrolls these rows, not a
+dump of the whole brief.
 
 | Query param | Meaning | Default |
 |---|---|---|
@@ -226,10 +227,10 @@ markdown json fences, then brace-slices:
 }
 ```
 
-Persist ticker/map only for `importance` in `critical`/`high`. Map rows also
+Persist ticker for critical/high first; if none, persist medium/low so the
-need valid coords; Unknown / invented places are dropped. Caps: 12 ticker
+tape is not empty. Map rows stay critical/high with valid coords; Unknown /
-(≤140 chars, no markdown), 20 map. Empty ticker is allowed. `summary_en`
+invented places are dropped. Caps: 12 ticker (≤140 chars, no markdown), 20
-lands in `article_summaries.summary_text`.
+map. `summary_en` lands in `article_summaries.summary_text`.
 
 ## Configuration (all via env / `.env`)
 

news/summerizer/intel.py

@@ -7,6 +7,7 @@ import re
 
 _EMPTY = {"summary_en": "", "ticker": [], "map_items": []}
 _KEEP = frozenset({"critical", "high"})
+_RANK = {"critical": 0, "high": 1, "medium": 2, "low": 3}
 _THINK_RE = re.compile(r"<think>.*?</think>", re.DOTALL)
 _FENCE_RE = re.compile(r"```(?:json)?", re.IGNORECASE)
 
@@ -58,19 +59,29 @@ def _trimmed_headline(row: dict, limit: int) -> str:
 
 
 def select_ticker(rows: list) -> list:
-    out = []
+    flagged = []
+    medium = []
+    low = []
     for row in rows:
-        if row.get("importance") not in _KEEP:
+        imp = row.get("importance")
+        if imp not in _RANK:
             continue
         headline = _trimmed_headline(row, TICKER_HEADLINE_MAX)
         if not headline:
             continue
         item = dict(row)
         item["headline"] = headline
-        out.append(item)
+        if imp in _KEEP:
-        if len(out) >= TICKER_CAP:
+            flagged.append(item)
+        elif imp == "medium":
+            medium.append(item)
+        else:
+            low.append(item)
+        if len(flagged) >= TICKER_CAP:
             break
-    return out
+    if flagged:
+        return flagged[:TICKER_CAP]
+    return (medium + low)[:TICKER_CAP]
 
 
 def select_map(items: list) -> list:

news/summerizer/nous_client.py

@@ -8,6 +8,10 @@ import httpx
 
 _DEFAULT_UA = "osint-dashboard-news-summarizer"
 _DEFAULT_BASE = "https://inference-api.nousresearch.com/v1"
+_JSON_SYSTEM = (
+    "You are an OSINT executive briefer. Reply with a single complete JSON object. "
+    "Never truncate mid-sentence. If you run out of room, drop the lowest-priority item."
+)
 
 
 def chat(prompt, *, api_key, model, base_url, json_mode=False) -> str:
@@ -17,20 +21,37 @@ def chat(prompt, *, api_key, model, base_url, json_mode=False) -> str:
         "Authorization": f"Bearer {api_key}",
         "User-Agent": os.environ.get("OSINT_USER_AGENT") or _DEFAULT_UA,
     }
+    max_tokens = 8192 if json_mode else 4096
+    timeout = 120.0 if json_mode else 60.0
+    messages = [{"role": "user", "content": prompt}]
+    if json_mode:
+        messages = [
+            {"role": "system", "content": _JSON_SYSTEM},
+            {"role": "user", "content": prompt},
+        ]
     payload = {
         "model": model,
-        "messages": [{"role": "user", "content": prompt}],
+        "messages": messages,
         "temperature": 0.2,
-        "max_tokens": 4096,
+        "max_tokens": max_tokens,
     }
     if json_mode:
         payload["response_format"] = {"type": "json_object"}
+    last_content = ""
     try:
-        with httpx.Client(timeout=60.0) as client:
+        for attempt in range(2):
-            resp = client.post(url, headers=headers, json=payload)
+            with httpx.Client(timeout=timeout) as client:
-        if resp.status_code == 401 or resp.status_code >= 500:
+                resp = client.post(url, headers=headers, json=payload)
-            return ""
+            if resp.status_code == 401 or resp.status_code >= 500:
-        data = resp.json()
+                return ""
-        return data["choices"][0]["message"]["content"]
+            data = resp.json()
+            choice = (data.get("choices") or [{}])[0]
+            last_content = (choice.get("message") or {}).get("content") or ""
+            finish = choice.get("finish_reason")
+            if finish == "length" and attempt == 0:
+                payload["max_tokens"] = min(int(payload["max_tokens"]) * 2, 16384)
+                continue
+            return last_content
+        return last_content
     except Exception:
         return ""

news/summerizer/summarizer.py

@@ -94,7 +94,7 @@ FUTURES_TICKERS = {
 MAP_PROMPT_DEFAULT = """\
 You are a precise, factual OSINT news processor. Your ONLY source of information is the articles provided below. Do NOT add external knowledge, assumptions, training data, or invented facts.
 
-Focus on breaking important news (geopolitical, military/conflict, security, disasters, major political developments). Ignore futures prices, commodity tape, ticker chatter, and routine market moves unless they themselves are the breaking event.
+Focus on breaking important news (geopolitical, military/conflict, security, disasters, major political developments). Ignore futures prices, commodity tape, ticker chatter, and routine market moves unless they themselves are the breaking event. If the batch has no critical/high stories, still extract minor incidents and crime reports.
 
 Write every field in English. Translate if the article is not English.
 
@@ -133,7 +133,11 @@ You are writing an English operator HUD brief from the article facts in DATA bel
 
 Always write a real summary_en that recaps the most important stories present in DATA. Rank geopolitics, military/conflict, security, disasters, and major political developments first. Ignore futures prices, commodity tape, ticker chatter, and routine market data — do not treat price ticks as news.
 
-ticker and map_items may be empty if nothing is critical or high. Never replace summary_en with a canned empty-brief sentence when DATA contains article facts.
+Lead with critical and high breaking events. If DATA has no critical/high stories, fill the brief with minor incidents and crime reports rather than writing an empty or unfinished brief. Never truncate mid-sentence; finish every sentence. If you run out of room, drop the lowest-priority item instead of cutting a line short.
+
+ticker: prefer critical and high. If nothing is critical or high, fill ticker with medium then low incidents and crime so the HUD is not blank.
+
+map_items may be empty if no located critical/high event is explicit in the data.
 
 Demand a single JSON object (no markdown fences) with this exact shape:
 
@@ -143,9 +147,9 @@ Demand a single JSON object (no markdown fences) with this exact shape:
   "map_items": [{"headline": "", "importance": "critical", "location_name": "", "lat": 0, "lon": 0, "location_confidence": "city", "category": "military/conflict", "url": ""}]
 }
 
-ticker: only critical and high, max 12, ≤140 chars, no markdown.
+ticker: max 12, ≤140 chars, no markdown. Rank critical > high > medium > low.
-map_items: only critical and high where a real-world location is explicit in the data. Estimate lat/lon. If location is Unknown or not in the data, omit the item. Never invent a place. Max 20.
+map_items: only where a real-world location is explicit in the data. Estimate lat/lon. If location is Unknown or not in the data, omit the item. Never invent a place. Max 20.
-summary_en: English markdown brief of breaking important news for an operator HUD (bullets or short paragraphs). Cover the actual stories in DATA.
+summary_en: English markdown executive brief for an operator HUD (4–8 complete bullets or short paragraphs). Cover the actual stories in DATA. Complete — never an unfinished sentence.
 
 DATA:
 {final_input}
@@ -156,7 +160,9 @@ You are writing a daily recap of the last 24 hours of news for an OSINT operator
 
 Always write a real summary_en daily recap of the most important stories in DATA. Rank geopolitics, military/conflict, security, disasters, and major political developments first. Ignore futures prices, commodity tape, ticker chatter, and routine market data — do not treat price ticks as news.
 
-ticker and map_items may be empty if nothing is critical or high. Never replace summary_en with a canned empty-brief sentence when DATA contains article facts.
+Lead with critical and high breaking events. If DATA has no critical/high stories, fill the recap with minor incidents and crime reports rather than writing an empty or unfinished recap. Never truncate mid-sentence; finish every sentence.
+
+ticker: prefer critical and high. If nothing is critical or high, fill ticker with medium then low incidents and crime so the HUD is not blank.
 
 Demand a single JSON object (no markdown fences) with this exact shape:
 
@@ -166,9 +172,9 @@ Demand a single JSON object (no markdown fences) with this exact shape:
   "map_items": [{"headline": "", "importance": "critical", "location_name": "", "lat": 0, "lon": 0, "location_confidence": "city", "category": "military/conflict", "url": ""}]
 }
 
-ticker: only critical and high, max 12, ≤140 chars, no markdown.
+ticker: max 12, ≤140 chars, no markdown. Rank critical > high > medium > low.
-map_items: only critical and high where a real-world location is explicit in the data. Estimate lat/lon. If location is Unknown or not in the data, omit the item. Never invent a place. Max 20.
+map_items: only where a real-world location is explicit in the data. Estimate lat/lon. If location is Unknown or not in the data, omit the item. Never invent a place. Max 20.
-summary_en: English markdown daily recap of the last 24 hours of breaking important news. Cover the actual stories in DATA.
+summary_en: English markdown daily recap of the last 24 hours. Complete sentences. Cover the actual stories in DATA.
 
 DATA:
 {final_input}

news/summerizer/tests/test_intel.py

@@ -32,6 +32,16 @@ def test_select_ticker_keeps_critical_high_caps_12():
     assert len(out) == 12
     assert all(r["importance"] in ("critical", "high") for r in out)
 
+
+def test_select_ticker_falls_back_to_medium_low_when_nothing_flagged():
+    rows = [
+        {"headline": "shop theft", "importance": "low"},
+        {"headline": "highway crash", "importance": "medium"},
+        {"headline": "none", "importance": "none"},
+    ]
+    out = select_ticker(rows)
+    assert [r["headline"] for r in out] == ["highway crash", "shop theft"]
+
 def test_select_map_requires_valid_coords_and_flag():
     items = [
         {"headline": "A", "importance": "critical", "lat": 50.45, "lon": 30.52, "location_name": "Kyiv"},

news/summerizer/tests/test_nous_client.py

@@ -67,6 +67,33 @@ def test_json_mode_sets_response_format(monkeypatch):
     captured = _install_fake(monkeypatch, lambda *a: _ok_response("{}"))
     chat("p", api_key="k", model="m", base_url=BASE, json_mode=True)
     assert captured["json"]["response_format"] == {"type": "json_object"}
+    assert captured["json"]["max_tokens"] >= 8192
+    roles = [m["role"] for m in captured["json"]["messages"]]
+    assert "system" in roles
+    assert "user" in roles
+
+
+def test_retries_once_when_finish_reason_is_length(monkeypatch):
+    calls = {"n": 0}
+
+    def post_impl(*a):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            resp = MagicMock()
+            resp.status_code = 200
+            resp.json.return_value = {
+                "choices": [{
+                    "message": {"content": "{\"summary_en\": \"cut off"},
+                    "finish_reason": "length",
+                }]
+            }
+            return resp
+        return _ok_response('{"summary_en": "complete brief."}')
+
+    _install_fake(monkeypatch, post_impl)
+    out = chat("p", api_key="k", model="m", base_url=BASE, json_mode=True)
+    assert calls["n"] == 2
+    assert "complete brief" in out
 
 
 def test_401_returns_empty_string(monkeypatch):

news/summerizer/tests/test_prompts.py

@@ -22,6 +22,14 @@ def test_summary_prompt_focuses_on_breaking_news_not_futures():
     assert "commodity" in p or "market" in p
 
 
+def test_summary_prompt_covers_critical_then_incidents():
+    p = SUMMARY_PROMPT_DEFAULT.lower()
+    assert "critical" in p
+    assert "crime" in p
+    assert "incident" in p
+    assert "complete" in p or "truncat" in p or "unfinished" in p or "mid-sentence" in p
+
+
 def test_summary_prompt_does_not_bail_out_with_canned_empty_brief():
     p = SUMMARY_PROMPT_DEFAULT
     assert "AND STOP" not in p

tests/test_api_news.py

@@ -230,6 +230,21 @@ def test_api_news_ticker_returns_only_flagged(clean_news):
     assert item["url"] == "https://example.com/ticker"
 
 
+@requires_db
+def test_api_news_ticker_falls_back_to_lesser_when_nothing_flagged(clean_news):
+    sid = _seed_summary("quiet brief", "2026-08-27T18:05:00+00:00", "Hermes-4.3-36B")
+    _seed_news_item(
+        sid, "ticker", "Shop theft downtown", "low",
+        location_name="Raleigh", url="https://example.com/theft",
+    )
+    resp = _get("/api/news/ticker")
+    assert resp.status_code == 200
+    body = resp.json()
+    assert len(body) == 1
+    assert body[0]["headline"] == "Shop theft downtown"
+    assert body[0]["importance"] == "low"
+
+
 @requires_db
 def test_api_news_map_returns_only_flagged_with_coords(clean_news):
     _seed_flagged_items()

tests/test_frontend_reliability.py

@@ -72,6 +72,14 @@ def test_chokepoint_skips_aisstream_subscribe_outside_conus():
     assert "minlat,minlon,maxlat,maxlon" in HTML.split("function chokepointLeafletBounds")[1][:400]
 
 
+def test_news_ticker_polls_more_often_than_summarizer_cycle():
+    assert "NEWS_REFRESH_MS" in HTML
+    # Summarizer is 15 min; ticker should refresh on a shorter cadence so
+    # lesser-news fills show up without waiting for the next brief.
+    line = [ln for ln in HTML.splitlines() if "NEWS_REFRESH_MS" in ln][0]
+    assert "900000" not in line
+
+
 def test_phone_chokepoints_use_select_not_buttons():
     mobile = HTML.split("@media (max-width: 820px)")[1].split("@media (prefers-reduced-motion")[0]
     assert "#chokepoint-select { display: block; }" in mobile

tests/test_geofence_frontend.py

@@ -0,0 +1,26 @@
+"""Geofence layer panel: draw + delete (DELETE /api/geofences/{id})."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+HTML = (ROOT / "app/static/index.html").read_text()
+
+
+def test_geofence_panel_has_list_and_delete_hook():
+    assert 'id="gf-draw"' in HTML
+    assert 'id="gf-list"' in HTML
+    assert "function deleteGeofence" in HTML
+    assert "method: 'DELETE'" in HTML or 'method: "DELETE"' in HTML
+    assert "/api/geofences/" in HTML
+
+
+def test_load_geofences_renders_delete_controls():
+    js = HTML.split("async function loadGeofences", 1)[1].split(
+        "async function loadFireAircraftHits", 1
+    )[0]
+    assert "gf-list" in js
+    assert "deleteGeofence" in js
+    assert "onEachFeature" in js
+    assert "bindPopup" in js

tests/test_live_layers.py

@@ -7,6 +7,7 @@ from live_layers import (
     filter_points_bbox,
     parse_bbox,
     quantize_bbox,
+    pick_sentinel_feature,
     rainviewer_tile_url,
     sign_cog_url,
     sentinel1_tile_url,
@@ -734,8 +735,9 @@ def test_fetch_sentinel1_vv_signed_tile_url(monkeypatch):
     post_url, post_json = calls[0][1], calls[0][2]
     assert post_url.endswith("/api/stac/v1/search")
     assert post_json["collections"] == ["sentinel-1-grd"]
-    assert post_json["limit"] == 1
+    assert post_json["limit"] >= 1
     assert post_json["sortby"][0]["direction"] == "desc"
+    assert "bbox" in out
 
 
 def test_fetch_sentinel1_uses_hh_when_vv_missing(monkeypatch):
@@ -783,3 +785,22 @@ def test_fetch_sentinel1_none_when_no_vv_or_hh(monkeypatch):
     _cache.clear()
 
     assert asyncio.run(fetch_sentinel1("-80,35,-79,36")) is None
+
+
+def test_pick_sentinel_feature_prefers_scene_covering_center():
+    features = [
+        {"id": "far", "bbox": [10.0, 10.0, 12.0, 12.0]},
+        {"id": "cover", "bbox": [-80.5, 34.5, -78.5, 36.5]},
+        {"id": "also-far", "bbox": [-10.0, 0.0, -8.0, 2.0]},
+    ]
+    picked = pick_sentinel_feature(features, -79.5, 35.5)
+    assert picked["id"] == "cover"
+
+
+def test_pick_sentinel_feature_falls_back_to_first_when_none_cover():
+    features = [
+        {"id": "a", "bbox": [10.0, 10.0, 12.0, 12.0]},
+        {"id": "b", "bbox": [20.0, 20.0, 22.0, 22.0]},
+    ]
+    assert pick_sentinel_feature(features, -79.5, 35.5)["id"] == "a"
+    assert pick_sentinel_feature([], -79.5, 35.5) is None

tests/test_sentinel1_frontend.py

@@ -39,3 +39,11 @@ def test_sentinel1_not_fetched_on_init_unless_on():
     assert "loadSentinel1()" not in init
     refresh = HTML.split("function refreshLiveOverlays", 1)[1].split("function addExtraAttrib", 1)[0]
     assert "if (sentinelOn) loadSentinel1();" in refresh
+
+
+def test_sentinel1_reuses_covering_scene_and_clips_tiles():
+    js = HTML.split("async function loadSentinel1", 1)[1].split("function loadThermal", 1)[0]
+    assert "sentinelStillCovers" in HTML
+    assert "itemId" in js
+    assert "L.latLngBounds" in js
+    assert "sentinelBounds" in HTML

tests/test_vesselapi.py

@@ -234,6 +234,8 @@ def _patch_side_effects(monkeypatch):
 
     monkeypatch.setattr("tracks.record_position", _noop)
     monkeypatch.setattr("geofence.record_and_notify", _noop)
+    monkeypatch.setattr(vesselapi, "persist_vessel_snapshot", _noop)
+    monkeypatch.setattr(vesselapi, "purge_old_vessels", _noop)
 
 
 def test_poll_once_lands_markers_in_vessel_last_known(monkeypatch):

tests/test_vessels_snapshot.py

@@ -0,0 +1,36 @@
+"""VesselAPI daily snapshot store — as-of DVR + UTC-day purge (no DB)."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from vesselapi import pick_poll_at, snapshot_as_of, utc_day_start
+
+
+def test_utc_day_start_floors_to_midnight_utc():
+    now = datetime(2026, 8, 29, 15, 30, 12, tzinfo=timezone.utc)
+    assert utc_day_start(now) == datetime(2026, 8, 29, 0, 0, tzinfo=timezone.utc)
+
+
+def test_pick_poll_at_returns_latest_snapshot_at_or_before_as_of():
+    t1 = datetime(2026, 8, 29, 0, 0, tzinfo=timezone.utc)
+    t2 = datetime(2026, 8, 29, 4, 48, tzinfo=timezone.utc)
+    t3 = datetime(2026, 8, 29, 9, 36, tzinfo=timezone.utc)
+    as_of = datetime(2026, 8, 29, 6, 0, tzinfo=timezone.utc)
+    assert pick_poll_at([t1, t2, t3], as_of) == t2
+    assert pick_poll_at([t1, t2, t3], t1) == t1
+    assert pick_poll_at([t1, t2, t3], datetime(2026, 8, 28, 23, tzinfo=timezone.utc)) is None
+
+
+def test_snapshot_as_of_returns_the_matching_poll_only():
+    t1 = datetime(2026, 8, 29, 0, 0, tzinfo=timezone.utc)
+    t2 = datetime(2026, 8, 29, 4, 48, tzinfo=timezone.utc)
+    rows = [
+        {"id": "1", "poll_at": t1, "lat": 26.5, "lon": 56.0},
+        {"id": "2", "poll_at": t1, "lat": 26.6, "lon": 56.1},
+        {"id": "1", "poll_at": t2, "lat": 26.7, "lon": 56.2},
+    ]
+    out = snapshot_as_of(rows, datetime(2026, 8, 29, 6, 0, tzinfo=timezone.utc))
+    assert {r["id"] for r in out} == {"1"}
+    assert out[0]["lat"] == 26.7
+    assert all(r["poll_at"] == t2 for r in out)