Stage 310: PR #1768 — fix(oauth): serialize Anthropic env fallback reads by @franksong2702

api/oauth.py

@@ -56,6 +56,30 @@ ANTHROPIC_PUBLIC_LINK_ERROR = "Claude Code credential linking failed. Check serv
 
 _OAUTH_FLOWS: dict[str, dict[str, Any]] = {}
 _OAUTH_FLOWS_LOCK = threading.Lock()
+_ANTHROPIC_ENV_KEYS = ("ANTHROPIC_TOKEN", "ANTHROPIC_API_KEY")
+
+
+def _clear_process_anthropic_env_values() -> None:
+    """Clear Anthropic process env fallbacks under the streaming env lock."""
+    from api.streaming import _ENV_LOCK
+
+    with _ENV_LOCK:
+        for key in _ANTHROPIC_ENV_KEYS:
+            os.environ.pop(key, None)
+
+
+def resolve_runtime_provider_with_anthropic_env_lock(resolver, *args, **kwargs):
+    """Resolve runtime credentials under the Anthropic onboarding env lock.
+
+    Request paths must resolve Anthropic env fallbacks per outbound request,
+    not cache ANTHROPIC_TOKEN or ANTHROPIC_API_KEY across onboarding. Sharing
+    the process-env lock prevents a chat stream from observing one stale
+    Anthropic env value while onboarding has already cleared the other.
+    """
+    from api.streaming import _ENV_LOCK
+
+    with _ENV_LOCK:
+        return resolver(*args, **kwargs)
 
 
 def _normalize_onboarding_oauth_provider(provider: str) -> str:
@@ -234,18 +258,22 @@ def _read_claude_code_credentials() -> dict[str, Any] | None:
 
 
 def _clear_anthropic_env_values(hermes_home: Path) -> None:
-    """Clear Anthropic API/setup-token env values in the active profile only."""
+    """Clear Anthropic API/setup-token env values in the active profile only.
+
+    The .env write path already clears os.environ while holding the streaming
+    env lock. Keep a locked process-env clear here too so import/write failures
+    cannot leave or partially clear stale Anthropic fallbacks.
+    """
     try:
         from api.providers import _write_env_file
 
         _write_env_file(
             Path(hermes_home) / ".env",
-            {"ANTHROPIC_TOKEN": None, "ANTHROPIC_API_KEY": None},
+            {key: None for key in _ANTHROPIC_ENV_KEYS},
         )
     except Exception as exc:
         logger.warning("Failed to clear Anthropic env values: %s", exc)
-    os.environ.pop("ANTHROPIC_TOKEN", None)
-    os.environ.pop("ANTHROPIC_API_KEY", None)
+    _clear_process_anthropic_env_values()
 
 
 def _link_anthropic_credentials(hermes_home: Path) -> None:

api/routes.py

@@ -6261,9 +6261,13 @@ def _handle_chat_sync(handler, body):
             # Resolve API key via Hermes runtime provider (matches gateway behaviour)
             _api_key = None
             try:
+                from api.oauth import resolve_runtime_provider_with_anthropic_env_lock
                 from hermes_cli.runtime_provider import resolve_runtime_provider
 
-                _rt = resolve_runtime_provider(requested=_provider)
+                _rt = resolve_runtime_provider_with_anthropic_env_lock(
+                    resolve_runtime_provider,
+                    requested=_provider,
+                )
                 _api_key = _rt.get("api_key")
                 # Also use runtime provider/base_url if the webui config didn't resolve them
                 if not _provider:
@@ -7015,6 +7019,7 @@ def _handle_session_compress(handler, body):
                 )
 
         import api.config as _cfg
+        from api.oauth import resolve_runtime_provider_with_anthropic_env_lock
         import hermes_cli.runtime_provider as _runtime_provider
         import run_agent as _run_agent
 
@@ -7024,7 +7029,10 @@ def _handle_session_compress(handler, body):
 
         resolved_api_key = None
         try:
-            _rt = _runtime_provider.resolve_runtime_provider(requested=resolved_provider)
+            _rt = resolve_runtime_provider_with_anthropic_env_lock(
+                _runtime_provider.resolve_runtime_provider,
+                requested=resolved_provider,
+            )
             resolved_api_key = _rt.get("api_key")
             if not resolved_provider:
                 resolved_provider = _rt.get("provider")
@@ -7616,6 +7624,7 @@ def _handle_handoff_summary(handler, body):
         # Call LLM for summary.
     try:
         import api.config as _cfg
+        from api.oauth import resolve_runtime_provider_with_anthropic_env_lock
         import hermes_cli.runtime_provider as _runtime_provider
         import run_agent as _run_agent
 
@@ -7634,7 +7643,10 @@ def _handle_handoff_summary(handler, body):
 
         resolved_api_key = None
         try:
-            _rt = _runtime_provider.resolve_runtime_provider(requested=resolved_provider)
+            _rt = resolve_runtime_provider_with_anthropic_env_lock(
+                _runtime_provider.resolve_runtime_provider,
+                requested=resolved_provider,
+            )
             resolved_api_key = _rt.get("api_key")
             if not resolved_provider:
                 resolved_provider = _rt.get("provider")

api/streaming.py

@@ -1741,7 +1741,10 @@ def _attempt_credential_self_heal(
        re-invoke ``run_conversation`` with these).
     """
     try:
-        from api.oauth import read_auth_json
+        from api.oauth import (
+            read_auth_json,
+            resolve_runtime_provider_with_anthropic_env_lock,
+        )
         from api.config import (
             SESSION_AGENT_CACHE, SESSION_AGENT_CACHE_LOCK,
             invalidate_credential_pool_cache,
@@ -1762,7 +1765,10 @@ def _attempt_credential_self_heal(
         invalidate_credential_pool_cache(provider_id)
 
         # 4. Re-resolve runtime provider with fresh credentials
-        _new_rt = resolve_runtime_provider(requested=provider_id)
+        _new_rt = resolve_runtime_provider_with_anthropic_env_lock(
+            resolve_runtime_provider,
+            requested=provider_id,
+        )
 
         logger.info(
             '[webui] self-heal: credential refresh succeeded for provider=%s session=%s',
@@ -2170,8 +2176,12 @@ def _run_agent_streaming(
             # Pass the resolved provider so non-default providers get their own credentials.
             resolved_api_key = None
             try:
+                from api.oauth import resolve_runtime_provider_with_anthropic_env_lock
                 from hermes_cli.runtime_provider import resolve_runtime_provider
-                _rt = resolve_runtime_provider(requested=resolved_provider)
+                _rt = resolve_runtime_provider_with_anthropic_env_lock(
+                    resolve_runtime_provider,
+                    requested=resolved_provider,
+                )
                 resolved_api_key = _rt.get("api_key")
                 if not resolved_provider:
                     resolved_provider = _rt.get("provider")

tests/test_issue1362_codex_oauth_onboarding.py

@@ -3,7 +3,9 @@
 from __future__ import annotations
 
 import json
+import os
 import stat
+import threading
 import time
 from pathlib import Path
 
@@ -474,7 +476,6 @@ def test_anthropic_worker_reports_link_errors(monkeypatch, tmp_path):
 
 
 def test_anthropic_link_clears_env_and_writes_secret_free_marker(monkeypatch, tmp_path):
-    import os
     import api.oauth as oauth
     from api.onboarding import _provider_oauth_authenticated
 
@@ -501,6 +502,55 @@ def test_anthropic_link_clears_env_and_writes_secret_free_marker(monkeypatch, tm
     assert _provider_oauth_authenticated("claude-code", tmp_path) is True
 
 
+def test_anthropic_env_clear_waits_for_chat_env_read_lock(monkeypatch, tmp_path):
+    import api.oauth as oauth
+    import api.providers as providers
+    from api.streaming import _ENV_LOCK
+
+    monkeypatch.setenv("ANTHROPIC_TOKEN", "old-token")
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "old-key")
+
+    def _fail_before_env_lock(_env_path, _updates):
+        raise RuntimeError("env write failed before process-env clear")
+
+    monkeypatch.setattr(providers, "_write_env_file", _fail_before_env_lock)
+
+    started = threading.Event()
+    done = threading.Event()
+    errors = []
+
+    def _onboarding_clear():
+        started.set()
+        try:
+            oauth._clear_anthropic_env_values(tmp_path)
+        except Exception as exc:  # pragma: no cover - assertion below reports it
+            errors.append(exc)
+        finally:
+            done.set()
+
+    with _ENV_LOCK:
+        worker = threading.Thread(target=_onboarding_clear)
+        worker.start()
+        assert started.wait(timeout=1)
+        assert not done.wait(timeout=0.1)
+        assert os.environ["ANTHROPIC_TOKEN"] == "old-token"
+        assert os.environ["ANTHROPIC_API_KEY"] == "old-key"
+
+    worker.join(timeout=1)
+    assert done.is_set()
+    assert errors == []
+    assert "ANTHROPIC_TOKEN" not in os.environ
+    assert "ANTHROPIC_API_KEY" not in os.environ
+
+
+def test_runtime_provider_reads_use_anthropic_env_lock():
+    streaming_src = (REPO / "api" / "streaming.py").read_text(encoding="utf-8")
+    routes_src = (REPO / "api" / "routes.py").read_text(encoding="utf-8")
+
+    assert "resolve_runtime_provider_with_anthropic_env_lock" in streaming_src
+    assert "resolve_runtime_provider_with_anthropic_env_lock" in routes_src
+
+
 def test_anthropic_onboarding_setup_allows_linked_oauth_without_api_key(monkeypatch, tmp_path):
     import api.onboarding as onboarding