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

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

Base provider handler and shared utilities for all provider implementations.

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 asyncio
import time
import os
import random
from typing import Dict, List, Optional, Union, Any
from ..models import Provider, Model, ErrorTracking
from ..config import config
from ..utils import count_messages_tokens
from ..database import DatabaseRegistry
from ..batching import get_request_batcher

# Check if debug mode is enabled
AISBF_DEBUG = os.environ.get('AISBF_DEBUG', '').lower() in ('true', '1', 'yes')

# Request spacing state, keyed by "provider:user" (and ":model" for per-model
# limits). Process-wide because a handler instance lives for exactly one request,
# so per-instance timestamps can never space anything apart. See _throttle().
_rate_limit_last_request: Dict[str, float] = {}
_rate_limit_locks: Dict[str, asyncio.Lock] = {}


# Newer OpenAI models (reasoning models and the GPT-5/codex families) rejected the
# legacy `max_tokens` parameter and only accept `max_completion_tokens`. Older
# models (gpt-4o, gpt-4.1, gpt-3.5, and most OpenAI-compatible third-party
# endpoints) still expect `max_tokens`, so we cannot simply rename it everywhere.
_MAX_COMPLETION_TOKENS_PREFIXES = (
    'o1', 'o3', 'o4', 'gpt-5', 'codex-', 'gpt-image',
)

# Models discovered at runtime to reject a parameter, so the retry below is paid
# at most once per model name per process.
_max_completion_tokens_models = set()
_no_temperature_models = set()


def _normalize_model_name(model: str) -> str:
    """Strip vendor/route prefixes ('openai/gpt-5', 'azure/o3') and date suffixes."""
    name = (model or '').strip().lower()
    if '/' in name:
        name = name.rsplit('/', 1)[1]
    return name


def model_requires_max_completion_tokens(model: str) -> bool:
    name = _normalize_model_name(model)
    if name in _max_completion_tokens_models:
        return True
    return name.startswith(_MAX_COMPLETION_TOKENS_PREFIXES)


def apply_max_tokens_param(request_params: Dict, model: str, max_tokens: Optional[int]) -> None:
    """Set the output-token limit under whichever name the model accepts."""
    if max_tokens is None:
        return
    if model_requires_max_completion_tokens(model):
        request_params['max_completion_tokens'] = max_tokens
    else:
        request_params['max_tokens'] = max_tokens


def apply_temperature_param(request_params: Dict, model: str, temperature: Optional[float]) -> None:
    """Set `temperature` only when the model accepts the requested value.

    Reasoning models accept the default (1) but reject any other value, and some
    deployments reject the parameter outright. In both cases we omit it rather
    than fail the request — the upstream then applies its own default, which is
    exactly the value those models insist on.
    """
    if temperature is None:
        return
    if model_rejects_temperature(model, temperature):
        import logging
        logging.warning(
            f"Model '{model}' does not accept temperature={temperature}; "
            f"omitting it, so the upstream default (1) applies instead"
        )
        return
    request_params['temperature'] = temperature


def model_rejects_temperature(model: str, temperature: Optional[float]) -> bool:
    name = _normalize_model_name(model)
    if name in _no_temperature_models:
        return True
    if temperature is None or float(temperature) == 1.0:
        return False
    return name.startswith(_MAX_COMPLETION_TOKENS_PREFIXES)


def adapt_request_for_unsupported_param(request_params: Dict, model: str, exc: Exception) -> Optional[str]:
    """Rewrite a request that the upstream rejected over an unsupported parameter.

    The prefix heuristics above cannot know every model that dropped support
    (new releases, custom deployments, proxies), so we also learn from the error
    and remember the model, making the retry a once-per-model cost.

    Returns a short description of what was changed, or None when the error is
    unrelated — in which case the caller must re-raise instead of retrying.
    """
    message = str(exc).lower()
    name = _normalize_model_name(model)

    if 'max_completion_tokens' in message and 'max_tokens' in request_params:
        request_params['max_completion_tokens'] = request_params.pop('max_tokens')
        _max_completion_tokens_models.add(name)
        return 'max_tokens -> max_completion_tokens'

    if ('temperature' in message and 'temperature' in request_params and
            ('not supported' in message or 'unsupported' in message or
             'does not support' in message)):
        dropped = request_params.pop('temperature')
        _no_temperature_models.add(name)
        import logging
        logging.warning(
            f"Model '{model}' rejected temperature={dropped}; retrying without it, "
            f"so the upstream default applies instead"
        )
        return 'dropped temperature'

    return None


def is_provider_disabled_cheap(provider_id: str, user_id: Optional[int] = None) -> bool:
    """Return True if a provider is currently disabled — either manually (via the
    dashboard toggle) or by an auto-disable cooldown — WITHOUT constructing the
    handler or validating credentials.

    Building a handler validates credentials, which for some provider types
    (e.g. kiro/kilo) performs a network round-trip. The rotation/autoselect scan
    uses this to skip already-disabled providers *before* paying that cost, so a
    disabled provider is never contacted and a dashboard enable/disable toggle
    takes effect on the very next request. It reads the same cache/DB keys the
    handler uses, so it stays in sync with is_rate_limited().

    Fails safe: any cache/DB error returns False so a hiccup can never wrongly
    hide a provider — the authoritative is_rate_limited() check still runs after.
    """
    # 1) Manual disable (dashboard toggle): cache fast-path, DB is source of truth.
    scope = user_id if user_id is not None else 'global'
    manual_key = f"aisbf:provider_manual_disabled:{scope}:{provider_id}"
    try:
        from ..cache import get_cache_manager
        cached = get_cache_manager().get(manual_key)
    except Exception:
        cached = None
    if cached is not None:
        if bool(cached):
            return True
    else:
        try:
            db = DatabaseRegistry.get_config_database()
            if db and db.is_provider_manually_disabled(user_id, provider_id):
                return True
        except Exception:
            pass
    # 2) Auto-disable cooldown (failure threshold / usage limits).
    try:
        from ..cache import get_cache_manager
        disabled_until = get_cache_manager().get(f"aisbf:provider_disabled:{provider_id}")
        if disabled_until and disabled_until > time.time():
            return True
    except Exception:
        pass
    return False


class RateLimitError(Exception):
    """Raised when a provider signals a rate limit, regardless of HTTP status code (e.g. 429, 402)."""
    def __init__(self, message: str, status_code: int = None):
        super().__init__(message)
        self.status_code = status_code


# HTTP statuses that describe a problem with THIS request — bad input, an
# unsupported capability for the chosen model, or a rate limit — rather than the
# provider being unhealthy. They must not count toward the consecutive-failure
# budget that disables the whole provider: otherwise a handful of request-scoped
# errors (e.g. a model that lacks an image endpoint returning 404 "No endpoints
# found that support image input", or a bad image returning 400) take the whole
# provider offline for every request for the cooldown window.
_REQUEST_SCOPED_STATUSES = frozenset({400, 404, 413, 415, 422, 429})


def is_request_scoped_status(status) -> bool:
    """True when an HTTP status code is about the request, not provider health."""
    try:
        return int(status) in _REQUEST_SCOPED_STATUSES
    except (TypeError, ValueError):
        return False


def extract_http_status(error) -> Optional[int]:
    """Best-effort HTTP status code from a provider exception."""
    status = getattr(error, 'status_code', None)
    if status is None:
        status = getattr(getattr(error, 'response', None), 'status_code', None)
    try:
        return int(status) if status is not None else None
    except (TypeError, ValueError):
        return None


def is_request_scoped_error(error) -> bool:
    """True when the error is about the specific request (bad input / unsupported
    capability / rate limit) rather than the provider's health, so it must not
    count toward the failure cooldown that disables the whole provider.

    Falls back to message sniffing for handlers that wrap the upstream error in a
    plain Exception without preserving the status code (e.g. the streaming path).
    """
    if is_request_scoped_status(extract_http_status(error)):
        return True
    text = str(error or '')
    return (
        'No endpoints found that support image input' in text
        or 'Cannot decode visions' in text
        or '429 Too Many Requests' in text
    )


class AnthropicFormatConverter:
    """
    Shared utility class for converting between OpenAI and Anthropic message formats.
    Used by both AnthropicProviderHandler and ClaudeProviderHandler.
    
    All methods are static to allow usage without instantiation.
    """
    
    # Anthropic stop_reason → OpenAI finish_reason mapping
    STOP_REASON_MAP = {
        'end_turn': 'stop',
        'max_tokens': 'length',
        'stop_sequence': 'stop',
        'tool_use': 'tool_calls'
    }
    
    @staticmethod
    def sanitize_tool_call_id(tool_call_id: str) -> str:
        """Sanitize tool call ID for Anthropic API (alphanumeric, underscore, hyphen only)."""
        import re
        return re.sub(r'[^a-zA-Z0-9_-]', '_', tool_call_id)
    
    @staticmethod
    def filter_empty_content(content) -> Union[str, list, None]:
        """Filter empty content from messages for Anthropic API compatibility."""
        if content is None:
            return None
        if isinstance(content, str):
            return None if content.strip() == "" else content
        if isinstance(content, list):
            filtered = []
            for block in content:
                if isinstance(block, dict):
                    if block.get('type') == 'text':
                        text = block.get('text', '')
                        if text and text.strip():
                            filtered.append(block)
                    else:
                        filtered.append(block)
                else:
                    filtered.append(block)
            return filtered if filtered else None
        return content
    
    @staticmethod
    def extract_images_from_content(content) -> list:
        """
        Convert OpenAI image_url content blocks to Anthropic image source format.
        
        Handles:
        - data:image/jpeg;base64,... → {"type": "image", "source": {"type": "base64", ...}}
        - https://... → {"type": "image", "source": {"type": "url", ...}}
        """
        import logging
        logger = logging.getLogger(__name__)
        
        if not isinstance(content, list):
            return []
        
        images = []
        max_image_size = 5 * 1024 * 1024  # 5MB
        
        for block in content:
            if not isinstance(block, dict):
                continue
            if block.get('type') != 'image_url':
                continue
            
            image_url_obj = block.get('image_url', {})
            url = image_url_obj.get('url', '') if isinstance(image_url_obj, dict) else ''
            if not url:
                continue
            
            if url.startswith('data:'):
                try:
                    header, data = url.split(',', 1)
                    media_type = header.split(';')[0].replace('data:', '')
                    if len(data) > max_image_size:
                        logger.warning(f"Image too large ({len(data)} bytes), skipping")
                        continue
                    images.append({
                        'type': 'image',
                        'source': {'type': 'base64', 'media_type': media_type, 'data': data}
                    })
                except (ValueError, IndexError) as e:
                    logger.warning(f"Failed to parse data URL: {e}")
            elif url.startswith(('http://', 'https://')):
                images.append({
                    'type': 'image',
                    'source': {'type': 'url', 'url': url}
                })
            elif block.get('type') == 'image' and 'source' in block:
                images.append(block)
        
        return images
    
    @staticmethod
    def convert_messages_to_anthropic(messages: list, sanitize_ids: bool = True) -> tuple:
        """
        Convert OpenAI messages to Anthropic format.
        
        Handles:
        - System message extraction (separate 'system' parameter)
        - Tool role → user message with tool_result content blocks
        - Assistant tool_calls → tool_use content blocks
        - Multimodal content (images)
        - Empty content filtering
        
        Args:
            messages: OpenAI format messages
            sanitize_ids: Whether to sanitize tool call IDs
            
        Returns:
            Tuple of (system_message: str|None, anthropic_messages: list)
        """
        import logging
        import json
        
        system_message = None
        anthropic_messages = []
        
        for msg in messages:
            role = msg.get('role')
            content = msg.get('content')
            
            if role == 'system':
                system_message = content
                logging.info(f"AnthropicFormatConverter: Extracted system message ({len(content) if content else 0} chars)")
            
            elif role == 'tool':
                tool_call_id = msg.get('tool_call_id', msg.get('name', 'unknown'))
                tool_result_block = {
                    'type': 'tool_result',
                    'tool_use_id': tool_call_id,
                    'content': content or ""
                }
                
                if anthropic_messages and anthropic_messages[-1]['role'] == 'user':
                    last_content = anthropic_messages[-1]['content']
                    if isinstance(last_content, str):
                        anthropic_messages[-1]['content'] = [
                            {'type': 'text', 'text': last_content},
                            tool_result_block
                        ]
                    elif isinstance(last_content, list):
                        anthropic_messages[-1]['content'].append(tool_result_block)
                else:
                    anthropic_messages.append({
                        'role': 'user',
                        'content': [tool_result_block]
                    })
            
            elif role == 'assistant':
                tool_calls = msg.get('tool_calls')
                
                if tool_calls:
                    content_blocks = []
                    filtered = AnthropicFormatConverter.filter_empty_content(content)
                    if filtered:
                        if isinstance(filtered, str):
                            content_blocks.append({'type': 'text', 'text': filtered})
                        elif isinstance(filtered, list):
                            content_blocks.extend(filtered)
                    
                    for tc in tool_calls:
                        raw_id = tc.get('id', f"toolu_{len(content_blocks)}")
                        tool_id = AnthropicFormatConverter.sanitize_tool_call_id(raw_id) if sanitize_ids else raw_id
                        function = tc.get('function', {})
                        arguments = function.get('arguments', {})
                        if isinstance(arguments, str):
                            try:
                                arguments = json.loads(arguments)
                            except json.JSONDecodeError:
                                arguments = {}
                        
                        content_blocks.append({
                            'type': 'tool_use',
                            'id': tool_id,
                            'name': function.get('name', ''),
                            'input': arguments
                        })
                    
                    if content_blocks:
                        anthropic_messages.append({
                            'role': 'assistant',
                            'content': content_blocks
                        })
                else:
                    filtered = AnthropicFormatConverter.filter_empty_content(content)
                    if filtered is None:
                        continue
                    
                    if isinstance(filtered, list):
                        text_parts = []
                        for block in filtered:
                            if isinstance(block, dict):
                                text_parts.append(block.get('text', ''))
                            elif isinstance(block, str):
                                text_parts.append(block)
                        content_str = '\n'.join(text_parts)
                    else:
                        content_str = filtered or ""
                    
                    if content_str:
                        anthropic_messages.append({
                            'role': 'assistant',
                            'content': content_str
                        })
            
            elif role == 'user':
                if isinstance(content, list):
                    content_blocks = []
                    images = AnthropicFormatConverter.extract_images_from_content(content)
                    
                    for block in content:
                        if isinstance(block, dict):
                            btype = block.get('type', '')
                            if btype == 'text':
                                content_blocks.append(block)
                            elif btype not in ('image_url', 'image'):
                                content_blocks.append(block)
                        elif isinstance(block, str):
                            content_blocks.append({'type': 'text', 'text': block})
                    
                    content_blocks.extend(images)
                    anthropic_messages.append({
                        'role': 'user',
                        'content': content_blocks if content_blocks else content or ""
                    })
                else:
                    anthropic_messages.append({
                        'role': 'user',
                        'content': content or ""
                    })
            
            else:
                logging.warning(f"AnthropicFormatConverter: Unknown role '{role}', treating as user")
                anthropic_messages.append({
                    'role': 'user',
                    'content': content or ""
                })
        
        logging.info(f"AnthropicFormatConverter: Converted {len(messages)} OpenAI → {len(anthropic_messages)} Anthropic messages")
        return system_message, anthropic_messages
    
    @staticmethod
    def convert_tools_to_anthropic(tools: list) -> Optional[list]:
        """
        Convert OpenAI tools to Anthropic format with schema normalization.
        
        Normalizes:
        - ["string", "null"] → "string"
        - Removes additionalProperties: false
        - Cleans up required array for nullable fields
        """
        import logging
        
        if not tools:
            return None
        
        def normalize_schema(schema):
            if not isinstance(schema, dict):
                return schema
            result = {}
            for key, value in schema.items():
                if key == "type" and isinstance(value, list):
                    non_null = [t for t in value if t != "null"]
                    result[key] = non_null[0] if len(non_null) == 1 else (non_null if non_null else "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):
                    props = schema.get("properties", {})
                    cleaned = [f for f in value if f in props and not (isinstance(props.get(f, {}), dict) and isinstance(props[f].get("type"), list) and "null" in props[f]["type"])]
                    if cleaned:
                        result[key] = cleaned
                else:
                    result[key] = value
            return result
        
        anthropic_tools = []
        for tool in tools:
            if tool.get("type") == "function":
                function = tool.get("function", {})
                anthropic_tools.append({
                    "name": function.get("name", ""),
                    "description": function.get("description", ""),
                    "input_schema": normalize_schema(function.get("parameters", {}))
                })
                logging.info(f"AnthropicFormatConverter: Converted tool: {function.get('name')}")
        
        return anthropic_tools if anthropic_tools else None
    
    @staticmethod
    def convert_tool_choice_to_anthropic(tool_choice) -> Optional[dict]:
        """
        Convert OpenAI tool_choice to Anthropic format.
        
        "auto" → {"type": "auto"}
        "none" → None
        "required" → {"type": "any"}
        {"type": "function", "function": {"name": "X"}} → {"type": "tool", "name": "X"}
        """
        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: {tool_choice}")
                return {"type": "auto"}
        
        if isinstance(tool_choice, dict):
            if tool_choice.get("type") == "function":
                name = tool_choice.get("function", {}).get("name")
                return {"type": "tool", "name": name} if name else {"type": "auto"}
            return tool_choice
        
        return {"type": "auto"}
    
    @staticmethod
    def convert_anthropic_response_to_openai(response_data: dict, provider_id: str, model: str) -> dict:
        """
        Convert Anthropic API response (dict) to OpenAI chat completion format.
        
        Handles text blocks, tool_use blocks, thinking blocks, usage metadata, stop reasons.
        """
        import json
        import logging
        logger = logging.getLogger(__name__)
        
        content_text = ""
        tool_calls = []
        thinking_text = ""
        
        for block in response_data.get('content', []):
            btype = block.get('type', '')
            if btype == 'text':
                content_text += block.get('text', '')
            elif btype == '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 btype == 'thinking':
                thinking_text = block.get('thinking', '')
            elif btype == 'redacted_thinking':
                logger.debug("Found redacted_thinking block")
        
        stop_reason = response_data.get('stop_reason', 'end_turn')
        finish_reason = AnthropicFormatConverter.STOP_REASON_MAP.get(stop_reason, 'stop')
        
        usage = response_data.get('usage', {})
        input_tokens = usage.get('input_tokens', 0)
        output_tokens = usage.get('output_tokens', 0)
        cache_read = usage.get('cache_read_input_tokens', 0)
        cache_creation = usage.get('cache_creation_input_tokens', 0)
        
        openai_response = {
            'id': f"{provider_id}-{model}-{int(time.time())}",
            'object': 'chat.completion',
            'created': int(time.time()),
            'model': f'{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, '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}
            }
        
        return openai_response
    
    @staticmethod
    def convert_anthropic_sdk_response_to_openai(response, provider_id: str, model: str) -> dict:
        """
        Convert Anthropic SDK response object (with attributes) to OpenAI format.
        """
        import json
        import logging
        logger = logging.getLogger(__name__)
        
        content_text = ""
        tool_calls = []
        thinking_text = ""
        
        for block in getattr(response, 'content', []):
            btype = getattr(block, 'type', '')
            if btype == 'text' or hasattr(block, 'text'):
                content_text += getattr(block, 'text', '')
            elif btype == 'tool_use':
                raw_input = getattr(block, 'input', {})
                tool_calls.append({
                    'id': getattr(block, 'id', f"call_{len(tool_calls)}"),
                    'type': 'function',
                    'function': {
                        'name': getattr(block, 'name', ''),
                        'arguments': json.dumps(raw_input) if isinstance(raw_input, dict) else str(raw_input)
                    }
                })
            elif btype == 'thinking':
                thinking_text = getattr(block, 'thinking', '')
        
        stop_reason = getattr(response, 'stop_reason', 'end_turn') or 'end_turn'
        finish_reason = AnthropicFormatConverter.STOP_REASON_MAP.get(stop_reason, 'stop')
        
        usage_obj = getattr(response, 'usage', None)
        input_tokens = getattr(usage_obj, 'input_tokens', 0) or 0 if usage_obj else 0
        output_tokens = getattr(usage_obj, 'output_tokens', 0) or 0 if usage_obj else 0
        cache_read = getattr(usage_obj, 'cache_read_input_tokens', 0) or 0 if usage_obj else 0
        cache_creation = getattr(usage_obj, 'cache_creation_input_tokens', 0) or 0 if usage_obj else 0
        
        openai_response = {
            'id': getattr(response, 'id', f"{provider_id}-{model}-{int(time.time())}"),
            'object': 'chat.completion',
            'created': int(time.time()),
            'model': f'{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, '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}
            }
        
        return openai_response


class AdaptiveRateLimiter:
    """
    Adaptive Rate Limiter that learns optimal rate limits from 429 responses.
    
    Features:
    - Tracks 429 patterns per provider
    - Implements exponential backoff with jitter for retries
    - Learns optimal rate limits from historical 429 data
    - Adds rate limit headroom (stays below limits)
    - Gradually recovers rate limits after cooldown periods
    """
    
    def __init__(self, provider_id: str, config: Dict = None):
        self.provider_id = provider_id
        
        # Configuration with defaults
        self.enabled = config.get('enabled', True) if config else True
        self.initial_rate_limit = config.get('initial_rate_limit', 0) if config else 0
        self.learning_rate = config.get('learning_rate', 0.1) if config else 0.1
        self.headroom_percent = config.get('headroom_percent', 10) if config else 10  # Stay 10% below learned limit
        self.recovery_rate = config.get('recovery_rate', 0.05) if config else 0.05  # 5% recovery per successful request
        self.max_rate_limit = config.get('max_rate_limit', 60) if config else 60  # Max 60 seconds between requests
        self.min_rate_limit = config.get('min_rate_limit', 0.1) if config else 0.1  # Min 0.1 seconds between requests
        self.backoff_base = config.get('backoff_base', 2) if config else 2
        self.jitter_factor = config.get('jitter_factor', 0.25) if config else 0.25  # 25% jitter
        self.history_window = config.get('history_window', 3600) if config else 3600  # 1 hour history window
        self.consecutive_successes_for_recovery = config.get('consecutive_successes_for_recovery', 10) if config else 10
        
        # Learned rate limit (starts with configured value)
        self.current_rate_limit = self.initial_rate_limit
        self.base_rate_limit = self.initial_rate_limit  # Original configured limit
        
        # 429 tracking
        self._429_history = []  # List of (timestamp, wait_seconds) tuples
        self._consecutive_429s = 0
        self._consecutive_successes = 0
        
        # Statistics
        self.total_429_count = 0
        self.total_requests = 0
        self.last_429_time = None
        
    def record_429(self, wait_seconds: int):
        """Record a 429 response and adjust rate limit accordingly."""
        import logging
        logger = logging.getLogger(__name__)
        
        current_time = time.time()
        
        # Record this 429 in history
        self._429_history.append((current_time, wait_seconds))
        self.total_429_count += 1
        self._consecutive_429s += 1
        self._consecutive_successes = 0
        self.last_429_time = current_time
        
        # Clean old history
        self._cleanup_history()
        
        # Calculate new rate limit using exponential backoff
        # New limit = current_limit * backoff_base + wait_seconds from server
        new_limit = self.current_rate_limit * self.backoff_base + wait_seconds
        
        # Apply learning rate adjustment
        new_limit = self.current_rate_limit + (new_limit - self.current_rate_limit) * self.learning_rate
        
        # Apply headroom (stay below the limit)
        new_limit = new_limit * (1 - self.headroom_percent / 100)
        
        # Clamp to min/max
        self.current_rate_limit = max(self.min_rate_limit, min(self.max_rate_limit, new_limit))
        
        logger.info(f"[AdaptiveRateLimiter {self.provider_id}] 429 recorded: wait_seconds={wait_seconds}, "
                   f"new_rate_limit={self.current_rate_limit:.2f}s, consecutive_429s={self._consecutive_429s}")
        self.save_to_cache()

    def record_success(self):
        """Record a successful request and gradually recover rate limit."""
        import logging
        logger = logging.getLogger(__name__)
        
        self.total_requests += 1
        self._consecutive_successes += 1
        self._consecutive_429s = 0
        
        # Gradually recover rate limit after successful requests
        if self._consecutive_successes >= self.consecutive_successes_for_recovery:
            # Recovery: move back towards base rate limit
            if self.current_rate_limit < self.base_rate_limit:
                old_limit = self.current_rate_limit
                self.current_rate_limit = self.current_rate_limit + (self.base_rate_limit - self.current_rate_limit) * self.recovery_rate
                # Clamp to not exceed base
                self.current_rate_limit = min(self.current_rate_limit, self.base_rate_limit)
                
                if old_limit != self.current_rate_limit:
                    logger.info(f"[AdaptiveRateLimiter {self.provider_id}] Rate limit recovery: "
                               f"{old_limit:.2f}s -> {self.current_rate_limit:.2f}s")
                
                # Reset consecutive successes counter after recovery
                self._consecutive_successes = 0
                self.save_to_cache()

    def get_rate_limit(self) -> float:
        """Get the current adaptive rate limit."""
        return self.current_rate_limit
    
    def get_wait_time(self) -> float:
        """Get the wait time before next request based on adaptive rate limiting."""
        if not self.enabled or self.current_rate_limit <= 0:
            return 0
        
        # Use current adaptive rate limit
        return self.current_rate_limit
    
    def calculate_backoff_with_jitter(self, attempt: int, base_wait: int = None) -> float:
        """
        Calculate exponential backoff wait time with jitter.
        
        Args:
            attempt: Current retry attempt number (0-indexed)
            base_wait: Optional base wait time from server response
            
        Returns:
            Wait time in seconds with jitter applied
        """
        import logging
        logger = logging.getLogger(__name__)
        
        # Calculate exponential backoff
        if base_wait is not None and base_wait > 0:
            # Use server-provided wait time as base
            wait_time = base_wait
        else:
            # Use exponential backoff: base * 2^attempt
            wait_time = self.backoff_base ** attempt
        
        # Apply jitter: random factor between (1 - jitter_factor) and (1 + jitter_factor)
        jitter_multiplier = 1 + random.uniform(-self.jitter_factor, self.jitter_factor)
        wait_time = wait_time * jitter_multiplier
        
        # Clamp to reasonable limits (1 second to 300 seconds)
        wait_time = max(1, min(300, wait_time))
        
        logger.info(f"[AdaptiveRateLimiter {self.provider_id}] Backoff calculation: attempt={attempt}, "
                   f"base_wait={base_wait}, jitter_multiplier={jitter_multiplier:.2f}, "
                   f"final_wait={wait_time:.2f}s")
        
        return wait_time
    
    def _cleanup_history(self):
        """Remove old entries from 429 history."""
        current_time = time.time()
        cutoff_time = current_time - self.history_window
        self._429_history = [(ts, ws) for ts, ws in self._429_history if ts > cutoff_time]
    
    def get_stats(self) -> Dict:
        """Get rate limiter statistics."""
        self._cleanup_history()
        
        return {
            'provider_id': self.provider_id,
            'enabled': self.enabled,
            'current_rate_limit': self.current_rate_limit,
            'base_rate_limit': self.base_rate_limit,
            'total_429_count': self.total_429_count,
            'total_requests': self.total_requests,
            'consecutive_429s': self._consecutive_429s,
            'consecutive_successes': self._consecutive_successes,
            'recent_429_count': len(self._429_history),
            'last_429_time': self.last_429_time
        }
    
    def reset(self):
        """Reset the adaptive rate limiter to initial state."""
        import logging
        logger = logging.getLogger(__name__)

        self.current_rate_limit = self.initial_rate_limit
        self._429_history = []
        self._consecutive_429s = 0
        self._consecutive_successes = 0
        self.total_429_count = 0
        self.total_requests = 0
        self.last_429_time = None

        logger.info(f"[AdaptiveRateLimiter {self.provider_id}] Reset to initial state")
        self._clear_cache()

    def _cache_key(self) -> str:
        return f"aisbf:adaptive_rate_limiter:{self.provider_id}"

    def save_to_cache(self):
        """Persist rate limiter state to the cache so it survives restarts."""
        try:
            from ..cache import get_cache_manager
            cache = get_cache_manager()
            state = {
                'current_rate_limit': self.current_rate_limit,
                'consecutive_429s': self._consecutive_429s,
                'total_429_count': self.total_429_count,
                'last_429_time': self.last_429_time,
                'history': self._429_history[-50:],  # keep last 50 entries
            }
            cache.set(self._cache_key(), state, ttl=86400 * 7)
        except Exception:
            pass

    def load_from_cache(self):
        """Restore rate limiter state from cache after a restart."""
        try:
            from ..cache import get_cache_manager
            cache = get_cache_manager()
            state = cache.get(self._cache_key())
            if not state:
                return
            self.current_rate_limit = state.get('current_rate_limit', self.initial_rate_limit)
            self._consecutive_429s = state.get('consecutive_429s', 0)
            self.total_429_count = state.get('total_429_count', 0)
            self.last_429_time = state.get('last_429_time')
            self._429_history = state.get('history', [])
        except Exception:
            pass

    def _clear_cache(self):
        try:
            from ..cache import get_cache_manager
            get_cache_manager().delete(self._cache_key())
        except Exception:
            pass


# Global adaptive rate limiters registry - now supports user-specific limiters
# Key format: "provider_id" (global) or "user:user_id:provider_id" (user-specific)
_adaptive_rate_limiters: Dict[str, AdaptiveRateLimiter] = {}


def get_adaptive_rate_limiter(provider_id: str, config: Dict = None, user_id: Optional[int] = None) -> AdaptiveRateLimiter:
    """Get or create an adaptive rate limiter for a provider, optionally user-specific."""
    global _adaptive_rate_limiters
    
    # Create key based on whether user_id is provided
    if user_id is not None:
        key = f"user:{user_id}:{provider_id}"
    else:
        key = provider_id
    
    if key not in _adaptive_rate_limiters:
        limiter = AdaptiveRateLimiter(key, config)
        limiter.load_from_cache()
        _adaptive_rate_limiters[key] = limiter

    return _adaptive_rate_limiters[key]


def get_all_adaptive_rate_limiters(user_id: Optional[int] = None) -> Dict[str, AdaptiveRateLimiter]:
    """Get all adaptive rate limiters, optionally filtered by user."""
    global _adaptive_rate_limiters
    
    if user_id is None:
        # Return all limiters (admin view)
        return _adaptive_rate_limiters
    else:
        # Return only limiters for this user + global ones
        user_prefix = f"user:{user_id}:"
        filtered = {}
        for key, limiter in _adaptive_rate_limiters.items():
            if key.startswith(user_prefix) or not key.startswith("user:"):
                # Strip the user prefix for display to user
                if key.startswith(user_prefix):
                    display_key = key[len(user_prefix):]
                else:
                    display_key = key
                filtered[display_key] = limiter
        return filtered


class BaseProviderHandler:
    @staticmethod
    def build_credentials_file(provider_type: str, provider_id: str) -> str:
        """Return the standard per-provider credential file path.

        Using both provider_type and provider_id guarantees a unique path
        even when multiple providers share the same type.
        """
        return os.path.expanduser(f"~/.aisbf/{provider_type}_{provider_id}_credentials.json")

    def __init__(self, provider_id: str, api_key: Optional[str] = None, user_id: Optional[int] = None):
        self.provider_id = provider_id
        self.api_key = api_key
        self.user_id = user_id
        self.user_provider_config = None
        
        # Get error tracking config - use user-specific or default to global
        if user_id is not None:
            # Default error tracking for user providers
            self.error_tracking = {
                "enabled": True,
                "max_errors": 5,
                "cooldown_seconds": 60,
                "failures": 0,
                "last_failure": 0,
                "disabled_until": None
            }
        else:
            # Fall back to defaults for pre-configured handler types not in providers.json
            self.error_tracking = config.error_tracking.get(provider_id, {
                "failures": 0,
                "last_failure": None,
                "disabled_until": None,
            })
            
        self.last_request_time = 0
        
        # Get rate limit - use user-specific or default to global
        if user_id is not None and hasattr(self, 'user_provider_config') and self.user_provider_config is not None:
            # User provider config (dict)
            self.rate_limit = self.user_provider_config.get('rate_limit', 60)
        elif user_id is not None:
            # Default rate limit for user providers
            self.rate_limit = 60
        else:
            _pcfg = config.providers.get(provider_id)
            self.rate_limit = _pcfg.rate_limit if _pcfg else 0
        # Add model-level rate limit tracking
        self.model_last_request_time = {}  # {model_name: timestamp}
        # Token usage tracking for rate limits
        self.token_usage = {}  # {model_name: {"TPM": [], "TPH": [], "TPD": []}}
        # Initialize batcher
        self.batcher = get_request_batcher()
        # Initialize adaptive rate limiter
        adaptive_config = None
        if config.aisbf and config.aisbf.adaptive_rate_limiting:
            adaptive_obj = config.aisbf.adaptive_rate_limiting
            adaptive_config = adaptive_obj.model_dump() if hasattr(adaptive_obj, 'model_dump') else adaptive_obj.dict()
        self.adaptive_limiter = get_adaptive_rate_limiter(provider_id, adaptive_config, user_id)
        # Load rate-limit disabled state from cache (persists across restarts)
        self._load_disabled_until_from_cache()
        # Load manual (admin/user-triggered) disable flag from cache
        self._manual_disabled: bool = False
        self._load_manual_disabled_from_cache()
        # Load usage-based disabled state from DB (persists across restarts)
        self._usage_disabled_until: Optional[float] = None
        try:
            db = DatabaseRegistry.get_config_database()
            if db:
                self._usage_disabled_until = db.get_provider_disabled_until(user_id, provider_id)
        except Exception:
            pass
    
    def validate_credentials(self) -> bool:
        """
        Validate provider credentials.
        
        Returns:
            True if credentials are valid or validation is not needed.
            False if credentials are invalid/missing.
        
        Base implementation checks only if api_key_required=True and api_key is missing/empty.
        Override in subclasses for provider-specific validation (e.g., Kiro credential files).
        """
        import logging
        logger = logging.getLogger(__name__)
        
        # Determine which config to use: user-specific or global
        is_user_context = self.user_id is not None and hasattr(self, 'user_provider_config') and self.user_provider_config is not None
        provider_config = self.user_provider_config if is_user_context else None
        
        if provider_config is None:
            from ..config import config
            provider_config = config.providers.get(self.provider_id)
        
        if provider_config is None:
            logger.error(f"[{self.provider_id}] Provider configuration not found")
            return False
        
        # Check if this provider requires authentication
        if isinstance(provider_config, dict):
            api_key_required = provider_config.get('api_key_required', False)
        else:
            api_key_required = getattr(provider_config, 'api_key_required', False)
        
        if not api_key_required:
            logger.debug(f"[{self.provider_id}] No API key required, skipping credential validation")
            return True
        
        # Check if API key is provided (either from config or passed to constructor)
        if not self.api_key:
            # Also check if it might be in config
            if isinstance(provider_config, dict):
                api_key_from_config = provider_config.get('api_key')
            else:
                api_key_from_config = getattr(provider_config, 'api_key', None)
            if api_key_from_config:
                self.api_key = api_key_from_config
            else:
                logger.error(f"[{self.provider_id}] API key required but not provided")
                return False
        
        # Check for placeholder/empty API key
        if isinstance(self.api_key, str):
            stripped = self.api_key.strip()
            if not stripped or stripped.startswith('YOUR_'):
                logger.error(f"[{self.provider_id}] Invalid API key format")
                return False
        
        logger.info(f"[{self.provider_id}] API key present, validation passed")
        return True

    def _native_cache_user_allows(self, model_name: Optional[str] = None) -> bool:
        """Return whether user-scoped native prompt caching is allowed for this provider/model."""
        if self.user_id is None:
            return True
        try:
            db = DatabaseRegistry.get_config_database()
            user_setting = db.get_user_cache_settings(self.user_id, self.provider_id, model_name)
            return bool(user_setting.get('cache_enabled', True))
        except Exception:
            return True
    
    def parse_429_response(self, response_data: Union[Dict, str], headers: Dict = None) -> Optional[int]:
        """
        Parse 429 rate limit response to extract wait time in seconds.
        
        Checks multiple sources:
        1. Retry-After header (seconds or HTTP date)
        2. X-RateLimit-Reset header (Unix timestamp)
        3. Response body fields (retry_after, reset_time, etc.)
        4. X-RateLimit-* headers for auto-configuration
        
        Returns:
            Wait time in seconds, or None if cannot be determined
        """
        import logging
        import re
        from email.utils import parsedate_to_datetime
        from datetime import datetime, timezone
        
        logger = logging.getLogger(__name__)
        logger.info("=== PARSING 429 RATE LIMIT RESPONSE ===")
        
        wait_seconds = None
        rate_limit_headers = {}  # Store rate limit headers for auto-configuration
        
        # Check for rate limit headers (for auto-configuration)
        if headers:
            rate_limit_headers = {
                'limit': headers.get('X-RateLimit-Limit') or headers.get('x-ratelimit-limit'),
                'remaining': headers.get('X-RateLimit-Remaining') or headers.get('x-ratelimit-remaining'),
                'reset': headers.get('X-RateLimit-Reset') or headers.get('x-ratelimit-reset'),
                'reset_at': headers.get('X-RateLimit-Reset-After') or headers.get('x-ratelimit-reset-after')
            }
            logger.info(f"Rate limit headers found: {rate_limit_headers}")
        
        # Check Retry-After header
        if headers:
            retry_after = headers.get('Retry-After') or headers.get('retry-after')
            if retry_after:
                logger.info(f"Found Retry-After header: {retry_after}")
                try:
                    # Try parsing as integer (seconds)
                    wait_seconds = int(retry_after)
                    logger.info(f"Parsed Retry-After as seconds: {wait_seconds}")
                except ValueError:
                    # Try parsing as HTTP date
                    try:
                        retry_date = parsedate_to_datetime(retry_after)
                        now = datetime.now(timezone.utc)
                        wait_seconds = int((retry_date - now).total_seconds())
                        logger.info(f"Parsed Retry-After as date, wait seconds: {wait_seconds}")
                    except Exception as e:
                        logger.warning(f"Failed to parse Retry-After header: {e}")
            
            # Check X-RateLimit-Reset header (Unix timestamp)
            if not wait_seconds:
                reset_time = headers.get('X-RateLimit-Reset') or headers.get('x-ratelimit-reset')
                if reset_time:
                    logger.info(f"Found X-RateLimit-Reset header: {reset_time}")
                    try:
                        reset_timestamp = int(reset_time)
                        now_timestamp = int(time.time())
                        wait_seconds = reset_timestamp - now_timestamp
                        logger.info(f"Calculated wait from reset timestamp: {wait_seconds} seconds")
                    except Exception as e:
                        logger.warning(f"Failed to parse X-RateLimit-Reset header: {e}")
        
        # Normalize response_data into a dict and/or raw string for parsing
        body_dict = None
        body_str = None
        if isinstance(response_data, dict):
            body_dict = response_data
        elif isinstance(response_data, (str, bytes)):
            body_str = response_data.decode('utf-8') if isinstance(response_data, bytes) else response_data
            try:
                import json as _json
                parsed = _json.loads(body_str)
                if isinstance(parsed, dict):
                    body_dict = parsed
                    logger.info("Parsed response body string as JSON dict")
            except (ValueError, TypeError):
                logger.info("Response body is not valid JSON, will apply regex to raw string")

        # Check response body (structured fields)
        if not wait_seconds and body_dict:
            logger.info(f"Checking response body for rate limit info: {body_dict}")

            # Common field names for retry/reset time
            retry_fields = [
                'retry_after', 'retryAfter', 'retry_after_seconds',
                'wait_seconds', 'waitSeconds', 'retry_in'
            ]
            reset_fields = [
                'reset_time', 'resetTime', 'reset_at', 'resetAt',
                'reset_timestamp', 'resetTimestamp'
            ]

            # Check retry fields (direct seconds)
            for field in retry_fields:
                if field in body_dict:
                    try:
                        wait_seconds = int(body_dict[field])
                        logger.info(f"Found {field} in response body: {wait_seconds} seconds")
                        break
                    except (ValueError, TypeError) as e:
                        logger.warning(f"Failed to parse {field}: {e}")

            # Check reset fields (timestamp)
            if not wait_seconds:
                for field in reset_fields:
                    if field in body_dict:
                        try:
                            reset_timestamp = int(body_dict[field])
                            now_timestamp = int(time.time())
                            wait_seconds = reset_timestamp - now_timestamp
                            logger.info(f"Found {field} in response body, calculated wait: {wait_seconds} seconds")
                            break
                        except (ValueError, TypeError) as e:
                            logger.warning(f"Failed to parse {field}: {e}")

            # Check reason field for known rate limit reason codes
            if not wait_seconds:
                reason = body_dict.get('reason') or body_dict.get('error_code') or body_dict.get('code', '')
                if isinstance(reason, str):
                    reason_upper = reason.upper()
                    reason_wait_map = {
                        'MONTHLY_REQUEST_COUNT': 86400,   # daily fallback; actual reset is monthly
                        'DAILY_REQUEST_COUNT': 3600,      # hourly fallback; actual reset is daily
                        'HOURLY_REQUEST_COUNT': 600,
                        'RATE_LIMIT_EXCEEDED': 60,
                        'TOO_MANY_REQUESTS': 60,
                        'QUOTA_EXCEEDED': 3600,
                    }
                    for key, secs in reason_wait_map.items():
                        if key in reason_upper:
                            wait_seconds = secs
                            logger.info(f"Inferred wait time from reason '{reason}': {wait_seconds} seconds")
                            break

            # Extract message string from dict for regex matching below
            if not body_str:
                error_field = body_dict.get('error')
                if isinstance(error_field, dict):
                    body_str = error_field.get('message', '') or body_dict.get('message', '')
                elif isinstance(error_field, str):
                    body_str = error_field
                else:
                    body_str = body_dict.get('message', '')

        # Apply regex patterns to any available string (raw body or extracted message)
        if not wait_seconds and body_str:
            logger.info(f"Checking string body for time patterns: {body_str[:500]}")
            time_patterns = [
                r'try again in (\d+)\s*(second|minute|hour|day)s?',
                r'retry after (\d+)\s*(second|minute|hour|day)s?',
                r'wait (\d+)\s*(second|minute|hour|day)s?',
                r'available in (\d+)\s*(second|minute|hour|day)s?',
            ]
            multipliers = {'second': 1, 'minute': 60, 'hour': 3600, 'day': 86400}
            for pattern in time_patterns:
                match = re.search(pattern, body_str, re.IGNORECASE)
                if match:
                    value = int(match.group(1))
                    unit = match.group(2).lower()
                    wait_seconds = value * multipliers.get(unit, 1)
                    logger.info(f"Extracted wait time from string body: {value} {unit}(s) = {wait_seconds} seconds")
                    break
        
        # Ensure wait_seconds is positive and reasonable
        if wait_seconds:
            if wait_seconds < 0:
                logger.warning(f"Calculated negative wait time: {wait_seconds}, setting to 60 seconds")
                wait_seconds = 60
            elif wait_seconds > 86400:  # More than 1 day
                logger.warning(f"Calculated very long wait time: {wait_seconds}, capping at 1 day")
                wait_seconds = 86400
            
            logger.info(f"Final parsed wait time: {wait_seconds} seconds")
        else:
            logger.warning("Could not determine wait time from 429 response, using default 60 seconds")
            wait_seconds = 60
        
        logger.info("=== END PARSING 429 RATE LIMIT RESPONSE ===")
        return wait_seconds
    
    def handle_429_error(self, response_data: Union[Dict, str] = None, headers: Dict = None):
        """
        Handle 429 rate limit error by parsing the response and disabling provider
        for the appropriate duration. Also records the 429 in the adaptive rate limiter.
        """
        import logging
        logger = logging.getLogger(__name__)
        
        logger.error("=== 429 RATE LIMIT ERROR DETECTED ===")
        logger.error(f"Provider: {self.provider_id}")
        
        # Parse the response to get wait time
        wait_seconds = self.parse_429_response(response_data, headers)
        
        # Record 429 in adaptive rate limiter for learning
        self.adaptive_limiter.record_429(wait_seconds)
        
        # Check for rate limit headers and auto-configure if not already set
        if headers:
            self._auto_configure_rate_limits(headers)
        
        # Disable provider for the calculated duration
        self._save_disabled_until(time.time() + wait_seconds)

        logger.error(f"!!! PROVIDER DISABLED DUE TO RATE LIMIT !!!")
        logger.error(f"Provider: {self.provider_id}")
        logger.error(f"Reason: 429 Too Many Requests")
        logger.error(f"Disabled for: {wait_seconds} seconds ({wait_seconds / 60:.1f} minutes)")
        logger.error(f"Disabled until: {self.error_tracking['disabled_until']}")
        logger.error(f"Adaptive rate limit: {self.adaptive_limiter.current_rate_limit:.2f}s")
        logger.error(f"Provider will be automatically re-enabled after cooldown")
        logger.error("=== END 429 RATE LIMIT ERROR ===")

    _RATE_LIMIT_REASON_KEYWORDS = ('limit', 'quota', 'count', 'exceeded', 'reached', 'allowance', 'throttl')

    def is_rate_limit_response(self, status_code: int, response_data: Union[Dict, str] = None, headers: Dict = None) -> bool:
        """
        Return True if the response should be treated as a rate limit regardless of status code.

        Handles status 429 (always) and status 402 / other codes when the body contains
        a recognisable limit/quota reason.
        """
        if status_code == 429:
            return True

        if isinstance(response_data, dict):
            message = response_data.get('message', '') or ''
            reason  = response_data.get('reason', '')  or ''
            error   = response_data.get('error', '')   or ''
            if isinstance(error, dict):
                error = error.get('message', '') or ''
            combined = (message + ' ' + reason + ' ' + error).lower()
            if any(kw in combined for kw in self._RATE_LIMIT_REASON_KEYWORDS):
                return True

        return False

    def handle_rate_limit_response(self, status_code: int, response_data: Union[Dict, str] = None, headers: Dict = None):
        """
        Detect a non-429 rate-limit response, disable the provider, and raise RateLimitError.

        Use this instead of handle_429_error when the status code may differ (e.g. 402).
        """
        import logging
        logger = logging.getLogger(__name__)

        reason_str = ''
        if isinstance(response_data, dict):
            reason_str = f" reason={response_data.get('reason', '')} message={response_data.get('message', '')}"

        logger.warning(f"=== RATE LIMIT DETECTED (HTTP {status_code}) ===")
        logger.warning(f"Provider: {self.provider_id}{reason_str}")
        logger.warning(f"Treating HTTP {status_code} as rate limit — disabling provider")

        self.handle_429_error(response_data, headers)

        raise RateLimitError(
            f"Provider {self.provider_id} rate limited (HTTP {status_code}){reason_str}",
            status_code=status_code
        )

    def _auto_configure_rate_limits(self, headers: Dict = None):
        """
        Auto-configure rate limits from response headers if not already configured.
        """
        import logging
        from ..config import config
        
        logger = logging.getLogger(__name__)
        
        if not headers:
            return
        
        # Extract rate limit headers
        rate_limit_header = headers.get('X-RateLimit-Limit') or headers.get('x-ratelimit-limit')
        remaining_header = headers.get('X-RateLimit-Remaining') or headers.get('x-ratelimit-remaining')
        reset_header = headers.get('X-RateLimit-Reset') or headers.get('x-ratelimit-reset')
        
        if not rate_limit_header:
            logger.debug("No X-RateLimit-Limit header found, skipping auto-configuration")
            return
        
        try:
            rate_limit_value = int(rate_limit_header)
            logger.info(f"Found rate limit header: {rate_limit_value} requests")
            
            # Get current provider config
            provider_config = config.providers.get(self.provider_id)
            if not provider_config:
                logger.debug(f"Provider {self.provider_id} not found in config")
                return
            
            # Check if we don't have a rate limit configured
            current_rate_limit = getattr(provider_config, 'rate_limit', None)
            if current_rate_limit is None or current_rate_limit == 0:
                # Calculate: use 80% of the limit to stay below it
                auto_rate_limit = rate_limit_value * 0.8
                
                logger.info(f"Auto-configuring rate limit for {self.provider_id}: {auto_rate_limit:.1f}s (from header limit: {rate_limit_value})")
                
                # Try to save to config (this may not persist if config is immutable)
                try:
                    # Update the in-memory config
                    if hasattr(provider_config, 'rate_limit'):
                        provider_config.rate_limit = auto_rate_limit
                        logger.info(f"✓ Auto-configured rate_limit: {auto_rate_limit:.1f}s for provider {self.provider_id}")
                except Exception as e:
                    logger.debug(f"Could not auto-configure rate limit: {e}")
            else:
                logger.debug(f"Rate limit already configured ({current_rate_limit}), skipping auto-configuration")
                
        except (ValueError, TypeError) as e:
            logger.debug(f"Could not parse rate limit header: {e}")

    def _disabled_cache_key(self) -> str:
        return f"aisbf:provider_disabled:{self.adaptive_limiter.provider_id}"

    def _save_disabled_until(self, disabled_until: float):
        """Set disabled_until in error_tracking and persist it to the cache."""
        self.error_tracking['disabled_until'] = disabled_until
        try:
            from ..cache import get_cache_manager
            ttl = max(1, int(disabled_until - time.time()) + 60)
            get_cache_manager().set(self._disabled_cache_key(), disabled_until, ttl=ttl)
        except Exception:
            pass

    def _clear_disabled_until(self):
        """Clear disabled_until from error_tracking and the cache."""
        self.error_tracking['disabled_until'] = None
        try:
            from ..cache import get_cache_manager
            get_cache_manager().delete(self._disabled_cache_key())
        except Exception:
            pass

    def _load_disabled_until_from_cache(self):
        """Restore disabled_until from cache on startup."""
        try:
            from ..cache import get_cache_manager
            disabled_until = get_cache_manager().get(self._disabled_cache_key())
            if disabled_until and disabled_until > time.time():
                self.error_tracking['disabled_until'] = disabled_until
        except Exception:
            pass

    def _manual_disabled_cache_key(self) -> str:
        # User-scoped so a user's manual toggle never affects the global provider.
        scope = self.user_id if self.user_id is not None else 'global'
        return f"aisbf:provider_manual_disabled:{scope}:{self.provider_id}"

    def _load_manual_disabled_from_cache(self):
        """Restore the manual-disable flag on startup.

        The cache is a fast path; the database is the source of truth so the state
        survives reboots and cache flushes. On a cache miss we read the DB and warm
        the cache (storing the boolean either way)."""
        try:
            from ..cache import get_cache_manager
            cached = get_cache_manager().get(self._manual_disabled_cache_key())
        except Exception:
            cached = None
        if cached is not None:
            self._manual_disabled = bool(cached)
            return
        # Cache miss → authoritative database lookup
        value = False
        try:
            db = DatabaseRegistry.get_config_database()
            if db:
                value = bool(db.is_provider_manually_disabled(self.user_id, self.provider_id))
        except Exception:
            value = False
        self._manual_disabled = value
        try:
            from ..cache import get_cache_manager
            get_cache_manager().set(self._manual_disabled_cache_key(), value, ttl=10 * 365 * 24 * 3600)
        except Exception:
            pass

    def manual_disable(self):
        """Manually disable this provider until it is manually re-enabled.

        Unlike the failure cooldown this never auto-expires. It is persisted in the
        database (source of truth, survives reboot) and mirrored to the cache."""
        import logging
        self._manual_disabled = True
        try:
            db = DatabaseRegistry.get_config_database()
            if db:
                db.set_provider_manually_disabled(self.user_id, self.provider_id)
        except Exception as e:
            logging.getLogger(__name__).warning(f"Failed to persist manual disable for {self.provider_id}: {e}")
        try:
            from ..cache import get_cache_manager
            get_cache_manager().set(self._manual_disabled_cache_key(), True, ttl=10 * 365 * 24 * 3600)
        except Exception:
            pass
        logging.getLogger(__name__).warning(f"Provider {self.provider_id} manually disabled (scope: {self.user_id if self.user_id is not None else 'global'})")

    def manual_enable(self):
        """Clear a manual disable AND any failure cooldown, restoring a fresh
        failure budget so the provider is immediately available again. Persisted
        to the database and mirrored to the cache."""
        import logging
        self._manual_disabled = False
        try:
            db = DatabaseRegistry.get_config_database()
            if db:
                db.clear_provider_manually_disabled(self.user_id, self.provider_id)
        except Exception as e:
            logging.getLogger(__name__).warning(f"Failed to persist manual enable for {self.provider_id}: {e}")
        try:
            from ..cache import get_cache_manager
            get_cache_manager().set(self._manual_disabled_cache_key(), False, ttl=10 * 365 * 24 * 3600)
        except Exception:
            pass
        # Also clear any auto-disable cooldown and reset the failure counter.
        self._clear_disabled_until()
        self.error_tracking['failures'] = 0
        logging.getLogger(__name__).info(f"Provider {self.provider_id} manually re-enabled (scope: {self.user_id if self.user_id is not None else 'global'})")

    def is_manually_disabled(self) -> bool:
        return bool(getattr(self, '_manual_disabled', False))

    def is_rate_limited(self) -> bool:
        # NB: the legacy global manual-disable flag is deliberately NOT consulted
        # here. Disabling a provider is expressed per rotation entry
        # (``enabled: false``) and honoured by the rotation scan; the global flag
        # gates nothing. Treating it as "rate limited" was actively harmful: every
        # provider's handle_request() calls this method and raises
        # Exception("Provider rate limited") when it is true, before any upstream
        # call. The rotation scan (which correctly ignores the manual flag) would
        # still select the provider, that synthetic error was then counted as a
        # transient failure, and three of them armed a real failure cooldown — so a
        # manually-flagged provider was permanently evicted from every rotation it
        # belonged to, reported to clients as an upstream 429.
        # Use is_manually_disabled() when you actually want the manual flag.
        disabled_until = self.error_tracking.get('disabled_until')
        if disabled_until:
            if disabled_until > time.time():
                return True
            # Cooldown has elapsed: this is a temporary measure, so reactivate the
            # provider with a fresh failure budget instead of leaving it on a
            # hair-trigger (failures still at 3) with a stale disabled_until.
            import logging
            logging.getLogger(__name__).info(
                f"Provider {self.provider_id} cooldown elapsed — reactivating with a fresh failure budget"
            )
            self._clear_disabled_until()
            self.error_tracking['failures'] = 0
        # Check usage-based disable (loaded from DB on init, persists across restarts)
        if self._usage_disabled_until and self._usage_disabled_until > time.time():
            return True
        return False
    
    def _get_model_config(self, model: str) -> Optional[Dict]:
        """Get model configuration from provider config"""
        # Use user provider config if available, otherwise global config
        if self.user_id is not None and hasattr(self, 'user_provider_config'):
            provider_config = self.user_provider_config
        else:
            provider_config = config.providers.get(self.provider_id)
            
        # Handle both dict (user) and object (global) config formats
        if provider_config:
            # Get models list depending on config type
            if isinstance(provider_config, dict):
                models = provider_config.get('models', [])
            else:
                models = provider_config.models if hasattr(provider_config, 'models') else []
                
            if models:
                for model_config in models:
                    # Handle both Pydantic objects and dictionaries
                    model_name_value = model_config.name if hasattr(model_config, 'name') else model_config.get('name')
                    if model_name_value == model:
                        # Convert Pydantic object to dict if needed
                        if hasattr(model_config, 'model_dump'):
                            return model_config.model_dump()
                        elif hasattr(model_config, 'dict'):
                            return model_config.dict()
                        else:
                            return model_config
        return None
    
    def _check_token_rate_limit(self, model: str, token_count: int) -> bool:
        """
        Check if a request would exceed token rate limits.
        
        Returns True if any rate limit would be exceeded, False otherwise.
        """
        model_config = self._get_model_config(model)
        if not model_config:
            return False
        
        import logging
        logger = logging.getLogger(__name__)
        
        current_time = time.time()
        
        # Check TPM (tokens per minute)
        if model_config.get('rate_limit_TPM'):
            tpm = model_config['rate_limit_TPM']
            # Get tokens used in the last minute
            tokens_used_tpm = self.token_usage.get(model, {}).get('TPM', [])
            # Filter to only include requests from the last 60 seconds
            one_minute_ago = current_time - 60
            recent_tokens_tpm = [t for t in tokens_used_tpm if t > one_minute_ago]
            total_tpm = sum(recent_tokens_tpm)
            
            if total_tpm + token_count > tpm:
                logger.warning(f"TPM limit would be exceeded: {total_tpm + token_count}/{tpm}")
                return True
        
        # Check TPH (tokens per hour)
        if model_config.get('rate_limit_TPH'):
            tph = model_config['rate_limit_TPH']
            # Get tokens used in the last hour
            tokens_used_tph = self.token_usage.get(model, {}).get('TPH', [])
            # Filter to only include requests from the last 3600 seconds
            one_hour_ago = current_time - 3600
            recent_tokens_tph = [t for t in tokens_used_tph if t > one_hour_ago]
            total_tph = sum(recent_tokens_tph)
            
            if total_tph + token_count > tph:
                logger.warning(f"TPH limit would be exceeded: {total_tph + token_count}/{tph}")
                return True
        
        # Check TPD (tokens per day)
        if model_config.get('rate_limit_TPD'):
            tpd = model_config['rate_limit_TPD']
            # Get tokens used in the last day
            tokens_used_tpd = self.token_usage.get(model, {}).get('TPD', [])
            # Filter to only include requests from the last 86400 seconds
            one_day_ago = current_time - 86400
            recent_tokens_tpd = [t for t in tokens_used_tpd if t > one_day_ago]
            total_tpd = sum(recent_tokens_tpd)
            
            if total_tpd + token_count > tpd:
                logger.warning(f"TPD limit would be exceeded: {total_tpd + token_count}/{tpd}")
                return True
        
        return False
    
    def _record_token_usage(self, model: str, token_count: int):
        """Record token usage for rate limit tracking"""
        import logging
        logger = logging.getLogger(__name__)
        
        if model not in self.token_usage:
            self.token_usage[model] = {"TPM": [], "TPH": [], "TPD": []}
        
        current_time = time.time()
        
        # Record for all three time windows
        self.token_usage[model]["TPM"].append((current_time, token_count))
        self.token_usage[model]["TPH"].append((current_time, token_count))
        self.token_usage[model]["TPD"].append((current_time, token_count))
        
        logger.debug(f"Recorded token usage for model {model}: {token_count} tokens")
    
    def _disable_provider_for_duration(self, duration: str):
        """
        Disable provider for a specific duration.
        
        Args:
            duration: "1m" (1 minute), "1h" (1 hour), or "1d" (1 day)
        """
        import logging
        logger = logging.getLogger(__name__)
        
        duration_map = {
            "1m": 60,
            "1h": 3600,
            "1d": 86400
        }
        
        if duration not in duration_map:
            logger.error(f"Invalid duration: {duration}")
            return
        
        disable_seconds = duration_map[duration]
        self._save_disabled_until(time.time() + disable_seconds)

        logger.error(f"!!! PROVIDER DISABLED !!!")
        logger.error(f"Provider: {self.provider_id}")
        logger.error(f"Reason: Token rate limit exceeded")
        logger.error(f"Disabled for: {duration}")
        logger.error(f"Disabled until: {self.error_tracking['disabled_until']}")
        logger.error(f"Provider will be automatically re-enabled after cooldown")

    async def _throttle(self, slot: str, rate_limit: float, label: str):
        """Space requests sharing `slot` at least `rate_limit` seconds apart.

        The timestamp lives in a process-wide registry rather than on the handler
        because get_provider_handler() builds a fresh handler for every request:
        an instance attribute is always 0 on arrival, so the spacing never applied
        and a configured rate_limit did nothing at all. The lock is what makes the
        limit hold under concurrency — without it, N simultaneous requests all
        read the same stale timestamp, decide no wait is due, and burst together.
        """
        import logging
        logger = logging.getLogger(__name__)

        lock = _rate_limit_locks.setdefault(slot, asyncio.Lock())
        async with lock:
            last_time = _rate_limit_last_request.get(slot, 0)
            required_wait = rate_limit - (time.time() - last_time)
            if required_wait > 0:
                logger.info(
                    f"[RateLimit] {label}: waiting {required_wait:.2f}s "
                    f"(limit: {rate_limit:.3f}s between requests, adaptive: {self.adaptive_limiter.enabled})"
                )
                await asyncio.sleep(required_wait)
            _rate_limit_last_request[slot] = time.time()

    def _effective_rate_limit(self, rate_limit: Optional[float]) -> Optional[float]:
        """Combine the caller's limit with the adaptive one, taking the wider."""
        if self.adaptive_limiter.enabled:
            adaptive_limit = self.adaptive_limiter.get_rate_limit()
            if rate_limit is None:
                return adaptive_limit
            # Use the higher of the two (more conservative)
            return max(rate_limit, adaptive_limit)
        if rate_limit is None:
            return self.rate_limit
        return rate_limit

    async def apply_rate_limit(self, rate_limit: Optional[float] = None):
        """Apply rate limiting by waiting if necessary, using adaptive rate limiting."""
        rate_limit = self._effective_rate_limit(rate_limit)
        if rate_limit and rate_limit > 0:
            slot = f"{self.provider_id}:{self.user_id}"
            await self._throttle(slot, rate_limit, f"Provider {self.provider_id}")
            self.last_request_time = time.time()

    async def apply_model_rate_limit(self, model: str, rate_limit: Optional[float] = None):
        """Apply rate limiting for a specific model, using adaptive rate limiting."""
        rate_limit = self._effective_rate_limit(rate_limit)
        if rate_limit and rate_limit > 0:
            slot = f"{self.provider_id}:{self.user_id}:{model}"
            await self._throttle(slot, rate_limit, f"Model {model}")
            self.model_last_request_time[model] = time.time()

    def record_failure(self):
        import logging
        logger = logging.getLogger(__name__)
        
        self.error_tracking['failures'] += 1
        self.error_tracking['last_failure'] = time.time()
        
        failure_count = self.error_tracking['failures']
        logger.warning(f"=== PROVIDER FAILURE RECORDED ===")
        logger.warning(f"Provider: {self.provider_id}")
        logger.warning(f"Failure count: {failure_count}/3")
        logger.warning(f"Last failure time: {self.error_tracking['last_failure']}")
        
        if self.error_tracking['failures'] >= 3:
            # Get cooldown period from provider config, default to 300 seconds (5 minutes)
            provider_config = config.providers.get(self.provider_id)
            cooldown_seconds = 300  # System default
            
            if provider_config and hasattr(provider_config, 'default_error_cooldown') and provider_config.default_error_cooldown is not None:
                cooldown_seconds = provider_config.default_error_cooldown
                logger.info(f"Using provider-configured cooldown: {cooldown_seconds} seconds")
            else:
                logger.info(f"Using system default cooldown: {cooldown_seconds} seconds")
            
            self._save_disabled_until(time.time() + cooldown_seconds)
            disabled_until_time = self.error_tracking['disabled_until']
            cooldown_remaining = int(disabled_until_time - time.time())
            logger.error(f"!!! PROVIDER DISABLED !!!")
            logger.error(f"Provider: {self.provider_id}")
            logger.error(f"Reason: 3 consecutive failures reached")
            logger.error(f"Disabled until: {disabled_until_time}")
            logger.error(f"Cooldown period: {cooldown_remaining} seconds ({cooldown_seconds / 60:.1f} minutes)")
            logger.error(f"Provider will be automatically re-enabled after cooldown")
        else:
            remaining_failures = 3 - failure_count
            logger.warning(f"Provider still active. {remaining_failures} more failure(s) will disable it")
        logger.warning(f"=== END FAILURE RECORDING ===")

    def record_success(self):
        import logging
        logger = logging.getLogger(__name__)
        
        was_disabled = self.error_tracking.get('disabled_until') is not None
        previous_failures = self.error_tracking.get('failures', 0)
        
        self.error_tracking['failures'] = 0
        self._clear_disabled_until()

        # Record success in adaptive rate limiter
        self.adaptive_limiter.record_success()
        
        logger.info(f"=== PROVIDER SUCCESS RECORDED ===")
        logger.info(f"Provider: {self.provider_id}")
        logger.info(f"Previous failure count: {previous_failures}")
        logger.info(f"Failure count reset to: 0")
        logger.info(f"Adaptive rate limit: {self.adaptive_limiter.current_rate_limit:.2f}s")
        
        if was_disabled:
            logger.info(f"!!! PROVIDER RE-ENABLED !!!")
            logger.info(f"Provider: {self.provider_id}")
            logger.info(f"Reason: Successful request after cooldown period")
            logger.info(f"Provider is now active and available for requests")
        else:
            logger.info(f"Provider remains active")
        logger.info(f"=== END SUCCESS RECORDING ===")

    def supports_usage(self) -> bool:
        """Return True if this provider supports a usage/quota endpoint."""
        return False

    async def get_usage(self) -> Optional[Dict]:
        """Fetch current usage/quota data from the provider. Returns None if unsupported."""
        return None

    def normalize_usage_data(self, usage_data: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
        """Normalize provider usage/quota data into a shared schema when possible."""
        if not usage_data or not isinstance(usage_data, dict):
            return usage_data

        normalized = dict(usage_data)
        free_tier = normalized.get('free_tier')
        if isinstance(free_tier, dict):
            normalized['free_tier'] = {
                'limit': free_tier.get('limit'),
                'used': free_tier.get('used'),
                'remaining': free_tier.get('remaining'),
                'period': free_tier.get('period'),
                'limit_type': free_tier.get('limit_type', 'requests'),
                'source': free_tier.get('source', 'provider')
            }
        return normalized

    async def handle_request_with_batching(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 a request with optional batching.
        """
        # Check if batching is enabled and not streaming
        if self.batcher.enabled and not stream:
            # Prepare request data
            request_data = {
                "model": model,
                "messages": messages,
                "max_tokens": max_tokens,
                "temperature": temperature,
                "stream": stream,
                "tools": tools,
                "tool_choice": tool_choice,
                "api_key": self.api_key
            }
            
            # Submit request for batching
            batched_result = await self.batcher.submit_request(
                provider_id=self.provider_id,
                model=model,
                request_data=request_data
            )
            
            # If batching returned None, it means batching is disabled or we should process directly
            if batched_result is not None:
                return batched_result
         
        # Fall back to direct processing (either batching disabled, streaming, or batching returned None)
        return await self._handle_request_direct(model, messages, max_tokens, temperature, stream, tools, tool_choice)

    def _get_models_cache_key(self) -> str:
        """Get unified cache key for provider models list"""
        if hasattr(self, 'user_id') and self.user_id is not None:
            return f"provider_models:{self.provider_id}:user:{self.user_id}"
        return f"provider_models:{self.provider_id}"

    def _save_models_cache(self, models: List[Any]) -> None:
        """Save models list to unified cache system"""
        import logging
        try:
            from ..cache import get_cache_manager
            cache = get_cache_manager()
            
            cache_data = {
                'models': [m.dict() if hasattr(m, 'dict') else dict(m) for m in models],
                'cached_at': time.time()
            }
            
            # Cache for 24 hours (86400 seconds)
            cache.set(self._get_models_cache_key(), cache_data, ttl=86400)
            logging.info(f"{self.__class__.__name__}: ✓ Saved {len(models)} models to cache")
        except Exception as e:
            logging.warning(f"{self.__class__.__name__}: Failed to save models cache: {e}")

    def _load_models_cache(self) -> Optional[List[Any]]:
        """Load models list from unified cache system if available and not expired"""
        import logging
        try:
            from ..cache import get_cache_manager
            from ..models import Model
            
            cache = get_cache_manager()
            cache_data = cache.get(self._get_models_cache_key())
            
            if not cache_data:
                return None
                
            models = []
            for model_dict in cache_data.get('models', []):
                models.append(Model(**model_dict))
            
            logging.info(f"{self.__class__.__name__}: ✓ Loaded {len(models)} models from cache")
            return models
        except Exception as e:
            logging.warning(f"{self.__class__.__name__}: Failed to load models cache: {e}")
            return None

    async def _handle_request_direct(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]:
        """
        Direct request handling without batching (original handle_request logic).
        This method should be overridden by subclasses with their specific implementation.
        """
        raise NotImplementedError("_handle_request_direct must be implemented by subclasses")
