"""
Base provider with integrated telemetry, events, and exception handling.
"""

import time
import uuid
import asyncio
import warnings
import base64
import json
import re
import socket
import hashlib
import inspect
import threading
from collections import deque, OrderedDict
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Union, Iterator, AsyncIterator, Type, TYPE_CHECKING, Tuple, Sequence
from abc import ABC, abstractmethod

try:
    from pydantic import BaseModel
    PYDANTIC_AVAILABLE = True
except ImportError:
    PYDANTIC_AVAILABLE = False
    BaseModel = None

from ..core.interface import AbstractCoreInterface
from ..core.types import GenerateResponse
from ..core.multimodal_generation import GeneratedItem, GeneratedResource, MultimodalGenerateResponse
from ..core.output_specs import (
    is_output_request,
    normalize_output_spec,
    normalize_output_specs,
    output_plugin_kwargs,
)
from ..capabilities.types import is_artifact_ref
from ..events import EventType, Event
from datetime import datetime
from ..utils.structured_logging import get_logger
from ..utils.jsonish import loads_dict_like
from ..utils.truncation import preview_text
from ..exceptions import (
    ProviderAPIError,
    AuthenticationError,
    RateLimitError,
    InvalidRequestError,
    UnsupportedFeatureError,
    ModelNotFoundError
)
from ..architectures import detect_architecture, get_architecture_format, get_model_capabilities
from ..architectures.response_postprocessing import (
    normalize_assistant_text,
    maybe_create_incremental_thinking_tag_stripper,
    strip_output_wrappers,
)
from ..tools import execute_tools
from ..core.retry import RetryManager, RetryConfig

if TYPE_CHECKING:  # pragma: no cover
    # Imported for type checking only to avoid hard dependencies in minimal installs.
    from ..media.types import MediaContent


@dataclass
class _PromptCacheEntry:
    value: Any
    created_at_s: float
    last_accessed_at_s: float
    ttl_s: Optional[float] = None
    meta: Dict[str, Any] = field(default_factory=dict)


class PromptCacheStore:
    """Best-effort in-process prompt cache store (LRU + optional TTL).

    Providers can store arbitrary backend-specific cache objects keyed by a caller-provided string
    (`prompt_cache_key`). This is primarily useful for local inference backends (MLX, llama.cpp).

    Notes:
    - This store is intentionally simple and in-process only.
    - Callers should treat prompt caches as potentially sensitive (they contain user prompt state).
    """

    def __init__(self, *, max_entries: int = 32, default_ttl_s: Optional[float] = None):
        self._max_entries = int(max_entries) if max_entries and int(max_entries) > 0 else 32
        self._default_ttl_s = default_ttl_s if default_ttl_s is None else float(default_ttl_s)
        self._entries: "OrderedDict[str, _PromptCacheEntry]" = OrderedDict()
        self._lock = threading.RLock()

    def _is_expired(self, entry: _PromptCacheEntry) -> bool:
        ttl_s = entry.ttl_s if entry.ttl_s is not None else self._default_ttl_s
        if ttl_s is None:
            return False
        return (time.time() - entry.last_accessed_at_s) > float(ttl_s)

    def get(self, key: str) -> Optional[Any]:
        if not isinstance(key, str) or not key.strip():
            return None
        key = key.strip()
        with self._lock:
            entry = self._entries.get(key)
            if entry is None:
                return None
            if self._is_expired(entry):
                self.delete(key)
                return None
            entry.last_accessed_at_s = time.time()
            self._entries.move_to_end(key)
            return entry.value

    def set(
        self,
        key: str,
        value: Any,
        *,
        ttl_s: Optional[float] = None,
        meta: Optional[Dict[str, Any]] = None,
    ) -> None:
        if not isinstance(key, str) or not key.strip():
            raise ValueError("prompt cache key must be a non-empty string")
        key = key.strip()
        now = time.time()
        with self._lock:
            self._entries[key] = _PromptCacheEntry(
                value=value,
                created_at_s=now,
                last_accessed_at_s=now,
                ttl_s=ttl_s,
                meta=dict(meta or {}),
            )
            self._entries.move_to_end(key)
            while len(self._entries) > self._max_entries:
                self._entries.popitem(last=False)

    def delete(self, key: str) -> bool:
        if not isinstance(key, str) or not key.strip():
            return False
        key = key.strip()
        with self._lock:
            return self._entries.pop(key, None) is not None

    def clear(self) -> None:
        with self._lock:
            self._entries.clear()

    def stats(self) -> Dict[str, Any]:
        # Opportunistically purge expired entries.
        with self._lock:
            expired = []
            for k, v in self._entries.items():
                if self._is_expired(v):
                    expired.append(k)
            for k in expired:
                self.delete(k)

            return {
                "entries": len(self._entries),
                "max_entries": self._max_entries,
                "default_ttl_s": self._default_ttl_s,
            }

    def keys(self) -> List[str]:
        with self._lock:
            return list(self._entries.keys())

    def meta(self, key: str) -> Optional[Dict[str, Any]]:
        if not isinstance(key, str) or not key.strip():
            return None
        with self._lock:
            entry = self._entries.get(key.strip())
            if entry is None:
                return None
            return dict(entry.meta or {})


@dataclass(frozen=True)
class ThinkingControlHandling:
    """Provider/model prompt thinking control handling details.

    This makes "partial handling" explicit (e.g., provider can enable/disable thinking
    but cannot enforce effort levels).
    """

    handled_enable_disable: bool = False
    handled_level: bool = False

    @property
    def handled(self) -> bool:
        return bool(self.handled_enable_disable or self.handled_level)


@dataclass(frozen=True)
class PromptCacheModule:
    """A single cacheable module of prompt context.

    This is intentionally generic and JSON-serializable so higher-level layers (runtime/agent/memory)
    can express cache intent without hard-coding provider-specific prompt formats.
    """

    module_id: str
    system_prompt: Optional[str] = None
    prompt: Optional[str] = None
    messages: Optional[List[Dict[str, Any]]] = None
    tools: Optional[List[Dict[str, Any]]] = None
    add_generation_prompt: bool = False
    scope: str = "private"  # "private" | "shared" (advisory; enforcement is host-dependent)
    meta: Dict[str, Any] = field(default_factory=dict)

    def normalized(self) -> "PromptCacheModule":
        module_id = str(self.module_id or "").strip()
        system_prompt = str(self.system_prompt).strip() if isinstance(self.system_prompt, str) and self.system_prompt else None
        prompt = str(self.prompt).strip() if isinstance(self.prompt, str) and self.prompt else None
        messages = None
        if isinstance(self.messages, list) and self.messages:
            out: List[Dict[str, Any]] = []
            for m in self.messages:
                if not isinstance(m, dict):
                    continue
                role = str(m.get("role") or "").strip().lower()
                if not role:
                    continue
                msg: Dict[str, Any] = {"role": role, "content": m.get("content")}
                for k in ("name", "tool_call_id", "tool_calls", "function_call"):
                    if k in m:
                        msg[k] = m.get(k)
                out.append(msg)
            messages = out or None
        tools = None
        if isinstance(self.tools, list) and self.tools:
            out_tools: List[Dict[str, Any]] = []
            for t in self.tools:
                if isinstance(t, dict):
                    out_tools.append(dict(t))
                    continue
                if hasattr(t, "to_dict") and callable(getattr(t, "to_dict")):
                    try:
                        td = t.to_dict()
                        if isinstance(td, dict):
                            out_tools.append(dict(td))
                            continue
                    except Exception:
                        pass
                if hasattr(t, "model_dump") and callable(getattr(t, "model_dump")):
                    try:
                        td = t.model_dump()
                        if isinstance(td, dict):
                            out_tools.append(dict(td))
                            continue
                    except Exception:
                        pass
            tools = out_tools or None
            if tools:
                def _tool_sort_key(tool: Dict[str, Any]) -> str:
                    func = tool.get("function") if isinstance(tool.get("function"), dict) else None
                    name = ""
                    if func:
                        name = str(func.get("name") or "").strip()
                    if not name:
                        name = str(tool.get("name") or "").strip()
                    return name.lower()

                tools.sort(key=_tool_sort_key)
        add_generation_prompt = bool(self.add_generation_prompt)
        scope = str(self.scope or "private").strip().lower() or "private"
        if scope not in {"private", "shared"}:
            scope = "private"
        meta = dict(self.meta or {})
        return PromptCacheModule(
            module_id=module_id,
            system_prompt=system_prompt,
            prompt=prompt,
            messages=messages,
            tools=tools,
            add_generation_prompt=add_generation_prompt,
            scope=scope,
            meta=meta,
        )

    def fingerprint(self, *, version: int = 1) -> str:
        """Stable module fingerprint for hierarchical cache keys (hex sha256)."""
        mod = self.normalized()
        payload = {
            "v": int(version),
            "module_id": mod.module_id,
            "system_prompt": mod.system_prompt,
            "prompt": mod.prompt,
            "messages": mod.messages,
            "tools": mod.tools,
            "add_generation_prompt": bool(mod.add_generation_prompt),
            "scope": mod.scope,
        }
        try:
            raw = json.dumps(payload, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
        except TypeError as e:
            def _first_issue(obj: Any, path: str = "$", seen: Optional[set[int]] = None) -> Optional[str]:
                if seen is None:
                    seen = set()
                try:
                    oid = id(obj)
                    if oid in seen:
                        return None
                    seen.add(oid)
                except Exception:
                    pass
                if obj is None or isinstance(obj, (str, int, float, bool)):
                    return None
                if isinstance(obj, dict):
                    for k, v in obj.items():
                        if not isinstance(k, str):
                            return f"{path} (non-string key type {type(k).__name__})"
                        issue = _first_issue(v, f"{path}.{k}", seen)
                        if issue:
                            return issue
                    return None
                if isinstance(obj, (list, tuple)):
                    for i, v in enumerate(obj):
                        issue = _first_issue(v, f"{path}[{i}]", seen)
                        if issue:
                            return issue
                    return None
                return f"{path} (type {type(obj).__name__})"

            issue = _first_issue(payload) or str(e)
            raise ValueError(
                "PromptCacheModule.fingerprint requires JSON-serializable values; "
                f"first non-serializable value at {issue}."
            ) from e
        return hashlib.sha256(raw.encode("utf-8")).hexdigest()


@dataclass(frozen=True)
class PromptCacheRenderedFragment:
    """Provider-rendered prompt fragment used for durable exact-prefix artifacts."""

    serialized_prompt: str
    serializer_version: str
    cache_backend: str = "provider"
    artifact_format: str = "prompt-cache"
    meta: Dict[str, Any] = field(default_factory=dict)


_PROMPT_CACHE_OPERATION_ALIASES: Dict[str, str] = {
    "set": "set",
    "clear": "clear",
    "update": "update",
    "fork": "fork",
    "save": "save",
    "load": "load",
    "prepare": "prepare_modules",
    "prepare_modules": "prepare_modules",
    "modules": "prepare_modules",
    "stats": "stats",
    "get_stats": "stats",
}


@dataclass(frozen=True)
class PromptCacheCapabilities:
    """Provider prompt-cache capability profile.

    `mode` is an intentionally small abstraction:
    - `none`: provider does not support prompt caching
    - `keyed`: provider accepts cache keys and local default-key control, but does not expose
      local module/update/fork control-plane semantics
    - `local_control_plane`: provider supports the full local prompt-cache control plane
    """

    supported: bool = False
    mode: str = "none"
    supports_set: bool = False
    supports_clear: bool = False
    supports_update: bool = False
    supports_fork: bool = False
    supports_prepare_modules: bool = False
    supports_stats: bool = False
    supports_save: bool = False
    supports_load: bool = False
    supports_ttl: bool = False
    notes: Tuple[str, ...] = ()

    def supports_operation(self, operation: str) -> bool:
        op = _PROMPT_CACHE_OPERATION_ALIASES.get(str(operation or "").strip().lower(), "")
        if op == "set":
            return bool(self.supports_set)
        if op == "clear":
            return bool(self.supports_clear)
        if op == "update":
            return bool(self.supports_update)
        if op == "fork":
            return bool(self.supports_fork)
        if op == "prepare_modules":
            return bool(self.supports_prepare_modules)
        if op == "stats":
            return bool(self.supports_stats)
        if op == "save":
            return bool(self.supports_save)
        if op == "load":
            return bool(self.supports_load)
        return False

    def to_dict(self) -> Dict[str, Any]:
        return {
            "supported": bool(self.supported),
            "mode": str(self.mode or "none"),
            "supports_set": bool(self.supports_set),
            "supports_clear": bool(self.supports_clear),
            "supports_update": bool(self.supports_update),
            "supports_fork": bool(self.supports_fork),
            "supports_prepare_modules": bool(self.supports_prepare_modules),
            "supports_stats": bool(self.supports_stats),
            "supports_save": bool(self.supports_save),
            "supports_load": bool(self.supports_load),
            "supports_ttl": bool(self.supports_ttl),
            "notes": list(self.notes or ()),
        }


class PromptCacheError(RuntimeError):
    """Structured prompt-cache error that higher layers can catch and inspect."""

    def __init__(
        self,
        message: str,
        *,
        operation: str,
        provider: Optional[str] = None,
        model: Optional[str] = None,
        code: str = "prompt_cache_error",
        capabilities: Optional[PromptCacheCapabilities] = None,
    ) -> None:
        super().__init__(message)
        self.operation = _PROMPT_CACHE_OPERATION_ALIASES.get(str(operation or "").strip().lower(), str(operation or ""))
        self.provider = str(provider or "").strip() or None
        self.model = str(model or "").strip() or None
        self.code = str(code or "prompt_cache_error").strip() or "prompt_cache_error"
        self.capabilities = capabilities

    def to_dict(self) -> Dict[str, Any]:
        payload: Dict[str, Any] = {
            "code": self.code,
            "operation": self.operation,
            "message": str(self),
        }
        if self.provider:
            payload["provider"] = self.provider
        if self.model:
            payload["model"] = self.model
        if isinstance(self.capabilities, PromptCacheCapabilities):
            payload["capabilities"] = self.capabilities.to_dict()
        return payload


class PromptCacheUnsupportedError(PromptCacheError):
    def __init__(
        self,
        *,
        operation: str,
        provider: Optional[str] = None,
        model: Optional[str] = None,
        capabilities: Optional[PromptCacheCapabilities] = None,
        detail: Optional[str] = None,
    ) -> None:
        op = _PROMPT_CACHE_OPERATION_ALIASES.get(str(operation or "").strip().lower(), str(operation or "prompt_cache"))
        msg = detail or f"Prompt cache operation '{op}' is not supported by this provider."
        super().__init__(
            msg,
            operation=op,
            provider=provider,
            model=model,
            code="prompt_cache_unsupported",
            capabilities=capabilities,
        )


class PromptCacheOperationError(PromptCacheError):
    def __init__(
        self,
        message: str,
        *,
        operation: str,
        provider: Optional[str] = None,
        model: Optional[str] = None,
        code: str = "prompt_cache_operation_failed",
        capabilities: Optional[PromptCacheCapabilities] = None,
    ) -> None:
        super().__init__(
            message,
            operation=operation,
            provider=provider,
            model=model,
            code=code,
            capabilities=capabilities,
        )


class BaseProvider(AbstractCoreInterface, ABC):
    """
    Base provider class with integrated telemetry and events.
    All providers should inherit from this class.
    """

    def __init__(self, model: str, **kwargs):
        AbstractCoreInterface.__init__(self, model, **kwargs)
        self.provider = None
        self._explicit_generation_params = frozenset(
            key
            for key, value in kwargs.items()
            if key in {"temperature", "top_p", "top_k"} and value is not None
        )
        for key in ("top_p", "top_k"):
            if key in kwargs and kwargs.get(key) is not None:
                setattr(self, key, kwargs[key])

        # Setup structured logging
        self.logger = get_logger(self.__class__.__name__)

        # Detect architecture and get model capabilities
        self.architecture = detect_architecture(model)
        self.architecture_config = get_architecture_format(self.architecture)
        self.model_capabilities = get_model_capabilities(model)
        self._apply_metadata_generation_defaults()

        # #[WARNING:TIMEOUT]
        # Setup timeout configuration (centralized defaults).
        #
        # Semantics:
        # - If the caller passes `timeout=...`, we respect it (including `None` for unlimited).
        # - If the caller omits `timeout`, we use AbstractCore's global config default.
        # - Same logic for `tool_timeout`.
        timeout_provided = "timeout" in kwargs
        tool_timeout_provided = "tool_timeout" in kwargs

        timeout_value = kwargs.get("timeout", None) if timeout_provided else None
        tool_timeout_value = kwargs.get("tool_timeout", None) if tool_timeout_provided else None

        if not timeout_provided or not tool_timeout_provided:
            try:
                from ..config.manager import get_config_manager

                cfg = get_config_manager()
            except Exception:
                cfg = None

            if not timeout_provided:
                try:
                    timeout_value = float(cfg.get_default_timeout()) if cfg is not None else None
                except Exception:
                    timeout_value = None

            if not tool_timeout_provided:
                try:
                    tool_timeout_value = float(cfg.get_tool_timeout()) if cfg is not None else None
                except Exception:
                    tool_timeout_value = None

        # Validate timeouts: non-positive numbers become "unlimited" (None).
        try:
            if isinstance(timeout_value, (int, float)) and float(timeout_value) <= 0:
                timeout_value = None
        except Exception:
            pass
        try:
            if isinstance(tool_timeout_value, (int, float)) and float(tool_timeout_value) <= 0:
                tool_timeout_value = None
        except Exception:
            pass

        self._timeout = timeout_value  # None = unlimited HTTP requests
        self._tool_timeout = tool_timeout_value  # None = unlimited tool execution

        # Setup tool execution mode
        # execute_tools: True = AbstractCore executes tools (legacy mode)
        #                False = Pass-through mode (default - for API server / agentic CLI)
        self.execute_tools = kwargs.get('execute_tools', False)
        if self.execute_tools:
            warnings.warn(
                "execute_tools=True is deprecated. Prefer passing tools explicitly to generate() "
                "and executing tool calls in the host/runtime via a ToolExecutor.",
                DeprecationWarning,
                stacklevel=2,
            )

        # Setup retry manager with optional configuration
        retry_config = kwargs.get('retry_config', None)
        if retry_config is None:
            # Use default retry configuration
            retry_config = RetryConfig()
        self.retry_manager = RetryManager(retry_config)

        # Create provider key for circuit breaker tracking
        self.provider_key = f"{self.__class__.__name__}:{self.model}"

        # Setup Glyph compression configuration
        self.glyph_config = kwargs.get('glyph_config', None)

        # Setup interaction tracing
        self.enable_tracing = kwargs.get('enable_tracing', False)
        self._traces = deque(maxlen=kwargs.get('max_traces', 100))  # Ring buffer for memory efficiency

        # Prompt caching (best-effort; provider-specific behavior).
        #
        # - Remote providers (OpenAI): supports `prompt_cache_key` pass-through (server-managed caching).
        # - Local runtimes (MLX / llama.cpp): can store KV/prefix caches in-process keyed by `prompt_cache_key`.
        self._default_prompt_cache_key: Optional[str] = None
        prompt_cache_max_entries = kwargs.get("prompt_cache_max_entries", kwargs.get("prompt_cache_max_items", 32))
        prompt_cache_ttl_s = kwargs.get("prompt_cache_ttl_s", None)
        self._prompt_cache_store = PromptCacheStore(
            max_entries=int(prompt_cache_max_entries) if prompt_cache_max_entries is not None else 32,
            default_ttl_s=prompt_cache_ttl_s,
        )

        # Provider created successfully - no event emission needed
        # (The simplified event system focuses on generation and tool events only)

        # Set default token limits if not provided
        self._initialize_token_limits()

    def _metadata_inference_parameters(self) -> Dict[str, Any]:
        """Return generation defaults declared by architecture/model metadata.

        Model-specific capabilities override architecture defaults.
        """
        merged: Dict[str, Any] = {}
        for source in (getattr(self, "architecture_config", None), getattr(self, "model_capabilities", None)):
            if not isinstance(source, dict):
                continue
            params = source.get("inference_parameters")
            if isinstance(params, dict):
                merged.update({k: v for k, v in params.items() if v is not None})
        return merged

    def _metadata_generation_default(self, name: str, fallback: Any = None) -> Any:
        params = self._metadata_inference_parameters()
        return params.get(name, fallback)

    def _apply_generation_parameter_defaults(self, params: Dict[str, Any]) -> None:
        """Apply generated/model defaults unless the caller set the value explicitly."""
        if not isinstance(params, dict) or not params:
            return
        explicit = set(getattr(self, "_explicit_generation_params", frozenset()))
        if "temperature" not in explicit and params.get("temperature") is not None:
            self.temperature = params["temperature"]
        for key in ("top_p", "top_k"):
            if key not in explicit and params.get(key) is not None:
                setattr(self, key, params[key])

    def _apply_metadata_generation_defaults(self) -> None:
        """Promote model/architecture generation defaults onto the provider instance."""
        self._apply_generation_parameter_defaults(self._metadata_inference_parameters())

    def __init_subclass__(cls, **kwargs):  # pragma: no cover
        super().__init_subclass__(**kwargs)
        # Enforce a single unload path: providers must implement `unload_model()` and must not define `unload()`.
        if "unload" in cls.__dict__:
            raise TypeError(
                f"{cls.__name__} defines unload(). "
                "Providers must implement unload_model(model_name) and must not provide any other unload entrypoint."
            )

    def _track_generation(self, prompt: str, response: Optional[GenerateResponse],
                         start_time: float, success: bool = True,
                         error: Optional[Exception] = None, stream: bool = False):
        """
        Track generation with telemetry and events.

        Args:
            prompt: Input prompt
            response: Generated response
            start_time: Generation start time
            success: Whether generation succeeded
            error: Error if failed
            stream: Whether this was a streaming generation
        """
        latency_ms = (time.time() - start_time) * 1000

        # Extract token and cost information from response
        tokens_input = None
        tokens_output = None
        cost_usd = None

        if response and response.usage:
            tokens_input = response.usage.get('prompt_tokens') or response.usage.get('input_tokens')
            tokens_output = response.usage.get('completion_tokens') or response.usage.get('output_tokens')
            # Calculate cost if possible (simplified - could be enhanced)
            total_tokens = response.usage.get('total_tokens', 0)
            if total_tokens > 0:
                # Very rough cost estimation - should be provider-specific
                cost_usd = total_tokens * 0.00002  # ~$0.02 per 1K tokens average

        # Emit comprehensive event with all data in one dict
        event_data = {
            "prompt": preview_text(prompt, max_chars=100),
            "success": success,
            "error": str(error) if error else None,
            "response_length": len(response.content) if response and response.content else 0,
            "stream": stream,
            "model": self.model,
            "provider": self.__class__.__name__,
            "duration_ms": latency_ms,
            "tokens_input": tokens_input,
            "tokens_output": tokens_output,
            "cost_usd": cost_usd
        }

        from ..events import emit_global
        emit_global(EventType.GENERATION_COMPLETED, event_data, source=self.__class__.__name__)

        # Track with structured logging (using formatted strings)
        if error:
            # Only log debug info for model not found errors to avoid duplication
            if isinstance(error, ModelNotFoundError):
                self.logger.debug(f"Model not found: {self.model}")
            else:
                self.logger.error(f"Generation failed for {self.model}: {error} (latency: {latency_ms:.2f}ms)")
        else:
            tokens_info = ""
            if response and response.usage:
                tokens_info = f" (tokens: {response.usage.get('total_tokens', 0)})"

            self.logger.info(f"Generation completed for {self.model}: {latency_ms:.2f}ms{tokens_info}")

    def _track_tool_call(self, tool_name: str, arguments: Dict[str, Any],
                        result: Optional[Any] = None, success: bool = True,
                        error: Optional[Exception] = None):
        """
        Track tool call with telemetry and events.

        Args:
            tool_name: Name of the tool
            arguments: Tool arguments
            result: Tool result
            success: Whether call succeeded
            error: Error if failed
        """
        # Emit comprehensive tool event
        event_type = EventType.TOOL_COMPLETED if success else EventType.ERROR
        event_data = {
            "tool_name": tool_name,
            "arguments": arguments,
            "result": preview_text(result, max_chars=100) if result else None,
            "error": str(error) if error else None,
            "success": success
        }

        # Add model and provider to event data
        event_data["model"] = self.model
        event_data["provider"] = self.__class__.__name__

        from ..events import emit_global
        emit_global(event_type, event_data, source=self.__class__.__name__)

        # Track with structured logging (using formatted strings)
        if error:
            self.logger.warning(f"Tool call failed: {tool_name} - {error}")
        else:
            result_info = f" (result length: {len(str(result))})" if result else ""
            self.logger.info(f"Tool call completed: {tool_name}{result_info}")

    def _capture_trace(self, prompt: str, messages: Optional[List[Dict[str, str]]],
                       system_prompt: Optional[str], tools: Optional[List[Dict[str, Any]]],
                       response: GenerateResponse, kwargs: Dict[str, Any]) -> str:
        """
        Capture interaction trace for observability.

        Args:
            prompt: Input prompt
            messages: Conversation history
            system_prompt: System prompt
            tools: Available tools
            response: Generated response
            kwargs: Additional generation parameters

        Returns:
            Trace ID (UUID string)
        """
        trace_id = str(uuid.uuid4())

        # If trace retention is disabled, still return a trace_id for correlation
        # without constructing/storing a full trace payload.
        maxlen = getattr(getattr(self, "_traces", None), "maxlen", None)
        if maxlen == 0:
            return trace_id

        # Extract generation parameters
        temperature = kwargs.get('temperature', self.temperature)
        if temperature is None:
            temperature = self.temperature
        max_tokens = kwargs.get('max_tokens', self.max_tokens)
        max_output_tokens = kwargs.get('max_output_tokens', self.max_output_tokens)
        seed = self._normalize_seed(kwargs.get('seed', self.seed))
        top_p = kwargs.get('top_p', getattr(self, 'top_p', None))
        top_k = kwargs.get('top_k', getattr(self, 'top_k', None))

        # Build parameters dict
        parameters = {
            'temperature': temperature,
            'max_tokens': max_tokens,
            'max_output_tokens': max_output_tokens,
        }
        if seed is not None:
            parameters['seed'] = seed
        if top_p is not None:
            parameters['top_p'] = top_p
        if top_k is not None:
            parameters['top_k'] = top_k

        # Create trace record
        trace = {
            'trace_id': trace_id,
            'timestamp': datetime.now().isoformat(),
            'provider': self.__class__.__name__,
            'model': self.model,
            'system_prompt': system_prompt,
            'prompt': prompt,
            'messages': messages,
            'tools': tools,
            'parameters': parameters,
            'response': {
                'content': response.content,
                'raw_response': None,  # Omit raw_response to save memory and avoid logging sensitive data
                'tool_calls': response.tool_calls,
                'finish_reason': response.finish_reason,
                'usage': response.usage,
                'generation_time_ms': response.gen_time,
            },
            'metadata': kwargs.get('trace_metadata', {})
        }

        # Store trace in ring buffer
        self._traces.append(trace)

        return trace_id

    def get_traces(self, trace_id: Optional[str] = None, last_n: Optional[int] = None) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
        """
        Retrieve interaction traces.

        Args:
            trace_id: Optional specific trace ID to retrieve
            last_n: Optional number of most recent traces to retrieve

        Returns:
            Single trace dict if trace_id provided, list of traces otherwise
        """
        if trace_id:
            # Find specific trace by ID
            for trace in self._traces:
                if trace['trace_id'] == trace_id:
                    return trace
            return None

        if last_n:
            # Return last N traces
            return list(self._traces)[-last_n:] if len(self._traces) >= last_n else list(self._traces)

        # Return all traces
        return list(self._traces)

    def _handle_api_error(self, error: Exception) -> Exception:
        """
        Convert API errors to custom exceptions.

        Args:
            error: Original exception

        Returns:
            Custom exception
        """
        def _provider_label() -> str:
            raw = getattr(self, "provider", None)
            if isinstance(raw, str) and raw.strip():
                return raw.strip()
            name = self.__class__.__name__
            return name[:-8] if name.endswith("Provider") else name

        def _configured_timeout_s() -> Optional[float]:
            v = getattr(self, "_timeout", None)
            if v is None:
                return None
            try:
                f = float(v)
            except Exception:
                return None
            return f if f > 0 else None

        def _looks_like_timeout(exc: Exception) -> bool:
            # Type-based (preferred)
            if isinstance(exc, (TimeoutError, asyncio.TimeoutError, socket.timeout)):
                return True
            cls = exc.__class__
            name = (getattr(cls, "__name__", "") or "").lower()
            mod = (getattr(cls, "__module__", "") or "").lower()
            if "timeout" in name:
                return True
            if mod.startswith(("httpx", "requests", "aiohttp")) and ("timeout" in name):
                return True

            # String-based fallback (covers wrapped SDK exceptions)
            msg = str(exc or "").lower()
            return ("timed out" in msg) or ("timeout" in msg) or ("time out" in msg)

        def _has_explicit_duration(msg: str) -> bool:
            # e.g. "... after 300s" or "... after 300.0s"
            return bool(re.search(r"\bafter\s+\d+(?:\.\d+)?\s*s\b", msg))

        # Preserve typed custom exceptions, but allow ProviderAPIError timeout messages
        # to be normalized centrally (avoid per-provider inconsistencies).
        if isinstance(error, ProviderAPIError):
            msg = str(error)
            if _looks_like_timeout(error) and not _has_explicit_duration(msg):
                t = _configured_timeout_s()
                if t is not None:
                    return ProviderAPIError(
                        f"{_provider_label()} API error: timed out after {t}s "
                        "(configured timeout; set timeout=None or default_timeout=0 for unlimited)"
                    )
                return ProviderAPIError(f"{_provider_label()} API error: timed out")
            return error

        if isinstance(error, (ModelNotFoundError, AuthenticationError, RateLimitError, InvalidRequestError)):
            return error

        # Central timeout normalization for all providers (httpx/requests/SDKs).
        if _looks_like_timeout(error):
            t = _configured_timeout_s()
            if t is not None:
                return ProviderAPIError(
                    f"{_provider_label()} API error: timed out after {t}s "
                    "(configured timeout; set timeout=None or default_timeout=0 for unlimited)"
                )
            return ProviderAPIError(f"{_provider_label()} API error: timed out")

        error_str = str(error).lower()

        if "rate" in error_str and "limit" in error_str:
            return RateLimitError(f"Rate limit exceeded: {error}")
        elif "auth" in error_str or "api key" in error_str or "unauthorized" in error_str:
            return AuthenticationError(f"Authentication failed: {error}")
        elif "invalid" in error_str or "bad request" in error_str:
            return InvalidRequestError(f"Invalid request: {error}")
        else:
            return ProviderAPIError(f"API error: {error}")

    @staticmethod
    def _normalize_thinking_request(thinking: Optional[Union[bool, str]]) -> Tuple[Optional[bool], Optional[str]]:
        """Normalize `thinking=` into (enabled, level).

        - enabled: True/False/None (None == "auto")
        - level: Optional[str] in {"minimal","low","medium","high","xhigh"} when requested

        Notes:
        - thinking="none" is accepted as an alias for thinking="off".
        - "extra high" / "extra_high" / "extra-high" are accepted as aliases for "xhigh".
        """
        if thinking is None:
            return None, None

        if isinstance(thinking, bool):
            return thinking, None

        if isinstance(thinking, str):
            s = thinking.strip().lower()
            if not s or s == "auto":
                return None, None
            if s in {"on", "true", "yes"}:
                return True, None
            if s in {"off", "false", "no", "none"}:
                return False, None
            # Normalize common separator variants for user ergonomics.
            s_norm = re.sub(r"[\\s\\-]+", "_", s).strip("_")
            if s_norm in {"extra_high", "x_high"}:
                return True, "xhigh"
            if s_norm in {"minimal", "low", "medium", "high", "xhigh"}:
                return True, s_norm

        raise ValueError(
            'thinking must be one of: None, bool, "auto", "on", "off", "none", '
            '"minimal", "low", "medium", "high", "xhigh" (aliases: "extra high" -> "xhigh")'
        )

    def _model_reasoning_levels(self) -> List[str]:
        levels = None
        for src in (self.model_capabilities, self.architecture_config):
            if not isinstance(src, dict):
                continue
            value = src.get("reasoning_levels")
            if isinstance(value, list) and value:
                levels = value
                break
        if not isinstance(levels, list):
            return []
        out: List[str] = []
        for x in levels:
            if isinstance(x, str) and x.strip():
                out.append(x.strip().lower())
        # Deduplicate while preserving order.
        seen: set[str] = set()
        uniq: List[str] = []
        for x in out:
            if x in seen:
                continue
            seen.add(x)
            uniq.append(x)
        return uniq

    def _is_reasoning_model(self) -> bool:
        """Check if this model has reasoning/thinking capabilities.

        Uses thinking_support from model_capabilities.json as the source of truth.
        Falls back to architecture-level indicators when model capabilities are absent.
        """
        caps = self.model_capabilities if isinstance(self.model_capabilities, dict) else {}
        if caps.get("thinking_support") is True:
            return True
        arch = self.architecture_config if isinstance(self.architecture_config, dict) else {}
        if arch.get("reasoning_support") is True:
            return True
        return False

    @staticmethod
    def _thinking_level_rank(level: Optional[str]) -> Optional[int]:
        order = ("minimal", "low", "medium", "high", "xhigh")
        if not isinstance(level, str):
            return None
        lvl = level.strip().lower()
        try:
            return order.index(lvl)
        except ValueError:
            return None

    @classmethod
    def _nearest_supported_thinking_level(cls, requested_level: str, supported_levels: List[str]) -> Optional[str]:
        """Map a requested thinking level to the nearest supported level (best-effort)."""
        if not isinstance(requested_level, str) or not requested_level.strip():
            return None
        req = requested_level.strip().lower()
        req_rank = cls._thinking_level_rank(req)
        if req_rank is None:
            return None

        ranked: List[Tuple[int, str]] = []
        for lvl in supported_levels or []:
            if not isinstance(lvl, str):
                continue
            s = lvl.strip().lower()
            if s == "none":
                # "none" is a disable signal for some providers, not an effort level.
                continue
            r = cls._thinking_level_rank(s)
            if r is None:
                continue
            ranked.append((r, s))
        if not ranked:
            return None

        # Nearest in ordinal space; ties prefer the lower effort (less surprising escalation).
        ranked.sort(key=lambda x: x[0])
        best: Optional[str] = None
        best_dist: Optional[int] = None
        best_rank: Optional[int] = None
        for r, s in ranked:
            dist = abs(r - req_rank)
            if best_dist is None or dist < best_dist or (dist == best_dist and best_rank is not None and r < best_rank):
                best = s
                best_dist = dist
                best_rank = r
        return best

    def _model_supports_reasoning_output(self) -> bool:
        """Return True if the model is expected to emit a separate reasoning/thinking channel."""
        caps = self.model_capabilities if isinstance(self.model_capabilities, dict) else {}
        arch = self.architecture_config if isinstance(self.architecture_config, dict) else {}

        resp_fmt = str((caps.get("response_format") or arch.get("response_format") or "")).strip().lower()
        if resp_fmt == "harmony":
            return True
        msg_fmt = str((arch.get("message_format") or "")).strip().lower()
        if msg_fmt == "harmony":
            return True

        if caps.get("thinking_support") is True:
            return True
        if isinstance(caps.get("thinking_tags"), (list, tuple)) and len(caps.get("thinking_tags")) == 2:
            return True
        if isinstance(caps.get("thinking_output_field"), str) and caps.get("thinking_output_field").strip():
            return True
        if self._model_reasoning_levels():
            return True

        if arch.get("reasoning_support") is True:
            return True
        if isinstance(arch.get("thinking_tags"), (list, tuple)) and len(arch.get("thinking_tags")) == 2:
            return True
        if isinstance(arch.get("thinking_output_field"), str) and arch.get("thinking_output_field").strip():
            return True
        if isinstance(arch.get("reasoning_levels"), list) and arch.get("reasoning_levels"):
            return True

        return False

    def _model_supports_reasoning_control(self) -> bool:
        """Return True if the model advertises a request-side reasoning/thinking control surface."""
        caps = self.model_capabilities if isinstance(self.model_capabilities, dict) else {}
        arch = self.architecture_config if isinstance(self.architecture_config, dict) else {}

        # Harmony response format implies a request-side Reasoning control via system prompt injection.
        resp_fmt = str((caps.get("response_format") or arch.get("response_format") or "")).strip().lower()
        if resp_fmt == "harmony":
            return True
        msg_fmt = str((arch.get("message_format") or "")).strip().lower()
        if msg_fmt == "harmony":
            return True

        # Explicit effort enums (OpenAI reasoning_effort, Harmony Reasoning: ..., etc.).
        if self._model_reasoning_levels():
            return True

        # Explicit budget knobs (e.g., Claude extended thinking budget).
        if caps.get("thinking_budget") is True:
            return True

        # Prompt-level disable tokens (e.g., GLM /nothink, Qwen /no_think).
        for src in (caps, arch):
            token = src.get("thinking_control")
            if isinstance(token, str) and token.strip():
                return True

        return False

    def _is_parameter_supported(self, param: str) -> bool:
        """Check if a generation parameter is supported by this model.

        Uses unsupported_parameters from model_capabilities.json.
        Absent field means all standard parameters are supported (backward-compatible default).
        """
        caps = self.model_capabilities if isinstance(self.model_capabilities, dict) else {}
        blocked = caps.get("unsupported_parameters")
        if blocked is None:
            return True
        return param not in blocked

    def _get_token_param_name(self) -> str:
        """Return the API parameter name for output token limit.

        Uses token_param_name from model_capabilities.json.
        Default when absent: 'max_tokens'.
        """
        caps = self.model_capabilities if isinstance(self.model_capabilities, dict) else {}
        return caps.get("token_param_name", "max_tokens")

    def _model_supports_thinking_control(self) -> bool:
        # Backward-compatible alias. Prefer `_model_supports_reasoning_control()` / `_model_supports_reasoning_output()`.
        return self._model_supports_reasoning_control()

    @staticmethod
    def _coerce_thinking_handling(value: Any) -> ThinkingControlHandling:
        if isinstance(value, ThinkingControlHandling):
            return value
        if value is True:
            return ThinkingControlHandling(handled_enable_disable=True, handled_level=True)
        return ThinkingControlHandling()

    def _apply_thinking_request(
        self,
        *,
        thinking: Optional[Union[bool, str]],
        prompt: str,
        messages: Optional[List[Dict[str, str]]],
        system_prompt: Optional[str],
        kwargs: Dict[str, Any],
    ) -> Tuple[str, Optional[List[Dict[str, str]]], Optional[str], Dict[str, Any], Optional[Dict[str, Any]]]:
        """Apply unified thinking controls to the request."""
        enabled, requested_level = self._normalize_thinking_request(thinking)
        requested_enabled = enabled
        if enabled is None and requested_level is None:
            return prompt, messages, system_prompt, kwargs, None

        reasoning_levels = self._model_reasoning_levels()
        effective_level = requested_level

        if requested_level is not None and reasoning_levels and requested_level not in reasoning_levels:
            mapped = self._nearest_supported_thinking_level(requested_level, reasoning_levels)
            if mapped:
                warnings.warn(
                    f"thinking level '{requested_level}' requested but not supported for model '{self.model}' "
                    f"(supported: {reasoning_levels}); using '{mapped}'.",
                    RuntimeWarning,
                    stacklevel=3,
                )
                effective_level = mapped
            else:
                warnings.warn(
                    f"thinking level '{requested_level}' requested but not supported for model '{self.model}' "
                    f"(supported: {reasoning_levels}); treating as thinking='on'.",
                    RuntimeWarning,
                    stacklevel=3,
                )
                effective_level = None
            enabled = True

        # When a specific thinking level is requested but the model does not advertise supported
        # reasoning levels, treat the level as a *best-effort hint* for provider-side token budgets.
        # We still enable thinking, but do not warn: many open-weight stacks expose the knob via
        # template kwargs (enable_thinking) and/or server-side budgets without a stable upstream
        # "effort enum" to record in assets.
        if effective_level is not None and not reasoning_levels:
            enabled = True

        supports_reasoning_output = self._model_supports_reasoning_output()
        supports_reasoning_control = self._model_supports_reasoning_control()

        handled_model_enable_disable = False
        handled_model_level = False
        harmony_effective_level: Optional[str] = None

        # Harmony (GPT-OSS): control via system message `Reasoning: low|medium|high`.
        msg_fmt = str((self.architecture_config or {}).get("message_format") or "").strip().lower()
        resp_fmt = str((self.model_capabilities or {}).get("response_format") or "").strip().lower()
        is_harmony = msg_fmt == "harmony" or resp_fmt == "harmony"
        if is_harmony:
            target_level: Optional[str] = None
            if effective_level is not None:
                target_level = effective_level
            elif enabled is False:
                warnings.warn(
                    f"thinking='off' requested for Harmony model '{self.model}', but GPT-OSS reasoning traces "
                    "cannot be fully disabled; using Reasoning: low.",
                    RuntimeWarning,
                    stacklevel=3,
                )
                target_level = "low"
            elif enabled is True:
                # Make the default explicit when the caller opts-in.
                target_level = "medium"

            if target_level:
                harmony_effective_level = target_level
                line = f"Reasoning: {target_level}"
                if isinstance(system_prompt, str) and system_prompt.strip():
                    # Replace any existing Reasoning line; otherwise prepend.
                    if re.search(r"(?mi)^\\s*Reasoning\\s*:\\s*(low|medium|high)\\s*$", system_prompt):
                        system_prompt = re.sub(
                            r"(?mi)^\\s*Reasoning\\s*:\\s*(low|medium|high)\\s*$",
                            line,
                            system_prompt,
                            count=1,
                        )
                    else:
                        system_prompt = f"{line}\n{system_prompt}"
                else:
                    system_prompt = line
                handled_model_enable_disable = True
                handled_model_level = True

        # Provider-specific hook: allow providers to see both requested and effective levels.
        hook = getattr(self, "_apply_provider_thinking_kwargs", None)
        raw_provider_handling: Any = False
        if callable(hook):
            call_kwargs: Dict[str, Any] = {
                "enabled": enabled,
                "level": effective_level,
                "kwargs": kwargs,
                "requested_level": requested_level,
                "supported_levels": reasoning_levels,
            }
            filtered_kwargs: Dict[str, Any] = {"enabled": enabled, "level": effective_level, "kwargs": kwargs}
            try:
                sig = inspect.signature(hook)
                params = sig.parameters
                accepts_var_kw = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values())
                if accepts_var_kw:
                    filtered_kwargs = call_kwargs
                else:
                    filtered_kwargs = {k: v for k, v in call_kwargs.items() if k in params}
            except Exception:
                filtered_kwargs = {"enabled": enabled, "level": effective_level, "kwargs": kwargs}

            kwargs, raw_provider_handling = hook(**filtered_kwargs)

        provider_handling = self._coerce_thinking_handling(raw_provider_handling)
        # IMPORTANT: Avoid prompt-injection based "reasoning effort" toggles here.
        #
        # Thinking controls should be mapped to backend-native request fields whenever possible
        # (e.g., vLLM `extra_body.chat_template_kwargs.enable_thinking`, LM Studio
        # `chat_template_kwargs.enable_thinking`, Ollama `think`). This keeps requests clean and
        # avoids polluting system prompts (and provider logs) with framework-specific directives.

        # Qwen3 / Qwen3.5 hard-switch (LM Studio + llama.cpp/GGUF): disable thinking without system-prompt hacks.
        #
        # Qwen's official guidance (and common Qwen chat templates) support a strict, stateless
        # "no-thinking" mode by inserting an empty think block in the *assistant generation prompt*
        # (i.e. right after the assistant start token) before generation:
        #   <think>\n\n</think>\n\n
        #
        # Some LM Studio runtimes do not reliably honor `chat_template_kwargs.enable_thinking`
        # for all model formats. As a robust fallback for `thinking="off"/"none"`, we append a
        # final assistant message containing only the empty think block and clear `prompt` so
        # OpenAI-compatible providers don't append an extra user turn after it.
        provider_id = str(getattr(self, "provider", "") or "").strip().lower()
        is_hf_gguf = provider_id == "huggingface" and str(getattr(self, "model_type", "") or "").strip().lower() == "gguf"
        if (
            enabled is False
            and (provider_id == "lmstudio" or is_hf_gguf)
            and self.architecture in {"qwen3", "qwen3_5", "qwen3_6"}
            and not provider_handling.handled_enable_disable
        ):
            marker = "<think>\n\n</think>\n\n"
            new_messages: List[Dict[str, str]] = []
            if isinstance(messages, list) and messages:
                for m in messages:
                    if isinstance(m, dict):
                        # Only keep string roles/contents for compatibility with OpenAI chat.
                        r = m.get("role")
                        c = m.get("content")
                        if isinstance(r, str) and r.strip() and isinstance(c, str):
                            new_messages.append({"role": r.strip(), "content": c})

            if isinstance(prompt, str) and prompt.strip():
                new_messages.append({"role": "user", "content": prompt.strip()})
                prompt = ""

            if new_messages:
                last = new_messages[-1]
                if not (
                    isinstance(last, dict)
                    and last.get("role") == "assistant"
                    and isinstance(last.get("content"), str)
                    and last.get("content") == marker
                ):
                    new_messages.append({"role": "assistant", "content": marker})
                messages = new_messages
                handled_model_enable_disable = True

        # Model-level control token for disabling thinking (e.g., GLM `/nothink`).
        #
        # Apply prompt-level control *only* when the provider did not already handle the request
        # with a backend-native knob (e.g., vLLM/LM Studio `enable_thinking`). This avoids injecting
        # tokens that some templates do not interpret as controls.
        thinking_control = None
        for src in (self.model_capabilities, self.architecture_config):
            if not isinstance(src, dict):
                continue
            token = src.get("thinking_control")
            if isinstance(token, str) and token.strip():
                thinking_control = token.strip()

        if enabled is False and thinking_control and not provider_handling.handled_enable_disable:
            handled_model_enable_disable = True

            def _append_control(text: str) -> str:
                if thinking_control in text:
                    return text
                return f"{text.rstrip()}\n{thinking_control}".strip()

            if isinstance(prompt, str) and prompt.strip():
                prompt = _append_control(prompt)
            elif isinstance(messages, list) and messages:
                # Append to the most recent user turn, if possible.
                new_messages: List[Dict[str, str]] = []
                appended = False
                for m in messages:
                    if not isinstance(m, dict):
                        continue
                    new_messages.append(dict(m))
                for m in reversed(new_messages):
                    if m.get("role") == "user" and isinstance(m.get("content"), str) and m["content"].strip():
                        m["content"] = _append_control(m["content"])
                        appended = True
                        break
                messages = new_messages
                if not appended:
                    warnings.warn(
                        f"thinking='off' requested for model '{self.model}', but no user prompt was available "
                        f"to append thinking_control='{thinking_control}'.",
                        RuntimeWarning,
                        stacklevel=3,
                    )

        requesting_enable = enabled is True or effective_level is not None
        requesting_disable = enabled is False
        requesting_level = requested_level is not None

        # Warn about unsupported thinking capability only when the caller is *enabling* thinking.
        # When thinking="off"/"none", a non-thinking model is already in the desired mode.
        if requesting_enable and not supports_reasoning_output:
            warnings.warn(
                f"thinking={thinking!r} requested but model '{self.model}' is not marked as thinking-capable "
                "in model_capabilities.json; the request may be ignored.",
                RuntimeWarning,
                stacklevel=3,
            )

        handled_enable_disable = handled_model_enable_disable or provider_handling.handled_enable_disable
        handled_level = handled_model_level or provider_handling.handled_level

        if requesting_level and not handled_level and effective_level is not None:
            warnings.warn(
                f"thinking level '{requested_level}' requested for model '{self.model}', but provider "
                f"'{self.provider or self.__class__.__name__}' cannot enforce effort scaling; treating as thinking='on'.",
                RuntimeWarning,
                stacklevel=3,
            )

        # Warn about missing mapping only when the request is expected to take effect.
        if requesting_enable and not handled_enable_disable and not handled_level:
            warnings.warn(
                f"thinking={thinking!r} requested but provider '{self.provider or self.__class__.__name__}' "
                "does not implement a thinking control mapping for this model; the request may be ignored.",
                RuntimeWarning,
                stacklevel=3,
            )
        if requesting_disable and supports_reasoning_output and not handled_enable_disable:
            if supports_reasoning_control:
                warnings.warn(
                    f"thinking={thinking!r} requested but provider '{self.provider or self.__class__.__name__}' "
                    "does not implement a thinking disable mapping for this model; the request may be ignored.",
                    RuntimeWarning,
                    stacklevel=3,
                )
            else:
                warnings.warn(
                    f"thinking={thinking!r} requested but model '{self.model}' does not advertise a thinking control "
                    "surface in assets; the request may be ignored.",
                    RuntimeWarning,
                    stacklevel=3,
                )

        def _request_repr(enabled_val: Optional[bool], level_val: Optional[str]) -> Optional[str]:
            if isinstance(level_val, str) and level_val:
                return level_val
            if enabled_val is True:
                return "on"
            if enabled_val is False:
                return "off"
            return None

        effective_enabled: Optional[bool] = None
        effective_level_meta: Optional[str] = None

        if isinstance(harmony_effective_level, str) and harmony_effective_level:
            effective_enabled = True
            effective_level_meta = harmony_effective_level

        # Best-effort infer the effective control after provider hooks.
        # Prefer explicit provider-native knobs over our internal request vars.
        effort = kwargs.get("reasoning_effort")
        if isinstance(effort, str) and effort.strip():
            effort_norm = effort.strip().lower()
            if effort_norm == "none":
                effective_enabled = False
                effective_level_meta = None
            else:
                effective_enabled = True
                effective_level_meta = effort_norm

        think = kwargs.get("think")
        if isinstance(think, bool):
            effective_enabled = bool(think)
            if not think:
                effective_level_meta = None
        elif isinstance(think, str) and think.strip():
            effective_enabled = True
            effective_level_meta = think.strip().lower()

        anthropic_thinking = kwargs.get("thinking")
        if isinstance(anthropic_thinking, dict):
            typ = str(anthropic_thinking.get("type") or "").strip().lower()
            if typ == "disabled":
                effective_enabled = False
                effective_level_meta = None
            elif typ:
                effective_enabled = True
                output_config = kwargs.get("output_config")
                if isinstance(output_config, dict):
                    eff = output_config.get("effort")
                    if isinstance(eff, str) and eff.strip():
                        effective_level_meta = eff.strip().lower()

        if effective_enabled is None:
            if enabled is False:
                effective_enabled = False
            elif enabled is True or effective_level is not None:
                effective_enabled = True

        if effective_level_meta is None and effective_level is not None and handled_level:
            effective_level_meta = effective_level

        thinking_requested = _request_repr(requested_enabled, requested_level)
        thinking_effective: Optional[str]
        if effective_enabled is False:
            thinking_effective = "off"
        elif isinstance(effective_level_meta, str) and effective_level_meta:
            thinking_effective = effective_level_meta
        elif effective_enabled is True:
            thinking_effective = "on"
        else:
            thinking_effective = None

        thinking_meta: Dict[str, Any] = {
            "thinking_requested": thinking_requested,
            "thinking_effective": thinking_effective,
            "thinking_level_requested": requested_level,
            "thinking_level_effective": effective_level_meta,
            "thinking_supported_levels": list(reasoning_levels) if reasoning_levels else None,
            "thinking_handled_enable_disable": bool(handled_enable_disable),
            "thinking_handled_level": bool(handled_level),
            "thinking_supports_output": bool(supports_reasoning_output),
            "thinking_supports_control": bool(supports_reasoning_control),
        }
        # Keep metadata clean (omit nulls).
        thinking_meta = {k: v for k, v in thinking_meta.items() if v is not None}

        return prompt, messages, system_prompt, kwargs, thinking_meta

    def _apply_provider_thinking_kwargs(
        self,
        *,
        enabled: Optional[bool],
        level: Optional[str],
        kwargs: Dict[str, Any],
    ) -> Tuple[Dict[str, Any], bool]:
        """Provider-specific thinking knob hook (default: unsupported)."""
        _ = (enabled, level)
        return kwargs, False

    def _supports_developer_messages(self) -> bool:
        """Return True when the provider API can accept OpenAI `developer` messages directly."""
        return False

    @staticmethod
    def _message_content_to_text(content: Any) -> str:
        if content is None:
            return ""
        if isinstance(content, str):
            return content
        if isinstance(content, list):
            parts: List[str] = []
            for item in content:
                if isinstance(item, dict):
                    typ = str(item.get("type") or "").strip()
                    if typ == "text" and item.get("text") is not None:
                        parts.append(str(item.get("text") or ""))
                    elif item.get("text") is not None:
                        parts.append(str(item.get("text") or ""))
                    else:
                        try:
                            parts.append(json.dumps(item, ensure_ascii=False))
                        except Exception:
                            parts.append(str(item))
                elif hasattr(item, "type") and getattr(item, "text", None) is not None:
                    parts.append(str(getattr(item, "text") or ""))
                else:
                    parts.append(str(item))
            return "\n".join(part for part in parts if part)
        try:
            return json.dumps(content, ensure_ascii=False)
        except Exception:
            return str(content)

    def _normalize_developer_messages(
        self,
        messages: Optional[List[Dict[str, Any]]],
        system_prompt: Optional[str],
    ) -> Tuple[Optional[List[Dict[str, Any]]], Optional[str]]:
        """Normalize OpenAI `developer` messages for providers that only support system/user roles."""
        if not isinstance(messages, list) or not messages:
            return messages, system_prompt
        if self._supports_developer_messages():
            return messages, system_prompt

        developer_parts: List[str] = []
        normalized_messages: List[Dict[str, Any]] = []
        changed = False

        for msg in messages:
            if not isinstance(msg, dict):
                normalized_messages.append(msg)
                continue
            role = msg.get("role")
            role_norm = str(role or "").strip().lower()
            if role_norm != "developer":
                normalized_messages.append(dict(msg))
                continue

            changed = True
            text = self._message_content_to_text(msg.get("content")).strip()
            if text:
                developer_parts.append(text)

        if not changed:
            return messages, system_prompt

        prompt_parts: List[str] = []
        if isinstance(system_prompt, str) and system_prompt.strip():
            prompt_parts.append(system_prompt.strip())
        prompt_parts.extend(developer_parts)
        merged_system_prompt = "\n\n".join(prompt_parts) if prompt_parts else system_prompt

        return normalized_messages, merged_system_prompt

    @staticmethod
    def _normalize_system_prompt_alias(
        system_prompt: Optional[str],
        kwargs: Dict[str, Any],
        *,
        stacklevel: int,
    ) -> Optional[str]:
        """Accept `system=` as a warned alias for `system_prompt=` at the provider boundary."""
        if "system" not in kwargs:
            return system_prompt

        system_alias = kwargs.pop("system")
        if system_prompt is None:
            warnings.warn(
                "`system=` is accepted as an alias for `system_prompt=`, but `system_prompt=` is the "
                "preferred AbstractCore parameter.",
                UserWarning,
                stacklevel=stacklevel,
            )
            return system_alias

        warnings.warn(
            "Both `system_prompt=` and `system=` were provided; using `system_prompt=` and ignoring `system=`.",
            UserWarning,
            stacklevel=stacklevel,
        )
        return system_prompt

    @staticmethod
    def _is_acore_output_request(output: Any) -> bool:
        """Return True when ``output`` is an AbstractCore multimodal selector."""
        return is_output_request(output)

    @staticmethod
    def _normalize_output_specs(output: Any) -> List[Dict[str, Any]]:
        return normalize_output_specs(output)

    @staticmethod
    def _normalize_output_spec(output: Any) -> Dict[str, Any]:
        return normalize_output_spec(output)

    @classmethod
    def _output_request_with_progress_callbacks(
        cls,
        output: Any,
        callbacks: Dict[str, Any],
    ) -> Tuple[Any, bool]:
        """Attach top-level generated-media progress callbacks to output specs."""
        effective_callbacks = {
            str(key): value
            for key, value in callbacks.items()
            if value is not None
        }
        if not effective_callbacks or not cls._is_acore_output_request(output):
            return output, False

        specs = cls._normalize_output_specs(output)
        changed = False
        patched_specs: List[Dict[str, Any]] = []
        for spec in specs:
            patched = dict(spec)
            modality = str(patched.get("modality") or "").strip().lower()
            if modality in {"image", "video"}:
                nested_extra = patched.get("extra")
                nested_keys = set(nested_extra) if isinstance(nested_extra, dict) else set()
                for key, value in effective_callbacks.items():
                    if key in patched or key in nested_keys:
                        continue
                    patched[key] = value
                    changed = True
            patched_specs.append(patched)

        if not changed:
            return output, False
        if isinstance(output, (list, tuple)):
            return patched_specs, True
        return patched_specs[0], True

    @staticmethod
    def _coerce_media_items(media: Any) -> List[Any]:
        if media is None:
            return []
        if isinstance(media, (str, bytes, bytearray, dict)) or hasattr(media, "media_type"):
            return [media]
        if isinstance(media, Sequence):
            return list(media)
        return [media]

    @staticmethod
    def _media_role(item: Any) -> Optional[str]:
        role = None
        if isinstance(item, dict):
            role = item.get("role")
            if role is None and isinstance(item.get("metadata"), dict):
                role = item["metadata"].get("role")
        elif hasattr(item, "metadata"):
            metadata = getattr(item, "metadata", None)
            if isinstance(metadata, dict):
                role = metadata.get("role")
        if isinstance(role, str) and role.strip():
            return role.strip().lower()
        return None

    @staticmethod
    def _media_type(item: Any, *, fallback: Optional[str] = None) -> Optional[str]:
        if isinstance(item, dict):
            raw = item.get("media_type", item.get("mediaType"))
            if raw is None:
                raw = item.get("type")
            if isinstance(raw, str) and raw.strip().lower() in {"image", "audio", "video", "document", "text"}:
                return raw.strip().lower()
            mime = item.get("mime_type", item.get("mimeType", item.get("mime")))
            if mime is None:
                mime = item.get("content_type", item.get("contentType"))
            path = item.get("file_path", item.get("filePath", item.get("path")))
        elif hasattr(item, "media_type"):
            raw = getattr(item, "media_type", None)
            value = getattr(raw, "value", raw)
            if isinstance(value, str) and value.strip():
                return value.strip().lower()
            mime = getattr(item, "mime_type", None)
            path = getattr(item, "file_path", None)
        elif isinstance(item, (bytes, bytearray)):
            return fallback
        else:
            mime = None
            path = item if isinstance(item, str) else None

        if isinstance(mime, str):
            ml = mime.lower()
            if ml.startswith("image/"):
                return "image"
            if ml.startswith("audio/"):
                return "audio"
            if ml.startswith("video/"):
                return "video"
            if ml.startswith("text/"):
                return "text"

        if isinstance(path, str) and path.strip():
            import mimetypes

            guessed, _ = mimetypes.guess_type(path)
            if isinstance(guessed, str):
                gl = guessed.lower()
                if gl.startswith("image/"):
                    return "image"
                if gl.startswith("audio/"):
                    return "audio"
                if gl.startswith("video/"):
                    return "video"
                if gl.startswith("text/"):
                    return "text"
        return fallback

    @staticmethod
    def _media_payload(item: Any) -> Any:
        def _maybe_decode_base64(content: Any, content_format: Any) -> Any:
            raw_format = getattr(content_format, "value", content_format)
            if not (isinstance(raw_format, str) and raw_format.strip().lower() == "base64"):
                return content
            if isinstance(content, (bytes, bytearray)):
                return bytes(content)
            if isinstance(content, str):
                raw = content.strip()
                if "," in raw and raw.lower().startswith("data:"):
                    raw = raw.split(",", 1)[1]
                return base64.b64decode("".join(raw.split()), validate=False)
            return content

        if isinstance(item, dict):
            for key in ("file_path", "filePath", "path"):
                value = item.get(key)
                if isinstance(value, str) and value.strip():
                    return value
            if "content" in item:
                return _maybe_decode_base64(
                    item.get("content"),
                    item.get("content_format", item.get("contentFormat")),
                )
            return item
        if hasattr(item, "file_path") and getattr(item, "file_path", None):
            return getattr(item, "file_path")
        if hasattr(item, "content"):
            return _maybe_decode_base64(getattr(item, "content"), getattr(item, "content_format", None))
        return item

    def _context_media_for_text_generation(self, media: Any) -> Any:
        """Return only media explicitly marked as text-model context."""
        items = self._coerce_media_items(media)
        if not items:
            return None
        context_items = [item for item in items if self._media_role(item) == "context"]
        return context_items or None

    @staticmethod
    def _output_plugin_kwargs(spec: Dict[str, Any], *, exclude: Optional[set[str]] = None) -> Dict[str, Any]:
        return output_plugin_kwargs(spec, exclude=exclude)

    @staticmethod
    def _artifact_or_data(value: Any) -> tuple[Optional[bytes], Optional[Dict[str, Any]], Dict[str, Any]]:
        if isinstance(value, (bytes, bytearray)):
            return bytes(value), None, {}
        if is_artifact_ref(value):
            return None, dict(value), {}
        if isinstance(value, dict):
            for key in ("data", "bytes", "content"):
                content = value.get(key)
                if isinstance(content, (bytes, bytearray)):
                    meta = {k: v for k, v in value.items() if k != key}
                    return bytes(content), None, meta
            return None, dict(value) if is_artifact_ref(value) else None, dict(value)
        content = getattr(value, "data", None)
        if isinstance(content, (bytes, bytearray)):
            metadata = getattr(value, "metadata", None)
            meta = dict(metadata) if isinstance(metadata, dict) else {}
            for attr, key in (("mime_type", "mime_type"), ("content_type", "content_type")):
                raw = getattr(value, attr, None)
                if raw is not None and key not in meta:
                    meta[key] = raw
            return bytes(content), None, meta
        return None, None, {"value": value}

    @staticmethod
    def _artifact_ref_from_store_result(value: Any) -> Optional[Dict[str, Any]]:
        if is_artifact_ref(value):
            return dict(value)
        if isinstance(value, str) and value.strip():
            return {"$artifact": value.strip()}
        raw = getattr(value, "artifact_id", None)
        if isinstance(raw, str) and raw.strip():
            return {"$artifact": raw.strip()}
        return None

    @staticmethod
    def _store_generated_data(
        data: Optional[bytes],
        *,
        artifact_store: Optional[Any],
        content_type: str,
        spec: Dict[str, Any],
    ) -> tuple[Optional[bytes], Optional[Dict[str, Any]]]:
        if data is None or artifact_store is None:
            return data, None
        store = getattr(artifact_store, "store", None)
        if not callable(store):
            return data, None
        stored = store(
            bytes(data),
            content_type=content_type,
            run_id=spec.get("run_id"),
            tags=spec.get("tags") if isinstance(spec.get("tags"), dict) else None,
            artifact_id=spec.get("artifact_id") if isinstance(spec.get("artifact_id"), str) else None,
        )
        ref = BaseProvider._artifact_ref_from_store_result(stored)
        if ref is not None:
            return None, ref
        return data, None

    @staticmethod
    def _resource_id_from_clone_result(value: Any) -> tuple[str, Dict[str, Any]]:
        if isinstance(value, str) and value.strip():
            return value.strip(), {"raw": value}
        if isinstance(value, dict):
            for key in ("resource_id", "voice_id", "id", "name"):
                raw = value.get(key)
                if isinstance(raw, str) and raw.strip():
                    return raw.strip(), dict(value)
            voice = value.get("voice")
            if isinstance(voice, dict):
                for key in ("resource_id", "voice_id", "id", "name"):
                    raw = voice.get(key)
                    if isinstance(raw, str) and raw.strip():
                        return raw.strip(), dict(value)
        for key in ("resource_id", "voice_id", "id", "name"):
            raw = getattr(value, key, None)
            if isinstance(raw, str) and raw.strip():
                return raw.strip(), {"raw": value}
        raise ValueError("Voice clone backend did not return a usable voice id.")

    def _generate_multimodal_response(
        self,
        *,
        prompt: str,
        output: Any,
        media: Any,
        stream: bool,
        response_model: Optional[Type[BaseModel]],
        artifact_store: Optional[Any],
        partial: bool,
        text_generation_kwargs: Dict[str, Any],
    ) -> MultimodalGenerateResponse:
        specs = self._normalize_output_specs(output)
        if any(s.get("modality") == "text" and not s.get("task") for s in specs) and not str(prompt or "").strip():
            audio_items = [
                item
                for item in self._coerce_media_items(media)
                if self._media_type(item, fallback="audio" if isinstance(item, (bytes, bytearray)) else None) == "audio"
            ]
            if audio_items:
                for spec in specs:
                    if spec.get("modality") == "text" and not spec.get("task"):
                        spec["task"] = "transcription"
        if stream:
            raise ValueError("generate(..., output=...) does not support stream=True for non-text outputs yet.")
        if response_model is not None and any(s.get("modality") != "text" for s in specs):
            raise ValueError("response_model cannot be combined with non-text multimodal outputs.")

        text_specs = [s for s in specs if s.get("modality") == "text" and s.get("task") != "transcription"]
        non_text_specs = [s for s in specs if not (s.get("modality") == "text" and s.get("task") != "transcription")]
        if len(text_specs) > 1 and any("source" not in s for s in non_text_specs):
            raise ValueError("Multiple text outputs require explicit source binding for non-text outputs.")

        result = MultimodalGenerateResponse(metadata={"model": self.model, "provider": self.__class__.__name__})

        generated_text: Optional[GenerateResponse] = None
        if text_specs and len(specs) > 1:
            text_kwargs = dict(text_generation_kwargs)
            text_kwargs["media"] = self._context_media_for_text_generation(media)
            generated_text = self.generate_with_telemetry(
                prompt=prompt,
                stream=False,
                **text_kwargs,
            )
            if not isinstance(generated_text, GenerateResponse):
                raise ValueError("Text output generation did not return GenerateResponse.")
            result.text = generated_text

        if len(specs) == 1 and specs[0].get("modality") == "text" and specs[0].get("task") != "transcription":
            generated_text = self.generate_with_telemetry(
                prompt=prompt,
                stream=False,
                **dict(text_generation_kwargs),
            )
            if not isinstance(generated_text, GenerateResponse):
                raise ValueError("Text output generation did not return GenerateResponse.")
            result.text = generated_text
            return result

        for spec in non_text_specs:
            try:
                source_text = str(spec.get("text") or spec.get("prompt") or "")
                if not source_text:
                    source = spec.get("source")
                    if source and generated_text is None:
                        raise ValueError(f"Unknown text source {source!r}; no generated text output is available.")
                    if generated_text is not None:
                        source_text = str(generated_text.content or "")
                    else:
                        source_text = str(prompt or "")
                self._run_multimodal_spec(
                    result=result,
                    spec=spec,
                    prompt=source_text,
                    media=media,
                    artifact_store=artifact_store,
                )
            except Exception as e:
                if partial:
                    from ..core.multimodal_generation import GenerationIssue

                    result.errors.append(
                        GenerationIssue(
                            modality=str(spec.get("modality") or "unknown"),
                            task=str(spec.get("task") or "unknown"),
                            message=f"{type(e).__name__}: {e}",
                        )
                    )
                    continue
                raise

        return result

    def _run_multimodal_spec(
        self,
        *,
        result: MultimodalGenerateResponse,
        spec: Dict[str, Any],
        prompt: str,
        media: Any,
        artifact_store: Optional[Any],
    ) -> None:
        modality = str(spec.get("modality") or "").lower()
        if modality == "image":
            self._run_image_output(result=result, spec=spec, prompt=prompt, media=media, artifact_store=artifact_store)
            return
        if modality == "video":
            self._run_video_output(result=result, spec=spec, prompt=prompt, media=media, artifact_store=artifact_store)
            return
        if modality == "voice":
            self._run_voice_output(result=result, spec=spec, prompt=prompt, media=media, artifact_store=artifact_store)
            return
        if modality == "music":
            self._run_music_output(result=result, spec=spec, prompt=prompt, media=media, artifact_store=artifact_store)
            return
        if modality == "text" and spec.get("task") == "transcription":
            self._run_transcription_output(result=result, spec=spec, media=media, artifact_store=artifact_store)
            return
        raise ValueError(f"Unsupported multimodal output modality: {modality!r}")

    def _run_image_output(
        self,
        *,
        result: MultimodalGenerateResponse,
        spec: Dict[str, Any],
        prompt: str,
        media: Any,
        artifact_store: Optional[Any],
    ) -> None:
        items = self._coerce_media_items(media)
        images = [item for item in items if self._media_type(item, fallback="image" if isinstance(item, (bytes, bytearray)) else None) == "image"]
        roles = [(item, self._media_role(item)) for item in images]
        source_items = [item for item, role in roles if role == "source"]
        mask_items = [item for item, role in roles if role == "mask"]
        reference_like = [item for item, role in roles if role in {"reference", "style", "context"}]
        unroled = [item for item, role in roles if role is None]
        task = str(spec.get("task") or "").lower()

        if mask_items and not (source_items or unroled):
            raise ValueError("Image edit mask media requires one source image.")
        if len(mask_items) > 1:
            raise ValueError("Image edit supports at most one mask in v1.")
        if len(source_items) > 1:
            raise ValueError("Image edit supports at most one source image in v1.")

        should_upscale = task == "image_upscale"
        should_edit = task == "image_edit"
        if not task:
            if source_items or mask_items:
                should_edit = True
            elif len(unroled) == 1 and not reference_like and not mask_items:
                should_edit = True
            elif len(unroled) > 1:
                raise ValueError("Multiple image media items require explicit roles.")
        if mask_items and len(unroled) > 1:
            raise ValueError("Image edit with a mask requires exactly one source image.")
        if should_upscale:
            source = source_items[0] if source_items else (unroled[0] if unroled else None)
            if source is None:
                raise ValueError("Image upscale requires one source image.")
            if mask_items or reference_like:
                raise ValueError("Image upscale accepts one source image and no mask/reference images.")
            if len(source_items) + len(unroled) != 1:
                raise ValueError("Image upscale requires exactly one source image.")

        kwargs = self._output_plugin_kwargs(
            spec,
            exclude={"format", "content_type", "mime_type", "provider", "response_format"},
        )
        if spec.get("provider") is not None:
            kwargs["provider"] = spec.get("provider")
        if artifact_store is not None:
            kwargs["artifact_store"] = artifact_store

        if should_upscale:
            raw = self.vision.upscale_image(self._media_payload(source), **kwargs)
            task_name = "image_upscale"
        elif should_edit:
            kwargs = {k: v for k, v in kwargs.items() if k not in {"width", "height"}}
            source = source_items[0] if source_items else (unroled[0] if unroled else None)
            if source is None:
                raise ValueError("Image edit requires one source image.")
            mask = mask_items[0] if mask_items else None
            reference_payloads = [self._media_payload(item) for item in reference_like]
            if reference_payloads:
                nested_extra = kwargs.get("extra")
                if isinstance(nested_extra, dict):
                    merged_extra = dict(nested_extra)
                    existing = merged_extra.get("reference_images")
                    if existing is None:
                        merged_extra["reference_images"] = reference_payloads
                    elif isinstance(existing, (list, tuple)):
                        merged_extra["reference_images"] = list(existing) + reference_payloads
                    else:
                        merged_extra["reference_images"] = [existing] + reference_payloads
                    kwargs["extra"] = merged_extra
                else:
                    existing = kwargs.get("reference_images")
                    if existing is None:
                        kwargs["reference_images"] = reference_payloads
                    elif isinstance(existing, (list, tuple)):
                        kwargs["reference_images"] = list(existing) + reference_payloads
                    else:
                        kwargs["reference_images"] = [existing] + reference_payloads
            raw = self.vision.i2i(
                prompt,
                self._media_payload(source),
                mask=self._media_payload(mask) if mask is not None else None,
                **kwargs,
            )
            task_name = "image_edit"
        else:
            raw = self.vision.t2i(prompt, **kwargs)
            task_name = "image_generation"

        data, artifact_ref, metadata = self._artifact_or_data(raw)
        fmt = str(spec.get("format") or "png")
        content_type = str(metadata.get("content_type") or metadata.get("mime_type") or f"image/{fmt}")
        if artifact_ref is None:
            data, stored_ref = self._store_generated_data(
                data,
                artifact_store=artifact_store,
                content_type=content_type,
                spec=spec,
            )
            artifact_ref = stored_ref
        item = GeneratedItem(
            modality="image",
            task=task_name,
            data=data,
            artifact_ref=artifact_ref,
            content_type=content_type,
            format=fmt,
            backend_id=getattr(self.vision, "backend_id", None),
            provider=str(spec.get("provider") or getattr(self.vision, "backend_id", None) or self.__class__.__name__),
            model=str(spec.get("model") or self.model),
            metadata=metadata,
        )
        result.add_output("image", item)

    def _run_video_output(
        self,
        *,
        result: MultimodalGenerateResponse,
        spec: Dict[str, Any],
        prompt: str,
        media: Any,
        artifact_store: Optional[Any],
    ) -> None:
        items = self._coerce_media_items(media)
        images = [item for item in items if self._media_type(item, fallback="image" if isinstance(item, (bytes, bytearray)) else None) == "image"]
        roles = [(item, self._media_role(item)) for item in images]
        source_items = [item for item, role in roles if role == "source"]
        mask_items = [item for item, role in roles if role == "mask"]
        reference_like = [item for item, role in roles if role in {"reference", "style", "context"}]
        unroled = [item for item, role in roles if role is None]
        task = str(spec.get("task") or "").lower()

        if mask_items:
            raise ValueError("Video generation does not support mask media in v1.")
        if len(source_items) > 1:
            raise ValueError("Image-to-video supports at most one source image in v1.")

        should_i2v = task == "image_to_video"
        if task == "text_to_video":
            should_i2v = False
        elif not should_i2v:
            if source_items:
                should_i2v = True
            elif len(unroled) == 1 and not reference_like:
                should_i2v = True
            elif len(unroled) > 1:
                raise ValueError("Multiple image media items require explicit roles for image-to-video.")

        kwargs = self._output_plugin_kwargs(
            spec,
            exclude={"format", "content_type", "mime_type", "provider", "response_format"},
        )
        if spec.get("provider") is not None:
            kwargs["provider"] = spec.get("provider")
        if artifact_store is not None:
            kwargs["artifact_store"] = artifact_store

        if should_i2v:
            source = source_items[0] if source_items else (unroled[0] if unroled else None)
            if source is None:
                raise ValueError("Image-to-video requires one source image.")
            raw = self.vision.i2v(self._media_payload(source), prompt=prompt, **kwargs)
            task_name = "image_to_video"
        else:
            raw = self.vision.t2v(prompt, **kwargs)
            task_name = "text_to_video"

        data, artifact_ref, metadata = self._artifact_or_data(raw)
        fmt = str(spec.get("format") or "mp4").strip().lower() or "mp4"
        content_type = str(metadata.get("content_type") or metadata.get("mime_type") or f"video/{fmt}")
        if artifact_ref is None:
            data, stored_ref = self._store_generated_data(
                data,
                artifact_store=artifact_store,
                content_type=content_type,
                spec=spec,
            )
            artifact_ref = stored_ref
        result.add_output(
            "video",
            GeneratedItem(
                modality="video",
                task=task_name,
                data=data,
                artifact_ref=artifact_ref,
                content_type=content_type,
                format=fmt,
                backend_id=getattr(self.vision, "backend_id", None),
                provider=str(spec.get("provider") or getattr(self.vision, "backend_id", None) or self.__class__.__name__),
                model=str(spec.get("model") or self.model),
                metadata=metadata,
            ),
        )

    def _run_voice_output(
        self,
        *,
        result: MultimodalGenerateResponse,
        spec: Dict[str, Any],
        prompt: str,
        media: Any,
        artifact_store: Optional[Any],
    ) -> None:
        items = self._coerce_media_items(media)
        audio_items = [item for item in items if self._media_type(item, fallback="audio" if isinstance(item, (bytes, bytearray)) else None) == "audio"]
        task = str(spec.get("task") or "").lower()
        voice_id = spec.get("voice_id", spec.get("voice"))
        if task in {"voice_clone", "clone"} or (audio_items and not task):
            if voice_id:
                raise ValueError("Voice output with audio media and voice/voice_id is ambiguous; set task='tts' or omit voice/voice_id.")
            if len(audio_items) != 1:
                raise ValueError("Voice cloning requires exactly one audio media item in v1.")
            kwargs = self._output_plugin_kwargs(spec, exclude={"voice", "voice_id", "format", "provider"})
            if spec.get("provider") is not None:
                kwargs["provider"] = spec.get("provider")
            reference_text = kwargs.pop("reference_text", None)
            if reference_text is None and prompt:
                reference_text = prompt
            if artifact_store is not None:
                kwargs["artifact_store"] = artifact_store
            raw = self.voice.clone(
                self._media_payload(audio_items[0]),
                reference_text=reference_text,
                **kwargs,
            )
            resource_id, metadata = self._resource_id_from_clone_result(raw)
            voice_provider = str(spec.get("provider") or spec.get("tts_provider") or getattr(self.voice, "backend_id", None) or self.__class__.__name__)
            voice_model = spec.get("model") or spec.get("tts_model")
            result.add_resource(
                "voice",
                GeneratedResource(
                    modality="voice",
                    task="voice_clone",
                    resource_type="voice",
                    resource_id=resource_id,
                    name=str(spec.get("name")) if spec.get("name") is not None else None,
                    backend_id=getattr(self.voice, "backend_id", None),
                    provider=voice_provider,
                    model=str(voice_model) if voice_model is not None else None,
                    metadata=metadata,
                ),
            )
            return

        fmt = str(spec.get("format") or "wav")
        kwargs = self._output_plugin_kwargs(spec, exclude={"voice", "voice_id", "format", "provider"})
        if spec.get("provider") is not None:
            kwargs["provider"] = spec.get("provider")
        kwargs["voice"] = str(voice_id) if voice_id is not None else None
        kwargs["format"] = fmt
        if audio_items:
            if len(audio_items) > 1:
                raise ValueError("Reference-guided TTS supports at most one audio reference in v1.")
            kwargs["reference_audio"] = self._media_payload(audio_items[0])
        if artifact_store is not None:
            kwargs["artifact_store"] = artifact_store
        raw = self.voice.tts(prompt, **kwargs)
        data, artifact_ref, metadata = self._artifact_or_data(raw)
        content_type = str(metadata.get("content_type") or metadata.get("mime_type") or f"audio/{fmt}")
        voice_provider = str(spec.get("provider") or spec.get("tts_provider") or getattr(self.voice, "backend_id", None) or self.__class__.__name__)
        voice_model = spec.get("model") or spec.get("tts_model") or metadata.get("model") or metadata.get("model_id")
        tts_meta = metadata.get("abstractvoice_tts")
        if voice_model is None and isinstance(tts_meta, dict):
            voice_model = tts_meta.get("model") or tts_meta.get("model_id") or tts_meta.get("tts_model")
        if voice_model is None and voice_id is not None:
            voice_model = voice_id
        if artifact_ref is None:
            data, stored_ref = self._store_generated_data(
                data,
                artifact_store=artifact_store,
                content_type=content_type,
                spec=spec,
            )
            artifact_ref = stored_ref
        result.add_output(
            "voice",
            GeneratedItem(
                modality="voice",
                task="tts",
                data=data,
                artifact_ref=artifact_ref,
                content_type=content_type,
                format=fmt,
                backend_id=getattr(self.voice, "backend_id", None),
                provider=voice_provider,
                model=str(voice_model) if voice_model is not None else None,
                metadata=metadata,
            ),
        )

    def _run_music_output(
        self,
        *,
        result: MultimodalGenerateResponse,
        spec: Dict[str, Any],
        prompt: str,
        media: Any,
        artifact_store: Optional[Any],
    ) -> None:
        items = self._coerce_media_items(media)
        audio_items = [
            item
            for item in items
            if self._media_type(item, fallback="audio" if isinstance(item, (bytes, bytearray)) else None) == "audio"
        ]
        if audio_items:
            raise ValueError("Music generation does not accept audio media in v1; pass lyrics/text fields instead.")

        if spec.get("backend") is not None or spec.get("music_backend") is not None:
            raise ValueError("Music output routing uses `provider` as the backend selector; `backend` and `music_backend` are not supported.")

        fmt = str(spec.get("format") or spec.get("response_format") or "wav").strip().lower() or "wav"
        kwargs = self._output_plugin_kwargs(
            spec,
            exclude={"lyrics", "format", "response_format", "provider", "backend", "music_backend"},
        )
        if spec.get("provider") is not None:
            kwargs["provider"] = spec.get("provider")
        if artifact_store is not None:
            kwargs["artifact_store"] = artifact_store

        provider_task = str(spec.get("task") or "music_generation").strip().lower().replace("-", "_") or "music_generation"
        if provider_task in {"music", "song", "t2m", "music_generation", "text_to_music", "lyrics_to_music"}:
            provider_task = "text_to_music"
        elif provider_task in {"sound", "sfx", "sound_generation", "audio_generation", "text_to_audio"}:
            provider_task = "text_to_audio"
        result_task = "sound_generation" if provider_task == "text_to_audio" else "music_generation"
        result_modality = "sound" if result_task == "sound_generation" else "music"

        raw = self.music.generate(
            prompt,
            task=provider_task,
            lyrics=spec.get("lyrics"),
            format=fmt,
            **kwargs,
        )
        data, artifact_ref, metadata = self._artifact_or_data(raw)
        content_type = str(metadata.get("content_type") or metadata.get("mime_type") or f"audio/{fmt}")
        backend_id = metadata.get("backend_id") or metadata.get("backend") or getattr(self.music, "backend_id", None)
        backend_id = str(backend_id).strip() if isinstance(backend_id, str) and str(backend_id).strip() else None

        # Provider/model should reflect the invoked backend/result when possible
        # (not a blind echo of request selectors).
        music_provider = metadata.get("provider") or metadata.get("provider_id") or backend_id or getattr(self.music, "backend_id", None) or self.__class__.__name__
        music_provider = str(music_provider)
        music_model = metadata.get("model") or metadata.get("model_id") or metadata.get("modelId") or spec.get("model")
        if artifact_ref is None:
            data, stored_ref = self._store_generated_data(
                data,
                artifact_store=artifact_store,
                content_type=content_type,
                spec=spec,
            )
            artifact_ref = stored_ref
        result.add_output(
            result_modality,
            GeneratedItem(
                modality=result_modality,
                task=result_task,
                data=data,
                artifact_ref=artifact_ref,
                content_type=content_type,
                format=fmt,
                backend_id=backend_id,
                provider=music_provider,
                model=str(music_model) if music_model is not None else None,
                metadata=metadata,
            ),
        )

    def _run_transcription_output(
        self,
        *,
        result: MultimodalGenerateResponse,
        spec: Dict[str, Any],
        media: Any,
        artifact_store: Optional[Any],
    ) -> None:
        items = self._coerce_media_items(media)
        audio_items = [item for item in items if self._media_type(item, fallback="audio" if isinstance(item, (bytes, bytearray)) else None) == "audio"]
        if len(audio_items) != 1:
            raise ValueError("Transcription requires exactly one audio media item in v1.")
        kwargs: Dict[str, Any] = {}
        if spec.get("language") is not None:
            kwargs["language"] = spec.get("language")
        if spec.get("provider") is not None:
            kwargs["provider"] = spec.get("provider")
        if spec.get("model") is not None:
            kwargs["model"] = spec.get("model")
        if artifact_store is not None:
            kwargs["artifact_store"] = artifact_store
        transcript = self.audio.transcribe(self._media_payload(audio_items[0]), **kwargs)
        result.text = GenerateResponse(
            content=str(transcript),
            model=self.model,
            metadata={"task": "transcription", "modality": "text"},
        )

    def generate_with_telemetry(self,
                               prompt: str = "",
                               messages: Optional[List[Dict[str, str]]] = None,
                               system_prompt: Optional[str] = None,
                               tools: Optional[List] = None,  # Accept both ToolDefinition and Dict
                               media: Optional[List[Union[str, Dict[str, Any], 'MediaContent']]] = None,  # Media files
                               stream: bool = False,
                               response_model: Optional[Type[BaseModel]] = None,
                               retry_strategy=None,  # Custom retry strategy for structured output
                               tool_call_tags: Optional[str] = None,  # Tool call tag rewriting
                               execute_tools: Optional[bool] = None,  # Tool execution control
                               glyph_compression: Optional[str] = None,  # Glyph compression preference
                               thinking: Optional[Union[bool, str]] = None,  # Unified reasoning/thinking control
                               **kwargs) -> Union[GenerateResponse, Iterator[GenerateResponse], BaseModel]:
        """
        Generate with integrated telemetry and error handling.
        Providers should override _generate_internal instead of generate.

        Args:
            prompt: The input prompt
            messages: Optional conversation history
            system_prompt: Optional system prompt
            tools: Optional list of available tools
            media: Optional list of media files (file paths, MediaContent objects, or dicts)
            stream: Whether to stream the response
            response_model: Optional Pydantic model for structured output
            retry_strategy: Optional retry strategy for structured output validation
            tool_call_tags: Optional tool call tag format for rewriting
            execute_tools: Whether to execute tools automatically (True) or let agent handle execution (False)
            glyph_compression: Glyph compression preference ("auto", "always", "never")
            thinking: Unified reasoning/thinking control (auto/on/off/none or low/medium/high/xhigh when supported)
        """
        system_prompt = self._normalize_system_prompt_alias(system_prompt, kwargs, stacklevel=4)

        output_request = kwargs.get("output", None)
        is_acore_output = self._is_acore_output_request(output_request)
        if (prompt is None or str(prompt) == "") and "text" in kwargs:
            text_value = kwargs.pop("text")
            prompt = "" if text_value is None else str(text_value)

        if is_acore_output:
            try:
                maybe_specs = self._normalize_output_specs(output_request)
            except Exception:
                maybe_specs = []
            if (
                len(maybe_specs) == 1
                and maybe_specs[0].get("modality") == "text"
                and maybe_specs[0].get("task") not in {"transcription"}
                and str(prompt or "").strip()
            ):
                kwargs.pop("output", None)
                is_acore_output = False

        if is_acore_output:
            output_request = kwargs.pop("output")
            output_request, progress_injected = self._output_request_with_progress_callbacks(
                output_request,
                {
                    "on_progress": kwargs.get("on_progress"),
                    "progress_event_callback": kwargs.get("progress_event_callback"),
                    "progress_callback": kwargs.get("progress_callback"),
                },
            )
            if progress_injected:
                kwargs.pop("on_progress", None)
                kwargs.pop("progress_event_callback", None)
                kwargs.pop("progress_callback", None)
            artifact_store = kwargs.pop("artifact_store", None)
            partial = bool(kwargs.pop("partial", False))
            return self._generate_multimodal_response(
                prompt=str(prompt or ""),
                output=output_request,
                media=media,
                stream=stream,
                response_model=response_model,
                artifact_store=artifact_store,
                partial=partial,
                text_generation_kwargs={
                    "messages": messages,
                    "system_prompt": system_prompt,
                    "tools": tools,
                    "media": media,
                    "retry_strategy": retry_strategy,
                    "tool_call_tags": tool_call_tags,
                    "execute_tools": execute_tools,
                    "glyph_compression": glyph_compression,
                    "thinking": thinking,
                    **kwargs,
                },
            )

        # Normalize token limit naming at the provider boundary.
        #
        # - OpenAI-style APIs use `max_tokens` for the output-token cap.
        # - AbstractCore's unified internal name is `max_output_tokens`.
        #
        # AbstractRuntime (and some hosts) may still emit `max_tokens` in effect payloads.
        # That translation is a provider integration concern, so keep it in AbstractCore.
        if "max_output_tokens" not in kwargs and "max_tokens" in kwargs and kwargs.get("max_tokens") is not None:
            kwargs["max_output_tokens"] = kwargs.pop("max_tokens")

        carried_thinking_meta = kwargs.pop("_acore_thinking_meta", None)
        if carried_thinking_meta is not None and not isinstance(carried_thinking_meta, dict):
            carried_thinking_meta = None

        # Prompt caching: apply a default `prompt_cache_key` if configured.
        self._apply_default_prompt_cache_key(kwargs)
        self._apply_prompt_cache_binding_request(kwargs)
        messages, system_prompt = self._normalize_developer_messages(messages, system_prompt)

        # Apply unified thinking controls (provider-agnostic + provider-specific mappings).
        prompt, messages, system_prompt, kwargs, thinking_meta = self._apply_thinking_request(
            thinking=thinking,
            prompt=prompt,
            messages=messages,
            system_prompt=system_prompt,
            kwargs=kwargs,
        )
        if thinking_meta is None and carried_thinking_meta is not None:
            thinking_meta = carried_thinking_meta

        # Handle structured output request
        if response_model is not None:
            if not PYDANTIC_AVAILABLE:
                raise ImportError(
                    "Pydantic is required for structured outputs. "
                    "Install with: pip install pydantic>=2.0.0"
                )

            # Handle hybrid case: tools + structured output.
            #
            # NOTE: `tools=[]` should behave like "no tools". Treating an empty list as
            # "tools present" triggers the hybrid 2-pass flow (unstructured call + structured
            # follow-up) which is both slower and can cause provider-side timeouts/unloads.
            if isinstance(tools, list) and len(tools) > 0:
                return self._handle_tools_with_structured_output(
                    prompt=prompt,
                    messages=messages,
                    system_prompt=system_prompt,
                    tools=tools,
                    media=media,
                    response_model=response_model,
                    retry_strategy=retry_strategy,
                    tool_call_tags=tool_call_tags,
                    execute_tools=execute_tools,
                    stream=stream,
                    **kwargs
                )

            # Standard structured output (no tools)
            from ..structured import StructuredOutputHandler
            handler = StructuredOutputHandler(retry_strategy=retry_strategy)
            return handler.generate_structured(
                provider=self,
                prompt=prompt,
                response_model=response_model,
                messages=messages,
                system_prompt=system_prompt,
                tools=None,  # No tools in this path
                media=media,
                stream=stream,
                **kwargs
            )

        # Process media content if provided
        processed_media = None
        media_metadata = None
        media_enrichment = None
        if media:
            compression_pref = glyph_compression or kwargs.get('glyph_compression', 'auto')
            processed_media = self._process_media_content(media, compression_pref)

            # Extract metadata from processed media for response
            if processed_media:
                media_metadata = []
                for media_content in processed_media:
                    if hasattr(media_content, 'metadata') and media_content.metadata:
                        media_metadata.append(media_content.metadata)

        def _configured_capability_default(kind: str, modality: str) -> Dict[str, Any]:
            key = f"{str(kind or '').strip().lower()}.{str(modality or '').strip().lower()}"

            scoped_routes = getattr(self, "_abstractcore_capability_defaults", None)
            if isinstance(scoped_routes, dict):
                route = scoped_routes.get(key)
                if isinstance(route, dict):
                    if route.get("source") == "not_configured":
                        return {}
                    if route.get("provider") or route.get("model") or route.get("base_url") or route.get("options"):
                        return dict(route)

            resolver = getattr(self, "resolve_capability_default", None)
            if callable(resolver):
                try:
                    route = resolver(kind, modality)
                except Exception:
                    route = None
                if isinstance(route, dict):
                    if route.get("source") == "not_configured":
                        return {}
                    if route.get("provider") or route.get("model") or route.get("base_url") or route.get("options"):
                        return dict(route)

            try:
                config_file = getattr(self, "_abstractcore_config_file", None)
                if isinstance(config_file, str) and config_file.strip():
                    from ..config.manager import ConfigurationManager

                    route = ConfigurationManager(config_file=config_file.strip(), apply_env=False).get_capability_default(kind, modality)
                else:
                    from ..config.manager import get_config_manager

                    route = get_config_manager().get_capability_default(kind, modality)
            except Exception:
                return {}
            if not isinstance(route, dict):
                return {}
            if route.get("source") == "not_configured":
                return {}
            if not (route.get("provider") or route.get("model") or route.get("base_url") or route.get("options")):
                return {}
            return route

        def _pop_stt_route_params(route: Dict[str, Any]) -> Dict[str, Any]:
            out: Dict[str, Any] = {}
            provider = kwargs.pop("stt_provider", None)
            if provider is None:
                provider = kwargs.pop("audio_provider", None)
            if provider is None:
                provider = route.get("provider")
            model = kwargs.pop("stt_model", None)
            if model is None:
                model = kwargs.pop("audio_model", None)
            if model is None:
                model = route.get("model")
            if isinstance(provider, str) and provider.strip():
                out["provider"] = provider.strip()
            if isinstance(model, str) and model.strip():
                out["model"] = model.strip()
            return out

        # Audio input policy (v0): avoid placeholder degradation and require explicit fallbacks.
        if processed_media:
            try:
                from ..media.types import ContentFormat, MediaType
                from ..media.enrichment import build_enrichment_item
                from ..capabilities.errors import CapabilityUnavailableError
            except Exception:
                ContentFormat = None  # type: ignore[assignment]
                MediaType = None  # type: ignore[assignment]
                build_enrichment_item = None  # type: ignore[assignment]
                CapabilityUnavailableError = Exception  # type: ignore[assignment]

            if MediaType is not None:
                audio_items = [mc for mc in processed_media if getattr(mc, "media_type", None) == MediaType.AUDIO]
            else:
                audio_items = []

            if audio_items:
                # Resolve policy: per-call kwarg > config default.
                policy_raw = kwargs.pop("audio_policy", None)
                policy_explicit = policy_raw is not None
                if policy_raw is None:
                    policy_raw = kwargs.pop("audio_handling_policy", None)
                    policy_explicit = policy_raw is not None
                if policy_raw is None:
                    try:
                        from ..config.manager import get_config_manager

                        policy_raw = getattr(get_config_manager().config, "audio", None).strategy  # type: ignore[union-attr]
                    except Exception:
                        policy_raw = "native_only"

                policy = str(policy_raw or "native_only").strip().lower()
                model_supports_audio = bool(getattr(self, "model_capabilities", {}).get("audio_support", False))
                stt_route_default = _configured_capability_default("input", "voice")

                if policy in ("native_only", "native", "disabled"):
                    if not model_supports_audio:
                        raise UnsupportedFeatureError(
                            f"Audio input is not supported by model '{self.model}'. "
                            "Choose an audio-capable model, configure the input.voice capability default, "
                            "or pass audio_policy='speech_to_text' for this call."
                        )
                    # Keep audio media for provider-native handling (provider support may still vary).

                elif policy in ("speech_to_text", "stt"):
                    if not policy_explicit and not stt_route_default:
                        raise UnsupportedFeatureError(
                            f"Audio input is not supported by model '{self.model}', and input.voice is not configured. "
                            "Configure the input.voice capability default for STT fallback, or pass audio_policy='speech_to_text' "
                            "with explicit stt_provider/stt_model for this call."
                        )
                    stt_route_params = _pop_stt_route_params(stt_route_default)
                    stt_language = kwargs.pop("audio_language", None)
                    if stt_language is None:
                        stt_language = kwargs.pop("stt_language", None)
                    if stt_language is None:
                        try:
                            from ..config.manager import get_config_manager

                            stt_language = getattr(get_config_manager().config, "audio", None).stt_language  # type: ignore[union-attr]
                        except Exception:
                            stt_language = None

                    audio_context_parts: List[str] = []
                    enrichments: List[Dict[str, Any]] = []

                    # Resolve backend id (best-effort) for transparency metadata.
                    backend_id = getattr(getattr(self, "audio", None), "backend_id", None)
                    backend = {"kind": "plugin"}
                    if isinstance(backend_id, str) and backend_id.strip():
                        backend["backend_id"] = backend_id.strip()

                    for idx, mc in enumerate(audio_items):
                        name = None
                        try:
                            name = mc.metadata.get("file_name") if hasattr(mc, "metadata") and isinstance(mc.metadata, dict) else None
                        except Exception:
                            name = None
                        if not isinstance(name, str) or not name.strip():
                            name = mc.file_path if getattr(mc, "file_path", None) else f"audio_{idx+1}"

                        # Prefer a file path when available.
                        audio_input: Any = None
                        try:
                            if getattr(mc, "file_path", None):
                                audio_input = str(mc.file_path)
                            elif getattr(mc, "content_format", None) == ContentFormat.FILE_PATH and isinstance(getattr(mc, "content", None), str):
                                audio_input = str(mc.content)
                            elif isinstance(getattr(mc, "content", None), (bytes, bytearray)):
                                audio_input = bytes(mc.content)
                        except Exception:
                            audio_input = None

                        if audio_input is None:
                            raise UnsupportedFeatureError("Audio STT fallback requires a file path or raw bytes for the audio input.")

                        try:
                            transcript = self.audio.transcribe(audio_input, language=stt_language, **stt_route_params)
                        except CapabilityUnavailableError as e:  # type: ignore[misc]
                            raise UnsupportedFeatureError(str(e))

                        transcript = str(transcript or "").strip()
                        audio_context_parts.append(f"Audio {idx+1} ({name}): {transcript}")
                        if build_enrichment_item is not None:
                            backend_for_item = dict(backend)
                            if stt_route_params.get("provider"):
                                backend_for_item["provider"] = stt_route_params.get("provider")
                            if stt_route_params.get("model"):
                                backend_for_item["model"] = stt_route_params.get("model")
                            enrichments.append(
                                build_enrichment_item(
                                    status="used",
                                    input_modality="audio",
                                    summary_kind="transcript",
                                    policy="speech_to_text",
                                    backend=backend_for_item,
                                    input_index=idx + 1,
                                    input_name=str(name),
                                    injected_text=transcript,
                                )
                            )

                    # Remove audio media from the provider call (we injected text context instead).
                    processed_media = [mc for mc in processed_media if getattr(mc, "media_type", None) != MediaType.AUDIO]

                    # Inject audio context into the prompt (similar recency semantics as vision fallback).
                    original_prompt = prompt.strip() if isinstance(prompt, str) else ""
                    parts: List[str] = []
                    parts.append(
                        "Audio context from attached audio file(s) "
                        "(treat as directly observed; do not mention this section):"
                    )
                    parts.extend(audio_context_parts)
                    if original_prompt:
                        parts.append("Now answer the user's request:")
                        parts.append(original_prompt)
                    prompt = "\n\n".join(parts) if parts else original_prompt

                    media_enrichment = enrichments

                elif policy == "auto":
                    if model_supports_audio:
                        pass  # provider-native path
                    else:
                        if not stt_route_default:
                            raise UnsupportedFeatureError(
                                f"Audio input is not supported by model '{self.model}', and input.voice is not configured. "
                                "Configure the input.voice capability default for STT fallback, or choose an audio-capable model."
                            )
                        stt_language = kwargs.pop("audio_language", None) or kwargs.pop("stt_language", None)
                        stt_route_params = _pop_stt_route_params(stt_route_default)
                        audio_context_parts: List[str] = []
                        enrichments: List[Dict[str, Any]] = []
                        backend_id = getattr(getattr(self, "audio", None), "backend_id", None)
                        backend = {"kind": "plugin"}
                        if isinstance(backend_id, str) and backend_id.strip():
                            backend["backend_id"] = backend_id.strip()
                        for idx, mc in enumerate(audio_items):
                            name = None
                            try:
                                name = mc.metadata.get("file_name") if hasattr(mc, "metadata") and isinstance(mc.metadata, dict) else None
                            except Exception:
                                name = None
                            if not isinstance(name, str) or not name.strip():
                                name = mc.file_path if getattr(mc, "file_path", None) else f"audio_{idx+1}"
                            audio_input: Any = None
                            try:
                                if getattr(mc, "file_path", None):
                                    audio_input = str(mc.file_path)
                                elif getattr(mc, "content_format", None) == ContentFormat.FILE_PATH and isinstance(getattr(mc, "content", None), str):
                                    audio_input = str(mc.content)
                                elif isinstance(getattr(mc, "content", None), (bytes, bytearray)):
                                    audio_input = bytes(mc.content)
                            except Exception:
                                audio_input = None
                            if audio_input is None:
                                raise UnsupportedFeatureError("Audio STT fallback requires a file path or raw bytes for the audio input.")
                            try:
                                transcript = self.audio.transcribe(audio_input, language=stt_language, **stt_route_params)
                            except CapabilityUnavailableError as e:  # type: ignore[misc]
                                raise UnsupportedFeatureError(str(e))
                            transcript = str(transcript or "").strip()
                            audio_context_parts.append(f"Audio {idx+1} ({name}): {transcript}")
                            if build_enrichment_item is not None:
                                backend_for_item = dict(backend)
                                if stt_route_params.get("provider"):
                                    backend_for_item["provider"] = stt_route_params.get("provider")
                                if stt_route_params.get("model"):
                                    backend_for_item["model"] = stt_route_params.get("model")
                                enrichments.append(
                                    build_enrichment_item(
                                        status="used",
                                        input_modality="audio",
                                        summary_kind="transcript",
                                        policy="auto",
                                        backend=backend_for_item,
                                        input_index=idx + 1,
                                        input_name=str(name),
                                        injected_text=transcript,
                                    )
                                )
                        processed_media = [mc for mc in processed_media if getattr(mc, "media_type", None) != MediaType.AUDIO]
                        original_prompt = prompt.strip() if isinstance(prompt, str) else ""
                        parts: List[str] = []
                        parts.append(
                            "Audio context from attached audio file(s) "
                            "(treat as directly observed; do not mention this section):"
                        )
                        parts.extend(audio_context_parts)
                        if original_prompt:
                            parts.append("Now answer the user's request:")
                            parts.append(original_prompt)
                        prompt = "\n\n".join(parts) if parts else original_prompt
                        media_enrichment = enrichments

                elif policy == "caption":
                    raise UnsupportedFeatureError(
                        "audio_policy='caption' is not configured in v0. "
                        "Use audio_policy='speech_to_text' for speech, or configure a future audio caption backend."
                    )
                else:
                    raise ValueError(f"Unknown audio_policy '{policy}'. Expected one of: native_only, speech_to_text, auto, caption.")

        # Video input policy (v0): allow native video where supported; otherwise fall back to sampled frames.
        # Note: most providers do not accept native video inputs; frame sampling provides a portable path.
        if processed_media:
            try:
                from ..media.types import MediaType
                from ..media.enrichment import build_enrichment_item
            except Exception:
                MediaType = None  # type: ignore[assignment]
                build_enrichment_item = None  # type: ignore[assignment]

            if MediaType is not None:
                video_items = [mc for mc in processed_media if getattr(mc, "media_type", None) == MediaType.VIDEO]
            else:
                video_items = []

            if video_items:
                policy_raw = kwargs.pop("video_policy", None)
                if policy_raw is None:
                    policy_raw = kwargs.pop("video_handling_policy", None)
                if policy_raw is None:
                    try:
                        from ..config.manager import get_config_manager

                        policy_raw = getattr(get_config_manager().config, "video", None).strategy  # type: ignore[union-attr]
                    except Exception:
                        policy_raw = "native_only"

                policy = str(policy_raw or "native_only").strip().lower()

                provider_name = str(getattr(self, "provider", "") or "").strip().lower()
                model_supports_native_video = bool(
                    provider_name == "huggingface"
                    and isinstance(getattr(self, "model_capabilities", None), dict)
                    and getattr(self, "model_capabilities", {}).get("video_support", False)
                )
                model_caps = getattr(self, "model_capabilities", {}) if isinstance(getattr(self, "model_capabilities", None), dict) else {}
                video_input_mode = str(model_caps.get("video_input_mode") or "").strip().lower()
                model_supports_frame_video = bool(
                    model_supports_native_video
                    or video_input_mode == "frames"
                    or (video_input_mode == "native" and bool(model_caps.get("video_support", False)))
                    or bool(model_caps.get("vision_support", False))
                )
                video_route_default = _configured_capability_default("input", "video")

                cfg_video = None
                try:
                    from ..config.manager import get_config_manager

                    cfg_video = getattr(get_config_manager().config, "video", None)
                except Exception:
                    cfg_video = None

                # Sampling controls (best-effort; keep small by default).
                # NOTE: do not `pop` here: native video backends may also need the resolved values.
                max_frames_raw = kwargs.get("video_max_frames", None)
                if max_frames_raw is None:
                    max_frames_raw = kwargs.get("max_video_frames", None)
                if max_frames_raw is None:
                    fallback_default = getattr(cfg_video, "max_frames", 3) if cfg_video is not None else 3
                    native_default = getattr(cfg_video, "max_frames_native", None) if cfg_video is not None else None
                    if native_default is None:
                        native_default = fallback_default

                    use_native_default = bool(
                        model_supports_native_video and policy in ("native_only", "native", "disabled", "auto")
                    )
                    max_frames_raw = native_default if use_native_default else fallback_default
                try:
                    max_frames = max(1, int(max_frames_raw))
                except Exception:
                    max_frames = 3

                frame_format_raw = kwargs.get("video_frame_format", None)
                if frame_format_raw is None:
                    try:
                        from ..config.manager import get_config_manager

                        frame_format_raw = getattr(get_config_manager().config, "video", None).frame_format  # type: ignore[union-attr]
                    except Exception:
                        frame_format_raw = "jpg"
                frame_format = str(frame_format_raw or "jpg").strip().lower()
                if frame_format not in {"jpg", "jpeg", "png"}:
                    frame_format = "jpg"
                if frame_format == "jpeg":
                    frame_format = "jpg"

                sampling_strategy_raw = kwargs.get("video_sampling_strategy", None)
                if sampling_strategy_raw is None:
                    try:
                        from ..config.manager import get_config_manager

                        sampling_strategy_raw = getattr(get_config_manager().config, "video", None).sampling_strategy  # type: ignore[union-attr]
                    except Exception:
                        sampling_strategy_raw = "uniform"
                sampling_strategy = str(sampling_strategy_raw or "uniform").strip().lower()
                if sampling_strategy not in {"uniform", "keyframes"}:
                    sampling_strategy = "uniform"

                max_frame_side_raw = kwargs.get("video_max_frame_side", None)
                if max_frame_side_raw is None:
                    max_frame_side_raw = kwargs.get("video_frame_max_side", None)
                if max_frame_side_raw is None:
                    max_frame_side_raw = getattr(cfg_video, "max_frame_side", 1024) if cfg_video is not None else 1024
                try:
                    max_frame_side = int(max_frame_side_raw) if max_frame_side_raw is not None else None
                except Exception:
                    max_frame_side = 1024
                if isinstance(max_frame_side, int) and max_frame_side <= 0:
                    max_frame_side = None

                # Expose normalized sampling values to provider-native implementations.
                kwargs["video_max_frames"] = max_frames
                kwargs["video_frame_format"] = frame_format
                kwargs["video_sampling_strategy"] = sampling_strategy
                kwargs["video_max_frame_side"] = max_frame_side

                def _video_route_matches_current(route: Dict[str, Any]) -> bool:
                    route_provider = str(route.get("provider") or "").strip().lower()
                    route_model = str(route.get("model") or "").strip()
                    current_model = str(getattr(self, "model", "") or "").strip()
                    return bool(route_provider and route_model and route_provider == provider_name and route_model == current_model)

                def _route_llm_kwargs(route: Dict[str, Any]) -> Dict[str, Any]:
                    out: Dict[str, Any] = {}
                    options = route.get("options")
                    if isinstance(options, dict):
                        out.update(dict(options))
                    base_url = route.get("base_url")
                    if isinstance(base_url, str) and base_url.strip():
                        out["base_url"] = base_url.strip()
                    return out

                def _resolve_endpoint_route_provider(provider_id: str, route: Dict[str, Any]) -> tuple[str, Dict[str, Any], Optional[Dict[str, Any]]]:
                    provider_s = str(provider_id or "").strip()
                    route_kwargs = _route_llm_kwargs(route)
                    if not provider_s.startswith("endpoint:"):
                        return provider_s, route_kwargs, None
                    resolver = getattr(self, "resolve_provider_endpoint_profile", None)
                    if not callable(resolver):
                        raise UnsupportedFeatureError(
                            f"input.video fallback uses Gateway endpoint profile {provider_s!r}, "
                            "but no endpoint-profile resolver is attached to this Core provider."
                        )
                    try:
                        resolved = resolver(provider_s)
                    except Exception as e:
                        raise UnsupportedFeatureError(f"Failed to resolve Gateway endpoint profile {provider_s!r}: {e}") from e
                    if not isinstance(resolved, dict):
                        raise UnsupportedFeatureError(f"Gateway endpoint profile {provider_s!r} is not configured or is disabled.")
                    resolved_provider = str(resolved.get("provider") or resolved.get("provider_family") or "").strip()
                    if not resolved_provider:
                        raise UnsupportedFeatureError(f"Gateway endpoint profile {provider_s!r} did not provide a concrete provider family.")
                    resolved_kwargs = dict(route_kwargs)
                    base_url = resolved.get("base_url")
                    if isinstance(base_url, str) and base_url.strip():
                        resolved_kwargs["base_url"] = base_url.strip()
                    api_key = resolved.get("api_key")
                    if isinstance(api_key, str) and api_key.strip():
                        resolved_kwargs["api_key"] = api_key.strip()
                    return resolved_provider, resolved_kwargs, resolved

                def _extract_response_text(value: Any) -> str:
                    if isinstance(value, str):
                        return value.strip()
                    if isinstance(value, dict):
                        for key in ("content", "text", "response", "message"):
                            item = value.get(key)
                            if isinstance(item, str) and item.strip():
                                return item.strip()
                    content = getattr(value, "content", None)
                    if isinstance(content, str) and content.strip():
                        return content.strip()
                    return str(value or "").strip()

                def _caption_videos_with_route(route: Dict[str, Any]) -> tuple[List[Any], List[Dict[str, Any]]]:
                    route_provider = str(route.get("provider") or "").strip()
                    route_model = str(route.get("model") or "").strip()
                    if not route_provider or not route_model:
                        raise UnsupportedFeatureError("input.video fallback requires both provider and model.")

                    try:
                        from pathlib import Path
                        import tempfile

                        from ..core.factory import create_llm
                        from ..media import AutoMediaHandler
                        from ..media.types import ContentFormat, MediaContent
                        from ..media.utils.video_frames import extract_video_frames, probe_duration_s
                    except Exception as e:
                        raise UnsupportedFeatureError(f"Video route fallback is not available: {e}")

                    routed_provider, route_kwargs, endpoint_profile = _resolve_endpoint_route_provider(route_provider, route)
                    try:
                        fallback_llm = create_llm(routed_provider, model=route_model, **route_kwargs)
                        resolver = getattr(self, "resolve_provider_endpoint_profile", None)
                        if callable(resolver):
                            try:
                                setattr(fallback_llm, "resolve_provider_endpoint_profile", resolver)
                            except Exception:
                                pass
                    except Exception as e:
                        raise UnsupportedFeatureError(
                            f"Failed to create input.video fallback provider {route_provider}/{route_model}: {e}"
                        )

                    new_media: List[Any] = []
                    enrichments: List[Dict[str, Any]] = []
                    video_group_index = 0
                    for idx, mc in enumerate(processed_media or []):
                        if getattr(mc, "media_type", None) != MediaType.VIDEO:  # type: ignore[operator]
                            new_media.append(mc)
                            continue

                        video_group_index += 1
                        video_path_raw = getattr(mc, "file_path", None) or getattr(mc, "content", None)
                        if not isinstance(video_path_raw, str) or not video_path_raw.strip():
                            raise UnsupportedFeatureError("Video route fallback requires a video file path.")
                        video_path = Path(video_path_raw)
                        if not video_path.exists():
                            raise UnsupportedFeatureError(f"Video file not found: {video_path}")

                        out_dir = Path(tempfile.mkdtemp(prefix="abstractcore_video_route_frames_"))
                        duration_s = probe_duration_s(video_path)
                        frames, timestamps_s = extract_video_frames(
                            video_path,
                            max_frames=max_frames,
                            frame_format=frame_format,
                            sampling_strategy=sampling_strategy,
                            max_side=max_frame_side,
                            output_dir=out_dir,
                        )
                        if not frames:
                            raise UnsupportedFeatureError("Video route fallback failed: no frames extracted.")

                        handler = AutoMediaHandler(enable_glyph_compression=False)
                        frame_paths: List[str] = []
                        max_res = None
                        if isinstance(max_frame_side, int) and max_frame_side > 0:
                            max_res = (max_frame_side, max_frame_side)
                        for fp in frames:
                            res = handler.process_file(
                                fp,
                                provider=routed_provider,
                                model=route_model,
                                glyph_compression="never",
                                max_resolution=max_res,
                            )
                            if res and getattr(res, "success", False) and getattr(res, "media_content", None) is not None:
                                frame_paths.append(str(fp))

                        if not frame_paths:
                            raise UnsupportedFeatureError("Video route fallback failed: extracted frames could not be processed as images.")

                        caption_prompt = (
                            "Provide grounded observations from these sampled video frames to help answer the user's request.\n"
                            "- Treat the frames as chronological samples from one video.\n"
                            "- Write 3-5 concise natural sentences.\n"
                            "- Include visible text verbatim when useful.\n"
                            "- Do not mention frames, sampling, extraction, tooling, or this instruction.\n\n"
                            f"User request: {prompt.strip() if isinstance(prompt, str) and prompt.strip() else 'Describe the video.'}"
                        )
                        try:
                            fallback_response = fallback_llm.generate(caption_prompt, media=frame_paths, stream=False)
                        except Exception as e:
                            raise UnsupportedFeatureError(
                                f"input.video fallback {route_provider}/{route_model} failed: {e}"
                            )
                        description = _extract_response_text(fallback_response)
                        if not description:
                            raise UnsupportedFeatureError(
                                f"input.video fallback {route_provider}/{route_model} returned no description."
                            )

                        new_media.append(
                            MediaContent(  # type: ignore[name-defined]
                                media_type=MediaType.TEXT,  # type: ignore[union-attr]
                                content=(
                                    f"Video {video_group_index} ({video_path.name}) observations: {description}\n"
                                    "Use these observations as direct context for the user's current video question. "
                                    "Do not mention fallback models, frames, extraction, or this marker."
                                ),
                                content_format=ContentFormat.TEXT,  # type: ignore[name-defined]
                                mime_type="text/plain",
                                file_path=None,
                                metadata={
                                    "processor": "VideoCapabilityDefaultFallback",
                                    "source_video": video_path.name,
                                    "frame_count": len(frame_paths),
                                    "timestamps_s": timestamps_s,
                                    "duration_s": duration_s,
                                    "provider": route_provider,
                                    "routed_provider": routed_provider,
                                    "model": route_model,
                                },
                            )
                        )
                        if build_enrichment_item is not None:
                            enrichments.append(
                                build_enrichment_item(
                                    status="used",
                                    input_modality="video",
                                    summary_kind="caption",
                                    policy="input.video",
                                    backend={
                                        "kind": "llm",
                                        "provider": route_provider,
                                        "routed_provider": routed_provider,
                                        "model": route_model,
                                        "source": "capability_default",
                                        **(
                                            {"endpoint_profile": str(endpoint_profile.get("virtual_provider") or endpoint_profile.get("id") or "")}
                                            if isinstance(endpoint_profile, dict)
                                            else {}
                                        ),
                                    },
                                    input_index=idx + 1,
                                    input_name=str(video_path.name),
                                    injected_text=description,
                                    artifact={
                                        "frame_count": len(frame_paths),
                                        "timestamps_s": timestamps_s,
                                        "duration_s": duration_s,
                                    },
                                )
                            )

                    return new_media, enrichments

                if policy in ("native_only", "native", "disabled"):
                    if not model_supports_native_video:
                        raise UnsupportedFeatureError(
                            f"Video input is not supported by model '{self.model}'. "
                            "Choose a video-capable model, or pass video_policy='frames_caption' "
                            "(samples frames and uses vision/image handling)."
                        )
                    # Keep video media for provider-native handling.
                    try:
                        from pathlib import Path

                        from ..media.utils.video_frames import probe_duration_s

                        for idx, mc in enumerate(video_items):
                            video_path_raw = getattr(mc, "file_path", None) or getattr(mc, "content", None)
                            if not isinstance(video_path_raw, str) or not video_path_raw.strip():
                                continue
                            vp = Path(video_path_raw)
                            duration_s = probe_duration_s(vp)
                            file_bytes = None
                            try:
                                file_bytes = int(vp.stat().st_size)
                            except Exception:
                                file_bytes = None

                            avg_gap_s = None
                            try:
                                if isinstance(duration_s, (int, float)) and duration_s > 0 and max_frames > 0:
                                    avg_gap_s = float(duration_s) / float(max_frames + 1)
                            except Exception:
                                avg_gap_s = None

                            self.logger.info(
                                "Video input policy: native video enabled (video will be sampled/budgeted for model input).",
                                provider=provider_name,
                                model=self.model,
                                video_policy=policy,
                                video_index=idx + 1,
                                video_name=vp.name,
                                video_duration_s=duration_s,
                                video_bytes=file_bytes,
                                video_max_frames=max_frames,
                                video_sampling_strategy=sampling_strategy,
                                video_max_frame_side=max_frame_side,
                                video_avg_gap_s=avg_gap_s,
                            )
                            if isinstance(avg_gap_s, float) and avg_gap_s >= 10.0:
                                self.logger.warning(
                                    "Video sampling is sparse; important events may be missed. "
                                    "Consider increasing video_max_frames/video.max_frames_native or using keyframes sampling.",
                                    provider=provider_name,
                                    model=self.model,
                                    video_policy=policy,
                                    video_name=vp.name,
                                    video_duration_s=duration_s,
                                    video_max_frames=max_frames,
                                    video_avg_gap_s=avg_gap_s,
                                )
                    except Exception:
                        pass

                    # Insert a short marker to disambiguate native-video inputs across turns.
                    #
                    # Without this, follow-ups like "and this one?" can be brittle for native
                    # video VLMs (they may over-weight the previous text-only answer and ignore
                    # that a *new* video is attached in the current call).
                    try:
                        from ..media.types import MediaContent, ContentFormat
                    except Exception:
                        MediaContent = None  # type: ignore[assignment]
                        ContentFormat = None  # type: ignore[assignment]

                    if MediaContent is not None and ContentFormat is not None:
                        try:
                            from pathlib import Path

                            from ..media.utils.video_frames import probe_duration_s
                        except Exception:
                            Path = None  # type: ignore[assignment]
                            probe_duration_s = None  # type: ignore[assignment]

                        new_media: List[Any] = []
                        video_group_index = 0
                        for mc in processed_media:
                            if getattr(mc, "media_type", None) != MediaType.VIDEO:  # type: ignore[operator]
                                new_media.append(mc)
                                continue

                            video_group_index += 1
                            video_path_raw = getattr(mc, "file_path", None) or getattr(mc, "content", None)

                            video_name = f"video_{video_group_index}"
                            duration_s = None
                            file_bytes = None
                            try:
                                if Path is not None and isinstance(video_path_raw, str) and video_path_raw.strip():
                                    vp = Path(video_path_raw)
                                    video_name = vp.name or video_name
                                    try:
                                        file_bytes = int(vp.stat().st_size)
                                    except Exception:
                                        file_bytes = None
                                    if callable(probe_duration_s):
                                        try:
                                            duration_s = probe_duration_s(vp)
                                        except Exception:
                                            duration_s = None
                            except Exception:
                                duration_s = None
                                file_bytes = None

                            marker = MediaContent(
                                media_type=MediaType.TEXT,
                                content=(
                                    f"Video {video_group_index} ({video_name}) is attached below. "
                                    "This is the current video for this user message. "
                                    "Answer the user's question about this video as if you watched it. "
                                    "If earlier turns mention other videos, images, or audio, ignore them unless the user explicitly asks you to compare. "
                                    "Do not mention tool activity, attachments lists, sampling, frames, extraction, or this marker."
                                ),
                                content_format=ContentFormat.TEXT,
                                mime_type="text/plain",
                                file_path=None,
                                metadata={
                                    "processor": "VideoNativeInputMarker",
                                    "source_video": video_name,
                                    "duration_s": duration_s,
                                    "bytes": file_bytes,
                                    "max_frames": max_frames,
                                    "sampling_strategy": sampling_strategy,
                                    "max_frame_side": max_frame_side,
                                },
                            )
                            new_media.append(marker)
                            new_media.append(mc)

                        processed_media = new_media

                elif policy in ("frames_caption", "frames", "frame_caption"):
                    # Convert each video into a small set of sampled frames (images).
                    try:
                        from pathlib import Path
                        import tempfile

                        from ..media import AutoMediaHandler
                        from ..media.utils.video_frames import extract_video_frames, probe_duration_s
                    except Exception as e:
                        raise UnsupportedFeatureError(f"Video frame fallback is not available: {e}")

                    enrichments: List[Dict[str, Any]] = []
                    new_media: List[Any] = []

                    video_group_index = 0
                    for idx, mc in enumerate(processed_media):
                        if getattr(mc, "media_type", None) != MediaType.VIDEO:  # type: ignore[operator]
                            new_media.append(mc)
                            continue

                        video_group_index += 1
                        video_path_raw = getattr(mc, "file_path", None) or getattr(mc, "content", None)
                        if not isinstance(video_path_raw, str) or not video_path_raw.strip():
                            raise UnsupportedFeatureError("Video frame fallback requires a video file path.")
                        video_path = Path(video_path_raw)
                        if not video_path.exists():
                            raise UnsupportedFeatureError(f"Video file not found: {video_path}")

                        out_dir = Path(tempfile.mkdtemp(prefix="abstractcore_video_frames_"))
                        duration_s = probe_duration_s(video_path)
                        file_bytes = None
                        try:
                            file_bytes = int(video_path.stat().st_size)
                        except Exception:
                            file_bytes = None
                        frames, timestamps_s = extract_video_frames(
                            video_path,
                            max_frames=max_frames,
                            frame_format=frame_format,
                            sampling_strategy=sampling_strategy,
                            max_side=max_frame_side,
                            output_dir=out_dir,
                        )
                        if not frames:
                            raise UnsupportedFeatureError("Video frame fallback failed: no frames extracted.")

                        handler = AutoMediaHandler(enable_glyph_compression=False)
                        frame_media: List[Any] = []
                        max_res = None
                        if isinstance(max_frame_side, int) and max_frame_side > 0:
                            max_res = (max_frame_side, max_frame_side)
                        for fp in frames:
                            res = handler.process_file(
                                fp,
                                provider=self.provider,
                                model=self.model,
                                glyph_compression="never",
                                max_resolution=max_res,
                            )
                            if res and getattr(res, "success", False) and getattr(res, "media_content", None) is not None:
                                frame_media.append(res.media_content)

                        if not frame_media:
                            raise UnsupportedFeatureError("Video frame fallback failed: extracted frames could not be processed as images.")

                        avg_gap_s = None
                        try:
                            if isinstance(duration_s, (int, float)) and duration_s > 0 and max_frames > 0:
                                avg_gap_s = float(duration_s) / float(max_frames + 1)
                        except Exception:
                            avg_gap_s = None

                        self.logger.info(
                            "Video input policy: frames_caption (sampling frames for downstream vision handling).",
                            provider=provider_name,
                            model=self.model,
                            video_policy="frames_caption",
                            video_index=video_group_index,
                            video_name=video_path.name,
                            video_duration_s=duration_s,
                            video_bytes=file_bytes,
                            extracted_frames=len(frame_media),
                            video_max_frames=max_frames,
                            video_sampling_strategy=sampling_strategy,
                            video_max_frame_side=max_frame_side,
                            video_avg_gap_s=avg_gap_s,
                        )
                        if isinstance(avg_gap_s, float) and avg_gap_s >= 10.0:
                            self.logger.warning(
                                "Video sampling is sparse; important events may be missed. "
                                "Consider increasing video_max_frames/video.max_frames or using keyframes sampling.",
                                provider=provider_name,
                                model=self.model,
                                video_policy="frames_caption",
                                video_name=video_path.name,
                                video_duration_s=duration_s,
                                extracted_frames=len(frame_media),
                                video_max_frames=max_frames,
                                video_avg_gap_s=avg_gap_s,
                            )

                        # Insert a short text marker to avoid the model treating sampled frames as
                        # unrelated standalone images (especially in follow-up prompts like "and this one?").
                        try:
                            from ..media.types import MediaContent, ContentFormat
                        except Exception:
                            MediaContent = None  # type: ignore[assignment]
                            ContentFormat = None  # type: ignore[assignment]

                        if MediaContent is not None and ContentFormat is not None:
                            marker = MediaContent(
                                media_type=MediaType.TEXT,
                                content=(
                                    f"Video {video_group_index} ({video_path.name}) — "
                                    "the following images belong to this video in chronological order. "
                                    "Answer the user's question about this video as if you watched it. "
                                    "Do not mention frames, timestamps, sampling, extraction, or this marker."
                                ),
                                content_format=ContentFormat.TEXT,
                                mime_type="text/plain",
                                file_path=None,
                                metadata={
                                    "processor": "VideoFrameFallback",
                                    "source_video": video_path.name,
                                    "frame_count": len(frame_media),
                                    "timestamps_s": timestamps_s,
                                    "duration_s": duration_s,
                                    "bytes": file_bytes,
                                },
                            )
                            new_media.append(marker)

                        new_media.extend(frame_media)

                        if build_enrichment_item is not None:
                            enrichments.append(
                                build_enrichment_item(
                                    status="used",
                                    input_modality="video",
                                    summary_kind="frames",
                                    policy="frames_caption",
                                    backend={"kind": "unknown", "source": "ffmpeg"},
                                    input_index=idx + 1,
                                    input_name=str(video_path.name),
                                    artifact={
                                        "frame_count": len(frame_media),
                                        "timestamps_s": timestamps_s,
                                        "duration_s": duration_s,
                                        "bytes": file_bytes,
                                    },
                                )
                            )

                    processed_media = new_media
                    if enrichments:
                        if media_enrichment is None:
                            media_enrichment = enrichments
                        else:
                            media_enrichment.extend(enrichments)

                elif policy == "auto":
                    explicit_video_route = bool(video_route_default and not video_route_default.get("covered_by"))
                    if explicit_video_route and (not _video_route_matches_current(video_route_default) or not model_supports_frame_video):
                        routed_media, routed_enrichments = _caption_videos_with_route(video_route_default)
                        processed_media = routed_media
                        if routed_enrichments:
                            if media_enrichment is None:
                                media_enrichment = routed_enrichments
                            else:
                                media_enrichment.extend(routed_enrichments)
                    elif model_supports_native_video:
                        # Use native video when available.
                        pass
                    elif model_supports_frame_video or video_route_default:
                        # Auto fallback: sample frames and proceed with existing image pipeline.
                        # This works for vision-capable models and explicit video-default routes.
                        policy_to_use = "frames_caption"
                        kwargs["video_policy"] = policy_to_use
                        # Re-run this branch once with explicit policy.
                        if thinking_meta is not None:
                            kwargs["_acore_thinking_meta"] = thinking_meta
                        return self.generate_with_telemetry(
                            prompt=prompt,
                            messages=messages,
                            system_prompt=system_prompt,
                            tools=tools,
                            media=processed_media,
                            response_model=response_model,
                            retry_strategy=retry_strategy,
                            tool_call_tags=tool_call_tags,
                            execute_tools=execute_tools,
                            stream=stream,
                            **kwargs,
                        )
                    else:
                        raise UnsupportedFeatureError(
                            f"Video input is not supported by model '{self.model}', and input.video is not configured. "
                            "Configure the input.video capability default, choose a video/vision-capable model, "
                            "or pass video_policy='frames_caption' for this call."
                        )

                else:
                    raise ValueError(f"Unknown video_policy '{policy}'. Expected one of: native_only, frames_caption, auto.")

        # Convert tools to ToolDefinition objects first (outside retry loop)
        converted_tools = None
        if tools:
            converted_tools = []
            for tool in tools:
                if hasattr(tool, 'to_dict'):  # ToolDefinition object
                    converted_tools.append(tool.to_dict())
                elif callable(tool):  # Function - check for enhanced metadata
                    if hasattr(tool, '_tool_definition'):
                        # Use the enhanced tool definition from @tool decorator
                        converted_tools.append(tool._tool_definition.to_dict())
                    else:
                        # Fall back to basic conversion
                        from ..tools.core import ToolDefinition
                        tool_def = ToolDefinition.from_function(tool)
                        converted_tools.append(tool_def.to_dict())
                elif isinstance(tool, dict):  # Already a dict
                    converted_tools.append(tool)
                else:
                    # Handle other types gracefully
                    self.logger.warning(f"Unknown tool type: {type(tool)}, skipping")

        # Handle tool execution control
        should_execute_tools = execute_tools if execute_tools is not None else self.execute_tools
        if should_execute_tools and converted_tools:
            warnings.warn(
                "execute_tools=True is deprecated. Prefer passing tools explicitly to generate() "
                "and executing tool calls in the host/runtime via a ToolExecutor.",
                DeprecationWarning,
                stacklevel=2,
            )
        if not should_execute_tools and converted_tools:
            # If tools are provided but execution is disabled,
            # we still pass them to the provider for generation but won't execute them
            self.logger.debug(
                "Provider-side tool execution disabled (expected for runtime/host tool execution); "
                "tools will be sent for generation only."
            )

        # Define generation function for retry wrapper
        def _execute_generation():
            start_time = time.time()
            start_perf = time.perf_counter()

            # Emit generation started event (covers request received)
            event_data = {
                "prompt": preview_text(prompt, max_chars=100),
                "has_tools": bool(tools),
                "stream": stream,
                "model": self.model,
                "provider": self.__class__.__name__
            }
            from ..events import emit_global
            emit_global(EventType.GENERATION_STARTED, event_data, source=self.__class__.__name__)

            try:
                # Call the actual generation (implemented by subclass)
                response = self._generate_internal(
                    prompt=prompt,
                    messages=messages,
                    system_prompt=system_prompt,
                    tools=converted_tools,
                    media=processed_media,
                    stream=stream,
                    execute_tools=should_execute_tools,
                    tool_call_tags=tool_call_tags,
                    media_metadata=media_metadata,
                    **kwargs
                )

                return response, start_time, start_perf

            except Exception as e:
                # Convert to custom exception and re-raise for retry handling
                custom_error = self._handle_api_error(e)
                raise custom_error

        # Execute with retry
        try:
            response, start_time, start_perf = self.retry_manager.execute_with_retry(
                _execute_generation,
                provider_key=self.provider_key
            )

            # Handle streaming with unified processor
            if stream:
                def unified_stream():
                    try:
                        # Import and create unified stream processor
                        from .streaming import UnifiedStreamProcessor

                        # Use the should_execute_tools value (defaults to False)
                        actual_execute_tools = should_execute_tools

                        processor = UnifiedStreamProcessor(
                            model_name=self.model,
                            execute_tools=actual_execute_tools,  # Default: False (pass-through mode)
                            tool_call_tags=tool_call_tags,
                            default_target_format="qwen3"  # Always rewrite to qwen3 format
                        )

                        # Process stream with incremental tool detection and execution
                        ttft_ms: Optional[float] = None
                        thinking_stripper = maybe_create_incremental_thinking_tag_stripper(
                            architecture_format=self.architecture_config,
                            model_capabilities=self.model_capabilities,
                        )
                        last_chunk: Optional[GenerateResponse] = None
                        for processed_chunk in processor.process_stream(response, converted_tools):
                            last_chunk = processed_chunk

                            # TTFT: measure "time to first token" from the provider stream,
                            # independent of downstream post-processing (wrapper/thinking stripping).
                            if ttft_ms is None:
                                raw_has_content = isinstance(processed_chunk.content, str) and bool(processed_chunk.content)
                                raw_has_tools = isinstance(processed_chunk.tool_calls, list) and bool(processed_chunk.tool_calls)
                                if raw_has_content or raw_has_tools:
                                    ttft_ms = round((time.perf_counter() - start_perf) * 1000, 1)
                                    meta = processed_chunk.metadata if isinstance(processed_chunk.metadata, dict) else {}
                                    timing = meta.get("_timing") if isinstance(meta.get("_timing"), dict) else {}
                                    merged = dict(timing)
                                    merged.setdefault("source", "client_wall")
                                    merged["ttft_ms"] = ttft_ms
                                    meta["_timing"] = merged
                                    processed_chunk.metadata = meta

                            if isinstance(processed_chunk.content, str) and processed_chunk.content:
                                processed_chunk.content = strip_output_wrappers(
                                    processed_chunk.content,
                                    architecture_format=self.architecture_config,
                                    model_capabilities=self.model_capabilities,
                                )
                                if thinking_stripper is not None:
                                    processed_chunk.content = thinking_stripper.process(processed_chunk.content)
                            if thinking_meta and isinstance(thinking_meta, dict):
                                meta = processed_chunk.metadata if isinstance(processed_chunk.metadata, dict) else {}
                                meta.update(thinking_meta)
                                processed_chunk.metadata = meta
                            yield processed_chunk

                        if thinking_stripper is not None:
                            tail, reasoning = thinking_stripper.finalize()
                            reasoning_text = reasoning.strip() if isinstance(reasoning, str) and reasoning.strip() else None

                            if (isinstance(tail, str) and tail) or reasoning_text:
                                meta: Dict[str, Any] = {}
                                if thinking_meta and isinstance(thinking_meta, dict):
                                    meta.update(thinking_meta)
                                if reasoning_text:
                                    meta["reasoning"] = reasoning_text

                                yield GenerateResponse(
                                    content=tail if isinstance(tail, str) else "",
                                    model=self.model,
                                    metadata=meta or None,
                                )

                        # Track generation after streaming completes
                        self._track_generation(prompt, None, start_time, success=True, stream=True)

                    except Exception as e:
                        # Track error
                        self._track_generation(prompt, None, start_time, success=False, error=e, stream=True)
                        raise

                return unified_stream()
            else:
                # Non-streaming: normalize tool calls into structured form.
                if response and converted_tools:
                    response = self._normalize_tool_calls_passthrough(
                        response=response,
                        tools=converted_tools,
                        tool_call_tags=tool_call_tags,
                    )

                    # Optional: rewrite tool-call tags in content for downstream clients that parse tags.
                    # Note: when tool_call_tags is None (default), we return cleaned content.
                    if tool_call_tags and response.content and not self._should_clean_tool_call_markup(tool_call_tags):
                        response = self._apply_non_streaming_tag_rewriting(response, tool_call_tags)

                # Normalize provider output (wrapper tokens, Harmony transcripts, think tags).
                if response and isinstance(response.content, str) and response.content:
                    cleaned, reasoning = normalize_assistant_text(
                        response.content,
                        architecture_format=self.architecture_config,
                        model_capabilities=self.model_capabilities,
                    )
                    response.content = cleaned
                    if isinstance(reasoning, str) and reasoning.strip():
                        if response.metadata is None or not isinstance(response.metadata, dict):
                            response.metadata = {}
                        existing = response.metadata.get("reasoning")
                        if isinstance(existing, str) and existing.strip():
                            if reasoning.strip() not in existing:
                                response.metadata["reasoning"] = f"{existing.strip()}\n\n{reasoning.strip()}"
                        else:
                            response.metadata["reasoning"] = reasoning.strip()

                # Attach media enrichment transparency metadata (caption/STT/etc.).
                if media_enrichment and response:
                    from ..media.enrichment import merge_enrichment_metadata

                    response.metadata = merge_enrichment_metadata(response.metadata, media_enrichment)

                # Add visual token calculation if media metadata is available
                if media_metadata and response:
                    self.logger.debug(f"Enhancing response with visual tokens from {len(media_metadata)} media items")
                    response = self._enhance_response_with_visual_tokens(response, media_metadata)

                # Capture interaction trace if enabled
                if self.enable_tracing and response:
                    trace_id = self._capture_trace(
                        prompt=prompt,
                        messages=messages,
                        system_prompt=system_prompt,
                        tools=converted_tools,
                        response=response,
                        kwargs=kwargs
                    )
                    # Attach trace_id to response metadata
                    if not response.metadata:
                        response.metadata = {}
                    response.metadata['trace_id'] = trace_id

                if thinking_meta and response:
                    if response.metadata is None or not isinstance(response.metadata, dict):
                        response.metadata = {}
                    response.metadata.update(thinking_meta)

                self._track_generation(prompt, response, start_time, success=True, stream=False)
                return response

        except Exception as e:
            # This exception comes from the retry manager after all attempts failed
            # Track final error (start_time may not be available, use current time)
            current_time = time.time()
            self._track_generation(prompt, None, current_time, success=False, error=e, stream=stream)

            # Emit error event
            from ..events import emit_global
            emit_global(EventType.ERROR, {
                "error": str(e),
                "error_type": type(e).__name__,
                "prompt": preview_text(prompt, max_chars=100),
                "model": self.model,
                "provider": self.__class__.__name__
            }, source=self.__class__.__name__)

            # Re-raise the exception
            raise e

    def _generate_internal(self,
                          prompt: str,
                          messages: Optional[List[Dict[str, str]]] = None,
                          system_prompt: Optional[str] = None,
                          tools: Optional[List[Dict[str, Any]]] = None,
                          media: Optional[List['MediaContent']] = None,
                          stream: bool = False,
                          response_model: Optional[Type[BaseModel]] = None,
                          execute_tools: Optional[bool] = None,
                          media_metadata: Optional[List[Dict[str, Any]]] = None,
                          **kwargs) -> Union[GenerateResponse, Iterator[GenerateResponse]]:
        """
        Internal generation method to be implemented by subclasses.
        This is called by generate_with_telemetry.

        Args:
            prompt: The input prompt
            messages: Optional conversation history
            system_prompt: Optional system prompt
            tools: Optional list of available tools
            media: Optional list of processed MediaContent objects
            stream: Whether to stream the response
            response_model: Optional Pydantic model for structured output
            execute_tools: Whether to execute tools automatically (True) or let agent handle execution (False)
            **kwargs: Additional provider-specific parameters
        """
        raise NotImplementedError("Subclasses must implement _generate_internal")

    def _enhance_response_with_visual_tokens(self, response: GenerateResponse, media_metadata: List[Dict[str, Any]]) -> GenerateResponse:
        """
        Enhance the response with visual token calculations for Glyph compression.
        This method is called automatically by BaseProvider for all providers.
        """
        try:
            # Calculate visual tokens using VLM token calculator
            provider_name = self.provider or self.__class__.__name__.lower().replace('provider', '')
            self.logger.debug(f"Calculating visual tokens for provider={provider_name}, model={self.model}")

            visual_tokens = self._calculate_visual_tokens(media_metadata, provider_name, self.model)
            self.logger.debug(f"Calculated visual tokens: {visual_tokens}")

            if visual_tokens > 0:
                # Ensure response has metadata
                if not response.metadata:
                    response.metadata = {}

                # Add visual token information to metadata
                response.metadata['visual_tokens'] = visual_tokens

                # Ensure response has usage dict
                if not response.usage:
                    response.usage = {}

                # Add visual tokens to usage
                response.usage['visual_tokens'] = visual_tokens

                # Update total tokens to include visual tokens
                original_total = response.usage.get('total_tokens', 0)
                response.usage['total_tokens'] = original_total + visual_tokens

                self.logger.info(f"Enhanced response with {visual_tokens} visual tokens (new total: {response.usage['total_tokens']})")
            else:
                self.logger.debug("No visual tokens calculated - skipping enhancement")

        except Exception as e:
            self.logger.warning(f"Failed to enhance response with visual tokens: {e}")

        return response

    def _calculate_visual_tokens(self, media_metadata: List[Dict[str, Any]], provider: str, model: str) -> int:
        """Calculate visual tokens from media metadata using VLM token calculator."""
        try:
            from ..utils.vlm_token_calculator import VLMTokenCalculator
            from pathlib import Path

            calculator = VLMTokenCalculator()
            total_visual_tokens = 0

            self.logger.debug(f"Processing {len(media_metadata)} media metadata items")

            for i, metadata in enumerate(media_metadata):
                self.logger.debug(f"Metadata {i}: processing_method={metadata.get('processing_method')}")

                # Check if this is Glyph compression
                if metadata.get('processing_method') == 'direct_pdf_conversion':
                    glyph_cache_dir = metadata.get('glyph_cache_dir')
                    total_images = metadata.get('total_images', 0)

                    self.logger.debug(f"Glyph metadata found: cache_dir={glyph_cache_dir}, total_images={total_images}")

                    if glyph_cache_dir and Path(glyph_cache_dir).exists():
                        # Get actual image paths
                        cache_dir = Path(glyph_cache_dir)
                        image_paths = list(cache_dir.glob("image_*.png"))

                        self.logger.debug(f"Found {len(image_paths)} images in cache directory")

                        if image_paths:
                            # Calculate tokens for all images
                            token_analysis = calculator.calculate_tokens_for_images(
                                image_paths=image_paths,
                                provider=provider,
                                model=model
                            )
                            total_visual_tokens += token_analysis['total_tokens']

                            self.logger.debug(f"Calculated {token_analysis['total_tokens']} visual tokens for {len(image_paths)} Glyph images")
                        else:
                            # Fallback: estimate based on total_images
                            base_tokens = calculator.PROVIDER_CONFIGS.get(provider, {}).get('base_tokens', 512)
                            estimated_tokens = total_images * base_tokens
                            total_visual_tokens += estimated_tokens

                            self.logger.debug(f"Estimated {estimated_tokens} visual tokens for {total_images} Glyph images (fallback)")
                    else:
                        self.logger.debug(f"Cache directory not found or doesn't exist: {glyph_cache_dir}")

            self.logger.debug(f"Total visual tokens calculated: {total_visual_tokens}")
            return total_visual_tokens

        except Exception as e:
            self.logger.warning(f"Failed to calculate visual tokens: {e}")
            return 0

    def _initialize_token_limits(self):
        """Initialize default token limits based on model capabilities"""
        # Set default max_tokens if not provided
        if self.max_tokens is None:
            self.max_tokens = self._get_default_context_window()

        # Set default max_output_tokens if not provided
        if self.max_output_tokens == 2048:  # Only if still default value
            default_max_output = self._get_default_max_output_tokens()
            if default_max_output != 2048:  # If we found a different default
                self.max_output_tokens = default_max_output

        # Validate parameters after setting defaults
        self._validate_token_parameters()

        # Run enhanced validation and warn about potential issues
        self._check_token_configuration_warnings()

    def _check_token_configuration_warnings(self):
        """Check token configuration and emit warnings for potential issues"""
        import warnings

        try:
            warnings_list = self.validate_token_constraints()
            for warning in warnings_list[:3]:  # Limit to first 3 warnings to avoid spam
                warnings.warn(f"Token configuration warning for {self.model}: {warning}",
                             UserWarning, stacklevel=4)

            # Also log warnings for debugging
            if warnings_list and hasattr(self, 'logger'):
                self.logger.debug(f"Token configuration warnings for {self.model}: {'; '.join(warnings_list)}")

        except Exception as e:
            # Don't fail provider initialization due to validation warnings
            if hasattr(self, 'logger'):
                self.logger.debug(f"Error checking token configuration warnings: {e}")

    def _get_default_max_output_tokens(self) -> int:
        """Get default max_output_tokens using JSON capabilities as single source of truth"""
        from ..architectures import get_model_capabilities

        # Fallback chain: Exact model → Model family → Provider defaults → Global defaults
        capabilities = get_model_capabilities(self.model)

        if capabilities:
            max_output_tokens = capabilities.get('max_output_tokens')
            if max_output_tokens:
                self.logger.debug(f"Using max_output_tokens {max_output_tokens} from model capabilities for {self.model}")
                return max_output_tokens

        # If no exact match, try model family/generation fallback
        model_lower = self.model.lower()

        # Family-based fallback patterns (same as context window)
        family_patterns = {
            'gpt-4': ['gpt-4', 'gpt4'],
            'gpt-3.5': ['gpt-3.5', 'gpt3.5'],
            'claude-3': ['claude-3'],
            'claude-3.5': ['claude-3.5'],
            'llama': ['llama'],
            'qwen': ['qwen'],
            'mistral': ['mistral']
        }

        for family, patterns in family_patterns.items():
            if any(pattern in model_lower for pattern in patterns):
                family_caps = get_model_capabilities(family)
                if family_caps and family_caps.get('max_output_tokens'):
                    max_output_tokens = family_caps['max_output_tokens']
                    self.logger.debug(f"Using max_output_tokens {max_output_tokens} from family {family} for {self.model}")
                    return max_output_tokens

        # Use JSON capabilities as single source of truth for defaults
        from ..architectures import get_context_limits
        limits = get_context_limits(self.model)
        max_output_tokens = limits['max_output_tokens']
        self.logger.debug(f"Using default max_output_tokens {max_output_tokens} from model_capabilities.json for {self.model}")
        return max_output_tokens

    def _get_default_context_window(self) -> int:
        """Get default context window using JSON capabilities as single source of truth"""
        from ..architectures import get_model_capabilities

        # Fallback chain: Exact model → Model family → Provider defaults → Global defaults
        capabilities = get_model_capabilities(self.model)

        if capabilities:
            max_tokens = capabilities.get('max_tokens')
            if max_tokens:
                self.logger.debug(f"Using max_tokens {max_tokens} from model capabilities for {self.model}")
                return max_tokens

        # If no exact match, try model family/generation fallback
        model_lower = self.model.lower()

        # Family-based fallback patterns
        family_patterns = {
            'gpt-4': ['gpt-4', 'gpt4'],
            'gpt-3.5': ['gpt-3.5', 'gpt3.5'],
            'claude-3': ['claude-3'],
            'claude-3.5': ['claude-3.5'],
            'llama': ['llama'],
            'qwen': ['qwen'],
            'mistral': ['mistral']
        }

        for family, patterns in family_patterns.items():
            if any(pattern in model_lower for pattern in patterns):
                family_caps = get_model_capabilities(family)
                if family_caps and family_caps.get('max_tokens'):
                    max_tokens = family_caps['max_tokens']
                    self.logger.debug(f"Using max_tokens {max_tokens} from family {family} for {self.model}")
                    return max_tokens

        # Use JSON capabilities as single source of truth for defaults
        from ..architectures import get_context_limits
        limits = get_context_limits(self.model)
        max_tokens = limits['max_tokens']
        self.logger.debug(f"Using default max_tokens {max_tokens} from model_capabilities.json for {self.model}")
        return max_tokens

    def _prepare_generation_kwargs(self, **kwargs) -> Dict[str, Any]:
        """
        Prepare generation kwargs by translating unified token parameters
        to provider-specific ones.

        Args:
            **kwargs: Generation parameters including unified token params

        Returns:
            Dictionary with provider-specific parameters
        """
        # Get effective token limits
        max_tokens, max_output_tokens, max_input_tokens = self._calculate_effective_token_limits()

        # Override max_output_tokens if provided in kwargs
        effective_max_output = kwargs.get("max_output_tokens", max_output_tokens)
        # Safety clamp: never exceed the provider/model's configured max_output_tokens.
        #
        # Upstream callers (runtimes/agents) may request large output budgets based on
        # stale capabilities or user configuration. Providers should not forward values
        # that violate the model's hard limits (Anthropic returns 400 for this).
        try:
            if effective_max_output is None:
                effective_max_output_i = int(max_output_tokens)
            else:
                effective_max_output_i = int(effective_max_output)
        except Exception:
            effective_max_output_i = int(max_output_tokens)
        if effective_max_output_i <= 0:
            effective_max_output_i = int(max_output_tokens)
        if effective_max_output_i > int(max_output_tokens):
            effective_max_output_i = int(max_output_tokens)

        # Return base kwargs with unified parameter
        result_kwargs = kwargs.copy()
        result_kwargs["max_output_tokens"] = effective_max_output_i

        # Add unified generation parameters with fallback hierarchy:
        # kwargs -> constructor/model metadata -> legacy defaults.
        temperature = result_kwargs.get("temperature", self.temperature)
        if temperature is None:
            temperature = self.temperature
        result_kwargs["temperature"] = temperature

        for key in ("top_p", "top_k"):
            if key in result_kwargs:
                if result_kwargs.get(key) is None:
                    result_kwargs.pop(key, None)
                continue
            value = getattr(self, key, None)
            if value is None:
                value = self._metadata_generation_default(key, None)
            if value is not None:
                result_kwargs[key] = value

        seed_value = self._normalize_seed(result_kwargs.get("seed", self.seed))
        if seed_value is not None:
            result_kwargs["seed"] = seed_value
        else:
            # Do not forward seed when unset/random (None or negative sentinel like -1).
            result_kwargs.pop("seed", None)

        return result_kwargs

    @staticmethod
    def _normalize_seed(seed: Any) -> Optional[int]:
        """Normalize seed semantics across providers.

        - None or any negative value -> None (meaning: don't send a provider seed / random).
        - Non-bool numeric-ish values -> int(seed) if >= 0.
        """
        try:
            if seed is None:
                return None
            if isinstance(seed, bool):
                return None
            seed_i = int(seed)
            return seed_i if seed_i >= 0 else None
        except Exception:
            return None

    def _extract_generation_params(self, **kwargs) -> Dict[str, Any]:
        """
        Extract generation parameters with consistent fallback hierarchy.

        Returns:
            Dict containing temperature, seed, and other generation parameters
        """
        params = {}

        # Temperature (always present)
        temperature = kwargs.get("temperature", self.temperature)
        if temperature is None:
            temperature = self.temperature
        params["temperature"] = temperature

        for key in ("top_p", "top_k"):
            if key in kwargs:
                value = kwargs.get(key)
            else:
                value = getattr(self, key, self._metadata_generation_default(key, None))
            if value is not None:
                params[key] = value

        # Seed (only if not None)
        seed_value = self._normalize_seed(kwargs.get("seed", self.seed))
        if seed_value is not None:
            params["seed"] = seed_value

        return params

    def _get_provider_max_tokens_param(self, kwargs: Dict[str, Any]) -> int:
        """
        Extract the appropriate max tokens parameter for this provider.
        This should be overridden by subclasses to return the provider-specific
        parameter name and value.

        Args:
            kwargs: Generation parameters

        Returns:
            Max output tokens for the provider's API
        """
        return kwargs.get("max_output_tokens", self.max_output_tokens)

    def _handle_prompted_tool_execution(self, response: GenerateResponse, tools: List[Dict[str, Any]], execute_tools_param: bool = None) -> GenerateResponse:
        """Handle tool execution for prompted responses (shared implementation)"""
        if not response.content:
            return response

        # Parse tool calls from response content using UniversalToolHandler
        tool_call_response = self.tool_handler.parse_response(response.content, mode="prompted")
        tool_calls = tool_call_response.tool_calls

        if not tool_calls:
            return response

        # Execute with events and return result
        return self._execute_tools_with_events(response, tool_calls, execute_tools_param)

    def _execute_tools_with_events(self, response: GenerateResponse, tool_calls: List, execute_tools_param: bool = None) -> GenerateResponse:
        """Core tool execution with event emission (shared implementation)"""
        # Check if tool execution is enabled
        should_execute = execute_tools_param if execute_tools_param is not None else self.execute_tools

        if not should_execute:
            # Tool execution disabled - return response with tool calls but don't execute
            self.logger.debug(
                "Provider-side tool execution disabled (expected for runtime/host tool execution); "
                "returning response with tool calls."
            )
            return response

        # Emit tool started event
        event_data = {
            "tool_calls": [{
                "name": call.name,
                "arguments": call.arguments
            } for call in tool_calls],
            "tool_count": len(tool_calls),
            "model": self.model,
            "provider": self.__class__.__name__
        }

        from ..events import emit_global
        emit_global(EventType.TOOL_STARTED, event_data, source=self.__class__.__name__)

        # Execute tools
        tool_results = execute_tools(tool_calls)

        # Emit tool completed event
        after_event_data = {
            "tool_results": [{
                "name": call.name,
                "success": result.success,
                "error": str(result.error) if result.error else None
            } for call, result in zip(tool_calls, tool_results)],
            "successful_count": sum(1 for r in tool_results if r.success),
            "failed_count": sum(1 for r in tool_results if not r.success),
            "model": self.model,
            "provider": self.__class__.__name__
        }

        emit_global(EventType.TOOL_COMPLETED, after_event_data, source=self.__class__.__name__)

        # Track tool calls
        for call, result in zip(tool_calls, tool_results):
            self._track_tool_call(
                tool_name=call.name,
                arguments=call.arguments,
                success=result.success,
                error=result.error if not result.success else None
            )

        # Format tool results and append to response
        results_text = self._format_tool_results(tool_calls, tool_results)

        # Return updated response with tool results
        # Use the cleaned content from tool parsing
        tool_call_response = self.tool_handler.parse_response(response.content, mode="prompted")
        cleaned_content = tool_call_response.content or ""
        return GenerateResponse(
            content=cleaned_content + results_text,
            model=response.model,
            finish_reason=response.finish_reason,
            raw_response=response.raw_response,
            usage=response.usage,
            tool_calls=response.tool_calls,  # Keep original format
            metadata=response.metadata,
        )

    def _format_tool_results(self, tool_calls: List, tool_results: List) -> str:
        """Format tool results with tool transparency (shared implementation)"""
        results_text = "\n\nTool Results:\n"
        for call, result in zip(tool_calls, tool_results):
            # Format parameters for display (limit size)
            params_str = preview_text(str(call.arguments) if call.arguments else "{}", max_chars=100)

            # Show tool name and parameters for transparency
            results_text += f"🔧 Tool: {call.name}({params_str})\n"

            # Show result
            if result.success:
                results_text += f"- {result.output}\n"
            else:
                results_text += f"- Error: {result.error}\n"
            results_text += "\n"  # Add spacing between tool calls

        return results_text

    def _convert_native_tool_calls_to_standard(self, native_tool_calls: List[Dict[str, Any]]) -> List:
        """Convert native API tool calls to standard ToolCall objects (shared implementation)"""
        from ..tools.core import ToolCall
        import json

        tool_calls = []
        for call in native_tool_calls:
            arguments = call.get('arguments', {})
            if isinstance(arguments, str):
                try:
                    arguments = json.loads(arguments)
                except json.JSONDecodeError:
                    arguments = {}

            tool_calls.append(ToolCall(
                name=call.get('name', ''),
                arguments=arguments,
                call_id=call.get('id')
            ))
        return tool_calls

    # Timeout management methods
    def get_timeout(self) -> float:
        """Get the current HTTP request timeout in seconds."""
        return self._timeout

    def set_timeout(self, timeout: float) -> None:
        """Set the HTTP request timeout in seconds."""
        self._timeout = timeout
        # Update HTTP clients if they exist
        self._update_http_client_timeout()

    def get_recovery_timeout(self) -> float:
        """Get the current circuit breaker recovery timeout in seconds."""
        return self.retry_manager.config.recovery_timeout

    def set_recovery_timeout(self, timeout: float) -> None:
        """Set the circuit breaker recovery timeout in seconds."""
        self.retry_manager.config.recovery_timeout = timeout

    def get_tool_timeout(self) -> float:
        """Get the current tool execution timeout in seconds."""
        return self._tool_timeout

    def set_tool_timeout(self, timeout: float) -> None:
        """Set the tool execution timeout in seconds."""
        self._tool_timeout = timeout

    def _update_http_client_timeout(self) -> None:
        """Update HTTP client timeout if the provider has one. Override in subclasses."""
        pass

    # Prompt cache management methods
    def supports_prompt_cache(self) -> bool:
        """Return True if this provider supports best-effort prompt caching.

        Semantics differ by provider:
        - Remote providers (OpenAI): `prompt_cache_key` is forwarded; cache is managed server-side.
        - Local providers (MLX / llama.cpp): in-process KV/prefix caches can be retained across calls.
        """
        return False

    def prompt_cache_supports_kv_source_of_truth(self) -> bool:
        """Return True if this provider can use the prompt cache as the context source-of-truth.

        When True, callers may omit `system_prompt`, `tools`, and `messages` and send only the
        delta user `prompt` each turn, relying on the in-process KV cache keyed by
        `prompt_cache_key` to retain prior context (append-only semantics).
        """
        return False

    def get_prompt_cache_capabilities(self) -> PromptCacheCapabilities:
        """Return the provider prompt-cache capability profile.

        Providers may override this for richer capability reporting. The default implementation keeps
        backward compatibility with existing providers that only override `supports_prompt_cache()`
        and the optional backend hooks.
        """
        supported = bool(self.supports_prompt_cache())
        if not supported:
            return PromptCacheCapabilities()

        create_overridden = type(self)._prompt_cache_backend_create is not BaseProvider._prompt_cache_backend_create
        clone_overridden = type(self)._prompt_cache_backend_clone is not BaseProvider._prompt_cache_backend_clone
        append_overridden = type(self)._prompt_cache_backend_append is not BaseProvider._prompt_cache_backend_append
        set_overridden = type(self).prompt_cache_set is not BaseProvider.prompt_cache_set
        save_overridden = type(self).prompt_cache_save is not BaseProvider.prompt_cache_save
        load_overridden = type(self).prompt_cache_load is not BaseProvider.prompt_cache_load

        supports_update = bool(create_overridden and append_overridden)
        supports_fork = bool(create_overridden and clone_overridden)
        supports_prepare_modules = bool(create_overridden and clone_overridden and append_overridden)
        supports_ttl = bool(supports_prepare_modules or supports_update or supports_fork or set_overridden)
        mode = "local_control_plane" if supports_prepare_modules else "keyed"

        notes: List[str] = []
        if mode == "keyed":
            notes.append(
                "Provider accepts prompt cache keys but does not expose the full local prompt-cache control plane."
            )

        return PromptCacheCapabilities(
            supported=True,
            mode=mode,
            supports_set=True,
            supports_clear=True,
            supports_update=supports_update,
            supports_fork=supports_fork,
            supports_prepare_modules=supports_prepare_modules,
            supports_stats=True,
            supports_save=bool(save_overridden),
            supports_load=bool(load_overridden),
            supports_ttl=supports_ttl,
            notes=tuple(notes),
        )

    def prompt_cache_supports_operation(self, operation: str) -> bool:
        return self.get_prompt_cache_capabilities().supports_operation(operation)

    def prompt_cache_artifact_extension(self) -> str:
        """Return the default on-disk extension for durable prompt-cache artifacts."""
        return ".safetensors"

    def prompt_cache_cache_backend(self) -> str:
        """Return a stable backend id for in-process prompt-cache artifacts."""
        provider = str(getattr(self, "provider", "") or "").strip().lower()
        return provider or self.__class__.__name__.lower()

    def prompt_cache_artifact_format(self) -> str:
        """Return a stable provider artifact format id for durable prompt-cache artifacts."""
        return f"abstractcore-{self.prompt_cache_cache_backend()}-prompt-cache/v1"

    def prompt_cache_render_fragment(
        self,
        *,
        prompt: str = "",
        messages: Optional[List[Dict[str, Any]]] = None,
        system_prompt: Optional[str] = None,
        tools: Optional[List[Dict[str, Any]]] = None,
        add_generation_prompt: bool = False,
        prefilled_modules: Optional[Sequence[str]] = None,
    ) -> Optional[PromptCacheRenderedFragment]:
        """Render prompt input exactly as this provider would append it to a prompt cache.

        Providers with durable bloc-artifact support override this. Returning None means the
        provider cannot safely expose an exact rendered-prefix artifact for this request.
        """
        _ = (prompt, messages, system_prompt, tools, add_generation_prompt, prefilled_modules)
        return None

    def prompt_cache_key_meta(self, key: Any) -> Dict[str, Any]:
        """Return JSON-safe metadata for an in-process prompt-cache key."""
        normalized = self._normalize_prompt_cache_key(key)
        if normalized is None:
            return {}
        try:
            meta = self._prompt_cache_store.meta(normalized)
            return dict(meta or {}) if isinstance(meta, dict) else {}
        except Exception:
            return {}

    def prompt_cache_update_key_meta(self, key: Any, **updates: Any) -> bool:
        """Best-effort: merge metadata into an existing prompt-cache key."""
        normalized = self._normalize_prompt_cache_key(key)
        if normalized is None:
            return False
        store = getattr(self, "_prompt_cache_store", None)
        lock = getattr(store, "_lock", None)
        entries = getattr(store, "_entries", None)
        if lock is None or entries is None:
            return False
        try:
            with lock:
                entry = entries.get(normalized)
                if entry is None:
                    return False
                meta = dict(getattr(entry, "meta", {}) or {})
                for field_name, field_value in updates.items():
                    if field_value is not None:
                        meta[str(field_name)] = field_value
                entry.meta = meta
            return True
        except Exception:
            return False

    def prompt_cache_validate_binding(self, key: Any, binding: Dict[str, Any]) -> Dict[str, Any]:
        """Validate an optional exact-artifact binding against a live prompt-cache key.

        `prompt_cache_key` remains best-effort. This method is only used when a caller supplies an
        explicit binding returned by a durable bloc artifact load operation.
        """
        normalized = self._normalize_prompt_cache_key(key)
        if normalized is None:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                "Prompt cache binding validation requires a non-empty prompt cache key.",
                operation="binding",
                provider=provider,
                model=model,
                code="prompt_cache_binding_invalid_key",
                capabilities=self.get_prompt_cache_capabilities(),
            )
        if not isinstance(binding, dict) or not binding:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                "Prompt cache binding validation requires a non-empty binding object.",
                operation="binding",
                provider=provider,
                model=model,
                code="prompt_cache_binding_invalid",
                capabilities=self.get_prompt_cache_capabilities(),
            )

        meta = self.prompt_cache_key_meta(normalized)
        if not meta:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                f"Prompt cache key '{normalized}' is not loaded or has no binding metadata.",
                operation="binding",
                provider=provider,
                model=model,
                code="prompt_cache_binding_missing",
                capabilities=self.get_prompt_cache_capabilities(),
            )

        expected_fields = (
            "binding_id",
            "artifact_sha256",
            "bloc_sha256",
            "content_sha256",
            "rendered_recipe_sha256",
            "manifest_version",
            "provider",
            "model",
        )
        mismatches: Dict[str, Dict[str, str]] = {}
        for field_name in expected_fields:
            expected = binding.get(field_name)
            if expected is None:
                continue
            actual = meta.get(field_name)
            if str(actual).strip() != str(expected).strip():
                mismatches[field_name] = {
                    "expected": str(expected).strip(),
                    "actual": str(actual).strip() if actual is not None else "",
                }
        if mismatches:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                f"Prompt cache key '{normalized}' does not match the expected durable bloc binding.",
                operation="binding",
                provider=provider,
                model=model,
                code="prompt_cache_binding_mismatch",
                capabilities=self.get_prompt_cache_capabilities(),
            )

        return {"key": normalized, "meta": meta}

    def _apply_prompt_cache_binding_request(self, kwargs: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        binding = kwargs.pop("expected_prompt_cache_binding", None)
        second = kwargs.pop("prompt_cache_binding", None)
        if binding is None:
            binding = second
        elif second is not None and second != binding:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                "expected_prompt_cache_binding and prompt_cache_binding must match when both are supplied.",
                operation="binding",
                provider=provider,
                model=model,
                code="prompt_cache_binding_conflict",
                capabilities=self.get_prompt_cache_capabilities(),
            )
        if binding is None:
            return None
        if isinstance(binding, str):
            binding = {"binding_id": binding.strip()}
        if not isinstance(binding, dict):
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                "prompt_cache_binding must be an object.",
                operation="binding",
                provider=provider,
                model=model,
                code="prompt_cache_binding_invalid",
                capabilities=self.get_prompt_cache_capabilities(),
            )
        binding_key = binding.get("key")
        current_key = kwargs.get("prompt_cache_key")
        if isinstance(binding_key, str) and binding_key.strip():
            binding_key_s = binding_key.strip()
            if isinstance(current_key, str) and current_key.strip() and current_key.strip() != binding_key_s:
                provider, model = self._prompt_cache_error_context()
                raise PromptCacheOperationError(
                    "prompt_cache_key and prompt_cache_binding.key must match.",
                    operation="binding",
                    provider=provider,
                    model=model,
                    code="prompt_cache_binding_key_mismatch",
                    capabilities=self.get_prompt_cache_capabilities(),
                )
            kwargs["prompt_cache_key"] = binding_key_s
            current_key = binding_key_s
        self.prompt_cache_validate_binding(current_key, binding)
        return dict(binding)

    def _prompt_cache_error_context(self) -> Tuple[Optional[str], Optional[str]]:
        provider = str(getattr(self, "provider", "") or "").strip() or self.__class__.__name__
        model = str(getattr(self, "model", "") or "").strip() or None
        return provider, model

    def _require_prompt_cache_operation(self, operation: str) -> PromptCacheCapabilities:
        caps = self.get_prompt_cache_capabilities()
        if not caps.supports_operation(operation):
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheUnsupportedError(
                operation=operation,
                provider=provider,
                model=model,
                capabilities=caps,
            )
        return caps

    # Provider-specific prompt cache backend hooks (optional)
    #
    # Providers that implement in-process KV caching (MLX, llama.cpp, etc.) can override these to enable
    # `prompt_cache_update`, `prompt_cache_fork`, and `prompt_cache_prepare_modules`.
    def _prompt_cache_backend_create(self) -> Optional[Any]:
        return None

    def _prompt_cache_backend_clone(self, cache_value: Any) -> Optional[Any]:
        _ = cache_value
        return None

    def _prompt_cache_backend_append(
        self,
        cache_value: Any,
        *,
        prompt: str = "",
        messages: Optional[List[Dict[str, Any]]] = None,
        system_prompt: Optional[str] = None,
        tools: Optional[List[Dict[str, Any]]] = None,
        add_generation_prompt: bool = False,
        **kwargs,
    ) -> bool:
        _ = (cache_value, prompt, messages, system_prompt, tools, add_generation_prompt, kwargs)
        return False

    def _prompt_cache_backend_token_count(self, cache_value: Any) -> Optional[int]:
        _ = cache_value
        return None

    def _normalize_prompt_cache_key(self, key: Any) -> Optional[str]:
        if not isinstance(key, str):
            return None
        key = key.strip()
        return key if key else None

    def _apply_default_prompt_cache_key(self, kwargs: Dict[str, Any]) -> None:
        # Explicit caller override wins (even if None / empty to disable).
        if "prompt_cache_key" in kwargs:
            kwargs["prompt_cache_key"] = self._normalize_prompt_cache_key(kwargs.get("prompt_cache_key"))
            return

        if self._default_prompt_cache_key and self.supports_prompt_cache():
            kwargs["prompt_cache_key"] = self._default_prompt_cache_key

    def get_prompt_cache_stats(self) -> Dict[str, Any]:
        """Return basic prompt cache stats (in-process store only)."""
        caps = self._require_prompt_cache_operation("stats")
        stats = self._prompt_cache_store.stats()
        stats["default_key"] = self._default_prompt_cache_key
        stats["capabilities"] = caps.to_dict()
        try:
            keys = self._prompt_cache_store.keys()
            if isinstance(keys, list):
                stats["keys"] = list(keys)
                meta_by_key: Dict[str, Any] = {}
                for k in keys:
                    meta = self._prompt_cache_store.meta(k)
                    if isinstance(meta, dict) and meta:
                        meta_by_key[str(k)] = dict(meta)
                if meta_by_key:
                    stats["meta_by_key"] = meta_by_key
        except Exception:
            pass
        return stats

    def prompt_cache_token_count(self, key: Optional[str] = None) -> Optional[int]:
        """Best-effort: return the current token count for an in-process cache key.

        Notes:
        - For local control-plane providers (MLX / GGUF / HF transformers KV reuse), the cache value
          stored in-process is mutated over time; this method queries the backend directly.
        - For server-managed caches (`mode="keyed"`), this typically returns None because there is
          no local KV/prefix state to introspect.
        """
        if not self.supports_prompt_cache():
            return None

        normalized = self._default_prompt_cache_key if key is None else self._normalize_prompt_cache_key(key)
        if normalized is None:
            return None

        cache_value = None
        try:
            cache_value = self._prompt_cache_store.get(normalized)
        except Exception:
            cache_value = None

        if cache_value is not None:
            try:
                tok = self._prompt_cache_backend_token_count(cache_value)
            except Exception:
                tok = None
            if isinstance(tok, int) and tok >= 0:
                return tok

        try:
            meta = self._prompt_cache_store.meta(normalized) or {}
            tok = meta.get("token_count")
            tok_i = int(tok)
            return tok_i if tok_i >= 0 else None
        except Exception:
            return None

    def prompt_cache_set(self, key: str, *, make_default: bool = True, **kwargs) -> bool:
        """Set the default prompt cache key for this provider instance.

        Provider-specific cache allocation/warming is implemented by subclasses when applicable.
        """
        caps = self._require_prompt_cache_operation("set")
        normalized = self._normalize_prompt_cache_key(key)
        if normalized is None:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                "Prompt cache key must be a non-empty string.",
                operation="set",
                provider=provider,
                model=model,
                code="prompt_cache_invalid_key",
                capabilities=caps,
            )
        _ = kwargs
        # Best-effort: allocate backend cache if the provider supports it.
        if self._prompt_cache_store.get(normalized) is None:
            created = self._prompt_cache_backend_create()
            if created is not None:
                try:
                    self._prompt_cache_store.set(normalized, created, meta={"backend": "provider"})
                except Exception:
                    pass
        if make_default:
            self._default_prompt_cache_key = normalized
        return True

    def prompt_cache_update(
        self,
        key: str,
        *,
        prompt: str = "",
        messages: Optional[List[Dict[str, Any]]] = None,
        system_prompt: Optional[str] = None,
        tools: Optional[List[Dict[str, Any]]] = None,
        add_generation_prompt: bool = False,
        ttl_s: Optional[float] = None,
        **kwargs,
    ) -> bool:
        """Append new prompt context into an existing cache key (best-effort).

        Semantics:
        - Local runtimes can implement true KV prefill updates (append-only).
        - Remote providers typically cannot be “pre-filled” explicitly; they may ignore this.
        - `thinking` is applied through the same unified reasoning-control path as `generate()`
          before the cache fragment is appended.

        Arguments are intentionally similar to `generate()` so higher-level code can reuse its own
        prompt/module construction logic.
        """
        caps = self._require_prompt_cache_operation("update")
        normalized = self._normalize_prompt_cache_key(key)
        if normalized is None:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                "Prompt cache key must be a non-empty string.",
                operation="update",
                provider=provider,
                model=model,
                code="prompt_cache_invalid_key",
                capabilities=caps,
            )

        # Ensure the cache exists if the provider can allocate a backend cache object.
        cache_value = self._prompt_cache_store.get(normalized)
        if cache_value is None:
            self.prompt_cache_set(normalized, make_default=False)
            cache_value = self._prompt_cache_store.get(normalized)
            if cache_value is None:
                provider, model = self._prompt_cache_error_context()
                raise PromptCacheOperationError(
                    f"Provider did not create prompt cache state for key '{normalized}'.",
                    operation="update",
                    provider=provider,
                    model=model,
                    code="prompt_cache_missing_backend_state",
                    capabilities=caps,
                )

        thinking = kwargs.pop("thinking", None)
        carried_thinking_meta = kwargs.pop("_acore_thinking_meta", None)
        if carried_thinking_meta is not None and not isinstance(carried_thinking_meta, dict):
            carried_thinking_meta = None

        prompt_text = str(prompt or "")
        messages, system_prompt = self._normalize_developer_messages(messages, system_prompt)
        prompt_text, messages, system_prompt, kwargs, thinking_meta = self._apply_thinking_request(
            thinking=thinking,
            prompt=prompt_text,
            messages=messages,
            system_prompt=system_prompt,
            kwargs=kwargs,
        )
        if thinking_meta is None and carried_thinking_meta is not None:
            thinking_meta = carried_thinking_meta
        _ = thinking_meta

        ok = self._prompt_cache_backend_append(
            cache_value,
            prompt=prompt_text,
            messages=messages,
            system_prompt=system_prompt,
            tools=tools,
            add_generation_prompt=bool(add_generation_prompt),
            **kwargs,
        )
        if not ok:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                f"Provider failed to append prompt context into cache key '{normalized}'.",
                operation="update",
                provider=provider,
                model=model,
                code="prompt_cache_append_failed",
                capabilities=caps,
            )

        # Update TTL/metadata best-effort.
        try:
            meta = self._prompt_cache_store.meta(normalized) or {}
            meta = dict(meta)
            tok = self._prompt_cache_backend_token_count(cache_value)
            if isinstance(tok, int) and tok >= 0:
                meta["token_count"] = tok
            if ttl_s is not None or meta:
                self._prompt_cache_store.set(normalized, cache_value, ttl_s=ttl_s, meta=meta)
        except Exception:
            pass
        return True

    def prompt_cache_fork(
        self,
        from_key: str,
        to_key: str,
        *,
        make_default: bool = False,
        ttl_s: Optional[float] = None,
        **kwargs,
    ) -> bool:
        """Create a new cache key by cloning another cache (best-effort).

        This is the primitive needed for hierarchical/module caches:
        - build stable shared prefixes (persona, memory blueprints, tool schemas)
        - fork them into per-session caches that can be appended/mutated safely.
        """
        caps = self._require_prompt_cache_operation("fork")
        _ = kwargs
        src = self._normalize_prompt_cache_key(from_key)
        dst = self._normalize_prompt_cache_key(to_key)
        if src is None or dst is None:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                "Source and destination prompt cache keys must be non-empty strings.",
                operation="fork",
                provider=provider,
                model=model,
                code="prompt_cache_invalid_key",
                capabilities=caps,
            )

        src_value = self._prompt_cache_store.get(src)
        if src_value is None:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                f"Source prompt cache key '{src}' does not exist.",
                operation="fork",
                provider=provider,
                model=model,
                code="prompt_cache_missing_source_key",
                capabilities=caps,
            )

        cloned = self._prompt_cache_backend_clone(src_value)
        if cloned is None:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                f"Provider failed to clone prompt cache key '{src}'.",
                operation="fork",
                provider=provider,
                model=model,
                code="prompt_cache_clone_failed",
                capabilities=caps,
            )

        try:
            meta = self._prompt_cache_store.meta(src) or {}
            meta = dict(meta)
            meta.setdefault("forked_from", src)
            tok = self._prompt_cache_backend_token_count(cloned)
            if isinstance(tok, int) and tok >= 0:
                meta["token_count"] = tok
            self._prompt_cache_store.set(dst, cloned, ttl_s=ttl_s, meta=meta)
        except Exception:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                f"Provider failed to persist forked prompt cache key '{dst}'.",
                operation="fork",
                provider=provider,
                model=model,
                code="prompt_cache_store_failed",
                capabilities=caps,
            )

        if make_default:
            self._default_prompt_cache_key = dst
        return True

    def prompt_cache_prepare_modules(
        self,
        *,
        namespace: str,
        modules: List[Union[PromptCacheModule, Dict[str, Any]]],
        make_default: bool = False,
        ttl_s: Optional[float] = None,
        version: int = 1,
    ) -> Dict[str, Any]:
        """Ensure hierarchical prefix caches exist for an ordered module list (best-effort).

        This builds immutable prefix caches (by derived keys) so callers can:
        - reuse stable sub-prefixes (persona, memory blueprints, etc.)
        - fork the final prefix into a per-session cache for incremental chat

        Returns a JSON-serializable dict containing per-module derived keys.
        """
        caps = self._require_prompt_cache_operation("prepare_modules")
        ns = str(namespace or "").strip()
        if not ns:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                "Prompt cache module preparation requires a non-empty namespace.",
                operation="prepare_modules",
                provider=provider,
                model=model,
                code="prompt_cache_invalid_namespace",
                capabilities=caps,
            )

        normalized_modules: List[PromptCacheModule] = []
        fingerprints: List[str] = []
        for idx, m in enumerate(modules or []):
            try:
                if isinstance(m, PromptCacheModule):
                    mod = m.normalized()
                elif isinstance(m, dict):
                    mod = PromptCacheModule(**m).normalized()
                else:
                    raise TypeError(f"Expected PromptCacheModule or dict, got {type(m).__name__}")

                if not mod.module_id:
                    raise ValueError("module_id must be a non-empty string")

                fp = mod.fingerprint(version=version)
                normalized_modules.append(mod)
                fingerprints.append(fp)
            except Exception as e:
                provider, model = self._prompt_cache_error_context()
                raise PromptCacheOperationError(
                    f"Invalid prompt cache module at index {idx}: {e}",
                    operation="prepare_modules",
                    provider=provider,
                    model=model,
                    code="prompt_cache_invalid_module",
                    capabilities=caps,
                ) from e

        if not normalized_modules:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                "Prompt cache module preparation requires at least one valid module.",
                operation="prepare_modules",
                provider=provider,
                model=model,
                code="prompt_cache_no_modules",
                capabilities=caps,
            )

        # Derive deterministic prefix keys per module boundary.
        prefix_hash = hashlib.sha256(f"acore-prompt-cache:{int(version)}".encode("utf-8")).hexdigest()
        derived: List[Dict[str, Any]] = []
        keys: List[str] = []
        for mod, fp in zip(normalized_modules, fingerprints):
            prefix_hash = hashlib.sha256((prefix_hash + fp).encode("utf-8")).hexdigest()
            key = f"{ns}:{prefix_hash[:32]}"
            keys.append(key)
            derived.append({"module_id": mod.module_id, "cache_key": key, "module_hash": fp})

        # Find the longest existing prefix cache.
        start_idx = -1
        for i, key in enumerate(keys):
            if self._prompt_cache_store.get(key) is None:
                break
            start_idx = i

        # Start from existing prefix (clone to avoid mutating the stored snapshot).
        current_cache: Optional[Any] = None
        if start_idx >= 0:
            existing = self._prompt_cache_store.get(keys[start_idx])
            if existing is not None:
                current_cache = self._prompt_cache_backend_clone(existing) or None

        # If we have no starting cache, start from empty backend cache.
        if current_cache is None:
            current_cache = self._prompt_cache_backend_create()
            if current_cache is None:
                provider, model = self._prompt_cache_error_context()
                raise PromptCacheOperationError(
                    "Provider does not implement an in-process prompt-cache backend.",
                    operation="prepare_modules",
                    provider=provider,
                    model=model,
                    code="prompt_cache_backend_unavailable",
                    capabilities=caps,
                )

        # Build missing caches.
        for j in range(start_idx + 1, len(keys)):
            mod = normalized_modules[j]
            ok = self._prompt_cache_backend_append(
                current_cache,
                prompt=str(mod.prompt or ""),
                messages=mod.messages,
                system_prompt=mod.system_prompt,
                tools=mod.tools,
                add_generation_prompt=bool(mod.add_generation_prompt),
            )
            if not ok:
                provider, model = self._prompt_cache_error_context()
                raise PromptCacheOperationError(
                    f"Provider failed to append prompt cache module '{mod.module_id}'.",
                    operation="prepare_modules",
                    provider=provider,
                    model=model,
                    code="prompt_cache_append_failed",
                    capabilities=caps,
                )

            snapshot = self._prompt_cache_backend_clone(current_cache) or None
            if snapshot is None:
                provider, model = self._prompt_cache_error_context()
                raise PromptCacheOperationError(
                    "Provider does not support prompt-cache cloning for module preparation.",
                    operation="prepare_modules",
                    provider=provider,
                    model=model,
                    code="prompt_cache_clone_failed",
                    capabilities=caps,
                )

            meta = {
                "namespace": ns,
                "module_id": mod.module_id,
                "module_hash": fingerprints[j],
                "index": j,
                "backend": "provider",
            }
            tok = self._prompt_cache_backend_token_count(snapshot)
            if isinstance(tok, int) and tok >= 0:
                meta["token_count"] = tok

            self._prompt_cache_store.set(keys[j], snapshot, ttl_s=ttl_s, meta=meta)

        if make_default:
            self._default_prompt_cache_key = keys[-1]

        return {
            "supported": True,
            "capabilities": caps.to_dict(),
            "namespace": ns,
            "version": int(version),
            "modules": derived,
            "final_cache_key": keys[-1],
        }

    def prompt_cache_clear(self, key: Optional[str] = None) -> bool:
        """Clear prompt caches for this provider instance (best-effort)."""
        caps = self._require_prompt_cache_operation("clear")
        normalized = self._normalize_prompt_cache_key(key) if key is not None else None

        if normalized is None:
            self._default_prompt_cache_key = None
            self._prompt_cache_store.clear()
            return True

        cleared = self._prompt_cache_store.delete(normalized)
        if self._default_prompt_cache_key == normalized:
            self._default_prompt_cache_key = None
        if not cleared:
            provider, model = self._prompt_cache_error_context()
            raise PromptCacheOperationError(
                f"Prompt cache key '{normalized}' does not exist.",
                operation="clear",
                provider=provider,
                model=model,
                code="prompt_cache_missing_key",
                capabilities=caps,
            )
        return True

    def prompt_cache_save(self, key: str, filename: str, **kwargs) -> Dict[str, Any]:
        """Save an in-process prompt cache to disk (provider-specific; best-effort).

        Providers that expose serializable KV/prefix caches (for example MLX and some llama.cpp
        backends) override this method.
        """
        _ = (kwargs, filename)
        provider, model = self._prompt_cache_error_context()
        raise PromptCacheUnsupportedError(
            operation="save",
            provider=provider,
            model=model,
            capabilities=self.get_prompt_cache_capabilities(),
            detail="This provider does not support saving prompt caches.",
        )

    def prompt_cache_load(
        self,
        filename: str,
        *,
        key: Optional[str] = None,
        make_default: bool = True,
        **kwargs,
    ) -> Dict[str, Any]:
        """Load a prompt cache from disk into this provider instance (provider-specific; best-effort)."""
        _ = (kwargs, filename, key, make_default)
        provider, model = self._prompt_cache_error_context()
        raise PromptCacheUnsupportedError(
            operation="load",
            provider=provider,
            model=model,
            capabilities=self.get_prompt_cache_capabilities(),
            detail="This provider does not support loading prompt caches.",
        )

    # Memory management methods
    @abstractmethod
    def unload_model(self, model_name: str) -> None:
        """
        Unload/cleanup resources for a specific model.

        This is the single canonical unload entrypoint across providers.
        Providers must implement this as a best-effort cleanup hook:

        - In-process providers (e.g. MLX, HuggingFace): free local model resources.
        - Some self-hosted servers (e.g. Ollama): may request server-side eviction/unload.
        - OpenAI-compatible servers (e.g. LMStudio, vLLM, openai-compatible): typically only close client
          connections; server-side model unloading may not be available and is controlled by the server (TTL/eviction).
        - Cloud APIs (e.g. OpenAI, Anthropic): usually a no-op (safe to call).
        """

    # Token configuration helpers - expose interface methods for user convenience
    def get_token_configuration_summary(self) -> str:
        """Get a human-readable summary of current token configuration"""
        return super().get_token_configuration_summary()

    def validate_token_constraints(self) -> List[str]:
        """Validate token configuration and return warnings/suggestions"""
        warnings_list = super().validate_token_constraints()

        # Embedding models are not text-generative: output token limits are irrelevant and can
        # legitimately be 0 (e.g. Nomic Embed). Suppress misleading output-token warnings.
        try:
            caps = getattr(self, "model_capabilities", None)
            model_type = caps.get("model_type") if isinstance(caps, dict) else None
            if isinstance(model_type, str) and model_type.strip().lower() == "embedding":
                warnings_list = [w for w in warnings_list if "max_output_tokens" not in str(w)]
        except Exception:
            pass

        return warnings_list

    def calculate_token_budget(self, input_text: str, desired_output_tokens: int,
                              safety_margin: float = 0.1) -> tuple[int, List[str]]:
        """Helper to estimate required max_tokens given input and desired output"""
        return super().calculate_token_budget(input_text, desired_output_tokens, safety_margin)

    def estimate_tokens(self, text: str) -> int:
        """Rough estimation of token count for given text"""
        return super().estimate_tokens(text)

    def _process_media_content(self, media: List[Union[str, Dict[str, Any], 'MediaContent']], 
                              glyph_compression: str = "auto") -> List['MediaContent']:
        """
        Process media content from various input formats into standardized MediaContent objects.

        Args:
            media: List of media inputs (file paths, MediaContent objects, or dicts)
            glyph_compression: Glyph compression preference (auto, always, never)

        Returns:
            List of processed MediaContent objects

        Raises:
            ImportError: If media processing dependencies are not available
            ValueError: If media input format is invalid
        """
        if not media:
            return []

        try:
            # Import media handler components
            from ..media import AutoMediaHandler
            from ..media.types import MediaContent
        except ImportError as e:
            raise ImportError(
                f"Media processing requires additional dependencies. "
                f"Install with: pip install \"abstractcore[media]\". Error: {e}"
            )

        processed_media = []

        media_items = self._coerce_media_items(media)
        for i, media_item in enumerate(media_items):
            try:
                if isinstance(media_item, str):
                    # File path - process with auto media handler
                    handler = AutoMediaHandler(
                        enable_glyph_compression=True,
                        glyph_config=getattr(self, 'glyph_config', None)
                    )
                    result = handler.process_file(
                        media_item,
                        provider=self.provider,
                        model=self.model,
                        glyph_compression=glyph_compression
                    )
                    if result.success:
                        processed_media.append(result.media_content)
                    else:
                        self.logger.warning(f"Failed to process media file {media_item}: {result.error_message}")
                        continue

                elif hasattr(media_item, 'media_type'):
                    # Already a MediaContent object
                    processed_media.append(media_item)

                elif isinstance(media_item, dict):
                    file_path_raw = media_item.get("file_path")
                    if file_path_raw is None:
                        file_path_raw = media_item.get("filePath")
                    if file_path_raw is None:
                        file_path_raw = media_item.get("path")
                    has_inline_content = media_item.get("content") is not None
                    if isinstance(file_path_raw, str) and file_path_raw.strip() and not has_inline_content:
                        handler = AutoMediaHandler(
                            enable_glyph_compression=True,
                            glyph_config=getattr(self, 'glyph_config', None)
                        )
                        result = handler.process_file(
                            file_path_raw.strip(),
                            provider=self.provider,
                            model=self.model,
                            glyph_compression=glyph_compression
                        )
                        if result.success:
                            media_content = result.media_content
                            metadata = dict(getattr(media_content, "metadata", None) or {})
                            for key in ("artifact_id", "$artifact", "filename", "role", "purpose", "kind"):
                                value = media_item.get(key)
                                if value is not None and str(value).strip():
                                    metadata[key] = str(value).strip()
                            media_content.metadata = metadata
                            content_type = media_item.get("content_type") or media_item.get("mime_type")
                            if isinstance(content_type, str) and content_type.strip():
                                media_content.mime_type = content_type.strip()
                            processed_media.append(media_content)
                        else:
                            self.logger.warning(
                                f"Failed to process media file {file_path_raw}: {result.error_message}"
                            )
                        continue

                    # Dictionary format - convert to MediaContent
                    try:
                        media_content = MediaContent.from_dict(media_item)
                        processed_media.append(media_content)
                    except Exception as e:
                        self.logger.warning(f"Failed to convert media dict at index {i}: {e}")
                        continue

                else:
                    self.logger.warning(f"Unsupported media type at index {i}: {type(media_item)}")
                    continue

            except Exception as e:
                self.logger.warning(f"Failed to process media item at index {i}: {e}")
                continue

        if not processed_media and media:
            self.logger.warning("No media items were successfully processed")

        return processed_media

    @abstractmethod
    def list_available_models(self, **kwargs) -> List[str]:
        """
        List available models for this provider.

        Each provider must implement this method to return a list of available models.
        The server will use this method to aggregate models across all providers.

        Args:
            **kwargs: Provider-specific parameters including:
                - api_key: API key for authentication (if required)
                - base_url: Base URL for API endpoint (if applicable)
                - input_capabilities: Optional list of ModelInputCapability enums to filter by input capability
                    (e.g., [ModelInputCapability.IMAGE] for vision models)
                - output_capabilities: Optional list of ModelOutputCapability enums to filter by output capability
                    (e.g., [ModelOutputCapability.EMBEDDINGS] for embedding models)

        Returns:
            List of model names available for this provider, optionally filtered by capabilities

        Examples:
            >>> from abstractcore.providers import OpenAIProvider
            >>> from abstractcore.providers.model_capabilities import ModelInputCapability, ModelOutputCapability
            >>>
            >>> # Get all models
            >>> all_models = OpenAIProvider.list_available_models(api_key="...")
            >>>
            >>> # Get models that can analyze images
            >>> vision_models = OpenAIProvider.list_available_models(
            ...     api_key="...",
            ...     input_capabilities=[ModelInputCapability.IMAGE]
            ... )
            >>>
            >>> # Get embedding models
            >>> embedding_models = OpenAIProvider.list_available_models(
            ...     api_key="...",
            ...     output_capabilities=[ModelOutputCapability.EMBEDDINGS]
            ... )
            >>>
            >>> # Get vision models that generate text (most common case)
            >>> vision_text_models = OpenAIProvider.list_available_models(
            ...     api_key="...",
            ...     input_capabilities=[ModelInputCapability.TEXT, ModelInputCapability.IMAGE],
            ...     output_capabilities=[ModelOutputCapability.TEXT]
            ... )

        Note:
            This is an abstract method that MUST be implemented by all provider subclasses.
            Each provider should implement its own discovery logic (API calls, local scanning, etc.).
            Providers should apply the capability filters if provided in kwargs.
        """
        pass

    def health(self, timeout: Optional[float] = 5.0) -> Dict[str, Any]:
        """
        Check provider health and connectivity.

        This method tests if the provider is online and accessible by attempting
        to list available models. A successful model listing indicates the provider
        is healthy and ready to serve requests.

        Args:
            timeout: Maximum time in seconds to wait for health check (default: 5.0).
                    None means unlimited timeout (not recommended for health checks).

        Returns:
            Dict with health status information:
            {
                "status": bool,              # True if provider is healthy/online
                "provider": str,             # Provider class name (e.g., "OpenAIProvider")
                "models": List[str] | None,  # Available models if online, None if offline
                "model_count": int,          # Number of models available (0 if offline)
                "error": str | None,         # Error message if offline, None if healthy
                "latency_ms": float          # Time taken for health check in milliseconds
            }

        Example:
            >>> provider = OllamaProvider(model="llama2")
            >>> health = provider.health(timeout=3.0)
            >>> if health["status"]:
            >>>     print(f"Healthy! {health['model_count']} models available")
            >>> else:
            >>>     print(f"Offline: {health['error']}")

        Note:
            - This method never raises exceptions; errors are captured in the response
            - Uses list_available_models() as the connectivity test
            - Providers can override this method for custom health check logic
        """
        import time as time_module

        start_time = time_module.time()
        provider_name = self.__class__.__name__

        try:
            # Attempt to list models as connectivity test
            # Store original timeout if provider has HTTP client
            original_timeout = None
            timeout_changed = False

            if timeout is not None and hasattr(self, '_timeout'):
                original_timeout = self._timeout
                if original_timeout != timeout:
                    self.set_timeout(timeout)
                    timeout_changed = True

            try:
                models = self.list_available_models()

                # Restore original timeout if changed
                if timeout_changed and original_timeout is not None:
                    self.set_timeout(original_timeout)

                latency_ms = (time_module.time() - start_time) * 1000

                return {
                    "status": True,
                    "provider": provider_name,
                    "models": models,
                    "model_count": len(models) if models else 0,
                    "error": None,
                    "latency_ms": round(latency_ms, 2)
                }

            except Exception as e:
                # Restore original timeout on error
                if timeout_changed and original_timeout is not None:
                    try:
                        self.set_timeout(original_timeout)
                    except:
                        pass  # Best effort restoration
                raise  # Re-raise to be caught by outer handler

        except Exception as e:
            latency_ms = (time_module.time() - start_time) * 1000

            # Extract meaningful error message
            error_message = str(e)
            if not error_message:
                error_message = f"{type(e).__name__} occurred during health check"

            return {
                "status": False,
                "provider": provider_name,
                "models": None,
                "model_count": 0,
                "error": error_message,
                "latency_ms": round(latency_ms, 2)
            }

    def _needs_tag_rewriting(self, tool_call_tags) -> bool:
        """Check if tag rewriting is needed (tags are non-standard)"""
        try:
            from ..tools.tag_rewriter import ToolCallTags

            if isinstance(tool_call_tags, str):
                # String format - handle comma-separated format
                if ',' in tool_call_tags:
                    # Comma-separated format like '<function_call>,</function_call>'
                    parts = tool_call_tags.split(',')
                    if len(parts) == 2:
                        opening_tag = parts[0].strip()
                        closing_tag = parts[1].strip()
                        if opening_tag == "<function_call>" and closing_tag == "</function_call>":
                            return False
                else:
                    # Single tag format
                    if tool_call_tags in ["<function_call>", "</function_call>"]:
                        return False
            elif isinstance(tool_call_tags, ToolCallTags):
                # ToolCallTags object - check if it contains standard tags
                if (hasattr(tool_call_tags, 'start_tag') and hasattr(tool_call_tags, 'end_tag')):
                    # Only standard if exactly matches the standard format
                    if (tool_call_tags.start_tag == "<function_call>" and tool_call_tags.end_tag == "</function_call>"):
                        return False

            # Any other format or non-standard tags need rewriting
            return True

        except Exception:
            # If we can't determine, err on the side of applying rewriting
            return True

    def _apply_non_streaming_tag_rewriting(self, response: GenerateResponse, tool_call_tags: Optional[str] = None) -> GenerateResponse:
        """Apply tag rewriting to non-streaming response content."""
        try:
            from .streaming import UnifiedStreamProcessor

            # Create a temporary processor for tag rewriting
            processor = UnifiedStreamProcessor(
                model_name=self.model,
                execute_tools=False,  # No execution, just rewriting
                tool_call_tags=tool_call_tags,
                default_target_format="qwen3"  # Always rewrite to qwen3 format
            )

            # Apply tag rewriting to the content
            if processor.tag_rewriter and response.content:
                rewritten_content = processor._apply_tag_rewriting_direct(response.content)

                # Return new response with rewritten content
                return GenerateResponse(
                    content=rewritten_content,
                    model=response.model,
                    finish_reason=response.finish_reason,
                    raw_response=response.raw_response,
                    usage=response.usage,
                    tool_calls=response.tool_calls,
                    metadata=response.metadata,
                    gen_time=response.gen_time,
                )

        except Exception as e:
            self.logger.debug(f"Non-streaming tag rewriting failed: {e}")

        # Return original response if rewriting fails
        return response

    def _normalize_tool_calls_passthrough(
        self,
        *,
        response: GenerateResponse,
        tools: List[Dict[str, Any]],
        tool_call_tags: Optional[str] = None,
    ) -> GenerateResponse:
        """Populate `response.tool_calls` (and usually clean `response.content`) in passthrough mode.

        Contract:
        - AbstractCore always returns structured `tool_calls` when tools are provided and the model emits tool syntax,
          even for prompted tool calling (tool calls embedded in `content`).
        - By default (`tool_call_tags is None`), tool-call markup is stripped from `content` for clean UX/history.
        - When `tool_call_tags` is set, we preserve `content` (for clients that parse tags) but still populate
          structured `tool_calls`.
        """

        # Only normalize when tools were actually provided.
        if not tools:
            return response

        allowed_names = self._get_allowed_tool_names(tools)

        # 1) If provider already returned tool_calls (native tools), normalize shape + args.
        normalized_existing = self._normalize_tool_calls_payload(
            response.tool_calls,
            allowed_tool_names=allowed_names,
        )
        if normalized_existing:
            response.tool_calls = normalized_existing

            # Clean any echoed tool syntax from content unless the caller explicitly requested tag passthrough.
            if self._should_clean_tool_call_markup(tool_call_tags) and isinstance(response.content, str) and response.content.strip():
                cleaned = self._clean_content_using_tool_calls(response.content, normalized_existing)
                response.content = cleaned

            return response

        # 2) Prompted tools: parse tool calls embedded in content.
        content = response.content
        if not isinstance(content, str) or not content.strip():
            return response

        tool_handler = getattr(self, "tool_handler", None)
        if tool_handler is None:
            return response

        try:
            parsed = tool_handler.parse_response(content, mode="prompted")
        except Exception:
            return response

        parsed_calls = getattr(parsed, "tool_calls", None)
        if not isinstance(parsed_calls, list) or not parsed_calls:
            return response

        normalized_parsed = self._normalize_tool_calls_payload(
            parsed_calls,
            allowed_tool_names=allowed_names,
        )
        if normalized_parsed:
            response.tool_calls = normalized_parsed

        # Always use the cleaned content from AbstractCore parsing when we are not explicitly preserving tags.
        if self._should_clean_tool_call_markup(tool_call_tags):
            cleaned_content = getattr(parsed, "content", None)
            if isinstance(cleaned_content, str):
                response.content = cleaned_content

        return response

    def _should_clean_tool_call_markup(self, tool_call_tags: Optional[str]) -> bool:
        """Return True when we should strip tool-call markup from assistant content."""
        if tool_call_tags is None:
            return True
        # OpenAI/Codex formats carry tool calls in structured fields, not in content.
        value = str(tool_call_tags).strip().lower()
        return value in {"openai", "codex"}

    def _get_allowed_tool_names(self, tools: List[Dict[str, Any]]) -> set[str]:
        """Extract allowed tool names from provider-normalized tool definitions."""
        names: set[str] = set()
        for tool in tools or []:
            if not isinstance(tool, dict):
                continue
            name = tool.get("name")
            if isinstance(name, str) and name.strip():
                names.add(name.strip())
                continue
            func = tool.get("function") if isinstance(tool.get("function"), dict) else None
            fname = func.get("name") if isinstance(func, dict) else None
            if isinstance(fname, str) and fname.strip():
                names.add(fname.strip())
        return names

    def _normalize_tool_calls_payload(
        self,
        tool_calls: Any,
        *,
        allowed_tool_names: Optional[set[str]] = None,
    ) -> Optional[List[Dict[str, Any]]]:
        """Normalize tool call shapes into a canonical dict form.

        Canonical shape:
            {"name": str, "arguments": dict, "call_id": Optional[str]}
        """
        if tool_calls is None or not isinstance(tool_calls, list):
            return None

        def _unwrap_arguments(arguments: Any, *, expected_tool_name: Optional[str]) -> Any:
            """Unwrap common wrapper payloads to get tool kwargs.

            Some providers/models emit nested wrappers like:
              {"name":"tool","arguments":{...},"call_id": "..."}
            inside the tool call `arguments` field (or even multiple times).

            We unwrap when the object looks like a wrapper (only wrapper keys) OR when
            it includes wrapper metadata fields (e.g. "name"/"call_id") and an inner
            "arguments" dict. When wrapper fields and tool kwargs are partially mixed,
            we merge the outer kwargs into the inner dict (inner takes precedence).
            """
            if not isinstance(arguments, dict):
                return arguments

            wrapper_keys = {"name", "arguments", "call_id", "id"}
            current = arguments
            for _ in range(4):
                if not isinstance(current, dict):
                    break
                keys = set(current.keys())
                if "arguments" not in current:
                    break
                inner = current.get("arguments")
                if isinstance(inner, dict) or isinstance(inner, str):
                    inner_dict: Any = inner
                    if isinstance(inner, str):
                        parsed = loads_dict_like(inner)
                        inner_dict = parsed if isinstance(parsed, dict) else None
                    if not isinstance(inner_dict, dict):
                        break

                    name_matches = False
                    raw_name = current.get("name")
                    if isinstance(raw_name, str) and expected_tool_name and raw_name.strip() == expected_tool_name:
                        name_matches = True

                    wrapperish = keys.issubset(wrapper_keys) or name_matches or bool(keys & {"call_id", "id"})
                    if not wrapperish:
                        break

                    # Merge any outer kwargs that were accidentally placed alongside wrapper fields.
                    extras = {k: v for k, v in current.items() if k not in wrapper_keys}
                    if extras:
                        merged = dict(inner_dict)
                        for k, v in extras.items():
                            merged.setdefault(k, v)
                        current = merged
                    else:
                        current = inner_dict
                    continue
                break

            return current

        def _map_wrapped_name_to_allowed(raw: str, allowed: set[str]) -> Optional[str]:
            """Best-effort mapping when a provider returns a wrapped tool name.

            Some OpenAI-compatible servers/models occasionally return tool names wrapped in
            extra tokens/text (e.g. "{function-name: write_file}"). If we can confidently
            detect an allowed tool name as a standalone token within the raw string, map it
            back to the exact allowed name so tool execution can proceed.
            """
            s = str(raw or "").strip()
            if not s:
                return None
            if s in allowed:
                return s

            try:
                import re

                # Prefer exact token-boundary matches (tool names are usually snake_case).
                candidates: List[str] = []
                for name in allowed:
                    if not isinstance(name, str) or not name:
                        continue
                    pat = r"(^|[^\w])" + re.escape(name) + r"([^\w]|$)"
                    if re.search(pat, s):
                        candidates.append(name)
                if candidates:
                    # Prefer the most specific (longest) match deterministically.
                    return max(candidates, key=lambda n: (len(n), n))
            except Exception:
                return None

            return None

        normalized: List[Dict[str, Any]] = []

        for tc in tool_calls:
            name: Optional[str] = None
            arguments: Any = None
            call_id: Any = None

            if isinstance(tc, dict):
                call_id = tc.get("call_id", None)
                if call_id is None:
                    call_id = tc.get("id", None)

                raw_name = tc.get("name")
                raw_args = tc.get("arguments")

                func = tc.get("function") if isinstance(tc.get("function"), dict) else None
                if func and (not isinstance(raw_name, str) or not raw_name.strip()):
                    raw_name = func.get("name")
                if func and raw_args is None:
                    raw_args = func.get("arguments")

                if isinstance(raw_name, str) and raw_name.strip():
                    name = raw_name.strip()
                arguments = raw_args if raw_args is not None else {}
            else:
                raw_name = getattr(tc, "name", None)
                raw_args = getattr(tc, "arguments", None)
                call_id = getattr(tc, "call_id", None)
                if isinstance(raw_name, str) and raw_name.strip():
                    name = raw_name.strip()
                arguments = raw_args if raw_args is not None else {}

            if not isinstance(name, str) or not name:
                continue
            if isinstance(allowed_tool_names, set) and allowed_tool_names and name not in allowed_tool_names:
                mapped = _map_wrapped_name_to_allowed(name, allowed_tool_names)
                if not isinstance(mapped, str) or not mapped:
                    continue
                name = mapped

            if isinstance(arguments, str):
                parsed = loads_dict_like(arguments)
                arguments = parsed if isinstance(parsed, dict) else {}

            # Recover tool kwargs from nested wrapper payloads when present.
            if isinstance(arguments, dict) and call_id is None:
                wrapper_id = arguments.get("call_id") or arguments.get("id")
                if isinstance(wrapper_id, str) and wrapper_id.strip():
                    call_id = wrapper_id.strip()
            arguments = _unwrap_arguments(arguments, expected_tool_name=name)
            if not isinstance(arguments, dict):
                arguments = {}

            try:
                from ..tools.arg_canonicalizer import canonicalize_tool_arguments

                arguments = canonicalize_tool_arguments(name, arguments)
            except Exception:
                pass

            normalized.append(
                {
                    "name": name,
                    "arguments": arguments,
                    "call_id": str(call_id) if call_id is not None else None,
                }
            )

        if not normalized:
            return None

        # Defense-in-depth: remove accidental duplicates introduced by overlapping parsing paths.
        unique: List[Dict[str, Any]] = []
        seen: set[tuple[str, str]] = set()
        for tc in normalized:
            try:
                args_key = json.dumps(tc.get("arguments", {}), sort_keys=True, ensure_ascii=False)
            except Exception:
                args_key = str(tc.get("arguments", {}))
            key = (str(tc.get("name") or ""), args_key)
            if key in seen:
                continue
            seen.add(key)
            unique.append(tc)

        return unique or None

    def _clean_content_using_tool_calls(self, content: str, tool_calls: List[Dict[str, Any]]) -> str:
        """Strip tool-call markup from assistant content using known tool calls."""
        try:
            from ..tools.core import ToolCall as CoreToolCall
            from ..tools.parser import clean_tool_syntax

            core_calls: List[CoreToolCall] = []
            for tc in tool_calls or []:
                if not isinstance(tc, dict):
                    continue
                name = tc.get("name")
                if not isinstance(name, str) or not name.strip():
                    continue
                args = tc.get("arguments")
                args_dict = dict(args) if isinstance(args, dict) else {}
                core_calls.append(CoreToolCall(name=name.strip(), arguments=args_dict, call_id=tc.get("call_id")))

            if not core_calls:
                return content
            return clean_tool_syntax(content, core_calls)
        except Exception:
            return content

    def _handle_tools_with_structured_output(self,
                                           prompt: str,
                                           messages: Optional[List[Dict[str, str]]] = None,
                                           system_prompt: Optional[str] = None,
                                           tools: Optional[List] = None,
                                           response_model: Optional[Type[BaseModel]] = None,
                                           retry_strategy=None,
                                           tool_call_tags: Optional[str] = None,
                                           execute_tools: Optional[bool] = None,
                                           stream: bool = False,
                                           **kwargs) -> BaseModel:
        """
        Handle the hybrid case: tools + structured output.

        Strategy: Sequential execution
        1. First, generate response with tools (may include tool calls)
        2. If tool calls are generated, execute them
        3. Then generate structured output using tool results as context

        Args:
            prompt: Input prompt
            messages: Optional message history
            system_prompt: Optional system prompt
            tools: List of available tools
            response_model: Pydantic model for structured output
            retry_strategy: Optional retry strategy for structured output
            tool_call_tags: Optional tool call tag format
            execute_tools: Whether to execute tools automatically
            stream: Whether to use streaming (not supported for hybrid mode)
            **kwargs: Additional parameters

        Returns:
            Validated instance of response_model

        Raises:
            ValueError: If streaming is requested (not supported for hybrid mode)
        """
        if stream:
            raise ValueError(
                "Streaming is not supported when combining tools with structured output. "
                "Please use either stream=True OR response_model, but not both."
            )

        # Step 1: Generate response with tools (normal tool execution flow)
        self.logger.info("Hybrid mode: Executing tools first, then structured output",
                        model=self.model,
                        response_model=response_model.__name__,
                        num_tools=len(tools) if tools else 0)

        # Force tool execution for hybrid mode
        should_execute_tools = execute_tools if execute_tools is not None else True

        # Generate response with tools using the normal flow (without response_model)
        tool_response = self.generate_with_telemetry(
            prompt=prompt,
            messages=messages,
            system_prompt=system_prompt,
            tools=tools,
            stream=False,  # Never stream in hybrid mode
            response_model=None,  # No structured output in first pass
            tool_call_tags=tool_call_tags,
            execute_tools=should_execute_tools,
            **kwargs
        )

        # Step 2: Generate structured output using tool results as context
        # Create enhanced prompt with tool execution context
        if hasattr(tool_response, 'content') and tool_response.content:
            enhanced_prompt = f"""{prompt}

Based on the following tool execution results:
{tool_response.content}

Please provide a structured response."""
        else:
            enhanced_prompt = prompt

        self.logger.info("Hybrid mode: Generating structured output with tool context",
                        model=self.model,
                        response_model=response_model.__name__,
                        has_tool_context=bool(hasattr(tool_response, 'content') and tool_response.content))

        # Generate structured output using the enhanced prompt
        from ..structured import StructuredOutputHandler
        handler = StructuredOutputHandler(retry_strategy=retry_strategy)

        structured_result = handler.generate_structured(
            provider=self,
            prompt=enhanced_prompt,
            response_model=response_model,
            messages=messages,
            system_prompt=system_prompt,
            tools=None,  # No tools in structured output pass
            stream=False,
            **kwargs
        )

        self.logger.info("Hybrid mode: Successfully completed tools + structured output",
                        model=self.model,
                        response_model=response_model.__name__,
                        success=True)

        return structured_result

    def generate(self,
                prompt: str = "",
                messages: Optional[List[Dict[str, str]]] = None,
                system_prompt: Optional[str] = None,
                tools: Optional[List[Dict[str, Any]]] = None,
                media: Optional[List[Union[str, Dict[str, Any], 'MediaContent']]] = None,
                stream: bool = False,
                **kwargs) -> Union[GenerateResponse, Iterator[GenerateResponse], BaseModel]:
        """
        Generate response from the LLM.

        This method implements the AbstractCoreInterface and delegates to generate_with_telemetry.

        Args:
            prompt: The input prompt
            messages: Optional conversation history
            system_prompt: Optional system prompt
            tools: Optional list of available tools
            stream: Whether to stream the response
            **kwargs: Additional provider-specific parameters (including response_model)

        Returns:
            GenerateResponse, iterator of GenerateResponse for streaming, or BaseModel for structured output
        """
        return self.generate_with_telemetry(
            prompt=prompt,
            messages=messages,
            system_prompt=system_prompt,
            tools=tools,
            media=media,
            stream=stream,
            **kwargs
        )

    async def agenerate(self,
                       prompt: str = "",
                       messages: Optional[List[Dict]] = None,
                       system_prompt: Optional[str] = None,
                       tools: Optional[List] = None,
                       media: Optional[List] = None,
                       stream: bool = False,
                       **kwargs) -> Union[GenerateResponse, AsyncIterator[GenerateResponse], BaseModel]:
        """
        Async generation - works with all providers.

        Calls _agenerate_internal() which can be overridden for native async.
        Default implementation uses asyncio.to_thread() fallback.

        Args:
            prompt: Text prompt
            messages: Conversation history
            system_prompt: System instructions
            tools: Available tools
            media: Media attachments
            stream: Enable streaming
            **kwargs: Additional generation parameters (including response_model)

        Returns:
            GenerateResponse, AsyncIterator[GenerateResponse] for streaming, or BaseModel for structured output
        """
        system_prompt = self._normalize_system_prompt_alias(system_prompt, kwargs, stacklevel=3)

        output_request = kwargs.get("output", None)
        is_acore_output = self._is_acore_output_request(output_request)
        if (prompt is None or str(prompt) == "") and "text" in kwargs:
            text_value = kwargs.pop("text")
            prompt = "" if text_value is None else str(text_value)

        if is_acore_output:
            try:
                maybe_specs = self._normalize_output_specs(output_request)
            except Exception:
                maybe_specs = []
            if (
                len(maybe_specs) == 1
                and maybe_specs[0].get("modality") == "text"
                and maybe_specs[0].get("task") not in {"transcription"}
                and str(prompt or "").strip()
            ):
                kwargs.pop("output", None)
                is_acore_output = False

        if is_acore_output:
            output_request = kwargs.pop("output")
            output_request, progress_injected = self._output_request_with_progress_callbacks(
                output_request,
                {
                    "on_progress": kwargs.get("on_progress"),
                    "progress_event_callback": kwargs.get("progress_event_callback"),
                    "progress_callback": kwargs.get("progress_callback"),
                },
            )
            if progress_injected:
                kwargs.pop("on_progress", None)
                kwargs.pop("progress_event_callback", None)
                kwargs.pop("progress_callback", None)
            response_model = kwargs.pop("response_model", None)
            artifact_store = kwargs.pop("artifact_store", None)
            partial = bool(kwargs.pop("partial", False))
            return await asyncio.to_thread(
                self._generate_multimodal_response,
                prompt=str(prompt or ""),
                output=output_request,
                media=media,
                stream=stream,
                response_model=response_model,
                artifact_store=artifact_store,
                partial=partial,
                text_generation_kwargs={
                    "messages": messages,
                    "system_prompt": system_prompt,
                    "tools": tools,
                    "media": media,
                    **kwargs,
                },
            )

        self._apply_default_prompt_cache_key(kwargs)
        self._apply_prompt_cache_binding_request(kwargs)
        messages, system_prompt = self._normalize_developer_messages(messages, system_prompt)
        response = await self._agenerate_internal(
            prompt, messages, system_prompt, tools, media, stream, **kwargs
        )

        # Capture interaction trace if enabled (match sync generate_with_telemetry behavior)
        # Only for non-streaming responses that are GenerateResponse objects
        if not stream and self.enable_tracing and response and isinstance(response, GenerateResponse):
            trace_id = self._capture_trace(
                prompt=prompt,
                messages=messages,
                system_prompt=system_prompt,
                tools=tools,
                response=response,
                kwargs=kwargs
            )
            # Attach trace_id to response metadata
            if not response.metadata:
                response.metadata = {}
            response.metadata['trace_id'] = trace_id

        return response

    async def _agenerate_internal(self,
                                   prompt: str,
                                   messages: Optional[List[Dict]],
                                   system_prompt: Optional[str],
                                   tools: Optional[List],
                                   media: Optional[List],
                                   stream: bool,
                                   **kwargs) -> Union[GenerateResponse, AsyncIterator[GenerateResponse], BaseModel]:
        """
        Internal async generation method.

        Default implementation: Uses asyncio.to_thread() to run sync generate().
        Providers override this for native async (3-10x faster for batch operations).

        Args:
            prompt: Text prompt
            messages: Conversation history
            system_prompt: System instructions
            tools: Available tools
            media: Media attachments
            stream: Enable streaming
            **kwargs: Additional generation parameters

        Returns:
            GenerateResponse, AsyncIterator[GenerateResponse] for streaming, or BaseModel for structured output
        """
        if stream:
            # Return async iterator for streaming
            return self._async_stream_generate(
                prompt, messages, system_prompt, tools, media, **kwargs
            )
        else:
            # Run sync generate in thread pool (fallback)
            return await asyncio.to_thread(
                self.generate,
                prompt=prompt,
                messages=messages,
                system_prompt=system_prompt,
                tools=tools,
                media=media,
                stream=stream,
                **kwargs,
            )

    async def _async_stream_generate(self,
                                     prompt: str,
                                     messages: Optional[List[Dict]],
                                     system_prompt: Optional[str],
                                     tools: Optional[List],
                                     media: Optional[List],
                                     **kwargs) -> AsyncIterator[GenerateResponse]:
        """
        Async streaming generator.

        Wraps sync streaming in async iterator, yielding control to event loop.
        """
        # Get sync generator in thread pool
        def get_sync_stream():
            return self.generate(
                prompt, messages, system_prompt, tools,
                stream=True, **kwargs
            )

        sync_gen = await asyncio.to_thread(get_sync_stream)

        # Yield chunks asynchronously
        for chunk in sync_gen:
            yield chunk
            await asyncio.sleep(0)  # Yield control to event loop
