fix(4-1): resolve api_key alias from per-adapter settings, not class attr

The /adapters list view's "⚠️ API Key Missing" chip, the /adapters/{name}
edit form's disabled enable-checkbox, the POST error re-render path, AND
the supervisor's adapter-start precondition all compared the hardcoded
SourceAdapter class attribute `requires_api_key` against `config.api_keys`,
ignoring the per-row `settings[api_key_field]` alias the operator
actually selected via the form.

FIRMS' class attr is `requires_api_key = "firms"`; the api_keys_new.html
placeholder text steers operators toward aliases like `firms_production`
instead, and the FIRMSSettings.api_key_alias field is exactly the
overridable slot that the form writes. The four predicates ignored that
slot, so a working key under any non-default alias was treated as
missing — chip on, checkbox disabled, supervisor refusing to start with
`last_error = "missing api key: firms"`.

Audit: FIRMS is the only adapter today with `requires_api_key != None`.
Every other adapter is unaffected by either the route or supervisor
predicate.

Helper module:
- src/central/api_key_resolver.py exposes:
    resolve_api_key_alias(adapter_cls, settings) -> str | None
        Pure sync function. Returns the alias to consult, or None when no
        key is required. Supervisor uses this directly + its own
        get_api_key.
    adapter_has_resolved_api_key(conn, adapter_cls, settings) -> (bool, alias)
        Async wrapper that runs the SELECT 1 against config.api_keys.
        The three GUI routes use this.
  Resolution: settings[api_key_field] when set to a non-empty str,
  otherwise the class-attr default.

Four call sites swapped:
- routes.py:adapters_list           (/adapters list — warning chip)
- routes.py:adapters_edit_form      (/adapters/{name} edit GET — disabled checkbox)
- routes.py:adapters_edit_submit    (POST error re-render)
- supervisor.py:_start_adapter      (adapter-start precondition)

Side-effect tests/test_adapters.py fix:
- TestAdaptersJsonbRegression::test_adapters_edit_fetches_api_keys_into_context
  used `AsyncMock()` (no return_value) for mock_conn.__aexit__. AsyncMock
  without a return_value yields a MagicMock — which is truthy, and the
  async context manager protocol reads truthy from __aexit__ as
  "exception suppressed." That silently swallowed any error inside
  `async with` blocks. The route refactor moved an assignment inside the
  one async with at site 2, so a swallowed mock error left the variable
  unbound. Fixed: `AsyncMock(return_value=None)` + a comment so the next
  person doesn't re-introduce the bug. fetchval mock added because the
  resolver now issues it (the swallowed exception previously hid the
  missing mock).

Verification:
- pytest: 479 passed (was 469; +10 new resolver tests).
- grep -rn "adapter_cls.requires_api_key" /opt/central/src returns only
  the new helper (2 lines, same file).
- Resolver against live FIRMS settings: resolved_alias='firms_production',
  has_key=True, api_key_missing=False -> NO warning chip, checkbox
  CLICKABLE.
- Supervisor on live CT104: FIRMS flipped enabled=true via DB UPDATE;
  supervisor started the adapter with `api_key_present: true,
  api_key_alias: 'firms_production'`; last_error cleared from "missing
  api key: firms" -> NULL; two satellite polls completed (VIIRS_SNPP_NRT
  477 features, VIIRS_NOAA20_NRT 400 features); 869 new events published
  to JetStream.

NOTE: This commit's verification flipped FIRMS to enabled=true in the
running config — the adapter is now actively polling. Pause via the UI
if that's not intended for now; the bug fix itself does not require
FIRMS to be enabled.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/central/api_key_resolver.py

@@ -0,0 +1,82 @@
+"""Resolve an adapter row's effective api-key alias and check existence.
+
+Four call sites share this answer: the /adapters list view, the
+/adapters/{name} edit form (GET), the same edit form's POST error
+re-render path, and the supervisor's adapter-start precondition.
+Previously each inlined a predicate that compared the hardcoded
+SourceAdapter class attribute `requires_api_key` against `config.api_keys`,
+ignoring the per-row `settings[api_key_field]` value the operator
+actually selected. An operator could pick alias "firms_production" via
+the form, store a working key under that alias, and still see the
+"⚠️ API Key Missing" chip + a disabled enable checkbox + the supervisor
+refusing to start the adapter — all from the same root cause.
+
+Two entry points:
+
+- `resolve_api_key_alias`  — sync. Pure function of (cls, settings).
+  Returns the alias the row should consult, or None when no key is
+  required (no SQL needed). Supervisor uses this directly because it
+  already has a ConfigStore and does its own key fetch.
+
+- `adapter_has_resolved_api_key` — async. Wraps the sync resolver with
+  the existence SELECT against config.api_keys. The three GUI route
+  call sites use this.
+
+Resolution order (same as the adapter's runtime read in __init__):
+  1. requires_api_key is None  -> no key needed.
+  2. settings[api_key_field] is a non-empty string -> use that.
+  3. fall back to the requires_api_key class-attribute default.
+"""
+
+from typing import Any
+
+from central.adapter import SourceAdapter
+
+
+def resolve_api_key_alias(
+    adapter_cls: type[SourceAdapter] | None,
+    settings: dict | None,
+) -> str | None:
+    """Return the api-key alias for this adapter row, or None when no key is required.
+
+    Pure function — no IO, no SQL. Same resolution order the adapter's own
+    __init__ uses when reading the alias out of config.settings.
+    """
+    if adapter_cls is None or adapter_cls.requires_api_key is None:
+        return None
+    field = adapter_cls.api_key_field
+    if field and settings:
+        value = settings.get(field)
+        if isinstance(value, str) and value.strip():
+            return value.strip()
+    return adapter_cls.requires_api_key
+
+
+async def adapter_has_resolved_api_key(
+    conn: Any,
+    adapter_cls: type[SourceAdapter] | None,
+    settings: dict | None,
+) -> tuple[bool, str | None]:
+    """Return (has_key, resolved_alias) for one adapter row.
+
+    Args:
+        conn: asyncpg connection or any object exposing an awaitable fetchval
+            with the same signature.
+        adapter_cls: The discovered SourceAdapter subclass for this row, or
+            None when the row references an adapter no longer in the codebase.
+        settings: The row's jsonb settings dict, or None when the row has no
+            settings stored yet.
+
+    Returns:
+        (has_key, resolved_alias). When the adapter does not require an api
+        key (requires_api_key is None) or no adapter class is available,
+        returns (True, None) and no SQL is issued.
+    """
+    alias = resolve_api_key_alias(adapter_cls, settings)
+    if alias is None:
+        return True, None
+    has_key = await conn.fetchval(
+        "SELECT 1 FROM config.api_keys WHERE alias = $1",
+        alias,
+    )
+    return bool(has_key), alias

src/central/gui/routes.py

@@ -51,6 +51,7 @@ from pathlib import Path
 from central.config_models import AdapterConfig
 from central.gui.db import get_pool
 from central.gui.form_descriptors import describe_fields, FieldDescriptor
+from central.api_key_resolver import adapter_has_resolved_api_key
 from central.adapter_discovery import discover_adapters
 from central.streams import STREAMS as STREAM_REGISTRY
 from pydantic import ValidationError
@@ -1348,16 +1349,13 @@ async def adapters_list(
             settings = row["settings"] or {}
             adapter_cls = adapter_classes.get(row["name"])
 
-            # Check if required API key is missing
-            api_key_missing = False
-            requires_api_key_alias = None
-            if adapter_cls and adapter_cls.requires_api_key is not None:
-                requires_api_key_alias = adapter_cls.requires_api_key
-                has_key = await conn.fetchval(
-                    "SELECT 1 FROM config.api_keys WHERE alias = $1",
-                    requires_api_key_alias,
-                )
-                api_key_missing = not has_key
+            # Check if required API key is missing — resolve via the per-row
+            # settings[api_key_field] (operator-selected alias), falling back
+            # to the class-attribute default when settings hasn't been set.
+            has_key, requires_api_key_alias = await adapter_has_resolved_api_key(
+                conn, adapter_cls, settings,
+            )
+            api_key_missing = not has_key
 
             adapters.append({
                 "name": row["name"],
@@ -1445,23 +1443,15 @@ async def adapters_edit_form(
                 if f.name == adapter_cls.api_key_field:
                     f.widget = "api_key_select"
 
-    # Fetch API keys for api_key_select widget
-    api_keys = []
+    # Fetch API keys for api_key_select widget + resolve the per-adapter
+    # alias against the operator-set settings, not the class-attr default.
     async with pool.acquire() as conn:
         api_key_rows = await conn.fetch("SELECT alias FROM config.api_keys ORDER BY alias")
         api_keys = [{"alias": r["alias"]} for r in api_key_rows]
-
-    # Check if required API key is missing
-    api_key_missing = False
-    requires_api_key_alias = None
-    if adapter_cls and adapter_cls.requires_api_key is not None:
-        requires_api_key_alias = adapter_cls.requires_api_key
-        async with pool.acquire() as conn:
-            has_key = await conn.fetchval(
-                "SELECT 1 FROM config.api_keys WHERE alias = $1",
-                requires_api_key_alias,
-            )
-            api_key_missing = not has_key
+        has_key, requires_api_key_alias = await adapter_has_resolved_api_key(
+            conn, adapter_cls, settings,
+        )
+        api_key_missing = not has_key
 
     # Generic settings-driven preview. Adapters opt in by overriding
     # SourceAdapter.preview_for_settings; the framework is duck-typed on the
@@ -1700,20 +1690,15 @@ async def adapters_edit_submit(
                         if f.name == adapter_cls.api_key_field:
                             f.widget = "api_key_select"
 
-            # Fetch API keys for api_key_select widget
+            # Fetch API keys for api_key_select widget + resolve the per-adapter
+            # alias against the pre-edit settings (form validation failed, so
+            # the stored settings haven't been replaced).
             api_key_rows = await conn.fetch("SELECT alias FROM config.api_keys ORDER BY alias")
             api_keys = [{"alias": r["alias"]} for r in api_key_rows]
-
-            # Check if required API key is missing
-            api_key_missing = False
-            requires_api_key_alias = None
-            if adapter_cls and adapter_cls.requires_api_key is not None:
-                requires_api_key_alias = adapter_cls.requires_api_key
-                has_key = await conn.fetchval(
-                    "SELECT 1 FROM config.api_keys WHERE alias = $1",
-                    requires_api_key_alias,
-                )
-                api_key_missing = not has_key
+            has_key, requires_api_key_alias = await adapter_has_resolved_api_key(
+                conn, adapter_cls, current_settings,
+            )
+            api_key_missing = not has_key
 
             csrf_token = request.state.csrf_token
             response = templates.TemplateResponse(

src/central/supervisor.py

@@ -20,6 +20,7 @@ from central.config_models import AdapterConfig
 from central.config_source import ConfigSource, DbConfigSource
 from central.config_store import ConfigStore
 from central.bootstrap_config import get_settings
+from central.api_key_resolver import resolve_api_key_alias
 from central.stream_manager import StreamManager
 from central.streams import STREAMS as STREAM_REGISTRY
 CURSOR_DB_PATH = Path("/var/lib/central/cursors.db")
@@ -263,10 +264,13 @@ class Supervisor:
         If the adapter was previously stopped (state exists but task is not running),
         reuses the existing state to preserve last_completed_poll for rate limiting.
         """
-        # API key precondition
+        # API key precondition — resolve via per-row settings[api_key_field]
+        # (operator-selected alias), falling back to the class-attribute
+        # default when settings hasn't been set. Returns None when no key
+        # is required.
         adapter_cls = self._adapters.get(config.name)
-        if adapter_cls is not None and adapter_cls.requires_api_key is not None:
-            alias = adapter_cls.requires_api_key
+        alias = resolve_api_key_alias(adapter_cls, config.settings)
+        if alias is not None:
             key_value = await self._config_store.get_api_key(alias)
             if not key_value:
                 error_msg = f"missing api key: {alias}"

tests/test_adapters.py

@@ -452,8 +452,17 @@ class TestAdaptersJsonbRegression:
             {"alias": "firms_key"},
             {"alias": "other_key"},
         ])
+        # The /adapters/{name} edit handler also issues a fetchval against
+        # config.api_keys to resolve whether the adapter's key is present.
+        # Return 1 (truthy) so the handler proceeds — this test only asserts
+        # api_keys reaches the template context, not the warning state.
+        mock_conn.fetchval = AsyncMock(return_value=1)
         mock_conn.__aenter__ = AsyncMock(return_value=mock_conn)
-        mock_conn.__aexit__ = AsyncMock()
+        # AsyncMock() with no return_value yields a MagicMock — which is truthy,
+        # and the async context manager protocol reads a truthy __aexit__ return
+        # as "exception suppressed." That silently swallows any error inside the
+        # `async with` block. Pin return_value=None so exceptions propagate.
+        mock_conn.__aexit__ = AsyncMock(return_value=None)
 
         mock_pool = MagicMock()
         mock_pool.acquire = MagicMock(return_value=mock_conn)

tests/test_api_key_resolver.py

@@ -0,0 +1,249 @@
+"""Tests for central.gui._api_key_resolver.adapter_has_resolved_api_key.
+
+The /adapters list, /adapters/{name} edit, and POST error re-render flows
+previously inlined a predicate that compared the SourceAdapter class
+attribute `requires_api_key` (default literal string) against
+`config.api_keys` — ignoring the per-row `settings[api_key_field]` alias
+the operator actually selected. The helper consolidates the resolution
+logic; these tests pin it.
+
+No hardcoded adapter-name list literals: keyless-adapter case picks a
+real one via central.adapter_discovery. The settings-driven cases use
+inline minimal SourceAdapter subclasses because no live adapter today
+has the pertinent class-attr/field combinations.
+"""
+
+from collections.abc import AsyncIterator
+from typing import Any
+
+import pytest
+from pydantic import BaseModel
+
+from central.adapter import SourceAdapter
+from central.adapter_discovery import discover_adapters
+from central.api_key_resolver import (
+    adapter_has_resolved_api_key,
+    resolve_api_key_alias,
+)
+from central.models import Event
+
+
+class _FakeConn:
+    """Captures the (sql, alias) of each fetchval call and returns a scripted
+    result. Mirrors the asyncpg connection surface adapter_has_resolved_api_key
+    actually uses."""
+
+    def __init__(self, result: Any) -> None:
+        self.result = result
+        self.calls: list[tuple[str, tuple[Any, ...]]] = []
+
+    async def fetchval(self, sql: str, *params: Any) -> Any:
+        self.calls.append((sql, params))
+        return self.result
+
+
+def _pick_keyless_adapter() -> type[SourceAdapter] | None:
+    """First discovered adapter with requires_api_key is None.
+
+    Lets test 1 exercise the no-SQL short-circuit against a real subclass
+    without hardcoding the adapter's name.
+    """
+    for cls in discover_adapters().values():
+        if cls.requires_api_key is None:
+            return cls
+    return None
+
+
+class _BareKeyAdapter(SourceAdapter):
+    """Requires a key but does NOT expose an operator-overridable field.
+
+    Forces the helper to fall back to the class-attribute default.
+    """
+
+    name = "_bare_key"
+    display_name = "_BareKey"
+    description = "Test fixture: class-attr fallback path."
+    settings_schema = BaseModel
+    requires_api_key = "default"
+    api_key_field = None
+    default_cadence_s = 60
+
+    async def poll(self) -> AsyncIterator[Event]:  # pragma: no cover
+        if False:
+            yield  # type: ignore[unreachable]
+        return
+
+    async def apply_config(self, new_config) -> None:  # pragma: no cover
+        pass
+
+    def subject_for(self, event: Event) -> str:  # pragma: no cover
+        return "central.test._bare_key"
+
+
+class _OperatorOverridableAdapter(SourceAdapter):
+    """Requires a key AND exposes an operator-overridable alias field.
+
+    Models the FIRMS-shaped contract that previously misfired.
+    """
+
+    name = "_overridable"
+    display_name = "_Overridable"
+    description = "Test fixture: operator-overridable alias field."
+    settings_schema = BaseModel
+    requires_api_key = "default"
+    api_key_field = "api_key_alias"
+    default_cadence_s = 60
+
+    async def poll(self) -> AsyncIterator[Event]:  # pragma: no cover
+        if False:
+            yield  # type: ignore[unreachable]
+        return
+
+    async def apply_config(self, new_config) -> None:  # pragma: no cover
+        pass
+
+    def subject_for(self, event: Event) -> str:  # pragma: no cover
+        return "central.test._overridable"
+
+
+@pytest.mark.asyncio
+async def test_keyless_adapter_short_circuits_without_sql():
+    """An adapter with requires_api_key=None returns (True, None) and the
+    helper issues NO SQL — caller can assume the gate is open."""
+    keyless = _pick_keyless_adapter()
+    if keyless is None:
+        pytest.skip("no keyless adapter discovered in central.adapters")
+    conn = _FakeConn(result=1)
+    has_key, alias = await adapter_has_resolved_api_key(conn, keyless, {})
+    assert (has_key, alias) == (True, None)
+    assert conn.calls == [], "no SQL expected for keyless adapter"
+
+
+@pytest.mark.asyncio
+async def test_class_attr_fallback_when_no_api_key_field():
+    """Adapter exposes requires_api_key but no api_key_field — helper falls
+    back to the class-attribute default for the lookup alias."""
+    conn = _FakeConn(result=1)
+    has_key, alias = await adapter_has_resolved_api_key(conn, _BareKeyAdapter, {})
+    assert alias == "default"
+    assert has_key is True
+    assert len(conn.calls) == 1
+    sql, params = conn.calls[0]
+    assert "config.api_keys" in sql
+    assert params == ("default",)
+
+
+@pytest.mark.asyncio
+async def test_settings_override_takes_precedence_over_class_attr():
+    """Operator-selected alias in settings[api_key_field] wins over the
+    class-attribute default. Regression guard for the FIRMS-shaped bug:
+    requires_api_key='default' but operator stored a key under 'custom'."""
+    conn = _FakeConn(result=1)
+    settings = {"api_key_alias": "custom"}
+    has_key, alias = await adapter_has_resolved_api_key(
+        conn, _OperatorOverridableAdapter, settings,
+    )
+    assert alias == "custom", "operator-selected alias should win"
+    assert has_key is True
+    assert conn.calls[0][1] == ("custom",), (
+        f"SQL must query the operator-selected alias, got {conn.calls[0][1]!r}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_missing_settings_field_falls_back_to_class_attr():
+    """Operator hasn't visited the form yet — settings dict lacks the
+    api_key_field entry. Class-attr default is the safe fallback."""
+    conn = _FakeConn(result=1)
+    has_key, alias = await adapter_has_resolved_api_key(
+        conn, _OperatorOverridableAdapter, {"unrelated": "noise"},
+    )
+    assert alias == "default"
+    assert conn.calls[0][1] == ("default",)
+
+    # Same path when settings is None entirely.
+    conn2 = _FakeConn(result=1)
+    has_key2, alias2 = await adapter_has_resolved_api_key(
+        conn2, _OperatorOverridableAdapter, None,
+    )
+    assert alias2 == "default"
+    assert conn2.calls[0][1] == ("default",)
+
+
+@pytest.mark.asyncio
+async def test_empty_string_settings_field_falls_back_to_class_attr():
+    """Empty string in settings does NOT mean "look up alias '' "; it means
+    "not set, use default" — same shape the form parser yields for cleared
+    text inputs."""
+    conn = _FakeConn(result=1)
+    has_key, alias = await adapter_has_resolved_api_key(
+        conn, _OperatorOverridableAdapter, {"api_key_alias": "   "},
+    )
+    assert alias == "default", "whitespace-only must fall through to class attr"
+    assert conn.calls[0][1] == ("default",)
+
+    conn2 = _FakeConn(result=1)
+    has_key2, alias2 = await adapter_has_resolved_api_key(
+        conn2, _OperatorOverridableAdapter, {"api_key_alias": ""},
+    )
+    assert alias2 == "default"
+    assert conn2.calls[0][1] == ("default",)
+
+
+@pytest.mark.asyncio
+async def test_resolved_alias_absent_returns_has_key_false():
+    """If the resolved alias has no matching row, helper returns
+    (False, alias) so caller renders the warning chip with the right alias
+    shown in the title attribute."""
+    conn = _FakeConn(result=None)
+    has_key, alias = await adapter_has_resolved_api_key(
+        conn, _OperatorOverridableAdapter, {"api_key_alias": "ghost"},
+    )
+    assert (has_key, alias) == (False, "ghost")
+
+
+@pytest.mark.asyncio
+async def test_unknown_adapter_cls_short_circuits():
+    """Row references a deleted adapter — adapter_cls is None. Helper must
+    not crash, must not issue SQL, and must report no missing key (the warning
+    is for known adapters; orphaned rows are a separate concern)."""
+    conn = _FakeConn(result=None)
+    has_key, alias = await adapter_has_resolved_api_key(conn, None, {})
+    assert (has_key, alias) == (True, None)
+    assert conn.calls == []
+
+
+# ---------------------------------------------------------------------------
+# Supervisor uses the sync resolver directly (no DB round-trip — supervisor
+# has its own ConfigStore and calls get_api_key on the resolved alias).
+# These tests mirror the route coverage on the sync entry point.
+# ---------------------------------------------------------------------------
+
+
+def test_supervisor_uses_operator_selected_alias():
+    """Supervisor's adapter-start precondition must resolve to the same alias
+    the GUI's warning predicate resolves to. Regression guard for the second
+    half of the FIRMS-shaped bug: routes opened the gate but supervisor still
+    refused to start with `missing api key: <class_attr>`."""
+    alias = resolve_api_key_alias(
+        _OperatorOverridableAdapter, {"api_key_alias": "firms_production"},
+    )
+    assert alias == "firms_production"
+
+
+def test_supervisor_falls_back_to_class_attr_when_settings_unset():
+    """Operator hasn't touched settings yet — supervisor still finds the
+    class-attribute default (preserves pre-bug behavior for fresh installs)."""
+    assert resolve_api_key_alias(_OperatorOverridableAdapter, None) == "default"
+    assert resolve_api_key_alias(_OperatorOverridableAdapter, {}) == "default"
+    assert resolve_api_key_alias(_BareKeyAdapter, {}) == "default"
+
+
+def test_supervisor_returns_none_for_keyless_adapter():
+    """When the adapter doesn't need a key, supervisor should skip the
+    `get_api_key` call entirely — the sync resolver returns None to signal."""
+    keyless = _pick_keyless_adapter()
+    if keyless is None:
+        pytest.skip("no keyless adapter discovered in central.adapters")
+    assert resolve_api_key_alias(keyless, {}) is None
+    assert resolve_api_key_alias(None, {}) is None