feat: add GenericHttpAdapter (kind=generic_http, v0.15.0 PR2) (#121)

One config-driven adapter class that operators can instantiate many times
via distinct config.adapters rows (unique name, kind="generic_http").
Ships class + Pydantic settings schema + unit tests only; no GUI create
route, no migration, no seeded row (PR3). Class is dormant until an
operator row exists.

Key design decisions:
- self.name = config.name in __init__ scopes dedup to each instance
- domain validated against STREAMS registry at import time
- subject_for derives region from event.data["_enriched"]["geocoder"]
  identically to usgs_quake (no static subject_region setting)
- _dig() helper supports dotted paths into nested dicts/lists

Also adds: GUI partials (_event_summaries, _event_rows), CONSUMER-
INTEGRATION.md §6 subsection, and test_events_feed_frontend.py sample
entry — all required by the existing adapter-coverage consistency tests.

Co-authored-by: Ubuntu <zvx@cortex.echo6.co>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

docs/CONSUMER-INTEGRATION.md

@@ -2023,6 +2023,55 @@ at parameter `00060`, gage height (ft) at `00065`, water temperature (°C) at
 \
 ---
 
+### generic_http — Generic HTTP Source (v0.15.0)
+
+- **Source:** Any public REST / GeoJSON endpoint; configured per-instance
+  via `config.adapters.settings.url`.  v1 supports public (unauthenticated)
+  feeds only; auth is a planned follow-up.
+- **Stream:** Determined by the `domain` setting, which must match an
+  existing Central stream domain (e.g. `wx`, `fire`, `quake`).  Subject
+  filter: `central.<domain>.>`.
+- **Subject:** `central.<domain>.<region>` — `<region>` is derived from
+  `event.data["_enriched"]["geocoder"]` by the Photon enrichment pipeline
+  (identical convention to `usgs_quake`).  Returns `unknown` when enrichment
+  has not yet run or found no geo signal.
+- **Category:** `<domain>.<category_suffix>` (default suffix: `alert`, so
+  e.g. `wx.alert`).
+- **Dedup key:** the value resolved by `id_path` within each source item
+  (required field; coerced to string).
+- **Geo:** GeoJSON geometry from `geometry_path` (default `"geometry"`)
+  when it resolves to a dict; `geo.centroid` is also set for Point
+  geometries.  Falls back to `lat_path` / `lon_path` for non-GeoJSON
+  responses.
+- **Event.data fields:** operator-configured.  `title_path` → `data["title"]`
+  when set; additional fields via `field_mappings` (`source_path → dest_key`
+  pairs).  No guaranteed fixed schema — varies per operator configuration.
+- **Settings summary:**
+
+  | Setting | Required | Default | Notes |
+  |---|---|---|---|
+  | `url` | yes | — | Polled endpoint (GET, no auth in v1) |
+  | `domain` | yes | — | Must be an existing stream domain |
+  | `id_path` | yes | — | Dotted path to stable unique item id |
+  | `items_path` | no | `features` | Dotted path to the items array |
+  | `format` | no | `geojson` | `geojson` or `json` (informational) |
+  | `time_path` | no | `null` | ISO-8601 timestamp path; uses poll time if absent |
+  | `title_path` | no | `null` | Path → `data["title"]` |
+  | `geometry_path` | no | `geometry` | GeoJSON geometry path |
+  | `lat_path` / `lon_path` | no | `null` | Numeric lat/lon paths (non-GeoJSON fallback) |
+  | `severity_path` | no | `null` | Int severity (0–4) path |
+  | `category_suffix` | no | `alert` | Appended to domain for Event.category |
+  | `field_mappings` | no | `[]` | List of `{source_path, dest_key}` |
+
+- **Cadence:** 300s (5 min) default.
+- **Multiple instances:** one `generic_http` class supports many
+  `config.adapters` rows (each with a distinct instance `name`); dedup is
+  scoped per-instance.  The class is dormant until an operator row is
+  created — no built-in seeded instance.
+
+\
+---
+
 ## 7. Fall-off / removal semantics
 
 Central adapters fall into three buckets for handling upstream events that

src/central/adapters/generic_http.py

@@ -0,0 +1,442 @@
+"""Generic HTTP / GeoJSON adapter — operator-instantiable, no Python required.
+
+One class (kind="generic_http") supports many config.adapters rows.  Each
+row carries a distinct ``name`` (instance identity) and a ``settings`` dict
+that drives a separate poll loop.  PR3 ships the GUI create route; this PR
+ships only the class.
+
+v1: public feeds only; auth is a follow-up.
+"""
+
+import logging
+import sqlite3
+from collections.abc import AsyncIterator
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Literal
+
+import aiohttp
+from pydantic import BaseModel, field_validator
+from tenacity import (
+    retry,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_exponential_jitter,
+)
+
+from central.adapter import SourceAdapter
+from central.adapters._subject_helpers import subject_for_region
+from central.config_models import AdapterConfig
+from central.config_store import ConfigStore
+from central.models import Event, Geo
+from central.streams import STREAMS
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Derive the valid domain set from the stream registry at import time.
+# Subject filters are always "central.<domain>.>" so we split on "." and
+# take index 1.
+# ---------------------------------------------------------------------------
+_VALID_DOMAINS: frozenset[str] = frozenset(
+    s.subject_filter.split(".")[1] for s in STREAMS
+)
+
+_DEDUP_DDL = (
+    "CREATE TABLE IF NOT EXISTS published_ids ("
+    "adapter TEXT NOT NULL, event_id TEXT NOT NULL, "
+    "first_seen TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, "
+    "last_seen TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, "
+    "PRIMARY KEY (adapter, event_id))"
+)
+_DEDUP_IDX = (
+    "CREATE INDEX IF NOT EXISTS published_ids_last_seen "
+    "ON published_ids (last_seen)"
+)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _dig(obj: Any, path: str) -> Any:
+    """Walk a nested dict/list by a dotted path string.
+
+    Each segment is tried as a dict key first; if the current node is a
+    list/tuple the segment is coerced to an integer index.  Returns None on
+    any miss (missing key, out-of-range index, wrong node type).
+
+    Examples::
+
+        _dig({"a": {"b": 1}}, "a.b")       # 1
+        _dig({"a": [10, 20]}, "a.1")        # 20
+        _dig({"a": 1}, "a.b")              # None
+        _dig(None, "anything")              # None
+    """
+    parts = path.split(".")
+    cur = obj
+    for part in parts:
+        if cur is None:
+            return None
+        if isinstance(cur, dict):
+            cur = cur.get(part)
+        elif isinstance(cur, (list, tuple)):
+            try:
+                cur = cur[int(part)]
+            except (ValueError, IndexError):
+                return None
+        else:
+            return None
+    return cur
+
+
+# ---------------------------------------------------------------------------
+# Settings schema
+# ---------------------------------------------------------------------------
+
+class FieldMapping(BaseModel):
+    """Map one path inside a source item to a key in Event.data.
+
+    Mirrors TomTomFlowSettings / TileCoord so the GUI model_list widget
+    renders it identically.
+    """
+    source_path: str  # dotted path into the source item
+    dest_key: str     # key written into Event.data
+
+
+class GenericHttpSettings(BaseModel):
+    """Settings schema for GenericHttpAdapter instances."""
+
+    url: str
+    """Endpoint to poll (GET, no auth in v1)."""
+
+    format: Literal["geojson", "json"] = "geojson"
+    """Response format hint (currently informational; both paths use the same
+    JSON fetch; the geometry_path / lat_path controls shape extraction)."""
+
+    items_path: str = "features"
+    """Dotted path to the array of items inside the response object."""
+
+    domain: str
+    """Event domain; must match an existing NATS stream (e.g. ``wx``, ``fire``,
+    ``quake``).  Determines which JetStream stream the event lands in."""
+
+    @field_validator("domain")
+    @classmethod
+    def _validate_domain(cls, v: str) -> str:
+        if v not in _VALID_DOMAINS:
+            valid = ", ".join(sorted(_VALID_DOMAINS))
+            raise ValueError(
+                f"unknown domain '{v}'; must be one of: {valid}"
+            )
+        return v
+
+    category_suffix: str = "alert"
+    """Appended to domain to form Event.category: ``{domain}.{category_suffix}``."""
+
+    id_path: str
+    """Dotted path to a stable, unique identifier within each item (required
+    for deduplication)."""
+
+    time_path: str | None = None
+    """Dotted path to an ISO-8601 timestamp string.  When None, poll time
+    (UTC now) is used."""
+
+    title_path: str | None = None
+    """Dotted path to a human-readable title → written to ``data['title']``."""
+
+    geometry_path: str = "geometry"
+    """(GeoJSON) dotted path to a GeoJSON geometry object within each item.
+    Ignored when the resolved value is not a dict."""
+
+    lat_path: str | None = None
+    """(JSON) dotted path to a numeric latitude.  Used only when
+    geometry_path does not resolve to a geometry dict."""
+
+    lon_path: str | None = None
+    """(JSON) dotted path to a numeric longitude.  See lat_path."""
+
+    severity_path: str | None = None
+    """Dotted path to an integer severity (0-4).  When None, severity is
+    omitted from the event."""
+
+    field_mappings: list[FieldMapping] = []
+    """Extra source-path → dest-key pairs written into Event.data."""
+
+
+# ---------------------------------------------------------------------------
+# Adapter class
+# ---------------------------------------------------------------------------
+
+class GenericHttpAdapter(SourceAdapter):
+    """Config-driven REST/GeoJSON source adapter.
+
+    A single class that operators can instantiate many times via distinct
+    ``config.adapters`` rows (each row has a unique ``name`` / instance
+    identity and ``kind="generic_http"``).  No Python required to add a
+    new source.
+    """
+
+    # Class-level identity (kind) — used by discover_adapters() as the
+    # registry key.  Each *instance* overrides self.name = config.name in
+    # __init__ so dedup and logs are scoped to the instance.
+    name = "generic_http"
+    display_name = "Generic HTTP Source"
+    description = (
+        "Config-driven adapter that polls any public REST or GeoJSON endpoint "
+        "and maps response fields to Central events.  Create one row per "
+        "source; no Python required."
+    )
+    settings_schema = GenericHttpSettings
+    default_cadence_s = 300
+    data_class = "event"
+    wizard_order = None   # not in the setup wizard; created via operator GUI
+    enrichment_locations = []
+    bypass_bbox_filter = False
+
+    def __init__(
+        self,
+        config: AdapterConfig,
+        config_store: ConfigStore,   # unused; accepted for signature uniformity
+        cursor_db_path: Path,
+    ) -> None:
+        # Override class-level name with the *instance* name so that the
+        # inherited dedup helpers (is_published / mark_published / sweep_old_ids)
+        # scope their SQL queries to this instance, not the generic_http class.
+        self.name = config.name
+        self._cursor_db_path = cursor_db_path
+        self._session: aiohttp.ClientSession | None = None
+        self._db: sqlite3.Connection | None = None
+        self._settings = GenericHttpSettings(**config.settings)
+
+    # ------------------------------------------------------------------
+    # Lifecycle
+    # ------------------------------------------------------------------
+
+    async def startup(self) -> None:
+        """Initialize HTTP session and dedup tracker."""
+        self._session = aiohttp.ClientSession(
+            headers={"User-Agent": "Central/1.0 (generic_http adapter)"},
+            timeout=aiohttp.ClientTimeout(total=30),
+        )
+        self._db = sqlite3.connect(str(self._cursor_db_path))
+        self._db.execute(_DEDUP_DDL)
+        self._db.execute(_DEDUP_IDX)
+        self._db.commit()
+        self.sweep_old_ids()
+        logger.info(
+            "generic_http adapter started",
+            extra={"adapter": self.name, "url": self._settings.url},
+        )
+
+    async def shutdown(self) -> None:
+        """Close HTTP session and database."""
+        if self._session:
+            await self._session.close()
+            self._session = None
+        if self._db:
+            self._db.close()
+            self._db = None
+        logger.info("generic_http adapter shut down", extra={"adapter": self.name})
+
+    # ------------------------------------------------------------------
+    # Configuration hot-reload
+    # ------------------------------------------------------------------
+
+    async def apply_config(self, new_config: AdapterConfig) -> None:
+        """Re-parse settings in place (no reconstruction needed)."""
+        self._settings = GenericHttpSettings(**new_config.settings)
+        logger.info(
+            "generic_http config applied",
+            extra={"adapter": self.name, "url": self._settings.url},
+        )
+
+    # ------------------------------------------------------------------
+    # Fetch
+    # ------------------------------------------------------------------
+
+    @retry(
+        stop=stop_after_attempt(3),
+        wait=wait_exponential_jitter(initial=1, max=15),
+        retry=retry_if_exception_type((aiohttp.ClientError,)),
+        reraise=True,
+    )
+    async def _fetch(self, url: str) -> dict[str, Any]:
+        """GET ``url`` and return parsed JSON.  Retried up to 3 times."""
+        if not self._session:
+            raise RuntimeError("Session not initialized")
+        async with self._session.get(url) as resp:
+            resp.raise_for_status()
+            return await resp.json(content_type=None)
+
+    # ------------------------------------------------------------------
+    # Poll
+    # ------------------------------------------------------------------
+
+    async def poll(self) -> AsyncIterator[Event]:
+        """Fetch the configured URL and yield new Events."""
+        self.sweep_old_ids()
+
+        s = self._settings
+        try:
+            raw = await self._fetch(s.url)
+        except Exception as exc:
+            logger.error(
+                "generic_http fetch failed",
+                extra={"adapter": self.name, "url": s.url, "error": str(exc)},
+            )
+            raise
+
+        items = _dig(raw, s.items_path)
+        if not isinstance(items, list):
+            logger.warning(
+                "generic_http items_path did not resolve to a list",
+                extra={
+                    "adapter": self.name,
+                    "items_path": s.items_path,
+                    "got": type(items).__name__,
+                },
+            )
+            return
+
+        new_count = 0
+        for item in items:
+            event = self._item_to_event(item)
+            if event is None:
+                continue
+            if self.is_published(event.id):
+                continue
+            yield event
+            self.mark_published(event.id)
+            new_count += 1
+
+        logger.info(
+            "generic_http poll completed",
+            extra={"adapter": self.name, "count": new_count},
+        )
+
+    # ------------------------------------------------------------------
+    # Item → Event
+    # ------------------------------------------------------------------
+
+    def _item_to_event(self, item: Any) -> Event | None:
+        """Convert a single source item to an Event, or None to skip."""
+        s = self._settings
+
+        # --- id (required) ---
+        raw_id = _dig(item, s.id_path)
+        if raw_id is None:
+            logger.warning(
+                "generic_http item missing id",
+                extra={"adapter": self.name, "id_path": s.id_path},
+            )
+            return None
+        event_id = str(raw_id)
+
+        # --- category ---
+        category = f"{s.domain}.{s.category_suffix}"
+
+        # --- time ---
+        if s.time_path:
+            raw_time = _dig(item, s.time_path)
+            if raw_time is not None:
+                try:
+                    # Python 3.11+ fromisoformat handles "Z"; older needs replace.
+                    event_time = datetime.fromisoformat(
+                        str(raw_time).replace("Z", "+00:00")
+                    )
+                    if event_time.tzinfo is None:
+                        event_time = event_time.replace(tzinfo=timezone.utc)
+                except (ValueError, TypeError):
+                    logger.warning(
+                        "generic_http could not parse time; using now",
+                        extra={
+                            "adapter": self.name,
+                            "time_path": s.time_path,
+                            "raw": raw_time,
+                        },
+                    )
+                    event_time = datetime.now(timezone.utc)
+            else:
+                event_time = datetime.now(timezone.utc)
+        else:
+            event_time = datetime.now(timezone.utc)
+
+        # --- geo ---
+        geo_kwargs: dict[str, Any] = {}
+
+        # Try GeoJSON geometry first.
+        if s.geometry_path:
+            geom = _dig(item, s.geometry_path)
+            if isinstance(geom, dict):
+                geo_kwargs["geometry"] = geom
+                # Also set centroid when geometry is a simple Point.
+                if geom.get("type") == "Point":
+                    coords = geom.get("coordinates") or []
+                    if len(coords) >= 2:
+                        try:
+                            geo_kwargs["centroid"] = (
+                                float(coords[0]),
+                                float(coords[1]),
+                            )
+                        except (TypeError, ValueError):
+                            pass
+
+        # Fall back to explicit lat/lon paths.
+        if "geometry" not in geo_kwargs and s.lat_path and s.lon_path:
+            raw_lat = _dig(item, s.lat_path)
+            raw_lon = _dig(item, s.lon_path)
+            if raw_lat is not None and raw_lon is not None:
+                try:
+                    geo_kwargs["centroid"] = (float(raw_lon), float(raw_lat))
+                except (TypeError, ValueError):
+                    pass
+
+        geo = Geo(**geo_kwargs)
+
+        # --- data ---
+        data: dict[str, Any] = {}
+        if s.title_path:
+            title = _dig(item, s.title_path)
+            if title is not None:
+                data["title"] = title
+
+        for fm in s.field_mappings:
+            data[fm.dest_key] = _dig(item, fm.source_path)
+
+        # --- severity ---
+        severity: int | None = None
+        if s.severity_path:
+            raw_sev = _dig(item, s.severity_path)
+            if raw_sev is not None:
+                try:
+                    severity = int(raw_sev)
+                except (TypeError, ValueError):
+                    pass
+
+        return Event(
+            id=event_id,
+            adapter=self.name,
+            category=category,
+            time=event_time,
+            geo=geo,
+            data=data,
+            severity=severity,
+        )
+
+    # ------------------------------------------------------------------
+    # Subject routing
+    # ------------------------------------------------------------------
+
+    def subject_for(self, event: Event) -> str:
+        """Return the NATS subject for a generic-http event.
+
+        Pattern: ``central.<domain>.<region>``
+
+        Region is derived from ``event.data["_enriched"]["geocoder"]``
+        identically to usgs_quake — the Photon enrichment pipeline populates
+        that key at publish time.  Returns "unknown" when enrichment hasn't
+        run yet or found no geo signal.
+        """
+        region = subject_for_region(event.data)
+        return f"central.{self._settings.domain}.{region}"

src/central/gui/templates/_event_rows/generic_http.html

@@ -0,0 +1,3 @@
+{# Generic HTTP source. Fields from payload->data->data. #}
+{% set d = (event.data.get('data') or {}).get('data') or {} %}
+{% if d.get('title') is not none %}<dt>Title</dt><dd>{{ d.title }}</dd>{% endif %}

src/central/gui/templates/_event_summaries/generic_http.html

@@ -0,0 +1,2 @@
+{% set d = (event.data.get('data') or {}).get('data') or {} %}
+{%- if d.get('title') %}{{ d.title }}{% endif -%}

tests/test_events_feed_frontend.py

@@ -1189,6 +1189,7 @@ _SAMPLE_INNER = {
         "current_lat_deg": -17.1487,
         "current_alt_km": 417.4,
     },
+    "generic_http": {"title": "Generic source alert"},
 }
 
 # Exact expected subjects for the deterministic adapters. swpc_alerts is omitted

tests/test_generic_http.py

@@ -0,0 +1,666 @@
+"""Unit tests for GenericHttpAdapter.
+
+Pure unit tests: no database mocking required (we use a real sqlite tmp file
+for dedup tests, same as test_usgs_quake.py), and HTTP is patched.
+"""
+
+import tempfile
+from datetime import datetime, timezone
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from pydantic import ValidationError
+
+from central.adapters.generic_http import (
+    FieldMapping,
+    GenericHttpAdapter,
+    GenericHttpSettings,
+    _dig,
+)
+from central.config_models import AdapterConfig
+from central.models import Event, Geo
+from central.streams import STREAMS
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _valid_domain() -> str:
+    """Return a known valid domain (first event-bearing stream)."""
+    for s in STREAMS:
+        domain = s.subject_filter.split(".")[1]
+        if domain != "meta":
+            return domain
+    raise RuntimeError("No non-meta stream found in STREAMS")
+
+
+def make_config(
+    name: str = "test_generic",
+    domain: str | None = None,
+    extra_settings: dict | None = None,
+) -> AdapterConfig:
+    domain = domain or _valid_domain()
+    settings: dict = {
+        "url": "https://example.com/feed.json",
+        "domain": domain,
+        "id_path": "id",
+        "items_path": "features",
+        "geometry_path": "geometry",
+    }
+    if extra_settings:
+        settings.update(extra_settings)
+    return AdapterConfig(
+        name=name,
+        kind="generic_http",
+        enabled=True,
+        cadence_s=300,
+        settings=settings,
+        updated_at=datetime.now(timezone.utc),
+    )
+
+
+@pytest.fixture
+def temp_db_path():
+    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+        yield Path(f.name)
+
+
+@pytest.fixture
+def mock_config_store():
+    return MagicMock()
+
+
+# ---------------------------------------------------------------------------
+# _dig
+# ---------------------------------------------------------------------------
+
+class TestDig:
+    def test_simple_key(self):
+        assert _dig({"a": 1}, "a") == 1
+
+    def test_nested_keys(self):
+        assert _dig({"a": {"b": {"c": 42}}}, "a.b.c") == 42
+
+    def test_list_index(self):
+        assert _dig({"a": [10, 20, 30]}, "a.1") == 20
+
+    def test_list_index_zero(self):
+        assert _dig({"items": [{"x": 99}]}, "items.0.x") == 99
+
+    def test_missing_key_returns_none(self):
+        assert _dig({"a": 1}, "b") is None
+
+    def test_missing_nested_returns_none(self):
+        assert _dig({"a": {"b": 1}}, "a.c") is None
+
+    def test_out_of_range_index_returns_none(self):
+        assert _dig({"a": [1, 2]}, "a.5") is None
+
+    def test_none_root_returns_none(self):
+        assert _dig(None, "a.b") is None
+
+    def test_non_dict_mid_path_returns_none(self):
+        assert _dig({"a": 42}, "a.b") is None
+
+    def test_empty_list(self):
+        assert _dig([], "0") is None
+
+
+# ---------------------------------------------------------------------------
+# Domain validation
+# ---------------------------------------------------------------------------
+
+class TestDomainValidation:
+    def test_unknown_domain_raises(self):
+        with pytest.raises(ValidationError) as exc_info:
+            GenericHttpSettings(
+                url="https://example.com/feed",
+                domain="totally_unknown_domain_xyz",
+                id_path="id",
+            )
+        err_str = str(exc_info.value)
+        assert "unknown domain" in err_str
+        # Should list valid options in the error message
+        assert "wx" in err_str or "fire" in err_str
+
+    def test_known_domain_ok(self):
+        s = GenericHttpSettings(
+            url="https://example.com/feed",
+            domain=_valid_domain(),
+            id_path="id",
+        )
+        assert s.domain == _valid_domain()
+
+    def test_wx_domain_valid(self):
+        s = GenericHttpSettings(
+            url="https://example.com/feed",
+            domain="wx",
+            id_path="id",
+        )
+        assert s.domain == "wx"
+
+    def test_fire_domain_valid(self):
+        s = GenericHttpSettings(
+            url="https://example.com/feed",
+            domain="fire",
+            id_path="id",
+        )
+        assert s.domain == "fire"
+
+
+# ---------------------------------------------------------------------------
+# Category composition
+# ---------------------------------------------------------------------------
+
+class TestCategoryComposition:
+    @pytest.mark.asyncio
+    async def test_category_default_suffix(self, temp_db_path, mock_config_store):
+        config = make_config(domain="wx")
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        geojson = {
+            "features": [
+                {
+                    "id": "item-1",
+                    "geometry": {"type": "Point", "coordinates": [-116.0, 43.0]},
+                }
+            ]
+        }
+        with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mock_fetch:
+            mock_fetch.return_value = geojson
+            events = [e async for e in adapter.poll()]
+
+        assert len(events) == 1
+        assert events[0].category == "wx.alert"
+
+        await adapter.shutdown()
+
+    @pytest.mark.asyncio
+    async def test_category_custom_suffix(self, temp_db_path, mock_config_store):
+        config = make_config(domain="wx", extra_settings={"category_suffix": "warning"})
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        geojson = {
+            "features": [
+                {
+                    "id": "item-wx-warn",
+                    "geometry": {"type": "Point", "coordinates": [-116.0, 43.0]},
+                }
+            ]
+        }
+        with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf:
+            mf.return_value = geojson
+            events = [e async for e in adapter.poll()]
+
+        assert events[0].category == "wx.warning"
+        await adapter.shutdown()
+
+
+# ---------------------------------------------------------------------------
+# GeoJSON path — geometry + centroid
+# ---------------------------------------------------------------------------
+
+GEOJSON_FIXTURE = {
+    "type": "FeatureCollection",
+    "features": [
+        {
+            "id": "feat-001",
+            "type": "Feature",
+            "properties": {
+                "title": "Test Alert",
+                "severity": 2,
+                "updated": "2025-01-15T12:00:00Z",
+            },
+            "geometry": {
+                "type": "Point",
+                "coordinates": [-116.2, 43.7],
+            },
+        },
+        {
+            "id": "feat-002",
+            "type": "Feature",
+            "properties": {
+                "title": "Polygon Alert",
+                "severity": 3,
+                "updated": "2025-01-15T13:00:00Z",
+            },
+            "geometry": {
+                "type": "Polygon",
+                "coordinates": [[[-116, 43], [-115, 43], [-115, 44], [-116, 44], [-116, 43]]],
+            },
+        },
+    ],
+}
+
+
+class TestGeoJsonPath:
+    @pytest.mark.asyncio
+    async def test_point_geometry_and_centroid(self, temp_db_path, mock_config_store):
+        config = make_config(
+            domain="wx",
+            extra_settings={
+                "items_path": "features",
+                "id_path": "id",
+                "geometry_path": "geometry",
+                "title_path": "properties.title",
+                "severity_path": "properties.severity",
+                "field_mappings": [
+                    {"source_path": "properties.title", "dest_key": "title"},
+                ],
+            },
+        )
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf:
+            mf.return_value = GEOJSON_FIXTURE
+            events = [e async for e in adapter.poll()]
+
+        assert len(events) == 2
+
+        point_event = next(e for e in events if e.id == "feat-001")
+        # geo.geometry set
+        assert point_event.geo.geometry == {
+            "type": "Point",
+            "coordinates": [-116.2, 43.7],
+        }
+        # centroid extracted from Point
+        assert point_event.geo.centroid == (-116.2, 43.7)
+
+        await adapter.shutdown()
+
+    @pytest.mark.asyncio
+    async def test_polygon_geometry_no_centroid(self, temp_db_path, mock_config_store):
+        config = make_config(domain="wx")
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf:
+            mf.return_value = GEOJSON_FIXTURE
+            events = [e async for e in adapter.poll()]
+
+        poly_event = next(e for e in events if e.id == "feat-002")
+        # geometry is set
+        assert poly_event.geo.geometry is not None
+        assert poly_event.geo.geometry["type"] == "Polygon"
+        # centroid is NOT set (non-Point geometry)
+        assert poly_event.geo.centroid is None
+
+        await adapter.shutdown()
+
+    @pytest.mark.asyncio
+    async def test_field_mappings_populate_data(self, temp_db_path, mock_config_store):
+        config = make_config(
+            domain="wx",
+            extra_settings={
+                "id_path": "id",
+                "title_path": "properties.title",
+                "field_mappings": [
+                    {"source_path": "properties.severity", "dest_key": "level"},
+                    {"source_path": "properties.updated", "dest_key": "updated_at"},
+                ],
+            },
+        )
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf:
+            mf.return_value = GEOJSON_FIXTURE
+            events = [e async for e in adapter.poll()]
+
+        e = next(ev for ev in events if ev.id == "feat-001")
+        assert e.data["title"] == "Test Alert"
+        assert e.data["level"] == 2
+        assert e.data["updated_at"] == "2025-01-15T12:00:00Z"
+
+        await adapter.shutdown()
+
+
+# ---------------------------------------------------------------------------
+# JSON (non-GeoJSON) path — lat_path / lon_path
+# ---------------------------------------------------------------------------
+
+JSON_FIXTURE = {
+    "alerts": [
+        {"uid": "a1", "name": "Alert One", "lat": 43.5, "lon": -116.1, "sev": 1},
+        {"uid": "a2", "name": "Alert Two", "lat": 44.0, "lon": -115.5, "sev": 3},
+    ]
+}
+
+
+class TestJsonLatLonPath:
+    @pytest.mark.asyncio
+    async def test_lat_lon_centroid(self, temp_db_path, mock_config_store):
+        config = make_config(
+            domain="fire",
+            extra_settings={
+                "items_path": "alerts",
+                "id_path": "uid",
+                "geometry_path": "",          # disable geometry_path extraction
+                "lat_path": "lat",
+                "lon_path": "lon",
+                "title_path": "name",
+                "severity_path": "sev",
+            },
+        )
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf:
+            mf.return_value = JSON_FIXTURE
+            events = [e async for e in adapter.poll()]
+
+        assert len(events) == 2
+        e1 = next(e for e in events if e.id == "a1")
+        # centroid is (lon, lat) per GeoJSON convention
+        assert e1.geo.centroid == (-116.1, 43.5)
+        assert e1.data["title"] == "Alert One"
+        assert e1.severity == 1
+
+        await adapter.shutdown()
+
+
+# ---------------------------------------------------------------------------
+# Dedup
+# ---------------------------------------------------------------------------
+
+class TestDedup:
+    @pytest.mark.asyncio
+    async def test_second_poll_yields_nothing(self, temp_db_path, mock_config_store):
+        config = make_config(domain="wx")
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        geojson = {
+            "features": [
+                {"id": "dedup-1", "geometry": None},
+                {"id": "dedup-2", "geometry": None},
+            ]
+        }
+        with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf:
+            mf.return_value = geojson
+
+            first = [e async for e in adapter.poll()]
+            second = [e async for e in adapter.poll()]
+
+        assert len(first) == 2
+        assert len(second) == 0
+
+        await adapter.shutdown()
+
+    @pytest.mark.asyncio
+    async def test_new_item_in_second_poll_is_yielded(
+        self, temp_db_path, mock_config_store
+    ):
+        config = make_config(domain="wx")
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        first_batch = {"features": [{"id": "old-1", "geometry": None}]}
+        second_batch = {
+            "features": [
+                {"id": "old-1", "geometry": None},
+                {"id": "new-2", "geometry": None},
+            ]
+        }
+
+        with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf:
+            mf.return_value = first_batch
+            [e async for e in adapter.poll()]  # consume first
+
+            mf.return_value = second_batch
+            second = [e async for e in adapter.poll()]
+
+        assert len(second) == 1
+        assert second[0].id == "new-2"
+
+        await adapter.shutdown()
+
+    @pytest.mark.asyncio
+    async def test_instance_name_used_for_dedup(
+        self, temp_db_path, mock_config_store
+    ):
+        """Two adapter instances share the same db but scope by instance name."""
+        config_a = make_config(name="instance_a", domain="wx")
+        config_b = make_config(name="instance_b", domain="fire")
+
+        adapter_a = GenericHttpAdapter(config_a, mock_config_store, temp_db_path)
+        adapter_b = GenericHttpAdapter(config_b, mock_config_store, temp_db_path)
+        await adapter_a.startup()
+        await adapter_b.startup()
+
+        # Publish "shared-id" under instance_a
+        adapter_a.mark_published("shared-id")
+
+        # same id is NOT published under instance_b
+        assert not adapter_b.is_published("shared-id")
+        # and IS published under instance_a
+        assert adapter_a.is_published("shared-id")
+
+        await adapter_a.shutdown()
+        await adapter_b.shutdown()
+
+
+# ---------------------------------------------------------------------------
+# subject_for
+# ---------------------------------------------------------------------------
+
+class TestSubjectFor:
+    @pytest.mark.asyncio
+    async def test_subject_no_enrichment(self, temp_db_path, mock_config_store):
+        """Without enrichment data, subject should end in 'unknown'."""
+        config = make_config(domain="wx")
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        event = Event(
+            id="subj-1",
+            adapter="test_generic",
+            category="wx.alert",
+            time=datetime.now(timezone.utc),
+            geo=Geo(centroid=(-116.0, 43.0)),
+            data={},
+        )
+        subject = adapter.subject_for(event)
+        assert subject == "central.wx.unknown"
+
+        await adapter.shutdown()
+
+    @pytest.mark.asyncio
+    async def test_subject_with_us_enrichment(self, temp_db_path, mock_config_store):
+        """With US geocoder enrichment, subject should be central.<domain>.us.<state>."""
+        config = make_config(domain="fire")
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        event = Event(
+            id="subj-2",
+            adapter="test_generic",
+            category="fire.alert",
+            time=datetime.now(timezone.utc),
+            geo=Geo(centroid=(-116.0, 43.0)),
+            data={
+                "_enriched": {
+                    "geocoder": {
+                        "country": "United States",
+                        "state": "Idaho",
+                    }
+                }
+            },
+        )
+        subject = adapter.subject_for(event)
+        assert subject == "central.fire.us.id"
+
+        await adapter.shutdown()
+
+    @pytest.mark.asyncio
+    async def test_subject_format_domain_region(self, temp_db_path, mock_config_store):
+        """Subject always matches central.<domain>.<region> — NOT central.<category>.<region>."""
+        config = make_config(domain="quake", extra_settings={"category_suffix": "event.minor"})
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        event = Event(
+            id="subj-3",
+            adapter="test_generic",
+            category="quake.event.minor",
+            time=datetime.now(timezone.utc),
+            geo=Geo(),
+            data={},
+        )
+        subject = adapter.subject_for(event)
+        # Should be central.quake.unknown, NOT central.quake.event.minor.unknown
+        assert subject == "central.quake.unknown"
+        assert subject.startswith("central.quake.")
+
+        await adapter.shutdown()
+
+
+# ---------------------------------------------------------------------------
+# time_path parsing
+# ---------------------------------------------------------------------------
+
+class TestTimePath:
+    @pytest.mark.asyncio
+    async def test_time_path_none_uses_now(self, temp_db_path, mock_config_store):
+        before = datetime.now(timezone.utc)
+        config = make_config(domain="wx", extra_settings={"time_path": None})
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        geojson = {"features": [{"id": "t1", "geometry": None}]}
+        with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf:
+            mf.return_value = geojson
+            events = [e async for e in adapter.poll()]
+
+        after = datetime.now(timezone.utc)
+        assert len(events) == 1
+        assert before <= events[0].time <= after
+
+        await adapter.shutdown()
+
+    @pytest.mark.asyncio
+    async def test_time_path_parsed_iso(self, temp_db_path, mock_config_store):
+        config = make_config(
+            domain="wx",
+            extra_settings={
+                "time_path": "properties.updated",
+                "id_path": "id",
+                "items_path": "features",
+            },
+        )
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        geojson = {
+            "features": [
+                {
+                    "id": "t2",
+                    "geometry": None,
+                    "properties": {"updated": "2025-06-15T08:30:00Z"},
+                }
+            ]
+        }
+        with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf:
+            mf.return_value = geojson
+            events = [e async for e in adapter.poll()]
+
+        assert len(events) == 1
+        assert events[0].time == datetime(2025, 6, 15, 8, 30, 0, tzinfo=timezone.utc)
+
+        await adapter.shutdown()
+
+    @pytest.mark.asyncio
+    async def test_time_path_missing_value_uses_now(
+        self, temp_db_path, mock_config_store
+    ):
+        """When time_path is set but the value is absent, fall back to now."""
+        before = datetime.now(timezone.utc)
+        config = make_config(
+            domain="wx",
+            extra_settings={"time_path": "properties.ts"},
+        )
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        geojson = {"features": [{"id": "t3", "geometry": None, "properties": {}}]}
+        with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf:
+            mf.return_value = geojson
+            events = [e async for e in adapter.poll()]
+
+        after = datetime.now(timezone.utc)
+        assert len(events) == 1
+        assert before <= events[0].time <= after
+
+        await adapter.shutdown()
+
+
+# ---------------------------------------------------------------------------
+# apply_config
+# ---------------------------------------------------------------------------
+
+class TestApplyConfig:
+    @pytest.mark.asyncio
+    async def test_apply_config_updates_url(self, temp_db_path, mock_config_store):
+        config = make_config(domain="wx")
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        assert adapter._settings.url == "https://example.com/feed.json"
+
+        new_config = make_config(
+            domain="wx", extra_settings={"url": "https://other.example.com/data.json"}
+        )
+        await adapter.apply_config(new_config)
+
+        assert adapter._settings.url == "https://other.example.com/data.json"
+
+        await adapter.shutdown()
+
+    @pytest.mark.asyncio
+    async def test_apply_config_updates_domain(self, temp_db_path, mock_config_store):
+        config = make_config(domain="wx")
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        new_config = make_config(domain="fire")
+        await adapter.apply_config(new_config)
+
+        assert adapter._settings.domain == "fire"
+
+        await adapter.shutdown()
+
+
+# ---------------------------------------------------------------------------
+# Instance name scoping
+# ---------------------------------------------------------------------------
+
+class TestInstanceName:
+    def test_instance_name_is_config_name(self, temp_db_path, mock_config_store):
+        """adapter.name must be the instance name, not the class name."""
+        config = make_config(name="my_noaa_alerts", domain="wx")
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        assert adapter.name == "my_noaa_alerts"
+        assert GenericHttpAdapter.name == "generic_http"
+
+    @pytest.mark.asyncio
+    async def test_event_adapter_field_is_instance_name(
+        self, temp_db_path, mock_config_store
+    ):
+        config = make_config(name="my_source", domain="wx")
+        adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path)
+        await adapter.startup()
+
+        geojson = {"features": [{"id": "inst-1", "geometry": None}]}
+        with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf:
+            mf.return_value = geojson
+            events = [e async for e in adapter.poll()]
+
+        assert len(events) == 1
+        assert events[0].adapter == "my_source"
+
+        await adapter.shutdown()