feat(viewer): add /api/observer-status endpoint

codemem/observer.py

@@ -153,6 +153,29 @@ def _load_opencode_oauth_cache() -> dict[str, Any]:
     return data if isinstance(data, dict) else {}
 
 
+def probe_available_credentials() -> dict[str, dict[str, bool]]:
+    """Probe which credentials are available for each provider without creating clients.
+
+    Returns a dict keyed by provider name with booleans for each credential source.
+    """
+    oauth_cache = _load_opencode_oauth_cache()
+    now = _now_ms()
+    result: dict[str, dict[str, bool]] = {}
+    for provider_key in ("openai", "anthropic"):
+        oauth_access = _extract_oauth_access(oauth_cache, provider_key)
+        oauth_expires = _extract_oauth_expires(oauth_cache, provider_key)
+        oauth_valid = bool(oauth_access) and (oauth_expires is None or oauth_expires > now)
+        env_key = "OPENAI_API_KEY" if provider_key == "openai" else "ANTHROPIC_API_KEY"
+        env_present = bool(os.getenv(env_key))
+        explicit_present = bool(os.getenv("CODEMEM_OBSERVER_API_KEY"))
+        result[provider_key] = {
+            "oauth": oauth_valid,
+            "api_key": explicit_present,
+            "env_var": env_present,
+        }
+    return result
+
+
 @dataclass
 class ObserverResponse:
     raw: str | None
@@ -231,6 +254,32 @@ def __init__(self) -> None:
         if self.runtime == "api_http":
             self._init_provider_client(force_refresh=False)
 
+    def get_status(self) -> dict[str, Any]:
+        """Return the resolved runtime state of this observer client."""
+        method = "none"
+        if self.runtime == "claude_sidecar":
+            method = "claude_sidecar"
+        elif self.use_opencode_run:
+            method = "opencode_run"
+        elif self.anthropic_oauth_access:
+            method = "anthropic_consumer"
+        elif self.codex_access:
+            method = "codex_consumer"
+        elif self.client is not None:
+            method = "sdk_client"
+
+        token_present = bool(self.auth.token)
+        return {
+            "provider": self.provider,
+            "model": self.model,
+            "runtime": self.runtime,
+            "auth": {
+                "source": self.auth.source,
+                "method": method,
+                "token_present": token_present,
+            },
+        }
+
     def _init_provider_client(self, *, force_refresh: bool) -> None:
         self.client = None
         self.codex_access = None

codemem/viewer.py

@@ -23,6 +23,7 @@
 )
 from .viewer_routes import config as viewer_routes_config
 from .viewer_routes import memory as viewer_routes_memory
+from .viewer_routes import observer_status as viewer_routes_observer_status
 from .viewer_routes import raw_events as viewer_routes_raw_events
 from .viewer_routes import stats as viewer_routes_stats
 from .viewer_routes import sync as viewer_routes_sync
@@ -102,6 +103,8 @@ def do_GET(self) -> None:  # noqa: N802
                 return
             if viewer_routes_memory.handle_get(self, store, parsed.path, parsed.query):
                 return
+            if viewer_routes_observer_status.handle_get(self, parsed.path):
+                return
             if viewer_routes_config.handle_get(
                 self,
                 path=parsed.path,

codemem/viewer_routes/observer_status.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from typing import Any, Protocol
+
+from .. import plugin_ingest as _plugin_ingest
+from ..observer import probe_available_credentials
+
+
+class _ViewerHandler(Protocol):
+    def _send_json(self, payload: dict[str, Any], status: int = 200) -> None: ...
+
+
+def handle_get(handler: _ViewerHandler, path: str) -> bool:
+    if path != "/api/observer-status":
+        return False
+
+    available = probe_available_credentials()
+    active: dict[str, Any] | None = None
+    observer = _plugin_ingest.OBSERVER
+    if observer is not None:
+        active = observer.get_status()
+
+    handler._send_json(
+        {
+            "active": active,
+            "available_credentials": available,
+        }
+    )
+    return True

tests/test_observer_status.py

@@ -0,0 +1,189 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from unittest.mock import patch
+
+from codemem.config import OpencodeMemConfig
+from codemem.observer import ObserverClient, probe_available_credentials
+
+
+def test_probe_finds_openai_oauth(tmp_path: Path) -> None:
+    auth_path = tmp_path / "auth.json"
+    auth_path.write_text(
+        json.dumps(
+            {
+                "openai": {
+                    "type": "oauth",
+                    "access": "oa-access",
+                    "refresh": "oa-refresh",
+                    "expires": 9999999999999,
+                }
+            }
+        )
+    )
+    with (
+        patch("codemem.observer._get_opencode_auth_path", return_value=auth_path),
+        patch.dict("os.environ", {}, clear=True),
+    ):
+        result = probe_available_credentials()
+    assert result["openai"]["oauth"] is True
+    assert result["openai"]["env_var"] is False
+    assert result["anthropic"]["oauth"] is False
+
+
+def test_probe_finds_anthropic_oauth(tmp_path: Path) -> None:
+    auth_path = tmp_path / "auth.json"
+    auth_path.write_text(
+        json.dumps(
+            {
+                "anthropic": {
+                    "type": "oauth",
+                    "access": "anth-access",
+                    "refresh": "anth-refresh",
+                    "expires": 9999999999999,
+                }
+            }
+        )
+    )
+    with (
+        patch("codemem.observer._get_opencode_auth_path", return_value=auth_path),
+        patch.dict("os.environ", {}, clear=True),
+    ):
+        result = probe_available_credentials()
+    assert result["anthropic"]["oauth"] is True
+    assert result["openai"]["oauth"] is False
+
+
+def test_probe_finds_both_oauth(tmp_path: Path) -> None:
+    auth_path = tmp_path / "auth.json"
+    auth_path.write_text(
+        json.dumps(
+            {
+                "openai": {
+                    "type": "oauth",
+                    "access": "oa-access",
+                    "refresh": "oa-refresh",
+                    "expires": 9999999999999,
+                },
+                "anthropic": {
+                    "type": "oauth",
+                    "access": "anth-access",
+                    "refresh": "anth-refresh",
+                    "expires": 9999999999999,
+                },
+            }
+        )
+    )
+    with (
+        patch("codemem.observer._get_opencode_auth_path", return_value=auth_path),
+        patch.dict("os.environ", {}, clear=True),
+    ):
+        result = probe_available_credentials()
+    assert result["openai"]["oauth"] is True
+    assert result["anthropic"]["oauth"] is True
+
+
+def test_probe_detects_expired_oauth(tmp_path: Path) -> None:
+    auth_path = tmp_path / "auth.json"
+    auth_path.write_text(
+        json.dumps(
+            {
+                "anthropic": {
+                    "type": "oauth",
+                    "access": "anth-access",
+                    "refresh": "anth-refresh",
+                    "expires": 1000,  # long expired
+                }
+            }
+        )
+    )
+    with (
+        patch("codemem.observer._get_opencode_auth_path", return_value=auth_path),
+        patch.dict("os.environ", {}, clear=True),
+    ):
+        result = probe_available_credentials()
+    assert result["anthropic"]["oauth"] is False
+
+
+def test_probe_detects_env_vars(tmp_path: Path) -> None:
+    auth_path = tmp_path / "auth.json"
+    auth_path.write_text("{}")
+    with (
+        patch("codemem.observer._get_opencode_auth_path", return_value=auth_path),
+        patch.dict(
+            "os.environ",
+            {"OPENAI_API_KEY": "sk-test", "ANTHROPIC_API_KEY": "ant-test"},
+            clear=True,
+        ),
+    ):
+        result = probe_available_credentials()
+    assert result["openai"]["env_var"] is True
+    assert result["anthropic"]["env_var"] is True
+    assert result["openai"]["oauth"] is False
+    assert result["anthropic"]["oauth"] is False
+
+
+def test_probe_detects_explicit_api_key(tmp_path: Path) -> None:
+    auth_path = tmp_path / "auth.json"
+    auth_path.write_text("{}")
+    with (
+        patch("codemem.observer._get_opencode_auth_path", return_value=auth_path),
+        patch.dict("os.environ", {"CODEMEM_OBSERVER_API_KEY": "explicit"}, clear=True),
+    ):
+        result = probe_available_credentials()
+    assert result["openai"]["api_key"] is True
+    assert result["anthropic"]["api_key"] is True
+
+
+def test_get_status_anthropic_consumer(tmp_path: Path) -> None:
+    auth_path = tmp_path / "auth.json"
+    auth_path.write_text(
+        json.dumps(
+            {
+                "anthropic": {
+                    "type": "oauth",
+                    "access": "anth-access",
+                    "refresh": "anth-refresh",
+                    "expires": 9999999999999,
+                }
+            }
+        )
+    )
+    cfg = OpencodeMemConfig(observer_api_key=None, observer_provider="anthropic")
+    with (
+        patch("codemem.observer.load_config", return_value=cfg),
+        patch("codemem.observer._get_opencode_auth_path", return_value=auth_path),
+        patch.dict("os.environ", {}, clear=True),
+    ):
+        client = ObserverClient()
+        status = client.get_status()
+    assert status["provider"] == "anthropic"
+    assert status["auth"]["method"] == "anthropic_consumer"
+    assert status["auth"]["source"] == "oauth"
+    assert status["auth"]["token_present"] is True
+
+
+def test_get_status_claude_sidecar() -> None:
+    cfg = OpencodeMemConfig(observer_runtime="claude_sidecar")
+    with (
+        patch("codemem.observer.load_config", return_value=cfg),
+        patch("codemem.observer._load_opencode_oauth_cache", return_value={}),
+    ):
+        client = ObserverClient()
+        status = client.get_status()
+    assert status["runtime"] == "claude_sidecar"
+    assert status["auth"]["method"] == "claude_sidecar"
+
+
+def test_get_status_no_auth() -> None:
+    cfg = OpencodeMemConfig(observer_api_key=None, observer_provider="anthropic")
+    with (
+        patch("codemem.observer.load_config", return_value=cfg),
+        patch("codemem.observer._load_opencode_oauth_cache", return_value={}),
+        patch.dict("os.environ", {}, clear=True),
+    ):
+        client = ObserverClient()
+        status = client.get_status()
+    assert status["auth"]["method"] == "none"
+    assert status["auth"]["token_present"] is False