feat(vessels): present AISStream + VesselAPI as two independent providers

- keystore KEY_REGISTRY copy splits the two keys (open/shared US-coast WS vs
  commercial Hormuz 5x/day REST); VesselAPI no longer reads as an AIS backup.
- vesselapi.py docstring drops 'fallback'; states both are first-class.
- /api/vessels docstring: last-known is the union (extra.src aisstream vs
  vesselapi).
- tests: VesselAPI idles/polls independently of the AISStream key.

app/keystore.py

@@ -68,12 +68,12 @@ KEY_REGISTRY: dict[str, dict] = {
         "example": "123456789:AA… (bot token from @BotFather)",
     },
     "AISSTREAM_API_KEY": {
-        "description": "AISStream WebSocket key — live vessel positions (server-side only).",
+        "description": "AISStream (open/shared) — live US-coast AIS. Server-side WebSocket only.",
         "pattern": r"^.{8,}$",
         "example": "key from https://aisstream.io/account (GitHub login)",
     },
     "VESSELAPI_API_KEY": {
-        "description": "VesselAPI REST AIS (150 calls/mo free; server-side poller).",
+        "description": "VesselAPI (commercial) — Strait of Hormuz AIS, 5×/day cache. Paste the Bearer token from dashboard.vesselapi.com. Not a US-coast feed.",
         "pattern": r"^.{8,}$",
         "example": "Bearer token from https://dashboard.vesselapi.com/",
     },

app/main.py

@@ -1472,10 +1472,11 @@ async def list_vessels(
     limit: int = Query(2000, ge=1, le=5000),
     timestamp: str | None = Query(None, description="ISO time — DVR 1-min tracks instead of live"),
 ):
-    """AIS last-known from the AISStream worker and/or the VesselAPI poller.
+    """AIS last-known — union of two independent providers.
 
-    Empty without either key / until the first successful poll. VesselAPI
-    positions upsert into the same store (extra.src = "vesselapi").
+    AISStream (extra.src="aisstream", live US-coast WebSocket) and VesselAPI
+    (extra.src="vesselapi", Strait of Hormuz 5×/day poll) both upsert into the
+    same store. Empty without either key / until the first successful poll.
     """
     if bbox:
         _parse_bbox_query(bbox)

app/vesselapi.py

@@ -1,8 +1,10 @@
-"""VesselAPI REST poller — quota-capped AIS fallback (free tier 150 calls/mo).
+"""VesselAPI REST poller — quota-capped AIS for the Middle East (free tier 150 calls/mo).
 
-AISStream (WebSocket) keeps US coasts live; VesselAPI fills the Middle East
-blind spot (Strait of Hormuz default box). This worker polls the REST
-``GET /v1/location/vessels/bounding-box`` endpoint at most
+VesselAPI and AISStream are two independent, first-class vessel providers —
+not a primary/fallback pair. AISStream (WebSocket) owns live US-coast AIS;
+VesselAPI (REST) covers the Strait of Hormuz (default box) where AISStream
+has no coverage. Missing one key never disables the other. This worker polls
+the REST ``GET /v1/location/vessels/bounding-box`` endpoint at most
 ``VESSELAPI_MAX_CALLS_PER_DAY`` (default 5) *successful 2xx* calls per UTC day
 and upserts the results into the shared ``vessel_last_known`` store.
 

tests/test_vesselapi.py

@@ -296,3 +296,56 @@ def _make_get_client(client):
         return client
 
     return _get_client
+
+
+# ── Independent-provider idle behaviour ───────────────────────────────────
+# VesselAPI must never be gated on AISStream (or vice versa): a missing key on
+# one provider leaves the other running. ``_StopLoop`` is a BaseException so
+# the worker's ``except Exception`` handler can't swallow it — the first
+# ``asyncio.sleep`` aborts the loop after exactly one decision.
+
+class _StopLoop(BaseException):
+    pass
+
+
+async def _stop_sleep(*_a, **_k):
+    raise _StopLoop()
+
+
+async def _resolve(value: str):
+    return value
+
+
+def test_worker_idles_without_vesselapi_key_even_if_aisstream_set(monkeypatch):
+    # AISStream key present, VesselAPI key absent -> no poll, still idles.
+    monkeypatch.setenv("AISSTREAM_API_KEY", "unused-aisstream-key")
+    monkeypatch.setenv("VESSELAPI_API_KEY", "")
+    monkeypatch.setattr(vesselapi, "_resolve_key", lambda: _resolve(""))
+    monkeypatch.setattr(vesselapi.asyncio, "sleep", _stop_sleep)
+    poll_calls: list = []
+    async def _spy_poll(store, boxes, key):
+        poll_calls.append(key)
+        return True
+    monkeypatch.setattr(vesselapi, "poll_once", _spy_poll)
+
+    with pytest.raises(_StopLoop):
+        asyncio.run(vesselapi.run_vesselapi_worker(FakeStore()))
+
+    assert poll_calls == []
+
+
+def test_worker_polls_with_vesselapi_key_even_if_aisstream_unset(monkeypatch):
+    # AISStream key absent, VesselAPI key present -> still polls exactly once.
+    monkeypatch.setenv("AISSTREAM_API_KEY", "")
+    monkeypatch.setattr(vesselapi, "_resolve_key", lambda: _resolve("vesselapi-key"))
+    monkeypatch.setattr(vesselapi.asyncio, "sleep", _stop_sleep)
+    poll_calls: list = []
+    async def _spy_poll(store, boxes, key):
+        poll_calls.append(key)
+        return True
+    monkeypatch.setattr(vesselapi, "poll_once", _spy_poll)
+
+    with pytest.raises(_StopLoop):
+        asyncio.run(vesselapi.run_vesselapi_worker(FakeStore()))
+
+    assert poll_calls == ["vesselapi-key"]