"""
Unified backbone interface for CHNB evaluations.

Supports five vision-capable backbones (gemini, qwen3vl, glm4v, gpt, claude)
and two text-only siblings (qwen3, glm4) used by Socratic / subtitles-only.

Required env vars:
    GOOGLE_CLOUD_PROJECT  for 'gemini' (with `gcloud auth application-default login`)
    OPENAI_API_KEY        for 'gpt'
    ANTHROPIC_API_KEY     for 'claude'

Usage:

    from util.backbones import init_backbone, call_backbone

    init_backbone('gemini', project_id='your-gcp-project')
    init_backbone('claude')

    reply = call_backbone('gemini', 'What is in this image?', images=[pil_img])
    reply = call_backbone('claude', 'Answer this MCQ', images=frames, json_mode=True)

`call_backbone` returns the raw assistant text; parse JSON from it with
`parse_json_flexible`.
"""
from __future__ import annotations

import base64
import io
import json
import os
import re
import warnings
from typing import Any, Optional

from util.config import (
    DEFAULT_MODELS,
    DEFAULT_SYSTEM_INSTRUCTION,
    DEFAULT_VERTEX_LOCATION,
    GOOGLE_CLOUD_PROJECT_ENV,
)


class FatalBackboneError(BaseException):
    """Unrecoverable backbone condition (quota / bad key / blocked account).

    Subclasses BaseException so the eval loop's `except Exception` does NOT
    swallow it.
    """


def _detect_fatal_openai(exc: Exception) -> Optional[str]:
    """User-facing message for a hard-fail OpenAI exception, else None."""
    msg = str(exc)
    body = getattr(exc, 'body', None) or {}
    if isinstance(body, dict):
        err = body.get('error', {}) if isinstance(body.get('error'), dict) else {}
        err_type = (err.get('type') or '').lower()
        err_code = (err.get('code') or '').lower()
    else:
        err_type = err_code = ''

    if 'insufficient_quota' in err_type or 'insufficient_quota' in err_code or \
            'insufficient_quota' in msg.lower():
        return ('OpenAI returned insufficient_quota. Your account is out of '
                'credits or hit its monthly cap. Add credits / raise spend cap at '
                'https://platform.openai.com/account/billing, then re-run.')
    if 'invalid_api_key' in msg.lower() or 'incorrect api key' in msg.lower():
        return ('OpenAI rejected the API key. Check OPENAI_API_KEY -- '
                'might be expired, deleted, or for the wrong project.')
    return None


def _detect_fatal_anthropic(exc: Exception) -> Optional[str]:
    """User-facing message for a hard-fail Anthropic exception, else None."""
    msg_lower = str(exc).lower()
    cls = type(exc).__name__
    body = getattr(exc, 'body', None) or {}
    err_type = ''
    if isinstance(body, dict):
        err = body.get('error', {}) if isinstance(body.get('error'), dict) else {}
        err_type = (err.get('type') or '').lower()

    if cls == 'AuthenticationError' or err_type == 'authentication_error' or \
            'invalid x-api-key' in msg_lower:
        return ('Anthropic returned authentication_error (401). The '
                'ANTHROPIC_API_KEY is wrong, expired, or for a different account.')
    if cls == 'PermissionDeniedError' or err_type == 'permission_error':
        return ('Anthropic returned permission_denied (403), your key is valid '
                'but lacks access to this model.')
    if 'credit_balance_too_low' in msg_lower or 'insufficient_credit' in msg_lower:
        return ('Anthropic credit balance too low. Add credits at '
                'https://console.anthropic.com, then re-run.')
    if cls == 'NotFoundError' or err_type == 'not_found_error':
        if 'model' in msg_lower:
            return ('Anthropic returned model_not_found. Check --model.')
    return None


def _detect_fatal_gemini(exc: Exception) -> Optional[str]:
    """User-facing message for a hard-fail Vertex/Gemini exception, else None."""
    msg_lower = str(exc).lower()
    cls = type(exc).__name__.lower()

    if 'unauthenticated' in cls or 'unauthenticated' in msg_lower:
        return ('Vertex AI returned unauthenticated. Refresh credentials with:\n'
                '  gcloud auth application-default login\n'
                'Then verify GOOGLE_CLOUD_PROJECT points to the right project.')
    if 'permissiondenied' in cls or 'permission_denied' in msg_lower:
        return ('Vertex AI returned permission_denied. The active ADC account '
                'lacks the aiplatform.endpoints.predict permission on '
                f'project={os.environ.get(GOOGLE_CLOUD_PROJECT_ENV)!r}. '
                'Either switch projects (set GOOGLE_CLOUD_PROJECT) or grant '
                '`Vertex AI User` role to your account.')
    if 'invalid_argument' in msg_lower and 'model' in msg_lower:
        return ('Vertex AI returned invalid_argument for the model id. '
                'Check --model -- preview model names change, the alias may be retired.')
    return None


_clients: dict[str, Any] = {}
_configs: dict[str, dict] = {}


def is_initialized(backbone: str) -> bool:
    return backbone in _clients


def get_model_id(backbone: str) -> str:
    return _configs.get(backbone, {}).get('model_id', DEFAULT_MODELS.get(backbone, ''))


def _pil_to_jpeg_bytes(img, quality: int = 85) -> bytes:
    if img.mode != 'RGB':
        img = img.convert('RGB')
    buf = io.BytesIO()
    img.save(buf, format='JPEG', quality=quality)
    return buf.getvalue()


def _pil_to_b64(img, quality: int = 85) -> str:
    return base64.standard_b64encode(_pil_to_jpeg_bytes(img, quality)).decode('ascii')


def parse_json_flexible(text: str) -> Any:
    """Parse JSON from text, with regex fallback for prose-wrapped output."""
    if not text:
        return None
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass
    for pat in (r'\[.*\]', r'\{.*\}'):
        m = re.search(pat, text, re.DOTALL)
        if m:
            try:
                return json.loads(m.group(0))
            except json.JSONDecodeError:
                continue
    return None


def _init_gemini(
    project_id: Optional[str] = None,
    location: str = DEFAULT_VERTEX_LOCATION,
    model_id: Optional[str] = None,
    **_: Any,
) -> None:
    from google import genai
    project_id = project_id or os.environ.get(GOOGLE_CLOUD_PROJECT_ENV)
    if not project_id:
        raise RuntimeError(
            f'Gemini backbone requires {GOOGLE_CLOUD_PROJECT_ENV} (env var or '
            'project_id=... kwarg). Run `gcloud auth application-default login` '
            'first.'
        )
    client = genai.Client(vertexai=True, project=project_id, location=location)
    _clients['gemini'] = client
    _configs['gemini'] = {
        'model_id': model_id or DEFAULT_MODELS['gemini'],
        'project_id': project_id,
        'location': location,
    }


def _call_gemini(
    prompt: str,
    images: Optional[list] = None,
    system_instruction: str = DEFAULT_SYSTEM_INSTRUCTION,
    json_mode: bool = True,
    max_output_tokens: int = 4096,
    temperature: float = 0.0,
    thinking_level: str = 'HIGH',
    **_: Any,
) -> str:
    from google.genai import types
    from google.genai.types import GenerateContentConfig

    model_id = _configs['gemini']['model_id']

    contents: list = []
    if images:
        for img in images:
            contents.append(types.Part.from_bytes(
                data=_pil_to_jpeg_bytes(img), mime_type='image/jpeg',
            ))
    contents.append(prompt)

    config_kwargs: dict = {
        'system_instruction': system_instruction,
        'temperature': temperature,
        'max_output_tokens': max_output_tokens,
    }

    if model_id.startswith(('gemini-3-', 'gemini-3.1-')):
        try:
            _level_map = {
                'HIGH': types.ThinkingLevel.HIGH,
                'LOW':  types.ThinkingLevel.LOW,
            }
            level_key = str(thinking_level).upper()
            if level_key in _level_map:
                config_kwargs['thinking_config'] = types.ThinkingConfig(
                    thinking_level=_level_map[level_key],
                )
        except (AttributeError, TypeError):
            pass

    if json_mode:
        config_kwargs['response_mime_type'] = 'application/json'

    try:
        response = _clients['gemini'].models.generate_content(
            model=model_id,
            contents=contents,
            config=GenerateContentConfig(**config_kwargs),
        )
    except Exception as e:
        fatal = _detect_fatal_gemini(e)
        if fatal:
            raise FatalBackboneError(f'_call_gemini({model_id}): {fatal}') from e
        raise

    text = (response.text or '').strip()
    if not text:
        um = getattr(response, 'usage_metadata', None)
        candidates = getattr(response, 'candidates', None) or []
        finish_reason = getattr(candidates[0], 'finish_reason', None) if candidates else None
        thoughts = getattr(um, 'thoughts_token_count', None) if um is not None else None
        warnings.warn(
            f"_call_gemini({model_id}): empty text. "
            f"finish_reason={finish_reason!r}, thoughts={thoughts}. "
            f"If finish_reason='MAX_TOKENS' or thoughts~max_output_tokens: raise --max-output-tokens.",
            UserWarning, stacklevel=2,
        )
    return text


def _init_qwen3vl(
    model_id: Optional[str] = None,
    device_map: str = 'auto',
    attn_implementation: str = 'flash_attention_2',
    min_pixels: int = 256 * 28 * 28,
    max_pixels: int = 512 * 28 * 28,
    **_: Any,
) -> None:
    import torch
    from transformers import AutoModelForImageTextToText, AutoProcessor

    mid = model_id or DEFAULT_MODELS['qwen3vl']
    processor = AutoProcessor.from_pretrained(
        mid, trust_remote_code=True,
        min_pixels=min_pixels, max_pixels=max_pixels,
    )
    model = AutoModelForImageTextToText.from_pretrained(
        mid, dtype=torch.bfloat16, device_map=device_map,
        attn_implementation=attn_implementation, trust_remote_code=True,
    )
    _clients['qwen3vl'] = (model, processor)
    _configs['qwen3vl'] = {'model_id': mid}


def _call_qwen3vl(
    prompt: str,
    images: Optional[list] = None,
    system_instruction: str = DEFAULT_SYSTEM_INSTRUCTION,
    json_mode: bool = True,
    max_output_tokens: int = 2048,
    temperature: float = 0.1,
    **_: Any,
) -> str:
    import torch
    from qwen_vl_utils import process_vision_info

    model, processor = _clients['qwen3vl']

    content: list = []
    if images:
        for img in images:
            content.append({'type': 'image', 'image': img})
    content.append({'type': 'text', 'text': prompt})
    messages = [
        {'role': 'system', 'content': system_instruction},
        {'role': 'user', 'content': content},
    ]
    text = processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)

    if images:
        image_inputs, video_inputs = process_vision_info(messages)
        inputs = processor(
            text=[text], images=image_inputs, videos=video_inputs,
            return_tensors='pt', padding=True,
        )
    else:
        inputs = processor(text=[text], return_tensors='pt', padding=True)
    inputs = {k: (v.to(model.device) if hasattr(v, 'to') else v) for k, v in inputs.items()}

    do_sample = temperature > 0
    with torch.no_grad():
        gen_ids = model.generate(
            **inputs,
            max_new_tokens=max_output_tokens,
            temperature=temperature if do_sample else 1.0,
            do_sample=do_sample,
        )
    in_len = inputs['input_ids'].shape[1]
    out_ids = gen_ids[:, in_len:]
    return processor.batch_decode(out_ids, skip_special_tokens=True)[0].strip()


def _init_qwen3(
    model_id: Optional[str] = None,
    device_map: str = 'auto',
    attn_implementation: str = 'flash_attention_2',
    **_: Any,
) -> None:
    import torch
    from transformers import AutoModelForCausalLM, AutoTokenizer

    mid = model_id or DEFAULT_MODELS['qwen3']
    tokenizer = AutoTokenizer.from_pretrained(mid, trust_remote_code=True)
    model = AutoModelForCausalLM.from_pretrained(
        mid, dtype=torch.bfloat16, device_map=device_map,
        attn_implementation=attn_implementation, trust_remote_code=True,
    )
    _clients['qwen3'] = (model, tokenizer)
    _configs['qwen3'] = {'model_id': mid}


def _call_qwen3(
    prompt: str,
    images: Optional[list] = None,
    system_instruction: str = DEFAULT_SYSTEM_INSTRUCTION,
    json_mode: bool = True,
    max_output_tokens: int = 2048,
    temperature: float = 0.1,
    **_: Any,
) -> str:
    import torch
    if images:
        warnings.warn(
            f'_call_qwen3: images={len(images)} were passed but Qwen3 (text-only) '
            f'cannot use them. Use qwen3vl backbone for vision input.',
            UserWarning, stacklevel=2,
        )
    model, tokenizer = _clients['qwen3']

    messages = [
        {'role': 'system', 'content': system_instruction},
        {'role': 'user',   'content': prompt},
    ]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True,
    )
    inputs = tokenizer(text, return_tensors='pt', padding=True).to(model.device)

    do_sample = temperature > 0
    with torch.no_grad():
        gen_ids = model.generate(
            **inputs,
            max_new_tokens=max_output_tokens,
            temperature=temperature if do_sample else 1.0,
            do_sample=do_sample,
        )
    in_len = inputs['input_ids'].shape[1]
    out_ids = gen_ids[:, in_len:]
    return tokenizer.batch_decode(out_ids, skip_special_tokens=True)[0].strip()


def _init_glm4(
    model_id: Optional[str] = None,
    device_map: str = 'auto',
    **_: Any,
) -> None:
    import torch
    from transformers import AutoModelForCausalLM, AutoTokenizer

    mid = model_id or DEFAULT_MODELS['glm4']
    tokenizer = AutoTokenizer.from_pretrained(mid, trust_remote_code=True)
    model = AutoModelForCausalLM.from_pretrained(
        mid, torch_dtype=torch.bfloat16, device_map=device_map,
        trust_remote_code=True,
    )
    _clients['glm4'] = (model, tokenizer)
    _configs['glm4'] = {'model_id': mid}


def _call_glm4(
    prompt: str,
    images: Optional[list] = None,
    system_instruction: str = DEFAULT_SYSTEM_INSTRUCTION,
    json_mode: bool = True,
    max_output_tokens: int = 2048,
    temperature: float = 0.1,
    **_: Any,
) -> str:
    import torch
    if images:
        warnings.warn(
            f'_call_glm4: images={len(images)} were passed but GLM-4 (text-only) '
            f'cannot use them. Use glm4v backbone for vision input.',
            UserWarning, stacklevel=2,
        )
    model, tokenizer = _clients['glm4']

    messages = [
        {'role': 'system', 'content': system_instruction},
        {'role': 'user',   'content': prompt},
    ]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True,
    )
    inputs = tokenizer(text, return_tensors='pt', padding=True).to(model.device)

    do_sample = temperature > 0
    with torch.no_grad():
        gen_ids = model.generate(
            **inputs,
            max_new_tokens=max_output_tokens,
            temperature=temperature if do_sample else 1.0,
            do_sample=do_sample,
        )
    in_len = inputs['input_ids'].shape[1]
    out_ids = gen_ids[:, in_len:]
    return tokenizer.batch_decode(out_ids, skip_special_tokens=True)[0].strip()


def _init_glm4v(
    model_id: Optional[str] = None,
    device_map: str = 'auto',
    **_: Any,
) -> None:
    import torch
    from transformers import AutoProcessor, Glm4vForConditionalGeneration

    mid = model_id or DEFAULT_MODELS['glm4v']
    processor = AutoProcessor.from_pretrained(mid)
    model = Glm4vForConditionalGeneration.from_pretrained(
        mid, torch_dtype=torch.bfloat16, device_map=device_map,
    )
    _clients['glm4v'] = (model, processor)
    _configs['glm4v'] = {
        'model_id': mid,
        'think_end_id':    processor.tokenizer.convert_tokens_to_ids('</think>'),
        'answer_start_id': processor.tokenizer.convert_tokens_to_ids('<answer>'),
        'answer_end_id':   processor.tokenizer.convert_tokens_to_ids('</answer>'),
    }


def _call_glm4v(
    prompt: str,
    images: Optional[list] = None,
    system_instruction: str = DEFAULT_SYSTEM_INSTRUCTION,
    json_mode: bool = True,
    max_output_tokens: int = 4096,
    temperature: float = 0.1,
    **_: Any,
) -> str:
    import torch
    model, processor = _clients['glm4v']
    cfg = _configs['glm4v']

    content: list = []
    if images:
        for img in images:
            content.append({'type': 'image', 'image': img})
    content.append({'type': 'text', 'text': prompt})
    messages = [{'role': 'user', 'content': content}]

    with warnings.catch_warnings():
        warnings.filterwarnings('ignore', message='.*processor_kwargs.*')
        inputs = processor.apply_chat_template(
            messages, tokenize=True, add_generation_prompt=True,
            return_dict=True, return_tensors='pt', padding=True,
        )
    if isinstance(inputs, str):
        inputs = processor.tokenizer(inputs, return_tensors='pt', padding=True)
    inputs = {k: (v.to(model.device) if hasattr(v, 'to') else v) for k, v in inputs.items()}
    inputs.pop('token_type_ids', None)
    in_len = inputs['input_ids'].shape[1]

    do_sample = temperature > 0
    gen_ids = model.generate(
        **inputs,
        max_new_tokens=max_output_tokens,
        temperature=temperature if do_sample else 1.0,
        do_sample=do_sample,
    )
    out_ids = gen_ids[0][in_len:]
    out_list = out_ids.tolist()

    # GLM-4.1V emits <think>...</think><answer>...</answer>. If the model
    # stopped before closing <answer>, force-feed the missing structural
    # tokens so the answer block can be extracted.
    if cfg['answer_end_id'] not in out_list:
        tokens_to_add: list[int] = []
        if cfg['think_end_id'] not in out_list:
            tokens_to_add.extend([cfg['think_end_id'], cfg['answer_start_id']])
        elif cfg['answer_start_id'] not in out_list:
            tokens_to_add.append(cfg['answer_start_id'])
        if tokens_to_add:
            extra = torch.tensor(tokens_to_add).unsqueeze(0).to(model.device)
            force_input = torch.cat([gen_ids, extra], dim=1)
            force_attn = torch.ones_like(force_input)
            force_kwargs: dict = {
                'input_ids': force_input,
                'attention_mask': force_attn,
            }
            for k in ['pixel_values', 'video_metadata']:
                if k in inputs:
                    force_kwargs[k] = inputs[k]
            gen2 = model.generate(
                **force_kwargs,
                max_new_tokens=min(2048, max_output_tokens),
                temperature=temperature if do_sample else 1.0,
                do_sample=do_sample,
            )
            out_ids = gen2[0][in_len:]

    full_text = processor.tokenizer.decode(out_ids, skip_special_tokens=False)

    m = re.search(r'<answer>(.*?)(?:</answer>|$)', full_text, re.DOTALL)
    if m:
        return m.group(1).strip()
    m = re.search(r'</think>(.*)', full_text, re.DOTALL)
    if m:
        return m.group(1).strip()
    return full_text.strip()


def _init_gpt(
    api_key: Optional[str] = None,
    model_id: Optional[str] = None,
    base_url: Optional[str] = None,
    max_retries: int = 8,
    **_: Any,
) -> None:
    # max_retries=8 (vs SDK default 2): the SDK honors Retry-After and uses
    # exponential backoff, so we ride out most transient 429s.
    from openai import OpenAI
    kwargs: dict = {'max_retries': max_retries}
    if api_key:
        kwargs['api_key'] = api_key
    if base_url:
        kwargs['base_url'] = base_url
    client = OpenAI(**kwargs)
    _clients['gpt'] = client
    _configs['gpt'] = {'model_id': model_id or DEFAULT_MODELS['gpt']}


def _call_gpt(
    prompt: str,
    images: Optional[list] = None,
    system_instruction: str = DEFAULT_SYSTEM_INSTRUCTION,
    json_mode: bool = True,
    max_output_tokens: int = 4096,
    temperature: float = 0.0,
    reasoning_effort: str = 'high',
    **_: Any,
) -> str:
    client = _clients['gpt']
    model_id = _configs['gpt']['model_id']
    is_reasoning = str(model_id).startswith(('gpt-5', 'o1', 'o3'))

    content: list = []
    if images:
        for img in images:
            b64 = _pil_to_b64(img)
            content.append({
                'type': 'input_image',
                'image_url': f'data:image/jpeg;base64,{b64}',
            })
    content.append({'type': 'input_text', 'text': prompt})

    kwargs: dict = {
        'model': model_id,
        'input': [{'role': 'user', 'content': content}],
        'max_output_tokens': max_output_tokens,
        'instructions': system_instruction,
    }
    if not is_reasoning:
        kwargs['temperature'] = temperature
    if is_reasoning:
        _valid_efforts = {'minimal', 'low', 'medium', 'high'}
        effort = str(reasoning_effort).lower()
        if effort not in _valid_efforts:
            effort = 'high'
        kwargs['reasoning'] = {'effort': effort}

    if json_mode:
        kwargs['text'] = {'format': {'type': 'json_object'}}

    try:
        try:
            response = client.responses.create(**kwargs)
        except TypeError as e:
            if json_mode and "'text'" in str(e):
                kwargs.pop('text', None)
                response = client.responses.create(**kwargs)
            else:
                raise
    except Exception as e:
        fatal = _detect_fatal_openai(e)
        if fatal:
            raise FatalBackboneError(f'_call_gpt({model_id}): {fatal}') from e
        raise

    text = getattr(response, 'output_text', None)
    if text is None:
        parts: list[str] = []
        for out in getattr(response, 'output', []) or []:
            if getattr(out, 'type', None) == 'reasoning':
                continue
            for block in getattr(out, 'content', []) or []:
                t = getattr(block, 'text', None)
                if t:
                    parts.append(t)
        text = ''.join(parts)
    text = (text or '').strip()

    if not text:
        usage = getattr(response, 'usage', None)
        status = getattr(response, 'status', None)
        incomplete = getattr(response, 'incomplete_details', None)
        incomplete_reason = getattr(incomplete, 'reason', None) if incomplete else None
        reasoning_tokens = None
        if usage is not None:
            details = getattr(usage, 'output_tokens_details', None)
            if details is not None:
                reasoning_tokens = getattr(details, 'reasoning_tokens', None)
        warnings.warn(
            f"_call_gpt({model_id}): empty text. "
            f"status={status!r}, incomplete_reason={incomplete_reason!r}, "
            f"reasoning_tokens={reasoning_tokens}. "
            f"If reasoning~max_output_tokens or incomplete_reason='max_output_tokens': raise --max-output-tokens.",
            UserWarning, stacklevel=2,
        )
    return text


def _init_claude(
    api_key: Optional[str] = None,
    model_id: Optional[str] = None,
    base_url: Optional[str] = None,
    max_retries: int = 8,
    **_: Any,
) -> None:
    # max_retries=8 (vs SDK default 2): the SDK honors Retry-After and uses
    # exponential backoff.
    import anthropic
    kwargs: dict = {'max_retries': max_retries}
    if api_key:
        kwargs['api_key'] = api_key
    if base_url:
        kwargs['base_url'] = base_url
    client = anthropic.Anthropic(**kwargs)
    _clients['claude'] = client
    _configs['claude'] = {'model_id': model_id or DEFAULT_MODELS['claude']}


def _call_claude(
    prompt: str,
    images: Optional[list] = None,
    system_instruction: str = DEFAULT_SYSTEM_INSTRUCTION,
    json_mode: bool = True,
    max_output_tokens: int = 4096,
    temperature: float = 0.1,
    **_: Any,
) -> str:
    client = _clients['claude']
    model_id = _configs['claude']['model_id']

    # Claude best practice: images BEFORE text.
    content: list = []
    if images:
        for img in images:
            content.append({
                'type': 'image',
                'source': {
                    'type': 'base64',
                    'media_type': 'image/jpeg',
                    'data': _pil_to_b64(img),
                },
            })
    text_prompt = prompt
    if json_mode and 'json' not in prompt.lower():
        text_prompt = f'{prompt}\n\nRespond with valid JSON only, no other text.'
    content.append({'type': 'text', 'text': text_prompt})

    import time as _time
    import anthropic as _anthropic
    _outer_retries = 6
    message = None
    for _attempt in range(_outer_retries + 1):
        try:
            with client.messages.stream(
                model=model_id,
                max_tokens=max_output_tokens,
                temperature=temperature,
                system=system_instruction,
                messages=[{'role': 'user', 'content': content}],
            ) as stream:
                for _ in stream.text_stream:
                    pass
                message = stream.get_final_message()
            break
        except _anthropic.RateLimitError as e:
            wait = 60.0
            try:
                resp = getattr(e, 'response', None)
                hdr = getattr(resp, 'headers', None) or {}
                ra = hdr.get('retry-after') or hdr.get('Retry-After')
                if ra:
                    wait = max(1.0, float(ra))
            except Exception:
                pass
            if _attempt >= _outer_retries:
                raise
            print(f'  [claude rate limit] sleeping {wait:.0f}s then retrying '
                  f'(attempt {_attempt + 2}/{_outer_retries + 1})')
            _time.sleep(wait)
            continue
        except Exception as e:
            fatal = _detect_fatal_anthropic(e)
            if fatal:
                raise FatalBackboneError(f'_call_claude({model_id}): {fatal}') from e
            raise
    assert message is not None
    parts: list[str] = []
    block_types: list[str] = []
    for block in message.content or []:
        block_types.append(getattr(block, 'type', '?'))
        if getattr(block, 'type', None) == 'thinking':
            continue
        t = getattr(block, 'text', None)
        if t:
            parts.append(t)
    text = ''.join(parts).strip()

    if not text:
        stop_reason = getattr(message, 'stop_reason', None)
        warnings.warn(
            f"_call_claude({model_id}): empty text. "
            f"stop_reason={stop_reason!r}, content_block_types={block_types}. "
            f"If stop_reason='max_tokens': raise --max-output-tokens.",
            UserWarning, stacklevel=2,
        )
    return text


_INIT_FUNCS = {
    'gemini':  _init_gemini,
    'qwen3vl': _init_qwen3vl,
    'qwen3':   _init_qwen3,
    'glm4v':   _init_glm4v,
    'glm4':    _init_glm4,
    'gpt':     _init_gpt,
    'claude':  _init_claude,
}

_CALL_FUNCS = {
    'gemini':  _call_gemini,
    'qwen3vl': _call_qwen3vl,
    'qwen3':   _call_qwen3,
    'glm4v':   _call_glm4v,
    'glm4':    _call_glm4,
    'gpt':     _call_gpt,
    'claude':  _call_claude,
}


def init_backbone(backbone: str, force: bool = False, **kwargs: Any) -> None:
    """Load / connect to a backbone. Idempotent unless `force=True`."""
    if backbone not in _INIT_FUNCS:
        raise ValueError(
            f'Unknown backbone: {backbone!r}. Available: {list(_INIT_FUNCS)}'
        )
    if backbone in _clients and not force:
        return
    _INIT_FUNCS[backbone](**kwargs)


def call_backbone(
    backbone: str,
    prompt: str,
    images: Optional[list] = None,
    system_instruction: str = DEFAULT_SYSTEM_INSTRUCTION,
    json_mode: bool = True,
    max_output_tokens: int = 4096,
    temperature: float = 0.0,
    **extra: Any,
) -> str:
    """Call a backbone with a unified signature. Returns the raw assistant text.

    `images` is a list of PIL.Image.Image objects. `**extra` is forwarded to
    the per-backbone implementation (e.g. `thinking_level` for Gemini,
    `reasoning_effort` for GPT).
    """
    if backbone not in _clients:
        raise RuntimeError(
            f'Backbone {backbone!r} not initialized. Call init_backbone({backbone!r}, ...) first.'
        )
    return _CALL_FUNCS[backbone](
        prompt=prompt,
        images=images,
        system_instruction=system_instruction,
        json_mode=json_mode,
        max_output_tokens=max_output_tokens,
        temperature=temperature,
        **extra,
    )