"""
Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>

AISBF - AI Service Broker Framework || AI Should Be Free

Claude Code OAuth2 provider handler.

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <https://www.gnu.org/licenses/>.

Why did the programmer quit his job? Because he didn't get arrays!
"""
import httpx
import asyncio
import time
import random
import os
import json
import shlex
import shutil
import sys
import tempfile
import logging as _logging
from typing import Dict, List, Optional, Union, Any, Tuple
from anthropic import Anthropic
from ..models import Model
from ..config import config
from .base import BaseProviderHandler, AnthropicFormatConverter, AISBF_DEBUG

# MCP server name for the tool shim; the CLI exposes its tools as
# mcp__<server>__<tool>, so this prefix is what we strip back off when
# converting a tool_use block into an OpenAI tool_call.
MCP_SHIM_SERVER = 'aisbf'
MCP_TOOL_PREFIX = f'mcp__{MCP_SHIM_SERVER}__'

# Each request starts a fresh CLI session, so earlier tool calls and their
# results can only be replayed as text.  Without this the model re-runs a tool
# it has already been given the answer for.
TOOL_REPLAY_INSTRUCTIONS = (
    'The conversation contains a transcript of tool use. A line of the form '
    '"[called tool NAME with arguments ...]" is a tool call that was already '
    'made on your behalf, and "[result of tool NAME: ...]" is the real value '
    'that tool returned. Treat those results as authoritative and current: use '
    'them to answer directly, and do not call a tool again to obtain a result '
    'that is already present in the transcript. Only call a tool when you need '
    'information the transcript does not already contain.'
)


class ClaudeCliSessionManager:
    """
    Manages per-user temporary config directories for the claude CLI subprocess.

    Each (user_id, provider_id) pair gets its own temp dir containing
    settings.json and .credentials.json.  The dir is reused across requests
    and cleaned up after 10 minutes of idle time.
    """

    _sessions: Dict[Tuple, Dict] = {}
    _cleanup_task: Optional[asyncio.Task] = None
    _lock: Optional[asyncio.Lock] = None
    _IDLE_TIMEOUT = 600  # 10 minutes

    @classmethod
    def _get_lock(cls) -> asyncio.Lock:
        if cls._lock is None:
            cls._lock = asyncio.Lock()
        return cls._lock

    @classmethod
    async def get_config_dir(cls, user_id, provider_id: str, cli_credentials: dict) -> str:
        """Return (creating if needed) the temp config dir for this user/provider."""
        logger = _logging.getLogger(__name__)
        key = (user_id, provider_id)

        async with cls._get_lock():
            if key in cls._sessions:
                session = cls._sessions[key]
                if os.path.exists(session['config_dir']):
                    session['last_used'] = time.time()
                    logger.debug(f"ClaudeCliMode: reusing session dir for {key}")
                    return session['config_dir']
                del cls._sessions[key]

            temp_dir = tempfile.mkdtemp(prefix='aisbf_claude_cli_')

            # settings.json – prevents interactive onboarding hang
            with open(os.path.join(temp_dir, 'settings.json'), 'w') as fh:
                json.dump({
                    'hasCompletedOnboarding': True,
                    'telemetry': 'off',
                    'theme': 'dark',
                }, fh, indent=2)

            # .credentials.json – the user's Claude CLI OAuth tokens
            with open(os.path.join(temp_dir, '.credentials.json'), 'w') as fh:
                json.dump(cli_credentials, fh, indent=2)

            cls._sessions[key] = {'config_dir': temp_dir, 'last_used': time.time()}
            logger.info(f"ClaudeCliMode: created session dir {temp_dir} for {key}")
            cls._ensure_cleanup_task()
            return temp_dir

    @classmethod
    def _ensure_cleanup_task(cls):
        try:
            loop = asyncio.get_event_loop()
            if cls._cleanup_task is None or cls._cleanup_task.done():
                cls._cleanup_task = loop.create_task(cls._cleanup_loop())
        except RuntimeError:
            pass

    @classmethod
    async def _cleanup_loop(cls):
        logger = _logging.getLogger(__name__)
        while True:
            await asyncio.sleep(60)
            async with cls._get_lock():
                if not cls._sessions:
                    cls._cleanup_task = None
                    return
                now = time.time()
                expired = [k for k, s in cls._sessions.items()
                           if now - s['last_used'] > cls._IDLE_TIMEOUT]
                for key in expired:
                    session = cls._sessions.pop(key)
                    try:
                        shutil.rmtree(session['config_dir'], ignore_errors=True)
                        logger.info(
                            f"ClaudeCliMode: cleaned up idle session "
                            f"{session['config_dir']} for {key}"
                        )
                    except Exception as exc:
                        logger.warning(
                            f"ClaudeCliMode: cleanup error for "
                            f"{session['config_dir']}: {exc}"
                        )


class ClaudeProviderHandler(BaseProviderHandler):
    """
    Handler for Claude Code OAuth2 integration using Anthropic SDK.
    
    This handler uses OAuth2 authentication to access Claude models through
    the official Anthropic Python SDK. OAuth2 access tokens are passed as
    the api_key parameter to the SDK, which handles proper message formatting,
    retries, and streaming.
    
    For admin users (user_id=None), credentials are loaded from file.
    For non-admin users, credentials are loaded from the database.
    """
    
    # NOTE: OAuth2 API uses its own model naming scheme that differs from standard Anthropic API
    
    def __init__(self, provider_id: str, api_key: Optional[str] = None, user_id: Optional[int] = None, provider_config: Optional[Any] = None):
        super().__init__(provider_id, api_key, user_id=user_id)
        if provider_config is not None:
            # Use provider config passed from factory (user-specific config)
            self.provider_config = provider_config
        else:
            # Fallback to global config
            self.provider_config = config.get_provider(provider_id)
        
        # Get credentials file path from config
        if isinstance(self.provider_config, dict):
            claude_config = self.provider_config.get('claude_config')
        else:
            claude_config = getattr(self.provider_config, 'claude_config', None)
        # Per-provider default so multiple admin claude providers don't share a file
        default_creds = BaseProviderHandler.build_credentials_file('claude', provider_id)
        credentials_file = default_creds
        if claude_config and isinstance(claude_config, dict):
            credentials_file = claude_config.get('credentials_file') or default_creds

        # Only the ONE config admin (user_id=None from aisbf.json) uses file-based credentials
        # All other users (including database admins with user_id) use database credentials
        if user_id is not None:
            self.auth = self._load_auth_from_db(provider_id, credentials_file)
        else:
            # Config admin (from aisbf.json): use file-based credentials
            from ..auth.claude import ClaudeAuth
            self.auth = ClaudeAuth(credentials_file=credentials_file)
        
        # HTTP client for direct API requests (OAuth2 requires direct HTTP, not SDK)
        self.client = httpx.AsyncClient(timeout=httpx.Timeout(300.0, connect=30.0))
        
        # Streaming idle watchdog configuration (Phase 1.3)
        self.stream_idle_timeout = 90.0  # seconds - matches vendors/claude
        
        # Cache token tracking for analytics (Phase 2.3)
        self.cache_stats = {
            'cache_hits': 0,
            'cache_misses': 0,
            'cache_tokens_read': 0,
            'cache_tokens_created': 0,
            'total_requests': 0,
        }
        
        # Session management for quota tracking
        self.session_state = {
            'initialized': False,
            'session_id': None,
            'device_id': None,
            'account_uuid': None,
            'organization_id': None,
            'last_initialized': None,
            'quota_5h_reset': None,
            'quota_5h_utilization': None,
            'quota_7d_reset': None,
            'quota_7d_utilization': None,
            'representative_claim': None,
            'status': None,
            'session_timeout': 3600,  # 1 hour session timeout
        }
        
        # Initialize persistent identifiers for metadata
        self._init_session_identifiers()

    def _init_session_identifiers(self):
        """Initialize persistent session identifiers (device_id, account_uuid)."""
        import uuid
        import hashlib
        if not self.session_state.get('device_id'):
            device_seed = f"{self.provider_id}-{time.time()}"
            self.session_state['device_id'] = hashlib.sha256(device_seed.encode()).hexdigest()
        if not self.session_state.get('account_uuid'):
            account_id = self.auth.get_account_id() if hasattr(self.auth, 'get_account_id') else None
            self.session_state['account_uuid'] = account_id if account_id else str(uuid.uuid4())

    def _get_api_token(self) -> Optional[str]:
        """Return the configured API token (api_key), or None if using OAuth2."""
        return self.api_key or None

    def _load_auth_from_db(self, provider_id: str, credentials_file: str):
        """
        Load OAuth2 credentials:
        - Admin users (user_id=None): ONLY load from file
        - Regular users: ONLY load from database, NO file fallback
        """
        from ..auth.claude import ClaudeAuth
        import logging
        
        if self.user_id is None:
            # Admin user: ONLY use file-based credentials
            logging.getLogger(__name__).info(f"ClaudeProviderHandler: Admin user, loading credentials from file: {credentials_file}")
            return ClaudeAuth(credentials_file=credentials_file)
        
        # Regular user: ONLY use database credentials, NO file fallback
        try:
            from ..database import DatabaseRegistry
            db = DatabaseRegistry.get_config_database()
            if db:
                db_creds = db.get_user_oauth2_credentials(
                    user_id=self.user_id,
                    provider_id=provider_id,
                    auth_type='claude_oauth2'
                )
                if db_creds and db_creds.get('credentials'):
                    # Create auth instance with skip_initial_load=True to avoid file read
                    # Pass save callback to save credentials back to database
                     auth = ClaudeAuth(
                         credentials_file=credentials_file, 
                         skip_initial_load=True,
                         save_callback=lambda creds: self._save_auth_to_db(creds)
                     )
                     # Set tokens directly from database, accepting either our
                     # own schema or a CLI credentials blob stored verbatim.
                     auth.tokens = ClaudeAuth.normalize_tokens(
                         db_creds['credentials'].get('tokens', {})
                     )
                     # Add expires_at if missing (for existing credentials saved before fix)
                     if auth.tokens and 'expires_at' not in auth.tokens and 'expires_in' in auth.tokens:
                         import time
                         auth.tokens['expires_at'] = time.time() + auth.tokens.get('expires_in', 3600)
                     import logging
                     logging.getLogger(__name__).info(f"ClaudeProviderHandler: Loaded credentials from database for user {self.user_id}")
                     return auth
        except Exception as e:
            logging.getLogger(__name__).warning(f"ClaudeProviderHandler: Failed to load credentials from database: {e}")
        
        # For regular users, NO file fallback - return empty auth instance
        logging.getLogger(__name__).info(f"ClaudeProviderHandler: No database credentials found for user {self.user_id}, returning unauthenticated instance")
        return ClaudeAuth(credentials_file=credentials_file, skip_initial_load=True)

    # ------------------------------------------------------------------ #
    # Claude CLI mode helpers                                              #
    # ------------------------------------------------------------------ #

    def _get_cli_credentials(self) -> Optional[dict]:
        """
        Return the Claude CLI .credentials.json content for this user/provider,
        or None if CLI credentials are not configured.

        Priority order:
        1. Explicit CLI credentials file (admin) or uploaded CLI credentials (DB user)
        2. If claude_config.use_cli_mode is true, derive from existing OAuth2 tokens
        """
        logger = _logging.getLogger(__name__)

        if isinstance(self.provider_config, dict):
            claude_cfg = self.provider_config.get('claude_config', {}) or {}
        else:
            claude_cfg = getattr(self.provider_config, 'claude_config', {}) or {}
        use_cli_mode = bool(claude_cfg.get('use_cli_mode')) if isinstance(claude_cfg, dict) else False

        if self.user_id is None:
            # ── Config admin ──────────────────────────────────────────────
            cli_file = claude_cfg.get('cli_credentials_file') if isinstance(claude_cfg, dict) else None
            if cli_file:
                expanded = os.path.expanduser(cli_file)
                if not os.path.exists(expanded):
                    logger.warning(f"ClaudeCliMode: CLI credentials file not found: {expanded}")
                else:
                    try:
                        with open(expanded) as fh:
                            return json.load(fh)
                    except Exception as exc:
                        logger.warning(f"ClaudeCliMode: failed to read CLI credentials file: {exc}")

            # Fall back: derive from existing OAuth2 tokens when use_cli_mode is set
            if use_cli_mode and self.auth and self.auth.tokens:
                logger.info("ClaudeCliMode: building CLI credentials from existing OAuth2 tokens (admin)")
                return self._oauth_tokens_to_cli_credentials(self.auth.tokens)

            return None

        else:
            # ── DB user ───────────────────────────────────────────────────
            try:
                from ..database import DatabaseRegistry
                db = DatabaseRegistry.get_config_database()
                if db:
                    # 1. Check for explicit uploaded CLI credentials
                    row = db.get_user_oauth2_credentials(
                        user_id=self.user_id,
                        provider_id=self.provider_id,
                        auth_type='claude_cli_credentials',
                    )
                    if row and row.get('credentials'):
                        return row['credentials'].get('credentials')

                    # 2. Derive from existing OAuth2 tokens when use_cli_mode is set
                    if use_cli_mode:
                        oauth_row = db.get_user_oauth2_credentials(
                            user_id=self.user_id,
                            provider_id=self.provider_id,
                            auth_type='claude_oauth2',
                        )
                        if oauth_row and oauth_row.get('credentials'):
                            tokens = oauth_row['credentials'].get('tokens', {})
                            if tokens:
                                logger.info(
                                    f"ClaudeCliMode: building CLI credentials from "
                                    f"OAuth2 tokens for user {self.user_id}"
                                )
                                return self._oauth_tokens_to_cli_credentials(tokens)
            except Exception as exc:
                logger.warning(f"ClaudeCliMode: failed to load credentials: {exc}")
            return None

    def _messages_to_cli_prompt(self, messages: List[Dict],
                                 tools: Optional[List[Dict]] = None) -> Tuple[str, str]:
        """
        Convert an OpenAI-style messages list into a
        ``(system_prompt, conversation_prompt)`` pair for the claude CLI.

        System messages are returned separately so the caller can pass them via
        the CLI's ``--system-prompt`` flag, which the CLI applies as a real
        system prompt.  Inlining them into the user turn (the previous
        behaviour) left them as ordinary user text that the model could ignore.

        Tool *definitions* are not rendered here — the MCP shim registers them
        with the CLI so the model can call them for real.  Past tool calls and
        their results are replayed as text: each request starts a fresh CLI
        session (--no-session-persistence), so without the replay the model
        would never see the result and would call the same tool forever.
        """
        system_parts: List[str] = []
        turn_parts: List[str] = []
        # tool_call_id -> function name, so a tool result can name its call
        call_names: Dict[str, str] = {}
        saw_tool_result = False

        for msg in messages:
            role = msg.get('role', '')
            content = msg.get('content', '')

            if isinstance(content, list):
                fragments = []
                for block in content:
                    if isinstance(block, dict) and block.get('type') == 'text':
                        fragments.append(block.get('text', ''))
                    elif isinstance(block, str):
                        fragments.append(block)
                content = '\n'.join(fragments)
            elif content is None:
                content = ''
            elif not isinstance(content, str):
                content = str(content)

            if role == 'system':
                system_parts.append(content.strip())
            elif role == 'user':
                turn_parts.append(f'Human: {content}')
            elif role == 'assistant':
                tool_calls = msg.get('tool_calls') or []
                rendered = [content] if content else []
                for call in tool_calls:
                    fn = call.get('function', {}) or {}
                    name = fn.get('name', '')
                    call_names[call.get('id', '')] = name
                    rendered.append(
                        f'[called tool {name} with arguments {fn.get("arguments", "{}")}]'
                    )
                if rendered:
                    turn_parts.append('Assistant: ' + '\n'.join(rendered))
            elif role == 'tool':
                name = call_names.get(msg.get('tool_call_id', ''), '') or msg.get('name', '')
                label = f'tool {name}' if name else 'tool'
                turn_parts.append(f'[result of {label}: {content}]')
                saw_tool_result = True

        if saw_tool_result:
            # Replayed results are only text, so the model treats them as
            # hearsay and calls the tool again unless told otherwise.
            system_parts.append(TOOL_REPLAY_INSTRUCTIONS)

        return '\n\n'.join(system_parts), '\n\n'.join(turn_parts)

    @staticmethod
    def _strip_mcp_prefix(name: str) -> str:
        """
        Map a CLI tool name back to the name the caller asked for.

        The CLI namespaces MCP tools as mcp__<server>__<tool>; the client only
        knows the bare function name it sent us.
        """
        if name.startswith(MCP_TOOL_PREFIX):
            return name[len(MCP_TOOL_PREFIX):]
        return name

    @staticmethod
    def _write_mcp_shim_config(tools: List[Dict], workdir: str) -> str:
        """
        Write the tool definitions and an --mcp-config pointing at the shim.

        Returns the path of the MCP config file.  Caller owns ``workdir`` and is
        responsible for removing it.
        """
        tools_path = os.path.join(workdir, 'tools.json')
        with open(tools_path, 'w') as fh:
            json.dump(tools, fh)

        shim_path = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                                 'claude_mcp_shim.py')
        config = {
            'mcpServers': {
                MCP_SHIM_SERVER: {
                    'command': sys.executable,
                    'args': [shim_path],
                    'env': {'AISBF_MCP_TOOLS_FILE': tools_path},
                    # Without this the CLI connects --mcp-config servers "fully
                    # async (nonblocking)" and starts the turn without waiting.
                    # Losing that race registers zero tools, and the model then
                    # describes tool calls in prose ("[Tool: list_dir]") instead
                    # of emitting tool_use — a silent wrong answer, not an error.
                    # alwaysLoad puts the shim on the CLI's blocking path, so
                    # the tools are always registered before the first request.
                    'alwaysLoad': True,
                },
            },
        }
        config_path = os.path.join(workdir, 'mcp.json')
        with open(config_path, 'w') as fh:
            json.dump(config, fh)
        return config_path

    @staticmethod
    def _build_cli_cmd(model: str, system_prompt: str = '',
                       mcp_config: str = '') -> List[str]:
        """
        Build the claude CLI argv.

        The flag set is deliberate and each entry matters:
          -p                          non-interactive print mode
          --output-format stream-json machine-readable event stream
          --input-format stream-json  prompt is fed as JSON on stdin (see below)
          --permission-prompt-tool stdio
          --verbose                   required for full stream-json events
          --dangerously-skip-permissions
          --tools ''                  disable every built-in tool.  MCP tools are
                                      registered separately and survive this.
          --disallowedTools 'mcp__*'  keep the host's MCP servers out of the
                                      broker's session

        With --input-format stream-json the CLI ignores any prompt passed as an
        argv positional; it is only read from stdin.

        When ``mcp_config`` is set the caller's tools are served by the MCP shim
        and --disallowedTools 'mcp__*' must be dropped: it is a blanket deny that
        also blocks the shim (deny beats --allowedTools, so there is no way to
        keep it and still expose the tools).  --strict-mcp-config preserves the
        isolation instead by loading *only* our config and ignoring every MCP
        server configured on the host.
        """
        cmd = [
            'stdbuf', '-oL',
            'claude', '-p',
            '--output-format', 'stream-json',
            '--input-format', 'stream-json',
            '--permission-prompt-tool', 'stdio',
            '--verbose',
            '--dangerously-skip-permissions',
            '--tools', '',
            # Emits token-level deltas for streaming, and — needed by both paths
            # — the message_delta event whose stop_reason marks the end of a
            # tool-calling message.  Without it there is no reliable signal that
            # every tool_use in a turn has arrived.
            '--include-partial-messages',
            '--no-session-persistence',
        ]
        if mcp_config:
            cmd += ['--mcp-config', mcp_config, '--strict-mcp-config']
        else:
            cmd += ['--disallowedTools', 'mcp__*']
        if system_prompt:
            cmd += ['--system-prompt', system_prompt]
        if model:
            cmd += ['--model', model]
        return cmd

    @staticmethod
    def _cli_stdin_message(prompt: str) -> str:
        """
        Frame a prompt as a stream-json stdin message.

        The CLI expects the Anthropic message envelope
        ``{"type": "user", "message": {"role": "user", "content": [...]}}``.
        Any other shape is silently discarded, which is why CLI mode previously
        produced no output at all.
        """
        return json.dumps({
            'type': 'user',
            'message': {
                'role': 'user',
                'content': [{'type': 'text', 'text': prompt}],
            },
        }) + '\n'

    async def _cli_discover_models(self, config_dir: str) -> List['Model']:
        """
        Ask the claude CLI which models it supports using --output-format json.
        Returns a list of Model objects parsed from the JSON result.
        The single-object JSON output format (not stream-json) is used here
        because it carries a `modelUsage` map with real contextWindow metadata,
        and the `result` text lists all models Claude knows about.
        """
        import re
        logger = _logging.getLogger(__name__)

        env = os.environ.copy()
        env['CLAUDE_CONFIG_DIR'] = config_dir
        env['CLAUDE_CODE_USE_KEYCHAIN'] = 'false'

        prompt = (
            "Which models are you compatible with? "
            "Give me only a JSON list without any other comment or word "
            "except for the list of the model IDs."
        )
        # Plain question, no tools needed: disable the built-ins and keep the
        # host's MCP servers out so discovery stays fast and side-effect free.
        cmd = [
            'claude', '-p', prompt,
            '--output-format', 'json',
            '--dangerously-skip-permissions',
            '--tools', '',
            '--disallowedTools', 'mcp__*',
            '--no-session-persistence',
        ]

        logger.info(
            "ClaudeCliMode: model discovery subprocess\n"
            f"  Replicate with: CLAUDE_CONFIG_DIR={config_dir} CLAUDE_CODE_USE_KEYCHAIN=false "
            + ' '.join(shlex.quote(c) for c in cmd)
        )

        process = await asyncio.create_subprocess_exec(
            *cmd,
            env=env,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        try:
            stdout_bytes, stderr_bytes = await asyncio.wait_for(
                process.communicate(), timeout=60.0
            )
        except asyncio.TimeoutError:
            logger.error("ClaudeCliMode: model discovery subprocess timed out")
            process.kill()
            await process.wait()
            return []

        if stderr_bytes:
            logger.debug(
                f"ClaudeCliMode: discovery stderr:\n"
                f"{stderr_bytes.decode('utf-8', errors='replace')[:2000]}"
            )

        stdout_str = stdout_bytes.decode('utf-8', errors='replace').strip()
        logger.debug(f"ClaudeCliMode: discovery raw output: {stdout_str[:1000]}")

        if not stdout_str:
            logger.warning("ClaudeCliMode: model discovery returned empty output")
            return []

        try:
            data = json.loads(stdout_str)
        except json.JSONDecodeError as e:
            logger.warning(f"ClaudeCliMode: model discovery JSON parse error: {e}")
            return []

        if data.get('is_error') or data.get('subtype') != 'success':
            logger.warning(
                f"ClaudeCliMode: model discovery error: {data.get('result', '')[:200]}"
            )
            return []

        # modelUsage keys → real metadata (contextWindow, maxOutputTokens)
        # Note: only models actually invoked in this call appear here; haiku is
        # used for internal routing so it shows up even though we didn't ask for it.
        model_usage: dict = data.get('modelUsage', {})

        # result text contains the JSON list we asked for, possibly wrapped in
        # a markdown code fence like ```json\n[...]\n```
        result_text: str = data.get('result', '')
        logger.info(f"ClaudeCliMode: discovery result: {result_text!r}")

        # Parse the JSON array from result (strip code fences if present)
        json_match = re.search(r'\[[\s\S]*?\]', result_text)
        result_ids: set = set()
        if json_match:
            try:
                parsed = json.loads(json_match.group())
                if isinstance(parsed, list):
                    result_ids = {m for m in parsed if isinstance(m, str) and m.startswith('claude-')}
            except json.JSONDecodeError:
                pass

        # Fall back to regex scan of the result text if JSON parse failed
        if not result_ids:
            result_ids = set(re.findall(r'claude-[a-z0-9][a-z0-9.\-]*[a-z0-9]', result_text))

        logger.info(f"ClaudeCliMode: model IDs from result: {sorted(result_ids)}")
        logger.info(f"ClaudeCliMode: model IDs from modelUsage: {sorted(model_usage.keys())}")

        # Known context window overrides — avoids a costly second prompt.
        # modelUsage carries real values for models used in this call; for the
        # rest we apply these known constants rather than querying Claude again.
        _known_context: dict = {
            'claude-opus-4-7': 1000000,
        }

        # Union: result_ids is the authoritative list; modelUsage adds metadata
        all_ids = result_ids | set(model_usage.keys())
        if not all_ids:
            return []

        models = []
        for mid in sorted(all_ids):
            usage_meta = model_usage.get(mid, {})
            context_size = (
                usage_meta.get('contextWindow')
                or _known_context.get(mid)
                or 200000
            )
            max_output = usage_meta.get('maxOutputTokens')
            m = Model(
                id=mid,
                name=mid,
                provider_id=self.provider_id,
                context_size=context_size,
                context_length=context_size,
            )
            if max_output:
                m.max_output_tokens = max_output
            models.append(m)

        return models

    async def _handle_cli_streaming_request(self, prompt: str, model: str, config_dir: str,
                                             system_prompt: str = '',
                                             tools: Optional[List[Dict]] = None):
        """
        Spawn a claude CLI subprocess, stream its JSON output, and yield
        OpenAI-compatible SSE chunks.  Multiple parallel calls each get their
        own subprocess; the config_dir is shared (read-only at runtime).

        When ``tools`` are supplied they are served to the CLI by the MCP shim,
        which makes the model emit real tool_use blocks.  Those are emitted as
        OpenAI tool_calls and the turn then ends, because the client — not the
        broker — executes tools (see claude_mcp_shim for the details).
        """
        logger = _logging.getLogger(__name__)
        clean_model = model.split('/')[-1] if '/' in model else model

        env = os.environ.copy()
        env['CLAUDE_CONFIG_DIR'] = config_dir
        env['CLAUDE_CODE_USE_KEYCHAIN'] = 'false'

        mcp_workdir = tempfile.mkdtemp(prefix='aisbf_claude_mcp_') if tools else ''
        mcp_config = self._write_mcp_shim_config(tools, mcp_workdir) if tools else ''

        cmd = self._build_cli_cmd(clean_model, system_prompt, mcp_config=mcp_config)
        input_msg = self._cli_stdin_message(prompt)

        # Log a shell-replicable command for debugging
        cmd_str = ' '.join(shlex.quote(c) for c in cmd)
        logger.info(
            f"ClaudeCliMode: launching subprocess model={clean_model} dir={config_dir}\n"
            f"  Replicate with: CLAUDE_CONFIG_DIR={config_dir} CLAUDE_CODE_USE_KEYCHAIN=false "
            f"{cmd_str} <<'EOF'\n{input_msg.strip()}\nEOF"
        )

        process = await asyncio.create_subprocess_exec(
            *cmd,
            env=env,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        process.stdin.write(input_msg.encode())
        await process.stdin.drain()
        process.stdin.close()

        completion_id = f'chatcmpl-cli-{int(time.time())}'
        created_time = int(time.time())
        first_chunk = True

        # State for accumulating tool_use blocks
        # { block_index: {"id": ..., "name": ..., "arguments": ""} }
        tool_blocks: dict = {}
        tool_header_sent: set = set()
        # tool_use ids already emitted incrementally via input_json_delta
        streamed_tool_ids: set = set()
        cli_prev_text_len: int = 0

        try:
            while True:
                try:
                    raw = await asyncio.wait_for(process.stdout.readline(), timeout=120.0)
                except asyncio.TimeoutError:
                    logger.error("ClaudeCliMode: subprocess read timeout (120 s)")
                    break

                if not raw:
                    break

                line_str = raw.decode('utf-8', errors='replace').strip()
                if not line_str:
                    continue

                logger.debug(f"ClaudeCliMode: raw event: {line_str}")

                try:
                    data = json.loads(line_str)
                except json.JSONDecodeError:
                    logger.debug(f"ClaudeCliMode: non-JSON line: {line_str}")
                    continue

                # The CLI wraps raw Anthropic stream events in a stream_event
                # envelope: {"type": "stream_event", "event": {...}}.  Unwrap it
                # so the content_block_* / message_stop branches below see the
                # real event type.
                if data.get('type') == 'stream_event':
                    data = data.get('event', {})

                event_type = data.get('type')

                if event_type == 'content_block_start':
                    cb = data.get('content_block', {})
                    if cb.get('type') == 'tool_use':
                        idx = data.get('index', 0)
                        tc_id = cb.get('id', f'call_{idx}')
                        tool_blocks[idx] = {
                            'id': tc_id,
                            'name': self._strip_mcp_prefix(cb.get('name', '')),
                            'arguments': '',
                            # OpenAI tool_calls are indexed by position in the
                            # tool_calls array; `idx` is an Anthropic content
                            # block index that also counts text/thinking blocks,
                            # so emitting it directly would leave holes that
                            # break client-side accumulation.
                            'seq': len(tool_blocks),
                        }
                        # Remember the id so the `assistant` event below does not
                        # re-emit arguments already streamed.
                        streamed_tool_ids.add(tc_id)
                        logger.debug(f"ClaudeCliMode: tool_use block started idx={idx} seq={tool_blocks[idx]['seq']} name={cb.get('name')}")

                elif event_type == 'content_block_delta':
                    delta = data.get('delta', {})
                    idx = data.get('index', 0)

                    if delta.get('type') == 'text_delta':
                        text = delta.get('text', '')
                        if not text:
                            continue

                        # Count text streamed here so the cumulative `assistant`
                        # event below does not re-emit it.
                        cli_prev_text_len += len(text)

                        if first_chunk:
                            yield f'data: {json.dumps({"id": completion_id, "object": "chat.completion.chunk", "created": created_time, "model": f"{self.provider_id}/{clean_model}", "choices": [{"index": 0, "delta": {"role": "assistant", "content": ""}, "finish_reason": None}]})}\n\n'
                            first_chunk = False

                        yield f'data: {json.dumps({"id": completion_id, "object": "chat.completion.chunk", "created": created_time, "model": f"{self.provider_id}/{clean_model}", "choices": [{"index": 0, "delta": {"content": text}, "finish_reason": None}]})}\n\n'

                    elif delta.get('type') == 'input_json_delta' and idx in tool_blocks:
                        partial = delta.get('partial_json', '')
                        tool_blocks[idx]['arguments'] += partial
                        seq = tool_blocks[idx]['seq']

                        # Emit streaming tool_calls delta
                        if idx not in tool_header_sent:
                            tool_header_sent.add(idx)
                            if first_chunk:
                                yield f'data: {json.dumps({"id": completion_id, "object": "chat.completion.chunk", "created": created_time, "model": f"{self.provider_id}/{clean_model}", "choices": [{"index": 0, "delta": {"role": "assistant", "content": None, "tool_calls": [{"index": seq, "id": tool_blocks[idx]["id"], "type": "function", "function": {"name": tool_blocks[idx]["name"], "arguments": ""}}]}, "finish_reason": None}]})}\n\n'
                                first_chunk = False
                            else:
                                yield f'data: {json.dumps({"id": completion_id, "object": "chat.completion.chunk", "created": created_time, "model": f"{self.provider_id}/{clean_model}", "choices": [{"index": 0, "delta": {"tool_calls": [{"index": seq, "id": tool_blocks[idx]["id"], "type": "function", "function": {"name": tool_blocks[idx]["name"], "arguments": ""}}]}, "finish_reason": None}]})}\n\n'

                        if partial:
                            yield f'data: {json.dumps({"id": completion_id, "object": "chat.completion.chunk", "created": created_time, "model": f"{self.provider_id}/{clean_model}", "choices": [{"index": 0, "delta": {"tool_calls": [{"index": seq, "function": {"arguments": partial}}]}, "finish_reason": None}]})}\n\n'

                elif event_type == 'assistant':
                    # Claude CLI stream-json format: partial or final assistant message
                    msg = data.get('message', {})
                    last_text = ''
                    for block in msg.get('content', []):
                        if not isinstance(block, dict):
                            continue
                        btype = block.get('type')
                        if btype == 'text':
                            last_text += block.get('text', '')
                        elif btype == 'tool_use':
                            # Fallback for a tool_use that never came through as
                            # content_block deltas; emits the complete input in
                            # one chunk.
                            tc_id = block.get('id', f'call_{len(tool_blocks)}')
                            tool_name = self._strip_mcp_prefix(block.get('name', ''))
                            if tc_id not in streamed_tool_ids and tc_id not in tool_header_sent:
                                tool_header_sent.add(tc_id)
                                streamed_tool_ids.add(tc_id)
                                # Key by a synthetic block index that cannot
                                # collide with a real content block index.
                                block_key = f'assistant:{tc_id}'
                                seq = len(tool_blocks)
                                tool_blocks[block_key] = {
                                    'id': tc_id,
                                    'name': tool_name,
                                    'arguments': json.dumps(block.get('input', {}), ensure_ascii=False),
                                    'seq': seq,
                                }
                                role_delta = {'role': 'assistant', 'content': None} if first_chunk else {}
                                first_chunk = False
                                yield f'data: {json.dumps({"id": completion_id, "object": "chat.completion.chunk", "created": created_time, "model": f"{self.provider_id}/{clean_model}", "choices": [{"index": 0, "delta": {**role_delta, "tool_calls": [{"index": seq, "id": tc_id, "type": "function", "function": {"name": tool_name, "arguments": tool_blocks[block_key]["arguments"]}}]}, "finish_reason": None}]})}\n\n'
                    if last_text:
                        # Content is cumulative; emit only new characters
                        new_text = last_text[cli_prev_text_len:]
                        cli_prev_text_len = len(last_text)
                        if new_text:
                            if first_chunk:
                                yield f'data: {json.dumps({"id": completion_id, "object": "chat.completion.chunk", "created": created_time, "model": f"{self.provider_id}/{clean_model}", "choices": [{"index": 0, "delta": {"role": "assistant", "content": ""}, "finish_reason": None}]})}\n\n'
                                first_chunk = False
                            yield f'data: {json.dumps({"id": completion_id, "object": "chat.completion.chunk", "created": created_time, "model": f"{self.provider_id}/{clean_model}", "choices": [{"index": 0, "delta": {"content": new_text}, "finish_reason": None}]})}\n\n'

                elif event_type == 'message_delta':
                    # stop_reason == 'tool_use' means every tool_use block in
                    # this message has arrived.  This — not the first tool_use —
                    # is where the turn ends: the CLI emits one assistant event
                    # per content block, so breaking earlier would drop any
                    # sibling calls the model made in parallel.
                    if data.get('delta', {}).get('stop_reason') == 'tool_use':
                        logger.info(
                            f"ClaudeCliMode: {len(tool_blocks)} tool call(s) emitted, "
                            "ending turn for client execution"
                        )
                        break

                elif event_type == 'result':
                    result_text = data.get('result', '')
                    logger.debug(f"ClaudeCliMode: result event, is_error={data.get('is_error')}, text_len={len(result_text)}")
                    # Only emit via result if we haven't already streamed content via other events
                    if result_text and first_chunk:
                        yield f'data: {json.dumps({"id": completion_id, "object": "chat.completion.chunk", "created": created_time, "model": f"{self.provider_id}/{clean_model}", "choices": [{"index": 0, "delta": {"role": "assistant", "content": ""}, "finish_reason": None}]})}\n\n'
                        yield f'data: {json.dumps({"id": completion_id, "object": "chat.completion.chunk", "created": created_time, "model": f"{self.provider_id}/{clean_model}", "choices": [{"index": 0, "delta": {"content": result_text}, "finish_reason": None}]})}\n\n'
                        first_chunk = False
                    break

                elif event_type == 'message_stop':
                    # End of one assistant message, not necessarily end of turn.
                    # The CLI's `result` event is the authoritative terminator,
                    # so keep reading until it arrives.
                    logger.debug("ClaudeCliMode: received message_stop")

                else:
                    logger.debug(f"ClaudeCliMode: unhandled event type={event_type}")

        except Exception as exc:
            logger.error(f"ClaudeCliMode: streaming error: {exc}", exc_info=True)
        finally:
            try:
                stderr_bytes = await asyncio.wait_for(process.stderr.read(), timeout=2.0)
                if stderr_bytes:
                    decoded = stderr_bytes.decode('utf-8', errors='replace')
                    logger.debug(f"ClaudeCliMode: stderr:\n{decoded[:2000]}")
            except Exception:
                pass
            try:
                process.terminate()
                await asyncio.wait_for(process.wait(), timeout=5.0)
            except Exception:
                try:
                    process.kill()
                except Exception:
                    pass
            if mcp_workdir:
                shutil.rmtree(mcp_workdir, ignore_errors=True)

        finish_reason = 'tool_calls' if tool_blocks else 'stop'
        yield f'data: {json.dumps({"id": completion_id, "object": "chat.completion.chunk", "created": created_time, "model": f"{self.provider_id}/{clean_model}", "choices": [{"index": 0, "delta": {}, "finish_reason": finish_reason}]})}\n\n'
        yield 'data: [DONE]\n\n'

    async def _handle_cli_request(self, prompt: str, model: str, config_dir: str,
                                   system_prompt: str = '',
                                   tools: Optional[List[Dict]] = None) -> dict:
        """
        Non-streaming CLI request.

        Uses the same stream-json transport as the streaming path and collects
        the final `result` event, rather than --output-format json.  The old
        implementation passed the caller's tool definitions to --tools, but that
        flag only selects built-in tools by name; a JSON blob there registers
        nothing.  Tools are served by the MCP shim instead.
        """
        logger = _logging.getLogger(__name__)
        clean_model = model.split('/')[-1] if '/' in model else model

        env = os.environ.copy()
        env['CLAUDE_CONFIG_DIR'] = config_dir
        env['CLAUDE_CODE_USE_KEYCHAIN'] = 'false'

        mcp_workdir = tempfile.mkdtemp(prefix='aisbf_claude_mcp_') if tools else ''
        mcp_config = self._write_mcp_shim_config(tools, mcp_workdir) if tools else ''

        cmd = self._build_cli_cmd(clean_model, system_prompt, mcp_config=mcp_config)
        input_msg = self._cli_stdin_message(prompt)

        logger.info(
            f"ClaudeCliMode: non-streaming subprocess model={clean_model} dir={config_dir}\n"
            f"  Replicate with: CLAUDE_CONFIG_DIR={config_dir} CLAUDE_CODE_USE_KEYCHAIN=false "
            + ' '.join(shlex.quote(c) for c in cmd)
            + f" <<'EOF'\n{input_msg.strip()}\nEOF"
        )

        process = await asyncio.create_subprocess_exec(
            *cmd,
            env=env,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        process.stdin.write(input_msg.encode())
        await process.stdin.drain()
        process.stdin.close()

        # stream-json output is newline-delimited JSON.  Read it incrementally
        # rather than via communicate(): a tool_use has to end the turn, and
        # waiting for exit would let the CLI call the shim and act on its
        # sentinel result.
        result_text = ''
        assistant_text = ''
        usage = {'prompt_tokens': 0, 'completion_tokens': 0, 'total_tokens': 0}
        tool_calls: List[Dict] = []
        seen_tool_ids: set = set()
        timed_out = False

        try:
            while True:
                try:
                    raw = await asyncio.wait_for(process.stdout.readline(), timeout=120.0)
                except asyncio.TimeoutError:
                    logger.error("ClaudeCliMode: non-streaming subprocess timed out")
                    timed_out = True
                    break
                if not raw:
                    break

                line = raw.decode('utf-8', errors='replace').strip()
                if not line:
                    continue
                try:
                    data = json.loads(line)
                except json.JSONDecodeError:
                    continue

                if data.get('type') == 'stream_event':
                    event = data.get('event', {})
                    # See the streaming path: stop_reason == 'tool_use' is the
                    # only reliable marker that every parallel tool_use in the
                    # message has arrived.
                    if (event.get('type') == 'message_delta'
                            and event.get('delta', {}).get('stop_reason') == 'tool_use'):
                        logger.info(
                            f"ClaudeCliMode: {len(tool_calls)} tool call(s) emitted, "
                            "ending turn for client execution"
                        )
                        break
                    continue

                dtype = data.get('type')
                if dtype == 'assistant':
                    # One assistant event per content block, so accumulate
                    # across events and dedupe by tool_use id.
                    for block in data.get('message', {}).get('content', []):
                        if not isinstance(block, dict):
                            continue
                        if block.get('type') == 'text':
                            assistant_text += block.get('text', '')
                        elif block.get('type') == 'tool_use':
                            tc_id = block.get('id', f'call_{len(tool_calls)}')
                            if tc_id in seen_tool_ids:
                                continue
                            seen_tool_ids.add(tc_id)
                            tool_calls.append({
                                'id': tc_id,
                                'type': 'function',
                                'function': {
                                    'name': self._strip_mcp_prefix(block.get('name', '')),
                                    'arguments': json.dumps(block.get('input', {}),
                                                            ensure_ascii=False),
                                },
                            })
                elif dtype == 'result':
                    if data.get('is_error'):
                        logger.warning(
                            f"ClaudeCliMode: CLI returned error: {str(data.get('result', ''))[:200]}"
                        )
                    result_text = data.get('result', '') or ''
                    raw_usage = data.get('usage', {}) or {}
                    prompt_tokens = (
                        raw_usage.get('input_tokens', 0)
                        + raw_usage.get('cache_read_input_tokens', 0)
                        + raw_usage.get('cache_creation_input_tokens', 0)
                    )
                    completion_tokens = raw_usage.get('output_tokens', 0)
                    usage = {
                        'prompt_tokens': prompt_tokens,
                        'completion_tokens': completion_tokens,
                        'total_tokens': prompt_tokens + completion_tokens,
                    }
                    break
        finally:
            try:
                stderr_bytes = await asyncio.wait_for(process.stderr.read(), timeout=2.0)
                if stderr_bytes:
                    logger.debug(
                        f"ClaudeCliMode: stderr:\n"
                        f"{stderr_bytes.decode('utf-8', errors='replace')[:2000]}"
                    )
            except Exception:
                pass
            try:
                process.terminate()
                await asyncio.wait_for(process.wait(), timeout=5.0)
            except Exception:
                try:
                    process.kill()
                except Exception:
                    pass
            if mcp_workdir:
                shutil.rmtree(mcp_workdir, ignore_errors=True)

        if timed_out and not (result_text or assistant_text or tool_calls):
            result_text = 'Request timed out.'

        message: Dict[str, Any] = {'role': 'assistant'}
        if tool_calls:
            # A tool-calling turn carries no user-visible text.
            message['content'] = assistant_text or None
            message['tool_calls'] = tool_calls
            finish_reason = 'tool_calls'
        else:
            message['content'] = result_text or assistant_text
            finish_reason = 'stop'

        return {
            'id': f'chatcmpl-cli-{int(time.time())}',
            'object': 'chat.completion',
            'created': int(time.time()),
            'model': f'{self.provider_id}/{clean_model}',
            'choices': [{'index': 0, 'message': message, 'finish_reason': finish_reason}],
            'usage': usage,
        }

    @staticmethod
    def _oauth_tokens_to_cli_credentials(tokens: dict) -> dict:
        """
        Convert AISBF OAuth2 token dict to the Claude CLI .credentials.json schema:

        AISBF stores:  access_token, refresh_token, expires_at (seconds float), scope
        CLI expects:   claudeAiOauth.accessToken, .refreshToken, .expiresAt (ms int),
                       .scopes (list), .subscriptionType, .rateLimitTier
        """
        default_scopes = [
            'user:file_upload',
            'user:inference',
            'user:mcp_servers',
            'user:profile',
            'user:sessions:claude_code',
        ]
        raw_scope = tokens.get('scope', '')
        scopes = raw_scope.split() if raw_scope.strip() else default_scopes

        expires_at_sec = tokens.get('expires_at', 0)
        expires_at_ms = int(expires_at_sec * 1000) if expires_at_sec else 0

        return {
            'claudeAiOauth': {
                'accessToken': tokens.get('access_token', ''),
                'refreshToken': tokens.get('refresh_token', ''),
                'expiresAt': expires_at_ms,
                'scopes': scopes,
                'subscriptionType': tokens.get('subscription_type', 'pro'),
                'rateLimitTier': tokens.get('rate_limit_tier', 'default_claude_ai'),
            }
        }

    def _save_auth_to_db(self, credentials: dict):
        """Save OAuth2 credentials back to the database."""
        try:
            from aisbf.database import DatabaseRegistry
            db = DatabaseRegistry.get_config_database()
            if db and self.user_id is not None:
                db.save_provider_credentials(self.user_id, self.provider_id, credentials, auth_type='claude_oauth2')
        except Exception as e:
            import logging
            logging.getLogger(__name__).warning(f"ClaudeProviderHandler: Failed to save credentials to database: {e}")

    def validate_credentials(self) -> bool:
        """
        Validate Claude credentials.
        
        Checks OAuth2 tokens via self.auth.is_authenticated() or API token if configured.
        
        Returns:
            True if credentials are valid, False otherwise.
        """
        import logging
        logger = logging.getLogger(__name__)
        
        # Check if standard API token is configured (bypasses OAuth2)
        api_token = self._get_api_token()
        if api_token:
            logger.info(f"[{self.provider_id}] Claude using API token authentication")
            return True
        
        # OAuth2 mode: check if auth object exists and is authenticated
        if hasattr(self, 'auth') and self.auth:
            is_auth = self.auth.is_authenticated()
            if is_auth:
                logger.info(f"[{self.provider_id}] Claude OAuth2 credentials are valid")
            else:
                logger.error(f"[{self.provider_id}] Claude OAuth2 credentials are invalid or missing")
            return is_auth
        
        logger.error(f"[{self.provider_id}] No authentication mechanism configured for Claude")
        return False
    
    async def _initialize_session(self):
        """Initialize session by sending a quota request to get rate limit information."""
        import logging
        import json
        
        logger = logging.getLogger(__name__)
        logger.info("ClaudeProviderHandler: Initializing session for quota tracking")
        
        try:
            headers = await self._get_auth_headers(stream=False)
            
            payload = {
                'model': 'claude-haiku-4-5-20251001',
                'max_tokens': 1,
                'messages': [
                    {
                        'role': 'user',
                        'content': 'quota'
                    }
                ],
                'metadata': {
                    'user_id': json.dumps({
                        'device_id': self.session_state['device_id'],
                        'account_uuid': self.session_state['account_uuid'],
                        'session_id': self.session_state['session_id']
                    })
                }
            }
            
            api_url = 'https://api.anthropic.com/v1/messages?beta=true'
            response = await self.client.post(api_url, headers=headers, json=payload)
            
            if response.status_code == 200:
                headers_dict = dict(response.headers)
                
                self.session_state.update({
                    'initialized': True,
                    'last_initialized': time.time(),
                    'organization_id': headers_dict.get('anthropic-organization-id'),
                    'quota_5h_reset': headers_dict.get('anthropic-ratelimit-unified-5h-reset'),
                    'quota_5h_utilization': headers_dict.get('anthropic-ratelimit-unified-5h-utilization'),
                    'quota_7d_reset': headers_dict.get('anthropic-ratelimit-unified-7d-reset'),
                    'quota_7d_utilization': headers_dict.get('anthropic-ratelimit-unified-7d-utilization'),
                    'representative_claim': headers_dict.get('anthropic-ratelimit-unified-representative-claim'),
                    'status': headers_dict.get('anthropic-ratelimit-unified-status'),
                })
                
                logger.info(f"ClaudeProviderHandler: Session initialized successfully")
                logger.info(f"  Organization ID: {self.session_state['organization_id']}")
                logger.info(f"  5h utilization: {self.session_state['quota_5h_utilization']}")
                logger.info(f"  7d utilization: {self.session_state['quota_7d_utilization']}")
                logger.info(f"  Representative claim: {self.session_state['representative_claim']}")
                logger.info(f"  Status: {self.session_state['status']}")
                
                return True
            else:
                logger.warning(f"ClaudeProviderHandler: Session initialization failed: {response.status_code}")
                return False
                
        except Exception as e:
            logger.error(f"ClaudeProviderHandler: Session initialization error: {e}", exc_info=True)
            return False
    
    def _should_refresh_session(self) -> bool:
        """Check if session should be refreshed based on timeout or rate limit status."""
        if not self.session_state['initialized']:
            return True
        
        if self.session_state['last_initialized']:
            age = time.time() - self.session_state['last_initialized']
            if age > self.session_state['session_timeout']:
                return True
        
        if self.session_state['status'] != 'allowed':
            return True
        
        return False

    def normalize_usage_data(self, usage_data: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
        normalized = super().normalize_usage_data(usage_data)
        if not normalized or not isinstance(normalized, dict):
            return normalized

        utilization = normalized.get('quota_7d_utilization')
        if utilization is not None:
            try:
                util_float = float(str(utilization).rstrip('%'))
                normalized['free_tier'] = {
                    'limit': 100,
                    'used': max(util_float, 0.0),
                    'remaining': max(100.0 - util_float, 0.0),
                    'period': 'week',
                    'limit_type': 'percent',
                    'source': 'provider'
                }
            except (TypeError, ValueError):
                pass

        return normalized
    
    async def _ensure_session(self):
        """Ensure session is initialized and valid before making requests."""
        if self._should_refresh_session():
            import logging
            logger = logging.getLogger(__name__)
            logger.info("ClaudeProviderHandler: Session needs refresh, initializing...")
            await self._initialize_session()
    
    def _update_session_from_headers(self, headers: Dict):
        """Update session state from response headers."""
        import logging
        logger = logging.getLogger(__name__)
        
        if 'anthropic-ratelimit-unified-5h-utilization' in headers:
            old_util = self.session_state.get('quota_5h_utilization')
            new_util = headers.get('anthropic-ratelimit-unified-5h-utilization')
            
            self.session_state.update({
                'quota_5h_reset': headers.get('anthropic-ratelimit-unified-5h-reset'),
                'quota_5h_utilization': new_util,
                'quota_7d_reset': headers.get('anthropic-ratelimit-unified-7d-reset'),
                'quota_7d_utilization': headers.get('anthropic-ratelimit-unified-7d-utilization'),
                'representative_claim': headers.get('anthropic-ratelimit-unified-representative-claim'),
                'status': headers.get('anthropic-ratelimit-unified-status'),
            })
            
            if old_util != new_util:
                logger.debug(f"ClaudeProviderHandler: Quota utilization updated: {old_util} -> {new_util}")
    
    async def _get_sdk_client(self):
        """Get or create an Anthropic SDK client configured with OAuth2 auth token."""
        import logging
        logger = logging.getLogger(__name__)

        access_token = await self.auth.get_valid_token_with_refresh()
        
        if not access_token:
            logger.error("ClaudeProviderHandler: No OAuth2 access token available")
            raise Exception("Claude authentication required. Token refresh failed. Please re-authenticate via /dashboard/claude/auth/start")
        
        self._sdk_client = Anthropic(
            auth_token=access_token,
            max_retries=3,
            timeout=httpx.Timeout(300.0, connect=30.0),
        )
        
        logger.info("ClaudeProviderHandler: Created SDK client with OAuth2 auth token")
        return self._sdk_client
    
    async def _get_auth_headers(self, stream: bool = False):
        """Get HTTP headers for the API request.

        When an api_token is configured, returns standard Anthropic API key headers.
        Otherwise returns OAuth2 Bearer token headers matching the Claude CLI client.
        """
        import logging
        import uuid
        import platform
        logger = logging.getLogger(__name__)

        # ── Standard Anthropic API key mode ──────────────────────────────
        api_token = self._get_api_token()
        if api_token:
            headers = {
                'x-api-key': api_token,
                'anthropic-version': '2023-06-01',
                'content-type': 'application/json',
                'accept': 'text/event-stream' if stream else 'application/json',
            }
            logger.info("ClaudeProviderHandler: Using standard Anthropic API key authentication")
            return headers

        # ── OAuth2 Bearer token mode ──────────────────────────────────────
        access_token = await self.auth.get_valid_token_with_refresh()

        if not access_token:
            raise Exception("Claude authentication required. Token refresh failed. Please re-authenticate via /dashboard/claude/auth/start")

        if not self.session_state.get('session_id'):
            self.session_state['session_id'] = str(uuid.uuid4())

        session_id = self.session_state['session_id']
        request_id = str(uuid.uuid4())

        headers = {
            'accept': 'application/json',
            'anthropic-beta': 'oauth-2025-04-20,interleaved-thinking-2025-05-14,redact-thinking-2026-02-12,context-management-2025-06-27,prompt-caching-scope-2026-01-05,structured-outputs-2025-12-15',
            'anthropic-dangerous-direct-browser-access': 'true',
            'anthropic-version': '2023-06-01',
            'authorization': f'Bearer {access_token}',
            'content-type': 'application/json',
            'user-agent': 'claude-cli/99.0.0 (undefined, cli)',
            'x-app': 'cli',
            'x-claude-code-session-id': session_id,
            'x-client-request-id': request_id,
            'x-stainless-arch': platform.machine() or 'x64',
            'x-stainless-lang': 'js',
            'x-stainless-os': platform.system() or 'Linux',
            'x-stainless-package-version': '0.81.0',
            'x-stainless-retry-count': '0',
            'x-stainless-runtime': 'node',
            'x-stainless-runtime-version': 'v22.22.0',
            'x-stainless-timeout': '600',
        }

        if stream:
            headers['accept'] = 'text/event-stream'
            headers['accept-encoding'] = 'identity'
        else:
            headers['accept-encoding'] = 'gzip, deflate, br, zstd'

        logger.info("ClaudeProviderHandler: Created auth headers matching claude-cli client")
        logger.debug(f"ClaudeProviderHandler: Session ID: {session_id}, Request ID: {request_id}")

        import json
        logger.debug(f"ClaudeProviderHandler: Full headers: {json.dumps(headers, indent=2)}")
        return headers
    
    def _sanitize_tool_call_id(self, tool_call_id: str) -> str:
        """Sanitize tool call ID for Claude API compatibility."""
        import re
        sanitized = re.sub(r'[^a-zA-Z0-9_-]', '_', tool_call_id)
        return sanitized
    
    def _filter_empty_content(self, content: Union[str, List, None]) -> Union[str, List, None]:
        """Filter empty content from messages for Claude API compatibility."""
        if content is None:
            return None
        
        if isinstance(content, str):
            if content.strip() == "":
                return None
            return content
        
        if isinstance(content, list):
            filtered = []
            for block in content:
                if isinstance(block, dict):
                    block_type = block.get('type', '')
                    if block_type == 'text':
                        text = block.get('text', '')
                        if text and text.strip():
                            filtered.append(block)
                    else:
                        filtered.append(block)
                else:
                    filtered.append(block)
            
            if not filtered:
                return None
            return filtered
        
        return content
    
    def _apply_cache_control(self, anthropic_messages: List[Dict], enable_caching: bool = True) -> List[Dict]:
        """Apply ephemeral cache_control to messages for prompt caching."""
        if not enable_caching or not anthropic_messages:
            return anthropic_messages
        
        import logging
        logger = logging.getLogger(__name__)
        
        if len(anthropic_messages) < 4:
            logger.debug(f"ClaudeProviderHandler: Skipping cache control (only {len(anthropic_messages)} messages)")
            return anthropic_messages
        
        cache_indices = []
        
        for i in range(max(0, len(anthropic_messages) - 2), len(anthropic_messages)):
            cache_indices.append(i)
        
        for idx in cache_indices:
            msg = anthropic_messages[idx]
            content = msg.get('content')
            
            if isinstance(content, str):
                if content.strip():
                    msg['content'] = [
                        {
                            'type': 'text',
                            'text': content,
                            'cache_control': {'type': 'ephemeral'}
                        }
                    ]
                    logger.debug(f"ClaudeProviderHandler: Applied cache_control to message {idx} (string content)")
            elif isinstance(content, list) and content:
                last_block = content[-1]
                if isinstance(last_block, dict):
                    last_block['cache_control'] = {'type': 'ephemeral'}
                    logger.debug(f"ClaudeProviderHandler: Applied cache_control to message {idx} (list content)")
        
        logger.info(f"ClaudeProviderHandler: Applied cache_control to {len(cache_indices)} messages for prompt caching")
        return anthropic_messages
    
    def _validate_messages(self, messages: List[Dict]) -> List[Dict]:
        """Validate and normalize message roles for Claude API compatibility."""
        import logging
        logger = logging.getLogger(__name__)
        
        if not messages:
            return messages
        
        valid_roles = {'user', 'assistant', 'system', 'tool'}
        normalized = []
        issues_found = 0
        
        for i, msg in enumerate(messages):
            role = msg.get('role', '')
            content = msg.get('content', '')
            
            if role not in valid_roles:
                logger.warning(f"ClaudeProviderHandler: Unknown message role '{role}' at index {i}, treating as 'user'")
                msg['role'] = 'user'
                role = 'user'
                issues_found += 1
            
            if role == 'system' and i > 0:
                logger.warning(f"ClaudeProviderHandler: System message at index {i} (not at start), converting to user")
                msg['role'] = 'user'
                role = 'user'
                issues_found += 1
            
            if role == 'tool':
                tool_call_id = msg.get('tool_call_id') or msg.get('name')
                if not tool_call_id:
                    logger.warning(f"ClaudeProviderHandler: Tool message at index {i} missing tool_call_id, adding placeholder")
                    msg['tool_call_id'] = f"placeholder_{i}"
                    issues_found += 1
            
            if normalized:
                last_role = normalized[-1].get('role', '')
                
                if role == 'user' and last_role == 'user':
                    logger.debug(f"ClaudeProviderHandler: Inserting synthetic assistant message between consecutive user messages at index {i}")
                    normalized.append({
                        'role': 'assistant',
                        'content': '(empty)'
                    })
                    issues_found += 1
                
                elif role == 'assistant' and last_role == 'assistant':
                    logger.debug(f"ClaudeProviderHandler: Merging consecutive assistant messages at index {i}")
                    prev_content = normalized[-1].get('content', '')
                    if isinstance(prev_content, str) and isinstance(content, str):
                        normalized[-1]['content'] = f"{prev_content}\n{content}"
                    else:
                        normalized[-1]['content'] = content
                    issues_found += 1
                    continue
            
            normalized.append(msg.copy())
        
        if issues_found:
            logger.info(f"ClaudeProviderHandler: Message validation fixed {issues_found} issue(s)")
        
        return normalized
    
    def _truncate_tool_result(self, content: str, max_chars: int = 100000) -> tuple:
        """Truncate tool result content if it exceeds the size limit."""
        import logging
        logger = logging.getLogger(__name__)
        
        if not content or len(content) <= max_chars:
            return content, False
        
        truncation_notice = f"\n\n[Tool result truncated: exceeded {max_chars} character limit. Original length: {len(content)} characters.]"
        truncated = content[:max_chars - len(truncation_notice)] + truncation_notice
        
        logger.warning(f"ClaudeProviderHandler: Tool result truncated from {len(content)} to {max_chars} characters")
        return truncated, True
    
    def _get_cache_config(self, user_id: int = None, provider_id: str = None, model_name: str = None) -> Dict:
        """Get prompt caching configuration from provider config and user settings."""
        cache_config = {
            'enabled': False,
            'min_messages': 4,
        }
        
        if self.provider_config:
            if isinstance(self.provider_config, dict):
                claude_config = self.provider_config.get('claude_config')
            else:
                claude_config = getattr(self.provider_config, 'claude_config', None)
            
            if claude_config and isinstance(claude_config, dict):
                cache_config['enabled'] = claude_config.get('enable_prompt_caching', False)
                cache_config['min_messages'] = claude_config.get('cache_min_messages', 4)
        
        # Check user's cache settings (overrides provider config)
        if user_id and cache_config['enabled']:
            if not self._native_cache_user_allows(model_name):
                cache_config['enabled'] = False
                import logging
                logging.getLogger(__name__).info(f"User {user_id} disabled cache for provider={provider_id}, model={model_name}")
        
        return cache_config
    
    def _get_fallback_models(self) -> List[str]:
        """Get list of fallback models from provider config."""
        fallback_models = []
        
        if self.provider_config:
            if isinstance(self.provider_config, dict):
                claude_config = self.provider_config.get('claude_config')
            else:
                claude_config = getattr(self.provider_config, 'claude_config', None)
            
            if claude_config and isinstance(claude_config, dict):
                fallback_models = claude_config.get('fallback_models', [])
        
        return fallback_models
    
    def _convert_tool_choice_to_anthropic(self, tool_choice: Optional[Union[str, Dict]]) -> Optional[Dict]:
        """Convert OpenAI tool_choice format to Anthropic format."""
        import logging
        
        if not tool_choice:
            return None
        
        if isinstance(tool_choice, str):
            if tool_choice == "auto":
                return {"type": "auto"}
            elif tool_choice == "none":
                return None
            elif tool_choice == "required":
                return {"type": "any"}
            else:
                logging.warning(f"Unknown tool_choice string: {tool_choice}, using auto")
                return {"type": "auto"}
        
        if isinstance(tool_choice, dict):
            if tool_choice.get("type") == "function":
                function = tool_choice.get("function", {})
                tool_name = function.get("name")
                if tool_name:
                    return {"type": "tool", "name": tool_name}
                else:
                    logging.warning(f"tool_choice dict missing function name: {tool_choice}")
                    return {"type": "auto"}
            else:
                logging.warning(f"Unknown tool_choice dict format: {tool_choice}, passing through")
                return tool_choice
        
        logging.warning(f"Unknown tool_choice type: {type(tool_choice)}, using auto")
        return {"type": "auto"}
    
    def _convert_tools_to_anthropic(self, tools: Optional[List[Dict]]) -> Optional[List[Dict]]:
        """Convert OpenAI tools format to Anthropic format."""
        import logging
        
        if not tools:
            return None
        
        def normalize_schema(schema: Dict) -> Dict:
            """Recursively normalize JSON Schema for Anthropic compatibility."""
            if not isinstance(schema, dict):
                return schema
            
            result = {}
            for key, value in schema.items():
                if key == "type" and isinstance(value, list):
                    non_null_types = [t for t in value if t != "null"]
                    if len(non_null_types) == 1:
                        result[key] = non_null_types[0]
                    elif len(non_null_types) > 1:
                        result[key] = non_null_types
                    else:
                        result[key] = "string"
                elif key == "properties" and isinstance(value, dict):
                    result[key] = {k: normalize_schema(v) for k, v in value.items()}
                elif key == "items" and isinstance(value, dict):
                    result[key] = normalize_schema(value)
                elif key == "additionalProperties" and value is False:
                    continue
                elif key == "required" and isinstance(value, list):
                    properties = schema.get("properties", {})
                    cleaned_required = []
                    for field in value:
                        if field in properties:
                            field_schema = properties[field]
                            if isinstance(field_schema, dict):
                                field_type = field_schema.get("type")
                                if isinstance(field_type, list) and "null" in field_type:
                                    continue
                            cleaned_required.append(field)
                    if cleaned_required:
                        result[key] = cleaned_required
                else:
                    result[key] = value
            
            return result
        
        anthropic_tools = []
        for tool in tools:
            if tool.get("type") == "function":
                function = tool.get("function", {})
                parameters = function.get("parameters", {})
                
                normalized_schema = normalize_schema(parameters)
                
                anthropic_tool = {
                    "name": function.get("name", ""),
                    "description": function.get("description", ""),
                    "input_schema": normalized_schema
                }
                anthropic_tools.append(anthropic_tool)
                logging.info(f"Converted tool to Anthropic format: {anthropic_tool['name']}")
            else:
                logging.warning(f"Unknown tool type: {tool.get('type')}, skipping")
        
        return anthropic_tools if anthropic_tools else None
    
    def _extract_images_from_content(self, content: Union[str, List, None]) -> List[Dict]:
        """Extract images from OpenAI message content format."""
        import logging
        logger = logging.getLogger(__name__)
        
        if not isinstance(content, list):
            return []
        
        images = []
        max_image_size = 5 * 1024 * 1024
        
        for block in content:
            if not isinstance(block, dict):
                continue
            
            block_type = block.get('type', '')
            
            if block_type == 'image_url':
                image_url_obj = block.get('image_url', {})
                url = image_url_obj.get('url', '') if isinstance(image_url_obj, dict) else ''
                
                if not url:
                    logger.warning("ClaudeProviderHandler: Empty image URL in content block")
                    continue
                
                if url.startswith('data:'):
                    try:
                        header, data = url.split(',', 1)
                        media_part = header.split(';')[0]
                        media_type = media_part.replace('data:', '')
                        
                        if len(data) > max_image_size:
                            logger.warning(f"ClaudeProviderHandler: Image too large ({len(data)} bytes), skipping")
                            continue
                        
                        image_block = {
                            'type': 'image',
                            'source': {
                                'type': 'base64',
                                'media_type': media_type,
                                'data': data
                            }
                        }
                        images.append(image_block)
                        logger.debug(f"ClaudeProviderHandler: Extracted base64 image ({media_type}, {len(data)} bytes)")
                        
                    except (ValueError, IndexError) as e:
                        logger.warning(f"ClaudeProviderHandler: Failed to parse data URL: {e}")
                
                elif url.startswith(('http://', 'https://')):
                    image_block = {
                        'type': 'image',
                        'source': {
                            'type': 'url',
                            'url': url
                        }
                    }
                    images.append(image_block)
                    logger.debug(f"ClaudeProviderHandler: Extracted URL image: {url[:80]}...")
                
                else:
                    logger.warning(f"ClaudeProviderHandler: Unsupported image URL format: {url[:80]}...")
            
            elif block_type == 'image':
                if 'source' in block:
                    images.append(block)
                    logger.debug("ClaudeProviderHandler: Passed through existing image block")
        
        return images
    
    def _convert_messages_to_anthropic(self, messages: List[Dict]) -> tuple:
        """
        Convert OpenAI messages format to Anthropic format.
        Delegates to shared AnthropicFormatConverter.convert_messages_to_anthropic().
        """
        return AnthropicFormatConverter.convert_messages_to_anthropic(messages, sanitize_ids=True)
    
    async def handle_request(self, model: str, messages: List[Dict], max_tokens: Optional[int] = None,
                           temperature: Optional[float] = 1.0, stream: Optional[bool] = False,
                           tools: Optional[List[Dict]] = None, tool_choice: Optional[Union[str, Dict]] = None) -> Union[Dict, object]:
        if self.is_rate_limited():
            raise Exception("Provider rate limited")

        # Get fallback models from config (Phase 3.3)
        fallback_models = self._get_fallback_models()
        models_to_try = [model] + fallback_models
        
        last_error = None
        
        for attempt, current_model in enumerate(models_to_try):
            try:
                if attempt > 0:
                    import logging
                    logger = logging.getLogger(__name__)
                    logger.warning(f"ClaudeProviderHandler: Retrying with fallback model: {current_model} (original: {model})")
                
                result = await self._handle_request_with_model(
                    model=current_model,
                    messages=messages,
                    max_tokens=max_tokens,
                    temperature=temperature,
                    stream=stream,
                    tools=tools,
                    tool_choice=tool_choice
                )
                
                if stream:
                    return self._wrap_streaming_with_retry(result, current_model, messages, max_tokens, temperature, tools, tool_choice, models_to_try, attempt)
                
                return result
                
            except Exception as e:
                last_error = e
                import logging
                logger = logging.getLogger(__name__)
                
                error_str = str(e).lower()
                is_retryable = any(keyword in error_str for keyword in [
                    'rate limit', 'overloaded', 'too many requests', '429', '529', '503'
                ])
                
                if is_retryable and attempt < len(models_to_try) - 1:
                    logger.warning(f"ClaudeProviderHandler: Retryable error with {current_model}, trying next fallback model")
                    wait_time = min(2 ** attempt + random.uniform(0, 1), 30)
                    logger.info(f"ClaudeProviderHandler: Waiting {wait_time:.1f}s before retry")
                    await asyncio.sleep(wait_time)
                    continue
                
                logger.error(f"ClaudeProviderHandler: Error with model {current_model}: {str(e)}", exc_info=True)
                self.record_failure()
                raise e
        
        import logging
        logger = logging.getLogger(__name__)
        logger.error(f"ClaudeProviderHandler: All models failed (tried: {models_to_try})")
        raise last_error
    
    async def _wrap_streaming_with_retry(self, stream_generator, current_model, messages, max_tokens, temperature, tools, tool_choice, models_to_try, attempt):
        """Wrapper that consumes the streaming generator and catches errors."""
        import logging
        logger = logging.getLogger(__name__)
        
        try:
            async for chunk in stream_generator:
                yield chunk
        except Exception as e:
            last_error = e
            error_str = str(e).lower()
            is_retryable = any(keyword in error_str for keyword in [
                'rate limit', 'overloaded', 'too many requests', '429', '529', '503'
            ])
            
            if is_retryable and attempt < len(models_to_try) - 1:
                next_model = models_to_try[attempt + 1]
                logger.warning(f"ClaudeProviderHandler: Streaming error with {current_model}, retrying with {next_model}")
                
                wait_time = min(2 ** (attempt + 1) + random.uniform(0, 1), 30)
                logger.info(f"ClaudeProviderHandler: Waiting {wait_time:.1f}s before retry")
                await asyncio.sleep(wait_time)
                
                try:
                    result = await self._handle_request_with_model(
                        model=next_model,
                        messages=messages,
                        max_tokens=max_tokens,
                        temperature=temperature,
                        stream=True,
                        tools=tools,
                        tool_choice=tool_choice
                    )
                    async for chunk in self._wrap_streaming_with_retry(result, next_model, messages, max_tokens, temperature, tools, tool_choice, models_to_try, attempt + 1):
                        yield chunk
                    return
                except Exception as retry_error:
                    logger.error(f"ClaudeProviderHandler: Retry with {next_model} also failed: {str(retry_error)}")
                    raise retry_error
            
            logger.error(f"ClaudeProviderHandler: Streaming error: {str(e)}", exc_info=True)
            raise e
    
    async def _handle_request_with_model(self, model: str, messages: List[Dict], max_tokens: Optional[int] = None,
                                        temperature: Optional[float] = 1.0, stream: Optional[bool] = False,
                                        tools: Optional[List[Dict]] = None, tool_choice: Optional[Union[str, Dict]] = None) -> Union[Dict, object]:
        """Handle request with a specific model.

        When the claude CLI is detected at startup and the user has CLI
        credentials configured, the request is proxied through a subprocess
        instead of calling the Anthropic HTTP API directly.
        """
        import logging
        logger = logging.getLogger(__name__)

        # ── Claude CLI mode (skipped when api_token is configured) ──────
        import aisbf.providers.claude_cli as cli_mode_mod
        if cli_mode_mod.CLAUDE_CLI_MODE and not self._get_api_token():
            cli_creds = self._get_cli_credentials()
            if cli_creds is not None:
                logger.info(f"ClaudeProviderHandler: using CLI subprocess mode for model {model}")
                anthropic_tools = self._convert_tools_to_anthropic(tools) if tools else None
                system_prompt, prompt = self._messages_to_cli_prompt(messages, tools=anthropic_tools)
                config_dir = await ClaudeCliSessionManager.get_config_dir(
                    self.user_id, self.provider_id, cli_creds
                )
                if stream:
                    return self._handle_cli_streaming_request(
                        prompt, model, config_dir,
                        system_prompt=system_prompt, tools=anthropic_tools,
                    )
                else:
                    return await self._handle_cli_request(
                        prompt, model, config_dir,
                        system_prompt=system_prompt, tools=anthropic_tools,
                    )
        # ── Fall through to HTTP API mode ────────────────────────────────

        logger.info(f"ClaudeProviderHandler: Handling request for model {model} (Direct HTTP mode)")

        if AISBF_DEBUG:
            logger.info(f"ClaudeProviderHandler: Messages: {messages}")
        else:
            logger.info(f"ClaudeProviderHandler: Messages count: {len(messages)}")

        using_api_token = bool(self._get_api_token())

        # Session init uses OAuth2-specific headers; skip when using API key
        if not using_api_token:
            await self._ensure_session()
        await self.apply_rate_limit()

        validated_messages = self._validate_messages(messages)

        system_message, anthropic_messages = self._convert_messages_to_anthropic(validated_messages)

        # Apply prompt caching based on user and provider settings
        cache_config = self._get_cache_config(
            user_id=getattr(self, 'user_id', None),
            provider_id=self.provider_id,
            model_name=model
        )

        if cache_config['enabled']:
            anthropic_messages = self._apply_cache_control(anthropic_messages)

        # Sanitize system message to avoid Claude's unofficial client detection
        # (only relevant for OAuth2 mode where we impersonate the CLI)
        if system_message and not using_api_token:
            import re
            system_message = re.sub(
                r'\bYou are (Kilo|Kiro),',
                'You are',
                system_message,
                flags=re.IGNORECASE
            )

        payload = {
            'model': model,
            'messages': anthropic_messages,
            'max_tokens': max_tokens or 4096,
        }

        if temperature is not None and temperature > 0:
            payload['temperature'] = temperature

        if system_message:
            if using_api_token:
                # Standard Anthropic API: plain system string
                payload['system'] = system_message
            else:
                # OAuth2 / CLI mode: include billing and identity headers
                payload['system'] = [
                    {
                        'type': 'text',
                        'text': 'x-anthropic-billing-header: cc_version=99.0.0.e8c; cc_entrypoint=cli;'
                    },
                    {
                        'type': 'text',
                        'text': "You are Claude Code, Anthropic's official CLI for Claude."
                    },
                    {
                        'type': 'text',
                        'text': system_message
                    },
                ]
        
        # Metadata with session identifiers is OAuth2/CLI specific
        if not using_api_token:
            payload['metadata'] = {
                'user_id': json.dumps({
                    'device_id': self.session_state['device_id'],
                    'account_uuid': self.session_state['account_uuid'],
                    'session_id': self.session_state['session_id']
                })
            }
        
        if tools:
            anthropic_tools = self._convert_tools_to_anthropic(tools)
            if anthropic_tools:
                payload['tools'] = anthropic_tools
        
        if tool_choice and tools:
            anthropic_tool_choice = self._convert_tool_choice_to_anthropic(tool_choice)
            if anthropic_tool_choice:
                payload['tool_choice'] = anthropic_tool_choice
        
        headers = await self._get_auth_headers(stream=stream)
        # Standard Anthropic API when api_token is set; OAuth2 endpoint otherwise
        if self._get_api_token():
            api_url = 'https://api.anthropic.com/v1/messages'
        else:
            api_url = 'https://api.anthropic.com/v1/messages?beta=true'
        
        logger.info(f"ClaudeProviderHandler: Request payload keys: {list(payload.keys())}")
        if AISBF_DEBUG:
            logger.info(f"ClaudeProviderHandler: Full payload: {json.dumps(payload, indent=2)}")
        
        try:
            if stream:
                payload['stream'] = True
                logger.info(f"ClaudeProviderHandler: Using direct HTTP streaming mode")
                return self._handle_streaming_request_with_retry(api_url, payload, headers, model)
            else:
                logger.info(f"ClaudeProviderHandler: Using direct HTTP non-streaming mode")
                response = await self._request_with_retry(api_url, headers, payload, max_retries=3)
                
                logger.info(f"ClaudeProviderHandler: HTTP response received successfully")
                
                self._update_session_from_headers(dict(response.headers))
                
                self.record_success()
                
                response_data = response.json()
                
                if AISBF_DEBUG:
                    logger.info(f"=== RAW CLAUDE RESPONSE ===")
                    logger.info(f"Raw response data: {json.dumps(response_data, indent=2, default=str)}")
                    logger.info(f"=== END RAW CLAUDE RESPONSE ===")
                
                openai_response = self._convert_to_openai_format(response_data, model)
                
                if AISBF_DEBUG:
                    logger.info(f"=== FINAL CLAUDE RESPONSE DICT ===")
                    logger.info(f"Final response: {json.dumps(openai_response, indent=2, default=str)}")
                    logger.info(f"=== END FINAL CLAUDE RESPONSE DICT ===")
                
                return openai_response
                
        except Exception as e:
            logger.error(f"ClaudeProviderHandler: HTTP request failed: {e}", exc_info=True)
            raise
    
    async def _request_with_retry(self, api_url: str, headers: Dict, payload: Dict, max_retries: int = 3):
        """Non-streaming request with automatic retry for transient errors."""
        import logging
        import json
        logger = logging.getLogger(__name__)
        
        last_error = None
        
        for attempt in range(max_retries):
            try:
                response = await self.client.post(api_url, headers=headers, json=payload)
                
                logger.info(f"ClaudeProviderHandler: Response status: {response.status_code} (attempt {attempt + 1}/{max_retries})")
                
                if response.status_code in (429, 529, 503):
                    should_retry = response.headers.get('x-should-retry', 'false').lower() == 'true'
                    
                    if should_retry or response.status_code in (529, 503):
                        if attempt < max_retries - 1:
                            wait_time = min(2 ** attempt + random.uniform(0, 1), 30)
                            
                            try:
                                error_data = response.json()
                                error_message = error_data.get('error', {}).get('message', '')
                                logger.warning(f"ClaudeProviderHandler: Retryable error: {error_message}")
                            except Exception:
                                pass
                            
                            logger.info(f"ClaudeProviderHandler: Retrying in {wait_time:.1f}s (attempt {attempt + 1}/{max_retries})")
                            await asyncio.sleep(wait_time)
                            continue
                        else:
                            try:
                                response_data = response.json()
                            except Exception:
                                response_data = response.text
                            
                            self.handle_429_error(response_data, dict(response.headers))
                            response.raise_for_status()
                
                if response.status_code >= 400:
                    try:
                        error_body = response.json()
                        error_message = error_body.get('error', {}).get('message', 'Unknown error')
                        error_type = error_body.get('error', {}).get('type', 'unknown')
                        logger.error(f"ClaudeProviderHandler: API error response: {json.dumps(error_body, indent=2)}")
                        logger.error(f"ClaudeProviderHandler: Error type: {error_type}")
                        logger.error(f"ClaudeProviderHandler: Error message: {error_message}")
                    except Exception:
                        logger.error(f"ClaudeProviderHandler: API error response (text): {response.text}")
                    
                    response.raise_for_status()
                
                return response
                
            except httpx.TimeoutException as e:
                last_error = e
                if attempt < max_retries - 1:
                    wait_time = min(2 ** attempt + random.uniform(0, 1), 30)
                    logger.warning(f"ClaudeProviderHandler: Request timeout, retrying in {wait_time:.1f}s")
                    await asyncio.sleep(wait_time)
                    continue
                else:
                    logger.error(f"ClaudeProviderHandler: Request timeout after {max_retries} attempts")
                    raise
            
            except httpx.HTTPError as e:
                last_error = e
                if attempt < max_retries - 1:
                    wait_time = min(2 ** attempt + random.uniform(0, 1), 30)
                    logger.warning(f"ClaudeProviderHandler: HTTP error, retrying in {wait_time:.1f}s: {e}")
                    await asyncio.sleep(wait_time)
                    continue
                else:
                    logger.error(f"ClaudeProviderHandler: HTTP error after {max_retries} attempts: {e}")
                    raise
        
        raise last_error or Exception("Request failed after max retries")
    
    async def _handle_streaming_request_with_retry(self, api_url: str, payload: Dict, headers: Dict, model: str):
        """Wrapper for streaming request that catches rate limit errors at the call site."""
        import logging
        logger = logging.getLogger(__name__)
        
        try:
            async for chunk in self._handle_streaming_request(api_url, payload, headers, model):
                yield chunk
        except Exception as e:
            error_str = str(e).lower()
            if '429' in error_str or 'rate limit' in error_str or 'too many requests' in error_str:
                logger.error(f"ClaudeProviderHandler: Streaming rate limit error: {e}")
                raise Exception(f"Rate limit error: {e}")
            raise
    
    async def _handle_streaming_request(self, api_url: str, payload: Dict, headers: Dict, model: str):
        """Handle streaming request to Claude API using direct HTTP."""
        import logging
        import json
        
        logger = logging.getLogger(__name__)
        logger.info(f"ClaudeProviderHandler: Starting streaming request to {api_url}")
        
        if AISBF_DEBUG:
            logger.info(f"=== STREAMING REQUEST DETAILS ===")
            logger.info(f"URL: {api_url}")
            logger.info(f"Headers (auth redacted): {json.dumps({k: v for k, v in headers.items() if k.lower() != 'authorization'}, indent=2)}")
            logger.info(f"Payload: {json.dumps(payload, indent=2)}")
            logger.info(f"=== END STREAMING REQUEST DETAILS ===")
        
        async with httpx.AsyncClient(timeout=httpx.Timeout(300.0, connect=30.0)) as streaming_client:
            async with streaming_client.stream(
                "POST",
                api_url,
                headers=headers,
                json=payload
            ) as response:
                logger.info(f"ClaudeProviderHandler: Streaming response status: {response.status_code}")
                
                self._update_session_from_headers(dict(response.headers))
                
                if response.status_code >= 400:
                    error_text = await response.aread()
                    logger.error(f"ClaudeProviderHandler: Streaming error response: {error_text}")
                    
                    try:
                        error_json = json.loads(error_text)
                        error_message = error_json.get('error', {}).get('message', 'Unknown error')
                        error_type = error_json.get('error', {}).get('type', 'unknown')
                        logger.error(f"ClaudeProviderHandler: Error type: {error_type}")
                        logger.error(f"ClaudeProviderHandler: Error message: {error_message}")
                        
                        if response.status_code == 429:
                            raise Exception(f"Rate limit error (429): {error_type} - {error_message}")
                        else:
                            raise Exception(f"Claude API error ({response.status_code}): {error_message}")
                    except json.JSONDecodeError:
                        logger.error(f"ClaudeProviderHandler: Could not parse error response as JSON")
                        if response.status_code == 429:
                            raise Exception(f"Rate limit error (429): Too Many Requests - {error_text.decode() if isinstance(error_text, bytes) else error_text}")
                        else:
                            raise Exception(f"Claude API error: {response.status_code} - {error_text.decode() if isinstance(error_text, bytes) else error_text}")
                
                completion_id = f"claude-{int(time.time())}"
                created_time = int(time.time())
                
                first_chunk = True
                accumulated_content = ""
                accumulated_tool_calls = []
                
                accumulated_thinking = ""
                thinking_signature = ""
                is_redacted_thinking = False
                
                content_block_index = 0
                current_tool_calls = []
                
                last_event_time = time.time()
                idle_timeout = self.stream_idle_timeout
                
                stream_stop_reason = None
                
                async for line in response.aiter_lines():
                    if time.time() - last_event_time > idle_timeout:
                        logger.error(f"ClaudeProviderHandler: Stream idle timeout ({idle_timeout}s)")
                        raise TimeoutError(f"Stream idle for {idle_timeout}s")
                    
                    if not line or not line.startswith('data: '):
                        continue
                    
                    data_str = line[6:]
                    
                    if data_str == '[DONE]':
                        break
                    
                    try:
                        chunk_data = json.loads(data_str)
                        
                        last_event_time = time.time()
                        
                        event_type = chunk_data.get('type')
                        
                        if event_type == 'content_block_start':
                            content_block = chunk_data.get('content_block', {})
                            block_type = content_block.get('type', '')
                            
                            if block_type == 'tool_use':
                                tool_call = {
                                    'index': content_block_index,
                                    'id': content_block.get('id', ''),
                                    'type': 'function',
                                    'function': {
                                        'name': content_block.get('name', ''),
                                        'arguments': ''
                                    }
                                }
                                current_tool_calls.append(tool_call)
                                logger.debug(f"ClaudeProviderHandler: Tool use block started: {tool_call['function']['name']}")
                            
                            elif block_type == 'thinking':
                                accumulated_thinking = ""
                                is_redacted_thinking = False
                                thinking_signature = ""
                                logger.debug(f"ClaudeProviderHandler: Thinking block started")
                            
                            elif block_type == 'redacted_thinking':
                                accumulated_thinking = ""
                                is_redacted_thinking = True
                                thinking_signature = ""
                                logger.debug(f"ClaudeProviderHandler: Redacted thinking block started")
                            
                            content_block_index += 1
                        
                        elif event_type == 'content_block_delta':
                            delta = chunk_data.get('delta', {})
                            delta_type = delta.get('type', '')
                            
                            if delta_type == 'text_delta':
                                text = delta.get('text', '')
                                accumulated_content += text
                                
                                openai_delta = {'content': text}
                                if first_chunk:
                                    openai_delta['role'] = 'assistant'
                                    first_chunk = False
                                
                                openai_chunk = {
                                    'id': completion_id,
                                    'object': 'chat.completion.chunk',
                                    'created': created_time,
                                    'model': f'{self.provider_id}/{model}',
                                    'choices': [{
                                        'index': 0,
                                        'delta': openai_delta,
                                        'finish_reason': None
                                    }]
                                }
                                
                                yield f"data: {json.dumps(openai_chunk, ensure_ascii=False)}\n\n".encode('utf-8')
                            
                            elif delta_type == 'input_json_delta':
                                partial_json = delta.get('partial_json', '')
                                if current_tool_calls:
                                    current_tool_calls[-1]['function']['arguments'] += partial_json
                            
                            elif delta_type == 'thinking_delta':
                                thinking_text = delta.get('thinking', '')
                                accumulated_thinking += thinking_text
                                logger.debug(f"ClaudeProviderHandler: Thinking delta: {len(thinking_text)} chars")
                            
                            elif delta_type == 'signature_delta':
                                signature = delta.get('signature', '')
                                thinking_signature = signature
                                logger.debug(f"ClaudeProviderHandler: Thinking signature received")
                        
                        elif event_type == 'content_block_stop':
                            if current_tool_calls:
                                tool_call = current_tool_calls[-1]
                                try:
                                    args = json.loads(tool_call['function']['arguments']) if tool_call['function']['arguments'] else {}
                                    tool_call['function']['arguments'] = json.dumps(args)
                                except json.JSONDecodeError:
                                    logger.warning(f"ClaudeProviderHandler: Invalid tool call arguments JSON")
                                    tool_call['function']['arguments'] = '{}'
                                
                                tool_call_chunk = {
                                    'id': completion_id,
                                    'object': 'chat.completion.chunk',
                                    'created': created_time,
                                    'model': f'{self.provider_id}/{model}',
                                    'choices': [{
                                        'index': 0,
                                        'delta': {
                                            'tool_calls': [{
                                                'index': tool_call['index'],
                                                'id': tool_call['id'],
                                                'type': tool_call['type'],
                                                'function': tool_call['function']
                                            }]
                                        },
                                        'finish_reason': None
                                    }]
                                }
                                
                                yield f"data: {json.dumps(tool_call_chunk, ensure_ascii=False)}\n\n".encode('utf-8')
                                logger.debug(f"ClaudeProviderHandler: Emitted tool call: {tool_call['function']['name']}")
                            
                            elif accumulated_thinking:
                                block_type = "redacted_thinking" if is_redacted_thinking else "thinking"
                                logger.info(f"ClaudeProviderHandler: {block_type} block completed ({len(accumulated_thinking)} chars)")
                                accumulated_thinking = ""
                                is_redacted_thinking = False
                                thinking_signature = ""
                        
                        elif event_type == 'message_delta':
                            delta_data = chunk_data.get('delta', {})
                            usage = chunk_data.get('usage', {})
                            
                            stream_stop_reason = delta_data.get('stop_reason')
                            if stream_stop_reason:
                                logger.debug(f"ClaudeProviderHandler: Stream stop_reason: {stream_stop_reason}")
                            
                            if usage:
                                logger.debug(f"ClaudeProviderHandler: Streaming usage update: {usage}")
                                
                                cache_read = usage.get('cache_read_input_tokens', 0)
                                cache_creation = usage.get('cache_creation_input_tokens', 0)
                                if cache_read > 0:
                                    self.cache_stats['cache_hits'] += 1
                                    self.cache_stats['cache_tokens_read'] += cache_read
                                if cache_creation > 0:
                                    self.cache_stats['cache_misses'] += 1
                                    self.cache_stats['cache_tokens_created'] += cache_creation
                        
                        elif event_type == 'message_stop':
                            stop_reason_map = {
                                'end_turn': 'stop',
                                'max_tokens': 'length',
                                'stop_sequence': 'stop',
                                'tool_use': 'tool_calls'
                            }
                            if stream_stop_reason:
                                finish_reason = stop_reason_map.get(stream_stop_reason, 'stop')
                            elif current_tool_calls:
                                finish_reason = 'tool_calls'
                            else:
                                finish_reason = 'stop'
                            logger.debug(f"ClaudeProviderHandler: Final finish_reason: {finish_reason}")
                            
                            final_chunk = {
                                'id': completion_id,
                                'object': 'chat.completion.chunk',
                                'created': created_time,
                                'model': f'{self.provider_id}/{model}',
                                'choices': [{
                                    'index': 0,
                                    'delta': {},
                                    'finish_reason': finish_reason
                                }]
                            }
                            
                            yield f"data: {json.dumps(final_chunk, ensure_ascii=False)}\n\n".encode('utf-8')
                            # Yield control to event loop to ensure chunk is flushed to client
                            await asyncio.sleep(0)
                            yield b"data: [DONE]\n\n"
                            # Final flush to ensure all buffered data reaches the client
                            await asyncio.sleep(0)
                    
                    except json.JSONDecodeError as e:
                        logger.warning(f"Failed to parse streaming chunk: {e}")
                        continue
    
    def _convert_to_openai_format(self, claude_response: Dict, model: str) -> Dict:
        """Convert Claude API response to OpenAI format."""
        import logging
        import json
        logger = logging.getLogger(__name__)
        
        logger.info(f"ClaudeProviderHandler: Converting response to OpenAI format")
        
        content_text = ""
        tool_calls = []
        thinking_text = ""
        
        if 'content' in claude_response:
            for block in claude_response['content']:
                block_type = block.get('type', '')
                
                if block_type == 'text':
                    content_text += block.get('text', '')
                elif block_type == 'tool_use':
                    tool_calls.append({
                        'id': block.get('id', f"call_{len(tool_calls)}"),
                        'type': 'function',
                        'function': {
                            'name': block.get('name', ''),
                            'arguments': json.dumps(block.get('input', {}))
                        }
                    })
                elif block_type == 'thinking':
                    thinking_text = block.get('thinking', '')
                    logger.debug(f"ClaudeProviderHandler: Extracted thinking block ({len(thinking_text)} chars)")
                elif block_type == 'redacted_thinking':
                    logger.debug(f"ClaudeProviderHandler: Found redacted_thinking block")
        
        stop_reason_map = {
            'end_turn': 'stop',
            'max_tokens': 'length',
            'stop_sequence': 'stop',
            'tool_use': 'tool_calls'
        }
        stop_reason = claude_response.get('stop_reason', 'end_turn')
        finish_reason = stop_reason_map.get(stop_reason, 'stop')
        
        usage = claude_response.get('usage', {})
        input_tokens = usage.get('input_tokens', 0)
        output_tokens = usage.get('output_tokens', 0)
        cache_read_tokens = usage.get('cache_read_input_tokens', 0)
        cache_creation_tokens = usage.get('cache_creation_input_tokens', 0)
        
        if cache_read_tokens or cache_creation_tokens:
            logger.info(f"ClaudeProviderHandler: Cache usage - read: {cache_read_tokens}, creation: {cache_creation_tokens}")
        
        openai_response = {
            'id': f"claude-{model}-{int(time.time())}",
            'object': 'chat.completion',
            'created': int(time.time()),
            'model': f'{self.provider_id}/{model}',
            'choices': [{
                'index': 0,
                'message': {
                    'role': 'assistant',
                    'content': content_text if content_text else None
                },
                'finish_reason': finish_reason
            }],
            'usage': {
                'prompt_tokens': input_tokens,
                'completion_tokens': output_tokens,
                'total_tokens': input_tokens + output_tokens,
                'prompt_tokens_details': {
                    'cached_tokens': cache_read_tokens,
                    'audio_tokens': 0
                },
                'completion_tokens_details': {
                    'reasoning_tokens': 0,
                    'audio_tokens': 0
                }
            }
        }
        
        if tool_calls:
            openai_response['choices'][0]['message']['tool_calls'] = tool_calls
        
        if thinking_text:
            openai_response['choices'][0]['message']['provider_options'] = {
                'anthropic': {
                    'thinking': thinking_text
                }
            }
            logger.debug(f"ClaudeProviderHandler: Added thinking content to response ({len(thinking_text)} chars)")
        
        return openai_response
    
    def _convert_sdk_response_to_openai(self, response, model: str) -> Dict:
        """Convert Anthropic SDK response object to OpenAI format."""
        import logging
        import json
        logger = logging.getLogger(__name__)
        
        message_content = ""
        tool_calls = []
        thinking_text = ""
        
        for block in response.content:
            block_type = getattr(block, 'type', '')
            
            if block_type == 'text' or hasattr(block, 'text'):
                message_content += getattr(block, 'text', '')
            elif block_type == 'tool_use':
                tool_calls.append({
                    'id': getattr(block, 'id', f"call_{len(tool_calls)}"),
                    'type': 'function',
                    'function': {
                        'name': getattr(block, 'name', ''),
                        'arguments': json.dumps(getattr(block, 'input', {}))
                    }
                })
            elif block_type == 'thinking':
                thinking_text = getattr(block, 'thinking', '')
                logger.debug(f"ClaudeProviderHandler: Extracted thinking block ({len(thinking_text)} chars)")
            elif block_type == 'redacted_thinking':
                logger.debug(f"ClaudeProviderHandler: Found redacted_thinking block")
        
        stop_reason_map = {
            'end_turn': 'stop',
            'max_tokens': 'length',
            'stop_sequence': 'stop',
            'tool_use': 'tool_calls'
        }
        stop_reason = getattr(response, 'stop_reason', 'end_turn') or 'end_turn'
        finish_reason = stop_reason_map.get(stop_reason, 'stop')
        
        usage = getattr(response, 'usage', None)
        input_tokens = getattr(usage, 'input_tokens', 0) if usage else 0
        output_tokens = getattr(usage, 'output_tokens', 0) if usage else 0
        cache_read_tokens = getattr(usage, 'cache_read_input_tokens', 0) if usage else 0
        cache_creation_tokens = getattr(usage, 'cache_creation_input_tokens', 0) if usage else 0
        
        if cache_read_tokens or cache_creation_tokens:
            logger.info(f"ClaudeProviderHandler: Cache usage - read: {cache_read_tokens}, creation: {cache_creation_tokens}")
        
        openai_response = {
            'id': getattr(response, 'id', f"claude-{model}-{int(time.time())}"),
            'object': 'chat.completion',
            'created': int(time.time()),
            'model': f'{self.provider_id}/{model}',
            'choices': [{
                'index': 0,
                'message': {
                    'role': 'assistant',
                    'content': message_content if message_content else None,
                },
                'finish_reason': finish_reason
            }],
            'usage': {
                'prompt_tokens': input_tokens,
                'completion_tokens': output_tokens,
                'total_tokens': input_tokens + output_tokens,
                'prompt_tokens_details': {
                    'cached_tokens': cache_read_tokens,
                    'audio_tokens': 0
                },
                'completion_tokens_details': {
                    'reasoning_tokens': 0,
                    'audio_tokens': 0
                }
            }
        }
        
        if tool_calls:
            openai_response['choices'][0]['message']['tool_calls'] = tool_calls
        
        if thinking_text:
            openai_response['choices'][0]['message']['provider_options'] = {
                'anthropic': {
                    'thinking': thinking_text
                }
            }
            logger.debug(f"ClaudeProviderHandler: Added thinking content to response ({len(thinking_text)} chars)")
        
        return openai_response
    
    async def _handle_streaming_request_sdk(self, client, request_kwargs: Dict, model: str):
        """Handle streaming request using Anthropic SDK's async streaming API."""
        import logging
        import json
        logger = logging.getLogger(__name__)
        
        logger.info(f"ClaudeProviderHandler: Starting SDK streaming request")
        
        completion_id = f"claude-{int(time.time())}"
        created_time = int(time.time())
        
        first_chunk = True
        accumulated_content = ""
        accumulated_thinking = ""
        thinking_signature = ""
        is_redacted_thinking = False
        content_block_index = 0
        current_tool_calls = []
        
        last_event_time = time.time()
        idle_timeout = self.stream_idle_timeout
        
        try:
            stream = await client.messages.create(**request_kwargs, stream=True)
            
            async for event in stream:
                last_event_time = time.time()
                
                if time.time() - last_event_time > idle_timeout:
                    logger.error(f"ClaudeProviderHandler: Stream idle timeout ({idle_timeout}s)")
                    raise TimeoutError(f"Stream idle for {idle_timeout}s")
                
                event_type = getattr(event, 'type', None)
                
                if event_type == 'content_block_start':
                    content_block = getattr(event, 'content_block', None)
                    if content_block:
                        block_type = getattr(content_block, 'type', '')
                        
                        if block_type == 'tool_use':
                            tool_call = {
                                'index': content_block_index,
                                'id': getattr(content_block, 'id', ''),
                                'type': 'function',
                                'function': {
                                    'name': getattr(content_block, 'name', ''),
                                    'arguments': ''
                                }
                            }
                            current_tool_calls.append(tool_call)
                            logger.debug(f"ClaudeProviderHandler: Tool use block started: {tool_call['function']['name']}")
                        
                        elif block_type == 'thinking':
                            accumulated_thinking = ""
                            is_redacted_thinking = False
                            thinking_signature = ""
                            logger.debug(f"ClaudeProviderHandler: Thinking block started")
                        
                        elif block_type == 'redacted_thinking':
                            accumulated_thinking = ""
                            is_redacted_thinking = True
                            thinking_signature = ""
                            logger.debug(f"ClaudeProviderHandler: Redacted thinking block started")
                        
                        content_block_index += 1
                
                elif event_type == 'content_block_delta':
                    delta = getattr(event, 'delta', None)
                    if delta:
                        delta_type = getattr(delta, 'type', '')
                        
                        if delta_type == 'text_delta':
                            text = getattr(delta, 'text', '')
                            accumulated_content += text
                            
                            openai_delta = {'content': text}
                            if first_chunk:
                                openai_delta['role'] = 'assistant'
                                first_chunk = False
                            
                            openai_chunk = {
                                'id': completion_id,
                                'object': 'chat.completion.chunk',
                                'created': created_time,
                                'model': f'{self.provider_id}/{model}',
                                'choices': [{
                                    'index': 0,
                                    'delta': openai_delta,
                                    'finish_reason': None
                                }]
                            }
                            
                            yield f"data: {json.dumps(openai_chunk, ensure_ascii=False)}\n\n".encode('utf-8')
                        
                        elif delta_type == 'input_json_delta':
                            partial_json = getattr(delta, 'partial_json', '')
                            if current_tool_calls:
                                current_tool_calls[-1]['function']['arguments'] += partial_json
                        
                        elif delta_type == 'thinking_delta':
                            thinking_text = getattr(delta, 'thinking', '')
                            accumulated_thinking += thinking_text
                            logger.debug(f"ClaudeProviderHandler: Thinking delta: {len(thinking_text)} chars")
                        
                        elif delta_type == 'signature_delta':
                            signature = getattr(delta, 'signature', '')
                            thinking_signature = signature
                            logger.debug(f"ClaudeProviderHandler: Thinking signature received")
                
                elif event_type == 'content_block_stop':
                    if current_tool_calls:
                        tool_call = current_tool_calls[-1]
                        try:
                            args = json.loads(tool_call['function']['arguments']) if tool_call['function']['arguments'] else {}
                            tool_call['function']['arguments'] = json.dumps(args)
                        except json.JSONDecodeError:
                            logger.warning(f"ClaudeProviderHandler: Invalid tool call arguments JSON")
                            tool_call['function']['arguments'] = '{}'
                        
                        tool_call_chunk = {
                            'id': completion_id,
                            'object': 'chat.completion.chunk',
                            'created': created_time,
                            'model': f'{self.provider_id}/{model}',
                            'choices': [{
                                'index': 0,
                                'delta': {
                                    'tool_calls': [{
                                        'index': tool_call['index'],
                                        'id': tool_call['id'],
                                        'type': tool_call['type'],
                                        'function': tool_call['function']
                                    }]
                                },
                                'finish_reason': None
                            }]
                        }
                        
                        yield f"data: {json.dumps(tool_call_chunk, ensure_ascii=False)}\n\n".encode('utf-8')
                        logger.debug(f"ClaudeProviderHandler: Emitted tool call: {tool_call['function']['name']}")
                    
                    elif accumulated_thinking:
                        block_type = "redacted_thinking" if is_redacted_thinking else "thinking"
                        logger.info(f"ClaudeProviderHandler: {block_type} block completed ({len(accumulated_thinking)} chars)")
                        accumulated_thinking = ""
                        is_redacted_thinking = False
                        thinking_signature = ""
                
                elif event_type == 'message_delta':
                    usage = getattr(event, 'usage', None)
                    if usage:
                        logger.debug(f"ClaudeProviderHandler: Streaming usage update: {usage}")
                        
                        cache_read = getattr(usage, 'cache_read_input_tokens', 0)
                        cache_creation = getattr(usage, 'cache_creation_input_tokens', 0)
                        if cache_read > 0:
                            self.cache_stats['cache_hits'] += 1
                            self.cache_stats['cache_tokens_read'] += cache_read
                        if cache_creation > 0:
                            self.cache_stats['cache_misses'] += 1
                            self.cache_stats['cache_tokens_created'] += cache_creation
                
                elif event_type == 'message_stop':
                    final_chunk = {
                        'id': completion_id,
                        'object': 'chat.completion.chunk',
                        'created': created_time,
                        'model': f'{self.provider_id}/{model}',
                        'choices': [{
                            'index': 0,
                            'delta': {},
                            'finish_reason': 'stop'
                        }]
                    }
                    
                    yield f"data: {json.dumps(final_chunk, ensure_ascii=False)}\n\n".encode('utf-8')
                    # Yield control to event loop to ensure chunk is flushed to client
                    await asyncio.sleep(0)
                    yield b"data: [DONE]\n\n"
                    # Final flush to ensure all buffered data reaches the client
                    await asyncio.sleep(0)
            
            logger.info(f"ClaudeProviderHandler: SDK streaming completed successfully")
            self.record_success()
            
        except Exception as e:
            logger.error(f"ClaudeProviderHandler: SDK streaming error: {str(e)}", exc_info=True)
            raise
    
    def get_cache_stats(self) -> Dict:
        """Get cache usage statistics (Phase 2.3)."""
        total = self.cache_stats['cache_hits'] + self.cache_stats['cache_misses']
        hit_rate = (self.cache_stats['cache_hits'] / total * 100) if total > 0 else 0
        
        return {
            **self.cache_stats,
            'total_cache_events': total,
            'cache_hit_rate_percent': round(hit_rate, 2),
        }
    
    # Model caching is now handled by the base class using the unified cache system
    # _get_models_cache_path(), _save_models_cache(), _load_models_cache() are inherited from BaseProviderHandler

    async def get_models(self) -> List[Model]:
        """Return list of available Claude models by querying the API."""
        try:
            import logging
            import json
            logging.info("=" * 80)
            logging.info("ClaudeProviderHandler: Starting model list retrieval")
            logging.info("=" * 80)

            # First try to load from cache
            cached_models = self._load_models_cache()
            if cached_models:
                logging.info(f"ClaudeProviderHandler: ✓ Returning {len(cached_models)} models from cache")
                return cached_models

            await self.apply_rate_limit()

            # [0/3] CLI subprocess model discovery
            import aisbf.providers.claude_cli as cli_mode_mod
            if cli_mode_mod.CLAUDE_CLI_MODE:
                cli_creds = self._get_cli_credentials()
                if cli_creds is not None:
                    try:
                        logging.info("ClaudeProviderHandler: [0/3] CLI subprocess model discovery...")
                        config_dir = await ClaudeCliSessionManager.get_config_dir(
                            self.user_id, self.provider_id, cli_creds
                        )
                        cli_models = await self._cli_discover_models(config_dir)
                        if cli_models:
                            self._save_models_cache(cli_models)
                            logging.info(
                                f"ClaudeProviderHandler: ✓ CLI discovery returned {len(cli_models)} models"
                            )
                            return cli_models
                        logging.warning(
                            "ClaudeProviderHandler: CLI discovery returned no models, falling through"
                        )
                    except Exception as cli_disc_err:
                        logging.warning(
                            f"ClaudeProviderHandler: CLI model discovery failed: {cli_disc_err}"
                        )

            try:
                logging.info("ClaudeProviderHandler: [1/3] Attempting primary API endpoint...")

                headers = await self._get_auth_headers(stream=False)
                
                api_endpoint = 'https://api.anthropic.com/v1/models'
                logging.info(f"ClaudeProviderHandler: Calling API endpoint: {api_endpoint}")
                logging.info(f"ClaudeProviderHandler: Using OAuth2 authentication with full headers")
                
                response = await self.client.get(api_endpoint, headers=headers)
                
                logging.info(f"ClaudeProviderHandler: API response status: {response.status_code}")
                
                if response.status_code == 200:
                    models_data = response.json()
                    logging.info(f"ClaudeProviderHandler: ✓ Primary API call successful!")
                    logging.info(f"ClaudeProviderHandler: Response data keys: {list(models_data.keys())}")
                    logging.info(f"ClaudeProviderHandler: Retrieved {len(models_data.get('data', []))} models from API")
                    
                    if AISBF_DEBUG:
                        logging.info(f"ClaudeProviderHandler: Full API response: {models_data}")
                    
                    models = []
                    for model_data in models_data.get('data', []):
                        model_id = model_data.get('id', '')
                        display_name = model_data.get('display_name') or model_data.get('name') or model_id
                        
                        context_size = (
                            model_data.get('max_input_tokens') or
                            model_data.get('context_window') or
                            model_data.get('context_length') or
                            model_data.get('max_tokens')
                        )
                        
                        description = model_data.get('description')
                        
                        models.append(Model(
                            id=model_id,
                            name=display_name,
                            provider_id=self.provider_id,
                            context_size=context_size,
                            context_length=context_size,
                            description=description
                        ))
                        logging.info(f"ClaudeProviderHandler:   - {model_id} ({display_name}, context: {context_size})")
                    
                    if models:
                        self._save_models_cache(models)
                        
                        logging.info("=" * 80)
                        logging.info(f"ClaudeProviderHandler: ✓ SUCCESS - Returning {len(models)} models from primary API")
                        logging.info(f"ClaudeProviderHandler: Source: Dynamic API retrieval (Anthropic)")
                        logging.info("=" * 80)
                        return models
                    else:
                        logging.warning("ClaudeProviderHandler: ✗ Primary API returned empty model list")
                else:
                    logging.warning(f"ClaudeProviderHandler: ✗ Primary API call failed with status {response.status_code}")
                    try:
                        error_body = response.json()
                        logging.warning(f"ClaudeProviderHandler: Error response: {error_body}")
                    except Exception:
                        logging.warning(f"ClaudeProviderHandler: Error response (text): {response.text[:200]}")
            
            except Exception as api_error:
                logging.warning(f"ClaudeProviderHandler: ✗ Exception during primary API call")
                logging.warning(f"ClaudeProviderHandler: Error type: {type(api_error).__name__}")
                logging.warning(f"ClaudeProviderHandler: Error message: {str(api_error)}")
                if AISBF_DEBUG:
                    logging.warning(f"ClaudeProviderHandler: Full traceback:", exc_info=True)
            
            try:
                logging.info("-" * 80)
                logging.info("ClaudeProviderHandler: [2/3] Attempting fallback endpoint...")
                
                fallback_endpoint = 'http://lisa.nexlab.net:5000/claude/models'
                logging.info(f"ClaudeProviderHandler: Calling fallback endpoint: {fallback_endpoint}")
                
                fallback_client = httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=5.0))
                
                try:
                    fallback_response = await fallback_client.get(fallback_endpoint)
                    logging.info(f"ClaudeProviderHandler: Fallback response status: {fallback_response.status_code}")
                    
                    if fallback_response.status_code == 200:
                        fallback_data = fallback_response.json()
                        logging.info(f"ClaudeProviderHandler: ✓ Fallback API call successful!")
                        
                        if AISBF_DEBUG:
                            logging.info(f"ClaudeProviderHandler: Fallback response: {fallback_data}")
                        
                        models_list = fallback_data if isinstance(fallback_data, list) else fallback_data.get('data', fallback_data.get('models', []))
                        
                        models = []
                        for model_data in models_list:
                            if isinstance(model_data, str):
                                models.append(Model(id=model_data, name=model_data, provider_id=self.provider_id))
                            elif isinstance(model_data, dict):
                                model_id = model_data.get('id', model_data.get('model', ''))
                                display_name = model_data.get('name', model_data.get('display_name', model_id))
                                
                                context_size = (
                                    model_data.get('max_input_tokens') or
                                    model_data.get('context_window') or
                                    model_data.get('context_length') or
                                    model_data.get('context_size') or
                                    model_data.get('max_tokens')
                                )
                                
                                description = model_data.get('description')
                                
                                models.append(Model(
                                    id=model_id,
                                    name=display_name,
                                    provider_id=self.provider_id,
                                    context_size=context_size,
                                    context_length=context_size,
                                    description=description
                                ))
                        
                        if models:
                            for model in models:
                                logging.info(f"ClaudeProviderHandler:   - {model.id} ({model.name})")
                            
                            self._save_models_cache(models)
                            
                            logging.info("=" * 80)
                            logging.info(f"ClaudeProviderHandler: ✓ SUCCESS - Returning {len(models)} models from fallback API")
                            logging.info(f"ClaudeProviderHandler: Source: Dynamic API retrieval (Fallback)")
                            logging.info("=" * 80)
                            return models
                        else:
                            logging.warning("ClaudeProviderHandler: ✗ Fallback API returned empty model list")
                    else:
                        logging.warning(f"ClaudeProviderHandler: ✗ Fallback API call failed with status {fallback_response.status_code}")
                        try:
                            error_body = fallback_response.json()
                            logging.warning(f"ClaudeProviderHandler: Fallback error response: {error_body}")
                        except Exception:
                            logging.warning(f"ClaudeProviderHandler: Fallback error response (text): {fallback_response.text[:200]}")
                finally:
                    await fallback_client.aclose()
                    
            except Exception as fallback_error:
                logging.warning(f"ClaudeProviderHandler: ✗ Exception during fallback API call")
                logging.warning(f"ClaudeProviderHandler: Error type: {type(fallback_error).__name__}")
                logging.warning(f"ClaudeProviderHandler: Error message: {str(fallback_error)}")
                if AISBF_DEBUG:
                    logging.warning(f"ClaudeProviderHandler: Full traceback:", exc_info=True)
            
            logging.info("-" * 80)
            logging.info("ClaudeProviderHandler: [3/3] Attempting to load from cache...")
            
            cached_models = self._load_models_cache()
            if cached_models:
                for model in cached_models:
                    logging.info(f"ClaudeProviderHandler:   - {model.id} ({model.name})")
                
                logging.info("=" * 80)
                logging.info(f"ClaudeProviderHandler: ✓ Returning {len(cached_models)} models from cache")
                logging.info(f"ClaudeProviderHandler: Source: Cached model list")
                logging.info("=" * 80)
                return cached_models
            
            logging.info("-" * 80)
            logging.info("ClaudeProviderHandler: Using static fallback model list")
            static_models = [
                Model(id="claude-3-7-sonnet-20250219", name="Claude 3.7 Sonnet", provider_id=self.provider_id, context_size=200000, context_length=200000),
                Model(id="claude-3-5-sonnet-20241022", name="Claude 3.5 Sonnet", provider_id=self.provider_id, context_size=200000, context_length=200000),
                Model(id="claude-3-5-haiku-20241022", name="Claude 3.5 Haiku", provider_id=self.provider_id, context_size=200000, context_length=200000),
                Model(id="claude-3-opus-20240229", name="Claude 3 Opus", provider_id=self.provider_id, context_size=200000, context_length=200000),
            ]
            
            for model in static_models:
                logging.info(f"ClaudeProviderHandler:   - {model.id} ({model.name})")
            
            logging.info("=" * 80)
            logging.info(f"ClaudeProviderHandler: ✓ Returning {len(static_models)} models from static list")
            logging.info(f"ClaudeProviderHandler: Source: Static fallback configuration")
            logging.info("=" * 80)
            
            return static_models
        except Exception as e:
            import logging
            logging.error("=" * 80)
            logging.error(f"ClaudeProviderHandler: ✗ FATAL ERROR getting models: {str(e)}")
            logging.error("=" * 80)
            logging.error(f"ClaudeProviderHandler: Error details:", exc_info=True)
            raise e
