"""AWS Bedrock Converse API adapter (boto3, optional dependency).

Works without API keys via the AWS credential chain, cross-region inference profiles, guardrails and
control-plane model discovery. OpenAI-format messages/tools are converted to Converse on the way in
and responses normalized back to OpenAI-shaped objects.
"""

from pm import install_hint
import base64
import importlib
import json
import logging
import os
import re
import time
import traceback
from contextlib import suppress
from types import SimpleNamespace
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import urlparse

import httpx

from agent.errors import EmptyStreamError

logger = logging.getLogger(__name__)

_bedrock_runtime_client_cache: dict[str, Any] = {}
_bedrock_control_client_cache: dict[str, Any] = {}
# Routed multiplex profiles: one client per (profile home, region). boto3 freezes the credential
# chain into the client at construction, so a region-only slot would sign profile B's calls with A's keys.
_bedrock_clients_by_home: dict[tuple[str, str, str], Any] = {}

# botocore session kwarg <- profile .env variable (the explicit sources of the default chain).
_AWS_SCOPED_CREDENTIAL_VARS: tuple[tuple[str, str], ...] = (
    ("aws_access_key_id", "AWS_ACCESS_KEY_ID"), ("aws_secret_access_key", "AWS_SECRET_ACCESS_KEY"),
    ("aws_session_token", "AWS_SESSION_TOKEN"), ("profile_name", "AWS_PROFILE"),
)


def scoped_aws_session_kwargs() -> dict[str, str]:
    """``boto3.session.Session`` kwargs from the routed profile's secret scope, ``{}`` when unscoped.

    Under a HERMES_HOME override the process env holds the LAUNCH profile's ``AWS_*`` (or nothing), so
    every Bedrock client for a served profile must be built from that profile's own ``.env`` values.
    Under multiplexing a profile that sets none of its own must NOT get ``{}`` — ``boto3.Session()``
    would then resolve the ambient default chain (process env, ~/.aws, instance metadata), i.e. the
    launch context's identity, exactly the borrow the Entra adapter refuses. ``AWS_PROFILE`` counts as
    an explicit per-profile choice (it names an entry in the shared AWS config, like the Entra
    ``AZURE_CLIENT_ID``-only managed-identity opt-in)."""
    from hermes_constants import get_hermes_home_override
    if get_hermes_home_override() is None:
        return {}
    from agent.secret_scope import current_secret_scope, is_multiplex_active
    scope = current_secret_scope() or {}
    kwargs = {kw: scope[var].strip() for kw, var in _AWS_SCOPED_CREDENTIAL_VARS
              if (scope.get(var) or "").strip()}
    # Under multiplexing a partial set is not enough: boto3 fills whatever is missing from the
    # ambient chain (process env, ~/.aws, instance metadata), i.e. the launch context's identity —
    # the same borrow the Entra adapter refuses. Require a COMPLETE credential: the key pair, or
    # AWS_PROFILE naming an entry in the shared AWS config (an explicit per-profile choice, like
    # the Entra AZURE_CLIENT_ID-only managed-identity opt-in).
    complete = ("aws_access_key_id" in kwargs and "aws_secret_access_key" in kwargs) \
        or "profile_name" in kwargs
    if not complete and is_multiplex_active():
        raise RuntimeError(
            "Bedrock auth is refused for this profile: it sets no complete AWS credential of its "
            "own, and under multiplexed profiles the ambient default chain (process env, ~/.aws, "
            "instance metadata) would sign this profile's calls with the LAUNCH context's identity. "
            "Set AWS_ACCESS_KEY_ID + AWS_SECRET_ACCESS_KEY in this profile's own config, or "
            "AWS_PROFILE to name a shared AWS config profile."
        )
    return kwargs

# Geo/global cross-Region inference-profile prefixes Bedrock prepends to a model ID.
_BEDROCK_PROFILE_PREFIXES: tuple[str, ...] = ("global", "us", "eu", "apac", "ap", "au", "jp", "ca", "sa", "me", "af")
# The GPT-5.6 Terra/Luna cards also document India geo profiles (``in.openai.gpt-5.6-terra``).
_BEDROCK_OPENAI_PROFILE_PREFIXES: tuple[str, ...] = _BEDROCK_PROFILE_PREFIXES + ("in",)

# Bedrock-hosted OpenAI GPT models (bare in-Region IDs). The bare ID is served by the Bedrock Mantle
# OpenAI-compatible endpoint; its geo/global profile form is served by bedrock-runtime, where these
# cards list Converse as supported. Narrow allowlist so GPT-OSS models stay on the native path.
BEDROCK_OPENAI_RESPONSES_MODEL_IDS: tuple[str, ...] = (
    "openai.gpt-6-astra", "openai.gpt-6.1-sol", "openai.gpt-6-sol", "openai.gpt-6-luna",
    "openai.gpt-5.5", "openai.gpt-5.6-sol", "openai.gpt-5.6-terra", "openai.gpt-5.6-luna",
)
_BEDROCK_OPENAI_MODEL_IDS_LOWER = frozenset(m.lower() for m in BEDROCK_OPENAI_RESPONSES_MODEL_IDS)
_BEDROCK_OPENAI_HOST_RE = re.compile(r"^bedrock-mantle\.([a-z0-9-]+)\.api\.aws$", re.IGNORECASE)
# Bedrock-hosted xAI Grok (any regional inference-profile prefix) rejects temperature/topP in Converse
# with a hard 400 ("This model doesn't support the temperature field"); reasoning-first, same
# restriction as Claude Opus 4.6+ but _forbids_sampling_params is Claude-only, so it needs its own gate.
_BEDROCK_XAI_GROK_NO_SAMPLING_RE = re.compile(r"^(?:[a-z]+\.)?xai\.grok", re.IGNORECASE)
_MIN_BOTO3_VERSION = (1, 34, 59)


def _require_boto3():
    """Import boto3; converse_stream() needs >= 1.34.59 (a system boto3 can shadow the venv pin)."""
    install_error = None
    try:
        # boto3 left [all] (PRs #24220, #24515); PM installs the [bedrock] extra on first use. This
        # runs at the first client build, never at import: an import-time sync would rebuild the
        # dependency environment of whatever process happens to import this module.
        try:
            from pm import ensure_import
            ensure_import("bedrock")
        except Exception as exc:  # the import below decides; exc explains a miss
            logger.warning("boto3 lazy install did not complete: %s", exc)
            install_error = exc
        import boto3
    except ImportError:
        # A completed install that needs a restart must not be reported as "install it".
        reason = f": {install_error}" if install_error else f". Run: {install_hint('bedrock')}"
        raise ImportError(f"The 'boto3' package is required for the AWS Bedrock provider{reason}") from install_error
    try:
        version = tuple(int(x) for x in boto3.__version__.split(".")[:3])
    except (AttributeError, ValueError):
        return boto3  # can't parse — don't block on version check
    if version < _MIN_BOTO3_VERSION:
        raise RuntimeError(
            f"boto3 {boto3.__version__} does not support converse_stream "
            f"(minimum 1.34.59 required). Run: hermes pm repair"
        )
    return boto3


def _cached_client(cache: dict[str, Any], service: str, region: str):
    """Get or create a per-region boto3 client. Unscoped: the default credential chain, one client per
    region. Routed profile: one client per (home, service, region), built from that profile's scoped
    ``AWS_*`` (falling back to the default chain only for what the profile does not set)."""
    from hermes_constants import get_hermes_home_override, hermes_home_key
    if get_hermes_home_override() is None:
        if region not in cache:
            cache[region] = _require_boto3().client(service, region_name=region)
        return cache[region]
    key = (hermes_home_key(), service, region)
    client = _bedrock_clients_by_home.get(key)
    if client is None:
        # Scope check first: a cred-less multiplex profile must hit the ambient-chain
        # refusal, not a boto3 ImportError on hosts that lack the package.
        kwargs = scoped_aws_session_kwargs()
        client = _require_boto3().Session(**kwargs).client(service, region_name=region)
        _bedrock_clients_by_home[key] = client
    return client


def _get_bedrock_runtime_client(region: str):
    return _cached_client(_bedrock_runtime_client_cache, "bedrock-runtime", region)


def _get_bedrock_control_client(region: str):
    return _cached_client(_bedrock_control_client_cache, "bedrock", region)


def reset_client_cache():
    """Clear cached boto3 clients. Used in tests and profile switches."""
    _bedrock_runtime_client_cache.clear()
    _bedrock_control_client_cache.clear()
    _bedrock_clients_by_home.clear()
    _inference_profile_model_cache.clear()
    _DEFAULT_WINDOW_WARNED.clear()


def invalidate_runtime_client(region: str) -> bool:
    """Evict one region's cached ``bedrock-runtime`` client (stale HTTP pool); True if evicted."""
    from hermes_constants import get_hermes_home_override, hermes_home_key
    if get_hermes_home_override() is not None:
        return _bedrock_clients_by_home.pop((hermes_home_key(), "bedrock-runtime", region), None) is not None
    return _bedrock_runtime_client_cache.pop(region, None) is not None



# --- Bedrock Mantle / OpenAI Responses support ---

def parse_bedrock_openai_model_id(model_id: str) -> Optional[tuple[str, str]]:
    """``(bare_id, profile_prefix)`` for a Bedrock-hosted OpenAI model, else None. ``profile_prefix``
    is ``""`` for the bare in-Region ID and e.g. ``"us"`` / ``"global"`` for a cross-Region profile."""
    normalized = str(model_id or "").strip().lower()
    prefix, sep, rest = normalized.partition(".")
    if sep and prefix in _BEDROCK_OPENAI_PROFILE_PREFIXES and rest.startswith("openai."):
        normalized = rest
    else:
        prefix = ""
    return (normalized, prefix) if normalized in _BEDROCK_OPENAI_MODEL_IDS_LOWER else None


def is_openai_bedrock_model(model_id: str) -> bool:
    """True for Bedrock-hosted OpenAI GPT models in any endpoint form (GPT-OSS excluded)."""
    return parse_bedrock_openai_model_id(model_id) is not None


def bedrock_openai_uses_mantle(model_id: str) -> bool:
    """True when the model is served by Mantle's OpenAI Responses endpoint: the bare in-Region ID.
    A geo/global profile ID is a bedrock-runtime ID (Mantle rejects it) and rides Converse."""
    parsed = parse_bedrock_openai_model_id(model_id)
    return parsed is not None and not parsed[1]


def merge_bedrock_openai_model_ids(model_ids: list[str]) -> list[str]:
    """Append Mantle-only OpenAI models, which control-plane discovery never lists."""
    merged = list(model_ids or [])
    seen = {str(m).lower() for m in merged}
    return merged + [m for m in BEDROCK_OPENAI_RESPONSES_MODEL_IDS if m.lower() not in seen]


def bedrock_openai_base_url(region: str) -> str:
    """Return Bedrock Mantle's OpenAI-compatible base URL for *region*."""
    resolved = (region or "").strip() or resolve_bedrock_runtime_region()
    return f"https://bedrock-mantle.{resolved}.api.aws/openai/v1"


def _mantle_url_parts(base_url: str) -> tuple[Optional[str], str]:
    """(region or None if not a Mantle host, normalized path) for a base URL."""
    parsed = urlparse(str(base_url or ""))
    match = _BEDROCK_OPENAI_HOST_RE.match(parsed.hostname or "")
    return (match.group(1) if match else None), (parsed.path or "").rstrip("/").lower()


def bedrock_openai_region_from_base_url(base_url: str) -> Optional[str]:
    """Extract the AWS region from a Bedrock Mantle OpenAI base URL."""
    return _mantle_url_parts(base_url)[0]


def is_bedrock_openai_base_url(base_url: str) -> bool:
    """True for Bedrock Mantle endpoints (bare host or /openai[/v1] path)."""
    region, path = _mantle_url_parts(base_url)
    return region is not None and path in {"", "/openai", "/openai/v1"}


def resolve_bedrock_bearer_token(env: Optional[dict[str, str]] = None) -> str:
    """Return AWS_BEARER_TOKEN_BEDROCK when Bedrock API-key auth is configured.

    Under a HERMES_HOME override the read goes through the profile secret scope so a
    served profile never inherits the launch profile's bearer from the process env."""
    if env is not None:
        return (env.get("AWS_BEARER_TOKEN_BEDROCK", "") or "").strip()
    from hermes_constants import get_hermes_home_override
    if get_hermes_home_override() is not None:
        from agent.secret_scope import get_secret
        return (get_secret("AWS_BEARER_TOKEN_BEDROCK", "") or "").strip()
    return (os.environ.get("AWS_BEARER_TOKEN_BEDROCK", "") or "").strip()


class BedrockOpenAISigV4Auth(httpx.Auth):
    """httpx auth hook that SigV4-signs Bedrock Mantle OpenAI requests."""

    requires_request_body = True

    def __init__(self, region: str, service: str = "bedrock"):
        self.region = (region or "").strip() or resolve_bedrock_runtime_region()
        self.service = service

    def auth_flow(self, request):  # pragma: no cover - exercised by live call
        from botocore.auth import SigV4Auth
        from botocore.awsrequest import AWSRequest
        kwargs = scoped_aws_session_kwargs()
        credentials = _require_boto3().Session(**kwargs).get_credentials()
        if credentials is None:
            raise RuntimeError(
                "No AWS credentials available for Bedrock OpenAI Responses. "
                "Configure AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY, AWS_PROFILE, SSO, or an instance/task role."
            )
        # SigV4 must own Authorization: drop the SDK's placeholder bearer header.
        headers = {str(k): str(v) for k, v in request.headers.items()
                   if str(k).lower() not in {"authorization", "x-amz-date", "x-amz-security-token"}}
        aws_request = AWSRequest(method=request.method, url=str(request.url), data=request.content or b"", headers=headers)
        SigV4Auth(credentials.get_frozen_credentials(), self.service, self.region).add_auth(aws_request)
        request.headers.update(dict(aws_request.headers.items()))
        yield request


def build_bedrock_openai_http_client(region: str, *, timeout: Optional[float] = None):
    """Build an httpx client that SigV4-signs Bedrock OpenAI requests."""
    kwargs: dict[str, Any] = {"auth": BedrockOpenAISigV4Auth(region)}
    if isinstance(timeout, (int, float)) and not isinstance(timeout, bool) and timeout > 0:
        kwargs["timeout"] = timeout
    return httpx.Client(**kwargs)


def configure_bedrock_openai_client_kwargs(client_kwargs: dict[str, Any], *, timeout: Optional[float] = None) -> dict[str, Any]:
    """Install SigV4 auth on OpenAI SDK kwargs for Bedrock Mantle; a real API key keeps the SDK's
    bearer auth, the ``aws-sdk``/``no-key-required`` placeholders mean IAM chain auth."""
    base_url = str(client_kwargs.get("base_url") or "")
    api_key = client_kwargs.get("api_key")
    if not is_bedrock_openai_base_url(base_url) or (
        isinstance(api_key, str) and api_key.strip() and api_key not in {"aws-sdk", "no-key-required"}
    ):
        return client_kwargs
    region = bedrock_openai_region_from_base_url(base_url) or resolve_bedrock_runtime_region()
    client_kwargs["api_key"] = "aws-sdk"
    client_kwargs["http_client"] = build_bedrock_openai_http_client(region, timeout=timeout)
    return client_kwargs


# --- Stale-connection detection ---
# A pooled connection killed under boto3 (NAT timeout, VPN flap, RST) surfaces as a botocore/urllib3
# transport error or a bare AssertionError from urllib3's pool checks; retrying the same client
# reproduces it, so the fix is to evict the client.

_STALE_LIB_MODULE_PREFIXES = ("urllib3.", "botocore.", "boto3.")


def _stale_error_types() -> tuple:
    """botocore + urllib3 transport-failure exception classes (best-effort import)."""
    types: list = []
    for module, names in (
        ("botocore.exceptions", ("ConnectionError", "HTTPClientError")),
        ("urllib3.exceptions", ("ProtocolError", "NewConnectionError", "ConnectionError")),
    ):
        # AttributeError too: ``from mod import Name`` raised ImportError for a missing name.
        with suppress(ImportError, AttributeError):  # pragma: no cover — both present with boto3
            types += [getattr(importlib.import_module(module), name) for name in names]
    return tuple(types)


def is_stale_connection_error(exc: BaseException) -> bool:
    """True for botocore/urllib3 transport errors or AssertionErrors raised inside those libs."""
    if isinstance(exc, _stale_error_types()):
        return True
    return isinstance(exc, AssertionError) and any(
        (frame.f_globals.get("__name__", "") or "").startswith(_STALE_LIB_MODULE_PREFIXES)
        for frame, _lineno in traceback.walk_tb(exc.__traceback__)
    )


def is_streaming_access_denied_error(exc: BaseException) -> bool:
    """True when IAM denied ``bedrock:InvokeModelWithResponseStream`` (permanent: callers fall back
    to converse()). Message-based: the AnthropicBedrock SDK wraps the response but keeps the action name."""
    msg = str(exc).lower()
    if "invokemodelwithresponsestream" not in msg:
        return False
    with suppress(ImportError):  # pragma: no cover — botocore always present with boto3
        from botocore.exceptions import ClientError
        if isinstance(exc, ClientError):
            code = (getattr(exc, "response", None) or {}).get("Error", {}).get("Code", "")
            return code in ("AccessDeniedException", "UnauthorizedException")
    return "not authorized" in msg or "accessdenied" in msg


# --- AWS credential detection ---
# Priority order; the first group whose vars are ALL set names the auth source.
_AWS_AUTH_ENV_CHAIN: tuple[tuple[str, ...], ...] = (
    ("AWS_BEARER_TOKEN_BEDROCK",),                    # Bedrock bearer token
    ("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"),   # explicit IAM key pair
    ("AWS_PROFILE",),                                 # named profile (SSO, assume-role)
    ("AWS_CONTAINER_CREDENTIALS_RELATIVE_URI",),      # ECS / CodeBuild
    ("AWS_WEB_IDENTITY_TOKEN_FILE",),                 # EKS IRSA
)


def _boto3_chain_has_credentials(*, instance_metadata: bool = True) -> bool:
    """True if boto3's default chain resolves credentials (IMDS, task role, ...). *instance_metadata*
    False skips the EC2 metadata lookup (169.254.169.254), which off EC2 only costs its timeout."""
    with suppress(Exception):
        import botocore.session
        session = botocore.session.get_session()
        if not instance_metadata:
            session.set_config_variable("ec2_metadata_disabled", "true")  # this session only, not os.environ
        credentials = session.get_credentials()
        resolved = credentials.get_frozen_credentials() if credentials is not None else None
        return bool(resolved and resolved.access_key)
    return False


def resolve_aws_auth_env_var(
    env: Optional[dict[str, str]] = None, *, instance_metadata: bool = True,
) -> Optional[str]:
    """Name of the active AWS auth source: env vars first (no I/O), then ``"iam-role"`` via boto3's chain, else None."""
    env = env if env is not None else os.environ
    for group in _AWS_AUTH_ENV_CHAIN:
        if all(env.get(var, "").strip() for var in group):
            return group[0]
    return "iam-role" if _boto3_chain_has_credentials(instance_metadata=instance_metadata) else None


def has_aws_credentials(env: Optional[dict[str, str]] = None, *, instance_metadata: bool = True) -> bool:
    """True if any AWS credential source (env vars or boto3 chain) is detected.

    This two-tier approach mirrors the pattern from OpenClaw PR #62673: cloud environments (EC2, ECS,
    Lambda) provide credentials via instance metadata, not environment variables. The env-var check is a
    fast path for local development; the boto3 fallback covers all cloud deployments.
    """
    return resolve_aws_auth_env_var(env, instance_metadata=instance_metadata) is not None


def resolve_bedrock_region(env: Optional[dict[str, str]] = None) -> str:
    """AWS_REGION → AWS_DEFAULT_REGION → botocore configured region (~/.aws/config profiles) → us-east-1."""
    env = env if env is not None else os.environ
    explicit = env.get("AWS_REGION", "").strip() or env.get("AWS_DEFAULT_REGION", "").strip()
    if explicit:
        return explicit
    with suppress(Exception):
        import botocore.session
        return botocore.session.get_session().get_config_variable("region") or "us-east-1"
    return "us-east-1"


def resolve_bedrock_runtime_region(config: Optional[dict[str, Any]] = None) -> str:
    """``bedrock.region`` from config.yaml, else :func:`resolve_bedrock_region`. Every non-runtime Bedrock
    endpoint must use this so auxiliary calls never leave the primary runtime's region. *config* skips disk."""
    if config is None:
        with suppress(Exception):
            from hermes_cli.config import load_config_readonly
            config = load_config_readonly()
    cfg_region = str(((config or {}).get("bedrock") or {}).get("region") or "").strip()
    return cfg_region or resolve_bedrock_region()


def bedrock_region_from_runtime_url(base_url: str) -> str:
    """AWS region from a ``bedrock-runtime.<region>.amazonaws.com`` URL (default us-east-1)."""
    m = re.search(r"bedrock-runtime\.([a-z0-9-]+)\.", base_url or "")
    return m.group(1) if m else "us-east-1"


def bedrock_guardrail_config(config: Optional[dict[str, Any]] = None) -> Optional[dict[str, Any]]:
    """Converse ``guardrailConfig`` from ``bedrock.guardrail`` in config.yaml (None when unset)."""
    if config is None:
        config = {}
        with suppress(Exception):
            from hermes_cli.config import load_config_readonly
            config = load_config_readonly()
    gr = ((config or {}).get("bedrock") or {}).get("guardrail") or {}
    if not (gr.get("guardrail_identifier") and gr.get("guardrail_version")):
        return None
    out = {"guardrailIdentifier": gr["guardrail_identifier"], "guardrailVersion": gr["guardrail_version"]}
    for src, dst in (("stream_processing_mode", "streamProcessingMode"), ("trace", "trace")):
        if gr.get(src):
            out[dst] = gr[src]
    return out


def bedrock_guardrail_headers(config: Optional[dict[str, Any]] = None) -> dict[str, str]:
    """InvokeModel/Messages-wire form of the configured guardrail. The AnthropicBedrock SDK speaks
    InvokeModel, which has no ``guardrailConfig`` body field; Bedrock reads the guardrail from these
    headers instead (same enforcement, keeps prompt caching / thinking / 1M context)."""
    gr = bedrock_guardrail_config(config)
    if not gr:
        return {}
    headers = {
        "X-Amzn-Bedrock-GuardrailIdentifier": str(gr["guardrailIdentifier"]),
        "X-Amzn-Bedrock-GuardrailVersion": str(gr["guardrailVersion"]),
    }
    if str(gr.get("trace", "")).lower() in {"enabled", "enabled_full", "true"}:
        headers["X-Amzn-Bedrock-Trace"] = "ENABLED"
    return headers


GUARDRAIL_ACTION_FIELD = "amazon-bedrock-guardrailAction"


def anthropic_response_guardrail_intervened(response: Any) -> bool:
    """True when Bedrock substituted the InvokeModel reply with guardrail messaging. Unlike Converse
    (``stopReason=guardrail_intervened``), InvokeModel keeps ``stop_reason=end_turn`` and signals the
    block only via an unmodelled body field the Anthropic SDK keeps in ``model_extra``."""
    extra = getattr(response, "model_extra", None) or {}
    return str(extra.get(GUARDRAIL_ACTION_FIELD, "")).upper() == "INTERVENED"


def bind_bedrock_runtime(agent, base_url: str, api_mode: str) -> None:
    """Point *agent* at a non-Mantle Bedrock wire: ``bedrock_converse`` (boto3 direct, no SDK client) or
    ``anthropic_messages`` (AnthropicBedrock SDK, SigV4 via the boto3 chain). ``aws-sdk`` is a sentinel,
    never a credential, so the generic Anthropic/OpenAI client builders must not see it. Startup and every
    later rebuild (/model switch, fallback restore, fallback-to-Bedrock) share this so region and guardrail
    state never lag the active endpoint."""
    agent._bedrock_region = bedrock_region_from_runtime_url(base_url)
    agent._bedrock_guardrail_config = bedrock_guardrail_config()
    agent.client = None
    agent._client_kwargs = {}
    agent.api_key = agent._anthropic_api_key = "aws-sdk"
    agent._anthropic_base_url = base_url
    agent._is_anthropic_oauth = False
    if api_mode == "anthropic_messages":
        from agent.anthropic_adapter import build_anthropic_bedrock_client
        agent._anthropic_client = build_anthropic_bedrock_client(agent._bedrock_region)
    else:
        agent._anthropic_client = None


def bedrock_model_ids_or_none() -> Optional[list[str]]:
    """Live-discover Bedrock model IDs; None on failure/empty so callers use the static list."""
    with suppress(Exception):
        discovered = discover_bedrock_models(resolve_bedrock_runtime_region())
        if discovered:
            return merge_bedrock_openai_model_ids([m["id"] for m in discovered])
    return None


# --- Tool-calling / prompt-cache capability detection ---
# Models known to reject toolConfig with a ValidationException; unknown models assumed OK.
_NON_TOOL_CALLING_PATTERNS = [
    "deepseek.r1", "deepseek-r1",  # DeepSeek R1 (both ID formats) — reasoning only
    "stability.",  # image generation
    "cohere.embed", "amazon.titan-embed",  # embeddings
]

# cachePoint allowlist — inverted policy vs tools: unknown models get NO cache markers (they reject
# cachePoint). Claude only reaches build_converse_kwargs under bearer auth.
_CACHE_POINT_PATTERNS = ["anthropic.claude", "amazon.nova"]


def _model_supports_tool_use(model_id: str) -> bool:
    """False for denylisted models; unknown models default to True."""
    return not any(pattern in model_id.lower() for pattern in _NON_TOOL_CALLING_PATTERNS)


def _model_supports_prompt_cache(model_id: str) -> bool:
    # An application-inference-profile ARN names no model: match on the wrapped model (cached lookup).
    if _APPLICATION_PROFILE_ARN_RE.search(model_id):
        model_id = _resolve_inference_profile_model_id(model_id)
    return any(pattern in model_id.lower() for pattern in _CACHE_POINT_PATTERNS)


# --- Server-verdict cachePoint suppression ---
# Bedrock's cachePoint rule is per-family AND per-field (Nova accepts it in system/messages but hard-fails
# on toolConfig.tools) and any static table drifts, so when Bedrock names a placement as unpermitted we
# record the verdict, drop the marker there for the rest of the process, and retry once without it.

CACHE_POINT_PLACEMENTS = ("tools", "system", "messages")
_CACHE_POINT_REJECTIONS: dict[str, set] = {}  # model_id (lowercased) → placements Bedrock has rejected this process
# e.g. "#/toolConfig/tools/18: extraneous key [cachePoint] is not permitted"
_CACHE_POINT_PATH_PATTERN = re.compile(r"#/(?P<path>[A-Za-z0-9_./\[\]-]*)", re.IGNORECASE)
_CACHE_POINT = {"cachePoint": {"type": "default"}}


def cache_point_rejection_placement(exc: BaseException) -> Optional[str]:
    """Converse section whose cachePoint Bedrock refused, or None. Message-based: the JSON pointer in the
    ValidationException is the only thing naming the section (raw or SDK-wrapped). Unlocalisable → "tools"."""
    msg = str(exc)
    lowered = msg.lower()
    if "cachepoint" not in lowered or ("not permitted" not in lowered and "extraneous" not in lowered):
        return None
    match = _CACHE_POINT_PATH_PATTERN.search(msg)
    path = (match.group("path") if match else "").lower()
    if "toolconfig" in path or "tools" in path:
        return "tools"
    return next((placement for placement in ("system", "messages") if placement in path), "tools")


def note_cache_point_rejection(model_id: str, placement: str) -> None:
    """Record that ``model_id`` refuses cachePoint blocks in ``placement``."""
    if placement in CACHE_POINT_PLACEMENTS:
        _CACHE_POINT_REJECTIONS.setdefault(model_id.lower(), set()).add(placement)


def cache_point_allowed(model_id: str, placement: str) -> bool:
    """False once Bedrock has refused this placement for this model."""
    return placement not in _CACHE_POINT_REJECTIONS.get(model_id.lower(), ())


def reset_cache_point_rejections() -> None:
    """Clear recorded cachePoint rejections. Used in tests."""
    _CACHE_POINT_REJECTIONS.clear()


def _without_cache_points(blocks: Any) -> Optional[list]:
    """``blocks`` minus cachePoint entries, or None if not a list / nothing removed."""
    if not isinstance(blocks, list):
        return None
    cleaned = [b for b in blocks if not (isinstance(b, dict) and set(b.keys()) == {"cachePoint"})]
    return None if len(cleaned) == len(blocks) else cleaned


def strip_cache_points(kwargs: dict[str, Any], placement: str) -> dict[str, Any]:
    """Copy of Converse kwargs with ``placement``'s cachePoint removed; the SAME object
    back when nothing was stripped (callers use identity to decide a retry cannot help)."""
    if placement == "messages":
        messages = kwargs.get("messages")
        cleaned_contents = [
            _without_cache_points(msg.get("content") if isinstance(msg, dict) else None) for msg in messages
        ] if isinstance(messages, list) else []
        if all(content is None for content in cleaned_contents):
            return kwargs
        return {**kwargs, "messages": [
            msg if content is None else {**msg, "content": content} for msg, content in zip(messages, cleaned_contents)
        ]}
    if placement == "system":
        cleaned = _without_cache_points(kwargs.get("system"))
        return kwargs if cleaned is None else {**kwargs, "system": cleaned}
    if placement == "tools":
        tool_config = kwargs.get("toolConfig")
        cleaned = _without_cache_points((tool_config or {}).get("tools"))
        return kwargs if cleaned is None else {**kwargs, "toolConfig": {**tool_config, "tools": cleaned}}
    return kwargs


def recover_from_cache_point_rejection(exc: BaseException, kwargs: dict[str, Any]) -> Optional[dict[str, Any]]:
    """Record Bedrock's cachePoint verdict and return retry kwargs, or None when the error
    was not a cachePoint rejection / the marker was already absent (caller re-raises)."""
    placement = cache_point_rejection_placement(exc)
    if placement is None:
        return None
    retry_kwargs = strip_cache_points(kwargs, placement)
    if retry_kwargs is kwargs:
        return None
    model_id = str(kwargs.get("modelId", ""))
    note_cache_point_rejection(model_id, placement)
    logger.warning(
        "bedrock: %s rejected a cachePoint block in %s — dropping that cache marker for this model and "
        "retrying. Prompt caching stays active for the remaining sections.", model_id or "model", placement,
    )
    return retry_kwargs


# --- Encrypted-content redacted-reasoning suppression ---
# Redacted thinking blobs are sealed to the issuing model/flow; replaying them after a model switch or
# across regions fails with an encrypted-content ValidationException. Drop the redacted blocks and
# resend once (mirrors the cachePoint self-heal above: same-object return means no retry can help).
_REDACTED_REASONING_REJECTION_PATTERN = re.compile(
    r"ValidationException.*(?:redacted|encrypt)", re.IGNORECASE | re.DOTALL,
)


def _without_redacted_reasoning(blocks):
    """``blocks`` minus reasoningContent entries carrying redactedContent, or None if not a list /
    nothing removed. Pure reasoningText blocks are kept."""
    if not isinstance(blocks, list):
        return None
    cleaned = [b for b in blocks if not (
        isinstance(b, dict) and isinstance(b.get("reasoningContent"), dict)
        and "redactedContent" in b["reasoningContent"]
    )]
    return None if len(cleaned) == len(blocks) else cleaned


def strip_redacted_reasoning(kwargs):
    """Copy of Converse kwargs with redacted reasoning blocks removed; the SAME object
    back when nothing was stripped (callers use identity to decide a retry cannot help).
    Turns left with empty content are dropped (they carried no replayable signal)."""
    messages = kwargs.get("messages")
    if not isinstance(messages, list):
        return kwargs
    cleaned_contents = [
        _without_redacted_reasoning(msg.get("content") if isinstance(msg, dict) else None)
        for msg in messages
    ]
    if all(content is None for content in cleaned_contents):
        return kwargs
    return {**kwargs, "messages": [
        msg if content is None else {**msg, "content": content}
        for msg, content in zip(messages, cleaned_contents)
        if content is None or len(content) > 0
    ]}


def recover_from_redacted_reasoning_rejection(exc, kwargs):
    """Return retry kwargs with redacted reasoning blocks stripped, or None when the error
    was not an encrypted-content rejection / nothing redacted remained (caller re-raises)."""
    if not _REDACTED_REASONING_REJECTION_PATTERN.search(str(exc)):
        return None
    retry_kwargs = strip_redacted_reasoning(kwargs)
    if retry_kwargs is kwargs:
        return None
    logger.warning(
        "bedrock: %s rejected replayed redacted reasoning (encrypted content is sealed to the issuing "
        "model/flow) - stripping redacted blocks and resending once.",
        str(kwargs.get("modelId", "")) or "model",
    )
    return retry_kwargs


# One optional regional/global inference-profile prefix, then the Claude model family.
_ANTHROPIC_BEDROCK_MODEL_RE = re.compile(
    rf"^(?:(?:{'|'.join(_BEDROCK_PROFILE_PREFIXES)})\.)?anthropic\.claude", re.IGNORECASE,
)


def is_anthropic_bedrock_model(model_id: str) -> bool:
    """True for Claude on Bedrock (``anthropic.claude-*``, any regional prefix): AnthropicBedrock SDK path."""
    return _ANTHROPIC_BEDROCK_MODEL_RE.match(model_id) is not None


# --- Message format conversion: OpenAI → Bedrock Converse ---

def convert_tools_to_converse(tools: list[dict]) -> list[dict]:
    """OpenAI ``{"function": {...}}`` tool defs → Converse ``{"toolSpec": {...}}``."""
    return [{"toolSpec": {
        "name": fn.get("name", ""), "description": fn.get("description", ""),
        "inputSchema": {"json": fn.get("parameters", {"type": "object", "properties": {}})},
    }} for fn in (t.get("function", {}) for t in tools or [])]


# Converse rejects empty OR whitespace-only text blocks, so the placeholder must be non-whitespace.
# A lone space is whitespace and is rejected too — the placeholder MUST itself be non-whitespace. Ref: issue
# #9486.
_EMPTY_TEXT_PLACEHOLDER = "(empty)"
_PLACEHOLDER_BLOCK = {"text": _EMPTY_TEXT_PLACEHOLDER}


def _safe_text(text) -> str:
    """``text`` if it has non-whitespace content, else the placeholder (None/non-str ok)."""
    text = "" if text is None else str(text)
    return text if text.strip() else _EMPTY_TEXT_PLACEHOLDER


def _image_block_from_data_url(url: str) -> dict:
    """``data:<mime>;base64,...`` → Converse image block with RAW bytes (boto3 base64-encodes on the
    wire; passing the string through double-encodes and Bedrock rejects it)."""
    header, _, data = url.partition(",")
    media_type = (header[5:].split(";")[0] if header.startswith("data:") else "") or "image/jpeg"
    try:
        # Ref: #33317.
        raw_bytes = base64.b64decode(data)
    except Exception:
        raw_bytes = data.encode("utf-8")
    return {"image": {"format": media_type.split("/")[-1] if "/" in media_type else "jpeg", "source": {"bytes": raw_bytes}}}


def _convert_content_to_converse(content) -> list[dict]:
    """OpenAI content → Converse blocks; blank text → placeholder, remote image URLs → text reference."""
    if not isinstance(content, list):
        return [{"text": _safe_text(content)}]
    blocks = []
    for part in content:
        if isinstance(part, str):
            blocks.append({"text": _safe_text(part)})
        elif isinstance(part, dict) and part.get("type", "") == "text":
            blocks.append({"text": _safe_text(part.get("text", ""))})
        elif isinstance(part, dict) and part.get("type", "") == "image_url":
            image_url = part.get("image_url", {})
            url = image_url.get("url", "") if isinstance(image_url, dict) else ""
            blocks.append(_image_block_from_data_url(url) if url.startswith("data:") else {"text": f"[Image: {url}]"})
    return blocks or [dict(_PLACEHOLDER_BLOCK)]


def _system_blocks(content) -> list[dict]:
    """System content → text blocks; blank parts are dropped, not placeholder-filled."""
    parts = [content] if isinstance(content, str) else content if isinstance(content, list) else []
    texts = [part.get("text", "") if isinstance(part, dict) and part.get("type") == "text" else part for part in parts]
    return [{"text": text} for text in texts if isinstance(text, str) and text.strip()]


def _tool_use_block(tool_use_id, name, input_dict) -> dict:
    return {"toolUse": {"toolUseId": tool_use_id, "name": name, "input": input_dict}}


def _decode_redacted(encoded) -> Optional[bytes]:
    """Strict base64 → bytes; None for empty/non-str/undecodable input."""
    try:
        return base64.b64decode(encoded, validate=True) if isinstance(encoded, str) and encoded else None
    except (ValueError, TypeError):
        return None


def _replay_ordered_blocks(ordered_blocks: list) -> list[dict]:
    """Rebuild the exact Bedrock block sequence captured at normalization time; redacted reasoning is
    stored base64 (JSON-safe sidecar) and undecodable entries are skipped."""
    content_blocks: list[dict] = []
    for block in ordered_blocks:
        if not isinstance(block, dict):
            continue
        if "text" in block and isinstance(block["text"], str):
            content_blocks.append({"text": block["text"]})
        elif "reasoningContent" in block:
            reasoning = block["reasoningContent"]
            if not isinstance(reasoning, dict):
                continue
            # ReasoningContentBlock is a tagged union: reasoningText and redactedContent must go
            # out as separate blocks (#115865). Undecodable redacted entries are skipped alone.
            if isinstance(reasoning.get("text"), str):
                reasoning_text: dict[str, str] = {"text": reasoning["text"]}
                if isinstance(reasoning.get("signature"), str) and reasoning["signature"]:
                    reasoning_text["signature"] = reasoning["signature"]  # models that sign thinking reject unsigned replay
                content_blocks.append({"reasoningContent": {"reasoningText": reasoning_text}})
            encoded = reasoning.get("redactedContentBase64")
            if isinstance(encoded, str) and encoded:
                redacted = _decode_redacted(encoded)
                if redacted is not None:
                    content_blocks.append({"reasoningContent": {"redactedContent": redacted}})
        elif "toolUse" in block and isinstance(block["toolUse"], dict):
            tu = block["toolUse"]
            content_blocks.append(_tool_use_block(tu.get("toolUseId", ""), tu.get("name", ""), tu.get("input", {})))
    return content_blocks


def _parse_tool_args(args) -> Any:
    """JSON-decode a tool-call argument string; {} on failure; non-str passes through."""
    try:
        return json.loads(args) if isinstance(args, str) else args
    except (json.JSONDecodeError, TypeError):
        return {}


def _redacted_reasoning_block(encoded) -> Optional[dict]:
    redacted = _decode_redacted(encoded)
    return {"reasoningContent": {"redactedContent": redacted}} if redacted is not None else None


def _reasoning_text_block(text, signature) -> dict:
    reasoning_text = {"text": text if isinstance(text, str) else ""}
    if isinstance(signature, str) and signature:
        reasoning_text["signature"] = signature
    return {"reasoningContent": {"reasoningText": reasoning_text}}


# Anthropic Messages block -> Converse block, for turns produced on the AnthropicBedrock path and sent
# over Converse after the sticky stream-denied fallback (same Claude signatures, Converse union shape).
_ANTHROPIC_TO_CONVERSE = {
    "thinking": lambda b, args: _reasoning_text_block(b.get("thinking"), b.get("signature")),
    "redacted_thinking": lambda b, args: _redacted_reasoning_block(b.get("data")),
    "text": lambda b, args: {"text": b["text"]} if isinstance(b.get("text"), str) and b["text"].strip() else None,
    # Input is re-sourced from the redacted tool_calls, never the raw captured block.
    "tool_use": lambda b, args: _tool_use_block(b.get("id", ""), b.get("name", ""), args.get(b.get("id"), b.get("input", {}))),
}


def _anthropic_ordered_blocks(msg: dict) -> list[dict]:
    args = {
        tc.get("id"): _parse_tool_args((tc.get("function") or {}).get("arguments", "{}"))
        for tc in (msg.get("tool_calls") or []) if isinstance(tc, dict)
    }
    converted = (
        _ANTHROPIC_TO_CONVERSE[b["type"]](b, args)
        for b in (msg.get("anthropic_content_blocks") or [])
        if isinstance(b, dict) and b.get("type") in _ANTHROPIC_TO_CONVERSE
    )
    return [block for block in converted if block is not None]


def _reasoning_details_blocks(msg: dict, claude: bool, own: bool = False) -> list[dict]:
    """Thinking from the persisted ``reasoning_details`` (Anthropic shape, as Converse captures it), in stored
    order: Claude gets signed thinking + redacted blobs, other models readable text only (falling back to
    ``reasoning_content``). This is the carrier a turn reloaded from state.db replays from, since the
    ordered ``bedrock_content_blocks`` sidecar is live-only. Converse wants reasoning ahead of toolUse."""
    blocks = []
    for d in msg.get("reasoning_details") or []:
        if not isinstance(d, dict):
            continue
        if d.get("type") == "thinking" and isinstance(d.get("thinking"), str) and d["thinking"].strip():
            if (claude or own) and d.get("signature"):
                blocks.append(_reasoning_text_block(d["thinking"], d["signature"]))
            elif not claude:
                blocks.append(_reasoning_text_block(d["thinking"], None))
        elif d.get("type") == "redacted_thinking" and (
                block := _redacted_reasoning_block(d.get("data") or d.get("redactedContentBase64"))):
            blocks.append(block)
    text = msg.get("reasoning_content")
    if not blocks and not claude and isinstance(text, str) and text.strip():
        blocks.append(_reasoning_text_block(text, None))
    return blocks


def _assistant_blocks(msg: dict, content, claude: bool = True, own: bool = False) -> list[dict]:
    """Assistant message → Converse blocks, from the first carrier that yields any: the ordered
    ``bedrock_content_blocks`` sidecar, (Claude only) the ordered ``anthropic_content_blocks`` sidecar,
    else thinking from ``reasoning_details`` followed by text, then tool calls. Anthropic-shaped carriers
    hold Claude signatures, so other models never read them."""
    ordered_blocks = msg.get("bedrock_content_blocks")
    if isinstance(ordered_blocks, list) and (content_blocks := _replay_ordered_blocks(ordered_blocks)):
        return content_blocks
    if claude and (content_blocks := _anthropic_ordered_blocks(msg)):
        return content_blocks
    content_blocks = _reasoning_details_blocks(msg, claude, own)
    if isinstance(content, str) and content.strip():
        content_blocks.append({"text": content})
    elif isinstance(content, list):
        content_blocks.extend(_convert_content_to_converse(content))
    for tc in (msg.get("tool_calls", []) or []):
        fn = tc.get("function", {})
        content_blocks.append(_tool_use_block(tc.get("id", ""), fn.get("name", ""), _parse_tool_args(fn.get("arguments", "{}"))))
    return content_blocks


# Per-model reasoning replay on Converse: (turns before the in-flight tool loop, turns inside it).
#   signed   — Claude: everything verbatim, signatures and redactedContent included (AWS: "You must include
#              the signature and all previous messages"; Bedrock ignores older turns server-side for free).
#   strip    — DeepSeek-R1 (AWS's multi-turn sample removes reasoningContent) and Kimi K3 (Converse raises
#              InternalServerException "when reasoning content from earlier turns is included"): prior turns
#              lose reasoning, the in-flight loop replays the model's own blocks as captured.
#   issuer   — OpenAI's reasoning GPTs (gpt-6 family) reject readable reasoningText in assistant history ("This
#              model doesn't support the reasoningContent.reasoningText.text field") and stall on another GPT's
#              sealed blobs: only turns the answering model stamped keep reasoning, verbatim so the replayed
#              prefix stays stable; every other turn replays none, in the in-flight loop too.
#   unsigned — every other model: prior turns carry readable reasoningText only, since their signatures and
#              redacted blobs may be another model's (a /model switch from Claude) and non-Claude models
#              reject them ("This model doesn't support the reasoningContent.reasoningText.signature field");
#              the in-flight loop was produced by the model now answering and replays as captured.
# A turn stamped by another model (CONVERSE_ISSUER_TYPE) never gets the in-flight verbatim replay: a fallback or
# switch mid-loop hands the loop to a model that cannot open the previous one's sealed blobs (_reasoning_filter).
_CONVERSE_REPLAY_STRIP_PATTERNS = ("deepseek.r1", "deepseek-r1", "moonshotai.kimi-k3")
_CONVERSE_REPLAY_ISSUER_PATTERNS = ("openai.gpt-6",)  # not gpt-oss: it takes readable reasoningText


# reasoning_details record naming the model that minted a turn's Converse reasoning: its signatures are that
# model's own, so it gets them back unmodified on every later turn and after a reload (AWS: "include the text and
# its signature unmodified"); any other model gets the per-policy prior-turn treatment, and Claude gets no
# reasoning at all from a turn another family minted.
CONVERSE_ISSUER_TYPE = "bedrock.native_assistant"


def _issuer(msg: dict) -> Optional[str]:
    return next((d["model"] for d in msg.get("reasoning_details") or []
                 if isinstance(d, dict) and d.get("type") == CONVERSE_ISSUER_TYPE and isinstance(d.get("model"), str)), None)


def _converse_replay_policy(model: Optional[str]) -> str:
    if model is None:
        return "signed"
    if _APPLICATION_PROFILE_ARN_RE.search(model):
        model = _resolve_inference_profile_model_id(model)  # cached; the profile ARN names no model
    lowered = model.lower()
    if "claude" in lowered:
        return "signed"
    if any(p in lowered for p in _CONVERSE_REPLAY_ISSUER_PATTERNS):
        return "issuer"
    return "strip" if any(p in lowered for p in _CONVERSE_REPLAY_STRIP_PATTERNS) else "unsigned"


def _unsigned_reasoning(block: dict) -> Optional[dict]:
    reasoning = block.get("reasoningContent")
    if not isinstance(reasoning, dict):
        return block
    text = (reasoning.get("reasoningText") or {}).get("text")
    return {"reasoningContent": {"reasoningText": {"text": text}}} if isinstance(text, str) and text.strip() else None


def _verbatim(block: dict) -> dict:
    return block


def _without_reasoning(block: dict) -> Optional[dict]:
    return None if "reasoningContent" in block else block


_REASONING_FILTERS = {  # policy -> (before the in-flight tool loop, inside it)
    "signed": (_verbatim, _verbatim),
    "strip": (_without_reasoning, _verbatim),
    "unsigned": (_unsigned_reasoning, _verbatim),
    "issuer": (_without_reasoning, _without_reasoning),
}


def _filter_reasoning(blocks: list[dict], keep) -> list[dict]:
    return [kept for kept in map(keep, blocks) if kept is not None]


def _reasoning_filter(policy: str, issuer: Optional[str], own: bool, *, in_loop: bool):
    """Which reasoning one assistant turn may replay to the model now answering. A turn stamped by ANOTHER
    model (fallback or ``/model`` switch, including mid tool loop) loses the in-flight loop's verbatim replay:
    sealed ``redactedContent`` opens only for its issuer, and Claude rejects another family's blobs outright
    (``Invalid `data` in `redacted_thinking` block``), so a non-Claude turn replays no reasoning to Claude.
    Unstamped turns (older sessions) keep the policy, and the rejection-driven resend covers them; under
    ``issuer`` only the answering model's own stamped turns keep any reasoning."""
    prior, loop = _REASONING_FILTERS[policy]
    if own:
        return _verbatim if in_loop or policy in ("unsigned", "issuer") else prior
    if issuer is not None:
        return _without_reasoning if policy == "signed" and _converse_replay_policy(issuer) != "signed" else prior
    return loop if in_loop else prior


def convert_messages_to_converse(
    messages: list[dict], model: Optional[str] = None,
) -> tuple[Optional[list[dict]], list[dict]]:
    """OpenAI messages → ``(system_blocks_or_None, converse_messages)``; tool results become ``toolResult``
    user blocks. Converse needs strict user/assistant alternation with a user turn first and last:
    same-role neighbours merge, placeholder user turns pad the ends. ``model`` selects the reasoning
    replay policy for turns before the in-flight tool loop (None replays every carrier verbatim)."""
    policy = _converse_replay_policy(model)
    # Tool results are role "tool" here, so the last role "user" message opens the in-flight tool loop.
    loop_start = next((i for i in range(len(messages) - 1, -1, -1) if messages[i].get("role") == "user"), -1)
    system_blocks: list[dict] = []
    converse_msgs: list[dict] = []

    def append_turn(role: str, blocks: list[dict]) -> None:
        if converse_msgs and converse_msgs[-1]["role"] == role:
            converse_msgs[-1]["content"].extend(blocks)
        else:
            converse_msgs.append({"role": role, "content": blocks})

    for idx, msg in enumerate(messages):
        role = msg.get("role", "")
        content = msg.get("content")
        if role == "system":
            system_blocks.extend(_system_blocks(content))
        elif role == "tool":
            result_content = content if isinstance(content, str) else json.dumps(content)
            append_turn("user", [{"toolResult": {
                "toolUseId": msg.get("tool_call_id", ""), "content": [{"text": _safe_text(result_content)}]}}])
        elif role == "assistant":
            issuer = _issuer(msg) if model is not None else None
            own = issuer is not None and issuer == model
            keep = _reasoning_filter(policy, issuer, own, in_loop=idx >= loop_start)
            blocks = _filter_reasoning(_assistant_blocks(msg, content, claude=policy == "signed", own=own), keep)
            append_turn("assistant", blocks or [dict(_PLACEHOLDER_BLOCK)])
        elif role == "user":
            append_turn("user", _convert_content_to_converse(content))
    if converse_msgs and converse_msgs[0]["role"] != "user":
        converse_msgs.insert(0, {"role": "user", "content": [dict(_PLACEHOLDER_BLOCK)]})
    if converse_msgs and converse_msgs[-1]["role"] != "user":
        converse_msgs.append({"role": "user", "content": [dict(_PLACEHOLDER_BLOCK)]})
    return (system_blocks or None, converse_msgs)


# --- Response format conversion: Bedrock Converse → OpenAI ---

# Bedrock stopReason → OpenAI finish_reason (unknown → "stop").
_STOP_REASON_TO_FINISH_REASON = {
    "end_turn": "stop", "stop_sequence": "stop", "tool_use": "tool_calls", "max_tokens": "length",
    "content_filtered": "content_filter", "guardrail_intervened": "content_filter",
}


def _encode_redacted(redacted) -> Optional[str]:
    """Redacted reasoning payload → base64 str (bytes encoded, str passed through, else None)."""
    if isinstance(redacted, (bytes, bytearray)):
        return base64.b64encode(bytes(redacted)).decode("ascii")
    return redacted if isinstance(redacted, str) else None


def _tool_call_ns(tool_use_id: str, name: str, input_dict) -> SimpleNamespace:
    return SimpleNamespace(
        id=tool_use_id, type="function", function=SimpleNamespace(name=name, arguments=json.dumps(input_dict)),
    )


def _persisted_reasoning_details(ordered_blocks: list[dict[str, Any]]) -> list[dict[str, Any]]:
    details: list[dict[str, Any]] = []
    for block in ordered_blocks:
        reasoning = block.get("reasoningContent") if isinstance(block, dict) else None
        if not isinstance(reasoning, dict):
            continue
        if isinstance(reasoning.get("text"), str) and reasoning["text"]:
            details.append({"type": "thinking", "thinking": reasoning["text"],
                            **({"signature": reasoning["signature"]} if reasoning.get("signature") else {})})
        if reasoning.get("redactedContentBase64"):
            details.append({"type": "redacted_thinking", "data": reasoning["redactedContentBase64"]})
    return details


class _ResponseParts:
    """Accumulator shared by the sync and streaming normalizers."""

    def __init__(self) -> None:
        self.text_parts: list[str] = []
        self.reasoning_details: list[dict[str, Any]] = []
        self.tool_calls: list[SimpleNamespace] = []

    def absorb_reasoning(self, reasoning: Any, block: dict[str, Any], on_text=None) -> None:
        """Fold a Converse ``reasoningContent`` payload into the accumulators and ``block``. The sync response
        nests ``reasoningText: {text, signature}``; stream deltas carry ``text`` / ``signature`` flat."""
        if not isinstance(reasoning, dict):
            return
        if isinstance(reasoning.get("reasoningText"), dict):
            reasoning = {**reasoning, **reasoning["reasoningText"]}
        thinking_text = reasoning.get("text", "")
        if thinking_text:
            if on_text:
                on_text(thinking_text)
            block["text"] = block.get("text", "") + str(thinking_text)
        signature = reasoning.get("signature")
        if isinstance(signature, str) and signature:
            block["signature"] = block.get("signature", "") + signature
        encoded = _encode_redacted(reasoning.get("redactedContent"))
        if encoded:
            self.reasoning_details.append({"type": "redacted_thinking", "data": encoded})
            block["redactedContentBase64"] = encoded

    def build(self, ordered_blocks: list[dict[str, Any]], usage_data: dict[str, int], stop_reason: str, model: str) -> SimpleNamespace:
        """Assemble the OpenAI-shaped response. Converse's inputTokens EXCLUDES cache read/write tokens
        (OpenAI's prompt_tokens includes them), so they are added back. ``reasoning_details`` carries every
        reasoning block in Anthropic shape (signed thinking included) because it is the persisted column."""
        msg = SimpleNamespace(
            role="assistant", content="\n".join(self.text_parts) if self.text_parts else None,
            tool_calls=self.tool_calls or None,
            reasoning_details=_persisted_reasoning_details(ordered_blocks) or self.reasoning_details or None,
            reasoning_content="\n\n".join(texts) if (texts := [b["reasoningContent"]["text"] for b in ordered_blocks
                if isinstance(b.get("reasoningContent"), dict) and b["reasoningContent"].get("text")]) else None,
            bedrock_content_blocks=ordered_blocks or None,
        )
        cache_read_tokens, cache_write_tokens, output_tokens = (
            usage_data.get(k, 0) for k in ("cacheReadInputTokens", "cacheWriteInputTokens", "outputTokens")
        )
        prompt_tokens = usage_data.get("inputTokens", 0) + cache_read_tokens + cache_write_tokens
        usage = SimpleNamespace(
            prompt_tokens=prompt_tokens, completion_tokens=output_tokens, total_tokens=prompt_tokens + output_tokens,
            cache_read_input_tokens=cache_read_tokens, cache_creation_input_tokens=cache_write_tokens,
        )
        finish_reason = _STOP_REASON_TO_FINISH_REASON.get(stop_reason, "stop")
        if self.tool_calls and finish_reason == "stop":
            finish_reason = "tool_calls"
        return SimpleNamespace(
            choices=[SimpleNamespace(index=0, message=msg, finish_reason=finish_reason)], usage=usage, model=model,
        )


def normalize_converse_response(response: dict) -> SimpleNamespace:
    """Bedrock Converse response → OpenAI ``ChatCompletion``-shaped SimpleNamespace (``.choices[0].message.
    {content,tool_calls,reasoning_content,reasoning_details,bedrock_content_blocks}``, ``finish_reason``, ``.usage``)."""
    parts = _ResponseParts()
    ordered_blocks = []
    for block in response.get("output", {}).get("message", {}).get("content", []):
        if "text" in block:
            parts.text_parts.append(block["text"])
            ordered_blocks.append({"text": block["text"]})
        elif "reasoningContent" in block:
            ordered_reasoning: dict[str, Any] = {}
            parts.absorb_reasoning(block["reasoningContent"], ordered_reasoning)
            if ordered_reasoning:
                ordered_blocks.append({"reasoningContent": ordered_reasoning})
        elif "toolUse" in block:
            tu = block["toolUse"]
            ordered_blocks.append(_tool_use_block(tu.get("toolUseId", ""), tu.get("name", ""), tu.get("input", {})))
            parts.tool_calls.append(_tool_call_ns(tu.get("toolUseId", ""), tu.get("name", ""), tu.get("input", {})))
    return parts.build(
        ordered_blocks, response.get("usage", {}), response.get("stopReason", "end_turn"), response.get("modelId", ""),
    )


# --- Streaming response conversion ---

def normalize_converse_stream_events(event_stream) -> SimpleNamespace:
    """Consume a ConverseStream event stream (no callbacks) → same shape as ``normalize_converse_response()``."""
    return stream_converse_with_callbacks(event_stream)


def stream_converse_with_callbacks(
    event_stream, on_text_delta=None, on_tool_start=None, on_reasoning_delta=None,
    on_interrupt_check=None, on_event=None,
) -> SimpleNamespace:
    """boto3 ``converse_stream()`` response + callbacks → the ``normalize_converse_response()`` shape.
    ``on_text_delta`` only fires while no toolUse block has been seen (as on the Anthropic/chat_completions
    paths); ``on_interrupt_check`` True stops streaming; ``on_event`` fires for EVERY event before branching
    and its exceptions are swallowed so a watchdog hook can never abort the stream.

    Blocks are keyed by the ``contentBlockIndex`` Bedrock stamps on every contentBlockStart/Delta/Stop:
    text blocks get NO contentBlockStart, so a counter keyed on starts shredded them (#108200)."""
    parts = _ResponseParts()
    stream_blocks: dict[int, dict[str, Any]] = {}
    current_block_index: Optional[int] = None
    current_tool: Optional[dict] = None
    current_text_buffer: list[str] = []
    has_tool_use = False
    stop_reason = None
    interrupted = False
    usage_data: dict[str, int] = {}

    def block_index(payload: dict[str, Any], *, new_block: bool = False) -> int:
        """Index of the block a contentBlock* event addresses. Without ``contentBlockIndex`` (test doubles,
        proxies) a start opens a fresh slot and a delta/stop continues the current one."""
        idx = payload.get("contentBlockIndex")
        if isinstance(idx, int):
            return idx
        return len(stream_blocks) if new_block or current_block_index is None else current_block_index

    def flush_text() -> None:
        if current_text_buffer:
            parts.text_parts.append("".join(current_text_buffer))
            current_text_buffer.clear()

    for event in event_stream.get("stream", []):
        if on_event is not None:
            with suppress(Exception):
                on_event()
        if on_interrupt_check and on_interrupt_check():
            interrupted = True
            break
        if "contentBlockStart" in event:
            start_event = event["contentBlockStart"]
            idx = current_block_index = block_index(start_event, new_block=True)
            start = start_event.get("start", {})
            if "toolUse" in start:
                has_tool_use = True
                flush_text()
                current_tool = {"toolUseId": start["toolUse"].get("toolUseId", ""), "name": start["toolUse"].get("name", ""), "input_json": ""}
                stream_blocks[idx] = _tool_use_block(current_tool["toolUseId"], current_tool["name"], {})
                if on_tool_start:
                    on_tool_start(current_tool["name"])
        elif "contentBlockDelta" in event:
            delta_event = event["contentBlockDelta"]
            idx = current_block_index = block_index(delta_event)
            delta = delta_event.get("delta", {})
            if "text" in delta:
                text = delta["text"]
                block = stream_blocks.setdefault(idx, {"text": ""})
                block["text"] = block.get("text", "") + text
                current_text_buffer.append(text)
                if on_text_delta and not has_tool_use:
                    on_text_delta(text)
            elif "toolUse" in delta and current_tool is not None:
                current_tool["input_json"] += delta["toolUse"].get("input", "")
            elif "reasoningContent" in delta:
                reasoning = delta["reasoningContent"]
                if isinstance(reasoning, dict) and (reasoning.get("text", "") or reasoning.get("signature") or _encode_redacted(reasoning.get("redactedContent"))):
                    block = stream_blocks.setdefault(idx, {"reasoningContent": {}}).setdefault("reasoningContent", {})
                    parts.absorb_reasoning(reasoning, block, on_reasoning_delta)
        elif "contentBlockStop" in event:
            idx = block_index(event["contentBlockStop"])
            current_block_index = None  # a following index-less delta opens a fresh slot, not this one
            if current_tool is not None:
                input_dict = _parse_tool_args(current_tool["input_json"])  # "" → {} via the JSON-error path
                parts.tool_calls.append(_tool_call_ns(current_tool["toolUseId"], current_tool["name"], input_dict))
                if "toolUse" in stream_blocks.get(idx, {}):
                    stream_blocks[idx]["toolUse"]["input"] = input_dict
                current_tool = None
            else:
                flush_text()
        elif "messageStop" in event:
            stop_reason = event["messageStop"].get("stopReason", "end_turn")
        elif "metadata" in event:
            meta_usage = event["metadata"].get("usage", {})
            usage_data = {key: meta_usage.get(key, 0) for key in ("inputTokens", "outputTokens", "cacheReadInputTokens", "cacheWriteInputTokens")}
    if stop_reason is None and not interrupted:
        raise EmptyStreamError("Bedrock Converse stream ended before messageStop; response is incomplete")
    flush_text()
    return parts.build([stream_blocks[i] for i in sorted(stream_blocks)], usage_data, stop_reason or "end_turn", "")


# --- High-level API: call Bedrock Converse ---

def _claude_converse_reasoning(model: str, reasoning_config: dict, max_tokens: Optional[int]) -> tuple[dict, dict]:
    """Claude thinking over Converse: the same ``thinking``/``output_config`` the AnthropicBedrock path sends,
    carried in ``additionalModelRequestFields`` (AWS: "pass the thinking and effort parameters inside
    additionalModelRequestFields"); manual budgets also need temperature 1 and room above the budget."""
    from agent.anthropic_adapter import _thinking_kwargs

    thinking = _thinking_kwargs(reasoning_config, model, max_tokens or 0)
    fields = {k: thinking[k] for k in ("thinking", "output_config") if k in thinking}
    inference: dict[str, Any] = {}
    if (fields.get("thinking") or {}).get("type") == "enabled":
        if "claude-3" not in model.lower():  # 3.7 has no interleaved thinking
            fields["anthropic_beta"] = ["interleaved-thinking-2025-05-14"]
        inference["temperature"] = thinking.get("temperature", 1)
        if max_tokens is not None:
            inference["maxTokens"] = thinking.get("max_tokens", max_tokens)
    return fields, inference


_NOVA_EFFORT = {"minimal": "low", "low": "low", "medium": "medium"}  # anything stronger -> "high"


def _nova_converse_reasoning(model: str, reasoning_config: dict, max_tokens: Optional[int]) -> tuple[dict, dict]:
    """Nova 2 ``reasoningConfig`` (off by default). At ``high``, temperature/topP/maxTokens must be unset."""
    if reasoning_config.get("enabled") is False:
        return {}, {}
    effort = _NOVA_EFFORT.get(str(reasoning_config.get("effort", "medium")).lower(), "high")
    return {"reasoningConfig": {"type": "enabled", "maxReasoningEffort": effort}}, (
        {"temperature": None, "topP": None, "maxTokens": None} if effort == "high" else {}
    )


_CONVERSE_REASONING_BUILDERS = (
    (is_anthropic_bedrock_model, _claude_converse_reasoning),
    (lambda model: "amazon.nova-2-lite" in model.lower(), _nova_converse_reasoning),
)


def _apply_converse_reasoning(kwargs: dict, inference_config: dict, model: str, reasoning_config: dict, max_tokens) -> None:
    builder = next((b for matches, b in _CONVERSE_REASONING_BUILDERS if matches(model or "")), None)
    if builder is None:
        return
    fields, inference = builder(model, reasoning_config, max_tokens)
    if fields:
        kwargs["additionalModelRequestFields"] = fields
    for key, value in inference.items():
        if value is None:
            inference_config.pop(key, None)
        else:
            inference_config[key] = value


def build_converse_kwargs(
    model: str, messages: list[dict], tools: Optional[list[dict]] = None, max_tokens: Optional[int] = 4096,
    temperature: Optional[float] = None, top_p: Optional[float] = None,
    stop_sequences: Optional[list[str]] = None, guardrail_config: Optional[dict] = None,
    reasoning_config: Optional[dict] = None, prompt_cache: bool = True,
) -> dict[str, Any]:
    """Build kwargs for ``bedrock-runtime.converse()`` / ``converse_stream()``. ``max_tokens=None`` omits
    ``maxTokens`` (model maximum; default stays 4096). cachePoint markers go on system, tools and the
    second-newest message (survives as the tail grows — mirrors Anthropic system_and_3), each only if the
    model supports caching and Bedrock has not rejected that placement. ``reasoning_config`` becomes the
    model's ``additionalModelRequestFields`` reasoning switch (Claude ``thinking``, Nova 2 ``reasoningConfig``)."""
    system_prompt, converse_messages = convert_messages_to_converse(messages, model=model)
    cache_at = {p for p in CACHE_POINT_PLACEMENTS if cache_point_allowed(model, p)} if prompt_cache and _model_supports_prompt_cache(model) else set()
    inference_config: dict[str, Any] = {} if max_tokens is None else {"maxTokens": max_tokens}
    kwargs: dict[str, Any] = {"modelId": model, "messages": converse_messages, "inferenceConfig": inference_config}
    if system_prompt:
        kwargs["system"] = system_prompt + [dict(_CACHE_POINT)] if "system" in cache_at else system_prompt
    from agent.anthropic_adapter import _forbids_sampling_params
    if not _forbids_sampling_params(model) and not _BEDROCK_XAI_GROK_NO_SAMPLING_RE.match(model or ""):
        inference_config.update({k: v for k, v in (("temperature", temperature), ("topP", top_p)) if v is not None})
    if stop_sequences:
        inference_config["stopSequences"] = stop_sequences
    converse_tools = convert_tools_to_converse(tools) if tools else []
    if converse_tools and not _model_supports_tool_use(model):
        # Non-tool-calling models reject toolConfig (ValidationException → retry loop): strip and warn.
        logger.warning(
            "Model %s does not support tool calling — tools stripped. "
            "The agent will operate in text-only mode.", model
        )
    elif converse_tools:
        kwargs["toolConfig"] = {"tools": converse_tools + [dict(_CACHE_POINT)] if "tools" in cache_at else converse_tools}
    if "messages" in cache_at and len(converse_messages) >= 2:
        content = converse_messages[-2].get("content")
        if isinstance(content, list) and content:
            content.append(dict(_CACHE_POINT))
    if guardrail_config:
        kwargs["guardrailConfig"] = guardrail_config
    if reasoning_config and isinstance(reasoning_config, dict):
        _apply_converse_reasoning(kwargs, inference_config, model, reasoning_config, max_tokens)
    if not inference_config:
        del kwargs["inferenceConfig"]  # optional on the wire; don't send {}
    return kwargs


def call_converse(
    region: str, model: str, messages: list[dict], tools: Optional[list[dict]] = None,
    max_tokens: Optional[int] = 4096, temperature: Optional[float] = None, top_p: Optional[float] = None,
    stop_sequences: Optional[list[str]] = None, guardrail_config: Optional[dict] = None,
) -> SimpleNamespace:
    """Non-streaming Converse call → OpenAI-compatible response. Retries once without a rejected cachePoint
    placement; evicts the cached client on stale-connection errors."""
    from agent.prompt_caching import is_one_shot_request
    client = _get_bedrock_runtime_client(region)
    # Auxiliary-only entry point: a one-shot's cachePoint is a write no later call reads.
    kwargs = build_converse_kwargs(model, messages, tools, max_tokens, temperature, top_p, stop_sequences, guardrail_config,
                                   prompt_cache=not is_one_shot_request(messages, tools))
    try:
        response = client.converse(**kwargs)
    except Exception as exc:
        retry_kwargs = recover_from_cache_point_rejection(exc, kwargs)
        if retry_kwargs is not None:
            return normalize_converse_response(client.converse(**retry_kwargs))
        redacted_retry_kwargs = recover_from_redacted_reasoning_rejection(exc, kwargs)
        if redacted_retry_kwargs is not None:
            return normalize_converse_response(client.converse(**redacted_retry_kwargs))
        if is_stale_connection_error(exc):
            logger.warning(
                "bedrock: stale-connection error on converse(region=%s, model=%s): "
                "%s — evicting cached client so the next call reconnects.", region, model, type(exc).__name__,
            )
            invalidate_runtime_client(region)
        raise
    return normalize_converse_response(response)


# --- Model discovery ---

_discovery_cache: dict[str, Any] = {}
_DISCOVERY_CACHE_TTL_SECONDS = 3600


def reset_discovery_cache():
    """Clear the model discovery cache. Used in tests."""
    _discovery_cache.clear()


def _model_entry(model_id: str, name: Any, provider: str, input_mods: list, output_mods: list) -> dict[str, Any]:
    return {"id": model_id, "name": (name or model_id).strip(), "provider": provider,
            "input_modalities": input_mods, "output_modalities": output_mods, "streaming": True}


def _list_foundation_models(client, filter_set: set, models: list[dict[str, Any]]) -> None:
    """Append active, streaming-capable, text-output foundation models (optionally provider-filtered)."""
    for summary in client.list_foundation_models().get("modelSummaries", []):
        model_id = (summary.get("modelId") or "").strip()
        if not model_id:
            continue
        provider_name = summary.get("providerName") or ""
        model_prefix = model_id.split(".")[0].lower() if "." in model_id else ""
        if filter_set and provider_name.lower() not in filter_set and model_prefix not in filter_set:
            continue
        output_mods = summary.get("outputModalities", [])
        if (summary.get("modelLifecycle", {}).get("status", "").upper() != "ACTIVE"
                or not summary.get("responseStreamingSupported", False) or "TEXT" not in output_mods):
            continue
        models.append(_model_entry(
            model_id, summary.get("modelName"), provider_name.strip(), summary.get("inputModalities", []), output_mods,
        ))


def _list_inference_profiles(client, filter_set: set, models: list[dict[str, Any]]) -> None:
    """Append active cross-region inference profiles whose IDs are not already present (paginated)."""
    profiles, next_token = [], None
    while True:
        response = client.list_inference_profiles(**({"nextToken": next_token} if next_token else {}))
        profiles.extend(response.get("inferenceProfileSummaries", []))
        if not (next_token := response.get("nextToken")):
            break
    seen_ids = {m["id"].lower() for m in models}
    for profile in profiles:
        profile_id = (profile.get("inferenceProfileId") or "").strip()
        if not profile_id or profile.get("status") != "ACTIVE" or profile_id.lower() in seen_ids:
            continue
        if filter_set and not any(
            _extract_provider_from_arn(m.get("modelArn", "")).lower() in filter_set for m in profile.get("models", [])
        ):
            continue
        models.append(_model_entry(profile_id, profile.get("inferenceProfileName"), "inference-profile", ["TEXT"], ["TEXT"]))
        seen_ids.add(profile_id.lower())


def discover_bedrock_models(region: str, provider_filter: Optional[list[str]] = None) -> list[dict[str, Any]]:
    """Foundation models + inference profiles (cached 1h per region/filter), ``global.`` profiles first then
    by name; [] when the client cannot be built."""
    # The list is account-scoped (whichever credentials the control client signs with), so a routed
    # profile gets its own entry; unscoped keeps the region:filter key byte-for-byte.
    from hermes_constants import get_hermes_home_override, hermes_home_key
    cache_key = f"{region}:{','.join(sorted(provider_filter or []))}"
    if get_hermes_home_override() is not None:
        cache_key = f"{hermes_home_key()}|{cache_key}"
    cached = _discovery_cache.get(cache_key)
    if cached and (time.time() - cached["timestamp"]) < _DISCOVERY_CACHE_TTL_SECONDS:
        return cached["models"]
    try:
        client = _get_bedrock_control_client(region)
    except Exception as e:
        logger.warning("Failed to create Bedrock client for model discovery: %s", e)
        return []
    models: list[dict[str, Any]] = []
    filter_set = {f.lower() for f in (provider_filter or [])}
    for step, log, message in (
        (_list_foundation_models, logger.warning, "Failed to list Bedrock foundation models: %s"),
        (_list_inference_profiles, logger.debug, "Skipping inference profile discovery: %s"),
    ):
        try:
            step(client, filter_set, models)
        except Exception as e:
            log(message, e)
    models.sort(key=lambda m: (0 if m["id"].startswith("global.") else 1, m["name"].lower()))
    _discovery_cache[cache_key] = {"timestamp": time.time(), "models": models}
    return models


def _extract_provider_from_arn(arn: str) -> str:
    """``arn:aws:bedrock:...:foundation-model/anthropic.claude-v2`` → ``"anthropic"``."""
    match = re.search(r"foundation-model/([^.]+)", arn)
    return match.group(1) if match else ""


# --- Bedrock model context lengths ---
# Static fallback when the live probe is unavailable (agent/model_metadata.py). Keys match by longest
# substring, so versioned entries win over the generic "anthropic.claude-opus-4".

BEDROCK_CONTEXT_LENGTHS: dict[str, int] = {
    # https://docs.aws.amazon.com/bedrock/latest/userguide/model-card-xai-grok-4-6.html
    "xai.grok-4.6": 500_000,
    # Anthropic Claude: 1M GA vs 200K. The 1M entries must match agent/model_metadata.py
    # DEFAULT_CONTEXT_LENGTHS or context compresses early — Opus 5 reached that table and not this
    # one, so the offline path resolved 128K for a 1M model (#74263); the pairing is now tested.
    **dict.fromkeys((
        "anthropic.claude-fable-5", "anthropic.claude-fable", "anthropic.claude-sonnet-5", "anthropic.claude-opus-5",
        "anthropic.claude-opus-4-8", "anthropic.claude-opus-4-7",
        "anthropic.claude-opus-4-6", "anthropic.claude-sonnet-4-6",
    ), 1_000_000),
    **dict.fromkeys((
        "anthropic.claude-sonnet-4-5", "anthropic.claude-haiku-4-5", "anthropic.claude-opus-4", "anthropic.claude-sonnet-4",
        "anthropic.claude-3-5-sonnet", "anthropic.claude-3-5-haiku", "anthropic.claude-3-opus", "anthropic.claude-3-sonnet",
        "anthropic.claude-3-haiku",
    ), 200_000),
    # Amazon Nova
    **dict.fromkeys(("amazon.nova-pro", "amazon.nova-lite"), 300_000), "amazon.nova-micro": 128_000,
    # Meta Llama / Mistral / DeepSeek
    **dict.fromkeys((
        "meta.llama4-maverick", "meta.llama4-scout", "meta.llama3-3-70b-instruct", "mistral.mistral-large", "deepseek.v3",
    ), 128_000),
    # OpenAI on Bedrock: docs.aws.amazon.com/bedrock/latest/userguide/model-cards-openai.html. GPT-6.1 Sol's
    # card states "1M tokens"; the other cards state 1,050,000. 272K is the long-context PRICING tier
    # boundary on these cards, not the window. Each entry stays <= model_metadata.DEFAULT_CONTEXT_LENGTHS.
    "openai.gpt-6.1-sol": 1_000_000,
    **dict.fromkeys((
        "openai.gpt-6-astra", "openai.gpt-6-sol", "openai.gpt-6-luna",
        "openai.gpt-5.5", "openai.gpt-5.6-sol", "openai.gpt-5.6-terra", "openai.gpt-5.6-luna",
    ), 1_050_000),
}

BEDROCK_DEFAULT_CONTEXT_LENGTH = 128_000  # unknown Bedrock models
_DEFAULT_WINDOW_WARNED: set = set()  # one fallback WARNING per model id per process

# Probe padding tiers (tokens): a wildly oversized payload yields an opaque InternalServerException
# instead of a clean ValidationException.
_BEDROCK_PROBE_TIERS = (1_300_000, 2_200_000)
_WORDS_PER_TOKEN = 0.9  # conservative: ensures the padded prompt clears the tier


def probe_bedrock_context_length(model_id: str, region: str) -> Optional[int]:
    """Discover a model's real context window by provoking a length error — the only authoritative source
    ("prompt is too long: 1300032 tokens > 1000000 maximum"); length validation runs before inference so the
    probe costs nothing. An accepted tier is a safe lower bound; None (no creds/network/unparseable) → static table."""
    from agent.model_metadata import parse_context_limit_from_error
    try:
        client = _get_bedrock_runtime_client(region)
    except Exception as exc:  # boto3 missing / credential resolution failure
        logger.debug("Bedrock context probe skipped for %s: %s", model_id, exc)
        return None
    last_error = ""
    for tier_tokens in _BEDROCK_PROBE_TIERS:
        oversized = "data " * int(tier_tokens / _WORDS_PER_TOKEN)
        try:
            client.converse(modelId=model_id, messages=[{"role": "user", "content": [{"text": oversized}]}],
                            inferenceConfig={"maxTokens": 8})
            logger.debug("Bedrock context probe for %s accepted ~%s-token prompt; "
                         "window is at least that", model_id, f"{tier_tokens:,}")
            return tier_tokens
        except Exception as exc:
            last_error = str(exc)
            limit = parse_context_limit_from_error(last_error)
            if limit and limit >= 1024:
                logger.info("Probed Bedrock context window for %s: %s tokens", model_id, f"{limit:,}")
                return limit
            # Opaque server error / auth / throttle at this tier — try the next.
    logger.debug("Bedrock context probe for %s returned no parseable limit: %s", model_id, last_error[:200])
    return None


def get_bedrock_context_length(model_id: str, region: str = "", probe: bool = True) -> int:
    """Context window: live probe (if ``probe`` and ``region``) → static table → default. The table is fallback
    only: a stale substring match silently caps the window (a 1M Opus pinned to 200K via "opus-4").
    An application-inference-profile ARN is first resolved to the model it wraps (#114476)."""
    profile_arn = model_id if _APPLICATION_PROFILE_ARN_RE.search(model_id) else ""
    if profile_arn:
        model_id = _resolve_inference_profile_model_id(profile_arn, region)
    if probe and region and (probed := probe_bedrock_context_length(model_id, region)):
        return probed
    matches = [key for key in BEDROCK_CONTEXT_LENGTHS if key in model_id.lower()]
    if matches:
        return BEDROCK_CONTEXT_LENGTHS[max(matches, key=len)]
    if profile_arn:
        logger.warning(
            "Bedrock inference profile %s resolved no known model window; using the %s default. "
            "Grant bedrock:GetInferenceProfile or set model.context_length explicitly if the "
            "wrapped model has a larger window.",
            profile_arn,
            f"{BEDROCK_DEFAULT_CONTEXT_LENGTH:,}",
        )
    elif model_id not in _DEFAULT_WINDOW_WARNED:
        _DEFAULT_WINDOW_WARNED.add(model_id)
        logger.warning(
            "Bedrock model %s has no known context window; using the static %s-token fallback "
            "(not a detected limit). Set model.context_length in config.yaml if the model's window is larger.",
            model_id, f"{BEDROCK_DEFAULT_CONTEXT_LENGTH:,}",
        )
    return BEDROCK_DEFAULT_CONTEXT_LENGTH


# An application-inference-profile ARN (cost-allocation wrapper) carries an opaque id, so the probe
# error text and the static substring table both miss and the 128k default silently applies
# (#114476). System-defined `inference-profile/us.anthropic...` ARNs embed the model id and need no
# lookup. The ARN's own region (field 4) is authoritative for the control-plane call: the runtime
# region / base_url may differ, and an empty region must not skip the lookup because the
# production caller (agent/model_metadata.py::_resolve_bedrock_context_length) passes none.
_APPLICATION_PROFILE_ARN_RE = re.compile(r":application-inference-profile/")
_ARN_REGION_RE = re.compile(r"^arn:[^:]+:bedrock:([a-z0-9-]+):", re.IGNORECASE)
_inference_profile_model_cache: dict[str, str] = {}


def _resolve_inference_profile_model_id(profile_arn: str, region: str = "") -> str:
    """Application-profile ARN → the wrapped model's ARN (its ``foundation-model/<id>`` tail satisfies the
    static-table substring match); the profile ARN itself when ``bedrock:GetInferenceProfile`` is not
    granted or unavailable, so callers keep the default-window behaviour. Both outcomes are cached per
    process: this runs on every context-length resolution, not once per model."""
    if profile_arn in _inference_profile_model_cache:
        return _inference_profile_model_cache[profile_arn]
    arn_region = _ARN_REGION_RE.match(profile_arn)
    region = (arn_region.group(1) if arn_region else "") or region or resolve_bedrock_region()
    resolved = profile_arn
    try:
        client = _get_bedrock_control_client(region)
        models = client.get_inference_profile(inferenceProfileIdentifier=profile_arn).get("models") or []
        resolved = next((m["modelArn"] for m in models if m.get("modelArn")), profile_arn)
    except Exception as exc:  # no boto3 / credentials / GetInferenceProfile not granted
        logger.debug("Inference profile resolution skipped for %s: %s", profile_arn, exc)
    _inference_profile_model_cache[profile_arn] = resolved
    return resolved