feat(brain): real Claude backend via Max OAuth token (auth_token + Claude Code system block)

Wires the deployment's Claude Max login (OAuth token from
$CLAUDE_CREDENTIALS_PATH) into ClaudeBrain/ClaudeVision. OAuth tokens
authenticate as Bearer (auth_token=), not x-api-key, and only answer
when the first system block is the Claude Code identity string, so the
real persona moves to a second system block. Token is re-read per
request so a host-side refresh is picked up without a restart. Falls
back to ANTHROPIC_API_KEY when set.

Verified: WSAI_BRAIN=claude returns a real Korean reply through the
factory; 7 smoke tests still pass.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

wsai/backends/claude.py

@@ -1,31 +1,90 @@
 """Cloud brain + vision via the Anthropic (Claude) API.
 
-Both share one client. Vision sends the frame as a base64 image; the brain is a
+Both share one auth resolver. Vision sends the frame as a base64 image; the
-plain chat call that receives the latest screen description as context.
+brain is a plain chat call that receives the latest screen description as
+context.
+
+Auth (two ways, tried in this order):
+  1. ANTHROPIC_API_KEY  -> standard API-key auth.
+  2. A Claude Code OAuth token (this deployment's Max login) read from the
+     credentials file at $CLAUDE_CREDENTIALS_PATH. OAuth tokens require the
+     first system block to be exactly the Claude Code identity string and are
+     sent as a Bearer token (auth_token=), not x-api-key. The token is
+     re-read before each request so a refresh rotated into the file by the
+     host is picked up without a restart.
 
 Requires: pip install anthropic
-Env: ANTHROPIC_API_KEY
 """
 
 from __future__ import annotations
 
 import base64
+import json
 import os
 import time
 
 from ..interfaces import Frame, Reply, ScreenObservation
 
+# Claude Code OAuth tokens only answer when the first system block is exactly
+# this identity string; the real persona/instructions go in later blocks.
+_CLAUDE_CODE_ID = "You are Claude Code, Anthropic's official CLI for Claude."
 
-def _client(api_key: str | None):
-    import anthropic
 
-    return anthropic.AsyncAnthropic(api_key=api_key or os.environ.get("ANTHROPIC_API_KEY"))
+def _load_oauth_token() -> str | None:
+    path = os.environ.get("CLAUDE_CREDENTIALS_PATH")
+    if not path or not os.path.exists(path):
+        return None
+    try:
+        with open(path) as f:
+            return json.load(f)["claudeAiOauth"]["accessToken"]
+    except (OSError, KeyError, ValueError):
+        return None
+
+
+class _Auth:
+    """Resolves a Claude client, preferring an explicit/env API key and falling
+    back to the deployment's OAuth token. The OAuth client is rebuilt whenever
+    the token in the credentials file changes (host-side refresh)."""
+
+    def __init__(self, api_key: str | None = None) -> None:
+        import anthropic  # lazy so mock mode needs no dependency
+
+        self._anthropic = anthropic
+        self._explicit_key = api_key
+        self._client = None
+        self._token: str | None = None
+        self._oauth = False
+
+    def client(self):
+        key = self._explicit_key or os.environ.get("ANTHROPIC_API_KEY")
+        if key:
+            if self._client is None:
+                self._client = self._anthropic.AsyncAnthropic(api_key=key)
+            self._oauth = False
+            return self._client
+        tok = _load_oauth_token()
+        if not tok:
+            raise RuntimeError(
+                "No Claude auth: set ANTHROPIC_API_KEY or provide a Claude "
+                "OAuth login via CLAUDE_CREDENTIALS_PATH."
+            )
+        if tok != self._token:
+            self._token = tok
+            self._client = self._anthropic.AsyncAnthropic(auth_token=tok)
+        self._oauth = True
+        return self._client
+
+    def system(self, *blocks: str) -> list[dict]:
+        """Build the system prompt, prepending the Claude Code identity block
+        when authing via OAuth (required) — harmless to include either way."""
+        texts = [_CLAUDE_CODE_ID, *blocks] if self._oauth else list(blocks)
+        return [{"type": "text", "text": t} for t in texts if t]
 
 
 class ClaudeVision:
     def __init__(self, *, model: str = "claude-sonnet-4-5", api_key: str | None = None) -> None:
         self.model = model
-        self._client = _client(api_key)
+        self._auth = _Auth(api_key)
 
     async def describe(self, frame: Frame, hint: str | None = None) -> ScreenObservation:
         prompt = hint or (
@@ -33,9 +92,11 @@ class ClaudeVision:
             "2~3문장으로 한국어로 간결하게 설명해줘. 코드/에러/게임/문서 등 맥락을 짚어줘."
         )
         b64 = base64.b64encode(frame.data).decode()
-        resp = await self._client.messages.create(
+        client = self._auth.client()
+        resp = await client.messages.create(
             model=self.model,
             max_tokens=300,
+            system=self._auth.system(),
             messages=[
                 {
                     "role": "user",
@@ -54,7 +115,7 @@ class ClaudeVision:
 
 
 class ClaudeBrain:
-    SYSTEM = (
+    PERSONA = (
         "너는 사용자의 디스코드 화면공유를 실시간으로 함께 보는 AI 파트너야. "
         "화면 설명을 참고해 자연스러운 반말/존댓말은 사용자에 맞추고, 짧고 대화하듯 답해. "
         "화면을 못 봤으면 솔직히 말해."
@@ -62,7 +123,7 @@ class ClaudeBrain:
 
     def __init__(self, *, model: str = "claude-sonnet-4-5", api_key: str | None = None) -> None:
         self.model = model
-        self._client = _client(api_key)
+        self._auth = _Auth(api_key)
 
     async def respond(self, user_text, screen, history) -> Reply:
         msgs = []
@@ -71,10 +132,11 @@ class ClaudeBrain:
             msgs.append({"role": "assistant", "content": ai})
         screen_note = f"[지금 화면] {screen.text}\n\n" if screen else "[지금 화면] (아직 못 읽음)\n\n"
         msgs.append({"role": "user", "content": screen_note + user_text})
-        resp = await self._client.messages.create(
+        client = self._auth.client()
+        resp = await client.messages.create(
             model=self.model,
             max_tokens=400,
-            system=self.SYSTEM,
+            system=self._auth.system(self.PERSONA),
             messages=msgs,
         )
         text = "".join(b.text for b in resp.content if b.type == "text")