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

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

Request handlers for AISBF.

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!

Request handlers for AISBF.
"""
import asyncio
import base64
import json
import logging
import re
import uuid
import hashlib
import threading
import time
import time as time_module
from concurrent.futures import ThreadPoolExecutor
from typing import Dict, List, Optional, Union
from pathlib import Path
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse, StreamingResponse, Response
from .models import ChatCompletionRequest, ChatCompletionResponse
from .providers import get_provider_handler, RateLimitError
from .providers.base import is_provider_disabled_cheap
from .config import config
from .studio import infer_model_capabilities
from .studio_adapters import effective_studio_adapter, infer_studio_adapter_profile, adapt_studio_payload_with_profile
from .utils import (
    count_messages_tokens,
    split_messages_into_chunks,
    get_max_request_tokens_for_model,
    get_max_completion_tokens_for_model
)
from .context import ContextManager, get_context_config_for_model
from .classifier import content_classifier
from .classifier import SemanticClassifier
from .cache import get_response_cache
from .analytics import get_analytics
from .streaming_optimization import (
    get_streaming_optimizer,
    StreamingConfig,
    calculate_google_delta,
    KiroSSEParser,
    OptimizedTextAccumulator,
    optimize_sse_chunk
)
from .database import DatabaseRegistry
from .prompt_analysis import analyze_prompt_payload

_autoselect_result_cache: dict = {}
_autoselect_result_cache_ttl: int = 3600  # seconds


def _is_coderai_warmup_error(error) -> bool:
    """True when the error is just a missing CoderAI broker session (worker warming up).

    CoderAI workers often run on small/edge hardware that drops the broker session
    while cooling down. That is transient, not a provider fault: callers must not
    record a failure for it (which would disable the provider for a long cooldown).
    The marker string is CoderAI-specific, so this is safe to check on any path.
    """
    return 'No active CoderAI broker session' in str(error or '')


def _is_upstream_rate_limit(error) -> bool:
    """True when the upstream refused on quota rather than the provider breaking.

    A 429 says one quota bucket is exhausted right now; it says nothing about
    whether the next request will succeed, and providers routinely keep serving
    from another bucket while one is capped. Counting these toward the
    three-strikes cooldown makes aisbf reject traffic itself that the upstream
    would have answered, so they are excluded from failure recording.
    """
    if isinstance(error, RateLimitError):
        return True
    status = getattr(getattr(error, 'response', None), 'status_code', None)
    if status == 429:
        return True
    return '429 Too Many Requests' in str(error or '') or 'usage_limit_reached' in str(error or '')


def _should_record_failure(error) -> bool:
    """Whether an exception should count against the provider's failure budget."""
    return not _is_coderai_warmup_error(error) and not _is_upstream_rate_limit(error)


def _raise_if_provider_unavailable(handler, provider_id: str) -> None:
    """Reject the request with 503 when the provider is disabled, saying why.

    is_rate_limited() collapses three very different states — the dashboard
    disable toggle, the failure cooldown and the usage-limit cooldown — into one
    boolean. Returning a bare "Provider temporarily unavailable" with no log line
    made a provider that had simply been toggled off in the dashboard look
    identical to an upstream outage, with nothing in debug.log to tell them
    apart. Log the reason and put it in the response detail.
    """
    if not handler.is_rate_limited():
        return

    if handler.is_manually_disabled():
        reason = "manually disabled from the dashboard (re-enable it there)"
    else:
        now = time.time()
        cooldown_until = handler.error_tracking.get('disabled_until') if getattr(handler, 'error_tracking', None) else None
        usage_until = getattr(handler, '_usage_disabled_until', None)
        if cooldown_until and cooldown_until > now:
            reason = f"in failure cooldown for another {int(cooldown_until - now)}s"
        elif usage_until and usage_until > now:
            reason = f"disabled by a usage limit for another {int(usage_until - now)}s"
        else:
            reason = "temporarily unavailable"

    logging.getLogger(__name__).warning(
        f"[{provider_id}] Rejecting request with 503: provider is {reason}"
    )
    raise HTTPException(status_code=503, detail=f"Provider '{provider_id}' is {reason}")


# Registry for proxied content — shared across all RequestHandler instances.
# Values: {"type": "broker", "provider_id": str, "path": str}
#      or {"type": "http",   "url": str}
_proxy_file_registry: dict = {}
_proxy_file_registry_lock = threading.Lock()
# In-flight Future registry for single-flight deduplication of identical concurrent selections
_autoselect_inflight: dict = {}  # cache_key → asyncio.Future

# Incremental conversation summaries — keyed by a stable session identifier derived from
# the first messages of the conversation.  Each entry records the fingerprints of the
# messages that were summarized so that the next turn can detect a prefix match and send
# only the delta (new messages) instead of re-compacting the full history.
_conversation_summaries: dict = {}
_conversation_summary_ttl: int = 14400  # 4 hours

# Internal HuggingFace model singleton — shared across all AutoselectHandler instances
# so the model is loaded once and kept in RAM regardless of how many user handlers exist.
_internal_model_singleton = None
_internal_tokenizer_singleton = None
# Serializes concurrent initialization (double-checked locking pattern)
_internal_model_init_lock = threading.Lock()
# Single-worker executor: queues inference calls so the model is never called concurrently.
# max_workers=1 makes the queue implicit — no separate inference lock needed.
_internal_model_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="aisbf-internal-model")


def generate_system_fingerprint(provider_id: str, seed: Optional[int] = None) -> str:
    """
    Generate a unique system_fingerprint for OpenAI-compatible responses.
    
    The fingerprint is:
    - Unique per provider (based on provider_id)
    - Different for every request if seed is present in the request
    - Consistent for the same provider_id + seed combination
    
    Args:
        provider_id: The provider identifier from configuration
        seed: Optional seed from the request (if present, generates unique fingerprint per request)
    
    Returns:
        A fingerprint string in format "fp_<hash>"
    """
    if seed is not None:
        # If seed is provided, generate a unique fingerprint for this specific request
        # Combine provider_id, seed, and a timestamp component for uniqueness
        unique_data = f"{provider_id}:{seed}:{uuid.uuid4()}"
        hash_value = hashlib.md5(unique_data.encode()).hexdigest()[:24]
    else:
        # Without seed, generate a consistent fingerprint per provider
        # This is still unique per provider but consistent across requests
        unique_data = f"{provider_id}:aisbf:fingerprint"
        hash_value = hashlib.md5(unique_data.encode()).hexdigest()[:24]
    
    return f"fp_{hash_value}"

class RequestHandler:
    def __init__(self, user_id=None):
        self.user_id = user_id
        self.config = config
        # Load user-specific configs if user_id is provided
        if user_id:
            self._load_user_configs()
        else:
            self.user_providers = {}
            self.user_rotations = {}
            self.user_autoselects = {}

    def _get_provider_config(self, provider_id: str):
        if self.user_id and provider_id in getattr(self, 'user_providers', {}):
            return self.user_providers[provider_id]
        return self.config.get_provider(provider_id)

    def _get_provider_type(self, provider_id: str) -> str:
        provider_config = self._get_provider_config(provider_id)
        if isinstance(provider_config, dict):
            return provider_config.get('type', 'openai') or 'openai'
        return getattr(provider_config, 'type', 'openai') or 'openai'

    def _record_dashboard_proxy_event(self, request, provider_id: str, model_name: str, success: bool, latency_ms: float, metadata: Dict | None = None):
        try:
            from aisbf.database import DatabaseRegistry
            db = DatabaseRegistry.get_config_database()
            status_code = (metadata or {}).get('status_code')
            if status_code is None:
                status_code = 200 if success else 500
            # Rotation/MCP code paths invoke this without an HTTP request object;
            # fall back to handler-level context when request is unavailable.
            path = request.url.path if request is not None else '/rotation'
            user_id = getattr(request.state, 'user_id', None) if request is not None else getattr(self, 'user_id', None)
            username = request.session.get('username') if (request is not None and hasattr(request, 'session')) else None
            method = request.method if request is not None else 'POST'
            db.record_dashboard_event(
                event_type='request_proxied',
                path=path,
                user_id=user_id,
                username=username,
                method=method,
                status_code=status_code,
                provider_id=provider_id,
                rotation_id=(metadata or {}).get('rotation_id'),
                autoselect_id=(metadata or {}).get('autoselect_id'),
                metadata={
                    'model_name': model_name,
                    'latency_ms': round(float(latency_ms or 0), 2),
                    'stream': bool((metadata or {}).get('stream')),
                    **(metadata or {}),
                },
            )
        except Exception:
            logger.debug("Failed to record proxied request dashboard event", exc_info=True)

    def _run_prompt_analysis(self, request, provider_id: str, model_name: str, request_data: Dict, prompt_tokens: Optional[int] = None, effective_context: Optional[int] = None):
        aisbf_config = self.config.get_aisbf_config()
        if not aisbf_config:
            return None

        provider_config = self._get_provider_config(provider_id)
        model_config = None
        models = provider_config.get('models', []) if isinstance(provider_config, dict) else getattr(provider_config, 'models', None) or []
        for candidate in models:
            candidate_name = candidate.get('name') if isinstance(candidate, dict) else getattr(candidate, 'name', None)
            if candidate_name == model_name:
                model_config = candidate
                break

        scan_enabled = self.config.resolve_feature_enabled(
            'prompt_security',
            model_config=model_config,
            provider_config=provider_config,
        )
        analytics_enabled = self.config.resolve_feature_enabled(
            'context_lens',
            model_config=model_config,
            provider_config=provider_config,
        )
        block_high_risk = self.config.resolve_feature_enabled(
            'block_high_risk_prompts',
            model_config=model_config,
            provider_config=provider_config,
        )

        if not scan_enabled and not analytics_enabled:
            return None

        analysis = analyze_prompt_payload(request_data, model_name=model_name)
        blocked = bool(block_high_risk and analysis.get('risk_level') == 'high')

        db = DatabaseRegistry.get_config_database()
        run_id = db.record_prompt_analysis_run(
            user_id=getattr(request.state, 'user_id', None) if request is not None else self.user_id,
            token_id=getattr(request.state, 'token_id', None) if request is not None else None,
            provider_id=provider_id,
            model_name=model_name,
            rotation_id=request_data.get('_rotation_id'),
            autoselect_id=request_data.get('_autoselect_id'),
            request_path=request.url.path if request is not None else None,
            request_hash=analysis.get('request_hash'),
            prompt_tokens=prompt_tokens,
            effective_context=effective_context,
            scan_enabled=scan_enabled,
            analytics_enabled=analytics_enabled,
            blocked=blocked,
            risk_level=analysis.get('risk_level'),
            risk_score=analysis.get('risk_score'),
            findings_count=analysis.get('summary', {}).get('findings_count', 0),
            summary_json={
                'security_summary': analysis.get('summary', {}),
                'composition': analysis.get('composition', {}),
            },
        )
        if scan_enabled:
            db.record_prompt_analysis_findings(run_id, analysis.get('findings', []))
        analysis['blocked'] = blocked
        return analysis

    def _adapt_studio_workflow_payload(self, provider_id: str, endpoint_path: str, body: Dict) -> Dict:
        payload = dict(body or {})
        provider_type = (self._get_provider_type(provider_id) or '').lower()
        model_metadata = payload.get('_studio_model_metadata') if isinstance(payload.get('_studio_model_metadata'), dict) else {}
        if provider_id and '_studio_provider_id' not in payload:
            payload['_studio_provider_id'] = provider_id
        if model_metadata.get('provider_endpoint') and '_studio_provider_endpoint' not in payload:
            payload['_studio_provider_endpoint'] = model_metadata.get('provider_endpoint')
        adapter = effective_studio_adapter(provider_type, model_metadata)
        profile = infer_studio_adapter_profile(provider_id, provider_type, model_metadata)

        def first_value(*keys):
            for key in keys:
                value = payload.get(key)
                if isinstance(value, str) and value.strip():
                    return value.strip()
            return None

        if endpoint_path == 'v1/video/dub':
            selected = first_value('video_model', 'stt_model', 'tts_model', 'model')
            if selected:
                payload['model'] = selected
        elif endpoint_path == 'v1/audio/clone':
            selected = first_value('model', 'tts_model')
            if selected:
                payload['model'] = selected
        elif endpoint_path == 'v1/audio/convert':
            selected = first_value('model', 'audio_model', 'tts_model', 'stt_model')
            if selected:
                payload['model'] = selected
        elif endpoint_path in {'v1/audio/split', 'v1/audio/denoise'}:
            selected = first_value('model', 'audio_model')
            if selected:
                payload['model'] = selected
        elif endpoint_path in {'v1/images/faceswap', 'v1/images/outfit'}:
            selected = first_value('model', 'image_model', 'video_model')
            if selected:
                payload['model'] = selected
        elif endpoint_path in {'v1/images/to3d', 'v1/images/from3d', 'v1/video/to3d', 'v1/video/from3d', 'v1/3d/generate'}:
            selected = first_value('model', 'render_model', 'image_model', 'video_model')
            if selected:
                payload['model'] = selected

        payload = adapt_studio_payload_with_profile(adapter, profile, endpoint_path, payload)
        payload.pop('_studio_model_metadata', None)
        payload.pop('_studio_provider_id', None)
        payload.pop('_studio_provider_endpoint', None)
        return payload

    def _get_market_source_details(self, provider_id: str):
        provider_config = None
        if self.user_id and provider_id in getattr(self, 'user_providers', {}):
            provider_config = self.user_providers.get(provider_id)
        elif provider_id in getattr(self.config, 'providers', {}):
            cfg = self.config.get_provider(provider_id)
            provider_config = cfg.model_dump() if hasattr(cfg, 'model_dump') else cfg
        if isinstance(provider_config, dict):
            return provider_config.get('market_source')
        return None

    @staticmethod
    def _market_request_id(request_data: Optional[Dict], fallback_provider_id: str) -> str:
        payload = {
            'provider_id': fallback_provider_id,
            'model': (request_data or {}).get('model'),
            'messages': (request_data or {}).get('messages'),
            'input': (request_data or {}).get('input'),
            'prompt': (request_data or {}).get('prompt'),
            'voice': (request_data or {}).get('voice'),
            'endpoint_path': (request_data or {}).get('_market_endpoint_path'),
        }
        serialized = json.dumps(payload, sort_keys=True, default=str)
        return hashlib.sha256(serialized.encode('utf-8')).hexdigest()

    def _settle_market_result(self, provider_id: str, usage: Optional[Dict], requests_count: int = 1, metadata: Optional[Dict] = None, request_data: Optional[Dict] = None):
        market_source = self._get_market_source_details(provider_id)
        if not self.user_id or not market_source:
            return None
        listing_id = market_source.get('listing_id')
        if not listing_id:
            return None
        db = DatabaseRegistry.get_config_database()
        settlement_metadata = {
            'provider_id': provider_id,
            'market_imported': True,
            'market_request_id': self._market_request_id(request_data, provider_id),
        }
        if metadata:
            settlement_metadata.update(metadata)
        return db.settle_market_usage(
            consumer_user_id=self.user_id,
            listing_id=int(listing_id),
            prompt_tokens=(usage or {}).get('prompt_tokens', 0),
            completion_tokens=(usage or {}).get('completion_tokens', 0),
            requests_count=requests_count,
            metadata=settlement_metadata,
        )

    def _extract_usage_from_sse_chunk(self, chunk_payload) -> Optional[Dict[str, int]]:
        try:
            if isinstance(chunk_payload, bytes):
                chunk_payload = chunk_payload.decode('utf-8', errors='ignore')
            if isinstance(chunk_payload, str):
                for line in chunk_payload.splitlines():
                    line = line.strip()
                    if not line.startswith('data: '):
                        continue
                    raw = line[6:].strip()
                    if not raw or raw == '[DONE]':
                        continue
                    try:
                        parsed = json.loads(raw)
                    except Exception:
                        continue
                    if isinstance(parsed, dict) and isinstance(parsed.get('usage'), dict):
                        usage = parsed.get('usage') or {}
                        if usage.get('total_tokens') is not None or usage.get('prompt_tokens') is not None or usage.get('completion_tokens') is not None:
                            return usage
            elif isinstance(chunk_payload, dict):
                usage = chunk_payload.get('usage')
                if isinstance(usage, dict):
                    return usage
        except Exception:
            return None
        return None

    def _load_user_configs(self):
        """Load user-specific configurations from database"""
        self.reload_user_config()

    def reload_user_config(self):
        """Reload user configuration from database"""
        import logging
        logger = logging.getLogger(__name__)
        from .database import DatabaseRegistry
        db = DatabaseRegistry.get_config_database()
        
        # Convert list to dictionary with id as key
        providers = db.get_user_providers(self.user_id)
        self.user_providers = {p['provider_id']: p['config'] for p in providers}
        
        rotations = db.get_user_rotations(self.user_id)
        self.user_rotations = {r['rotation_id']: r['config'] for r in rotations}
        
        autoselects = db.get_user_autoselects(self.user_id)
        self.user_autoselects = {a['autoselect_id']: a['config'] for a in autoselects}
        
        logger.info(f"Reloaded user configuration for user_id={self.user_id}")
        logger.info(f"  Loaded {len(self.user_providers)} user providers")
        logger.info(f"  Loaded {len(self.user_rotations)} user rotations")
        logger.info(f"  Loaded {len(self.user_autoselects)} user autoselects")

    def _should_cache_response(self, provider_config=None, model_config=None, rotation_config=None, autoselect_config=None):
        """
        Determine if response caching should be enabled based on configuration hierarchy.
        
        Priority order (highest to lowest):
        1. Model-level enable_response_cache setting
        2. Provider-level enable_response_cache setting
        3. Rotation-level enable_response_cache setting
        4. Autoselect-level enable_response_cache setting
        5. Global response_cache.enabled setting
        
        Args:
            provider_config: Provider configuration object
            model_config: Model configuration object or dict
            rotation_config: Rotation configuration object
            autoselect_config: Autoselect configuration object
            
        Returns:
            bool: True if caching should be enabled, False otherwise
        """
        import logging
        logger = logging.getLogger(__name__)
        
        resolved = self.config.resolve_feature_enabled(
            'response_cache',
            model_config=model_config,
            provider_config=provider_config,
            rotation_config=rotation_config,
            autoselect_config=autoselect_config,
        )
        logger.debug(f"Resolved response cache setting: {resolved}")
        return resolved

    async def _handle_chunked_request(
        self,
        handler,
        model: str,
        messages: List[Dict],
        max_tokens: Optional[int],
        temperature: float,
        stream: bool,
        tools: Optional[List[Dict]],
        tool_choice: Optional[Union[str, Dict]],
        max_request_tokens: int,
        provider_id: str,
        logger
    ) -> Dict:
        """
        Handle a request that needs to be split into multiple chunks due to token limits.
        
        This method splits the request into chunks, sends each chunk sequentially,
        and combines the responses into a single response.
        
        Args:
            handler: The provider handler
            model: The model name
            messages: The messages to send
            max_tokens: Max output tokens
            temperature: Temperature setting
            stream: Whether to stream (not supported for chunked requests)
            tools: Tool definitions
            tool_choice: Tool choice setting
            max_request_tokens: Maximum tokens per request
            provider_id: Provider identifier
            logger: Logger instance
        
        Returns:
            Combined response from all chunks
        """
        import time
        
        logger.info(f"=== CHUNKED REQUEST HANDLING START ===")
        logger.info(f"Max request tokens per chunk: {max_request_tokens}")
        
        # Split messages into chunks
        message_chunks = split_messages_into_chunks(messages, max_request_tokens, model)
        logger.info(f"Split into {len(message_chunks)} message chunks")
        
        if stream:
            logger.warning("Streaming is not supported for chunked requests, falling back to non-streaming")
        
        # Process each chunk and collect responses
        all_responses = []
        combined_content = ""
        combined_tool_calls = []
        total_prompt_tokens = 0
        total_completion_tokens = 0
        created_time = int(time_module.time())
        response_id = f"chunked-{provider_id}-{model}-{created_time}"
        
        for chunk_idx, chunk_messages in enumerate(message_chunks):
            logger.info(f"Processing chunk {chunk_idx + 1}/{len(message_chunks)}")
            logger.info(f"Chunk messages count: {len(chunk_messages)}")
            
            # Apply rate limiting between chunks
            if chunk_idx > 0:
                await handler.apply_rate_limit()
            
            try:
                chunk_response = await handler.handle_request(
                    model=model,
                    messages=chunk_messages,
                    max_tokens=max_tokens,
                    temperature=temperature,
                    stream=False,  # Always non-streaming for chunked requests
                    tools=tools if chunk_idx == 0 else None,  # Only first chunk uses tools
                    tool_choice=tool_choice if chunk_idx == 0 else None
                )
                
                # Extract content from response
                if isinstance(chunk_response, dict):
                    choices = chunk_response.get('choices', [])
                    if choices:
                        chunk_message = choices[0].get('message') or {}
                        # Tool-call responses carry content=None (the key exists, so a
                        # .get default won't kick in) – coalesce to '' or the += raises
                        # TypeError, which records a provider failure and can disable
                        # the provider even though upstream returned 200.
                        combined_content += chunk_message.get('content') or ''
                        chunk_tool_calls = chunk_message.get('tool_calls')
                        if chunk_tool_calls:
                            combined_tool_calls.extend(chunk_tool_calls)

                        # Accumulate token usage
                        usage = chunk_response.get('usage', {})
                        total_prompt_tokens += usage.get('prompt_tokens', 0)
                        total_completion_tokens += usage.get('completion_tokens', 0)

                all_responses.append(chunk_response)
                logger.info(f"Chunk {chunk_idx + 1} processed successfully")
                
            except Exception as e:
                logger.error(f"Error processing chunk {chunk_idx + 1}: {e}")
                # If a chunk fails, we still try to return what we have
                if all_responses:
                    logger.warning("Returning partial results from successful chunks")
                    break
                else:
                    raise e
        
        # Build combined response.  Tool calls from any chunk must be passed through
        # (chunk 0 is the one sent with tools) or agent clients would receive an
        # empty completion and stall.
        combined_message = {"role": "assistant", "content": combined_content}
        finish_reason = "stop"
        if combined_tool_calls:
            combined_message["tool_calls"] = combined_tool_calls
            if not combined_content:
                combined_message["content"] = None
            finish_reason = "tool_calls"
        combined_response = {
            "id": response_id,
            "object": "chat.completion",
            "created": created_time,
            "model": model,
            "choices": [{
                "index": 0,
                "message": combined_message,
                "finish_reason": finish_reason
            }],
            "usage": {
                "prompt_tokens": total_prompt_tokens,
                "completion_tokens": total_completion_tokens,
                "total_tokens": total_prompt_tokens + total_completion_tokens
            },
            "aisbf_chunked": True,
            "aisbf_total_chunks": len(message_chunks)
        }

        logger.info(f"=== CHUNKED REQUEST HANDLING END ===")
        logger.info(f"Combined content length: {len(combined_content)} characters")
        logger.info(f"Total chunks processed: {len(all_responses)}")
        
        return combined_response

    async def handle_chat_completion(self, request: Request, provider_id: str, request_data: Dict) -> Dict:
        import logging
        import time
        logger = logging.getLogger(__name__)
        logger.info(f"=== RequestHandler.handle_chat_completion START ===")
        logger.info(f"Provider ID: {provider_id}")
        logger.info(f"User ID: {self.user_id}")
        logger.info(f"Request data: {request_data}")
        
        # Track request start time for analytics
        request_start_time = time.time()
        model_name = request_data.get('model', 'unknown')
        autoselect_id = request_data.get('_autoselect_id')

        # Check for user-specific provider config first
        if self.user_id and provider_id in self.user_providers:
            provider_config = self.user_providers[provider_id]
            logger.info(f"Using user-specific provider config for {provider_id}")
        else:
            provider_config = self.config.get_provider(provider_id)
            logger.info(f"Using global provider config for {provider_id}")

        # Inject backend endpoint and normalised model name so the cache key is
        # provider-agnostic: two providers sharing the same endpoint+model share
        # the same cache entries regardless of their provider_id.
        if isinstance(provider_config, dict):
            _backend_endpoint = provider_config.get('endpoint', '')
        else:
            _backend_endpoint = getattr(provider_config, 'endpoint', '')
        _raw_model = request_data.get('model', '')
        # Models are exposed as "{provider_id}/{model_name}"; strip the prefix.
        if _raw_model.startswith(f"{provider_id}/"):
            _backend_model = _raw_model[len(provider_id) + 1:]
        else:
            _backend_model = _raw_model
        request_data['_backend_endpoint'] = _backend_endpoint
        request_data['_backend_model'] = _backend_model

        # Check response cache for non-streaming requests
        stream = request_data.get('stream', False)
        if not stream:
            try:
                aisbf_config = self.config.get_aisbf_config()
                if aisbf_config and aisbf_config.response_cache and aisbf_config.response_cache.enabled:
                    response_cache = get_response_cache(aisbf_config.response_cache.model_dump())
                    cached_response = response_cache.get(request_data, user_id=self.user_id)
                    if cached_response:
                        logger.info(f"Cache hit for request to provider {provider_id}")
                        return cached_response
                    else:
                        logger.debug(f"Cache miss for request to provider {provider_id}")
            except Exception as cache_error:
                logger.warning(f"Response cache check failed: {cache_error}")

        logger.info(f"Provider config: {provider_config}")
        # Handle both dict (user providers) and object (global providers) formats
        if isinstance(provider_config, dict):
            provider_type = provider_config.get('type')
            provider_endpoint = provider_config.get('endpoint')
            api_key_required = provider_config.get('api_key_required', False)
        else:
            provider_type = provider_config.type
            provider_endpoint = provider_config.endpoint
            api_key_required = provider_config.api_key_required
            
        logger.info(f"Provider type: {provider_type}")
        logger.info(f"Provider endpoint: {provider_endpoint}")
        logger.info(f"API key required: {api_key_required}")

        if api_key_required:
            api_key = request_data.get('api_key') or request.headers.get('Authorization', '').replace('Bearer ', '')
            logger.info(f"API key from request: {'***' if api_key else 'None'}")
            if not api_key:
                # Record analytics for authentication failure
                try:
                    analytics = get_analytics()
                    # Calculate latency for auth failure
                    latency_ms = (time.time() - request_start_time) * 1000
                    
                    # Estimate tokens for the request
                    try:
                        messages = request_data.get('messages', [])
                        model_name = request_data.get('model', 'unknown')
                        estimated_tokens = count_messages_tokens(messages, model_name)
                    except Exception:
                        estimated_tokens = 50
                    
                    analytics.record_request(
                        provider_id=provider_id,
                        model_name=request_data.get('model', 'unknown'),
                        tokens_used=estimated_tokens,
                        latency_ms=latency_ms,
                        success=False,
                        error_type='AuthenticationError',
                        user_id=getattr(request.state, 'user_id', None),
                        token_id=getattr(request.state, 'token_id', None),
                        prompt_tokens=estimated_tokens,
                        completion_tokens=0,
                        actual_cost=None
                    )
                except Exception as analytics_error:
                    logger.warning(f"Analytics recording for auth failure failed: {analytics_error}")
                
                raise HTTPException(status_code=401, detail="API key required")
        else:
            api_key = None
            logger.info("No API key required for this provider")

        logger.info(f"Getting provider handler for {provider_id}")
        handler = get_provider_handler(provider_id, api_key, user_id=self.user_id)
        logger.info(f"Provider handler obtained: {handler.__class__.__name__}")

        _raise_if_provider_unavailable(handler, provider_id)

        try:
            model = request_data.get('model')
            messages = request_data.get('messages', [])
            
            logger.info(f"Model requested: {model}")
            logger.info(f"Messages count: {len(messages)}")
            logger.info(f"Max tokens: {request_data.get('max_tokens')}")
            logger.info(f"Temperature: {request_data.get('temperature', 1.0)}")
            logger.info(f"Stream: {request_data.get('stream', False)}")
            
            # Get context configuration
            context_config = get_context_config_for_model(
                model_name=model,
                provider_config=provider_config,
                rotation_model_config=None,
                rotation_config=None,
                autoselect_config=None,
            )
            logger.info(f"Context config: {context_config}")
            
            # Check for max_request_tokens in provider config
            max_request_tokens = get_max_request_tokens_for_model(
                model_name=model,
                provider_config=provider_config,
                rotation_model_config=None
            )

            # Apply provider-level default max output tokens when the client didn't set one
            if request_data.get('max_tokens') is None:
                default_max_tokens = get_max_completion_tokens_for_model(
                    model_name=model,
                    provider_config=provider_config,
                    rotation_model_config=None,
                )
                # Lowest-priority fallback: autoselect default threaded in when this
                # request is dispatched directly from an autoselect (provider/model).
                if not default_max_tokens:
                    default_max_tokens = request_data.get('_autoselect_default_max_tokens')
                if default_max_tokens:
                    request_data['max_tokens'] = default_max_tokens
                    logger.info(f"Applied provider default max_tokens: {default_max_tokens}")

            # Calculate effective context (total tokens used)
            effective_context = count_messages_tokens(messages, model)
            logger.info(f"Effective context: {effective_context} tokens")

            prompt_analysis = self._run_prompt_analysis(
                request,
                provider_id,
                model,
                request_data,
                prompt_tokens=effective_context,
                effective_context=effective_context,
            )
            if prompt_analysis and prompt_analysis.get('blocked'):
                raise HTTPException(status_code=400, detail="Prompt blocked by local security policy")
            
            # Apply context condensation if needed
            if context_config.get('condense_context', 0) > 0:
                context_manager = ContextManager(context_config, handler, self.config.get_condensation(), self.user_id)
                if context_manager.should_condense(messages, model):
                    logger.info("Context condensation triggered")
                    messages = await context_manager.condense_context(messages, model)
                    effective_context = count_messages_tokens(messages, model)
                    logger.info(f"Condensed effective context: {effective_context} tokens")
            
            if max_request_tokens:
                # Count tokens in the request
                request_tokens = count_messages_tokens(messages, model)
                logger.info(f"Request tokens: {request_tokens}, max_request_tokens: {max_request_tokens}")
                
                if request_tokens > max_request_tokens:
                    logger.info(f"Request exceeds max_request_tokens, will split into chunks")
                    
                    # Apply rate limiting
                    logger.info("Applying rate limiting...")
                    await handler.apply_rate_limit()
                    logger.info("Rate limiting applied")
                    
                    # Handle as chunked request
                    response = await self._handle_chunked_request(
                        handler=handler,
                        model=model,
                        messages=messages,
                        max_tokens=request_data.get('max_tokens'),
                        temperature=request_data.get('temperature', 1.0),
                        stream=request_data.get('stream', False),
                        tools=request_data.get('tools'),
                        tool_choice=request_data.get('tool_choice'),
                        max_request_tokens=max_request_tokens,
                        provider_id=provider_id,
                        logger=logger
                    )
                    
                    handler.record_success()
                    
                    # Cache the response for non-streaming chunked requests
                    if not stream:
                        try:
                            aisbf_config = self.config.get_aisbf_config()
                            if aisbf_config and aisbf_config.response_cache and aisbf_config.response_cache.enabled:
                                response_cache = get_response_cache(aisbf_config.response_cache.model_dump())
                                response_cache.set(request_data, response, user_id=self.user_id)
                                logger.debug(f"Cached chunked response for request to provider {provider_id}")
                        except Exception as cache_error:
                            logger.warning(f"Response cache set failed for chunked request: {cache_error}")
                    
                    logger.info(f"=== RequestHandler.handle_chat_completion END ===")
                    return response
            
            # Apply rate limiting
            logger.info("Applying rate limiting...")
            await handler.apply_rate_limit()
            logger.info("Rate limiting applied")

            logger.info(f"Sending request to provider handler...")
            response = await handler.handle_request(
                model=model,
                messages=messages,
                max_tokens=request_data.get('max_tokens'),
                temperature=request_data.get('temperature', 1.0),
                stream=request_data.get('stream', False),
                tools=request_data.get('tools'),
                tool_choice=request_data.get('tool_choice')
            )
            logger.info(f"Response received from provider")
            logger.info(f"Response type: {type(response)}")
            logger.info(f"Response: {response}")
            
            # Add effective context to response for non-streaming
            if isinstance(response, dict) and 'usage' in response:
                response['usage']['effective_context'] = effective_context
                logger.info(f"Added effective_context to response: {effective_context}")
            
            # For OpenAI-compatible providers, the response is already a response object
            # Just return it as-is without any parsing or modification
            
            # Cache the response for non-streaming requests
            if not stream:
                try:
                    aisbf_config = self.config.get_aisbf_config()
                    if aisbf_config and aisbf_config.response_cache and aisbf_config.response_cache.enabled:
                        response_cache = get_response_cache(aisbf_config.response_cache.model_dump())
                        response_cache.set(request_data, response, user_id=self.user_id)
                        logger.debug(f"Cached response for request to provider {provider_id}")
                except Exception as cache_error:
                    logger.warning(f"Response cache set failed: {cache_error}")
            
            handler.record_success()

            if isinstance(response, dict):
                try:
                    self._settle_market_result(provider_id, response.get('usage', {}), 1, {'kind': 'chat_completion'}, request_data=request_data)
                except Exception as market_error:
                    logger.warning(f"Market settlement failed for {provider_id}: {market_error}")
            
            # Record analytics for token usage
            try:
                analytics = get_analytics()
                latency_ms = (time.time() - request_start_time) * 1000
                logger.info(f"Analytics: latency_ms={latency_ms:.2f}, request_start_time={request_start_time}, current_time={time.time()}")
                
                if response:
                    # Handle both dict responses and OpenAI objects
                    usage = None
                    if isinstance(response, dict):
                        # Dict response (traditional API format)
                        usage = response.get('usage', {})
                        total_tokens = usage.get('total_tokens', 0)
                        prompt_tokens = usage.get('prompt_tokens', 0)
                        completion_tokens = usage.get('completion_tokens', 0)
                    elif hasattr(response, 'usage') and response.usage:
                        # OpenAI/Pydantic object with usage attribute
                        usage = response.usage
                        total_tokens = getattr(usage, 'total_tokens', 0)
                        prompt_tokens = getattr(usage, 'prompt_tokens', 0)
                        completion_tokens = getattr(usage, 'completion_tokens', 0)
                        logger.debug(f"Extracted usage from OpenAI object: total={total_tokens}, prompt={prompt_tokens}, completion={completion_tokens}")
                    
                     # Try to extract actual cost from provider response
                    try:
                        from .cost_extractor import extract_cost_from_response
                        actual_cost = extract_cost_from_response(response, provider_id)
                    except ImportError:
                        actual_cost = None
                    
                    # Calculate estimated cost and log breakdown
                    try:
                        estimated_cost = analytics.estimate_cost(provider_id, total_tokens, prompt_tokens, completion_tokens)
                        logger.info(f"💰 Cost calculation breakdown for {provider_id}:")
                        logger.info(f"  Tokens: total={total_tokens}, prompt={prompt_tokens}, completion={completion_tokens}")
                        logger.info(f"  Estimated cost: ${estimated_cost:.8f} USD")
                        if actual_cost is not None:
                            logger.info(f"  Actual cost: ${actual_cost:.8f} USD")
                    except Exception as cost_error:
                        logger.debug(f"Cost calculation failed: {cost_error}")
                        estimated_cost = 0.0
                    
                    # If no token usage provided, estimate it with improved accuracy
                    if total_tokens == 0:
                        try:
                            messages = request_data.get('messages', [])
                            estimated_prompt_tokens = count_messages_tokens(messages, model_name)
                            
                            # Count actual completion tokens from response instead of estimating
                            response_content = response.get('choices', [{}])[0].get('message', {}).get('content', '')
                            if response_content:
                                completion_tokens = count_messages_tokens([{
                                    "role": "assistant",
                                    "content": response_content
                                }], model_name)
                            else:
                                # Fallback to estimation if no content
                                max_tokens = request_data.get('max_tokens') or 0
                                if max_tokens > 0:
                                    completion_tokens = min(max_tokens, estimated_prompt_tokens * 2)
                                else:
                                    completion_tokens = max(estimated_prompt_tokens, 50)
                            
                            total_tokens = estimated_prompt_tokens + completion_tokens
                            prompt_tokens = estimated_prompt_tokens
                            logger.debug(f"Counted token usage: {total_tokens} (prompt: {estimated_prompt_tokens}, completion: {completion_tokens})")
                        except Exception as est_error:
                            logger.debug(f"Token counting failed: {est_error}")
                            # Use a more realistic default if counting fails
                            total_tokens = 150
                            prompt_tokens = 0
                            completion_tokens = 0
                    
                    # Always record analytics, even with estimated tokens
                    analytics.record_request(
                        provider_id=provider_id,
                        model_name=model_name,
                        tokens_used=total_tokens,
                        latency_ms=latency_ms,
                        success=True,
                        user_id=getattr(request.state, 'user_id', None),
                        token_id=getattr(request.state, 'token_id', None),
                        prompt_tokens=prompt_tokens if prompt_tokens > 0 else None,
                        completion_tokens=completion_tokens if completion_tokens > 0 else None,
                        actual_cost=actual_cost,
                        analytics_kind='execution'
                    )
                    self._record_dashboard_proxy_event(
                        request,
                        provider_id,
                        model_name,
                        True,
                        latency_ms,
                        metadata={
                            'prompt_tokens': prompt_tokens,
                            'completion_tokens': completion_tokens,
                            'total_tokens': total_tokens,
                            'actual_cost': actual_cost,
                            'stream': bool(stream),
                            'autoselect_id': autoselect_id,
                            'prompt_risk_level': (prompt_analysis or {}).get('risk_level'),
                            'prompt_findings_count': (prompt_analysis or {}).get('summary', {}).get('findings_count', 0),
                        },
                    )
                    
                    # Record context dimensions for model performance tracking
                    try:
                        from aisbf.database import DatabaseRegistry
                        db = DatabaseRegistry.get_config_database()
                        context_config = context_config or {}
                        db.record_context_dimension(
                            provider_id=provider_id,
                            model_name=model_name,
                            context_size=context_config.get('context_size'),
                            condense_context=context_config.get('condense_context'),
                            condense_method=context_config.get('condense_method')
                        )
                    except Exception as context_error:
                        logger.debug(f"Context dimension recording failed: {context_error}")
            except Exception as analytics_error:
                logger.warning(f"Analytics recording failed: {analytics_error}")
            
            logger.info(f"=== RequestHandler.handle_chat_completion END ===")
            return response
        except Exception as e:
            if _should_record_failure(e):
                handler.record_failure()

            # Record failed request analytics
            try:
                analytics = get_analytics()
                latency_ms = (time.time() - request_start_time) * 1000
                
                # Estimate tokens for failed request
                try:
                    messages = request_data.get('messages', [])
                    estimated_tokens = count_messages_tokens(messages, model_name)
                    total_tokens = estimated_tokens
                    prompt_tokens = estimated_tokens
                    completion_tokens = 0  # No completion for failed requests
                except Exception:
                    total_tokens = 50  # Minimal estimate for failed requests
                    prompt_tokens = 50
                    completion_tokens = 0
                
                analytics.record_request(
                    provider_id=provider_id,
                    model_name=model_name,
                    tokens_used=total_tokens,
                    latency_ms=latency_ms,
                    success=False,
                    error_type=type(e).__name__,
                    user_id=getattr(request.state, 'user_id', None),
                    token_id=getattr(request.state, 'token_id', None),
                    prompt_tokens=prompt_tokens,
                    completion_tokens=completion_tokens,
                    actual_cost=None,
                    analytics_kind='execution'
                )
                self._record_dashboard_proxy_event(
                    request,
                    provider_id,
                    model_name,
                    False,
                    latency_ms,
                        metadata={
                            'error_type': type(e).__name__,
                            'prompt_tokens': prompt_tokens,
                            'completion_tokens': completion_tokens,
                            'total_tokens': total_tokens,
                            'stream': bool(request_data.get('stream', False)),
                            'autoselect_id': autoselect_id,
                            'prompt_blocked': 'local security policy' in str(e).lower(),
                        },
                    )
            except Exception as analytics_error:
                logger.warning(f"Analytics recording for failed request failed: {analytics_error}")

            # Forward an upstream quota refusal as a 429 so the caller can back off
            # or fail over on a real signal instead of a generic server error.
            if _is_upstream_rate_limit(e):
                raise HTTPException(status_code=429, detail=str(e))
            raise HTTPException(status_code=500, detail=str(e))

    async def handle_streaming_chat_completion(self, request: Request, provider_id: str, request_data: Dict):
        autoselect_id = request_data.get('_autoselect_id')
        # Check for user-specific provider config first
        if self.user_id and provider_id in self.user_providers:
            provider_config = self.user_providers[provider_id]
        else:
            provider_config = self.config.get_provider(provider_id)

        if isinstance(provider_config, dict):
            api_key_required = provider_config.get('api_key_required', False)
            _provider_type = provider_config.get('type', '')
        else:
            api_key_required = provider_config.api_key_required
            _provider_type = provider_config.type

        if api_key_required:
            api_key = request_data.get('api_key') or request.headers.get('Authorization', '').replace('Bearer ', '')
            if not api_key:
                raise HTTPException(status_code=401, detail="API key required")
        else:
            api_key = None

        handler = get_provider_handler(provider_id, api_key, user_id=self.user_id)

        _raise_if_provider_unavailable(handler, provider_id)

        # Generate system_fingerprint for this request
        # If seed is present in request, generate unique fingerprint per request
        seed = request_data.get('seed')
        system_fingerprint = generate_system_fingerprint(provider_id, seed)

        # Get context configuration and calculate effective context
        model = request_data.get('model')
        messages = request_data.get('messages', [])

        context_config = get_context_config_for_model(
            model_name=model,
            provider_config=provider_config,
            rotation_model_config=None,
            rotation_config=None,
            autoselect_config=None,
        )

        # Apply provider-level default max output tokens when the client didn't set one
        if request_data.get('max_tokens') is None:
            default_max_tokens = get_max_completion_tokens_for_model(
                model_name=model,
                provider_config=provider_config,
                rotation_model_config=None,
            )
            # Lowest-priority fallback: autoselect default threaded in when this
            # request is dispatched directly from an autoselect (provider/model).
            if not default_max_tokens:
                default_max_tokens = request_data.get('_autoselect_default_max_tokens')
            if default_max_tokens:
                request_data['max_tokens'] = default_max_tokens

        effective_context = count_messages_tokens(messages, model)

        prompt_analysis = self._run_prompt_analysis(
            request,
            provider_id,
            model,
            request_data,
            prompt_tokens=effective_context,
            effective_context=effective_context,
        )
        if prompt_analysis and prompt_analysis.get('blocked'):
            raise HTTPException(status_code=400, detail="Prompt blocked by local security policy")
        
        # Apply context condensation if needed
        if context_config.get('condense_context', 0) > 0:
            context_manager = ContextManager(context_config, handler, self.config.get_condensation())
            if context_manager.should_condense(messages, model):
                import logging
                logger = logging.getLogger(__name__)
                logger.info("Context condensation triggered for streaming request")
                messages = await context_manager.condense_context(messages, model)
                effective_context = count_messages_tokens(messages, model)
                logger.info(f"Condensed effective context: {effective_context} tokens")
        
        # Update request_data with condensed messages
        request_data['messages'] = messages

        # Initialize streaming optimizer for this request
        stream_config = StreamingConfig(
            enable_chunk_pooling=True,
            max_pooled_chunks=20,
            chunk_reuse_enabled=True,
            enable_backpressure=True,
            max_pending_chunks=15,
            google_delta_calculation=True,
            kiro_sse_optimization=True
        )
        optimizer = get_streaming_optimizer(stream_config)

        async def stream_generator(effective_context):
            import logging
            import time
            import json
            logger = logging.getLogger(__name__)
            # Track request start time for latency calculation
            request_start_time = time.time()
            try:
                # Apply rate limiting
                await handler.apply_rate_limit()

                response = await handler.handle_request(
                    model=request_data['model'],
                    messages=request_data['messages'],
                    max_tokens=request_data.get('max_tokens'),
                    temperature=request_data.get('temperature', 1.0),
                    stream=True,
                    tools=request_data.get('tools'),
                    tool_choice=request_data.get('tool_choice')
                )
                
                # Check if this is a Google streaming response by checking provider type from config
                # This is more reliable than checking response iterability which can cause false positives
                is_google_stream = _provider_type == 'google'
                is_kiro_stream = _provider_type == 'kiro'
                is_kilo_stream = _provider_type in ('kilo', 'kilocode')
                logger.info(f"Is Google streaming response: {is_google_stream} (provider type: {_provider_type})")
                logger.info(f"Is Kiro streaming response: {is_kiro_stream} (provider type: {_provider_type})")
                logger.info(f"Is Kilo streaming response: {is_kilo_stream} (provider type: {_provider_type})")

                if is_kilo_stream:
                    # Handle Kilo/KiloCode streaming response
                    # Kilo returns an async generator that yields OpenAI-compatible SSE bytes directly
                    # We parse these and pass through with minimal processing
                    
                    accumulated_response_text = ""  # Track full response for token counting
                    chunk_count = 0
                    tool_calls_from_stream = []  # Track tool calls from stream

                    async for chunk in response:
                        chunk_count += 1
                        try:
                            logger.debug(f"Kilo chunk type: {type(chunk)}")

                            # Parse SSE chunk to extract JSON data
                            chunk_data = None
                            
                            if isinstance(chunk, bytes):
                                try:
                                    chunk_str = chunk.decode('utf-8')
                                    # May contain multiple SSE lines
                                    for sse_line in chunk_str.split('\n'):
                                        sse_line = sse_line.strip()
                                        if sse_line.startswith('data: '):
                                            data_str = sse_line[6:].strip()
                                            if data_str and data_str != '[DONE]':
                                                try:
                                                    chunk_data = json.loads(data_str)
                                                except json.JSONDecodeError:
                                                    pass
                                except (UnicodeDecodeError, Exception) as e:
                                    logger.warning(f"Failed to parse Kilo bytes chunk: {e}")
                            elif isinstance(chunk, str):
                                if chunk.startswith('data: '):
                                    data_str = chunk[6:].strip()
                                    if data_str and data_str != '[DONE]':
                                        try:
                                            chunk_data = json.loads(data_str)
                                        except json.JSONDecodeError:
                                            pass
                            
                            if chunk_data:
                                # Extract content and tool calls from chunk
                                choices = chunk_data.get('choices', [])
                                if choices:
                                    delta = choices[0].get('delta', {})
                                    
                                    # Track content
                                    delta_content = delta.get('content', '')
                                    if delta_content:
                                        accumulated_response_text += delta_content
                                    
                                    # Track tool calls
                                    delta_tool_calls = delta.get('tool_calls', [])
                                    if delta_tool_calls:
                                        for tc in delta_tool_calls:
                                            tool_calls_from_stream.append(tc)
                            
                            # Pass through the chunk as-is
                            if isinstance(chunk, bytes):
                                yield chunk
                            elif isinstance(chunk, str):
                                yield chunk.encode('utf-8')
                            else:
                                yield f"data: {json.dumps(chunk)}\n\n".encode('utf-8')

                        except Exception as chunk_error:
                            logger.warning(f"Error processing Kilo chunk: {chunk_error}")
                            continue

                    logger.info(f"Kilo streaming processed {chunk_count} chunks total")

                elif is_kiro_stream:
                    # Handle Kiro streaming response
                    # Kiro returns an async generator that yields OpenAI-compatible SSE strings directly
                    # We need to parse these and handle tool calls properly
                    
                    # Use optimized SSE parser for Kiro
                    if stream_config.kiro_sse_optimization:
                        kiro_parser = KiroSSEParser(buffer_size=stream_config.kiro_buffer_size)
                    else:
                        kiro_parser = None
                    
                    accumulated_response_text = ""  # Track full response for token counting
                    chunk_count = 0
                    tool_calls_from_stream = []  # Track tool calls from stream
                    completion_id = f"chatcmpl-{uuid.uuid4().hex[:24]}"
                    created_time = int(time.time())

                    # Use optimized text accumulator for Kiro
                    kiro_text_accumulator = OptimizedTextAccumulator(
                        max_size=stream_config.max_accumulated_text,
                        enable_truncation=stream_config.enable_text_truncation
                    )

                    async for chunk in response:
                        chunk_count += 1
                        try:
                            logger.debug(f"Kiro chunk type: {type(chunk)}")
                            logger.debug(f"Kiro chunk: {chunk}")

                            # Parse SSE chunk to extract JSON data
                            chunk_data = None
                            
                            if kiro_parser and isinstance(chunk, bytes):
                                # Use optimized parser
                                events = kiro_parser.feed(chunk)
                                for event in events:
                                    if event.get('type') == 'data':
                                        chunk_data = event.get('data')
                                        break
                            elif isinstance(chunk, str) and chunk.startswith('data: '):
                                data_str = chunk[6:].strip()  # Remove 'data: ' prefix
                                if data_str and data_str != '[DONE]':
                                    try:
                                        chunk_data = json.loads(data_str)
                                    except json.JSONDecodeError:
                                        logger.warning(f"Failed to parse Kiro chunk JSON: {data_str}")
                                        continue
                            elif isinstance(chunk, bytes):
                                # Try to decode bytes as SSE
                                try:
                                    chunk_str = chunk.decode('utf-8')
                                    if chunk_str.startswith('data: '):
                                        data_str = chunk_str[6:].strip()
                                        if data_str and data_str != '[DONE]':
                                            chunk_data = json.loads(data_str)
                                except (UnicodeDecodeError, json.JSONDecodeError):
                                    logger.warning(f"Failed to parse Kiro bytes chunk")
                                    continue
                            
                            if chunk_data:
                                # Extract content and tool calls from chunk
                                choices = chunk_data.get('choices', [])
                                if choices:
                                    delta = choices[0].get('delta', {})
                                    
                                    # Track content using optimized accumulator
                                    delta_content = delta.get('content', '')
                                    if delta_content:
                                        accumulated_response_text = kiro_text_accumulator.append(delta_content)
                                    
                                    # Track tool calls
                                    delta_tool_calls = delta.get('tool_calls', [])
                                    if delta_tool_calls:
                                        for tc in delta_tool_calls:
                                            tool_calls_from_stream.append(tc)
                                            logger.debug(f"Collected tool call from Kiro stream: {tc}")
                                
                                # Pass through the chunk as-is
                                if isinstance(chunk, str):
                                    yield chunk.encode('utf-8')
                                elif isinstance(chunk, bytes):
                                    yield chunk
                                else:
                                    yield f"data: {json.dumps(chunk_data)}\n\n".encode('utf-8')
                            else:
                                # Pass through non-data chunks as-is (like [DONE])
                                if isinstance(chunk, str):
                                    yield chunk.encode('utf-8')
                                elif isinstance(chunk, bytes):
                                    yield chunk
                                else:
                                    yield f"data: {json.dumps(chunk)}\n\n".encode('utf-8')

                        except Exception as chunk_error:
                            error_msg = str(chunk_error)
                            logger.warning(f"Error processing Kiro chunk: {error_msg}")
                            logger.warning(f"Chunk type: {type(chunk)}")
                            logger.warning(f"Chunk content: {chunk}")
                            continue

                    # After stream ends, process collected tool calls
                    if tool_calls_from_stream:
                        logger.info(f"Processing {len(tool_calls_from_stream)} tool calls from Kiro stream")
                        
                        # Add required index field to each tool_call
                        # according to OpenAI API specification for streaming
                        indexed_tool_calls = []
                        for idx, tc in enumerate(tool_calls_from_stream):
                            # Extract function with None protection
                            func = tc.get("function") or {}
                            # Use "or" for protection against explicit None in values
                            tool_name = func.get("name") or ""
                            tool_args = func.get("arguments") or "{}"
                            
                            logger.debug(f"Tool call [{idx}] '{tool_name}': id={tc.get('id')}, args_length={len(tool_args)}")
                            
                            indexed_tc = {
                                "index": idx,
                                "id": tc.get("id"),
                                "type": tc.get("type", "function"),
                                "function": {
                                    "name": tool_name,
                                    "arguments": tool_args
                                }
                            }
                            indexed_tool_calls.append(indexed_tc)
                        
                        # Send tool calls chunk
                        tool_calls_chunk = {
                            "id": completion_id,
                            "object": "chat.completion.chunk",
                            "created": created_time,
                            "model": request_data['model'],
                            "choices": [{
                                "index": 0,
                                "delta": {"tool_calls": indexed_tool_calls},
                                "finish_reason": None
                            }]
                        }
                        yield f"data: {json.dumps(tool_calls_chunk, ensure_ascii=False)}\n\n".encode('utf-8')

                    logger.info(f"Kiro streaming processed {chunk_count} chunks total")

                elif is_google_stream:
                    # Handle Google's streaming response
                    # Google provider returns an async generator
                    # Note: Google returns accumulated text, so we need to track and send only deltas
                    chunk_id = 0
                    accumulated_text = ""  # Track text we've already sent
                    seen_tool_call_ids = set()  # Track which tool call IDs we've already sent
                    last_chunk_id = None  # Track the last chunk for finish_reason
                    created_time = int(time.time())
                    response_id = f"google-{request_data['model']}-{created_time}"
                    
                    # Track completion tokens for Google responses (since Google doesn't provide them)
                    completion_tokens = 0
                    accumulated_response_text = ""  # Track full response for token counting
                    
                    # Use optimized text accumulator for memory efficiency
                    text_accumulator = OptimizedTextAccumulator(
                        max_size=stream_config.google_accumulated_text_limit,
                        enable_truncation=stream_config.enable_text_truncation
                    )
                    
                    # Collect all chunks first to know when we're at the last one
                    chunks_list = []
                    async for chunk in response:
                        chunks_list.append(chunk)
                    
                    total_chunks = len(chunks_list)
                    chunk_idx = 0
                    
                    for chunk in chunks_list:
                        try:
                            logger.debug(f"Google chunk type: {type(chunk)}")
                            logger.debug(f"Google chunk: {chunk}")
                            
                            # Extract text and tool calls from Google chunk (this is accumulated text)
                            chunk_text = ""
                            chunk_tool_calls = []
                            finish_reason = None
                            try:
                                if hasattr(chunk, 'candidates') and chunk.candidates:
                                    candidate = chunk.candidates[0] if chunk.candidates else None
                                    if candidate and hasattr(candidate, 'content') and candidate.content:
                                        if hasattr(candidate.content, 'parts') and candidate.content.parts:
                                            for part in candidate.content.parts:
                                                # Extract text content
                                                if hasattr(part, 'text') and part.text:
                                                    chunk_text += part.text
                                                # Extract function calls (Google's format)
                                                if hasattr(part, 'function_call') and part.function_call:
                                                    function_call = part.function_call
                                                    # Convert Google function call to OpenAI format
                                                    import json
                                                    openai_tool_call = {
                                                        "id": f"call_{len(chunk_tool_calls)}",
                                                        "type": "function",
                                                        "function": {
                                                            "name": function_call.name,
                                                            "arguments": json.dumps(function_call.args) if hasattr(function_call, 'args') else "{}"
                                                        }
                                                    }
                                                    chunk_tool_calls.append(openai_tool_call)
                                                    logger.info(f"Extracted tool call from Google chunk: {openai_tool_call}")
                                    # Check for finish reason in candidate
                                    if hasattr(candidate, 'finish_reason'):
                                        google_finish = str(candidate.finish_reason)
                                        if google_finish in ('STOP', 'END_TURN', 'FINISH_REASON_UNSPECIFIED'):
                                            finish_reason = "stop"
                                        elif google_finish == 'MAX_TOKENS':
                                            finish_reason = "length"
                            except Exception as e:
                                logger.error(f"Error extracting text from Google chunk: {e}")
                            
                            # Calculate the delta (only the new text since last chunk) using optimized function
                            if stream_config.google_delta_calculation:
                                delta_text = calculate_google_delta(chunk_text, accumulated_text)
                            else:
                                delta_text = chunk_text[len(accumulated_text):] if chunk_text.startswith(accumulated_text) else chunk_text
                            accumulated_text = chunk_text  # Update accumulated text for next iteration
                            
                            # Calculate delta tool calls (only tool calls we haven't seen before)
                            delta_tool_calls = []
                            for tool_call in chunk_tool_calls:
                                if tool_call["id"] not in seen_tool_call_ids:
                                    delta_tool_calls.append(tool_call)
                                    seen_tool_call_ids.add(tool_call["id"])
                                    logger.info(f"Adding tool call to delta: {tool_call}")
                            
                            # Check if this is the last chunk
                            is_last_chunk = (chunk_idx == total_chunks - 1)
                            chunk_finish_reason = finish_reason if is_last_chunk else None
                            
                            # Debug logging
                            logger.debug(f"Chunk {chunk_idx}/{total_chunks}: delta_text='{delta_text}', delta_tool_calls={len(delta_tool_calls)}, is_last={is_last_chunk}, condition={bool(delta_text or delta_tool_calls or is_last_chunk)}")
                            
                            # Only send if there's new content, new tool calls, or it's the last chunk with finish_reason
                            if delta_tool_calls or delta_text or is_last_chunk:
                                # Use optimized chunk from pool
                                openai_chunk = optimizer.chunk_pool.acquire()
                                try:
                                    openai_chunk.update({
                                        "id": response_id,
                                        "object": "chat.completion.chunk",
                                        "created": created_time,
                                        "model": request_data['model'],
                                        "service_tier": None,
                                        "system_fingerprint": system_fingerprint,
                                        "usage": None,
                                        "provider": provider_id,
                                        "choices": [{
                                            "index": 0,
                                            "delta": {
                                                "content": delta_text if delta_text else "",
                                                "refusal": None,
                                                "role": "assistant",
                                                "tool_calls": delta_tool_calls if len(delta_tool_calls) > 0 else None
                                            },
                                            "finish_reason": chunk_finish_reason,
                                            "logprobs": None,
                                            "native_finish_reason": chunk_finish_reason
                                        }]
                                    })
                                    
                                    chunk_id += 1
                                    logger.debug(f"OpenAI chunk (delta length: {len(delta_text)}, finish: {chunk_finish_reason})")
                                    
                                    # Track completion tokens for Google responses using optimized accumulator
                                    if delta_text:
                                        accumulated_response_text = text_accumulator.append(delta_text)
                                    
                                    # Serialize as JSON and yield
                                    yield f"data: {json.dumps(openai_chunk)}\n\n".encode('utf-8')
                                finally:
                                    optimizer.chunk_pool.release(openai_chunk)
                            
                            chunk_idx += 1
                        except Exception as chunk_error:
                            error_msg = str(chunk_error)
                            logger.error(f"Error processing Google chunk: {error_msg}")
                            logger.error(f"Chunk type: {type(chunk)}")
                            logger.error(f"Chunk content: {chunk}")
                            # Skip this chunk and continue
                            chunk_idx += 1
                            continue
                    
                    # Send final chunk with usage statistics (empty content)
                    # Calculate completion tokens for Google responses (count tokens in full response)
                    if accumulated_response_text:
                        completion_tokens = count_messages_tokens([{"role": "assistant", "content": accumulated_response_text}], request_data['model'])
                    total_tokens = effective_context + completion_tokens
                    final_chunk = {
                        "id": response_id,
                        "object": "chat.completion.chunk",
                        "created": created_time,
                        "model": request_data['model'],
                        "service_tier": None,
                        "system_fingerprint": system_fingerprint,
                        "usage": {
                            "prompt_tokens": effective_context,
                            "completion_tokens": completion_tokens,
                            "total_tokens": total_tokens,
                            "effective_context": effective_context
                        },
                        "provider": provider_id,
                        "choices": [{
                            "index": 0,
                            "delta": {
                                "content": "",
                                "function_call": None,
                                "refusal": None,
                                "role": "assistant",
                                "tool_calls": None
                            },
                            "finish_reason": None,
                            "logprobs": None,
                            "native_finish_reason": None
                        }]
                    }
                    yield f"data: {json.dumps(final_chunk)}\n\n".encode('utf-8')
                else:
                    # Handle OpenAI/Anthropic/Claude streaming responses
                    # Some providers return async generators, others return sync iterables
                    accumulated_response_text = ""  # Track full response for token counting
                    
                    # Check if response is an async generator
                    import inspect
                    if inspect.iscoroutinefunction(response) or hasattr(response, '__aiter__'):
                        # Handle async generator (like Claude, Kiro)
                        logger.info(f"Detected async generator response, using async for loop")
                        async for chunk in response:
                            try:
                                logger.debug(f"Async chunk type: {type(chunk)}")
                                logger.debug(f"Async chunk: {chunk}")
                                
                                # For async generators, chunks might be bytes or pre-formatted SSE strings
                                if isinstance(chunk, bytes):
                                    logger.debug(f"Yielding raw bytes chunk: {len(chunk)} bytes")
                                    yield chunk
                                elif isinstance(chunk, str):
                                    # Already SSE-formatted (e.g. "data: {...}\n\n") — pass through directly
                                    yield chunk.encode('utf-8')
                                else:
                                    # Fallback: treat as dict and serialize
                                    chunk_dict = chunk.model_dump() if hasattr(chunk, 'model_dump') else chunk
                                    
                                    # Track response content for token calculation
                                    if isinstance(chunk_dict, dict):
                                        choices = chunk_dict.get('choices', [])
                                        if choices:
                                            delta = choices[0].get('delta', {})
                                            delta_content = delta.get('content', '')
                                            if delta_content:
                                                accumulated_response_text += delta_content
                                    
                                    # Add effective_context to the last chunk (when finish_reason is present)
                                    if isinstance(chunk_dict, dict):
                                        choices = chunk_dict.get('choices', [])
                                        if choices and choices[0].get('finish_reason') is not None:
                                            # This is the last chunk, add effective_context
                                            if 'usage' not in chunk_dict:
                                                chunk_dict['usage'] = {}
                                            chunk_dict['usage']['effective_context'] = effective_context
                                            
                                            # If provider doesn't provide token counts, calculate them
                                            if chunk_dict['usage'].get('total_tokens') is None:
                                                # Calculate completion tokens from accumulated response
                                                if accumulated_response_text:
                                                    completion_tokens = count_messages_tokens([{"role": "assistant", "content": accumulated_response_text}], request_data['model'])
                                                else:
                                                    completion_tokens = 0
                                                total_tokens = effective_context + completion_tokens
                                                chunk_dict['usage']['prompt_tokens'] = effective_context
                                                chunk_dict['usage']['completion_tokens'] = completion_tokens
                                                chunk_dict['usage']['total_tokens'] = total_tokens
                                    
                                    yield f"data: {json.dumps(chunk_dict)}\n\n".encode('utf-8')
                            except Exception as chunk_error:
                                error_msg = str(chunk_error)
                                logger.warning(f"Error processing async chunk: {error_msg}")
                                logger.warning(f"Chunk type: {type(chunk)}")
                                logger.warning(f"Chunk content: {chunk}")
                                continue
                    else:
                        # Handle sync iterable (like OpenAI SDK)
                        logger.info(f"Detected sync iterable response, using regular for loop")
                        for chunk in response:
                            try:
                                # Debug: Log chunk type and content before serialization
                                logger.debug(f"Chunk type: {type(chunk)}")
                                logger.debug(f"Chunk: {chunk}")
                                
                                # For OpenAI-compatible providers, just pass through the raw chunk
                                # Convert chunk to dict and serialize as JSON
                                chunk_dict = chunk.model_dump() if hasattr(chunk, 'model_dump') else chunk
                                
                                # Track response content for token calculation
                                if isinstance(chunk_dict, dict):
                                    choices = chunk_dict.get('choices', [])
                                    if choices:
                                        delta = choices[0].get('delta', {})
                                        delta_content = delta.get('content', '')
                                        if delta_content:
                                            accumulated_response_text += delta_content
                                
                                # Add effective_context to the last chunk (when finish_reason is present)
                                if isinstance(chunk_dict, dict):
                                    choices = chunk_dict.get('choices', [])
                                    if choices and choices[0].get('finish_reason') is not None:
                                        # This is the last chunk, add effective_context
                                        if 'usage' not in chunk_dict:
                                            chunk_dict['usage'] = {}
                                        chunk_dict['usage']['effective_context'] = effective_context
                                        
                                        # If provider doesn't provide token counts, calculate them
                                        if chunk_dict['usage'].get('total_tokens') is None:
                                            # Calculate completion tokens from accumulated response
                                            if accumulated_response_text:
                                                completion_tokens = count_messages_tokens([{"role": "assistant", "content": accumulated_response_text}], request_data['model'])
                                            else:
                                                completion_tokens = 0
                                            total_tokens = effective_context + completion_tokens
                                            chunk_dict['usage']['prompt_tokens'] = effective_context
                                            chunk_dict['usage']['completion_tokens'] = completion_tokens
                                            chunk_dict['usage']['total_tokens'] = total_tokens
                                
                                yield f"data: {json.dumps(chunk_dict)}\n\n".encode('utf-8')
                            except Exception as chunk_error:
                                # Handle errors during chunk serialization
                                error_msg = str(chunk_error)
                                logger.warning(f"Error serializing chunk: {error_msg}")
                                logger.warning(f"Chunk type: {type(chunk)}")
                                logger.warning(f"Chunk content: {chunk}")
                                # Skip this chunk and continue with the next one
                                continue
                handler.record_success()
                
                # Record analytics for streaming request
                try:
                    analytics = get_analytics()
                    # Calculate latency
                    latency_ms = (time.time() - request_start_time) * 1000
                    logger.info(f"Streaming Analytics: latency_ms={latency_ms:.2f}")
                    
                    # Calculate total tokens from accumulated response
                    if accumulated_response_text:
                        completion_tokens = count_messages_tokens([{"role": "assistant", "content": accumulated_response_text}], request_data['model'])
                    else:
                        completion_tokens = 0
                    total_tokens = effective_context + completion_tokens
                    
                    analytics.record_request(
                        provider_id=provider_id,
                        model_name=request_data['model'],
                        tokens_used=total_tokens,
                        latency_ms=latency_ms,
                        success=True,
                        user_id=getattr(request.state, 'user_id', None),
                        token_id=getattr(request.state, 'token_id', None),
                        prompt_tokens=effective_context,
                        completion_tokens=completion_tokens,
                        actual_cost=None,
                        analytics_kind='execution'
                    )
                    self._record_dashboard_proxy_event(
                        request,
                        provider_id,
                        request_data['model'],
                        True,
                        latency_ms,
                        metadata={
                            'prompt_tokens': effective_context,
                            'completion_tokens': completion_tokens,
                            'total_tokens': total_tokens,
                            'stream': True,
                            'autoselect_id': autoselect_id,
                            'prompt_risk_level': (prompt_analysis or {}).get('risk_level'),
                            'prompt_findings_count': (prompt_analysis or {}).get('summary', {}).get('findings_count', 0),
                        },
                    )
                except Exception as analytics_error:
                    logger.warning(f"Analytics recording for streaming request failed: {analytics_error}")
                
            except Exception as e:
                if _should_record_failure(e):
                    handler.record_failure()

                # Record analytics for failed streaming request
                try:
                    analytics = get_analytics()
                    # Calculate latency
                    latency_ms = (time.time() - request_start_time) * 1000
                    logger.info(f"Failed Streaming Analytics: latency_ms={latency_ms:.2f}")
                    
                    # Estimate tokens for failed request
                    try:
                        messages = request_data.get('messages', [])
                        estimated_tokens = count_messages_tokens(messages, request_data['model'])
                        total_tokens = estimated_tokens
                        prompt_tokens = estimated_tokens
                        completion_tokens = 0
                    except Exception:
                        total_tokens = 50  # Minimal estimate for failed requests
                        prompt_tokens = 50
                        completion_tokens = 0
                    
                    analytics.record_request(
                        provider_id=provider_id,
                        model_name=request_data['model'],
                        tokens_used=total_tokens,
                        latency_ms=latency_ms,
                        success=False,
                        error_type=type(e).__name__,
                        user_id=getattr(request.state, 'user_id', None),
                        token_id=getattr(request.state, 'token_id', None),
                        prompt_tokens=prompt_tokens,
                        completion_tokens=completion_tokens,
                        actual_cost=None,
                        analytics_kind='execution'
                    )
                    self._record_dashboard_proxy_event(
                        request,
                        provider_id,
                        request_data['model'],
                        False,
                        latency_ms,
                        metadata={
                            'error_type': type(e).__name__,
                            'prompt_tokens': prompt_tokens,
                            'completion_tokens': completion_tokens,
                            'total_tokens': total_tokens,
                            'stream': True,
                            'autoselect_id': autoselect_id,
                            'prompt_blocked': 'local security policy' in str(e).lower(),
                        },
                    )
                except Exception as analytics_error:
                    logger.warning(f"Analytics recording for failed streaming request failed: {analytics_error}")
                
                error_dict = {"error": str(e)}
                yield f"data: {json.dumps(error_dict)}\n\n".encode('utf-8')

        return StreamingResponse(stream_generator(effective_context), media_type="text/event-stream")

    async def handle_model_list(self, request: Request, provider_id: str) -> List[Dict]:
        provider_config = self.config.get_provider(provider_id)

        if provider_config.api_key_required:
            api_key = request.headers.get('Authorization', '').replace('Bearer ', '')
            if not api_key:
                raise HTTPException(status_code=401, detail="API key required")
        else:
            api_key = None

        # First check if we already have models for same provider type + same endpoint from same user
        import logging
        logger = logging.getLogger(__name__)
        
        # Check all other providers with same type, same endpoint and same user
        same_providers = []
        if self.user_id:
            # Check user providers
            for pid, pconfig in self.user_providers.items():
                if pid != provider_id and pconfig.type == provider_config.type and pconfig.endpoint == provider_config.endpoint:
                    same_providers.append(pid)
        else:
            # Check global providers
            for pid, pconfig in self.config.providers.items():
                if pid != provider_id and pconfig.type == provider_config.type and pconfig.endpoint == provider_config.endpoint:
                    same_providers.append(pid)
        
        # If there are matching providers, check if they have cached models
        for same_pid in same_providers:
            # Check if this provider already has models loaded
            try:
                same_handler = get_provider_handler(same_pid, user_id=self.user_id)
                if hasattr(same_handler, '_cached_models') and same_handler._cached_models and len(same_handler._cached_models) > 0:
                    logger.info(f"Reusing models from existing provider {same_pid} for {provider_id} (same type and endpoint)")
                    # Copy models and apply correct provider_id
                    models = []
                    for model in same_handler._cached_models:
                        # Create new model instance with correct provider_id
                        model_copy = model.copy()
                        model_copy.provider_id = provider_id
                        models.append(model_copy)
                    
                    # Also cache on this handler
                    handler = get_provider_handler(provider_id, api_key, user_id=self.user_id)
                    handler._cached_models = models
                    
                    # Skip rate limit and direct model fetch
                    model_filter = getattr(provider_config, 'model_filter', None)
                    if model_filter and (not getattr(provider_config, "models", []) or len(getattr(provider_config, "models", [])) == 0):
                        logger.info(f"Applying model filter '{model_filter}' to provider {provider_id}")
                        original_count = len(models)
                        models = [m for m in models if model_filter.lower() in m.id.lower()]
                        logger.info(f"Model filter applied: {original_count} -> {len(models)} models")
                    
                    # Enhance model information with context window and capabilities
                    enhanced_models = []
                    current_time = int(time_module.time())
                    for model in models:
                        model_dict = model.model_dump() if hasattr(model, 'model_dump') else model.dict()
                        model_name = model_dict.get('id', '')
                        
                        # Add OpenAI-compatible required fields
                        model_dict['object'] = 'model'
                        model_dict['created'] = current_time
                        model_dict['owned_by'] = provider_config.name
                        
                        # Try to find model config in provider config
                        model_config = None
                        if getattr(provider_config, "models", []):
                            for m in getattr(provider_config, "models", []):
                                if m.name == model_name:
                                    model_config = m
                                    break
                        
                        # Add context window information - use dynamically fetched value unless manually configured
                        # Priority: manually configured > dynamically fetched > inferred
                        if model_config and hasattr(model_config, 'context_size') and model_config.context_size:
                            # Manually configured - use this value
                            model_dict['context_window'] = model_config.context_size
                        elif model_dict.get('context_size'):
                            # Dynamically fetched from provider - use this value
                            model_dict['context_window'] = model_dict['context_size']
                        elif getattr(provider_config, 'default_context_size', None):
                            # Provider-level default (applies when no per-model config exists)
                            model_dict['context_window'] = provider_config.default_context_size
                        else:
                            # Fall back to inference
                            model_dict['context_window'] = self._infer_context_window(model_name, provider_config.type)

                        # Add context_length for compatibility - same priority order as context_window
                        if model_config and hasattr(model_config, 'context_size') and model_config.context_size:
                            model_dict['context_length'] = model_config.context_size
                        elif model_dict.get('context_size'):
                            model_dict['context_length'] = model_dict['context_size']
                        elif getattr(provider_config, 'default_context_size', None):
                            model_dict['context_length'] = provider_config.default_context_size
                        elif model_dict.get('context_length'):
                            model_dict['context_length'] = model_dict['context_length']
                        
                        # Add pricing if available (from dynamic fetch)
                        if model_dict.get('pricing'):
                            model_dict['pricing'] = model_dict['pricing']
                        
                        # Add description if available (from dynamic fetch)
                        if model_dict.get('description'):
                            model_dict['description'] = model_dict['description']
                        
                        # Add top_provider info if available (from dynamic fetch)
                        if model_dict.get('top_provider'):
                            model_dict['top_provider'] = model_dict['top_provider']
                        
                        # Add supported_parameters if available (from dynamic fetch)
                        if model_dict.get('supported_parameters'):
                            model_dict['supported_parameters'] = model_dict['supported_parameters']
                        
                        # Add capabilities information
                        if model_config and hasattr(model_config, 'capabilities'):
                            model_dict['capabilities'] = model_config.capabilities
                        elif 'capabilities' not in model_dict:
                            # Auto-detect capabilities based on model name and provider type
                            model_dict['capabilities'] = self._detect_capabilities(model_name, provider_config.type)
                        
                        enhanced_models.append(model_dict)
                    
                    return enhanced_models
            except Exception as e:
                logger.debug(f"Failed to reuse models from {same_pid}: {e}")
                continue
        
        # No existing models found, proceed normally
        handler = get_provider_handler(provider_id, api_key, user_id=self.user_id)
        try:
            # Apply rate limiting
            await handler.apply_rate_limit()

            models = await handler.get_models()
            
            # Check if this is an auth status response (dictionary instead of Model list)
            # Some providers (Kilo, Qwen, Claude) return auth status instead of models when not authenticated
            if isinstance(models, dict) and 'status' in models:
                logger.info(f"Provider {provider_id} returned auth status instead of models: {models.get('status')}")
                # Return empty models list but include auth status in response headers
                models = []
            
            # Cache the models on the handler
            handler._cached_models = models
            
            # Apply model filter if configured and no models are manually specified
            model_filter = getattr(provider_config, 'model_filter', None)
            if model_filter and (not getattr(provider_config, "models", []) or len(getattr(provider_config, "models", [])) == 0):
                import logging
                logger = logging.getLogger(__name__)
                logger.info(f"Applying model filter '{model_filter}' to provider {provider_id}")
                
                # Filter models whose ID contains the filter word (case-insensitive)
                original_count = len(models)
                models = [m for m in models if model_filter.lower() in m.id.lower()]
                logger.info(f"Model filter applied: {original_count} -> {len(models)} models")
            
            # Enhance model information with context window and capabilities
            enhanced_models = []
            current_time = int(time_module.time())
            for model in models:
                model_dict = model.model_dump() if hasattr(model, 'model_dump') else model.dict()
                model_name = model_dict.get('id', '')
                
                # Add OpenAI-compatible required fields
                model_dict['object'] = 'model'
                model_dict['created'] = current_time
                model_dict['owned_by'] = provider_config.name
                
                # Try to find model config in provider config
                model_config = None
                if getattr(provider_config, "models", []):
                    for m in getattr(provider_config, "models", []):
                        if m.name == model_name:
                            model_config = m
                            break
                
                # Add context window information - use dynamically fetched value unless manually configured
                # Priority: manually configured > dynamically fetched > inferred
                if model_config and hasattr(model_config, 'context_size') and model_config.context_size:
                    # Manually configured - use this value
                    model_dict['context_window'] = model_config.context_size
                elif model_dict.get('context_size'):
                    # Dynamically fetched from provider - use this value
                    model_dict['context_window'] = model_dict['context_size']
                elif getattr(provider_config, 'default_context_size', None):
                    # Provider-level default (applies when no per-model config exists)
                    model_dict['context_window'] = provider_config.default_context_size
                else:
                    # Fall back to inference
                    model_dict['context_window'] = self._infer_context_window(model_name, provider_config.type)

                # Add context_length for compatibility - same priority order as context_window
                if model_config and hasattr(model_config, 'context_size') and model_config.context_size:
                    model_dict['context_length'] = model_config.context_size
                elif model_dict.get('context_size'):
                    model_dict['context_length'] = model_dict['context_size']
                elif getattr(provider_config, 'default_context_size', None):
                    model_dict['context_length'] = provider_config.default_context_size
                elif model_dict.get('context_length'):
                    model_dict['context_length'] = model_dict['context_length']
                
                # Add pricing if available (from dynamic fetch)
                if model_dict.get('pricing'):
                    model_dict['pricing'] = model_dict['pricing']
                
                # Add description if available (from dynamic fetch)
                if model_dict.get('description'):
                    model_dict['description'] = model_dict['description']
                
                # Add top_provider info if available (from dynamic fetch)
                if model_dict.get('top_provider'):
                    model_dict['top_provider'] = model_dict['top_provider']
                
                # Add supported_parameters if available (from dynamic fetch)
                if model_dict.get('supported_parameters'):
                    model_dict['supported_parameters'] = model_dict['supported_parameters']
                
                # Add capabilities information
                if model_config and hasattr(model_config, 'capabilities'):
                    model_dict['capabilities'] = model_config.capabilities
                elif 'capabilities' not in model_dict:
                    # Auto-detect capabilities based on model name and provider type
                    model_dict['capabilities'] = self._detect_capabilities(model_name, provider_config.type)
                
                enhanced_models.append(model_dict)
            
            return enhanced_models
        except Exception as e:
            raise HTTPException(status_code=500, detail=str(e))
    
    def _infer_context_window(self, model_name: str, provider_type: str) -> int:
        """Infer context window size from model name or provider type"""
        model_lower = model_name.lower()
        
        # Known model patterns
        if 'gpt-4' in model_lower:
            if 'turbo' in model_lower or '1106' in model_lower or '0125' in model_lower:
                return 128000
            return 8192
        elif 'gpt-3.5' in model_lower:
            if 'turbo' in model_lower and ('1106' in model_lower or '0125' in model_lower):
                return 16385
            return 4096
        elif 'claude-3' in model_lower:
            return 200000
        elif 'claude-2' in model_lower:
            return 100000
        elif 'gemini' in model_lower:
            if '1.5' in model_lower:
                return 2000000 if 'pro' in model_lower else 1000000
            elif '2.0' in model_lower:
                return 1000000
            return 32000
        elif 'llama' in model_lower:
            if '3' in model_lower:
                return 128000
            return 4096
        elif 'mistral' in model_lower:
            if 'large' in model_lower:
                return 32000
            return 8192
        
        # Default based on provider type
        if provider_type == 'google':
            return 32000
        elif provider_type == 'anthropic':
            return 100000
        elif provider_type == 'openai':
            return 8192
        
        # Generic default
        return 4096
    
    def _detect_capabilities(self, model_name: str, provider_type: str) -> List[str]:
        """Auto-detect model capabilities based on model name and provider type."""
        return infer_model_capabilities(model_name=model_name, provider_type=provider_type).capabilities
    
    async def handle_generic_proxy(self, request: Request, provider_id: str, endpoint_path: str, body: dict, method: str = "POST") -> JSONResponse:
        """Forward a request to the provider's native endpoint and return the response."""
        import httpx
        import logging
        logger = logging.getLogger(__name__)
        body = self._adapt_studio_workflow_payload(provider_id, endpoint_path, body)
        incoming_headers = {
            key: value for key, value in request.headers.items()
            if key.lower() not in {"host", "content-length", "connection"}
        }
        query_params = dict(request.query_params)
        content_type = request.headers.get('content-type', '')
        multipart_payload = None

        if 'multipart/form-data' in content_type.lower():
            form = await request.form()
            fields = []
            files = []
            for key, value in form.multi_items():
                filename = getattr(value, 'filename', None)
                if filename is not None:
                    content = await value.read()
                    files.append({
                        'name': key,
                        'filename': filename,
                        'content_type': getattr(value, 'content_type', None),
                        'data_base64': base64.b64encode(content).decode('ascii'),
                    })
                else:
                    fields.append({'name': key, 'value': str(value)})
            multipart_payload = {'fields': fields, 'files': files}

        # Support user-defined providers (dict format) and global providers (object format)
        if self.user_id and provider_id in self.user_providers:
            provider_config = self.user_providers[provider_id]
            base_url = (provider_config.get('endpoint') or '').rstrip('/')
            api_key_required = provider_config.get('api_key_required', False)
            config_api_key = provider_config.get('api_key')
        else:
            provider_config = self.config.get_provider(provider_id)
            base_url = (getattr(provider_config, 'endpoint', '') or '').rstrip('/')
            api_key_required = getattr(provider_config, 'api_key_required', False)
            config_api_key = getattr(provider_config, 'api_key', None)

        # Strip trailing /chat/completions or /completions to get the real base
        for suffix in ['/chat/completions', '/completions']:
            if base_url.endswith(suffix):
                base_url = base_url[:-len(suffix)]
                break

        url = f"{base_url}/{endpoint_path.lstrip('/')}"

        headers = {'Content-Type': 'application/json'}
        if api_key_required:
            api_key = request.headers.get('Authorization', '').replace('Bearer ', '') or config_api_key
            if api_key:
                headers['Authorization'] = f'Bearer {api_key}'

        logger.info(f"Generic proxy [{method}]: {provider_id} -> {url}")
        try:
            provider_type = provider_config.get('type') if isinstance(provider_config, dict) else getattr(provider_config, 'type', None)
            if provider_type == 'coderai':
                provider_handler = get_provider_handler(provider_id, config_api_key, user_id=self.user_id)
                wants_stream = 'text/event-stream' in request.headers.get('accept', '').lower() or endpoint_path in {"v1/audio/progress", "v1/video/progress", "v1/images/progress"}
                status_code, payload = await provider_handler.proxy_native_request(
                    endpoint_path,
                    body,
                    method=method,
                    headers=incoming_headers,
                    query_params=query_params,
                    content_type=content_type,
                    multipart=multipart_payload,
                    stream=wants_stream,
                )
                _skip_headers = {'content-length', 'transfer-encoding', 'content-encoding'}
                response_headers = {k: v for k, v in (payload.get('headers') or {}).items() if k.lower() not in _skip_headers}
                if payload.get('stream_chunks'):
                    media_type = payload.get('content_type') or 'text/event-stream'

                    async def iter_stream():
                        for chunk in payload.get('stream_chunks', []):
                            yield base64.b64decode(chunk)

                    return StreamingResponse(iter_stream(), status_code=status_code, media_type=media_type, headers=response_headers)
                if payload.get('body_base64'):
                    media_type = payload.get('content_type') or 'application/octet-stream'
                    return Response(content=base64.b64decode(payload['body_base64']), status_code=status_code, media_type=media_type, headers=response_headers)
                body_content = payload.get('body', payload)
                # Rewrite any /v1/files/ URLs in JSON responses so the client
                # fetches them through our proxy (via broker tunnel when applicable).
                if isinstance(body_content, str) and '/v1/files/' in body_content:
                    uses_broker = await provider_handler._use_broker()
                    body_content = self._rewrite_coderai_file_urls(
                        body_content, provider_id, request, via_broker=uses_broker
                    )
                # body from the broker dispatcher is already a JSON string — return it
                # directly to avoid double-encoding by JSONResponse.
                if isinstance(body_content, str):
                    return Response(
                        content=body_content.encode("utf-8"),
                        status_code=status_code,
                        media_type="application/json",
                        headers=response_headers,
                    )
                return JSONResponse(status_code=status_code, content=body_content, headers=response_headers)

            async with httpx.AsyncClient(timeout=300) as client:
                if method == "GET":
                    resp = await client.get(url, headers=headers, params=query_params)
                elif method == "DELETE":
                    resp = await client.delete(url, headers=headers, params=query_params)
                else:
                    if multipart_payload is not None:
                        files = []
                        data = []
                        for field in multipart_payload['fields']:
                            data.append((field['name'], field['value']))
                        for file_entry in multipart_payload['files']:
                            files.append((file_entry['name'], (file_entry['filename'], base64.b64decode(file_entry['data_base64']), file_entry.get('content_type') or 'application/octet-stream')))
                        resp = await client.post(url, data=data, files=files, headers={k: v for k, v in headers.items() if k.lower() != 'content-type'}, params=query_params)
                    else:
                        resp = await client.post(url, json=body, headers=headers, params=query_params)
                try:
                    content = resp.json()
                except Exception:
                    content = {"detail": resp.text}
                try:
                    usage = content.get('usage', {}) if isinstance(content, dict) else {}
                    self._settle_market_result(provider_id, usage, 1, {'kind': 'generic_proxy', 'endpoint_path': endpoint_path}, request_data=request_data)
                except Exception as market_error:
                    logger.warning(f"Market settlement failed for generic proxy {provider_id}: {market_error}")
                return JSONResponse(status_code=resp.status_code, content=content)
        except Exception as e:
            logger.error(f"Generic proxy error: {e}", exc_info=True)
            raise HTTPException(status_code=502, detail=str(e))

    async def handle_audio_transcription(self, request: Request, provider_id: str, form_data) -> Dict:
        """Handle audio transcription requests"""
        import logging
        logger = logging.getLogger(__name__)
        logger.info(f"=== Audio Transcription Handler START ===")
        
        provider_config = self.config.get_provider(provider_id)
        
        if provider_config.api_key_required:
            api_key = request.headers.get('Authorization', '').replace('Bearer ', '')
            if not api_key:
                raise HTTPException(status_code=401, detail="API key required")
        else:
            api_key = None
        
        handler = get_provider_handler(provider_id, api_key, user_id=self.user_id)
        
        _raise_if_provider_unavailable(handler, provider_id)
        
        try:
            await handler.apply_rate_limit()
            result = await handler.handle_audio_transcription(form_data)
            handler.record_success()
            try:
                usage = result.get('usage', {}) if isinstance(result, dict) else {}
                self._settle_market_result(provider_id, usage, 1, {'kind': 'audio_transcription'}, request_data=request_data)
            except Exception as market_error:
                logger.warning(f"Market settlement failed for audio transcription {provider_id}: {market_error}")
            return result
        except Exception as e:
            if _should_record_failure(e):
                handler.record_failure()
            raise HTTPException(status_code=500, detail=str(e))
    
    async def handle_text_to_speech(self, request: Request, provider_id: str, request_data: Dict) -> StreamingResponse:
        """Handle text-to-speech requests"""
        import logging
        logger = logging.getLogger(__name__)
        logger.info(f"=== Text-to-Speech Handler START ===")
        
        provider_config = self.config.get_provider(provider_id)
        
        if provider_config.api_key_required:
            api_key = request_data.get('api_key') or request.headers.get('Authorization', '').replace('Bearer ', '')
            if not api_key:
                raise HTTPException(status_code=401, detail="API key required")
        else:
            api_key = None
        
        handler = get_provider_handler(provider_id, api_key, user_id=self.user_id)
        
        _raise_if_provider_unavailable(handler, provider_id)
        
        try:
            await handler.apply_rate_limit()
            result = await handler.handle_text_to_speech(request_data)
            handler.record_success()
            try:
                usage = result.get('usage', {}) if isinstance(result, dict) else {}
                self._settle_market_result(provider_id, usage, 1, {'kind': 'text_to_speech'}, request_data=request_data)
            except Exception as market_error:
                logger.warning(f"Market settlement failed for text to speech {provider_id}: {market_error}")
            return result
        except Exception as e:
            if _should_record_failure(e):
                handler.record_failure()
            raise HTTPException(status_code=500, detail=str(e))
    
    async def handle_image_generation(self, request: Request, provider_id: str, request_data: Dict) -> Dict:
        """Handle image generation requests with URL rewriting"""
        import logging
        logger = logging.getLogger(__name__)
        logger.info(f"=== Image Generation Handler START ===")
        
        provider_config = self.config.get_provider(provider_id)
        
        if provider_config.api_key_required:
            api_key = request_data.get('api_key') or request.headers.get('Authorization', '').replace('Bearer ', '')
            if not api_key:
                raise HTTPException(status_code=401, detail="API key required")
        else:
            api_key = None
        
        provider_type = getattr(provider_config, 'type', None)
        if provider_type == 'coderai':
            # coderai image generation goes through the same generic proxy path so that
            # broker-tunnelled file URLs are rewritten correctly.
            return await self.handle_generic_proxy(request, provider_id, "v1/images/generations", request_data)

        handler = get_provider_handler(provider_id, api_key, user_id=self.user_id)

        _raise_if_provider_unavailable(handler, provider_id)

        try:
            await handler.apply_rate_limit()
            result = await handler.handle_image_generation(request_data)

            # Rewrite URLs in the response to point to our proxy
            result = self._rewrite_content_urls(result, request)

            handler.record_success()
            try:
                usage = result.get('usage', {}) if isinstance(result, dict) else {}
                self._settle_market_result(provider_id, usage, 1, {'kind': 'image_generation'}, request_data=request_data)
            except Exception as market_error:
                logger.warning(f"Market settlement failed for image generation {provider_id}: {market_error}")
            return result
        except Exception as e:
            if _should_record_failure(e):
                handler.record_failure()
            raise HTTPException(status_code=500, detail=str(e))
    
    async def handle_embeddings(self, request: Request, provider_id: str, request_data: Dict) -> Dict:
        """Handle embeddings requests"""
        import logging
        logger = logging.getLogger(__name__)
        logger.info(f"=== Embeddings Handler START ===")
        
        provider_config = self.config.get_provider(provider_id)
        
        if provider_config.api_key_required:
            api_key = request_data.get('api_key') or request.headers.get('Authorization', '').replace('Bearer ', '')
            if not api_key:
                raise HTTPException(status_code=401, detail="API key required")
        else:
            api_key = None
        
        handler = get_provider_handler(provider_id, api_key, user_id=self.user_id)
        
        _raise_if_provider_unavailable(handler, provider_id)
        
        try:
            await handler.apply_rate_limit()
            result = await handler.handle_embeddings(request_data)
            handler.record_success()
            try:
                usage = result.get('usage', {}) if isinstance(result, dict) else {}
                self._settle_market_result(provider_id, usage, 1, {'kind': 'embeddings'}, request_data=request_data)
            except Exception as market_error:
                logger.warning(f"Market settlement failed for embeddings {provider_id}: {market_error}")
            return result
        except Exception as e:
            if _should_record_failure(e):
                handler.record_failure()
            raise HTTPException(status_code=500, detail=str(e))
    
    def _rewrite_coderai_file_urls(self, body_str: str, provider_id: str, request: Request, *, via_broker: bool) -> str:
        """Rewrite /v1/files/ URLs in a JSON body string to go through our proxy endpoint.

        For broker-connected providers the proxy entry records (provider_id, path) so that
        handle_content_proxy can fetch the file back through the WSS tunnel.  For direct
        providers the proxy entry records the original URL for a plain HTTP fetch.
        """
        try:
            data = json.loads(body_str)
        except Exception:
            return body_str

        scheme = request.url.scheme
        host = request.headers.get('host', request.url.netloc)
        root_path = (request.scope.get('root_path') or '').rstrip('/')
        base_url = f"{scheme}://{host}{root_path}"

        from urllib.parse import urlparse as _urlparse

        def rewrite_url(url: str) -> str:
            if '/v1/files/' not in url:
                return url
            parsed = _urlparse(url)
            file_path = parsed.path  # e.g. /v1/files/image_abc.png
            key = f"{provider_id}:{file_path}" if via_broker else url
            content_id = hashlib.md5(key.encode()).hexdigest()[:16]
            entry = (
                {"type": "broker", "provider_id": provider_id, "path": file_path}
                if via_broker
                else {"type": "http", "url": url}
            )
            with _proxy_file_registry_lock:
                _proxy_file_registry[content_id] = entry
            return f"{base_url}/api/proxy/{content_id}"

        def _walk(obj):
            if isinstance(obj, dict):
                for key in list(obj.keys()):
                    v = obj[key]
                    if key in ('url', 'image_url', 'video_url', 'audio_url') and isinstance(v, str):
                        obj[key] = rewrite_url(v)
                    else:
                        _walk(v)
            elif isinstance(obj, list):
                for i, item in enumerate(obj):
                    if isinstance(item, str) and '/v1/files/' in item:
                        obj[i] = rewrite_url(item)
                    else:
                        _walk(item)

        _walk(data)
        return json.dumps(data)

    def _rewrite_content_urls(self, response: Dict, request: Request) -> Dict:
        """Rewrite content URLs to point to our proxy endpoint"""
        import logging
        import hashlib
        import json
        logger = logging.getLogger(__name__)
        
        # Get the base URL from the request
        scheme = request.url.scheme
        host = request.headers.get('host', request.url.netloc)
        base_url = f"{scheme}://{host}"
        
        # Store URL mappings in a simple in-memory cache (in production, use Redis or similar)
        if not hasattr(self, '_url_cache'):
            self._url_cache = {}
        
        def rewrite_url(original_url: str) -> str:
            """Rewrite a single URL"""
            # Check if URL is already public and accessible
            if self._is_public_url(original_url):
                logger.info(f"URL is public, passing through: {original_url}")
                return original_url
            
            # Generate a unique ID for this URL
            url_hash = hashlib.md5(original_url.encode()).hexdigest()[:16]
            
            # Store the mapping
            self._url_cache[url_hash] = original_url
            
            # Return the proxy URL
            proxy_url = f"{base_url}/api/proxy/{url_hash}"
            logger.info(f"Rewrote URL: {original_url} -> {proxy_url}")
            return proxy_url
        
        # Recursively rewrite URLs in the response
        def rewrite_recursive(obj):
            if isinstance(obj, dict):
                for key, value in obj.items():
                    if key in ['url', 'image_url', 'audio_url', 'video_url'] and isinstance(value, str):
                        obj[key] = rewrite_url(value)
                    else:
                        obj[key] = rewrite_recursive(value)
            elif isinstance(obj, list):
                return [rewrite_recursive(item) for item in obj]
            return obj
        
        return rewrite_recursive(response)
    
    def _is_public_url(self, url: str) -> bool:
        """Check if a URL is publicly accessible (doesn't need proxying)"""
        # URLs from major CDNs and public services don't need proxying
        public_domains = [
            'cloudflare.com',
            'amazonaws.com',
            'googleusercontent.com',
            'azure.com',
            'cdn.',
            'storage.googleapis.com'
        ]
        
        return any(domain in url.lower() for domain in public_domains)
    
    async def handle_content_proxy(self, content_id: str) -> Response:
        """Proxy content — fetches via broker WSS tunnel for broker coderai providers,
        or via plain HTTP for everything else."""
        import logging
        import httpx
        logger = logging.getLogger(__name__)

        with _proxy_file_registry_lock:
            entry = _proxy_file_registry.get(content_id)
        # Legacy per-instance cache fallback (old _rewrite_content_urls path)
        if entry is None:
            original_url = getattr(self, '_url_cache', {}).get(content_id)
            if original_url:
                entry = {"type": "http", "url": original_url}

        if entry is None:
            raise HTTPException(status_code=404, detail="Content not found")

        if entry.get("type") == "broker":
            provider_id = entry["provider_id"]
            file_path = entry["path"]
            logger.info(f"Proxy content via broker tunnel: provider={provider_id} path={file_path}")
            try:
                provider_handler = get_provider_handler(provider_id, None, user_id=self.user_id)
                content_bytes, content_type = await provider_handler.fetch_file(file_path)
                return Response(
                    content=content_bytes,
                    media_type=content_type,
                    headers={"Cache-Control": "public, max-age=3600"},
                )
            except Exception as e:
                logger.error(f"Error fetching file via broker {provider_id} {file_path}: {e}")
                raise HTTPException(status_code=500, detail=f"Error fetching file: {e}")

        original_url = entry.get("url", "")
        logger.info(f"Proxy content via HTTP: {content_id} -> {original_url}")
        try:
            async with httpx.AsyncClient(timeout=120) as client:
                response = await client.get(original_url, follow_redirects=True)
                response.raise_for_status()
                content_type = response.headers.get('content-type', 'application/octet-stream')
                return Response(
                    content=response.content,
                    media_type=content_type,
                    headers={
                        'Content-Disposition': response.headers.get('content-disposition', ''),
                        'Cache-Control': 'public, max-age=3600',
                    },
                )
        except Exception as e:
            logger.error(f"Error proxying content {original_url}: {e}")
            raise HTTPException(status_code=500, detail=f"Error fetching content: {e}")

# Round-robin state for even load distribution across equal-priority models
_rotation_round_robin: Dict[str, int] = {}
_rotation_round_robin_lock = threading.Lock()


async def run_meta_target(target_id, messages, max_tokens=None, temperature=0.0, stop=None):
    """Run a GLOBAL rotation or autoselect as an override for an internal/local model.

    Used by the internal-model overrides (compaction, classification, autoselect
    selection). Resolves ``target_id`` against the global rotations/autoselects,
    forwards a non-streaming chat request, and returns the assistant message
    content string — or ``None`` on any failure so callers can fall back to their
    local model.
    """
    import logging
    logger = logging.getLogger(__name__)
    target_id = (target_id or '').strip()
    if not target_id:
        return None
    is_rotation = bool(getattr(config, 'rotations', None)) and target_id in config.rotations
    is_autoselect = bool(getattr(config, 'autoselect', None)) and target_id in config.autoselect
    if not is_rotation and not is_autoselect:
        logger.warning(f"Internal-model override target '{target_id}' is neither a rotation nor an autoselect; using local model")
        return None
    request_data = {
        'model': target_id,
        'messages': messages,
        'stream': False,
        'temperature': temperature,
    }
    if max_tokens:
        request_data['max_tokens'] = max_tokens
    if stop:
        request_data['stop'] = stop
    try:
        if is_rotation:
            result = await RotationHandler().handle_rotation_request(target_id, request_data)
        else:
            result = await AutoselectHandler().handle_autoselect_request(target_id, request_data)
    except Exception as e:
        logger.warning(f"Internal-model override '{target_id}' failed: {e}")
        return None
    if not isinstance(result, dict) or result.get('aisbf_error'):
        logger.warning(f"Internal-model override '{target_id}' returned no usable response")
        return None
    try:
        return result['choices'][0]['message']['content']
    except Exception:
        return None


async def classify_with_override(target_id, text, kind):
    """Classify ``text`` with a chat rotation/autoselect override.

    ``kind`` is 'nsfw' or 'privacy'. Returns True if the content is FLAGGED
    (unsafe), False if safe, or None when the override is unavailable/unparseable
    so the caller can fall back to the local classifier pipeline.
    """
    target_id = (target_id or '').strip()
    if not target_id:
        return None
    text_to_check = text[:512] if len(text) > 512 else text
    if kind == 'nsfw':
        question = "Does the following text contain NSFW content (sexual, explicit, or adult material)?"
    else:
        question = ("Does the following text contain privacy-sensitive personal information (PII) such as "
                    "real names, addresses, emails, phone numbers, or financial/identity data?")
    messages = [
        {"role": "system", "content": "You are a strict binary content classifier. Reply with exactly one word: YES or NO. Do not explain."},
        {"role": "user", "content": f"{question}\n\nTEXT:\n{text_to_check}\n\nAnswer YES or NO."},
    ]
    content = await run_meta_target(target_id, messages, max_tokens=5, temperature=0.0)
    if not content:
        return None
    ans = content.strip().upper()
    if ans.startswith('YES'):
        return True
    if ans.startswith('NO'):
        return False
    has_yes, has_no = ('YES' in ans), ('NO' in ans)
    if has_yes and not has_no:
        return True
    if has_no and not has_yes:
        return False
    return None  # unparseable → fall back to local classifier


async def classify_content_with_overrides(text, check_nsfw, check_privacy, aisbf_config):
    """Run NSFW/privacy checks, preferring per-functionality rotation/autoselect
    overrides and falling back to the local classifier pipelines. Returns
    (is_safe, message), matching ContentClassifier.check_content."""
    internal_model = (aisbf_config.internal_model or {}) if aisbf_config else {}
    nsfw_override = internal_model.get('nsfw_classifier_override')
    privacy_override = internal_model.get('privacy_classifier_override')

    # Ensure local pipelines are available as a fallback for enabled checks.
    if (check_nsfw or check_privacy):
        content_classifier.initialize(internal_model.get('nsfw_classifier'), internal_model.get('privacy_classifier'))

    if check_nsfw:
        flagged = await classify_with_override(nsfw_override, text, 'nsfw') if nsfw_override else None
        if flagged is None:
            is_safe, message = content_classifier.check_nsfw(text)
            if not is_safe:
                return False, f"NSFW: {message}"
        elif flagged:
            return False, "NSFW: flagged by override classifier"

    if check_privacy:
        flagged = await classify_with_override(privacy_override, text, 'privacy') if privacy_override else None
        if flagged is None:
            is_safe, message = content_classifier.check_privacy(text)
            if not is_safe:
                return False, f"Privacy: {message}"
        elif flagged:
            return False, "Privacy: flagged by override classifier"

    return True, "All content is safe"


class RotationHandler:
    def __init__(self, user_id=None):
        self.user_id = user_id
        self.config = config
        # Load user-specific configs if user_id is provided
        if user_id:
            self._load_user_configs()
            # Override config to only use user-specific configs with NO global fallback
            self.rotations = {}
            for rotation in self.user_rotations:
                self.rotations[rotation['rotation_id']] = rotation['config']
        else:
            self.user_providers = {}
            self.user_rotations = {}
            self.user_autoselects = {}
            self.rotations = self.config.rotations if hasattr(self.config, 'rotations') else {}

    def _load_user_configs(self):
        """Load user-specific configurations from database"""
        from .database import DatabaseRegistry
        db = DatabaseRegistry.get_config_database()
        self.user_providers = db.get_user_providers(self.user_id)
        self.user_rotations = db.get_user_rotations(self.user_id)
        self.user_autoselects = db.get_user_autoselects(self.user_id)

    def _get_provider_config(self, provider_id: str):
        """Resolve a provider config, preferring user-specific providers.

        In RotationHandler, user_providers is a list of {'provider_id', 'config'}
        dicts (unlike RequestHandler where it is a dict), so handle both forms.
        """
        up = getattr(self, 'user_providers', None)
        if self.user_id and up:
            if isinstance(up, dict):
                if provider_id in up:
                    return up[provider_id]
            else:
                for p in up:
                    if p.get('provider_id') == provider_id:
                        return p.get('config')
        return self.config.get_provider(provider_id, warn=False)

    def reload_user_configs(self):
        """Reload user-specific configurations from database"""
        if self.user_id:
            self._load_user_configs()
            # Refresh rotations dict after reload
            self.rotations = {}
            for rotation in self.user_rotations:
                self.rotations[rotation['rotation_id']] = rotation['config']

    def _record_dashboard_proxy_event(self, request, provider_id: str, model_name: str, success: bool, latency_ms: float, metadata: Dict | None = None):
        try:
            from aisbf.database import DatabaseRegistry
            db = DatabaseRegistry.get_config_database()
            status_code = (metadata or {}).get('status_code')
            if status_code is None:
                status_code = 200 if success else 500
            # Rotation/MCP code paths invoke this without an HTTP request object;
            # fall back to handler-level context when request is unavailable.
            path = request.url.path if request is not None else '/rotation'
            user_id = getattr(request.state, 'user_id', None) if request is not None else getattr(self, 'user_id', None)
            username = request.session.get('username') if (request is not None and hasattr(request, 'session')) else None
            method = request.method if request is not None else 'POST'
            db.record_dashboard_event(
                event_type='request_proxied',
                path=path,
                user_id=user_id,
                username=username,
                method=method,
                status_code=status_code,
                provider_id=provider_id,
                rotation_id=(metadata or {}).get('rotation_id'),
                autoselect_id=(metadata or {}).get('autoselect_id'),
                metadata={
                    'model_name': model_name,
                    'latency_ms': round(float(latency_ms or 0), 2),
                    'stream': bool((metadata or {}).get('stream')),
                    **(metadata or {}),
                },
            )
        except Exception:
            logger.debug("Failed to record proxied request dashboard event", exc_info=True)

    def _get_market_source_details(self, provider_id: str):
        provider_config = None
        if self.user_id and provider_id in getattr(self, 'user_providers', {}):
            provider_config = self.user_providers.get(provider_id)
        elif provider_id in getattr(self.config, 'providers', {}):
            cfg = self.config.get_provider(provider_id)
            provider_config = cfg.model_dump() if hasattr(cfg, 'model_dump') else cfg
        if isinstance(provider_config, dict):
            return provider_config.get('market_source')
        return None

    @staticmethod
    def _market_request_id(request_data: Optional[Dict], fallback_provider_id: str) -> str:
        payload = {
            'provider_id': fallback_provider_id,
            'model': (request_data or {}).get('model'),
            'messages': (request_data or {}).get('messages'),
            'input': (request_data or {}).get('input'),
            'prompt': (request_data or {}).get('prompt'),
            'voice': (request_data or {}).get('voice'),
            'endpoint_path': (request_data or {}).get('_market_endpoint_path'),
        }
        serialized = json.dumps(payload, sort_keys=True, default=str)
        return hashlib.sha256(serialized.encode('utf-8')).hexdigest()

    def _settle_market_result(self, provider_id: str, usage: Optional[Dict], requests_count: int = 1, metadata: Optional[Dict] = None, request_data: Optional[Dict] = None):
        market_source = self._get_market_source_details(provider_id)
        if not self.user_id or not market_source:
            return None
        listing_id = market_source.get('listing_id')
        if not listing_id:
            return None
        db = DatabaseRegistry.get_config_database()
        settlement_metadata = {
            'provider_id': provider_id,
            'market_imported': True,
            'market_request_id': self._market_request_id(request_data, provider_id),
        }
        if metadata:
            settlement_metadata.update(metadata)
        return db.settle_market_usage(
            consumer_user_id=self.user_id,
            listing_id=int(listing_id),
            prompt_tokens=(usage or {}).get('prompt_tokens', 0),
            completion_tokens=(usage or {}).get('completion_tokens', 0),
            requests_count=requests_count,
            metadata=settlement_metadata,
        )

    def _extract_usage_from_sse_chunk(self, chunk_payload) -> Optional[Dict[str, int]]:
        try:
            if isinstance(chunk_payload, bytes):
                chunk_payload = chunk_payload.decode('utf-8', errors='ignore')
            if isinstance(chunk_payload, str):
                for line in chunk_payload.splitlines():
                    line = line.strip()
                    if not line.startswith('data: '):
                        continue
                    raw = line[6:].strip()
                    if not raw or raw == '[DONE]':
                        continue
                    try:
                        parsed = json.loads(raw)
                    except Exception:
                        continue
                    if isinstance(parsed, dict) and isinstance(parsed.get('usage'), dict):
                        usage = parsed.get('usage') or {}
                        if usage.get('total_tokens') is not None or usage.get('prompt_tokens') is not None or usage.get('completion_tokens') is not None:
                            return usage
            elif isinstance(chunk_payload, dict):
                usage = chunk_payload.get('usage')
                if isinstance(usage, dict):
                    return usage
        except Exception:
            return None
        return None

    def _get_provider_type(self, provider_id: str) -> str:
        """Get the provider type from configuration"""
        provider_config = self.config.get_provider(provider_id)
        if provider_config:
            return provider_config.type
        return None
    
    def _apply_defaults_to_model(self, model: Dict, provider_config, rotation_config) -> Dict:
        """
        Apply default settings to a model configuration.
        
        Priority order:
        1. Model-specific settings (highest priority)
        2. Rotation default settings
        3. Provider default settings
        4. Auto-derived from first model in provider (lowest priority)
        
        Args:
            model: The model configuration dict
            provider_config: The provider configuration
            rotation_config: The rotation configuration
            
        Returns:
            Model dict with defaults applied
        """
        # List of fields that can have defaults
        default_fields = [
            'rate_limit',
            'max_request_tokens',
            'rate_limit_TPM',
            'rate_limit_TPH',
            'rate_limit_TPD',
            'context_size',
            'condense_context',
            'condense_method'
        ]
        
        for field in default_fields:
            # If field is not set in model, try rotation defaults, then provider defaults
            if field not in model or model[field] is None:
                # Try rotation defaults first
                rotation_default = getattr(rotation_config, f'default_{field}', None)
                if rotation_default is not None:
                    model[field] = rotation_default
                else:
                    # Try provider defaults
                    provider_default = getattr(provider_config, f'default_{field}', None)
                    if provider_default is not None:
                        model[field] = provider_default
                    else:
                        # Auto-derive from first model in provider config if available
                        if provider_config and getattr(provider_config, "models", []) and len(getattr(provider_config, "models", [])) > 0:
                            first_model = getattr(provider_config, "models", [])[0]
                            # For context_size, check multiple field names (from dynamic fetch)
                            if field == 'context_size':
                                model_field = getattr(first_model, 'context_size', None)
                                if model_field is None:
                                    model_field = getattr(first_model, 'context_window', None)
                                if model_field is None:
                                    model_field = getattr(first_model, 'context_length', None)
                            else:
                                model_field = getattr(first_model, field, None)
                            if model_field is not None:
                                model[field] = model_field
        
        return model

    def _apply_defaults_to_autoselect_model(self, model_config: Dict, autoselect_config) -> Dict:
        """
        Apply default settings to an autoselect model configuration.
        
        Priority order:
        1. Model-specific settings (highest priority)
        2. Autoselect default settings
        3. Auto-derived from first model in rotation (lowest priority)
        
        Args:
            model_config: The model configuration dict (typically a rotation_id from autoselect)
            autoselect_config: The autoselect configuration
            
        Returns:
            Model dict with defaults applied
        """
        import logging
        logger = logging.getLogger(__name__)
        
        # List of fields that can have defaults
        default_fields = [
            'rate_limit',
            'max_request_tokens',
            'rate_limit_TPM',
            'rate_limit_TPH',
            'rate_limit_TPD',
            'context_size',
            'condense_context',
            'condense_method'
        ]
        
        # First, check if the model_config is a rotation ID and get its settings
        model_id = model_config.get('model_id') or model_config.get('name') or model_config.get('id', '')
        
        # Try to get defaults from the referenced rotation (first model in the rotation)
        if self.user_id and model_id in self.rotations:
            rotation_config = self.rotations[model_id]
        elif model_id in self.config.rotations:
            rotation_config = self.config.rotations[model_id]
            
            # Check each default field
            for field in default_fields:
                # If field is not set in model, try autoselect defaults, then rotation defaults
                if field not in model_config or model_config.get(field) is None:
                    # Try autoselect defaults first
                    autoselect_default = getattr(autoselect_config, f'default_{field}', None)
                    if autoselect_default is not None:
                        model_config[field] = autoselect_default
                        logger.debug(f"Applied autoselect default_{field}: {autoselect_default} to model {model_id}")
                    else:
                        # Try rotation defaults
                        rotation_default = getattr(rotation_config, f'default_{field}', None)
                        if rotation_default is not None:
                            model_config[field] = rotation_default
                            logger.debug(f"Applied rotation default_{field}: {rotation_default} to model {model_id}")
                        else:
                            # Auto-derive from first provider in rotation, then first model
                            if rotation_config.providers and len(rotation_config.providers) > 0:
                                first_provider = rotation_config.providers[0]
                                provider_id = first_provider.get('provider_id')
                                provider_config = self.config.get_provider(provider_id)
                                
                                if provider_config and getattr(provider_config, "models", []) and len(getattr(provider_config, "models", [])) > 0:
                                    first_model = getattr(provider_config, "models", [])[0]
                                    # Check for context_size, context_window, or context_length
                                    if field == 'context_size':
                                        model_field = getattr(first_model, 'context_size', None)
                                        if model_field is None:
                                            model_field = getattr(first_model, 'context_window', None)
                                        if model_field is None:
                                            model_field = getattr(first_model, 'context_length', None)
                                    else:
                                        model_field = getattr(first_model, field, None)
                                    if model_field is not None:
                                        model_config[field] = model_field
                                        logger.debug(f"Auto-derived default_{field}: {model_field} from first model in {provider_id}")
        else:
            # Not a rotation, apply autoselect defaults directly
            for field in default_fields:
                if field not in model_config or model_config.get(field) is None:
                    autoselect_default = getattr(autoselect_config, f'default_{field}', None)
                    if autoselect_default is not None:
                        model_config[field] = autoselect_default
        
        return model_config

    async def _handle_chunked_rotation_request(
        self,
        handler,
        model_name: str,
        messages: List[Dict],
        max_tokens: Optional[int],
        temperature: float,
        stream: bool,
        tools: Optional[List[Dict]],
        tool_choice: Optional[Union[str, Dict]],
        max_request_tokens: int,
        provider_id: str,
        logger
    ) -> Dict:
        """
        Handle a rotation request that needs to be split into multiple chunks due to token limits.
        
        This method splits the request into chunks, sends each chunk sequentially,
        and combines the responses into a single response.
        
        Args:
            handler: The provider handler
            model_name: The model name
            messages: The messages to send
            max_tokens: Max output tokens
            temperature: Temperature setting
            stream: Whether to stream (not supported for chunked requests)
            tools: Tool definitions
            tool_choice: Tool choice setting
            max_request_tokens: Maximum tokens per request
            provider_id: Provider identifier
            logger: Logger instance
        
        Returns:
            Combined response from all chunks
        """
        logger.info(f"=== ROTATION CHUNKED REQUEST HANDLING START ===")
        logger.info(f"Max request tokens per chunk: {max_request_tokens}")
        
        # Split messages into chunks
        message_chunks = split_messages_into_chunks(messages, max_request_tokens, model_name)
        logger.info(f"Split into {len(message_chunks)} message chunks")
        
        if stream:
            logger.warning("Streaming is not supported for chunked rotation requests, falling back to non-streaming")
        
        # Process each chunk and collect responses
        all_responses = []
        combined_content = ""
        combined_tool_calls = []
        total_prompt_tokens = 0
        total_completion_tokens = 0
        created_time = int(time_module.time())
        response_id = f"chunked-rotation-{provider_id}-{model_name}-{created_time}"
        
        for chunk_idx, chunk_messages in enumerate(message_chunks):
            logger.info(f"Processing chunk {chunk_idx + 1}/{len(message_chunks)}")
            logger.info(f"Chunk messages count: {len(chunk_messages)}")
            
            # Apply rate limiting between chunks
            if chunk_idx > 0:
                await handler.apply_rate_limit()
            
            try:
                chunk_response = await handler.handle_request(
                    model=model_name,
                    messages=chunk_messages,
                    max_tokens=max_tokens,
                    temperature=temperature,
                    stream=False,  # Always non-streaming for chunked requests
                    tools=tools if chunk_idx == 0 else None,  # Only first chunk uses tools
                    tool_choice=tool_choice if chunk_idx == 0 else None
                )
                
                # Extract content from response
                if isinstance(chunk_response, dict):
                    choices = chunk_response.get('choices', [])
                    if choices:
                        chunk_message = choices[0].get('message') or {}
                        # Tool-call responses carry content=None (the key exists, so a
                        # .get default won't kick in) – coalesce to '' or the += raises
                        # TypeError, which records a provider failure and can disable
                        # the provider even though upstream returned 200.
                        combined_content += chunk_message.get('content') or ''
                        chunk_tool_calls = chunk_message.get('tool_calls')
                        if chunk_tool_calls:
                            combined_tool_calls.extend(chunk_tool_calls)

                        # Accumulate token usage
                        usage = chunk_response.get('usage', {})
                        chunk_total_tokens = usage.get('total_tokens', 0)
                        total_prompt_tokens += usage.get('prompt_tokens', 0)
                        total_completion_tokens += usage.get('completion_tokens', 0)
                        
                        # Record token usage for rate limit tracking
                        if chunk_total_tokens > 0:
                            handler._record_token_usage(model_name, chunk_total_tokens)
                            logger.info(f"Recorded {chunk_total_tokens} tokens for chunk {chunk_idx + 1}")
                
                all_responses.append(chunk_response)
                logger.info(f"Chunk {chunk_idx + 1} processed successfully")
                
            except Exception as e:
                logger.error(f"Error processing chunk {chunk_idx + 1}: {e}")
                # If a chunk fails, we still try to return what we have
                if all_responses:
                    logger.warning("Returning partial results from successful chunks")
                    break
                else:
                    raise e
        
        # Build combined response.  Tool calls from any chunk must be passed through
        # (chunk 0 is the one sent with tools) or agent clients would receive an
        # empty completion and stall.
        combined_message = {"role": "assistant", "content": combined_content}
        finish_reason = "stop"
        if combined_tool_calls:
            combined_message["tool_calls"] = combined_tool_calls
            if not combined_content:
                combined_message["content"] = None
            finish_reason = "tool_calls"
        combined_response = {
            "id": response_id,
            "object": "chat.completion",
            "created": created_time,
            "model": model_name,
            "choices": [{
                "index": 0,
                "message": combined_message,
                "finish_reason": finish_reason
            }],
            "usage": {
                "prompt_tokens": total_prompt_tokens,
                "completion_tokens": total_completion_tokens,
                "total_tokens": total_prompt_tokens + total_completion_tokens
            },
            "aisbf_chunked": True,
            "aisbf_total_chunks": len(message_chunks)
        }

        logger.info(f"=== ROTATION CHUNKED REQUEST HANDLING END ===")
        logger.info(f"Combined content length: {len(combined_content)} characters")
        logger.info(f"Total chunks processed: {len(all_responses)}")
        
        return combined_response

    def _get_api_key(self, provider_id: str, rotation_api_key: Optional[str] = None) -> Optional[str]:
        """
        Get the API key for a provider.
        
        Priority order:
        1. API key from provider config (providers.json)
        2. API key from rotation config (rotations.json)
        
        Args:
            provider_id: The provider identifier
            rotation_api_key: Optional API key from rotation configuration
            
        Returns:
            The API key to use, or None if not found
        """
        import logging
        logger = logging.getLogger(__name__)
        
        # First check provider config for api_key
        provider_config = self.config.get_provider(provider_id)
        if provider_config and hasattr(provider_config, 'api_key') and provider_config.api_key:
            logger.info(f"Using API key from provider config for {provider_id}")
            return provider_config.api_key
        
        # Fall back to rotation api_key
        if rotation_api_key:
            logger.info(f"Using API key from rotation config for {provider_id}")
            return rotation_api_key
        
        logger.info(f"No API key found for {provider_id}")
        return None

    async def handle_rotation_request(self, rotation_id: str, request_data: Dict, user_id: Optional[int] = None, token_id: Optional[int] = None):
        """
        Handle a rotation request.

        For streaming requests, returns a StreamingResponse with proper handling
        based on the selected provider's type (google vs others).
        For non-streaming requests, returns the response dict directly.
        """
        import logging
        import time
        logger = logging.getLogger(__name__)
        # Track request start time for latency calculation
        request_start_time = time.time()
        logger.info(f"=== RotationHandler.handle_rotation_request START ===")
        logger.info(f"Rotation ID: {rotation_id}")
        logger.info(f"User ID: {self.user_id}")

        # Check for user-specific rotation config first
        if self.user_id:
            # Database user: ONLY use user-specific configs - NO global fallback
            if rotation_id not in self.rotations:
                logger.error(f"User rotation {rotation_id} not found - NO global fallback")
                raise HTTPException(status_code=400, detail=f"Rotation {rotation_id} not found for this user")
            rotation_config = self.rotations[rotation_id]
            logger.info(f"Using user-specific rotation config for {rotation_id}")
        else:
            # Admin user: use global config
            rotation_config = self.config.get_rotation(rotation_id)
            logger.info(f"Using global rotation config for {rotation_id}")

        # Check response cache for non-streaming requests
        stream = request_data.get('stream', False)
        if not stream:
            try:
                aisbf_config = self.config.get_aisbf_config()
                if aisbf_config and aisbf_config.response_cache and aisbf_config.response_cache.enabled:
                    response_cache = get_response_cache(aisbf_config.response_cache.model_dump())
                    cached_response = response_cache.get(request_data, user_id=self.user_id)
                    if cached_response:
                        logger.info(f"Cache hit for rotation request {rotation_id}")
                        return cached_response
                    else:
                        logger.debug(f"Cache miss for rotation request {rotation_id}")
            except Exception as cache_error:
                logger.warning(f"Response cache check failed: {cache_error}")

        if not rotation_config:
            logger.error(f"Rotation {rotation_id} not found")
            raise HTTPException(status_code=400, detail=f"Rotation {rotation_id} not found")
        
        # Check if notifyerrors is enabled for this rotation
        notify_errors = getattr(rotation_config, 'notifyerrors', False)
        logger.info(f"notifyerrors setting for rotation '{rotation_id}': {notify_errors}")

        # Check if failover notifications are enabled for this rotation (banner/metadata/headers
        # emitted when a failover changes the served provider/model vs. the preferred one)
        notify_on_failover = getattr(rotation_config, 'notify_on_failover', False)
        logger.info(f"notify_on_failover setting for rotation '{rotation_id}': {notify_on_failover}")
        
        # Extract stream setting early - needed for error handling
        stream = request_data.get('stream', False)
        logger.info(f"Request stream mode: {stream}")

        # Delegation loop protection: a rotation entry may forward to another
        # rotation/autoselect. Track the chain of meta-targets already traversed in
        # this request so nothing can recurse into itself (A -> B -> A) or nest too
        # deeply. The chain rides on request_data and is copied into child requests.
        MAX_DELEGATION_DEPTH = 8
        delegation_chain = list(request_data.get('_delegation_chain') or [])
        self_delegation_key = f"rotation:{rotation_id}"
        if self_delegation_key in delegation_chain:
            logger.error(f"Delegation loop detected: '{self_delegation_key}' already in chain {delegation_chain}")
            raise HTTPException(status_code=508, detail=f"Rotation delegation loop detected for '{rotation_id}' (chain: {' -> '.join(delegation_chain)})")
        delegation_chain = delegation_chain + [self_delegation_key]
        request_data['_delegation_chain'] = delegation_chain
        logger.info(f"Delegation chain: {delegation_chain}")

        logger.info(f"Rotation config loaded successfully")
        providers = rotation_config.providers
        logger.info(f"Number of providers in rotation: {len(providers)}")
        
        # Collect all available models with their weights
        available_models = []
        skipped_providers = []
        total_models_considered = 0

        logger.info(f"=== MODEL SELECTION PROCESS START ===")
        logger.info(f"Scanning providers for available models...")
        
        for provider in providers:
            provider_id = provider['provider_id']
            logger.info(f"")
            logger.info(f"--- Processing provider: {provider_id} ---")

            # Meta-provider: this entry forwards to another rotation/autoselect.
            # Each model 'name' is the id of the target rotation/autoselect.
            delegate_kind = self._delegation_kind(provider_id)
            if delegate_kind:
                provider_weight = provider.get('weight', 1)
                targets = provider.get('models') or []
                if not targets:
                    logger.warning(f"  [SKIPPED] Meta-provider '{provider_id}' lists no target {delegate_kind}(s) in models")
                    skipped_providers.append(provider_id)
                    continue
                for tmodel in targets:
                    tname = (tmodel.get('name') or '').strip()
                    if not tname:
                        continue
                    target_key = f"{delegate_kind}:{tname}"
                    # Pre-filter targets that would loop or exceed the depth limit so
                    # they are never selected (the retry loop guards again at runtime).
                    if target_key in delegation_chain:
                        logger.warning(f"  [SKIPPED] Delegation target '{target_key}' would create a loop (chain: {delegation_chain})")
                        continue
                    if len(delegation_chain) >= MAX_DELEGATION_DEPTH:
                        logger.warning(f"  [SKIPPED] Delegation depth limit ({MAX_DELEGATION_DEPTH}) reached; skipping '{target_key}'")
                        continue
                    target_weight = tmodel.get('weight', provider_weight)
                    available_models.append({
                        'name': tname,
                        'weight': target_weight,
                        'provider_id': provider_id,
                        'api_key': None,
                        '_delegate_kind': delegate_kind,
                        '_delegate_target': tname,
                    })
                    total_models_considered += 1
                    logger.info(f"  [DELEGATION] -> {delegate_kind} '{tname}' (weight {target_weight})")
                continue

            # Check if provider exists in configuration (user-specific first, then global)
            if self.user_id and provider_id in self.user_providers:
                provider_config = self.user_providers[provider_id]
                logger.info(f"  [USER] Using user-specific provider config for {provider_id}")
            else:
                provider_config = self.config.get_provider(provider_id)
                logger.info(f"  [GLOBAL] Using global provider config for {provider_id}")

            if not provider_config:
                logger.error(f"  [ERROR] Provider {provider_id} not found in providers configuration")
                logger.error(f"  Available providers: {list(self.config.providers.keys()) if not self.user_id else list(self.user_providers.keys())}")
                logger.error(f"  Skipping this provider")
                skipped_providers.append(provider_id)
                continue
            
            # Cheap pre-check: skip providers that are already disabled — manually
            # (dashboard toggle) or by an auto-disable cooldown — BEFORE building the
            # handler and validating credentials, which for some provider types
            # performs a network round-trip. This keeps disabled providers from being
            # contacted at all and makes a dashboard enable/disable take effect on the
            # very next request.
            if is_provider_disabled_cheap(provider_id, self.user_id):
                logger.warning(f"  [SKIPPED] Provider {provider_id} is disabled (manual toggle or cooldown)")
                skipped_providers.append(provider_id)
                continue

            # Get API key: first from provider config, then from rotation config
            api_key = self._get_api_key(provider_id, provider.get('api_key'))

            # Check if provider is rate limited/deactivated (authoritative check)
            provider_handler = get_provider_handler(provider_id, api_key, user_id=self.user_id)
            if provider_handler.is_rate_limited():
                logger.warning(f"  [SKIPPED] Provider {provider_id} is rate limited/deactivated")
                logger.warning(f"  Reason: Provider has exceeded failure threshold or is in cooldown period")
                skipped_providers.append(provider_id)
                continue
            
            logger.info(f"  [AVAILABLE] Provider {provider_id} is active and ready")
            
            # Check if provider-level weight is specified
            provider_weight = provider.get('weight', 1)  # Default to 1 if not specified
            if provider.get('weight') is not None:
                logger.info(f"  Provider-level weight: {provider_weight}")
            
            # Check if models are specified in rotation config
            # If not, use models from provider config
            rotation_models = provider.get('models')
            if not rotation_models:
                logger.info(f"  No models specified in rotation config for {provider_id}")
                logger.info(f"  Will use models from provider configuration")
                
                # Get models from provider config
                if getattr(provider_config, "models", []):
                    # Use models from provider config with provider-level weight
                    rotation_models = []
                    for provider_model in getattr(provider_config, "models", []):
                        model_dict = {
                            'name': provider_model.name,
                            'weight': provider_weight,  # Use provider-level weight
                            'rate_limit': provider_model.rate_limit,
                            'max_request_tokens': provider_model.max_request_tokens,
                            'nsfw': getattr(provider_model, 'nsfw', False),
                            'privacy': getattr(provider_model, 'privacy', False)
                        }
                        rotation_models.append(model_dict)
                    logger.info(f"  Loaded {len(rotation_models)} model(s) from provider config with weight {provider_weight}")
                else:
                    logger.warning(f"  No models defined in provider config for {provider_id}")
                    logger.warning(f"  Skipping this provider")
                    skipped_providers.append(provider_id)
                    continue
            
            models_in_provider = len(rotation_models)
            total_models_considered += models_in_provider
            logger.info(f"  Found {models_in_provider} model(s) in this provider")
            
            for model in rotation_models:
                # Apply defaults: model-specific > rotation defaults > provider defaults
                model = self._apply_defaults_to_model(model, provider_config, rotation_config)

                model_name = (model.get('name') or '').strip()
                if not model_name:
                    logger.warning(f"  [SKIPPED] Provider {provider_id}: model entry has empty name, skipping")
                    total_models_considered -= 1
                    continue

                model_weight = model['weight']
                model_rate_limit = model.get('rate_limit', 'N/A')
                
                logger.info(f"    - Model: {model_name}")
                logger.info(f"      Weight (Priority): {model_weight}")
                logger.info(f"      Rate Limit: {model_rate_limit}")
                
                # Add provider_id and api_key to model for later use
                # Use resolved api_key (from provider config or rotation config)
                model_with_provider = model.copy()
                model_with_provider['provider_id'] = provider_id
                model_with_provider['api_key'] = api_key
                available_models.append(model_with_provider)

        logger.info(f"")
        logger.info(f"=== MODEL SELECTION SUMMARY ===")
        logger.info(f"Total providers scanned: {len(providers)}")
        logger.info(f"Providers skipped (rate limited): {len(skipped_providers)}")
        if skipped_providers:
            logger.info(f"Skipped providers: {', '.join(skipped_providers)}")
        logger.info(f"Total models considered: {total_models_considered}")
        logger.info(f"Total models available: {len(available_models)}")
        
        # Apply NSFW/Privacy content classification filtering
        # Only classify the immediate intent (last 3 messages + current query) to avoid
        # huge context issues and long classification times
        aisbf_config = self.config.get_aisbf_config()
        if aisbf_config and (aisbf_config.classify_nsfw or aisbf_config.classify_privacy):
            logger.info(f"=== CONTENT CLASSIFICATION FILTERING ===")
            
            # Get messages for classification - only last 3 user messages + current query
            messages = request_data.get('messages', [])
            
            # Extract last 3 user messages for classification window
            recent_user_messages = []
            for msg in reversed(messages):
                if msg.get('role') == 'user':
                    content = msg.get('content', '')
                    if isinstance(content, str):
                        recent_user_messages.append(content)
                        if len(recent_user_messages) >= 3:
                            break
            
            # Build the classification prompt from recent messages only
            # Reverse to get correct order (oldest first)
            recent_user_messages.reverse()
            user_prompt = " ".join(recent_user_messages)
            
            logger.info(f"Classifying only recent context ({len(recent_user_messages)} messages))")
            logger.info(f"Recent context preview: {user_prompt[:200]}..." if len(user_prompt) > 200 else f"Recent context: {user_prompt}")
            logger.info(f"Classify NSFW: {aisbf_config.classify_nsfw}")
            logger.info(f"Classify privacy: {aisbf_config.classify_privacy}")
            
            # Check if content classification is needed
            check_nsfw = aisbf_config.classify_nsfw
            check_privacy = aisbf_config.classify_privacy
            
            if check_nsfw or check_privacy:
                # Classify, preferring per-functionality rotation/autoselect overrides
                # (internal_model.nsfw_classifier_override / privacy_classifier_override)
                # and falling back to the local classifier pipelines.
                is_safe, message = await classify_content_with_overrides(
                    user_prompt,
                    check_nsfw=check_nsfw,
                    check_privacy=check_privacy,
                    aisbf_config=aisbf_config,
                )

                logger.info(f"Content classification result: {message}")
                
                if not is_safe:
                    # Content is flagged - filter to only nsfw=True or privacy=True models
                    logger.info(f"Content flagged - filtering available models")
                    
                    if check_nsfw and not check_privacy:
                        # Only NSFW filtering needed
                        original_count = len(available_models)
                        available_models = [m for m in available_models if m.get('nsfw', False)]
                        logger.info(f"NSFW filtering: {original_count} -> {len(available_models)} models")
                    elif check_privacy and not check_nsfw:
                        # Only privacy filtering needed
                        original_count = len(available_models)
                        available_models = [m for m in available_models if m.get('privacy', False)]
                        logger.info(f"Privacy filtering: {original_count} -> {len(available_models)} models")
                    elif check_nsfw and check_privacy:
                        # Both filtering - need models with EITHER flag
                        original_count = len(available_models)
                        available_models = [m for m in available_models if m.get('nsfw', False) or m.get('privacy', False)]
                        logger.info(f"NSFW+Privacy filtering: {original_count} -> {len(available_models)} models")
            
            logger.info(f"=== CONTENT CLASSIFICATION FILTERING END ===")
        
        # Check if rotation-level nsfw/privacy flags also apply
        rotation_nsfw = getattr(rotation_config, 'nsfw', False)
        rotation_privacy = getattr(rotation_config, 'privacy', False)
        
        if rotation_nsfw or rotation_privacy:
            logger.info(f"=== ROTATION-LEVEL CONTENT FLAGS ===")
            logger.info(f"Rotation nsfw: {rotation_nsfw}, privacy: {rotation_privacy}")
            
            # If rotation explicitly allows NSFW content, keep models that can handle it
            # If rotation explicitly allows Privacy content, keep models that can handle it
            if rotation_nsfw:
                logger.info(f"Rotation allows NSFW content - keeping models that support it")
            if rotation_privacy:
                logger.info(f"Rotation allows Privacy content - keeping models that support it")
        
        if not available_models:
            logger.error("No models available in rotation (all providers may be rate limited)")
            logger.error("All providers in this rotation are currently deactivated")
            logger.info(f"notifyerrors setting: {notify_errors}")
            
            # Build detailed error message with provider status information
            error_details = []
            error_details.append(f"No models available in rotation '{rotation_id}'. Details:")
            error_details.append("")
            error_details.append(f"**Total providers in rotation:** {len(providers)}")
            error_details.append(f"**Providers skipped (rate limited):** {len(skipped_providers)}")
            
            if skipped_providers:
                error_details.append("")
                error_details.append("**Skipped providers:**")
                for provider_id in skipped_providers:
                    provider_config = self.config.get_provider(provider_id)
                    if provider_config:
                        error_tracking = config.error_tracking.get(provider_id, {})
                        disabled_until = error_tracking.get('disabled_until')
                        failures = error_tracking.get('failures', 0)
                        
                        if disabled_until:
                            import time
                            cooldown_remaining = int(disabled_until - time.time())
                            if cooldown_remaining > 0:
                                error_details.append(f"• {provider_id}: Rate limited (cooldown: {cooldown_remaining}s remaining, failures: {failures})")
                            else:
                                error_details.append(f"• {provider_id}: Rate limited (cooldown expired, failures: {failures})")
                        else:
                            error_details.append(f"• {provider_id}: Rate limited (failures: {failures})")
                    else:
                        error_details.append(f"• {provider_id}: Not configured")
            
            # Check if notifyerrors is enabled - if so, return error as normal message instead of HTTP 503
            logger.info(f"Checking notifyerrors: {notify_errors}")
            if notify_errors:
                logger.info(f"notifyerrors is enabled for rotation '{rotation_id}', returning error as normal message")
                # Return a normal response with error message instead of HTTP 503
                error_message = f"All providers in rotation '{rotation_id}' failed. Details:\n{chr(10).join(error_details[1:])}"
                error_response = {
                    "id": f"error-{rotation_id}-{int(time.time())}",
                    "object": "chat.completion",
                    "created": int(time.time()),
                    "model": rotation_id,
                    "choices": [{
                        "index": 0,
                        "message": {
                            "role": "assistant",
                            "content": error_message
                        },
                        "finish_reason": "stop"
                    }],
                    "usage": {
                        "prompt_tokens": 0,
                        "completion_tokens": len(error_message),
                        "total_tokens": len(error_message)
                    },
                    "aisbf_error": True,
                    "rotation_id": rotation_id,
                    "error_details": error_details
                }
                # For streaming requests, wrap in a simple streaming response
                if stream:
                    return self._create_error_streaming_response(error_response)
                else:
                    return error_response
            else:
                logger.info(f"notifyerrors is disabled for rotation '{rotation_id}', returning error with status code 429")
                # Return a normal response with error message and status code 429
                error_message = f"All providers in rotation '{rotation_id}' failed. Details:\n{chr(10).join(error_details[1:])}"
                error_response = {
                    "id": f"error-{rotation_id}-{int(time.time())}",
                    "object": "chat.completion",
                    "created": int(time.time()),
                    "model": rotation_id,
                    "choices": [{
                        "index": 0,
                        "message": {
                            "role": "assistant",
                            "content": error_message
                        },
                        "finish_reason": "stop"
                    }],
                    "usage": {
                        "prompt_tokens": 0,
                        "completion_tokens": len(error_message),
                        "total_tokens": len(error_message)
                    },
                    "aisbf_error": True,
                    "rotation_id": rotation_id,
                    "error_details": error_details
                }
                # For streaming requests, wrap in a simple streaming response
                if stream:
                    return self._create_error_streaming_response(error_response, status_code=429)
                else:
                    return JSONResponse(status_code=429, content=error_response)

        # Sort models by weight in descending order (higher weight = higher priority)
        available_models.sort(key=lambda m: m['weight'], reverse=True)
        
        logger.info(f"")
        logger.info(f"=== PRIORITY-BASED SELECTION ===")
        logger.info(f"Models sorted by weight (descending priority):")
        for idx, model in enumerate(available_models, 1):
            logger.info(f"  {idx}. {model['name']} (provider: {model['provider_id']}, weight: {model['weight']})")

        # Find the highest weight
        highest_weight = available_models[0]['weight']
        logger.info(f"")
        logger.info(f"Highest priority weight: {highest_weight}")

        # Filter models with the highest weight
        highest_weight_models = [m for m in available_models if m['weight'] == highest_weight]
        other_models = [m for m in available_models if m['weight'] != highest_weight]
        logger.info(f"Models with highest priority ({highest_weight}): {len(highest_weight_models)}")
        for model in highest_weight_models:
            logger.info(f"  - {model['name']} (provider: {model['provider_id']})")

        # Round-robin selection among equal-priority models for even load distribution
        if len(highest_weight_models) > 1:
            rr_key = f"{rotation_id}:{self.user_id or ''}"
            with _rotation_round_robin_lock:
                idx = _rotation_round_robin.get(rr_key, 0) % len(highest_weight_models)
                _rotation_round_robin[rr_key] = idx + 1
            highest_weight_models = highest_weight_models[idx:] + highest_weight_models[:idx]
            logger.info(f"Round-robin selection: slot {idx} of {len(highest_weight_models)} candidates")

        # Reassemble available_models with round-robin order for the top-priority tier
        available_models = highest_weight_models + other_models
        selected_model = available_models[0]

        logger.info(f"")
        logger.info(f"=== FINAL SELECTION ===")
        logger.info(f"Selected model: {selected_model['name']}")
        logger.info(f"Selected provider: {selected_model['provider_id']}")
        logger.info(f"Model weight (priority): {selected_model['weight']}")
        logger.info(f"Model rate limit: {selected_model.get('rate_limit', 'N/A')}")
        logger.info(f"=== MODEL SELECTION PROCESS END ===")

        # Remember the preferred (highest-weight) pick so we can detect, on success,
        # whether a failover ended up serving a different provider/model and notify
        # the client when notify_on_failover is enabled.
        preferred_provider_id = selected_model['provider_id']
        preferred_model_name = selected_model['name']

        # Retry logic: Try up to 5 times, allowing model retries with rate limiting
        max_retries = 5
        tried_models = []  # Track which models have been tried
        model_retry_counts = {}  # Track retry count per model
        last_error = None
        successful_model = None
        successful_handler = None
        successful_response = None
        
        for attempt in range(max_retries):
            logger.info(f"")
            logger.info(f"=== ATTEMPT {attempt + 1}/{max_retries} ===")
            
            # Select a model that hasn't been tried yet, or retry a failed model with rate limiting
            remaining_models = [m for m in available_models if m not in tried_models]
            
            if not remaining_models:
                logger.error(f"No more models available to try")
                logger.error(f"All {len(available_models)} models have been attempted")
                break
            
            # Sort remaining models by weight and select the best one
            remaining_models.sort(key=lambda m: m['weight'], reverse=True)
            current_model = remaining_models[0]
            
            # Check if this model has been retried too many times
            model_key = f"{current_model['provider_id']}:{current_model['name']}"
            retry_count = model_retry_counts.get(model_key, 0)
            
            if retry_count >= 2:  # Max 2 retries per model
                logger.warning(f"Model {current_model['name']} has reached max retry count, skipping")
                tried_models.append(current_model)
                continue
            
            logger.info(f"Trying model: {current_model['name']} (provider: {current_model['provider_id']})")
            logger.info(f"Attempt {attempt + 1} of {max_retries}")
            logger.info(f"Model retry count: {retry_count}")
            
            provider_id = current_model['provider_id']
            api_key = current_model.get('api_key')
            model_name = current_model['name']

            # If this entry delegates to another rotation/autoselect, forward the
            # whole request there (with loop protection) instead of treating it as a
            # real provider. The downstream handler returns a final response object
            # (dict / StreamingResponse) which we pass straight back to the client;
            # a downstream total failure makes us fail over to the next entry.
            if current_model.get('_delegate_kind'):
                delegate_kind = current_model['_delegate_kind']
                target = current_model['_delegate_target']
                target_key = f"{delegate_kind}:{target}"
                try:
                    if target_key in delegation_chain:
                        raise RuntimeError(f"delegation loop: {target_key} already in {delegation_chain}")
                    if len(delegation_chain) >= MAX_DELEGATION_DEPTH:
                        raise RuntimeError(f"delegation depth limit ({MAX_DELEGATION_DEPTH}) reached")
                    logger.info(f"=== DELEGATING rotation '{rotation_id}' -> {delegate_kind} '{target}' (chain: {delegation_chain}) ===")
                    # Child request carries the (already-extended) delegation chain.
                    child_request = dict(request_data)
                    if delegate_kind == 'rotation':
                        result = await RotationHandler(user_id=self.user_id).handle_rotation_request(
                            target, child_request, user_id, token_id
                        )
                    else:
                        autoselect_handler = AutoselectHandler(user_id=self.user_id)
                        if request_data.get('stream', False):
                            result = await autoselect_handler.handle_autoselect_streaming_request(target, child_request)
                        else:
                            result = await autoselect_handler.handle_autoselect_request(target, child_request, user_id, token_id)
                    if self._is_failed_delegation_result(result):
                        raise RuntimeError(f"delegated {target_key} returned a failure result")
                    logger.info(f"=== DELEGATION to {target_key} succeeded ===")
                    return result
                except Exception as delegation_error:
                    last_error = str(delegation_error)
                    logger.error(f"Delegation to {target_key} failed: {delegation_error}")
                    model_retry_counts[model_key] = retry_count + 1
                    tried_models.append(current_model)
                    continue

            logger.info(f"Getting provider handler for {provider_id}")
            handler = get_provider_handler(provider_id, api_key, user_id=self.user_id)
            logger.info(f"Provider handler obtained: {handler.__class__.__name__}")

            if handler.is_rate_limited():
                logger.warning(f"Provider {provider_id} is rate limited, skipping to next model")
                continue
            
            # Check token rate limits for this model
            request_tokens = count_messages_tokens(request_data['messages'], model_name)
            if handler._check_token_rate_limit(model_name, request_tokens):
                logger.warning(f"Model {model_name} would exceed token rate limit, skipping to next model")
                # Determine which limit was exceeded and disable provider accordingly
                model_config = current_model
                if model_config.get('rate_limit_TPM'):
                    handler._disable_provider_for_duration("1m")
                    logger.warning(f"Provider {provider_id} disabled for 1 minute due to TPM limit")
                elif model_config.get('rate_limit_TPH'):
                    handler._disable_provider_for_duration("1h")
                    logger.warning(f"Provider {provider_id} disabled for 1 hour due to TPH limit")
                elif model_config.get('rate_limit_TPD'):
                    handler._disable_provider_for_duration("1d")
                    logger.warning(f"Provider {provider_id} disabled for 1 day due to TPD limit")
                continue
            
            try:
                logger.info(f"Model requested: {model_name}")
                logger.info(f"Messages count: {len(request_data.get('messages', []))}")
                logger.info(f"Max tokens: {request_data.get('max_tokens')}")
                logger.info(f"Temperature: {request_data.get('temperature', 1.0)}")
                logger.info(f"Stream: {request_data.get('stream', False)}")
                
                # Get context configuration
                context_config = get_context_config_for_model(
                    model_name=model_name,
                    provider_config=None,
                    rotation_model_config=current_model,
                    rotation_config=rotation_config,
                    autoselect_config=None,
                )
                logger.info(f"Context config: {context_config}")
                
                # Calculate effective context
                messages = request_data['messages']
                effective_context = count_messages_tokens(messages, model_name)
                logger.info(f"Effective context: {effective_context} tokens")
                
                # Apply context condensation if needed
                if context_config.get('condense_context', 0) > 0:
                    context_manager = ContextManager(context_config, handler, self.config.get_condensation(), self.user_id)
                    if context_manager.should_condense(messages, model_name):
                        logger.info("Context condensation triggered")
                        messages = await context_manager.condense_context(messages, model_name)
                        effective_context = count_messages_tokens(messages, model_name)
                        logger.info(f"Condensed effective context: {effective_context} tokens")
                    # Update request_data with condensed messages
                    request_data['messages'] = messages
                
                # Apply default max output tokens when the client didn't set one.
                # Consult the selected provider's config so a provider-level
                # default_max_tokens (or per-model max_tokens) applies even though
                # rotations cannot configure max_tokens themselves.
                if request_data.get('max_tokens') is None:
                    selected_provider_config = self._get_provider_config(provider_id)
                    default_max_tokens = get_max_completion_tokens_for_model(
                        model_name=model_name,
                        provider_config=selected_provider_config,
                        rotation_model_config=current_model,
                        rotation_config=rotation_config,
                    )
                    # Lowest-priority fallback: autoselect-level default threaded
                    # in by the autoselect handler (below provider/rotation defaults).
                    if not default_max_tokens:
                        default_max_tokens = request_data.get('_autoselect_default_max_tokens')
                    if default_max_tokens:
                        request_data['max_tokens'] = default_max_tokens
                        logger.info(f"Applied default max_tokens: {default_max_tokens} (provider {provider_id})")

                # Check for max_request_tokens in rotation model config
                max_request_tokens = current_model.get('max_request_tokens')
                if max_request_tokens:
                    # Count tokens in the request
                    request_tokens = count_messages_tokens(request_data['messages'], model_name)
                    logger.info(f"Request tokens: {request_tokens}, max_request_tokens: {max_request_tokens}")
                    
                    if request_tokens > max_request_tokens:
                        logger.info(f"Request exceeds max_request_tokens, will split into chunks")
                        
                        # Apply rate limiting
                        logger.info("Applying rate limiting...")
                        await handler.apply_rate_limit()
                        logger.info("Rate limiting applied")
                        
                        # Handle as chunked request
                        response = await self._handle_chunked_rotation_request(
                            handler=handler,
                            model_name=model_name,
                            messages=request_data['messages'],
                            max_tokens=request_data.get('max_tokens'),
                            temperature=request_data.get('temperature', 1.0),
                            stream=request_data.get('stream', False),
                            tools=request_data.get('tools'),
                            tool_choice=request_data.get('tool_choice'),
                            max_request_tokens=max_request_tokens,
                            provider_id=provider_id,
                            logger=logger
                        )

                        # Prime streaming chunked responses so immediate provider
                        # errors trigger failover instead of being streamed to the
                        # client (see _prime_stream / the non-chunked path below).
                        if request_data.get('stream', False):
                            logger.info("Priming chunked streaming response to surface immediate provider errors")
                            response = await self._prime_stream(response)
                            logger.info("Chunked streaming response primed successfully (first chunk received)")

                        handler.record_success()

                        # Build a failover notification if the served provider/model
                        # differs from the preferred pick (opt-in via notify_on_failover).
                        failover_notice = None
                        if notify_on_failover:
                            failover_notice = self._build_failover_notice(
                                rotation_id, preferred_provider_id, preferred_model_name, provider_id, model_name
                            )

                        logger.info(f"=== RotationHandler.handle_rotation_request END ===")
                        logger.info(f"Request succeeded on attempt {attempt + 1}")
                        logger.info(f"Successfully used model: {model_name} (provider: {provider_id})")

                        # Check if response is a streaming request
                        is_streaming = request_data.get('stream', False)
                        if is_streaming:
                            # Get provider type from configuration for proper streaming handling
                            provider_type = self._get_provider_type(provider_id)
                            logger.info(f"Returning streaming response for provider type: {provider_type}")
                            return self._create_streaming_response(
                                response=response,
                                provider_type=provider_type,
                                provider_id=provider_id,
                                model_name=model_name,
                                handler=handler,
                                request_data=request_data,
                                effective_context=effective_context,
                                rotation_id=rotation_id,
                                user_id=user_id,
                                token_id=token_id,
                                request_start_time=request_start_time,
                                failover_info=failover_notice
                            )
                        else:
                            # Cache the clean response for non-streaming chunked requests
                            # (before injecting the failover banner — see non-chunked path).
                            try:
                                aisbf_config = self.config.get_aisbf_config()
                                if aisbf_config and aisbf_config.response_cache and aisbf_config.response_cache.enabled:
                                    response_cache = get_response_cache(aisbf_config.response_cache.model_dump())
                                    response_cache.set(request_data, response, user_id=self.user_id)
                                    logger.debug(f"Cached chunked response for rotation request {rotation_id}")
                            except Exception as cache_error:
                                logger.warning(f"Response cache set failed for chunked request: {cache_error}")

                            self._apply_failover_notice_to_dict(response, failover_notice)
                            logger.info("Returning non-streaming response")
                            return response
                
                # Apply model-specific rate limiting
                rate_limit = current_model.get('rate_limit')
                logger.info(f"Model-specific rate limit: {rate_limit}")
                logger.info("Applying model-level rate limiting...")
                await handler.apply_model_rate_limit(model_name, rate_limit)
                logger.info("Model-level rate limiting applied")

                logger.info(f"Sending request to provider handler...")
                response = await handler.handle_request(
                    model=model_name,
                    messages=request_data['messages'],
                    max_tokens=request_data.get('max_tokens'),
                    temperature=request_data.get('temperature', 1.0),
                    stream=request_data.get('stream', False),
                    tools=request_data.get('tools'),
                    tool_choice=request_data.get('tool_choice')
                )
                logger.info(f"Response received from provider")
                logger.info(f"Response type: {type(response)}")

                # For streaming requests the provider call is lazy: the actual
                # upstream HTTP request is only made when the generator is first
                # iterated. If we returned the StreamingResponse now, an immediate
                # provider error (e.g. HTTP 400/429) would surface *inside* the
                # response generator after the 200 OK headers were already sent to
                # the client — making transparent failover impossible.
                #
                # Prime the stream here (pull the first chunk) so that such errors
                # are raised within this try/except and trigger failover to the
                # next provider by weight, completely transparently to the client.
                if request_data.get('stream', False):
                    logger.info("Priming streaming response to surface immediate provider errors")
                    response = await self._prime_stream(response)
                    logger.info("Streaming response primed successfully (first chunk received)")

                # Record token usage for rate limit tracking
                if isinstance(response, dict):
                    usage = response.get('usage', {})
                    total_tokens = usage.get('total_tokens', 0)
                    if total_tokens > 0:
                        handler._record_token_usage(model_name, total_tokens)
                        logger.info(f"Recorded {total_tokens} tokens for model {model_name}")
                    
                    # Add effective context to response for non-streaming
                    usage['effective_context'] = effective_context
                    logger.info(f"Added effective_context to response: {effective_context}")
                
                handler.record_success()
                
                # Update successful variables to the ones that worked
                successful_model = current_model
                successful_handler = handler
                successful_response = response

                # If a failover changed the served provider/model vs. the preferred
                # one, build a notification for the client (opt-in via notify_on_failover).
                failover_notice = None
                if notify_on_failover:
                    failover_notice = self._build_failover_notice(
                        rotation_id, preferred_provider_id, preferred_model_name, provider_id, model_name
                    )
                    if failover_notice:
                        logger.info(f"Failover notice: preferred {preferred_provider_id}/{preferred_model_name} -> used {provider_id}/{model_name}")

                logger.info(f"=== RotationHandler.handle_rotation_request END ===")
                logger.info(f"Request succeeded on attempt {attempt + 1}")
                logger.info(f"Successfully used model: {successful_model['name']} (provider: {successful_model['provider_id']})")

                # Check if response is a streaming request
                is_streaming = request_data.get('stream', False)
                if is_streaming:
                    # Get provider type from configuration for proper streaming handling
                    provider_type = self._get_provider_type(provider_id)
                    logger.info(f"Returning streaming response for provider type: {provider_type}")
                    return self._create_streaming_response(
                        response=response,
                        provider_type=provider_type,
                        provider_id=provider_id,
                        model_name=model_name,
                        handler=handler,
                        request_data=request_data,
                        effective_context=effective_context,
                        rotation_id=rotation_id,
                        user_id=user_id,
                        token_id=token_id,
                        request_start_time=request_start_time,
                        failover_info=failover_notice
                    )
                else:
                    # Cache the response for non-streaming requests (store the clean
                    # response BEFORE injecting the failover banner, so a later cache
                    # hit doesn't replay a stale "provider switched" notice).
                    try:
                        aisbf_config = self.config.get_aisbf_config()
                        if aisbf_config and aisbf_config.response_cache and aisbf_config.response_cache.enabled:
                            response_cache = get_response_cache(aisbf_config.response_cache.model_dump())
                            response_cache.set(request_data, response, user_id=self.user_id)
                            logger.debug(f"Cached response for rotation request {rotation_id}")
                    except Exception as cache_error:
                        logger.warning(f"Response cache set failed: {cache_error}")

                    # Notify the client of the provider/model change (body metadata + banner)
                    self._apply_failover_notice_to_dict(response, failover_notice)

                    logger.info("Returning non-streaming response")

                    try:
                        self._settle_market_result(provider_id, response.get('usage', {}) if isinstance(response, dict) else {}, 1, {'kind': 'rotation', 'rotation_id': rotation_id}, request_data=request_data)
                    except Exception as market_error:
                        logger.warning(f"Market settlement failed for rotation provider {provider_id}: {market_error}")
                    
                    # Record analytics for token usage
                    try:
                        analytics = get_analytics()
                        if response and isinstance(response, dict):
                            usage = response.get('usage', {})
                            total_tokens = usage.get('total_tokens', 0)
                            prompt_tokens = usage.get('prompt_tokens', 0)
                            completion_tokens = usage.get('completion_tokens', 0)
                            
                            # If no token usage provided, estimate it
                            if total_tokens == 0:
                                try:
                                    messages = request_data.get('messages', [])
                                    estimated_prompt_tokens = count_messages_tokens(messages, model_name)
                                    
                                    # More realistic completion estimate
                                    max_tokens = request_data.get('max_tokens') or 0
                                    if max_tokens > 0:
                                        estimated_completion = min(max_tokens, estimated_prompt_tokens * 2)
                                    else:
                                        estimated_completion = max(estimated_prompt_tokens, 50)

                                    total_tokens = estimated_prompt_tokens + estimated_completion
                                    prompt_tokens = estimated_prompt_tokens
                                    completion_tokens = estimated_completion
                                    logger.debug(f"Estimated token usage for rotation: {total_tokens}")
                                except Exception as est_error:
                                    logger.debug(f"Token estimation failed: {est_error}")
                                    total_tokens = 150
                                    prompt_tokens = 0
                                    completion_tokens = 0
                            
                            # Try to extract actual cost from provider response
                            from .cost_extractor import extract_cost_from_response
                            actual_cost = extract_cost_from_response(response, provider_id)
                            
                            # Always record analytics
                            analytics.record_request(
                                provider_id=provider_id,
                                model_name=model_name,
                                tokens_used=total_tokens,
                                latency_ms=(time.time() - request_start_time) * 1000,
                                success=True,
                                rotation_id=rotation_id,
                                user_id=user_id,
                                token_id=token_id,
                                prompt_tokens=prompt_tokens if prompt_tokens > 0 else None,
                                completion_tokens=completion_tokens if completion_tokens > 0 else None,
                                actual_cost=actual_cost,
                                analytics_kind='execution'
                            )
                            self._record_dashboard_proxy_event(
                                None,
                                provider_id,
                                model_name,
                                True,
                                (time.time() - request_start_time) * 1000,
                                metadata={
                                    'prompt_tokens': prompt_tokens,
                                    'completion_tokens': completion_tokens,
                                    'total_tokens': total_tokens,
                                    'actual_cost': actual_cost,
                                    'stream': False,
                                    'rotation_id': rotation_id,
                                    'autoselect_id': request_data.get('_autoselect_id'),
                                },
                            )
                    except Exception as analytics_error:
                        logger.warning(f"Analytics recording failed: {analytics_error}")
                    
                    return response
            except RateLimitError as e:
                last_error = str(e)
                # Provider already disabled by handle_rate_limit_response — do NOT call record_failure()
                logger.warning(f"")
                logger.warning(f"=== ROTATION FAILOVER: rate limit on attempt {attempt + 1} ===")
                logger.warning(f"Provider {provider_id} / model {model_name} is rate limited: {str(e)}")
                logger.warning(f"Skipping to next model in rotation (transparent to client)")
                logger.warning(f"=== END ROTATION FAILOVER ===")
                tried_models.append(current_model)
                continue

            except Exception as e:
                last_error = str(e)

                # CoderAI "warming up": the broker session is temporarily gone while
                # the (often small/edge) worker cools down. The provider handler
                # already waited and retried internally; if it still failed, this is
                # NOT a provider fault — do not record a failure (which would disable
                # the provider for a long cooldown), just fail over to the next
                # provider, fully transparent to the client.
                if self._is_coderai_warming_error(e, provider_id):
                    logger.warning(
                        f"CoderAI provider {provider_id} still warming up after internal retries; "
                        f"failing over to the next provider (not recorded as a failure)"
                    )
                    tried_models.append(current_model)
                    continue

                # Increment retry count for this model
                model_retry_counts[model_key] = retry_count + 1

                logger.error(f"Attempt {attempt + 1} failed: {str(e)}")
                logger.error(f"Error type: {type(e).__name__}")
                logger.error(f"Model retry count: {model_retry_counts[model_key]}")

                # Non-retryable errors (e.g. HTTP 400/401/403/404/422 — bad request,
                # unsupported model, auth failure) are client/configuration problems,
                # NOT provider-health problems: the provider is up and answering, it
                # just rejected this particular request. They will fail identically if
                # we retry the same provider, so don't waste attempts — move straight to
                # the next provider by weight. Crucially, do NOT record a failure here:
                # counting it would trip the consecutive-failure auto-disable and pull a
                # perfectly healthy provider out of the rotation for the whole cooldown
                # (e.g. a rotation pointing codex at an unsupported model name would
                # disable codex after 3 requests). Only transient errors (5xx, timeouts,
                # network) count against provider health and are worth retrying.
                if self._is_non_retryable_error(e):
                    logger.warning(f"Non-retryable error on {model_name} (provider: {provider_id}); skipping to next provider by weight (not counted as a provider failure, transparent to client)")
                    tried_models.append(current_model)
                    continue

                # Transient/retryable error: count it against the provider's health.
                handler.record_failure()

                # If this is the first failure for this model, allow retry with rate limiting
                if model_retry_counts[model_key] < 2:
                    logger.info(f"Will retry model {model_name} with rate limiting...")
                    continue
                else:
                    logger.error(f"Model {model_name} has failed too many times, moving to next model...")
                    tried_models.append(current_model)
                    continue
        
        # All retries exhausted
        logger.error(f"")
        logger.error(f"=== ALL RETRIES EXHAUSTED ===")
        logger.error(f"Attempted {len(tried_models)} different model(s): {[m['name'] for m in tried_models]}")
        logger.error(f"Last error: {last_error}")
        logger.error(f"Max retries ({max_retries}) reached without success")
        
        # Record analytics for failed rotation request
        try:
            analytics = get_analytics()
            # Calculate latency
            latency_ms = (time.time() - request_start_time) * 1000
            logger.info(f"Failed Rotation Analytics: latency_ms={latency_ms:.2f}")
            
            # Estimate tokens for failed request
            try:
                messages = request_data.get('messages', [])
                estimated_tokens = count_messages_tokens(messages, rotation_id)
                total_tokens = estimated_tokens
                prompt_tokens = estimated_tokens
                completion_tokens = 0
            except Exception:
                total_tokens = 50  # Minimal estimate for failed requests
                prompt_tokens = 50
                completion_tokens = 0
            
            analytics.record_request(
                provider_id='rotation',
                model_name=rotation_id,
                tokens_used=total_tokens,
                latency_ms=latency_ms,
                success=False,
                error_type='RotationFailure',
                rotation_id=rotation_id,
                user_id=user_id,
                token_id=token_id,
                prompt_tokens=prompt_tokens,
                completion_tokens=completion_tokens,
                actual_cost=None,
                analytics_kind='selector'
            )
        except Exception as analytics_error:
            logger.warning(f"Analytics recording for failed rotation failed: {analytics_error}")
        
        # Build detailed error message
        error_details = []
        error_details.append(f"All providers in rotation '{rotation_id}' failed after {max_retries} attempts. Details:")
        error_details.append("")
        error_details.append(f"**Attempted models:**")
        error_details.append(f"{[m['name'] for m in tried_models]}")
        error_details.append("")
        
        # Format last error with JSON indentation if it contains JSON
        try:
            # Check if last_error contains JSON-like structure
            if '{' in last_error or '[' in last_error:
                # Try to extract and format JSON
                import json
                # Find JSON start and end
                json_start = last_error.find('{') if '{' in last_error else last_error.find('[')
                if json_start != -1:
                    json_end = last_error.rfind('}') + 1 if '{' in last_error else last_error.rfind(']') + 1
                    json_str = last_error[json_start:json_end]
                    try:
                        # Prettify JSON
                        parsed_json = json.loads(json_str)
                        formatted_json = json.dumps(parsed_json, indent=2)
                        # Replace JSON part with formatted version
                        error_part = last_error[:json_start]
                        error_details.append(f"**Last error:** {error_part}")
                        error_details.append(f"```json\n{formatted_json}\n```")
                    except Exception:
                        error_details.append(f"**Last error:**")
                        error_details.append(f"{last_error}")
                else:
                    error_details.append(f"**Last error:**")
                    error_details.append(f"{last_error}")
            else:
                error_details.append(f"**Last error:**")
                error_details.append(f"{last_error}")
        except Exception:
            error_details.append(f"**Last error:**")
            error_details.append(f"{last_error}")
        
        error_details.append("")
        error_details.append("**Provider status:**")
        for provider in providers:
            provider_id = provider['provider_id']
            provider_config = self.config.get_provider(provider_id)
            if provider_config:
                error_tracking = config.error_tracking.get(provider_id, {})
                disabled_until = error_tracking.get('disabled_until')
                failures = error_tracking.get('failures', 0)
                
                if disabled_until:
                    import time
                    cooldown_remaining = int(disabled_until - time.time())
                    if cooldown_remaining > 0:
                        error_details.append(f"• {provider_id}: Rate limited (cooldown: {cooldown_remaining}s remaining, failures: {failures})")
                    else:
                        error_details.append(f"• {provider_id}: Rate limited (cooldown expired, failures: {failures})")
                else:
                    error_details.append(f"• {provider_id}: Available (failures: {failures})")
            else:
                error_details.append(f"• {provider_id}: Not configured")
        
        # Check if notifyerrors is enabled - if so, return error as normal message instead of HTTP 503
        # Get stream parameter from request_data to determine response type
        stream = request_data.get('stream', False)
        logger.info(f"Request stream mode: {stream}")
        
        if notify_errors:
            logger.info(f"notifyerrors is enabled for rotation '{rotation_id}', returning error as normal message")
            # Return a normal response with error message instead of HTTP 503
            error_message = f"All providers in rotation '{rotation_id}' failed after {max_retries} attempts. Details:\n{chr(10).join(error_details[1:])}"
            error_response = {
                "id": f"error-{rotation_id}-{int(time.time())}",
                "object": "chat.completion",
                "created": int(time.time()),
                "model": rotation_id,
                "choices": [{
                    "index": 0,
                    "message": {
                        "role": "assistant",
                        "content": error_message
                    },
                    "finish_reason": "stop"
                }],
                "usage": {
                    "prompt_tokens": 0,
                    "completion_tokens": len(error_message),
                    "total_tokens": len(error_message)
                },
                "aisbf_error": True,
                "rotation_id": rotation_id,
                "attempted_models": [m['name'] for m in tried_models],
                "attempted_count": len(tried_models),
                "max_retries": max_retries,
                "last_error": last_error,
                "error_details": error_details
            }
            # For streaming requests, wrap in a simple streaming response
            if stream:
                return self._create_error_streaming_response(error_response)
            else:
                return error_response
        else:
            logger.info(f"notifyerrors is disabled for rotation '{rotation_id}', returning error with status code 429")
            # Return a normal response with error message and status code 429
            error_message = f"All providers in rotation '{rotation_id}' failed after {max_retries} attempts. Details:\n{chr(10).join(error_details[1:])}"
            error_response = {
                "id": f"error-{rotation_id}-{int(time.time())}",
                "object": "chat.completion",
                "created": int(time.time()),
                "model": rotation_id,
                "choices": [{
                    "index": 0,
                    "message": {
                        "role": "assistant",
                        "content": error_message
                    },
                    "finish_reason": "stop"
                }],
                "usage": {
                    "prompt_tokens": 0,
                    "completion_tokens": len(error_message),
                    "total_tokens": len(error_message)
                },
                "aisbf_error": True,
                "rotation_id": rotation_id,
                "attempted_models": [m['name'] for m in tried_models],
                "attempted_count": len(tried_models),
                "max_retries": max_retries,
                "last_error": last_error,
                "error_details": error_details
            }
            # For streaming requests, wrap in a simple streaming response
            if stream:
                return self._create_error_streaming_response(error_response, status_code=429)
            else:
                return JSONResponse(status_code=429, content=error_response)

    def _create_error_streaming_response(self, error_response: Dict, status_code: int = 200):
        """
        Create a simple StreamingResponse for error messages.
        
        This is used when notifyerrors is enabled and we need to return
        an error as a streaming response instead of raising an HTTPException.
        
        Args:
            error_response: The error response dict to stream
            status_code: The HTTP status code to return (default: 200)
        
        Returns:
            StreamingResponse with the error response in OpenAI-compatible streaming format
        """
        import json
        import time
        
        # Extract values from error_response for proper streaming format
        response_id = error_response.get('id', f"error-{int(time.time())}")
        created = error_response.get('created', int(time.time()))
        model = error_response.get('model', 'unknown')
        content = error_response.get('choices', [{}])[0].get('message', {}).get('content', '')
        
        async def error_stream_generator():
            # First chunk: role and content
            chunk_with_role = {
                "id": response_id,
                "object": "chat.completion.chunk",
                "created": created,
                "model": model,
                "choices": [{
                    "index": 0,
                    "delta": {
                        "role": "assistant",
                        "content": content
                    },
                    "finish_reason": None
                }]
            }
            yield f"data: {json.dumps(chunk_with_role)}\n\n".encode('utf-8')
            
            # Final chunk: finish_reason and usage
            final_chunk = {
                "id": response_id,
                "object": "chat.completion.chunk",
                "created": created,
                "model": model,
                "choices": [{
                    "index": 0,
                    "delta": {},
                    "finish_reason": "stop"
                }],
                "usage": error_response.get('usage', {
                    "prompt_tokens": 0,
                    "completion_tokens": len(content),
                    "total_tokens": len(content)
                })
            }
            yield f"data: {json.dumps(final_chunk)}\n\n".encode('utf-8')
            # Yield control to event loop to ensure chunk is flushed to client
            await asyncio.sleep(0)
            
            # Send [DONE] marker
            yield b"data: [DONE]\n\n"
            # Final flush to ensure all buffered data reaches the client
            await asyncio.sleep(0)
        
        sr = StreamingResponse(error_stream_generator(), media_type="text/event-stream", status_code=status_code)
        sr.aisbf_error = True
        return sr

    def _build_failover_notice(self, rotation_id, preferred_provider, preferred_model, used_provider, used_model):
        """Build the client-facing failover notification payloads.

        Returns a dict with ``metadata`` (JSON body field), ``headers`` (X-AISBF-*
        HTTP headers) and ``banner`` (human-readable content prefix), or ``None``
        when no real failover occurred (the served provider/model matches the
        preferred one). Header values are sanitised to be HTTP/latin-1 safe.
        """
        if str(used_provider) == str(preferred_provider) and str(used_model) == str(preferred_model):
            return None

        def _hsafe(value):
            return str(value).encode('ascii', 'ignore').decode('ascii')

        metadata = {
            "switched": True,
            "rotation": rotation_id,
            "preferred": {"provider": preferred_provider, "model": preferred_model},
            "used": {"provider": used_provider, "model": used_model},
        }
        headers = {
            "X-AISBF-Failover": "true",
            "X-AISBF-Provider": _hsafe(used_provider),
            "X-AISBF-Model": _hsafe(used_model),
            "X-AISBF-Preferred-Provider": _hsafe(preferred_provider),
            "X-AISBF-Preferred-Model": _hsafe(preferred_model),
        }
        banner = (
            f"ℹ️ Provider switched to {used_provider} ({used_model}) — "
            f"preferred {preferred_provider} ({preferred_model}) was unavailable.\n\n"
        )
        return {"metadata": metadata, "headers": headers, "banner": banner}

    def _apply_failover_notice_to_dict(self, response, failover_notice):
        """Inject failover metadata and a visible banner into a non-streaming
        chat-completion dict in place. Safe no-op for non-dict responses or when
        there is no notice."""
        if not failover_notice or not isinstance(response, dict):
            return response
        response['aisbf_failover'] = failover_notice['metadata']
        try:
            choices = response.get('choices') or []
            if choices:
                message = choices[0].setdefault('message', {})
                existing = message.get('content') or ''
                # OpenAI allows content alongside tool_calls, so prefixing the
                # banner is safe even for tool-call responses.
                message['content'] = failover_notice['banner'] + existing
        except Exception as inject_error:
            import logging
            logging.getLogger(__name__).warning(f"Failed to inject failover banner: {inject_error}")
        return response

    def _delegation_kind(self, provider_id):
        """Return 'rotation' or 'autoselect' when a rotation provider entry is a
        meta-provider that forwards to a named rotation/autoselect, else None."""
        pid = (provider_id or '').strip().lower()
        if pid in ('rotation', 'rotations'):
            return 'rotation'
        if pid in ('autoselect', 'autoselects', 'autoselection', 'autoselections'):
            return 'autoselect'
        return None

    def _is_failed_delegation_result(self, result) -> bool:
        """Return True when a delegated rotation/autoselect result represents a
        total failure, so the parent rotation can fail over to the next entry."""
        if result is None:
            return True
        # Error responses tagged by the rotation/autoselect error paths
        if isinstance(result, dict) and result.get('aisbf_error'):
            return True
        if getattr(result, 'aisbf_error', False):
            return True
        # JSONResponse / Response carrying an error status code
        status_code = getattr(result, 'status_code', None)
        if isinstance(status_code, int) and status_code >= 400:
            return True
        return False

    def _is_non_retryable_error(self, error) -> bool:
        """Return True when retrying the same provider for this error is pointless.

        Client errors that reflect the request/credentials/billing rather than a
        transient condition (HTTP 400/401/402/403/404/405/422) will fail identically
        on retry, so the rotation should fail over to the next provider immediately.
        402 Payment Required (e.g. Kilo "usage_limit_exceeded"/out of credits) is
        deterministic too: retrying the same provider only burns the failure budget
        and needlessly disables it. Rate limits (408/429) and server/network errors
        (5xx, timeouts) remain retryable. Rate-limit handling has its own
        RateLimitError path and never reaches here.
        """
        non_retryable_codes = {400, 401, 402, 403, 404, 405, 422}

        # Prefer the structured status code when the exception carries an HTTP
        # response (httpx.HTTPStatusError and similar).
        status_code = None
        response = getattr(error, 'response', None)
        if response is not None:
            status_code = getattr(response, 'status_code', None)
        if status_code is None:
            status_code = getattr(error, 'status_code', None)
        if isinstance(status_code, int):
            return status_code in non_retryable_codes

        # Fall back to scanning the error text for an explicit status code.
        msg = str(error)
        import re as _re
        for match in _re.findall(r'\b([45]\d{2})\b', msg):
            try:
                code = int(match)
            except ValueError:
                continue
            if code in non_retryable_codes:
                return True
        return False

    def _is_coderai_warming_error(self, error, provider_id) -> bool:
        """Return True when the error is a CoderAI provider that is merely 'warming up'.

        CoderAI workers often run on small/edge hardware that disconnects its broker
        session while cooling down (to avoid overheating). A missing broker session is
        therefore a transient condition, not a provider fault: instead of recording a
        failure (which would count toward disabling the provider for a long cooldown),
        the rotation should wait a few seconds and retry the same provider.
        """
        msg = str(error or '')
        if 'No active CoderAI broker session' not in msg:
            return False
        # The message is already CoderAI-specific; confirm the provider type when we
        # can, but don't require it (the snapshot lookup may be unavailable).
        try:
            ptype = (self._get_provider_type(provider_id) or '').lower()
            if ptype and ptype != 'coderai':
                return False
        except Exception:
            pass
        return True

    async def _prime_stream(self, response):
        """Pull the first chunk of a streaming response so that immediate provider
        errors are raised here (within the rotation retry loop) instead of after the
        client's StreamingResponse has started.

        Returns an object with the same iteration semantics as ``response`` (async
        generator stays async, sync iterable stays sync) with the consumed first
        chunk re-injected at the front. Non-iterable responses (e.g. dicts) are
        returned unchanged. Any exception raised while reading the first chunk
        propagates to the caller so it can fail over to the next provider.
        """
        # Async generators / async iterables (codex, kiro, google, etc.)
        if hasattr(response, '__aiter__'):
            aiter = response.__aiter__()
            try:
                first_chunk = await aiter.__anext__()
            except StopAsyncIteration:
                # Valid but empty stream — hand back an empty async generator.
                async def _empty_async():
                    return
                    yield  # pragma: no cover - makes this an async generator
                return _empty_async()

            async def _rewound_async():
                yield first_chunk
                async for chunk in aiter:
                    yield chunk
            return _rewound_async()

        # Sync iterables (e.g. OpenAI SDK Stream) — exclude mapping/bytes/str.
        if hasattr(response, '__iter__') and not isinstance(response, (dict, bytes, str)):
            iterator = iter(response)
            try:
                first_chunk = next(iterator)
            except StopIteration:
                return iter(())

            def _rewound_sync():
                yield first_chunk
                for chunk in iterator:
                    yield chunk
            return _rewound_sync()

        # Not a stream (dict / unexpected type) — nothing to prime.
        return response

    def _create_streaming_response(self, response, provider_type: str, provider_id: str, model_name: str, handler, request_data: Dict, effective_context: int, rotation_id: Optional[str] = None, user_id: Optional[int] = None, token_id: Optional[int] = None, request_start_time: Optional[float] = None, failover_info: Optional[Dict] = None):
        """
        Create a StreamingResponse with proper handling based on provider type.
        
        Args:
            response: The streaming response from the provider handler
            provider_type: The type of provider (e.g., 'google', 'openai', 'anthropic')
            provider_id: The provider identifier from configuration
            model_name: The model name being used
            handler: The provider handler (for recording success/failure)
            request_data: The original request data
            effective_context: The effective context (total tokens used)
        
        Returns:
            StreamingResponse with appropriate generator for the provider type
        """
        import logging
        import time
        import json
        logger = logging.getLogger(__name__)
        
        # Check if this is a Google or Kilo provider based on configuration
        is_google_provider = provider_type == 'google'
        is_kilo_provider = provider_type in ('kilo', 'kilocode')
        is_coderai_provider = provider_type == 'coderai'
        logger.info(f"Creating streaming response for provider type: {provider_type}, is_google: {is_google_provider}, is_kilo: {is_kilo_provider}, is_coderai: {is_coderai_provider}")
        
        # Generate system_fingerprint for this request
        # If seed is present in request, generate unique fingerprint per request
        seed = request_data.get('seed')
        system_fingerprint = generate_system_fingerprint(provider_id, seed)
        
        async def stream_generator(effective_context):
            import json
            accumulated_response_text = ""
            final_usage = None
            try:
                # If a failover changed the served provider/model, emit a leading
                # notification chunk carrying both a visible banner (delta content)
                # and the structured aisbf_failover metadata, before the real stream.
                if failover_info:
                    notice_chunk = {
                        "id": f"failover-{provider_id}-{int(time.time())}",
                        "object": "chat.completion.chunk",
                        "created": int(time.time()),
                        "model": model_name,
                        "system_fingerprint": system_fingerprint,
                        "provider": provider_id,
                        "aisbf_failover": failover_info["metadata"],
                        "choices": [{
                            "index": 0,
                            "delta": {"role": "assistant", "content": failover_info["banner"]},
                            "finish_reason": None
                        }]
                    }
                    yield f"data: {json.dumps(notice_chunk)}\n\n".encode('utf-8')

                if isinstance(response, dict):
                    # Complete non-streamed completion handed to the streaming path
                    # (e.g. the chunked-request handler always returns a plain dict).
                    # Iterating a dict yields its KEYS as string "chunks" and clients
                    # blow up ('str' object has no attribute 'choices') – synthesize a
                    # proper single-chunk SSE stream from it instead.
                    first_choice = (response.get('choices') or [{}])[0]
                    message = first_choice.get('message') or {}
                    delta = {"role": "assistant"}
                    if message.get('content') is not None:
                        delta['content'] = message.get('content')
                        accumulated_response_text = message.get('content') or ""
                    tool_calls = message.get('tool_calls')
                    if tool_calls:
                        # Streaming tool_calls carry a per-entry index.
                        delta['tool_calls'] = [
                            {**tc, 'index': i} if isinstance(tc, dict) and 'index' not in tc else tc
                            for i, tc in enumerate(tool_calls)
                        ]
                    synth_chunk = {
                        "id": response.get('id') or f"chatcmpl-{provider_id}-{int(time.time())}",
                        "object": "chat.completion.chunk",
                        "created": response.get('created') or int(time.time()),
                        "model": response.get('model') or model_name,
                        "system_fingerprint": system_fingerprint,
                        "choices": [{
                            "index": 0,
                            "delta": delta,
                            "finish_reason": first_choice.get('finish_reason') or "stop",
                        }],
                    }
                    if isinstance(response.get('usage'), dict):
                        synth_chunk['usage'] = response['usage']
                        final_usage = response['usage']
                    logger.info("Synthesized SSE stream from non-streaming dict response (chunked fallback)")
                    yield f"data: {json.dumps(synth_chunk)}\n\n".encode('utf-8')
                    yield b"data: [DONE]\n\n"
                    return

                if is_google_provider:
                    # Handle Google's streaming response
                    # Google provider returns an async generator
                    # Note: Google returns accumulated text, so we need to track and send only deltas
                    chunk_id = 0
                    accumulated_text = ""  # Track text we've already sent
                    accumulated_tool_calls = []  # Track tool calls we've already sent
                    created_time = int(time.time())
                    response_id = f"google-{model_name}-{created_time}"
                    
                    # Track completion tokens for Google responses (since Google doesn't provide them)
                    completion_tokens = 0
                    accumulated_response_text = ""  # Track full response for token counting
                    
                    # Whether the client requested tools. Text-encoded tool-call
                    # detection (see the buffered path below) needs the COMPLETE
                    # response, so we only stream incrementally when no tools were
                    # requested -- the dominant chat case.
                    request_has_tools = bool(request_data.get('tools'))

                    if not request_has_tools:
                        # TRUE INCREMENTAL streaming: forward each text delta to
                        # the client as soon as Google yields it.
                        sent_role = False
                        async for chunk in response:
                            chunk_text = ""
                            chunk_finish = None
                            try:
                                if hasattr(chunk, 'candidates') and chunk.candidates:
                                    candidate = chunk.candidates[0] if chunk.candidates else None
                                    if candidate and hasattr(candidate, 'content') and candidate.content:
                                        if hasattr(candidate.content, 'parts') and candidate.content.parts:
                                            for part in candidate.content.parts:
                                                if hasattr(part, 'text') and part.text:
                                                    chunk_text += part.text
                                        if hasattr(candidate, 'finish_reason'):
                                            google_finish = str(candidate.finish_reason)
                                            if google_finish in ('STOP', 'END_TURN', 'FINISH_REASON_UNSPECIFIED'):
                                                chunk_finish = "stop"
                                            elif google_finish == 'MAX_TOKENS':
                                                chunk_finish = "length"
                            except Exception as e:
                                logger.error(f"Error extracting text from Google stream chunk: {e}")
                                continue

                            # Google chunk text is the incremental delta; guard
                            # against providers that send accumulated text.
                            delta_text = chunk_text[len(accumulated_text):] if chunk_text.startswith(accumulated_text) and accumulated_text else chunk_text
                            if delta_text:
                                accumulated_text += delta_text
                                accumulated_response_text += delta_text
                                delta = {"content": delta_text, "refusal": None, "role": None, "tool_calls": None}
                                if not sent_role:
                                    delta["role"] = "assistant"
                                    sent_role = True
                                out_chunk = {
                                    "id": response_id,
                                    "object": "chat.completion.chunk",
                                    "created": created_time,
                                    "model": model_name,
                                    "service_tier": None,
                                    "system_fingerprint": system_fingerprint,
                                    "usage": None,
                                    "provider": provider_id,
                                    "choices": [{
                                        "index": 0,
                                        "delta": delta,
                                        "finish_reason": None,
                                        "logprobs": None,
                                        "native_finish_reason": None
                                    }]
                                }
                                yield f"data: {json.dumps(out_chunk)}\n\n".encode('utf-8')
                                await asyncio.sleep(0)

                        # Final chunk with finish reason and usage statistics.
                        if accumulated_response_text:
                            completion_tokens = count_messages_tokens([{"role": "assistant", "content": accumulated_response_text}], model_name)
                        total_tokens = effective_context + completion_tokens
                        final_chunk = {
                            "id": response_id,
                            "object": "chat.completion.chunk",
                            "created": created_time,
                            "model": model_name,
                            "service_tier": None,
                            "system_fingerprint": system_fingerprint,
                            "usage": {
                                "prompt_tokens": effective_context,
                                "completion_tokens": completion_tokens,
                                "total_tokens": total_tokens,
                                "effective_context": effective_context
                            },
                            "provider": provider_id,
                            "choices": [{
                                "index": 0,
                                "delta": {"content": "", "function_call": None, "refusal": None, "role": None, "tool_calls": None},
                                "finish_reason": "stop",
                                "logprobs": None,
                                "native_finish_reason": "stop"
                            }]
                        }
                        yield f"data: {json.dumps(final_chunk)}\n\n".encode('utf-8')
                        await asyncio.sleep(0)
                    else:
                        # Collect all chunks first to know when we're at the last one
                        chunks_list = []
                        async for chunk in response:
                            chunks_list.append(chunk)
                    
                        total_chunks = len(chunks_list)
                        chunk_idx = 0
                    
                        for chunk in chunks_list:
                            try:
                                logger.debug(f"Google chunk type: {type(chunk)}")
                                logger.debug(f"Google chunk: {chunk}")
                            
                                # Extract text and tool calls from Google chunk (this is accumulated text)
                                chunk_text = ""
                                chunk_tool_calls = []
                                finish_reason = None
                                try:
                                    if hasattr(chunk, 'candidates') and chunk.candidates:
                                        candidate = chunk.candidates[0] if chunk.candidates else None
                                        if candidate and hasattr(candidate, 'content') and candidate.content:
                                            if hasattr(candidate.content, 'parts') and candidate.content.parts:
                                                for part in candidate.content.parts:
                                                    # Extract text content
                                                    if hasattr(part, 'text') and part.text:
                                                        chunk_text += part.text
                                                    # Extract function calls (Google's format)
                                                    if hasattr(part, 'function_call') and part.function_call:
                                                        function_call = part.function_call
                                                        # Convert Google function call to OpenAI format
                                                        import json
                                                        openai_tool_call = {
                                                            "id": f"call_{len(chunk_tool_calls)}",
                                                            "type": "function",
                                                            "function": {
                                                                "name": function_call.name,
                                                                "arguments": json.dumps(function_call.args) if hasattr(function_call, 'args') else "{}"
                                                            }
                                                        }
                                                        chunk_tool_calls.append(openai_tool_call)
                                                        logger.info(f"Extracted tool call from Google chunk: {openai_tool_call}")
                                        # Check for finish reason in candidate
                                        if hasattr(candidate, 'finish_reason'):
                                            google_finish = str(candidate.finish_reason)
                                            if google_finish in ('STOP', 'END_TURN', 'FINISH_REASON_UNSPECIFIED'):
                                                finish_reason = "stop"
                                            elif google_finish == 'MAX_TOKENS':
                                                finish_reason = "length"
                                except Exception as e:
                                    logger.error(f"Error extracting text from Google chunk: {e}")
                            
                                # Calculate the delta (only the new text since last chunk)
                                delta_text = chunk_text[len(accumulated_text):] if chunk_text.startswith(accumulated_text) else chunk_text
                                accumulated_text = chunk_text  # Update accumulated text for next iteration
                            
                                # Track completion tokens for Google responses
                                if delta_text:
                                    accumulated_response_text += delta_text
                            
                                chunk_idx += 1
                            except Exception as chunk_error:
                                error_msg = str(chunk_error)
                                logger.error(f"Error processing Google chunk: {error_msg}")
                                logger.error(f"Chunk type: {type(chunk)}")
                                logger.error(f"Chunk content: {chunk}")
                                chunk_idx += 1
                                continue
                    
                        # After collecting all chunks, check if the accumulated text contains a tool call pattern
                        # This handles models that return tool calls as text instead of using function_call attributes
                        tool_calls = None
                        final_text = accumulated_response_text
                    
                        # Check for tool call patterns in the accumulated text
                        if accumulated_response_text:
                            import re as re_module
                        
                            # Simple approach: just look for "tool: {...}" pattern and extract the JSON
                            # This avoids complex nested parsing issues
                            tool_pattern = r'tool:\s*(\{[^{}]*\{[^{}]*\}[^{}]*\}|\{[^{}]+\})'
                            tool_match = re_module.search(tool_pattern, accumulated_response_text, re_module.DOTALL)
                        
                            if tool_match:
                                try:
                                    # Extract the tool JSON using brace counting for robustness
                                    tool_start = accumulated_response_text.find('tool:')
                                    if tool_start != -1:
                                        json_start = accumulated_response_text.find('{', tool_start)
                                        if json_start != -1:
                                            brace_count = 0
                                            json_end = json_start
                                            for i, c in enumerate(accumulated_response_text[json_start:], json_start):
                                                if c == '{':
                                                    brace_count += 1
                                                elif c == '}':
                                                    brace_count -= 1
                                                    if brace_count == 0:
                                                        json_end = i + 1
                                                        break
                                        
                                            tool_json_str = accumulated_response_text[json_start:json_end]
                                            logger.debug(f"Extracted tool JSON: {tool_json_str[:200]}...")
                                        
                                            try:
                                                parsed_tool = json.loads(tool_json_str)
                                            except json.JSONDecodeError:
                                                # Try fixing common issues: single quotes, trailing commas
                                                fixed_json = tool_json_str.replace("'", '"')
                                                fixed_json = re_module.sub(r',\s*}', '}', fixed_json)
                                                fixed_json = re_module.sub(r',\s*]', ']', fixed_json)
                                                parsed_tool = json.loads(fixed_json)
                                        
                                            # Convert to OpenAI tool_calls format
                                            tool_calls = [{
                                                "id": f"call_0",
                                                "type": "function",
                                                "function": {
                                                    "name": parsed_tool.get('action', parsed_tool.get('name', 'unknown')),
                                                    "arguments": json.dumps({k: v for k, v in parsed_tool.items() if k not in ['action', 'name']})
                                                }
                                            }]
                                            logger.info(f"Converted streaming tool call to OpenAI format: {tool_calls}")
                                        
                                            # Extract final assistant text after the tool JSON
                                            # Look for pattern: }\\nassistant: [{'type': 'text', 'text': "..."}]
                                            # or just return empty since the tool call is the main content
                                            after_tool = accumulated_response_text[json_end:]
                                            assistant_pattern = r"assistant:\s*\[.*'text':\s*['\"](.+?)['\"].*\]\s*\]?\s*$"
                                            assistant_match = re_module.search(assistant_pattern, after_tool, re_module.DOTALL)
                                            if assistant_match:
                                                final_text = assistant_match.group(1)
                                                # Unescape common escape sequences
                                                final_text = final_text.replace("\\n", "\n").replace("\\'", "'").replace('\\"', '"')
                                            else:
                                                final_text = ""
                                except (json.JSONDecodeError, ValueError, SyntaxError, Exception) as e:
                                    logger.debug(f"Failed to parse tool JSON in streaming: {e}")
                    
                        # Now send the response chunks
                        # If we detected tool calls, send them in the first chunk with role
                        if tool_calls:
                            # First chunk with tool_calls
                            tool_chunk = {
                                "id": response_id,
                                "object": "chat.completion.chunk",
                                "created": created_time,
                                "model": model_name,
                                "service_tier": None,
                                "system_fingerprint": system_fingerprint,
                                "usage": None,
                                "provider": provider_id,
                                "choices": [{
                                    "index": 0,
                                    "delta": {
                                        "content": None,
                                        "refusal": None,
                                        "role": "assistant",
                                        "tool_calls": tool_calls
                                    },
                                    "finish_reason": None,
                                    "logprobs": None,
                                    "native_finish_reason": None
                                }]
                            }
                            yield f"data: {json.dumps(tool_chunk)}\n\n".encode('utf-8')
                        
                            # If there's final assistant text, send it
                            if final_text:
                                text_chunk = {
                                    "id": response_id,
                                    "object": "chat.completion.chunk",
                                    "created": created_time,
                                    "model": model_name,
                                    "service_tier": None,
                                    "system_fingerprint": system_fingerprint,
                                    "usage": None,
                                    "provider": provider_id,
                                    "choices": [{
                                        "index": 0,
                                        "delta": {
                                            "content": final_text,
                                            "refusal": None,
                                            "role": None,
                                            "tool_calls": None
                                        },
                                        "finish_reason": None,
                                        "logprobs": None,
                                        "native_finish_reason": None
                                    }]
                                }
                                yield f"data: {json.dumps(text_chunk)}\n\n".encode('utf-8')
                        else:
                            # No tool calls detected, send text normally
                            # Send the accumulated text as a single chunk
                            if accumulated_response_text:
                                text_chunk = {
                                    "id": response_id,
                                    "object": "chat.completion.chunk",
                                    "created": created_time,
                                    "model": model_name,
                                    "service_tier": None,
                                    "system_fingerprint": system_fingerprint,
                                    "usage": None,
                                    "provider": provider_id,
                                    "choices": [{
                                        "index": 0,
                                        "delta": {
                                            "content": accumulated_response_text,
                                            "refusal": None,
                                            "role": "assistant",
                                            "tool_calls": None
                                        },
                                        "finish_reason": None,
                                        "logprobs": None,
                                        "native_finish_reason": None
                                    }]
                                }
                                yield f"data: {json.dumps(text_chunk)}\n\n".encode('utf-8')
                    
                        # Send final chunk with finish reason and usage statistics
                        if accumulated_response_text:
                            completion_tokens = count_messages_tokens([{"role": "assistant", "content": accumulated_response_text}], model_name)
                        total_tokens = effective_context + completion_tokens
                        final_chunk = {
                            "id": response_id,
                            "object": "chat.completion.chunk",
                            "created": created_time,
                            "model": model_name,
                            "service_tier": None,
                            "system_fingerprint": system_fingerprint,
                            "usage": {
                                "prompt_tokens": effective_context,
                                "completion_tokens": completion_tokens,
                                "total_tokens": total_tokens,
                                "effective_context": effective_context
                            },
                            "provider": provider_id,
                            "choices": [{
                                "index": 0,
                                "delta": {
                                    "content": "",
                                    "function_call": None,
                                    "refusal": None,
                                    "role": None,
                                    "tool_calls": None
                                },
                                "finish_reason": "stop",
                                "logprobs": None,
                                "native_finish_reason": "stop"
                            }]
                        }
                        yield f"data: {json.dumps(final_chunk)}\n\n".encode('utf-8')
                        # Yield control to event loop to ensure final chunk is flushed to client
                        await asyncio.sleep(0)
                elif is_kilo_provider or is_coderai_provider:
                    # Handle Kilo/KiloCode streaming response
                    # Kilo/CoderAI return async generators that yield OpenAI-compatible SSE bytes
                    accumulated_response_text = ""
                    chunk_count = 0

                    async for chunk in response:
                        chunk_count += 1
                        try:
                            # Pass through the chunk as-is (already in SSE format)
                            if isinstance(chunk, bytes):
                                # Parse to track content for token counting
                                try:
                                    chunk_str = chunk.decode('utf-8')
                                    for sse_line in chunk_str.split('\n'):
                                        sse_line = sse_line.strip()
                                        if sse_line.startswith('data: '):
                                            data_str = sse_line[6:].strip()
                                            if data_str and data_str != '[DONE]':
                                                try:
                                                    chunk_data = json.loads(data_str)
                                                    usage_data = chunk_data.get('usage') if isinstance(chunk_data, dict) else None
                                                    if isinstance(usage_data, dict):
                                                        final_usage = usage_data
                                                    choices = chunk_data.get('choices', [])
                                                    if choices:
                                                        delta = choices[0].get('delta', {})
                                                        delta_content = delta.get('content', '')
                                                        if delta_content:
                                                            accumulated_response_text += delta_content
                                                except json.JSONDecodeError:
                                                    pass
                                except (UnicodeDecodeError, Exception):
                                    pass
                                yield chunk
                            elif isinstance(chunk, str):
                                yield chunk.encode('utf-8')
                            else:
                                yield f"data: {json.dumps(chunk)}\n\n".encode('utf-8')
                        except Exception as chunk_error:
                            logger.warning(f"Error processing {provider_type} chunk: {chunk_error}")
                            continue

                    logger.info(f"{provider_type} streaming processed {chunk_count} chunks total")
                else:
                    # Handle OpenAI/Anthropic/Kiro streaming responses
                    # Some providers return async generators, others return sync iterables
                    accumulated_response_text = ""  # Track full response for token counting

                    # Check if response is an async generator
                    import inspect
                    if inspect.iscoroutinefunction(response) or hasattr(response, '__aiter__'):
                        # Handle async generator (like Kiro)
                        logger.info(f"Detected async generator response, using async for loop")
                        chunk_count = 0
                        try:
                            async for chunk in response:
                                chunk_count += 1
                                try:
                                    logger.debug(f"Async chunk type: {type(chunk)}")
                                    logger.debug(f"Async chunk: {chunk}")

                                    # For Kiro/Claude CLI, chunks may be pre-formatted SSE bytes or strings
                                    if isinstance(chunk, bytes):
                                        logger.debug(f"Yielding raw bytes chunk: {len(chunk)} bytes")
                                        yield chunk
                                    elif isinstance(chunk, str):
                                        final_usage = self._extract_usage_from_sse_chunk(chunk) or final_usage
                                        yield chunk.encode('utf-8')
                                    else:
                                        # Fallback: treat as dict and serialize
                                        chunk_dict = chunk.model_dump() if hasattr(chunk, 'model_dump') else chunk
                                        if isinstance(chunk_dict, dict) and isinstance(chunk_dict.get('usage'), dict):
                                            final_usage = chunk_dict.get('usage')
                                        yield f"data: {json.dumps(chunk_dict)}\n\n".encode('utf-8')
                                except Exception as chunk_error:
                                    error_msg = str(chunk_error)
                                    logger.warning(f"Error processing async chunk: {error_msg}")
                                    logger.warning(f"Chunk type: {type(chunk)}")
                                    logger.warning(f"Chunk content: {chunk}")
                                    continue
                        except Exception as async_error:
                            logger.error(f"Error in async for loop: {async_error}")
                            logger.error(f"Response type: {type(response)}")
                            logger.error(f"Response has __aiter__: {hasattr(response, '__aiter__')}")
                            logger.error(f"Response is coroutine function: {inspect.iscoroutinefunction(response)}")
                            raise async_error
                        finally:
                            logger.info(f"Async generator processed {chunk_count} chunks total")
                    else:
                        # Handle sync iterable (like OpenAI SDK)
                        logger.info(f"Detected sync iterable response, using regular for loop")
                        for chunk in response:
                            try:
                                logger.debug(f"Sync chunk type: {type(chunk)}")
                                logger.debug(f"Sync chunk: {chunk}")

                                # For OpenAI-compatible providers, just pass through the raw chunk
                                # Convert chunk to dict and serialize as JSON
                                chunk_dict = chunk.model_dump() if hasattr(chunk, 'model_dump') else chunk

                                # Track response content for token calculation
                                if isinstance(chunk_dict, dict):
                                    choices = chunk_dict.get('choices', [])
                                    if choices:
                                        delta = choices[0].get('delta', {})
                                        delta_content = delta.get('content', '')
                                        if delta_content:
                                            accumulated_response_text += delta_content

                                # Add effective_context to the last chunk (when finish_reason is present)
                                if isinstance(chunk_dict, dict):
                                    choices = chunk_dict.get('choices', [])
                                    if choices and choices[0].get('finish_reason') is not None:
                                        # This is the last chunk, add effective_context
                                        if 'usage' not in chunk_dict:
                                            chunk_dict['usage'] = {}
                                        chunk_dict['usage']['effective_context'] = effective_context

                                        # If provider doesn't provide token counts, calculate them
                                        if chunk_dict['usage'].get('total_tokens') is None:
                                            # Calculate completion tokens from accumulated response
                                            if accumulated_response_text:
                                                completion_tokens = count_messages_tokens([{"role": "assistant", "content": accumulated_response_text}], request_data['model'])
                                            else:
                                                completion_tokens = 0
                                            total_tokens = effective_context + completion_tokens
                                            chunk_dict['usage']['prompt_tokens'] = effective_context
                                            chunk_dict['usage']['completion_tokens'] = completion_tokens
                                            chunk_dict['usage']['total_tokens'] = total_tokens
                                        final_usage = chunk_dict.get('usage')

                                yield f"data: {json.dumps(chunk_dict)}\n\n".encode('utf-8')
                            except Exception as chunk_error:
                                error_msg = str(chunk_error)
                                logger.warning(f"Error serializing sync chunk: {error_msg}")
                                logger.warning(f"Chunk type: {type(chunk)}")
                                logger.warning(f"Chunk content: {chunk}")
                                continue
                
                handler.record_success()

                # Record analytics after stream completes
                try:
                    analytics = get_analytics()
                    prompt_tokens = (final_usage or {}).get('prompt_tokens')
                    completion_tokens_count = (final_usage or {}).get('completion_tokens')
                    total_tokens = (final_usage or {}).get('total_tokens')
                    if prompt_tokens is None:
                        prompt_tokens = effective_context
                    if completion_tokens_count is None:
                        completion_tokens_count = count_messages_tokens([{"role": "assistant", "content": accumulated_response_text}], model_name) if accumulated_response_text else 0
                    if total_tokens is None:
                        total_tokens = prompt_tokens + completion_tokens_count
                    latency_ms = (time.time() - request_start_time) * 1000 if request_start_time else 0
                    analytics.record_request(
                        provider_id=provider_id,
                        model_name=model_name,
                        tokens_used=total_tokens,
                        latency_ms=latency_ms,
                        success=True,
                        rotation_id=rotation_id,
                        user_id=user_id,
                        token_id=token_id,
                        prompt_tokens=prompt_tokens if prompt_tokens > 0 else None,
                        completion_tokens=completion_tokens_count if completion_tokens_count > 0 else None,
                        actual_cost=None,
                        analytics_kind='execution'
                    )
                    self._record_dashboard_proxy_event(
                        None,
                        provider_id,
                        model_name,
                        True,
                        latency_ms,
                        metadata={
                            'prompt_tokens': prompt_tokens,
                            'completion_tokens': completion_tokens_count,
                            'total_tokens': total_tokens,
                            'stream': True,
                            'rotation_id': rotation_id,
                            'autoselect_id': request_data.get('_autoselect_id'),
                        },
                    )
                    try:
                        self._settle_market_result(provider_id, {
                            'prompt_tokens': prompt_tokens,
                            'completion_tokens': completion_tokens_count,
                            'total_tokens': total_tokens,
                        }, 1, {'kind': 'streaming_chat_completion', 'rotation_id': rotation_id}, request_data=request_data)
                    except Exception as market_error:
                        logger.warning(f"Market settlement failed for streaming {provider_id}: {market_error}")
                except Exception as analytics_error:
                    logger.warning(f"Analytics recording for streaming rotation failed: {analytics_error}")

            except RateLimitError as e:
                # Provider already disabled by handle_429 — do NOT call record_failure()
                latency_ms = (time.time() - request_start_time) * 1000 if request_start_time else 0
                self._record_dashboard_proxy_event(
                    None,
                    provider_id,
                    model_name,
                    False,
                    latency_ms,
                    metadata={
                        'error_type': type(e).__name__,
                        'stream': True,
                        'rotation_id': rotation_id,
                        'status_code': 503,
                        'autoselect_id': request_data.get('_autoselect_id'),
                    },
                )
                error_dict = {"error": str(e)}
                yield f"data: {json.dumps(error_dict)}\n\n".encode('utf-8')
            except Exception as e:
                # Don't count non-provider-health errors against the provider:
                #  - CoderAI warm-up (no broker session) is transient.
                #  - Non-retryable client/config errors (HTTP 400/401/403/404/422 —
                #    bad request, unsupported model, auth) mean the provider is up and
                #    simply rejected this request; disabling it would pull a healthy
                #    provider out of rotation for the whole cooldown.
                if _should_record_failure(e) and not self._is_non_retryable_error(e):
                    handler.record_failure()
                latency_ms = (time.time() - request_start_time) * 1000 if request_start_time else 0
                self._record_dashboard_proxy_event(
                    None,
                    provider_id,
                    model_name,
                    False,
                    latency_ms,
                    metadata={
                        'error_type': type(e).__name__,
                        'stream': True,
                        'rotation_id': rotation_id,
                        'autoselect_id': request_data.get('_autoselect_id'),
                    },
                )
                error_dict = {"error": str(e)}
                yield f"data: {json.dumps(error_dict)}\n\n".encode('utf-8')

        stream_headers = failover_info["headers"] if failover_info else None
        return StreamingResponse(stream_generator(effective_context), media_type="text/event-stream", headers=stream_headers)

    async def handle_rotation_model_list(self, rotation_id: str) -> List[Dict]:
        rotation_config = self.config.get_rotation(rotation_id)
        if not rotation_config:
            raise HTTPException(status_code=400, detail=f"Rotation {rotation_id} not found")

        all_models = []
        for provider in rotation_config.providers:
            provider_id = provider['provider_id']
            # Meta-providers forward to a named rotation/autoselect; they don't
            # expose concrete provider models, so skip them in the model listing.
            if self._delegation_kind(provider_id):
                continue
            provider_config = self.config.get_provider(provider_id)

            for model in provider['models']:
                model_name = model['name']
                model_dict = {
                    "id": f"{provider_id}/{model_name}",
                    "name": model_name,
                    "provider_id": provider_id,
                    "weight": model['weight'],
                    "rate_limit": model.get('rate_limit')
                }
                
                # Add context window information
                # Priority: model config in rotation > provider config > first model in provider
                if model.get('context_size'):
                    model_dict['context_window'] = model['context_size']
                elif provider_config:
                    # Try to find in provider config
                    found_in_provider = False
                    for pm in getattr(provider_config, "models", []) or []:
                        if pm.name == model_name and hasattr(pm, 'context_size') and pm.context_size:
                            model_dict['context_window'] = pm.context_size
                            found_in_provider = True
                            break
                    if not found_in_provider:
                        provider_default = getattr(provider_config, 'default_context_size', None)
                        if provider_default:
                            # Provider-level default takes precedence over heuristic auto-derivation
                            model_dict['context_window'] = provider_default
                        # Auto-derive from first model in provider (which has context_size from dynamic fetch)
                        elif getattr(provider_config, "models", []) and len(getattr(provider_config, "models", [])) > 0:
                            first_model = getattr(provider_config, "models", [])[0]
                            if hasattr(first_model, 'context_size') and first_model.context_size:
                                model_dict['context_window'] = first_model.context_size
                            elif hasattr(first_model, 'context_window') and first_model.context_window:
                                model_dict['context_window'] = first_model.context_window
                            elif hasattr(first_model, 'context_length') and first_model.context_length:
                                model_dict['context_window'] = first_model.context_length
                            else:
                                model_dict['context_window'] = self._infer_context_window(model_name, provider_config.type)
                        else:
                            model_dict['context_window'] = self._infer_context_window(model_name, provider_config.type)
                
                # Add capabilities information
                if model.get('capabilities'):
                    model_dict['capabilities'] = model['capabilities']
                elif provider_config:
                    # Try to find in provider config
                    for pm in getattr(provider_config, "models", []) or []:
                        if pm.name == model_name and hasattr(pm, 'capabilities'):
                            model_dict['capabilities'] = pm.capabilities
                            break
                    if 'capabilities' not in model_dict:
                        model_dict['capabilities'] = self._detect_capabilities(model_name, provider_config.type)
                
                all_models.append(model_dict)

        return all_models
    
    def _infer_context_window(self, model_name: str, provider_type: str) -> int:
        """Infer context window size from model name or provider type"""
        model_lower = model_name.lower()
        
        # Known model patterns
        if 'gpt-4' in model_lower:
            if 'turbo' in model_lower or '1106' in model_lower or '0125' in model_lower:
                return 128000
            return 8192
        elif 'gpt-3.5' in model_lower:
            if 'turbo' in model_lower and ('1106' in model_lower or '0125' in model_lower):
                return 16385
            return 4096
        elif 'claude-3' in model_lower:
            return 200000
        elif 'claude-2' in model_lower:
            return 100000
        elif 'gemini' in model_lower:
            if '1.5' in model_lower:
                return 2000000 if 'pro' in model_lower else 1000000
            elif '2.0' in model_lower:
                return 1000000
            return 32000
        elif 'llama' in model_lower:
            if '3' in model_lower:
                return 128000
            return 4096
        elif 'mistral' in model_lower:
            if 'large' in model_lower:
                return 32000
            return 8192
        
        # Default based on provider type
        if provider_type == 'google':
            return 32000
        elif provider_type == 'anthropic':
            return 100000
        elif provider_type == 'openai':
            return 8192
        
        # Generic default
        return 4096
    
    def _detect_capabilities(self, model_name: str, provider_type: str) -> List[str]:
        """Auto-detect model capabilities based on model name and provider type."""
        return infer_model_capabilities(model_name=model_name, provider_type=provider_type).capabilities

class AutoselectHandler:
    def __init__(self, user_id=None):
        self.user_id = user_id
        self.config = config
        self._skill_file_content = None
        # Load user-specific configs if user_id is provided
        if user_id:
            self._load_user_configs()
            # Override config to only use user-specific configs with NO global fallback
            self.autoselects = {}
            for autoselect in self.user_autoselects:
                self.autoselects[autoselect['autoselect_id']] = autoselect['config']
        else:
            self.user_providers = {}
            self.user_rotations = {}
            self.user_autoselects = {}
            self.autoselects = self.config.autoselect if hasattr(self.config, 'autoselect') else {}

    def _register_delegation(self, request_data: Dict, key: str, max_depth: int = 8):
        """Register this rotation/autoselect in the shared delegation chain for
        loop protection. Raises HTTP 508 if the key is already present (loop) or
        the depth limit is exceeded. Stores the extended chain on request_data
        (carried into child requests) and on the instance (for selection requests).
        """
        chain = list(request_data.get('_delegation_chain') or [])
        if key in chain:
            raise HTTPException(status_code=508, detail=f"Delegation loop detected for '{key}' (chain: {' -> '.join(chain)})")
        if len(chain) >= max_depth:
            raise HTTPException(status_code=508, detail=f"Delegation depth limit ({max_depth}) reached (chain: {' -> '.join(chain)})")
        chain = chain + [key]
        request_data['_delegation_chain'] = chain
        self._delegation_chain = chain
        return chain

    def _get_market_source_details(self, provider_id: str):
        provider_config = None
        if self.user_id and provider_id in getattr(self, 'user_providers', {}):
            provider_config = self.user_providers.get(provider_id)
        elif provider_id in getattr(self.config, 'providers', {}):
            cfg = self.config.get_provider(provider_id)
            provider_config = cfg.model_dump() if hasattr(cfg, 'model_dump') else cfg
        if isinstance(provider_config, dict):
            return provider_config.get('market_source')
        return None

    @staticmethod
    def _market_request_id(request_data: Optional[Dict], fallback_provider_id: str) -> str:
        payload = {
            'provider_id': fallback_provider_id,
            'model': (request_data or {}).get('model'),
            'messages': (request_data or {}).get('messages'),
            'input': (request_data or {}).get('input'),
            'prompt': (request_data or {}).get('prompt'),
            'voice': (request_data or {}).get('voice'),
            'endpoint_path': (request_data or {}).get('_market_endpoint_path'),
        }
        serialized = json.dumps(payload, sort_keys=True, default=str)
        return hashlib.sha256(serialized.encode('utf-8')).hexdigest()

    def _settle_market_result(self, provider_id: str, usage: Optional[Dict], requests_count: int = 1, metadata: Optional[Dict] = None, request_data: Optional[Dict] = None):
        market_source = self._get_market_source_details(provider_id)
        if not self.user_id or not market_source:
            return None
        listing_id = market_source.get('listing_id')
        if not listing_id:
            return None
        db = DatabaseRegistry.get_config_database()
        settlement_metadata = {
            'provider_id': provider_id,
            'market_imported': True,
            'market_request_id': self._market_request_id(request_data, provider_id),
        }
        if metadata:
            settlement_metadata.update(metadata)
        return db.settle_market_usage(
            consumer_user_id=self.user_id,
            listing_id=int(listing_id),
            prompt_tokens=(usage or {}).get('prompt_tokens', 0),
            completion_tokens=(usage or {}).get('completion_tokens', 0),
            requests_count=requests_count,
            metadata=settlement_metadata,
        )

    def _get_response_cache_backend(self):
        aisbf_config = self.config.get_aisbf_config()
        if not (aisbf_config and aisbf_config.response_cache and aisbf_config.response_cache.enabled):
            return None
        return get_response_cache(aisbf_config.response_cache.model_dump())

    def _should_cache_response(self, provider_config=None, model_config=None, rotation_config=None, autoselect_config=None):
        return self.config.resolve_feature_enabled(
            'response_cache',
            model_config=model_config,
            provider_config=provider_config,
            rotation_config=rotation_config,
            autoselect_config=autoselect_config,
        )

    def _load_user_configs(self):
        """Load user-specific configurations from database"""
        from .database import DatabaseRegistry
        db = DatabaseRegistry.get_config_database()
        self.user_providers = db.get_user_providers(self.user_id)
        self.user_rotations = db.get_user_rotations(self.user_id)
        self.user_autoselects = db.get_user_autoselects(self.user_id)
        
    def reload_user_configs(self):
        """Reload user-specific configurations from database"""
        if self.user_id:
            self._load_user_configs()
            # Refresh autoselects dict after reload
            self.autoselects = {}
            for autoselect in self.user_autoselects:
                self.autoselects[autoselect['autoselect_id']] = autoselect['config']

    def _get_skill_file_content(self) -> str:
        """Load the autoselect.md skill file content"""
        if self._skill_file_content is None:
            # Check for user-specific prompt first if user_id is present
            if self.user_id is not None:
                from .database import DatabaseRegistry
                db = DatabaseRegistry.get_config_database()
                user_prompt = db.get_user_prompt(self.user_id, 'autoselect')
                if user_prompt is not None:
                    self._skill_file_content = user_prompt
                    return self._skill_file_content

            # Try installed locations first
            installed_dirs = [
                Path('/usr/share/aisbf'),
                Path.home() / '.local' / 'share' / 'aisbf',
            ]
            
            for installed_dir in installed_dirs:
                skill_file = installed_dir / 'autoselect.md'
                if skill_file.exists():
                    with open(skill_file) as f:
                        self._skill_file_content = f.read()
                    return self._skill_file_content
            
            # Fallback to source tree config directory
            source_dir = Path(__file__).parent.parent / 'config'
            skill_file = source_dir / 'autoselect.md'
            if skill_file.exists():
                with open(skill_file) as f:
                    self._skill_file_content = f.read()
                return self._skill_file_content
            
            raise FileNotFoundError("Could not find autoselect.md skill file")
        
        return self._skill_file_content

    def reset_internal_model(self):
        """Unload the shared internal model from memory so it is re-loaded on next use."""
        global _internal_model_singleton, _internal_tokenizer_singleton
        # Acquire init lock so we don't race with a concurrent initialization
        with _internal_model_init_lock:
            _internal_model_singleton = None
            _internal_tokenizer_singleton = None

    def _initialize_internal_model(self):
        """Load the shared HuggingFace model once; safe to call from multiple concurrent threads."""
        global _internal_model_singleton, _internal_tokenizer_singleton
        import logging
        import json
        from pathlib import Path
        logger = logging.getLogger(__name__)

        # Fast path: model already loaded (no lock needed for read)
        if _internal_model_singleton is not None:
            return

        # Slow path: acquire init lock, then check again (double-checked locking)
        with _internal_model_init_lock:
            if _internal_model_singleton is not None:
                return  # Another thread loaded it while we waited

            try:
                import torch
                from transformers import AutoTokenizer, AutoModelForCausalLM

                logger.info("=== INITIALIZING INTERNAL SELECTION MODEL ===")

                config_path = Path.home() / '.aisbf' / 'aisbf.json'
                if not config_path.exists():
                    installed_dirs = [
                        Path('/usr/share/aisbf'),
                        Path.home() / '.local' / 'share' / 'aisbf',
                    ]
                    for installed_dir in installed_dirs:
                        test_path = installed_dir / 'aisbf.json'
                        if test_path.exists():
                            config_path = test_path
                            break
                    else:
                        config_path = Path(__file__).parent.parent / 'config' / 'aisbf.json'

                model_name = "huihui-ai/Qwen2.5-0.5B-Instruct-abliterated-v3"
                if config_path.exists():
                    try:
                        with open(config_path) as f:
                            aisbf_cfg = json.load(f)
                            model_name = aisbf_cfg.get('internal_model', {}).get('autoselect_model_id', model_name)
                    except Exception as e:
                        logger.warning(f"Error loading autoselect model config: {e}, using default")

                logger.info(f"Model: {model_name}")
                device = "cuda" if torch.cuda.is_available() else "cpu"
                logger.info(f"Device: {device}")

                logger.info("Loading tokenizer...")
                try:
                    _internal_tokenizer_singleton = AutoTokenizer.from_pretrained(model_name, local_files_only=True)
                    logger.info("Tokenizer loaded from local cache")
                except (OSError, EnvironmentError):
                    logger.info("Tokenizer not cached, downloading from HuggingFace...")
                    _internal_tokenizer_singleton = AutoTokenizer.from_pretrained(model_name)
                    logger.info("Tokenizer downloaded and cached")

                logger.info("Loading model...")
                try:
                    _internal_model_singleton = AutoModelForCausalLM.from_pretrained(
                        model_name,
                        dtype=torch.float16 if device == "cuda" else torch.float32,
                        device_map="auto" if device == "cuda" else None,
                        local_files_only=True
                    )
                    logger.info("Model loaded from local cache")
                except (OSError, EnvironmentError):
                    logger.info("Model not cached, downloading from HuggingFace...")
                    _internal_model_singleton = AutoModelForCausalLM.from_pretrained(
                        model_name,
                        dtype=torch.float16 if device == "cuda" else torch.float32,
                        device_map="auto" if device == "cuda" else None
                    )
                    logger.info("Model downloaded and cached")

                if device == "cpu":
                    _internal_model_singleton = _internal_model_singleton.to(device)

                logger.info("=== INTERNAL SELECTION MODEL READY ===")
            except ImportError as e:
                logger.error(f"Failed to import required libraries for internal model: {e}")
                logger.error("Please install: pip install torch transformers")
                raise
            except Exception as e:
                logger.error(f"Failed to initialize internal model: {e}", exc_info=True)
                raise
    
    async def _run_internal_model_selection(self, messages: list) -> str:
        """Run the shared internal instruct model for selection.

        Inference is submitted to a module-level single-worker executor so concurrent
        requests are automatically queued — only one inference runs at a time without
        any additional locking.  Initialization is protected separately by
        _internal_model_init_lock (double-checked locking).
        """
        global _internal_model_singleton, _internal_tokenizer_singleton
        import logging
        logger = logging.getLogger(__name__)

        if _internal_model_singleton is None:
            self._initialize_internal_model()

        def run_inference():
            try:
                import torch

                has_chat_template = (
                    getattr(_internal_tokenizer_singleton, 'chat_template', None) is not None
                )
                if has_chat_template:
                    logger.info("[internal] Instruct model — applying chat template")
                    prompt = _internal_tokenizer_singleton.apply_chat_template(
                        messages,
                        tokenize=False,
                        add_generation_prompt=True
                    )
                else:
                    logger.info("[internal] Base model — using raw prompt")
                    prompt = "\n\n".join(m["content"] for m in messages) + "\n"
                logger.info(f"[internal] Prompt length: {len(prompt)} chars")
                logger.debug(f"[internal] Full prompt:\n{prompt}")

                inputs = _internal_tokenizer_singleton(prompt, return_tensors="pt")
                input_length = inputs['input_ids'].shape[1]
                logger.info(f"[internal] Tokenized input: {input_length} tokens")

                aisbf_conf = self.config.get_aisbf_config()
                _max_input_tokens = int(
                    (aisbf_conf.internal_model or {}).get('autoselect_max_tokens', 8000)
                    if aisbf_conf else 8000
                )
                if input_length > _max_input_tokens:
                    logger.info(f"[internal] Truncating input from {input_length} to {_max_input_tokens} tokens")
                    inputs = {k: v[:, -_max_input_tokens:] for k, v in inputs.items()}
                    input_length = _max_input_tokens

                device = next(_internal_model_singleton.parameters()).device
                logger.info(f"[internal] Running inference on device: {device}")
                inputs = {k: v.to(device) for k, v in inputs.items()}

                aisbf_conf = self.config.get_aisbf_config()
                _max_new_tokens = int(
                    (aisbf_conf.internal_model or {}).get('autoselect_max_new_tokens', 100)
                    if aisbf_conf else 100
                )
                with torch.no_grad():
                    outputs = _internal_model_singleton.generate(
                        **inputs,
                        max_new_tokens=_max_new_tokens,
                        do_sample=False,
                        pad_token_id=_internal_tokenizer_singleton.eos_token_id
                    )

                generated_length = outputs[0].shape[0] - input_length
                logger.info(f"[internal] Generated {generated_length} new tokens")

                response = _internal_tokenizer_singleton.decode(
                    outputs[0][input_length:], skip_special_tokens=True
                ).strip()
                logger.info(f"[internal] Decoded response: {repr(response)}")
                return response
            except Exception as e:
                logger.error(f"[internal] Inference error: {e}", exc_info=True)
                return None

        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(_internal_model_executor, run_inference)


    def _compact_messages_for_selection(self, messages: List[Dict], max_tokens: int) -> str:
        """
        Compact a message list into a prompt string that fits within max_tokens.
        Keeps the first HEAD and last TAIL messages verbatim; the dropped middle
        section is replaced by a compact one-liner summary so no context is lost silently.

        Individual messages that are too large are truncated so the HEAD/TAIL
        content always fits within the budget even when there are fewer messages
        than HEAD+TAIL.
        """
        def _tokens(text: str) -> int:
            return len(text) // 4

        def _msg_text(msg: dict, max_content_chars: int = None) -> str:
            role = msg.get('role', 'user')
            content = msg.get('content') or ''
            if isinstance(content, list):
                content = str(content)
            if max_content_chars is not None and len(content) > max_content_chars:
                content = content[:max_content_chars] + "...[truncated]"
            return f"{role}: {content}\n"

        if not messages:
            return ""

        full = "".join(_msg_text(m) for m in messages)
        if _tokens(full) <= max_tokens:
            return full

        HEAD = 2  # first N messages to keep verbatim
        TAIL = 3  # last N messages to keep verbatim
        max_chars = max_tokens * 4

        head = messages[:HEAD]
        tail = messages[-TAIL:] if len(messages) > HEAD + TAIL else []
        middle = messages[HEAD: len(messages) - TAIL] if len(messages) > HEAD + TAIL else messages[HEAD:]

        # Budget each kept message so head+tail together never exceed the limit.
        kept_count = len(head) + len(tail)
        if kept_count > 0:
            # Reserve ~10 % for the summary block; split the rest equally.
            per_msg_budget_chars = max(256, (max_chars * 9 // 10) // kept_count)
        else:
            per_msg_budget_chars = max_chars

        head_text = "".join(_msg_text(m, per_msg_budget_chars) for m in head)
        tail_text = "".join(_msg_text(m, per_msg_budget_chars) for m in tail)
        fixed_tokens = _tokens(head_text) + _tokens(tail_text)

        summary_block = ""
        if middle:
            snippets = [
                f"{m.get('role','?')}: {str(m.get('content','')).replace(chr(10), ' ').strip()[:80]}"
                for m in middle
            ]
            summary_line = "; ".join(snippets)
            summary_block = f"[... {len(middle)} omitted messages — summary: {summary_line} ...]\n"
            available_for_summary = max_tokens - fixed_tokens
            if _tokens(summary_block) > available_for_summary > 0:
                summary_block = summary_block[:available_for_summary * 4 - 5] + "...]\n"

        result = head_text + summary_block + tail_text

        # Final safety cap: never exceed the hard limit regardless of edge cases.
        if len(result) > max_chars:
            result = result[:max_chars - 14] + "...[truncated]"

        return result

    @staticmethod
    def _detect_loop(text: str, ngram: int = 8, threshold: int = 3) -> bool:
        """
        Return True if `text` contains a repeated phrase indicating a generation loop.
        Splits into word-level n-grams and flags if any n-gram appears >= threshold times.
        """
        if not text:
            return False
        words = text.split()
        if len(words) < ngram * threshold:
            return False
        counts: dict = {}
        for i in range(len(words) - ngram + 1):
            key = tuple(words[i:i + ngram])
            counts[key] = counts.get(key, 0) + 1
            if counts[key] >= threshold:
                return True
        return False

    # ------------------------------------------------------------------
    # Incremental conversation summary helpers
    # ------------------------------------------------------------------

    def _msg_fingerprint(self, msg: dict) -> str:
        """Stable MD5 of a message's content — used to detect identical message prefixes."""
        c = msg.get('content', '')
        if isinstance(c, list):
            c = ' '.join(p.get('text', '') if isinstance(p, dict) else str(p) for p in c)
        return hashlib.md5(str(c).encode()).hexdigest()

    def _get_session_key(self, context_msgs: list) -> Optional[str]:
        """Derive a stable conversation-level key from the first 1-2 messages.

        These messages (usually the system prompt + first user turn) are constant
        for the lifetime of a conversation, giving a reliable session identifier.
        """
        if not context_msgs:
            return None
        parts = []
        for m in context_msgs[:2]:
            c = m.get('content', '')
            if isinstance(c, list):
                c = ' '.join(p.get('text', '') if isinstance(p, dict) else str(p) for p in c)
            parts.append(str(c)[:500])
        return hashlib.md5('||'.join(parts).encode()).hexdigest()

    def _find_conversation_summary(self, context_msgs: list):
        """Look for a cached summary whose fingerprints are a prefix of context_msgs.

        Returns (summary_dict, msgs_since_summary) if found, or (None, context_msgs).
        A prefix match means the conversation grew by appending new messages — the
        summarized part is identical to what was seen before.
        """
        global _conversation_summaries
        session_key = self._get_session_key(context_msgs)
        if not session_key:
            return None, context_msgs

        entry = _conversation_summaries.get(session_key)
        if not entry:
            return None, context_msgs

        if time_module.time() - entry['created_at'] > _conversation_summary_ttl:
            _conversation_summaries.pop(session_key, None)
            return None, context_msgs

        stored_fps = entry['fingerprints']
        K = len(stored_fps)
        if K > len(context_msgs):
            return None, context_msgs

        # Verify that stored fingerprints match the first K messages exactly
        current_fps = [self._msg_fingerprint(m) for m in context_msgs[:K]]
        if current_fps != stored_fps:
            return None, context_msgs  # conversation was edited — treat as new

        return entry, context_msgs[K:]

    def _build_summary_context_prompt(self, summary: dict, new_msgs: list, max_tokens: int) -> str:
        """Build context_prompt from a cached summary + messages added since that summary."""
        lines = [
            "[PRIOR CONTEXT SUMMARY]",
            summary['summary_text'],
        ]
        nsfw = summary.get('nsfw')
        pd   = summary.get('personal_data')
        prev = summary.get('previous_selection')
        lines.append(f"NSFW content: {'detected' if nsfw else 'not detected' if nsfw is False else 'not analyzed'}")
        lines.append(f"Personal data: {'detected' if pd else 'not detected' if pd is False else 'not analyzed'}")
        if prev:
            lines.append(f"Previous model selection: {prev}")

        if new_msgs:
            lines.append("\n[NEW MESSAGES SINCE SUMMARY]")
            # Budget the new messages portion: summary is already compact, give remaining tokens to new msgs
            header_chars = sum(len(l) for l in lines)
            remaining_tokens = max(500, max_tokens - header_chars // 4)
            lines.append(self._compact_messages_for_selection(new_msgs, remaining_tokens))

        return '\n'.join(lines)

    async def _store_conversation_summary(self, context_msgs: list, model_id: Optional[str],
                                          autoselect_config=None, context_text: str = ''):
        """Build and cache a compact conversation summary for the next autoselect turn.

        Classification (NSFW / personal-data) only runs when the classifier is already
        loaded in memory — no model download penalty.  Classification runs in the shared
        executor so it never blocks the event loop.
        """
        global _conversation_summaries
        if not context_msgs:
            return
        session_key = self._get_session_key(context_msgs)
        if not session_key:
            return

        fingerprints = [self._msg_fingerprint(m) for m in context_msgs]

        # Compact summary text: HEAD messages + message count
        summary_lines = []
        for m in context_msgs[:2]:
            role = m.get('role', 'user')
            c = m.get('content', '')
            if isinstance(c, list):
                c = ' '.join(p.get('text', '') if isinstance(p, dict) else str(p) for p in c)
            summary_lines.append(f"{role}: {str(c).strip()[:400]}")
        if len(context_msgs) > 2:
            summary_lines.append(f"[{len(context_msgs) - 2} additional messages in session]")
        summary_text = '\n'.join(summary_lines)

        classify_text = (context_text or summary_text)[:512]

        # NSFW / privacy — only if classifier is already initialised (fast path)
        nsfw: Optional[bool] = None
        personal_data: Optional[bool] = None

        loop = asyncio.get_running_loop()

        if autoselect_config and getattr(autoselect_config, 'classify_nsfw', False) \
                and content_classifier._nsfw_classifier is not None:
            try:
                is_safe, _ = await loop.run_in_executor(
                    None, content_classifier.check_nsfw, classify_text
                )
                nsfw = not is_safe
            except Exception:
                pass

        if autoselect_config and getattr(autoselect_config, 'classify_privacy', False) \
                and content_classifier._privacy_classifier is not None:
            try:
                is_safe, _ = await loop.run_in_executor(
                    None, content_classifier.check_privacy, classify_text
                )
                personal_data = not is_safe
            except Exception:
                pass

        _conversation_summaries[session_key] = {
            'fingerprints': fingerprints,
            'summary_text': summary_text,
            'nsfw': nsfw,
            'personal_data': personal_data,
            'previous_selection': model_id,
            'created_at': time_module.time(),
        }

    def _get_selection_max_tokens(self, autoselect_config) -> int:
        """Return the max-tokens budget for compacting messages before autoselection.

        - internal model   → aisbf_config.internal_model.autoselect_max_tokens (default 8000)
        - rotation         → rotation.default_context_size  (fallback 8000)
        - provider/model   → model.context_length  (fallback 8000)
        """
        selection_model = (getattr(autoselect_config, 'selection_model', None) or '').strip() or 'internal'

        if selection_model == 'internal':
            aisbf_conf = self.config.get_aisbf_config()
            if aisbf_conf and aisbf_conf.internal_model:
                return int(aisbf_conf.internal_model.get('autoselect_max_tokens', 8000))
            return 8000

        # Rotation: look in user rotations first, then global
        rotation_cfg = None
        if self.user_id and hasattr(self, 'user_rotations'):
            rotation_cfg = next(
                (r['config'] for r in self.user_rotations if r['rotation_id'] == selection_model),
                None
            )
        if rotation_cfg is None and hasattr(self.config, 'rotations'):
            rotation_cfg = self.config.rotations.get(selection_model)
        if rotation_cfg and getattr(rotation_cfg, 'default_context_size', None):
            return rotation_cfg.default_context_size

        # Direct provider/model
        if '/' in selection_model:
            provider_id, model_name = selection_model.split('/', 1)
            if hasattr(self.config, 'providers') and provider_id in self.config.providers:
                prov_cfg = self.config.get_provider(provider_id)
                for m in (getattr(prov_cfg, 'models', None) or []):
                    if m.name == model_name and getattr(m, 'context_length', None):
                        return m.context_length

        return 8000

    def _build_autoselect_messages(self, context_prompt: str, current_task: str, autoselect_config, failed_models: Optional[List[str]] = None) -> List[Dict]:
        """Build the messages for model selection (system + user).

        context_prompt  — compacted prior conversation (establishes domain/topic)
        current_task    — the last user message, i.e. the specific thing to do NOW
        """
        skill_content = self._get_skill_file_content()

        # Build the available models list, excluding already-failed ones
        models_list = ""
        for model_info in autoselect_config.available_models:
            if failed_models and model_info.model_id in failed_models:
                continue
            models_list += f"<model><model_id>{model_info.model_id}</model_id><model_description>{model_info.description}</model_description></model>\n"

        system_message = skill_content

        failed_note = ""
        if failed_models:
            failed_note = f"\n<aisbf_failed_models>{', '.join(failed_models)}</aisbf_failed_models>\n<aisbf_note>The above models have already failed or produced looping responses. Do NOT select them.</aisbf_note>"

        context_block = f"<aisbf_session_context>\n{context_prompt}\n</aisbf_session_context>\n" if context_prompt.strip() else ""

        user_message = f"""{context_block}<aisbf_current_task>{current_task}</aisbf_current_task>
<aisbf_autoselect_list>
{models_list}
</aisbf_autoselect_list>
<aisbf_autoselect_fallback>{autoselect_config.fallback}</aisbf_autoselect_fallback>{failed_note}"""

        return [
            {"role": "system", "content": system_message},
            {"role": "user", "content": user_message}
        ]

    def _extract_model_selection(self, response: str) -> Optional[str]:
        """Extract the model_id from the autoselection response"""
        match = re.search(r'<aisbf_model_autoselection>(.*?)</aisbf_model_autoselection>', response, re.DOTALL)
        if match:
            return match.group(1).strip()
        return None

    async def _get_model_selection(self, context_prompt: str, current_task: str, autoselect_config, failed_models: Optional[List[str]] = None) -> str:
        """Cache + coalescing wrapper around _run_selection.

        1. Returns immediately on a cache hit.
        2. If an identical selection is already in-flight, awaits its Future instead of
           making a redundant API/inference call (single-flight deduplication).
        3. Otherwise runs the selection, stores the result, and resolves all waiters.
        """
        import logging
        logger = logging.getLogger(__name__)
        logger.info("=== AUTOSELECT MODEL SELECTION START ===")

        global _autoselect_result_cache, _autoselect_inflight

        _available_ids = sorted(
            m.model_id for m in autoselect_config.available_models
            if not failed_models or m.model_id not in failed_models
        )
        _sel_model_key = (getattr(autoselect_config, 'selection_model', None) or 'internal') or 'internal'
        _cache_key = (
            hashlib.md5(context_prompt.encode()).hexdigest()
            + "|" + hashlib.md5(current_task.encode()).hexdigest()
            + "|" + hashlib.md5(",".join(_available_ids).encode()).hexdigest()
            + "|" + _sel_model_key
        )
        _now = time_module.time()

        # 1. Cache hit
        _cached = _autoselect_result_cache.get(_cache_key)
        if _cached:
            _cached_id, _cached_at = _cached
            if _now - _cached_at < _autoselect_result_cache_ttl:
                logger.info(f"=== AUTOSELECT CACHE HIT (age {_now - _cached_at:.0f}s) → {_cached_id} ===")
                return _cached_id

        # 2. Coalesce: await an identical in-flight selection rather than duplicating the call.
        #    Safe without a lock because asyncio is single-threaded — the check and the
        #    dict assignment below are both in the same synchronous segment (no await between).
        if _cache_key in _autoselect_inflight:
            logger.info("=== AUTOSELECT COALESCING: awaiting identical in-flight selection ===")
            try:
                result = await _autoselect_inflight[_cache_key]
                logger.info(f"=== AUTOSELECT COALESCED RESULT → {result} ===")
                return result
            except Exception as e:
                logger.warning(f"In-flight selection failed ({e}), running independent selection")

        # 3. Register as in-flight, run, then resolve all waiters atomically.
        _future = asyncio.get_running_loop().create_future()
        _autoselect_inflight[_cache_key] = _future
        try:
            model_id = await self._run_selection(context_prompt, current_task, autoselect_config, failed_models)
            if model_id:
                _autoselect_result_cache[_cache_key] = (model_id, _now)
            _future.set_result(model_id)
            return model_id
        except Exception as e:
            if not _future.done():
                _future.set_exception(e)
            raise
        finally:
            _autoselect_inflight.pop(_cache_key, None)

    async def _run_selection(self, context_prompt: str, current_task: str, autoselect_config, failed_models: Optional[List[str]] = None) -> Optional[str]:
        """Run the actual model selection — no caching, no coalescing.  Called only by _get_model_selection."""
        import logging
        logger = logging.getLogger(__name__)

        # For text/semantic search we combine both parts
        full_prompt = (context_prompt + "\n" + current_task).strip()

        # Check if semantic classification is enabled
        if autoselect_config.classify_semantic:
            logger.info("=== SEMANTIC CLASSIFICATION ENABLED ===")

            try:
                semantic_classifier = SemanticClassifier()
                semantic_classifier.initialize()

                model_library = {}
                for model_info in autoselect_config.available_models:
                    if failed_models and model_info.model_id in failed_models:
                        continue
                    model_library[model_info.model_id] = model_info.description

                chat_history = [current_task] if current_task else []
                if context_prompt:
                    lines = context_prompt.strip().split('\n')
                    for line in lines[-3:]:
                        if ': ' in line:
                            _, content = line.split(': ', 1)
                            chat_history.append(content)

                results = semantic_classifier.hybrid_model_search(full_prompt, chat_history, model_library, top_k=1)

                if results:
                    selected_model_id, score = results[0]
                    logger.info(f"=== SEMANTIC CLASSIFICATION SUCCESS === Selected: {selected_model_id} (score: {score:.4f})")
                    return selected_model_id
                else:
                    logger.warning("Semantic search returned no results, falling back to AI model selection")

            except Exception as e:
                logger.error(f"Semantic classification error: {e}")
                logger.warning("Falling back to AI model selection")

        logger.info(f"Using '{autoselect_config.selection_model}' for model selection")

        messages = self._build_autoselect_messages(context_prompt, current_task, autoselect_config, failed_models)

        selection_request = {
            "messages": messages,
            "temperature": 0,
            "max_tokens": 100,
            "stream": False,
            "stop": ["</aisbf_model_autoselection>"],
            # Carry the delegation chain so a selection-model rotation cannot loop back.
            "_delegation_chain": list(getattr(self, '_delegation_chain', []) or []),
        }

        selection_model = (getattr(autoselect_config, 'selection_model', None) or '').strip() or 'internal'

        try:
            if selection_model == "internal":
                aisbf_conf = self.config.get_aisbf_config()
                # Global override: route the selection step through a rotation/autoselect
                # instead of the local model, falling back to local on any failure.
                autoselect_override = (
                    (aisbf_conf.internal_model or {}).get('autoselect_override', '')
                    if aisbf_conf else ''
                )
                if (autoselect_override or '').strip():
                    logger.info(f"Autoselect selection using override target '{autoselect_override}'")
                    override_content = await run_meta_target(
                        autoselect_override, messages, max_tokens=100, temperature=0.0,
                        stop=["</aisbf_model_autoselection>"]
                    )
                    if override_content:
                        model_id = self._extract_model_selection(override_content)
                        if model_id:
                            logger.info(f"=== AUTOSELECT SUCCESS (override) === {model_id}")
                            return model_id
                        logger.warning("Could not extract model ID from override response; falling back to local model")
                    else:
                        logger.warning(f"Autoselect override '{autoselect_override}' unavailable; falling back to local model")
                internal_model_id = (
                    (aisbf_conf.internal_model or {}).get('autoselect_model_id', '')
                    if aisbf_conf else ''
                )
                if internal_model_id and '/' in internal_model_id:
                    internal_provider_id = internal_model_id.split('/', 1)[0]
                    if internal_provider_id in self.config.providers:
                        logger.info(f"Internal autoselect model '{internal_model_id}' → provider '{internal_provider_id}'")
                        selection_request['model'] = internal_model_id.split('/', 1)[1]
                        response = await RequestHandler().handle_chat_completion(
                            request=None, provider_id=internal_provider_id, request_data=selection_request
                        )
                        content = response.get('choices', [{}])[0].get('message', {}).get('content', '')
                        model_id = self._extract_model_selection(content)
                        if model_id:
                            logger.info(f"=== AUTOSELECT SUCCESS === {model_id}")
                        else:
                            logger.warning("Could not extract model ID from provider response")
                        return model_id

                logger.info("Selection model is 'internal' — using local HuggingFace model")
                response_content = await self._run_internal_model_selection(messages)
                if not response_content:
                    logger.error("Internal model returned no response")
                    return None
                logger.info(f"Internal model response: {response_content[:200]}..." if len(response_content) > 200 else f"Internal model response: {response_content}")
                model_id = self._extract_model_selection(response_content)
                if model_id:
                    logger.info(f"=== AUTOSELECT SUCCESS === {model_id}")
                else:
                    logger.warning("Could not extract model ID from internal model response")
                return model_id

            elif (self.user_id and hasattr(self, 'rotations') and selection_model in self.rotations) or selection_model in self.config.rotations:
                logger.info(f"Selection model '{selection_model}' is a rotation")
                response = await RotationHandler(user_id=self.user_id).handle_rotation_request(selection_model, selection_request)

            elif '/' in selection_model:
                provider_id, model_name = selection_model.split('/', 1)
                logger.info(f"Selection model '{selection_model}' is a direct provider model")
                if provider_id not in self.config.providers:
                    logger.error(f"Selection model provider '{provider_id}' not found")
                    return None
                selection_request['model'] = model_name
                response = await RequestHandler().handle_chat_completion(
                    request=None, provider_id=provider_id, request_data=selection_request
                )

            elif selection_model in self.config.providers:
                logger.info(f"Selection model '{selection_model}' is a provider")
                provider_config = self.config.get_provider(selection_model)
                models = getattr(provider_config, "models", []) or []
                if not models:
                    logger.error(f"Provider '{selection_model}' has no models configured")
                    return None
                selection_request['model'] = models[0].name
                response = await RequestHandler().handle_chat_completion(
                    request=None, provider_id=selection_model, request_data=selection_request
                )

            else:
                logger.error(f"Selection model '{selection_model}' not found in rotations or providers")
                return None

            content = response.get('choices', [{}])[0].get('message', {}).get('content', '')
            logger.info(f"Raw response: {content[:200]}..." if len(content) > 200 else f"Raw response: {content}")
            model_id = self._extract_model_selection(content)
            if model_id:
                logger.info(f"=== AUTOSELECT SUCCESS === {model_id}")
            else:
                logger.warning(f"Could not extract model ID from response: {content}")
            return model_id

        except Exception as e:
            logger.error(f"=== AUTOSELECT SELECTION ERROR === {e}")
            return None

    def _apply_autoselect_max_tokens(self, proxied_request_data: Dict, autoselect_config, selected_model_id: str, available_models) -> None:
        """Apply autoselect-level max output token settings to a proxied request.

        Only acts when the client did not specify max_tokens. Priority:
        - per-model max_tokens on the selected available_model (highest) -> set directly
        - autoselect default_max_tokens -> threaded as a lowest-priority fallback
          (``_autoselect_default_max_tokens``) so the downstream rotation/provider
          resolution (rotation per-model, provider per-model, provider default,
          rotation default) still wins over it.
        """
        if proxied_request_data.get('max_tokens') is not None:
            return
        model_info = next((m for m in (available_models or []) if getattr(m, 'model_id', None) == selected_model_id), None)
        per_model = getattr(model_info, 'max_tokens', None) if model_info else None
        if per_model:
            proxied_request_data['max_tokens'] = per_model
            return
        autoselect_default = getattr(autoselect_config, 'default_max_tokens', None)
        if autoselect_default:
            proxied_request_data['_autoselect_default_max_tokens'] = autoselect_default

    async def handle_autoselect_request(self, autoselect_id: str, request_data: Dict, user_id: Optional[int] = None, token_id: Optional[int] = None) -> Dict:
        """Handle an autoselect request"""
        import logging
        logger = logging.getLogger(__name__)
        logger.info(f"=== AUTOSELECT REQUEST START ===")
        logger.info(f"Autoselect ID: {autoselect_id}")
        logger.info(f"User ID: {self.user_id}")

        # Delegation loop protection (shared with rotations): register this
        # autoselect in the chain so a rotation/autoselect cannot recurse into it.
        self._register_delegation(request_data, f"autoselect:{autoselect_id}")

        # Check response cache for non-streaming requests
        stream = request_data.get('stream', False)
        if not stream:
            try:
                current_autoselect_config = self.autoselects.get(autoselect_id) if self.user_id else self.config.get_autoselect(autoselect_id)
                if self._should_cache_response(autoselect_config=current_autoselect_config):
                    response_cache = self._get_response_cache_backend()
                    if response_cache is None:
                        raise RuntimeError("Response cache backend unavailable")
                    cached_response = response_cache.get(request_data, user_id=self.user_id)
                    if cached_response:
                        logger.info(f"Cache hit for autoselect request {autoselect_id}")
                        return cached_response
                    else:
                        logger.debug(f"Cache miss for autoselect request {autoselect_id}")
            except Exception as cache_error:
                logger.warning(f"Response cache check failed: {cache_error}")

        # Check for user-specific autoselect config first
        if self.user_id:
            # Database user: ONLY use user-specific configs - NO global fallback
            autoselect_config = next((aut['config'] for aut in self.user_autoselects if aut['autoselect_id'] == autoselect_id), None)
            if autoselect_config:
                logger.info(f"Using user-specific autoselect config for {autoselect_id}")
            else:
                logger.error(f"User autoselect {autoselect_id} not found - NO global fallback")
                raise HTTPException(status_code=400, detail=f"Autoselect {autoselect_id} not found for this user")
        else:
            # Admin user: use global config
            autoselect_config = self.config.get_autoselect(autoselect_id)
            logger.info(f"Using global autoselect config for {autoselect_id}")

        if not autoselect_config:
            logger.error(f"Autoselect {autoselect_id} not found")
            raise HTTPException(status_code=400, detail=f"Autoselect {autoselect_id} not found")

        logger.info(f"Autoselect config loaded")
        logger.info(f"Available models for selection: {len(autoselect_config.available_models)}")
        for model_info in autoselect_config.available_models:
            logger.info(f"  - {model_info.model_id}: {model_info.description}")
        logger.info(f"Selection model: {autoselect_config.selection_model}")
        logger.info(f"Fallback model: {autoselect_config.fallback}")

        # Extract the user prompt from the request
        user_messages = request_data.get('messages', [])
        if not user_messages:
            logger.error("No messages provided")
            raise HTTPException(status_code=400, detail="No messages provided")

        logger.info(f"User messages count: {len(user_messages)}")

        # Split: last TAIL messages → current_task (what's happening now)
        #        earlier messages  → context_prompt (session domain/background)
        _TAIL = 5
        tail_msgs = user_messages[-_TAIL:]
        context_msgs = user_messages[:-_TAIL] if len(user_messages) > _TAIL else []
        _task_parts = []
        for _m in tail_msgs:
            _role = _m.get('role', 'user')
            _c = _m.get('content', '')
            if isinstance(_c, list):
                _c = ' '.join((p.get('text', '') if isinstance(p, dict) else str(p)) for p in _c).strip()
            _c = str(_c).strip()
            if len(_c) > 800:
                _c = _c[:800] + "...[truncated]"
            _task_parts.append(f"{_role}: {_c}")
        current_task = "\n".join(_task_parts)

        max_selection_tokens = self._get_selection_max_tokens(autoselect_config)

        # Try to reuse a cached conversation summary (incremental context)
        _summary, _new_msgs = self._find_conversation_summary(context_msgs)
        if _summary:
            context_prompt = self._build_summary_context_prompt(_summary, _new_msgs, max_selection_tokens)
            logger.info(f"Using cached conversation summary ({len(_summary['fingerprints'])} msgs summarized, {len(_new_msgs)} new)")
        else:
            context_prompt = self._compact_messages_for_selection(context_msgs, max_selection_tokens) if context_msgs else ""
            logger.info(f"No cached summary — full context compaction")

        logger.info(f"Context: {len(context_prompt)} chars; current task: {len(current_task)} chars ({len(tail_msgs)} tail msgs, limit {max_selection_tokens} tokens)")
        logger.info(f"Current task preview: {current_task[:300]}..." if len(current_task) > 300 else f"Current task: {current_task}")

        # Filter out entries with empty model_id (misconfiguration guard)
        valid_models = [m for m in autoselect_config.available_models if (m.model_id or '').strip()]
        invalid_count = len(autoselect_config.available_models) - len(valid_models)
        if invalid_count:
            logger.warning(f"Autoselect '{autoselect_id}': ignoring {invalid_count} available_model(s) with empty model_id")

        # Build escalation order: sort available models by priority ascending (lower priority first)
        available_models_ordered = sorted(valid_models, key=lambda m: m.priority)
        rotation_handler = RotationHandler()
        failed_models: List[str] = []
        last_exception = None
        _summary_stored = False
        _selection_latency_ms = 0.0

        while True:
            # Re-run selection excluding already-failed models
            logger.info(f"Requesting model selection from AI (failed so far: {failed_models})...")
            _sel_start = time.time()
            selected_model_id = await self._get_model_selection(context_prompt, current_task, autoselect_config, failed_models)
            _selection_latency_ms = (time.time() - _sel_start) * 1000

            # Store/update conversation summary after the first selection (fire-and-forget)
            if not _summary_stored:
                _summary_stored = True
                asyncio.create_task(self._store_conversation_summary(
                    context_msgs, selected_model_id, autoselect_config, context_prompt
                ))

            # Validate
            available_ids = [m.model_id for m in available_models_ordered if m.model_id not in failed_models]
            if not selected_model_id or selected_model_id not in available_ids:
                # Pick the lowest-priority untried model as fallback
                untried = [m.model_id for m in available_models_ordered if m.model_id not in failed_models]
                if not untried:
                    logger.error("All models have failed, no escalation target available")
                    if last_exception:
                        raise last_exception
                    raise HTTPException(status_code=502, detail="All autoselect models failed")
                selected_model_id = untried[0]
                logger.warning(f"Selection invalid, using next untried model: {selected_model_id}")
            else:
                logger.info(f"Selected model: {selected_model_id}")

            logger.info(f"Proxying request to: {selected_model_id}")
            try:
                proxied_request_data = dict(request_data)
                proxied_request_data['_autoselect_id'] = autoselect_id
                self._apply_autoselect_max_tokens(proxied_request_data, autoselect_config, selected_model_id, available_models_ordered)
                response = await rotation_handler.handle_rotation_request(selected_model_id, proxied_request_data, user_id, token_id)
            except Exception as e:
                logger.warning(f"Model '{selected_model_id}' raised exception: {e} — escalating")
                failed_models.append(selected_model_id)
                last_exception = e
                untried = [m.model_id for m in available_models_ordered if m.model_id not in failed_models]
                if not untried:
                    logger.error("All models exhausted after exception")
                    raise
                continue

            # Check if the rotation itself reported a full failure (all its internal models failed)
            response_dict = response if isinstance(response, dict) else (response.body and __import__('json').loads(response.body) if hasattr(response, 'body') else {})
            if isinstance(response_dict, dict) and response_dict.get('aisbf_error'):
                logger.warning(f"Rotation '{selected_model_id}' fully failed (aisbf_error) — escalating")
                failed_models.append(selected_model_id)
                untried = [m.model_id for m in available_models_ordered if m.model_id not in failed_models]
                if not untried:
                    logger.warning("All autoselect models exhausted after rotation failure, returning last response")
                    break
                continue

            # Loop detection on non-streaming response
            if isinstance(response, dict):
                content = response.get('choices', [{}])[0].get('message', {}).get('content', '') or ''
            else:
                try:
                    content = response.choices[0].message.content or ''
                except (AttributeError, IndexError):
                    content = ''
            if not content:
                logger.warning(f"Model '{selected_model_id}' returned empty content — escalating")
                failed_models.append(selected_model_id)
                untried = [m.model_id for m in available_models_ordered if m.model_id not in failed_models]
                if not untried:
                    logger.warning("All models exhausted after empty response, returning last response")
                    break
                continue

            if self._detect_loop(content):
                logger.warning(f"Model '{selected_model_id}' detected looping response — escalating")
                failed_models.append(selected_model_id)
                untried = [m.model_id for m in available_models_ordered if m.model_id not in failed_models]
                if not untried:
                    logger.warning("All models exhausted after loop detection, returning last response")
                    break
                continue

            # Good response
            break

        # Cache the response for non-streaming requests
        if not stream:
            try:
                if self._should_cache_response(autoselect_config=autoselect_config):
                    response_cache = self._get_response_cache_backend()
                    if response_cache is None:
                        raise RuntimeError("Response cache backend unavailable")
                    response_cache.set(request_data, response, user_id=self.user_id)
                    logger.debug(f"Cached response for autoselect request {autoselect_id}")
            except Exception as cache_error:
                logger.warning(f"Response cache set failed: {cache_error}")

        # Record analytics with autoselect_id so it appears in the autoselect column
        try:
            analytics = get_analytics()
            if response and isinstance(response, dict):
                usage = response.get('usage', {})
                total_tokens = usage.get('total_tokens', 0)
                prompt_tokens = usage.get('prompt_tokens', 0)
                completion_tokens = usage.get('completion_tokens', 0)
                if total_tokens == 0:
                    try:
                        messages = request_data.get('messages', [])
                        prompt_tokens = count_messages_tokens(messages, selected_model_id)
                        response_content = response.get('choices', [{}])[0].get('message', {}).get('content', '')
                        completion_tokens = count_messages_tokens([{"role": "assistant", "content": response_content}], selected_model_id) if response_content else 0
                        total_tokens = prompt_tokens + completion_tokens
                    except Exception:
                        total_tokens = 150
                        prompt_tokens = 0
                        completion_tokens = 0
                analytics.record_request(
                    provider_id='autoselect',
                    model_name=selected_model_id,
                    tokens_used=total_tokens,
                    latency_ms=_selection_latency_ms,
                    success=True,
                    autoselect_id=autoselect_id,
                    user_id=user_id,
                    token_id=token_id,
                    prompt_tokens=prompt_tokens if prompt_tokens > 0 else None,
                    completion_tokens=completion_tokens if completion_tokens > 0 else None,
                    actual_cost=None,
                    analytics_kind='selector'
                )
        except Exception as analytics_error:
            logger.warning(f"Analytics recording for autoselect failed: {analytics_error}")

        logger.info(f"=== AUTOSELECT REQUEST END ===")
        if isinstance(response, dict):
            selected_provider = None
            if isinstance(selected_model_id, str) and '/' in selected_model_id:
                selected_provider = selected_model_id.split('/', 1)[0]
            try:
                if selected_provider:
                    self._settle_market_result(selected_provider, response.get('usage', {}), 1, {'kind': 'autoselect', 'autoselect_id': autoselect_id}, request_data=request_data)
            except Exception as market_error:
                logger.warning(f"Market settlement failed for autoselect {autoselect_id}: {market_error}")
        return response

    async def handle_autoselect_streaming_request(self, autoselect_id: str, request_data: Dict):
        """
        Handle an autoselect streaming request.

        The rotation handler handles the streaming conversion internally based on
        the selected provider's type, so we just pass through the response.
        """
        import logging
        logger = logging.getLogger(__name__)
        logger.info(f"=== AUTOSELECT STREAMING REQUEST START ===")
        logger.info(f"Autoselect ID: {autoselect_id}")

        # Delegation loop protection (shared with rotations).
        self._register_delegation(request_data, f"autoselect:{autoselect_id}")

        autoselect_config = self.config.get_autoselect(autoselect_id)
        if not autoselect_config:
            logger.error(f"Autoselect {autoselect_id} not found")
            raise HTTPException(status_code=400, detail=f"Autoselect {autoselect_id} not found")

        logger.info(f"Autoselect config loaded")
        logger.info(f"Available models for selection: {len(autoselect_config.available_models)}")
        for model_info in autoselect_config.available_models:
            logger.info(f"  - {model_info.model_id}: {model_info.description}")
        logger.info(f"Fallback model: {autoselect_config.fallback}")

        # Extract the user prompt from the request
        user_messages = request_data.get('messages', [])
        if not user_messages:
            logger.error("No messages provided")
            raise HTTPException(status_code=400, detail="No messages provided")

        logger.info(f"User messages count: {len(user_messages)}")

        # Split: last TAIL messages → current_task (what's happening now)
        #        earlier messages  → context_prompt (session domain/background)
        _TAIL = 5
        tail_msgs = user_messages[-_TAIL:]
        context_msgs = user_messages[:-_TAIL] if len(user_messages) > _TAIL else []
        _task_parts = []
        for _m in tail_msgs:
            _role = _m.get('role', 'user')
            _c = _m.get('content', '')
            if isinstance(_c, list):
                _c = ' '.join((p.get('text', '') if isinstance(p, dict) else str(p)) for p in _c).strip()
            _c = str(_c).strip()
            if len(_c) > 800:
                _c = _c[:800] + "...[truncated]"
            _task_parts.append(f"{_role}: {_c}")
        current_task = "\n".join(_task_parts)

        max_selection_tokens = self._get_selection_max_tokens(autoselect_config)

        # Try to reuse a cached conversation summary (incremental context)
        _summary, _new_msgs = self._find_conversation_summary(context_msgs)
        if _summary:
            context_prompt = self._build_summary_context_prompt(_summary, _new_msgs, max_selection_tokens)
            logger.info(f"Using cached conversation summary ({len(_summary['fingerprints'])} msgs summarized, {len(_new_msgs)} new)")
        else:
            context_prompt = self._compact_messages_for_selection(context_msgs, max_selection_tokens) if context_msgs else ""
            logger.info(f"No cached summary — full context compaction")

        logger.info(f"Context: {len(context_prompt)} chars; current task: {len(current_task)} chars ({len(tail_msgs)} tail msgs, limit {max_selection_tokens} tokens)")
        logger.info(f"Current task preview: {current_task[:300]}..." if len(current_task) > 300 else f"Current task: {current_task}")

        # Filter out entries with empty model_id (misconfiguration guard)
        valid_models = [m for m in autoselect_config.available_models if (m.model_id or '').strip()]
        invalid_count = len(autoselect_config.available_models) - len(valid_models)
        if invalid_count:
            logger.warning(f"Autoselect '{autoselect_id}': ignoring {invalid_count} available_model(s) with empty model_id")

        # Build escalation order: sort available models by priority ascending (lower priority first)
        available_models_ordered = sorted(valid_models, key=lambda m: m.priority)
        request_data['stream'] = True
        failed_models: List[str] = []
        last_exception = None
        _summary_stored = False

        while True:
            logger.info(f"Requesting model selection from AI (failed so far: {failed_models})...")
            selected_model_id = await self._get_model_selection(context_prompt, current_task, autoselect_config, failed_models)

            # Store/update conversation summary after the first selection (fire-and-forget)
            if not _summary_stored:
                _summary_stored = True
                asyncio.create_task(self._store_conversation_summary(
                    context_msgs, selected_model_id, autoselect_config, context_prompt
                ))

            available_ids = [m.model_id for m in available_models_ordered if m.model_id not in failed_models]
            if not selected_model_id or selected_model_id not in available_ids:
                untried = [m.model_id for m in available_models_ordered if m.model_id not in failed_models]
                if not untried:
                    logger.error("All models have failed, no escalation target available")
                    if last_exception:
                        raise last_exception
                    raise HTTPException(status_code=502, detail="All autoselect models failed")
                selected_model_id = untried[0]
                logger.warning(f"Selection invalid, using next untried model: {selected_model_id}")
            else:
                logger.info(f"Selected model: {selected_model_id}")

            logger.info(f"Proxying streaming request to: {selected_model_id}")
            try:
                proxied_request_data = dict(request_data)
                proxied_request_data['_autoselect_id'] = autoselect_id
                self._apply_autoselect_max_tokens(proxied_request_data, autoselect_config, selected_model_id, available_models_ordered)
                # Check if it's a rotation first
                if (self.user_id and selected_model_id in self.rotations) or selected_model_id in self.config.rotations:
                    rotation_handler = RotationHandler(user_id=self.user_id)
                    response = await rotation_handler.handle_rotation_request(selected_model_id, proxied_request_data, self.user_id)
                elif '/' in selected_model_id:
                    provider_id, model_name = selected_model_id.split('/', 1)
                    if provider_id not in self.config.providers:
                        raise HTTPException(status_code=400, detail=f"Provider {provider_id} not found")
                    request_handler = RequestHandler()
                    proxied_request_data['model'] = model_name
                    response = await request_handler.handle_streaming_chat_completion(
                        request=None, provider_id=provider_id, request_data=proxied_request_data)
                elif selected_model_id in self.config.providers:
                    provider_config = self.config.get_provider(selected_model_id)
                    models = getattr(provider_config, "models", [])
                    if not models:
                        raise HTTPException(status_code=400, detail=f"Provider {selected_model_id} has no models configured")
                    request_handler = RequestHandler()
                    proxied_request_data['model'] = models[0].name
                    response = await request_handler.handle_streaming_chat_completion(
                        request=None, provider_id=selected_model_id, request_data=proxied_request_data)
                else:
                    raise HTTPException(status_code=400, detail=f"Model {selected_model_id} not found")
            except Exception as e:
                logger.warning(f"Streaming model '{selected_model_id}' raised exception: {e} — escalating")
                failed_models.append(selected_model_id)
                last_exception = e
                untried = [m.model_id for m in available_models_ordered if m.model_id not in failed_models]
                if not untried:
                    logger.error("All streaming models exhausted after exception")
                    raise
                continue

            # Check if the rotation itself reported a full failure (all its internal models failed).
            # Works for both dict responses (non-streaming) and StreamingResponse objects tagged
            # with aisbf_error=True by _create_error_streaming_response.
            if (isinstance(response, dict) and response.get('aisbf_error')) or getattr(response, 'aisbf_error', False):
                logger.warning(f"Rotation '{selected_model_id}' fully failed (aisbf_error) — escalating")
                failed_models.append(selected_model_id)
                untried = [m.model_id for m in available_models_ordered if m.model_id not in failed_models]
                if not untried:
                    logger.warning("All autoselect models exhausted after rotation failure, returning last response")
                    break
                continue

            # Good response
            break

        logger.info(f"=== AUTOSELECT STREAMING REQUEST END ===")
        return response

    async def handle_autoselect_model_list(self, autoselect_id: str) -> List[Dict]:
        """List the available models for an autoselect endpoint"""
        autoselect_config = self.config.get_autoselect(autoselect_id)
        if not autoselect_config:
            raise HTTPException(status_code=400, detail=f"Autoselect {autoselect_id} not found")

        # Return the available models that can be selected
        return [
            {
                "id": model_info.model_id,
                "name": model_info.model_id,
                "description": model_info.description
            }
            for model_info in autoselect_config.available_models
        ]
