From ba8b7a4e0d85307793d8fc258fc8b21546220882 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 11:21:11 -0700 Subject: [PATCH] refactor(agent/adapters): simplify bedrock adapter (-639 LOC) Drop dead call_converse_stream / classify_bedrock_error / is_context_overflow_error (zero refs) and their tests; stop-reason mapping becomes a dict; extract _cache_point/_assistant_blocks/_append_turn/_cached_client helpers; compact incident narratives to their invariants. Converse wire output byte-identical. --- agent/bedrock_adapter.py | 1847 +++++++++------------------ tests/agent/test_bedrock_adapter.py | 123 +- 2 files changed, 605 insertions(+), 1365 deletions(-) diff --git a/agent/bedrock_adapter.py b/agent/bedrock_adapter.py index 88235c1281..b62b231ffb 100644 --- a/agent/bedrock_adapter.py +++ b/agent/bedrock_adapter.py @@ -1,30 +1,14 @@ """AWS Bedrock Converse API adapter for Hermes Agent. -Provides native integration with Amazon Bedrock using the Converse API, -bypassing the OpenAI-compatible endpoint in favor of direct AWS SDK calls. -This enables full access to the Bedrock ecosystem: +Talks to Bedrock through the native Converse API (boto3) instead of an +OpenAI-compatible endpoint, so the AWS credential chain (IAM roles, SSO +profiles, env vars, instance metadata), cross-region inference profiles, +guardrails and control-plane model discovery all work without API keys. - - **Native Converse API**: Unified interface for all Bedrock models - (Claude, Nova, Llama, Mistral, etc.) with streaming support. - - **AWS credential chain**: IAM roles, SSO profiles, environment variables, - instance metadata — zero API key management for AWS-native environments. - - **Dynamic model discovery**: Auto-discovers available foundation models - and cross-region inference profiles via the Bedrock control plane. - - **Guardrails support**: Optional Bedrock Guardrails configuration for - content filtering and safety policies. - - **Inference profiles**: Supports cross-region inference profiles - (us.anthropic.claude-*, global.anthropic.claude-*) for better capacity - and automatic failover. - -Architecture follows the same pattern as ``anthropic_adapter.py``: - - All Bedrock-specific logic is isolated in this module. - - Messages/tools are converted between OpenAI format and Converse format. - - Responses are normalized back to OpenAI-compatible objects for the agent loop. - -Reference: OpenClaw's ``extensions/amazon-bedrock/`` plugin, which implements -the same Converse API integration in TypeScript via ``@aws-sdk/client-bedrock``. - -Requires: ``boto3`` (optional dependency — only needed when using the Bedrock provider). +Same layout as ``anthropic_adapter.py``: all Bedrock-specific logic lives +here; OpenAI-format messages/tools are converted to Converse format on the +way in and responses are normalized back to OpenAI-shaped objects for the +agent loop. Requires ``boto3`` (optional dependency). """ import base64 @@ -40,12 +24,8 @@ import httpx logger = logging.getLogger(__name__) -# --------------------------------------------------------------------------- -# Ensure boto3/botocore are installed before any code in this module runs. -# Upstream removed boto3 from [all] extras (PRs #24220, #24515); lazy_deps -# handles on-demand installation so the Bedrock provider still works in the -# EKS deployment without baking boto3 into the base image. -# --------------------------------------------------------------------------- +# boto3 is not in the [all] extras; lazy_deps installs it on demand so the +# provider works without baking boto3 into the base image. try: from tools.lazy_deps import ensure ensure("provider.bedrock", prompt=False) @@ -53,39 +33,31 @@ except Exception: pass # lazy_deps unavailable or install failed — let downstream imports surface the real error -# --------------------------------------------------------------------------- -# Lazy boto3 import — only loaded when the Bedrock provider is actually used. -# This keeps startup fast for users who don't use Bedrock. -# --------------------------------------------------------------------------- - _bedrock_runtime_client_cache: Dict[str, Any] = {} _bedrock_control_client_cache: Dict[str, Any] = {} -# Bedrock-hosted OpenAI GPT-5.5 is not exposed through the native Converse -# runtime. AWS serves it from the Bedrock Mantle OpenAI-compatible Responses -# endpoint instead (https://bedrock-mantle..api.aws/openai/v1). -# Keep the allowlist intentionally narrow so OpenAI GPT-OSS models that are -# Converse-capable continue to use the native Bedrock path. +# Bedrock-hosted OpenAI GPT-5.x models are not exposed through the Converse +# runtime; AWS serves them from the Bedrock Mantle OpenAI-compatible Responses +# endpoint (https://bedrock-mantle..api.aws/openai/v1). Keep the +# allowlist narrow so Converse-capable GPT-OSS models stay on the native path. BEDROCK_OPENAI_RESPONSES_MODEL_IDS: Tuple[str, ...] = ( "openai.gpt-5.5", - # GPT-5.6 family (GA on Bedrock 2026-07-13): Sol (frontier), Terra - # (balanced), Luna (fast/affordable). All are Mantle-only — the model - # cards list bedrock-runtime/Converse as unsupported. - # https://docs.aws.amazon.com/bedrock/latest/userguide/model-cards-openai.html "openai.gpt-5.6-sol", "openai.gpt-5.6-terra", "openai.gpt-5.6-luna", ) -_BEDROCK_OPENAI_HOST_RE = re.compile( - r"^bedrock-mantle\.([a-z0-9-]+)\.api\.aws$", re.IGNORECASE -) +_BEDROCK_OPENAI_HOST_RE = re.compile(r"^bedrock-mantle\.([a-z0-9-]+)\.api\.aws$", re.IGNORECASE) _MIN_BOTO3_VERSION = (1, 34, 59) def _require_boto3(): - """Import boto3, raising a clear error if not installed or too old.""" + """Import boto3, raising a clear error if not installed or too old. + + converse()/converse_stream() need boto3 >= 1.34.59; a system boto3 (e.g. + Ubuntu's 1.34.46) can shadow the venv pin under editable installs. + """ try: import boto3 except ImportError: @@ -94,10 +66,6 @@ def _require_boto3(): "Install it with: pip install boto3\n" "Or install Hermes with Bedrock support: pip install -e '.[bedrock]'" ) - # converse() / converse_stream() were added in boto3 1.34.59. - # When Hermes is installed editable into system Python, the system boto3 - # (e.g. Ubuntu 24.04 ships 1.34.46) may take precedence over the venv - # version pinned in pyproject.toml. try: version = tuple(int(x) for x in boto3.__version__.split(".")[:3]) except (AttributeError, ValueError): @@ -111,27 +79,19 @@ def _require_boto3(): return boto3 -def _get_bedrock_runtime_client(region: str): - """Get or create a cached ``bedrock-runtime`` client for the given region. +def _cached_client(cache: Dict[str, Any], service: str, region: str): + """Get or create a per-region boto3 client using the default credential chain.""" + if region not in cache: + cache[region] = _require_boto3().client(service, region_name=region) + return cache[region] - Uses the default AWS credential chain (env vars → profile → instance role). - """ - if region not in _bedrock_runtime_client_cache: - boto3 = _require_boto3() - _bedrock_runtime_client_cache[region] = boto3.client( - "bedrock-runtime", region_name=region, - ) - return _bedrock_runtime_client_cache[region] + +def _get_bedrock_runtime_client(region: str): + return _cached_client(_bedrock_runtime_client_cache, "bedrock-runtime", region) def _get_bedrock_control_client(region: str): - """Get or create a cached ``bedrock`` control-plane client for model discovery.""" - if region not in _bedrock_control_client_cache: - boto3 = _require_boto3() - _bedrock_control_client_cache[region] = boto3.client( - "bedrock", region_name=region, - ) - return _bedrock_control_client_cache[region] + return _cached_client(_bedrock_control_client_cache, "bedrock", region) def reset_client_cache(): @@ -141,44 +101,26 @@ def reset_client_cache(): def invalidate_runtime_client(region: str) -> bool: - """Evict the cached ``bedrock-runtime`` client for a single region. + """Evict one region's cached ``bedrock-runtime`` client (stale HTTP pool). - Per-region counterpart to :func:`reset_client_cache`. Used by the converse - call wrappers to discard clients whose underlying HTTP connection has - gone stale, so the next call allocates a fresh client (with a fresh - connection pool) instead of reusing a dead socket. - - Returns True if a cached entry was evicted, False if the region was not - cached. + Returns True if an entry was evicted. """ existed = region in _bedrock_runtime_client_cache _bedrock_runtime_client_cache.pop(region, None) return existed -# --------------------------------------------------------------------------- -# Bedrock Mantle / OpenAI Responses support -# --------------------------------------------------------------------------- +# --- Bedrock Mantle / OpenAI Responses support --- def is_openai_bedrock_model(model_id: str) -> bool: - """Return True for Bedrock-hosted OpenAI models that require Mantle. - - Bedrock's GPT-OSS models are Converse-capable and intentionally do not - match this helper. The allowlist tracks models served by the OpenAI - Responses-compatible ``bedrock-mantle`` route. - """ + """True for Bedrock-hosted OpenAI models that require Mantle (GPT-OSS excluded).""" normalized = str(model_id or "").strip().lower() return normalized in {m.lower() for m in BEDROCK_OPENAI_RESPONSES_MODEL_IDS} def merge_bedrock_openai_model_ids(model_ids: List[str]) -> List[str]: - """Append Bedrock OpenAI Responses models to a discovered Bedrock list. - - The Bedrock control plane's ListFoundationModels/ListInferenceProfiles - discovery covers Converse models but does not enumerate Mantle-only - OpenAI Responses models. The picker needs both surfaces under AWS Bedrock. - """ + """Append Mantle-only OpenAI models, which control-plane discovery never lists.""" merged = list(model_ids or []) seen = {str(m).lower() for m in merged} for model_id in BEDROCK_OPENAI_RESPONSES_MODEL_IDS: @@ -202,13 +144,10 @@ def bedrock_openai_region_from_base_url(base_url: str) -> Optional[str]: def is_bedrock_openai_base_url(base_url: str) -> bool: - """Return True for Bedrock Mantle OpenAI-compatible endpoints.""" + """True for Bedrock Mantle endpoints (bare host or /openai[/v1] path).""" parsed = urlparse(str(base_url or "")) - host = parsed.hostname or "" - if not _BEDROCK_OPENAI_HOST_RE.match(host): + if not _BEDROCK_OPENAI_HOST_RE.match(parsed.hostname or ""): return False - # The OpenAI GPT-5.5 Bedrock route lives under /openai/v1. Accept a bare - # host too so callers can normalize before appending the path. path = (parsed.path or "").rstrip("/").lower() return path in {"", "/openai", "/openai/v1"} @@ -240,10 +179,8 @@ class BedrockOpenAISigV4Auth(httpx.Auth): "Configure AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY, AWS_PROFILE, " "SSO, or an instance/task role." ) - frozen = credentials.get_frozen_credentials() - # Drop the OpenAI SDK's placeholder bearer header before signing; SigV4 - # must own Authorization. Keep all other SDK headers so AWS receives - # content-type, accept, request IDs, etc. + # SigV4 must own Authorization: drop the SDK's placeholder bearer header + # but keep every other SDK header (content-type, accept, request IDs). headers = { str(k): str(v) for k, v in request.headers.items() @@ -255,15 +192,13 @@ class BedrockOpenAISigV4Auth(httpx.Auth): data=request.content or b"", headers=headers, ) - SigV4Auth(frozen, self.service, self.region).add_auth(aws_request) + 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.""" - import httpx - kwargs: Dict[str, Any] = {"auth": BedrockOpenAISigV4Auth(region)} if isinstance(timeout, (int, float)) and not isinstance(timeout, bool) and timeout > 0: kwargs["timeout"] = timeout @@ -271,15 +206,13 @@ def build_bedrock_openai_http_client(region: str, *, timeout: Optional[float] = def configure_bedrock_openai_client_kwargs( - client_kwargs: Dict[str, Any], - *, - timeout: Optional[float] = None, + client_kwargs: Dict[str, Any], *, timeout: Optional[float] = None, ) -> Dict[str, Any]: """Install SigV4 auth on OpenAI SDK kwargs for Bedrock Mantle. - ``AWS_BEARER_TOKEN_BEDROCK``/explicit Bedrock API keys continue to use the - SDK's normal bearer auth. The special ``aws-sdk`` placeholder means IAM - credential-chain auth, so we attach a per-request SigV4 httpx client. + Real API keys (``AWS_BEARER_TOKEN_BEDROCK``) keep the SDK's bearer auth; + the ``aws-sdk``/``no-key-required`` placeholders mean IAM credential-chain + auth, so we attach a per-request SigV4 httpx client. """ base_url = str(client_kwargs.get("base_url") or "") if not is_bedrock_openai_base_url(base_url): @@ -293,121 +226,67 @@ def configure_bedrock_openai_client_kwargs( return client_kwargs -# --------------------------------------------------------------------------- -# Stale-connection detection -# --------------------------------------------------------------------------- -# -# boto3 caches its HTTPS connection pool inside the client object. When a -# pooled connection is killed out from under us (NAT timeout, VPN flap, -# server-side TCP RST, proxy idle cull, etc.), the next use surfaces as -# one of a handful of low-level exceptions — most commonly -# ``botocore.exceptions.ConnectionClosedError`` or -# ``urllib3.exceptions.ProtocolError``. urllib3 also trips an internal -# ``assert`` in a couple of paths (connection pool state checks, chunked -# response readers) which bubbles up as a bare ``AssertionError`` with an -# empty ``str(exc)``. -# -# In all of these cases the client is the problem, not the request: retrying -# with the same cached client reproduces the failure until the process -# restarts. The fix is to evict the region's cached client so the next -# attempt builds a new one. +# --- Stale-connection detection --- +# boto3 caches its HTTPS pool inside the client. A pooled connection killed +# out from under us (NAT timeout, VPN flap, RST, proxy idle cull) surfaces as +# botocore ConnectionClosedError / urllib3 ProtocolError, or as a bare +# AssertionError from urllib3's internal pool-state checks. Retrying with the +# same client reproduces the failure, so the fix is to evict the client. -_STALE_LIB_MODULE_PREFIXES = ( - "urllib3.", - "botocore.", - "boto3.", -) +_STALE_LIB_MODULE_PREFIXES = ("urllib3.", "botocore.", "boto3.") def _traceback_frames_modules(exc: BaseException): - """Yield ``__name__``-style module strings for each frame in exc's traceback.""" + """Yield the ``__name__`` of each frame in exc's traceback.""" tb = getattr(exc, "__traceback__", None) while tb is not None: - frame = tb.tb_frame - module = frame.f_globals.get("__name__", "") - yield module or "" + yield tb.tb_frame.f_globals.get("__name__", "") or "" tb = tb.tb_next -def is_stale_connection_error(exc: BaseException) -> bool: - """Return True if ``exc`` indicates a dead/stale Bedrock HTTP connection. - - Matches: - * ``botocore.exceptions.ConnectionError`` and subclasses - (``ConnectionClosedError``, ``EndpointConnectionError``, - ``ReadTimeoutError``, ``ConnectTimeoutError``). - * ``urllib3.exceptions.ProtocolError`` / ``NewConnectionError`` / - ``ConnectionError`` (best-effort import — urllib3 is a transitive - dependency of botocore so it is always available in practice). - * Bare ``AssertionError`` raised from a frame inside urllib3, botocore, - or boto3. These are internal-invariant failures (typically triggered - by corrupted connection-pool state after a dropped socket) and are - recoverable by swapping the client. - - Non-library ``AssertionError``s (from application code or tests) are - intentionally not matched — only library-internal asserts signal stale - connection state. - """ - # botocore: the canonical signal — HTTPClientError is the umbrella for - # ConnectionClosedError, ReadTimeoutError, EndpointConnectionError, - # ConnectTimeoutError, and ProxyConnectionError. ConnectionError covers - # the same family via a different branch of the hierarchy. +def _stale_error_types() -> tuple: + """botocore + urllib3 transport-failure exception classes (best-effort import).""" + types: list = [] try: - from botocore.exceptions import ( - ConnectionError as BotoConnectionError, - HTTPClientError, - ) - botocore_errors: tuple = (BotoConnectionError, HTTPClientError) + from botocore.exceptions import ConnectionError as BotoConnectionError, HTTPClientError + types += [BotoConnectionError, HTTPClientError] except ImportError: # pragma: no cover — botocore always present with boto3 - botocore_errors = () - if botocore_errors and isinstance(exc, botocore_errors): - return True - - # urllib3: low-level transport failures + pass try: - from urllib3.exceptions import ( - ProtocolError, - NewConnectionError, - ConnectionError as Urllib3ConnectionError, - ) - urllib3_errors = (ProtocolError, NewConnectionError, Urllib3ConnectionError) + from urllib3.exceptions import ConnectionError as Urllib3ConnectionError, NewConnectionError, ProtocolError + types += [ProtocolError, NewConnectionError, Urllib3ConnectionError] except ImportError: # pragma: no cover - urllib3_errors = () - if urllib3_errors and isinstance(exc, urllib3_errors): - return True + pass + return tuple(types) - # Library-internal AssertionError (urllib3 / botocore / boto3) + +def is_stale_connection_error(exc: BaseException) -> bool: + """True if ``exc`` indicates a dead/stale Bedrock HTTP connection. + + Matches botocore ConnectionError/HTTPClientError subclasses, urllib3 + transport errors, and bare ``AssertionError``s raised from inside + urllib3/botocore/boto3. Application-level AssertionErrors are NOT matched. + """ + if isinstance(exc, _stale_error_types()): + return True if isinstance(exc, AssertionError): for module in _traceback_frames_modules(exc): - if any(module.startswith(prefix) for prefix in _STALE_LIB_MODULE_PREFIXES): + if module.startswith(_STALE_LIB_MODULE_PREFIXES): return True - return False def is_streaming_access_denied_error(exc: BaseException) -> bool: - """Return True when AWS denied the ``bedrock:InvokeModelWithResponseStream`` action. + """True when IAM denied ``bedrock:InvokeModelWithResponseStream``. - IAM policies scoped to ``bedrock:InvokeModel`` only (a common least-privilege - setup) reject ``converse_stream()`` with an ``AccessDeniedException`` whose - message names the streaming action, e.g.:: - - User: arn:aws:iam::123456789012:user/x is not authorized to perform: - bedrock:InvokeModelWithResponseStream on resource: ... - - This is permanent for the session — retrying the stream can never succeed — - so callers should flip to the non-streaming ``converse()`` path (which maps - to ``bedrock:InvokeModel``) instead of burning retries. - - Detection is deliberately message-based: boto3 surfaces this as a - ``ClientError`` with ``Error.Code == "AccessDeniedException"``, and the - AnthropicBedrock SDK wraps the same AWS response in its own exception - types, but both preserve the action name in the message. + InvokeModel-only policies reject converse_stream() permanently, so callers + should fall back to non-streaming converse() instead of retrying. Detection + is message-based because the AnthropicBedrock SDK wraps the same AWS + response in its own exception types but preserves the action name. """ msg = str(exc).lower() if "invokemodelwithresponsestream" not in msg: return False - # ClientError with an explicit access-denied code is the canonical form. try: from botocore.exceptions import ClientError except ImportError: # pragma: no cover — botocore always present with boto3 @@ -415,123 +294,60 @@ def is_streaming_access_denied_error(exc: BaseException) -> bool: if ClientError is not None and isinstance(exc, ClientError): code = (getattr(exc, "response", None) or {}).get("Error", {}).get("Code", "") return code in ("AccessDeniedException", "UnauthorizedException") - # Wrapped forms (e.g. AnthropicBedrock SDK PermissionDeniedError) — match - # on the authorization-failure phrasing AWS uses. return "not authorized" in msg or "accessdenied" in msg -# --------------------------------------------------------------------------- -# AWS credential detection -# --------------------------------------------------------------------------- +# --- AWS credential detection --- -# Priority order matches OpenClaw's resolveAwsSdkEnvVarName(): -# 1. AWS_BEARER_TOKEN_BEDROCK (Bedrock-specific bearer token) -# 2. AWS_ACCESS_KEY_ID + AWS_SECRET_ACCESS_KEY (explicit IAM credentials) -# 3. AWS_PROFILE (named profile → SSO, assume-role, etc.) -# 4. Implicit: instance role, ECS task role, Lambda execution role -_AWS_CREDENTIAL_ENV_VARS = [ - "AWS_BEARER_TOKEN_BEDROCK", - "AWS_ACCESS_KEY_ID", - "AWS_PROFILE", - # These are checked by boto3's default chain but we list them for - # has_aws_credentials() detection: - "AWS_CONTAINER_CREDENTIALS_RELATIVE_URI", - "AWS_WEB_IDENTITY_TOKEN_FILE", -] +# 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 resolve_aws_auth_env_var(env: Optional[Dict[str, str]] = None) -> Optional[str]: - """Return the name of the AWS auth source that is active, or None. - - Checks environment variables first, then falls back to boto3's credential - chain for implicit sources (EC2 IMDS, ECS task role, etc.). - - This mirrors OpenClaw's ``resolveAwsSdkEnvVarName()`` — used to detect - whether the user has any AWS credentials configured without actually - attempting to authenticate. - """ - env = env if env is not None else os.environ - # Bearer token takes highest priority - if env.get("AWS_BEARER_TOKEN_BEDROCK", "").strip(): - return "AWS_BEARER_TOKEN_BEDROCK" - # Explicit access key pair - if (env.get("AWS_ACCESS_KEY_ID", "").strip() - and env.get("AWS_SECRET_ACCESS_KEY", "").strip()): - return "AWS_ACCESS_KEY_ID" - # Named profile (SSO, assume-role, etc.) - if env.get("AWS_PROFILE", "").strip(): - return "AWS_PROFILE" - # Container credentials (ECS, CodeBuild) - if env.get("AWS_CONTAINER_CREDENTIALS_RELATIVE_URI", "").strip(): - return "AWS_CONTAINER_CREDENTIALS_RELATIVE_URI" - # Web identity (EKS IRSA) - if env.get("AWS_WEB_IDENTITY_TOKEN_FILE", "").strip(): - return "AWS_WEB_IDENTITY_TOKEN_FILE" - # No env vars — check if boto3 can resolve credentials via IMDS or other - # implicit sources (EC2 instance role, ECS task role, Lambda, etc.) +def _boto3_chain_has_credentials() -> bool: + """True if boto3's default chain resolves credentials (IMDS, task role, ...).""" try: import botocore.session - session = botocore.session.get_session() - credentials = session.get_credentials() + credentials = botocore.session.get_session().get_credentials() if credentials is not None: resolved = credentials.get_frozen_credentials() - if resolved and resolved.access_key: - return "iam-role" - except Exception: - pass - return None - - -def has_aws_credentials(env: Optional[Dict[str, str]] = None) -> bool: - """Return True if any AWS credential source is detected. - - Checks environment variables first (fast, no I/O), then falls back to - boto3's credential chain which covers EC2 instance roles, ECS task roles, - Lambda execution roles, and other IMDS-based sources that don't set - environment variables. - - 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. - """ - if resolve_aws_auth_env_var(env) is not None: - return True - # Fall back to boto3's credential resolver — this covers EC2 instance - # metadata (IMDS), ECS container credentials, and other implicit sources - # that don't set environment variables. - try: - import botocore.session - session = botocore.session.get_session() - credentials = session.get_credentials() - if credentials is not None: - resolved = credentials.get_frozen_credentials() - if resolved and resolved.access_key: - return True + return bool(resolved and resolved.access_key) except Exception: pass return False -def resolve_bedrock_region(env: Optional[Dict[str, str]] = None) -> str: - """Resolve the AWS region for Bedrock API calls. +def resolve_aws_auth_env_var(env: Optional[Dict[str, str]] = None) -> Optional[str]: + """Return the name of the active AWS auth source, or None. - Priority: - 1. AWS_REGION env var - 2. AWS_DEFAULT_REGION env var - 3. boto3/botocore configured region (from ~/.aws/config or SSO profile) - 4. us-east-1 (hard fallback) - - The boto3 fallback is critical for EU/AP users who configure their region - in ~/.aws/config via a named profile rather than env vars — without it, - live model discovery would always return us.* profile IDs regardless of - the user's actual region. + Env vars first (fast, no I/O), then ``"iam-role"`` if boto3's chain + resolves implicit credentials (EC2 IMDS, ECS task role, Lambda). """ env = env if env is not None else os.environ - explicit = ( - env.get("AWS_REGION", "").strip() - or env.get("AWS_DEFAULT_REGION", "").strip() - ) + 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() else None + + +def has_aws_credentials(env: Optional[Dict[str, str]] = None) -> bool: + """True if any AWS credential source (env vars or boto3 chain) is detected.""" + return resolve_aws_auth_env_var(env) is not None or _boto3_chain_has_credentials() + + +def resolve_bedrock_region(env: Optional[Dict[str, str]] = None) -> str: + """AWS_REGION → AWS_DEFAULT_REGION → botocore configured region → us-east-1. + + The botocore fallback matters for EU/AP users who set the region in + ~/.aws/config via a named profile; without it discovery returns us.* IDs. + """ + 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 try: @@ -545,21 +361,12 @@ def resolve_bedrock_region(env: Optional[Dict[str, str]] = None) -> str: def resolve_bedrock_runtime_region(config: Optional[Dict[str, Any]] = None) -> str: - """Resolve the Bedrock region with the same priority as the main runtime. + """``bedrock.region`` from config.yaml, else :func:`resolve_bedrock_region`. - Priority (matches the runtime provider resolver in - ``hermes_cli/runtime_provider.py``): - 1. ``bedrock.region`` in config.yaml - 2. ``resolve_bedrock_region()`` (AWS_REGION / AWS_DEFAULT_REGION / - botocore profile / us-east-1) - - Callers that already hold a loaded config dict should pass it to avoid a - disk read; when *config* is None the config is loaded read-only. Every - non-runtime call site that constructs a Bedrock endpoint (auxiliary - client resolution, model discovery for the picker) must use this helper — - using bare ``resolve_bedrock_region()`` there lets auxiliary calls leave - the primary runtime's configured region when ``bedrock.region`` and the - ambient AWS env/profile disagree. + Matches ``hermes_cli/runtime_provider.py``. Every non-runtime Bedrock + endpoint (auxiliary clients, picker discovery) must use this, otherwise + auxiliary calls can leave the primary runtime's region when the config + and ambient AWS env/profile disagree. Pass *config* to avoid a disk read. """ if config is None: try: @@ -567,24 +374,12 @@ def resolve_bedrock_runtime_region(config: Optional[Dict[str, Any]] = None) -> s config = load_config_readonly() except Exception: config = {} - bedrock_cfg = (config or {}).get("bedrock") or {} - cfg_region = str(bedrock_cfg.get("region") or "").strip() - if cfg_region: - return cfg_region - return resolve_bedrock_region() + cfg_region = str(((config or {}).get("bedrock") or {}).get("region") or "").strip() + return cfg_region or resolve_bedrock_region() def bedrock_model_ids_or_none() -> Optional[List[str]]: - """Live-discover Bedrock model IDs for the active region. - - Returns a list of model ID strings if discovery succeeds and yields - at least one model, or ``None`` on failure / empty result. Callers - should fall back to the static curated list when ``None`` is returned. - - This helper consolidates the discover → extract-ids → fallback - pattern that was previously duplicated across ``provider_model_ids``, - ``list_authenticated_providers`` section 2, and section 3. - """ + """Live-discover Bedrock model IDs; None on failure/empty so callers use the static list.""" try: discovered = discover_bedrock_models(resolve_bedrock_runtime_region()) if discovered: @@ -594,45 +389,29 @@ def bedrock_model_ids_or_none() -> Optional[List[str]]: return None -# --------------------------------------------------------------------------- -# Tool-calling capability detection -# --------------------------------------------------------------------------- -# Some Bedrock models don't support tool/function calling. Sending toolConfig -# to these models causes ValidationException. We maintain a denylist of known -# non-tool-calling model patterns and strip tools for them. -# -# This is a conservative approach: unknown models are assumed to support tools. -# If a model fails with a tool-related ValidationException, add it here. +# --- Tool-calling / prompt-cache capability detection --- +# Models known to reject toolConfig with a ValidationException. Conservative: +# unknown models are assumed to support tools; add here when one fails. _NON_TOOL_CALLING_PATTERNS = [ - "deepseek.r1", # DeepSeek R1 — reasoning only, no tool support + "deepseek.r1", # DeepSeek R1 — reasoning only "deepseek-r1", # Alternate ID format - "stability.", # Image generation models - "cohere.embed", # Embedding models - "amazon.titan-embed", # Embedding models + "stability.", # Image generation + "cohere.embed", # Embeddings + "amazon.titan-embed", # Embeddings ] def _model_supports_tool_use(model_id: str) -> bool: - """Return True if the model is expected to support tool/function calling. - - Models in the denylist are known to reject toolConfig in the Converse API. - Unknown models default to True (assume tool support). - """ + """False for denylisted models; unknown models default to True.""" model_lower = model_id.lower() return not any(pattern in model_lower for pattern in _NON_TOOL_CALLING_PATTERNS) -# --------------------------------------------------------------------------- -# Prompt-cache capability detection (Converse API cachePoint) -# --------------------------------------------------------------------------- -# Claude on Bedrock already gets prompt caching through the AnthropicBedrock -# SDK path (see is_anthropic_bedrock_model / runtime_provider.py's dual-path -# routing) — it never reaches build_converse_kwargs unless bearer-token auth -# forces the Converse path (#28156). This allowlist covers the Converse API -# itself: sending an unsupported model a cachePoint block raises a -# ValidationException, so — like _model_supports_tool_use but inverted — -# unknown models default to NOT receiving cache markers until confirmed. +# Converse API cachePoint allowlist — inverted policy vs tools: an unsupported +# model rejects cachePoint with a ValidationException, so unknown models get +# NO cache markers until confirmed. Claude normally uses the AnthropicBedrock +# SDK path and only reaches build_converse_kwargs under bearer-token auth. # Ref: https://docs.aws.amazon.com/bedrock/latest/userguide/prompt-caching.html _CACHE_POINT_PATTERNS = [ "anthropic.claude", # bearer-token fallback path @@ -641,26 +420,18 @@ _CACHE_POINT_PATTERNS = [ def _model_supports_prompt_cache(model_id: str) -> bool: - """Return True if the model accepts a Converse API cachePoint block.""" + """True if the model accepts a Converse API cachePoint block.""" model_lower = model_id.lower() return any(pattern in model_lower for pattern in _CACHE_POINT_PATTERNS) -# --------------------------------------------------------------------------- -# Server-verdict cachePoint suppression -# --------------------------------------------------------------------------- -# The allowlist above is a static guess about *placement*, and Bedrock's real -# rule is per-model-family AND per-field: Amazon Nova accepts cachePoint in -# ``system``/``messages`` but rejects it inside ``toolConfig.tools`` with a -# hard ValidationException that fails the whole request (#97281). Any static -# table drifts the moment AWS ships a family whose placement rules differ, and -# the failure mode is 100% of turns with no recovery and no user workaround. -# -# So the table is not the only authority: when Bedrock names a placement as -# unpermitted, that verdict is recorded and the marker is dropped from that -# placement for the rest of the process, and the rejected request is retried -# once without it. Mirrors the existing self-heal idiom in this module -# (is_streaming_access_denied_error → non-streaming converse()). +# --- Server-verdict cachePoint suppression --- +# Bedrock's real 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 as AWS ships new families — failing 100% of turns with no workaround. +# So when Bedrock names a placement as unpermitted we record that verdict, +# drop the marker from that placement for the rest of the process, and retry +# the rejected request once without it. CACHE_POINT_PLACEMENTS = ("tools", "system", "messages") @@ -668,24 +439,17 @@ CACHE_POINT_PLACEMENTS = ("tools", "system", "messages") _CACHE_POINT_REJECTIONS: Dict[str, set] = {} # "#/toolConfig/tools/18: extraneous key [cachePoint] is not permitted" -_CACHE_POINT_PATH_PATTERN = re.compile( - r"#/(?P[A-Za-z0-9_./\[\]-]*)", re.IGNORECASE -) +_CACHE_POINT_PATH_PATTERN = re.compile(r"#/(?P[A-Za-z0-9_./\[\]-]*)", re.IGNORECASE) def cache_point_rejection_placement(exc: BaseException) -> Optional[str]: - """Return the Converse section whose cachePoint block Bedrock refused. + """Return the Converse section whose cachePoint Bedrock refused, or None. - Returns one of ``CACHE_POINT_PLACEMENTS``, or None when the error is not a - cachePoint rejection. Bedrock reports it as a ValidationException naming - the offending JSON pointer, e.g.:: - - Malformed input request: #/toolConfig/tools/18: extraneous key - [cachePoint] is not permitted, please reformat your input and try again. - - Detection is message-based on purpose: the pointer is the only part of the - response that says *which* section was rejected, and the same wording - reaches us both as a raw botocore ``ClientError`` and wrapped by SDKs. + Message-based on purpose: the JSON pointer in the ValidationException is + the only thing that says *which* section was rejected, and the same + wording arrives both as a raw botocore ClientError and wrapped by SDKs. + An unlocalisable rejection maps to "tools" — the only placement any + supported family is known to refuse while accepting the others. """ msg = str(exc) lowered = msg.lower() @@ -701,9 +465,6 @@ def cache_point_rejection_placement(exc: BaseException) -> Optional[str]: return "system" if "messages" in path: return "messages" - # A rejection we cannot localise: suppress the tool marker first, since - # toolConfig.tools is the only placement any supported family is known to - # refuse while still accepting the others. return "tools" @@ -715,7 +476,7 @@ def note_cache_point_rejection(model_id: str, placement: str) -> None: def cache_point_allowed(model_id: str, placement: str) -> bool: - """Return False once Bedrock has refused this placement for this model.""" + """False once Bedrock has refused this placement for this model.""" return placement not in _CACHE_POINT_REJECTIONS.get(model_id.lower(), ()) @@ -728,28 +489,28 @@ def _is_cache_point_block(block: Any) -> bool: return isinstance(block, dict) and set(block.keys()) == {"cachePoint"} +def _without_cache_points(blocks: Any) -> Optional[list]: + """Return ``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 _is_cache_point_block(b)] + return None if len(cleaned) == len(blocks) else cleaned + + def strip_cache_points(kwargs: Dict[str, Any], placement: str) -> Dict[str, Any]: - """Return a copy of Converse kwargs with ``placement``'s cachePoint removed. + """Copy of Converse kwargs with ``placement``'s cachePoint removed. Returns the input unchanged (same object) when there was nothing to strip, which is what callers use to decide a retry cannot help. """ if placement == "system": - system = kwargs.get("system") - if not isinstance(system, list): - return kwargs - cleaned = [b for b in system if not _is_cache_point_block(b)] - if len(cleaned) == len(system): - return kwargs - return {**kwargs, "system": cleaned} + 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") - tools = (tool_config or {}).get("tools") - if not isinstance(tools, list): - return kwargs - cleaned = [t for t in tools if not _is_cache_point_block(t)] - if len(cleaned) == len(tools): + cleaned = _without_cache_points((tool_config or {}).get("tools")) + if cleaned is None: return kwargs return {**kwargs, "toolConfig": {**tool_config, "tools": cleaned}} @@ -760,30 +521,22 @@ def strip_cache_points(kwargs: Dict[str, Any], placement: str) -> Dict[str, Any] changed = False cleaned_messages = [] for msg in messages: - content = msg.get("content") if isinstance(msg, dict) else None - if isinstance(content, list) and any(_is_cache_point_block(b) for b in content): + content = _without_cache_points(msg.get("content") if isinstance(msg, dict) else None) + if content is not None: changed = True - cleaned_messages.append({ - **msg, - "content": [b for b in content if not _is_cache_point_block(b)], - }) + cleaned_messages.append({**msg, "content": content}) else: cleaned_messages.append(msg) - if not changed: - return kwargs - return {**kwargs, "messages": cleaned_messages} + return {**kwargs, "messages": cleaned_messages} if changed else kwargs return kwargs -def recover_from_cache_point_rejection( - exc: BaseException, kwargs: Dict[str, Any] -) -> Optional[Dict[str, Any]]: +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. None means the error was not a cachePoint rejection, or the marker was - already absent — in which case retrying cannot change the outcome and the - caller must re-raise. + already absent — retrying cannot change the outcome; caller must re-raise. """ placement = cache_point_rejection_placement(exc) if placement is None: @@ -802,79 +555,47 @@ def recover_from_cache_point_rejection( return retry_kwargs +_REGIONAL_PREFIXES = ("global.", "us.", "eu.", "apac.", "ap.", "au.", "jp.", "ca.", "sa.", "me.", "af.") + + def is_anthropic_bedrock_model(model_id: str) -> bool: - """Return True if the model is an Anthropic Claude model on Bedrock. + """True for Claude on Bedrock (``anthropic.claude-*`` with any regional prefix). - These models should use the AnthropicBedrock SDK path for full feature - parity (prompt caching, thinking budgets, adaptive thinking). - Non-Claude models use the Converse API path. - - Matches: - - ``anthropic.claude-*`` (foundation model IDs) - - ``us.anthropic.claude-*`` (US inference profiles) - - ``global.anthropic.claude-*`` (global inference profiles) - - ``eu.anthropic.claude-*`` (EU inference profiles) + These use the AnthropicBedrock SDK path for full feature parity (prompt + caching, thinking budgets); non-Claude models use the Converse path. """ model_lower = model_id.lower() - # Strip regional prefix if present - for prefix in ( - "global.", "us.", "eu.", "apac.", "ap.", "au.", "jp.", - "ca.", "sa.", "me.", "af.", - ): + for prefix in _REGIONAL_PREFIXES: if model_lower.startswith(prefix): model_lower = model_lower[len(prefix):] break return model_lower.startswith("anthropic.claude") -# --------------------------------------------------------------------------- -# Message format conversion: OpenAI → Bedrock Converse -# --------------------------------------------------------------------------- +# --- Message format conversion: OpenAI → Bedrock Converse --- def convert_tools_to_converse(tools: List[Dict]) -> List[Dict]: - """Convert OpenAI-format tool definitions to Bedrock Converse ``toolConfig``. - - OpenAI format:: - - {"type": "function", "function": {"name": "...", "description": "...", - "parameters": {"type": "object", "properties": {...}}}} - - Converse format:: - - {"toolSpec": {"name": "...", "description": "...", - "inputSchema": {"json": {"type": "object", "properties": {...}}}}} - """ - if not tools: - return [] + """OpenAI ``{"function": {...}}`` tool defs → Converse ``{"toolSpec": {...}}``.""" result = [] - for t in tools: + for t in tools or []: fn = t.get("function", {}) - name = fn.get("name", "") - description = fn.get("description", "") - parameters = fn.get("parameters", {"type": "object", "properties": {}}) result.append({ "toolSpec": { - "name": name, - "description": description, - "inputSchema": {"json": parameters}, + "name": fn.get("name", ""), + "description": fn.get("description", ""), + "inputSchema": {"json": fn.get("parameters", {"type": "object", "properties": {}})}, } }) return result -# Bedrock's Converse API rejects any text content block whose text is empty -# OR whitespace-only (ValidationException: "text content blocks must contain -# non-whitespace text"). A lone space is whitespace and is rejected too — the -# placeholder MUST itself be non-whitespace. Ref: issue #9486. +# Converse rejects empty OR whitespace-only text blocks ("text content blocks +# must contain non-whitespace text"), so the placeholder must be non-whitespace. _EMPTY_TEXT_PLACEHOLDER = "(empty)" def _safe_text(text) -> str: - """Return ``text`` if it's non-whitespace, else a non-whitespace placeholder. - - Handles None, empty string, and whitespace-only string (spaces, tabs, - newlines) — all of which Bedrock's Converse API rejects as text content. - """ + """Return ``text`` if it has non-whitespace content, else the placeholder (None/non-str ok).""" if text is None: return _EMPTY_TEXT_PLACEHOLDER if not isinstance(text, str): @@ -882,86 +603,169 @@ def _safe_text(text) -> str: return text if text.strip() else _EMPTY_TEXT_PLACEHOLDER -def _convert_content_to_converse(content) -> List[Dict]: - """Convert OpenAI message content (string or list) to Converse content blocks. +def _image_block_from_data_url(url: str) -> Dict: + """``data:;base64,...`` → Converse image block with RAW bytes. - Handles: - - Plain text strings → [{"text": "..."}] - - Content arrays with text/image_url parts → mixed text/image blocks - - Replaces empty/whitespace-only text blocks with a non-whitespace - placeholder — Bedrock's Converse API rejects messages where a text - content block is empty or whitespace-only (ValidationException: - "text content blocks must contain non-whitespace text"). Ref: issue #9486. + boto3 base64-encodes at the wire layer, so passing the base64 string + through double-encodes and Bedrock rejects it ("Failed to sanitize image"). """ - if content is None: + header, _, data = url.partition(",") + media_type = "image/jpeg" + if header.startswith("data:"): + mime_part = header[5:].split(";")[0] + if mime_part: + media_type = mime_part + try: + 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 message content (str or parts list) → Converse content blocks. + + Empty/whitespace-only text becomes the placeholder; remote image URLs are + not supported by Converse and are included as a text reference instead. + """ + 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)}) + continue + if not isinstance(part, dict): + continue + part_type = part.get("type", "") + if part_type == "text": + blocks.append({"text": _safe_text(part.get("text", ""))}) + elif part_type == "image_url": + image_url = part.get("image_url", {}) + url = image_url.get("url", "") if isinstance(image_url, dict) else "" + if url.startswith("data:"): + blocks.append(_image_block_from_data_url(url)) + else: + blocks.append({"text": f"[Image: {url}]"}) + return blocks if blocks else [{"text": _EMPTY_TEXT_PLACEHOLDER}] + + +def _system_blocks(content) -> List[Dict]: + """System content → text blocks; blank parts are dropped, not placeholder-filled.""" if isinstance(content, str): - return [{"text": _safe_text(content)}] + return [{"text": content}] if content.strip() else [] + blocks: List[Dict] = [] if isinstance(content, list): - blocks = [] for part in content: - if isinstance(part, str): - blocks.append({"text": _safe_text(part)}) - continue - if not isinstance(part, dict): - continue - part_type = part.get("type", "") - if part_type == "text": + if isinstance(part, dict) and part.get("type") == "text": text = part.get("text", "") - blocks.append({"text": _safe_text(text)}) - elif part_type == "image_url": - image_url = part.get("image_url", {}) - url = image_url.get("url", "") if isinstance(image_url, dict) else "" - if url.startswith("data:"): - # data:image/jpeg;base64,/9j/4AAQ... - header, _, data = url.partition(",") - media_type = "image/jpeg" - if header.startswith("data:"): - mime_part = header[5:].split(";")[0] - if mime_part: - media_type = mime_part - # Decode base64 to raw bytes — boto3 re-encodes at the - # wire layer, so passing the base64 string directly - # results in double-encoding and Bedrock rejects it with - # "Failed to sanitize image". Ref: #33317. - import base64 - try: - raw_bytes = base64.b64decode(data) - except Exception: - raw_bytes = data.encode("utf-8") - blocks.append({ - "image": { - "format": media_type.split("/")[-1] if "/" in media_type else "jpeg", - "source": {"bytes": raw_bytes}, - } - }) - else: - # Remote URL — Converse doesn't support URLs directly, - # include as text reference for the model. - blocks.append({"text": f"[Image: {url}]"}) - return blocks if blocks else [{"text": _EMPTY_TEXT_PLACEHOLDER}] - return [{"text": _safe_text(content)}] + if isinstance(text, str) and text.strip(): + blocks.append({"text": text}) + elif isinstance(part, str) and part.strip(): + blocks.append({"text": part}) + return blocks -def convert_messages_to_converse( - messages: List[Dict], -) -> Tuple[Optional[List[Dict]], List[Dict]]: - """Convert OpenAI-format messages to Bedrock Converse format. +def _tool_use_block(tool_use_id, name, input_dict) -> Dict: + return {"toolUse": {"toolUseId": tool_use_id, "name": name, "input": input_dict}} - Returns ``(system_prompt, converse_messages)`` where: - - ``system_prompt`` is a list of system content blocks (or None) - - ``converse_messages`` is the conversation in Converse format - Handles: - - System messages → extracted as system prompt - - User messages → ``{"role": "user", "content": [...]}`` - - Assistant messages → ``{"role": "assistant", "content": [...]}`` - - Tool calls → ``{"toolUse": {"toolUseId": ..., "name": ..., "input": ...}}`` - - Tool results → ``{"toolResult": {"toolUseId": ..., "content": [...]}}`` +def _replay_ordered_blocks(ordered_blocks: List) -> List[Dict]: + """Rebuild the exact Bedrock block sequence captured at normalization time. - Converse requires strict user/assistant alternation. Consecutive messages - with the same role are merged into a single message. + Redacted reasoning bytes are stored base64-encoded (JSON-safe sidecar) and + decoded here; 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 + replay = {} + if isinstance(reasoning.get("text"), str): + replay["text"] = reasoning["text"] + encoded = reasoning.get("redactedContentBase64") + if isinstance(encoded, str) and encoded: + try: + replay["redactedContent"] = base64.b64decode(encoded, validate=True) + except (ValueError, TypeError): + continue + if replay: + content_blocks.append({"reasoningContent": replay}) + 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 _assistant_blocks(msg: Dict, content) -> List[Dict]: + """Assistant message → Converse blocks. + + An ordered ``bedrock_content_blocks`` sidecar is authoritative and replayed + verbatim. Otherwise: redacted thinking from ``reasoning_details`` (so + opaque encrypted reasoning round-trips byte-for-byte), then text, then + tool calls. + """ + ordered_blocks = msg.get("bedrock_content_blocks") + if isinstance(ordered_blocks, list) and ordered_blocks: + content_blocks = _replay_ordered_blocks(ordered_blocks) + if content_blocks: + return content_blocks + + content_blocks = [] + for detail in (msg.get("reasoning_details") or []): + if not isinstance(detail, dict) or detail.get("type") != "redacted_thinking": + continue + encoded = detail.get("data") or detail.get("redactedContentBase64") + if not isinstance(encoded, str) or not encoded: + continue + try: + redacted = base64.b64decode(encoded, validate=True) + except (ValueError, TypeError): + continue + content_blocks.append({"reasoningContent": {"redactedContent": redacted}}) + + 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", {}) + args_str = fn.get("arguments", "{}") + try: + args_dict = json.loads(args_str) if isinstance(args_str, str) else args_str + except (json.JSONDecodeError, TypeError): + args_dict = {} + content_blocks.append(_tool_use_block(tc.get("id", ""), fn.get("name", ""), args_dict)) + return content_blocks + + +def _append_turn(converse_msgs: List[Dict], role: str, blocks: List[Dict]) -> None: + """Append blocks as a new turn, merging into the previous turn of the same role (strict alternation).""" + if converse_msgs and converse_msgs[-1]["role"] == role: + converse_msgs[-1]["content"].extend(blocks) + else: + converse_msgs.append({"role": role, "content": blocks}) + + +def convert_messages_to_converse(messages: List[Dict]) -> Tuple[Optional[List[Dict]], List[Dict]]: + """OpenAI messages → ``(system_blocks_or_None, converse_messages)``. + + System messages become the system prompt; tool results become + ``toolResult`` blocks in a user turn. Converse requires strict + user/assistant alternation with a user turn first and last, so same-role + neighbours are merged and placeholder user turns are inserted at the ends. """ system_blocks: List[Dict] = [] converse_msgs: List[Dict] = [] @@ -971,191 +775,110 @@ def convert_messages_to_converse( content = msg.get("content") if role == "system": - # System messages become the system prompt. Blank/whitespace-only - # parts are dropped entirely (not placeholder-filled) since a - # system prompt made up of only placeholder text is meaningless. - if isinstance(content, str) and content.strip(): - system_blocks.append({"text": content}) - elif isinstance(content, list): - for part in content: - if isinstance(part, dict) and part.get("type") == "text": - text = part.get("text", "") - if isinstance(text, str) and text.strip(): - system_blocks.append({"text": text}) - elif isinstance(part, str) and part.strip(): - system_blocks.append({"text": part}) - continue - - if role == "tool": - # Tool result messages → merge into the preceding user turn - tool_call_id = msg.get("tool_call_id", "") + system_blocks.extend(_system_blocks(content)) + elif role == "tool": result_content = content if isinstance(content, str) else json.dumps(content) - tool_result_block = { + _append_turn(converse_msgs, "user", [{ "toolResult": { - "toolUseId": tool_call_id, + "toolUseId": msg.get("tool_call_id", ""), "content": [{"text": _safe_text(result_content)}], } - } - # In Converse, tool results go in a "user" role message - if converse_msgs and converse_msgs[-1]["role"] == "user": - converse_msgs[-1]["content"].append(tool_result_block) - else: - converse_msgs.append({ - "role": "user", - "content": [tool_result_block], - }) - continue + }]) + elif role == "assistant": + content_blocks = _assistant_blocks(msg, content) or [{"text": _EMPTY_TEXT_PLACEHOLDER}] + _append_turn(converse_msgs, "assistant", content_blocks) + elif role == "user": + _append_turn(converse_msgs, "user", _convert_content_to_converse(content)) - if role == "assistant": - content_blocks = [] - ordered_blocks = msg.get("bedrock_content_blocks") - if isinstance(ordered_blocks, list) and ordered_blocks: - # Rebuild the exact Bedrock block sequence captured at - # normalization time. Redacted bytes are stored as base64 so - # the sidecar remains JSON-safe in assistant history. - 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 - replay = {} - if isinstance(reasoning.get("text"), str): - replay["text"] = reasoning["text"] - encoded = reasoning.get("redactedContentBase64") - if isinstance(encoded, str) and encoded: - try: - replay["redactedContent"] = base64.b64decode(encoded, validate=True) - except (ValueError, TypeError): - continue - if replay: - content_blocks.append({"reasoningContent": replay}) - elif "toolUse" in block and isinstance(block["toolUse"], dict): - tu = block["toolUse"] - content_blocks.append({"toolUse": { - "toolUseId": tu.get("toolUseId", ""), - "name": tu.get("name", ""), - "input": tu.get("input", {}), - }}) - - if not content_blocks: - ordered_blocks = None - - if content_blocks: - # Ordered replay is authoritative; do not append parallel - # reasoning/text/tool lists a second time. - pass - else: - # Bedrock may return opaque encrypted reasoning instead of text. - # Preserve the payload in the provider-neutral reasoning_details - # envelope so the next tool turn can replay it byte-for-byte. - for detail in (msg.get("reasoning_details") or []): - if not isinstance(detail, dict) or detail.get("type") != "redacted_thinking": - continue - encoded = detail.get("data") or detail.get("redactedContentBase64") - if not isinstance(encoded, str) or not encoded: - continue - try: - redacted = base64.b64decode(encoded, validate=True) - except (ValueError, TypeError): - continue - content_blocks.append({"reasoningContent": {"redactedContent": redacted}}) - - # Convert text content - if isinstance(content, str) and content.strip(): - content_blocks.append({"text": content}) - elif isinstance(content, list): - content_blocks.extend(_convert_content_to_converse(content)) - - # Convert tool calls - tool_calls = msg.get("tool_calls", []) - for tc in (tool_calls or []): - fn = tc.get("function", {}) - args_str = fn.get("arguments", "{}") - try: - args_dict = json.loads(args_str) if isinstance(args_str, str) else args_str - except (json.JSONDecodeError, TypeError): - args_dict = {} - content_blocks.append({ - "toolUse": { - "toolUseId": tc.get("id", ""), - "name": fn.get("name", ""), - "input": args_dict, - } - }) - - if not content_blocks: - content_blocks = [{"text": _EMPTY_TEXT_PLACEHOLDER}] - - # Merge with previous assistant message if needed (strict alternation) - if converse_msgs and converse_msgs[-1]["role"] == "assistant": - converse_msgs[-1]["content"].extend(content_blocks) - else: - converse_msgs.append({ - "role": "assistant", - "content": content_blocks, - }) - continue - - if role == "user": - content_blocks = _convert_content_to_converse(content) - # Merge with previous user message if needed (strict alternation) - if converse_msgs and converse_msgs[-1]["role"] == "user": - converse_msgs[-1]["content"].extend(content_blocks) - else: - converse_msgs.append({ - "role": "user", - "content": content_blocks, - }) - continue - - # Converse requires the first message to be from the user if converse_msgs and converse_msgs[0]["role"] != "user": converse_msgs.insert(0, {"role": "user", "content": [{"text": _EMPTY_TEXT_PLACEHOLDER}]}) - - # Converse requires the last message to be from the user if converse_msgs and converse_msgs[-1]["role"] != "user": converse_msgs.append({"role": "user", "content": [{"text": _EMPTY_TEXT_PLACEHOLDER}]}) return (system_blocks if system_blocks else None, converse_msgs) -# --------------------------------------------------------------------------- -# Response format conversion: Bedrock Converse → OpenAI -# --------------------------------------------------------------------------- +# --- Response format conversion: Bedrock Converse → OpenAI --- -def _converse_stop_reason_to_openai(stop_reason: str) -> str: - """Map Bedrock Converse stop reasons to OpenAI finish_reason values.""" - mapping = { - "end_turn": "stop", - "stop_sequence": "stop", - "tool_use": "tool_calls", - "max_tokens": "length", - "content_filtered": "content_filter", - "guardrail_intervened": "content_filter", - } - return mapping.get(stop_reason, "stop") +# 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") + if isinstance(redacted, str): + return redacted + return 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 _build_openai_response( + text_parts: List[str], reasoning_parts: List[str], reasoning_details: List[Dict[str, Any]], + tool_calls: List[SimpleNamespace], ordered_blocks: List[Dict[str, Any]], + usage_data: Dict[str, int], stop_reason: str, model: str, +) -> SimpleNamespace: + """Assemble the OpenAI-shaped response shared by the sync and streaming paths. + + Converse's inputTokens EXCLUDES cache read/write tokens (OpenAI's + prompt_tokens includes them), so they are added back here and the + Anthropic-named cache fields are surfaced for downstream normalize_usage(). + """ + msg = SimpleNamespace( + role="assistant", + content="\n".join(text_parts) if text_parts else None, + tool_calls=tool_calls if tool_calls else None, + reasoning_content="\n\n".join(reasoning_parts) if reasoning_parts else None, + reasoning_details=reasoning_details or None, + bedrock_content_blocks=ordered_blocks or None, + ) + + input_tokens = usage_data.get("inputTokens", 0) + cache_read_tokens = usage_data.get("cacheReadInputTokens", 0) + cache_write_tokens = usage_data.get("cacheWriteInputTokens", 0) + output_tokens = usage_data.get("outputTokens", 0) + usage = SimpleNamespace( + prompt_tokens=input_tokens + cache_read_tokens + cache_write_tokens, + completion_tokens=output_tokens, + total_tokens=input_tokens + cache_read_tokens + cache_write_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 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: - """Convert a Bedrock Converse API response to an OpenAI-compatible object. + """Bedrock Converse response → OpenAI ``ChatCompletion``-shaped SimpleNamespace. - The agent loop in ``run_agent.py`` expects responses shaped like - ``openai.ChatCompletion`` — this function bridges the gap. - - Returns a SimpleNamespace with: - - ``.choices[0].message.content`` — text response - - ``.choices[0].message.tool_calls`` — tool call list (if any) - - ``.choices[0].finish_reason`` — stop/tool_calls/length - - ``.usage`` — token usage stats + Exposes ``.choices[0].message.{content,tool_calls,reasoning_content, + reasoning_details,bedrock_content_blocks}``, ``.choices[0].finish_reason`` + and ``.usage``. """ - output = response.get("output", {}) - message = output.get("message", {}) - content_blocks = message.get("content", []) - stop_reason = response.get("stopReason", "end_turn") + content_blocks = response.get("output", {}).get("message", {}).get("content", []) text_parts = [] reasoning_parts = [] @@ -1169,109 +892,35 @@ def normalize_converse_response(response: Dict) -> SimpleNamespace: ordered_blocks.append({"text": block["text"]}) elif "reasoningContent" in block: reasoning = block["reasoningContent"] - if isinstance(reasoning, dict): - thinking_text = reasoning.get("text", "") - encoded = None - if thinking_text: - reasoning_parts.append(str(thinking_text)) - redacted = reasoning.get("redactedContent") - if redacted is not None: - if isinstance(redacted, (bytes, bytearray)): - encoded = base64.b64encode(bytes(redacted)).decode("ascii") - elif isinstance(redacted, str): - encoded = redacted - else: - encoded = None - if encoded: - reasoning_details.append({ - "type": "redacted_thinking", - "data": encoded, - }) - if thinking_text or encoded: - ordered_reasoning = {} - if thinking_text: - ordered_reasoning["text"] = str(thinking_text) - if encoded: - ordered_reasoning["redactedContentBase64"] = encoded - ordered_blocks.append({"reasoningContent": ordered_reasoning}) + if not isinstance(reasoning, dict): + continue + thinking_text = reasoning.get("text", "") + encoded = _encode_redacted(reasoning.get("redactedContent")) + ordered_reasoning = {} + if thinking_text: + reasoning_parts.append(str(thinking_text)) + ordered_reasoning["text"] = str(thinking_text) + if encoded: + reasoning_details.append({"type": "redacted_thinking", "data": encoded}) + ordered_reasoning["redactedContentBase64"] = encoded + if ordered_reasoning: + ordered_blocks.append({"reasoningContent": ordered_reasoning}) elif "toolUse" in block: tu = block["toolUse"] - ordered_blocks.append({"toolUse": { - "toolUseId": tu.get("toolUseId", ""), - "name": tu.get("name", ""), - "input": tu.get("input", {}), - }}) - tool_calls.append(SimpleNamespace( - id=tu.get("toolUseId", ""), - type="function", - function=SimpleNamespace( - name=tu.get("name", ""), - arguments=json.dumps(tu.get("input", {})), - ), - )) + tool_use_id, name, tool_input = tu.get("toolUseId", ""), tu.get("name", ""), tu.get("input", {}) + ordered_blocks.append(_tool_use_block(tool_use_id, name, tool_input)) + tool_calls.append(_tool_call_ns(tool_use_id, name, tool_input)) - # Build the message object - msg = SimpleNamespace( - role="assistant", - content="\n".join(text_parts) if text_parts else None, - tool_calls=tool_calls if tool_calls else None, - reasoning_content="\n\n".join(reasoning_parts) if reasoning_parts else None, - reasoning_details=reasoning_details or None, - bedrock_content_blocks=ordered_blocks or None, - ) - - # Build usage stats. Converse's inputTokens excludes cache read/write - # tokens (unlike OpenAI's prompt_tokens, which includes them) — restore - # the OpenAI-style "total includes cache" convention here so downstream - # normalize_usage() can subtract them back out consistently, and surface - # the Anthropic-named fields it already falls back to for cache reads. - usage_data = response.get("usage", {}) - input_tokens = usage_data.get("inputTokens", 0) - cache_read_tokens = usage_data.get("cacheReadInputTokens", 0) - cache_write_tokens = usage_data.get("cacheWriteInputTokens", 0) - output_tokens = usage_data.get("outputTokens", 0) - usage = SimpleNamespace( - prompt_tokens=input_tokens + cache_read_tokens + cache_write_tokens, - completion_tokens=output_tokens, - total_tokens=input_tokens + cache_read_tokens + cache_write_tokens + output_tokens, - cache_read_input_tokens=cache_read_tokens, - cache_creation_input_tokens=cache_write_tokens, - ) - - finish_reason = _converse_stop_reason_to_openai(stop_reason) - if tool_calls and finish_reason == "stop": - finish_reason = "tool_calls" - - choice = SimpleNamespace( - index=0, - message=msg, - finish_reason=finish_reason, - ) - - return SimpleNamespace( - choices=[choice], - usage=usage, - model=response.get("modelId", ""), + return _build_openai_response( + text_parts, reasoning_parts, reasoning_details, tool_calls, ordered_blocks, + response.get("usage", {}), response.get("stopReason", "end_turn"), response.get("modelId", ""), ) -# --------------------------------------------------------------------------- -# Streaming response conversion -# --------------------------------------------------------------------------- +# --- Streaming response conversion --- def normalize_converse_stream_events(event_stream) -> SimpleNamespace: - """Consume a Bedrock ConverseStream event stream and build an OpenAI-compatible response. - - Processes the stream events in order: - - ``messageStart`` — role info - - ``contentBlockStart`` — new text or toolUse block - - ``contentBlockDelta`` — incremental text or toolUse input - - ``contentBlockStop`` — block complete - - ``messageStop`` — stop reason - - ``metadata`` — usage stats - - Returns the same shape as ``normalize_converse_response()``. - """ + """Consume a ConverseStream event stream (no callbacks) → same shape as ``normalize_converse_response()``.""" return stream_converse_with_callbacks(event_stream) @@ -1283,34 +932,14 @@ def stream_converse_with_callbacks( on_interrupt_check=None, on_event=None, ) -> SimpleNamespace: - """Process a Bedrock ConverseStream event stream with real-time callbacks. + """Process a boto3 ``converse_stream()`` response with real-time callbacks. - This is the core streaming function that powers both the CLI's live token - display and the gateway's progressive message updates. - - Args: - event_stream: The boto3 ``converse_stream()`` response containing a - ``stream`` key with an iterable of events. - on_text_delta: Called with each text chunk as it arrives. Only fires - when no tool_use blocks have been seen (same semantics as the - Anthropic and chat_completions streaming paths). - on_tool_start: Called with the tool name when a toolUse block begins. - Lets the TUI show a spinner while tool arguments are generated. - on_reasoning_delta: Called with reasoning/thinking text chunks. - Bedrock surfaces thinking via ``reasoning`` content block deltas - on supported models (Claude 4.6+). - on_interrupt_check: Called on each event. Should return True if the - agent has been interrupted and streaming should stop. - on_event: Called once at the top of the loop body for EVERY yielded - Bedrock event (text/tool-input/reasoning/metadata deltas alike), - before any branching. Provides a wire-level liveness signal so an - external watchdog can distinguish "still receiving events" from - "stream wedged with no data". Errors raised by the callback are - swallowed so a liveness hook can never abort the stream. - - Returns: - An OpenAI-compatible SimpleNamespace response, identical in shape to - ``normalize_converse_response()``. + ``on_text_delta`` only fires while no toolUse block has been seen (same + semantics as the Anthropic/chat_completions paths). ``on_interrupt_check`` + runs per event; True stops streaming. ``on_event`` is a wire-level liveness + signal fired for EVERY event before any branching (so a watchdog can tell + "still receiving" from "wedged"); its exceptions are swallowed so it can + never abort the stream. Returns the ``normalize_converse_response()`` shape. """ text_parts: List[str] = [] reasoning_parts: List[str] = [] @@ -1324,17 +953,22 @@ def stream_converse_with_callbacks( stop_reason = "end_turn" usage_data: Dict[str, int] = {} + def _current_block(default: Dict[str, Any]) -> Dict[str, Any]: + idx = current_block_index if current_block_index is not None else len(stream_blocks) + return stream_blocks.setdefault(idx, default) + + def _flush_text() -> None: + nonlocal current_text_buffer + if current_text_buffer: + text_parts.append("".join(current_text_buffer)) + current_text_buffer = [] + for event in event_stream.get("stream", []): - # Wire-level liveness signal: fire on EVERY yielded event (text, tool - # input, reasoning, metadata) before branching so an external watchdog - # can tell a still-flowing stream from a wedged one. Best-effort — a - # liveness callback must never be able to abort the stream. if on_event is not None: try: on_event() except Exception: pass - # Check for interrupt if on_interrupt_check and on_interrupt_check(): break @@ -1344,20 +978,13 @@ def stream_converse_with_callbacks( start = start_event.get("start", {}) if "toolUse" in start: has_tool_use = True - # Flush any accumulated text - if current_text_buffer: - text_parts.append("".join(current_text_buffer)) - current_text_buffer = [] + _flush_text() current_tool = { "toolUseId": start["toolUse"].get("toolUseId", ""), "name": start["toolUse"].get("name", ""), "input_json": "", } - stream_blocks[current_block_index] = {"toolUse": { - "toolUseId": current_tool["toolUseId"], - "name": current_tool["name"], - "input": {}, - }} + stream_blocks[current_block_index] = _tool_use_block(current_tool["toolUseId"], current_tool["name"], {}) if on_tool_start: on_tool_start(current_tool["name"]) @@ -1365,18 +992,15 @@ def stream_converse_with_callbacks( delta = event["contentBlockDelta"].get("delta", {}) if "text" in delta: text = delta["text"] - block = stream_blocks.setdefault(current_block_index if current_block_index is not None else len(stream_blocks), {"text": ""}) + block = _current_block({"text": ""}) block["text"] = block.get("text", "") + text current_text_buffer.append(text) - # Fire text delta callback only when no tool calls are present - # (same semantics as Anthropic/chat_completions streaming) if on_text_delta and not has_tool_use: on_text_delta(text) elif "toolUse" in delta: if current_tool is not None: current_tool["input_json"] += delta["toolUse"].get("input", "") elif "reasoningContent" in delta: - # Claude 4.6+ on Bedrock surfaces thinking via reasoningContent reasoning = delta["reasoningContent"] if isinstance(reasoning, dict): thinking_text = reasoning.get("text", "") @@ -1384,23 +1008,13 @@ def stream_converse_with_callbacks( reasoning_parts.append(str(thinking_text)) if on_reasoning_delta: on_reasoning_delta(thinking_text) - block = stream_blocks.setdefault(current_block_index if current_block_index is not None else len(stream_blocks), {"reasoningContent": {}}) + block = _current_block({"reasoningContent": {}}) block.setdefault("reasoningContent", {})["text"] = block["reasoningContent"].get("text", "") + str(thinking_text) - redacted = reasoning.get("redactedContent") - if redacted is not None: - if isinstance(redacted, (bytes, bytearray)): - encoded = base64.b64encode(bytes(redacted)).decode("ascii") - elif isinstance(redacted, str): - encoded = redacted - else: - encoded = None - if encoded: - reasoning_details.append({ - "type": "redacted_thinking", - "data": encoded, - }) - block = stream_blocks.setdefault(current_block_index if current_block_index is not None else len(stream_blocks), {"reasoningContent": {}}) - block.setdefault("reasoningContent", {})["redactedContentBase64"] = encoded + encoded = _encode_redacted(reasoning.get("redactedContent")) + if encoded: + reasoning_details.append({"type": "redacted_thinking", "data": encoded}) + block = _current_block({"reasoningContent": {}}) + block.setdefault("reasoningContent", {})["redactedContentBase64"] = encoded elif "contentBlockStop" in event: if current_tool is not None: @@ -1408,20 +1022,12 @@ def stream_converse_with_callbacks( input_dict = json.loads(current_tool["input_json"]) if current_tool["input_json"] else {} except (json.JSONDecodeError, TypeError): input_dict = {} - tool_calls.append(SimpleNamespace( - id=current_tool["toolUseId"], - type="function", - function=SimpleNamespace( - name=current_tool["name"], - arguments=json.dumps(input_dict), - ), - )) + tool_calls.append(_tool_call_ns(current_tool["toolUseId"], current_tool["name"], input_dict)) if current_block_index is not None and current_block_index in stream_blocks: stream_blocks[current_block_index]["toolUse"]["input"] = input_dict current_tool = None - elif current_text_buffer: - text_parts.append("".join(current_text_buffer)) - current_text_buffer = [] + else: + _flush_text() elif "messageStop" in event: stop_reason = event["messageStop"].get("stopReason", "end_turn") @@ -1435,51 +1041,19 @@ def stream_converse_with_callbacks( "cacheWriteInputTokens": meta_usage.get("cacheWriteInputTokens", 0), } - # Flush remaining text - if current_text_buffer: - text_parts.append("".join(current_text_buffer)) + _flush_text() - msg = SimpleNamespace( - role="assistant", - content="\n".join(text_parts) if text_parts else None, - tool_calls=tool_calls if tool_calls else None, - reasoning_content="\n\n".join(reasoning_parts) if reasoning_parts else None, - reasoning_details=reasoning_details or None, - bedrock_content_blocks=[stream_blocks[i] for i in sorted(stream_blocks)] or None, - ) - - input_tokens = usage_data.get("inputTokens", 0) - cache_read_tokens = usage_data.get("cacheReadInputTokens", 0) - cache_write_tokens = usage_data.get("cacheWriteInputTokens", 0) - output_tokens = usage_data.get("outputTokens", 0) - usage = SimpleNamespace( - prompt_tokens=input_tokens + cache_read_tokens + cache_write_tokens, - completion_tokens=output_tokens, - total_tokens=input_tokens + cache_read_tokens + cache_write_tokens + output_tokens, - cache_read_input_tokens=cache_read_tokens, - cache_creation_input_tokens=cache_write_tokens, - ) - - finish_reason = _converse_stop_reason_to_openai(stop_reason) - if tool_calls and finish_reason == "stop": - finish_reason = "tool_calls" - - choice = SimpleNamespace( - index=0, - message=msg, - finish_reason=finish_reason, - ) - - return SimpleNamespace( - choices=[choice], - usage=usage, - model="", + return _build_openai_response( + text_parts, reasoning_parts, reasoning_details, tool_calls, + [stream_blocks[i] for i in sorted(stream_blocks)], usage_data, stop_reason, "", ) -# --------------------------------------------------------------------------- -# High-level API: call Bedrock Converse -# --------------------------------------------------------------------------- +# --- High-level API: call Bedrock Converse --- + +def _cache_point() -> Dict[str, Any]: + return {"cachePoint": {"type": "default"}} + def build_converse_kwargs( model: str, @@ -1491,15 +1065,15 @@ def build_converse_kwargs( stop_sequences: Optional[List[str]] = None, guardrail_config: Optional[Dict] = None, ) -> Dict[str, Any]: - """Build kwargs for ``bedrock-runtime.converse()`` or ``converse_stream()``. + """Build kwargs for ``bedrock-runtime.converse()`` / ``converse_stream()``. - Converts OpenAI-format inputs to Converse API parameters. - - ``max_tokens=None`` omits ``inferenceConfig.maxTokens`` entirely, in which - case Bedrock defaults to the model's maximum allowed output — the Converse - field is optional per the AWS API reference. The default stays 4096 so - existing callers are unaffected; callers that want the model's full output - budget (e.g. uncapped auxiliary vision calls) pass ``None`` explicitly. + ``max_tokens=None`` omits ``inferenceConfig.maxTokens`` so Bedrock uses the + model's maximum output (callers wanting the full budget, e.g. uncapped + auxiliary vision calls, pass None explicitly; default stays 4096). + cachePoint markers go on system, tools and the second-newest message + (so the marker survives as only the tail grows — mirrors the Anthropic + system_and_3 strategy) — each only if the model supports caching and + Bedrock has not rejected that placement. """ system_prompt, converse_messages = convert_messages_to_converse(messages) cache_enabled = _model_supports_prompt_cache(model) @@ -1516,32 +1090,28 @@ def build_converse_kwargs( if system_prompt: if cache_enabled and cache_point_allowed(model, "system"): - system_prompt = system_prompt + [{"cachePoint": {"type": "default"}}] + system_prompt = system_prompt + [_cache_point()] kwargs["system"] = system_prompt from agent.anthropic_adapter import _forbids_sampling_params if not _forbids_sampling_params(model): if temperature is not None: - kwargs["inferenceConfig"]["temperature"] = temperature - + inference_config["temperature"] = temperature if top_p is not None: - kwargs["inferenceConfig"]["topP"] = top_p + inference_config["topP"] = top_p if stop_sequences: - kwargs["inferenceConfig"]["stopSequences"] = stop_sequences + inference_config["stopSequences"] = stop_sequences if tools: converse_tools = convert_tools_to_converse(tools) if converse_tools: - # Some Bedrock models don't support tool/function calling (e.g. - # DeepSeek R1, reasoning-only models). Sending toolConfig to - # these models causes a ValidationException → retry loop → failure. - # Strip tools for known non-tool-calling models and warn the user. - # Ref: PR #7920 feedback from @ptlally, pattern from PR #4346. + # Non-tool-calling models (e.g. DeepSeek R1) reject toolConfig with a + # ValidationException → retry loop → failure. Strip tools and warn. if _model_supports_tool_use(model): if cache_enabled and cache_point_allowed(model, "tools"): - converse_tools = converse_tools + [{"cachePoint": {"type": "default"}}] + converse_tools = converse_tools + [_cache_point()] kwargs["toolConfig"] = {"tools": converse_tools} else: logger.warning( @@ -1554,19 +1124,15 @@ def build_converse_kwargs( and cache_point_allowed(model, "messages") and len(converse_messages) >= 2 ): - # Checkpoint everything up to (not including) the newest turn, so the - # marker survives unchanged across requests as only the tail grows — - # mirroring the Anthropic system_and_3 strategy in prompt_caching.py. content = converse_messages[-2].get("content") if isinstance(content, list) and content: - content.append({"cachePoint": {"type": "default"}}) + content.append(_cache_point()) if guardrail_config: kwargs["guardrailConfig"] = guardrail_config - if not kwargs["inferenceConfig"]: - # inferenceConfig is optional on the wire; don't send an empty object. - del kwargs["inferenceConfig"] + if not inference_config: + del kwargs["inferenceConfig"] # optional on the wire; don't send {} return kwargs @@ -1582,20 +1148,14 @@ def call_converse( stop_sequences: Optional[List[str]] = None, guardrail_config: Optional[Dict] = None, ) -> SimpleNamespace: - """Call Bedrock Converse API (non-streaming) and return an OpenAI-compatible response. + """Non-streaming Converse call → OpenAI-compatible response. - This is the primary entry point for the agent loop when using the Bedrock provider. + Retries once without the rejected cachePoint placement; evicts the cached + client on stale-connection errors before re-raising. """ client = _get_bedrock_runtime_client(region) kwargs = build_converse_kwargs( - model=model, - messages=messages, - tools=tools, - max_tokens=max_tokens, - temperature=temperature, - top_p=top_p, - stop_sequences=stop_sequences, - guardrail_config=guardrail_config, + model, messages, tools, max_tokens, temperature, top_p, stop_sequences, guardrail_config, ) try: @@ -1615,66 +1175,7 @@ def call_converse( return normalize_converse_response(response) -def call_converse_stream( - 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: - """Call Bedrock ConverseStream API and return an OpenAI-compatible response. - - Consumes the full stream and returns the assembled response. For true - streaming with delta callbacks, use ``iter_converse_stream()`` instead. - """ - client = _get_bedrock_runtime_client(region) - kwargs = build_converse_kwargs( - model=model, - messages=messages, - tools=tools, - max_tokens=max_tokens, - temperature=temperature, - top_p=top_p, - stop_sequences=stop_sequences, - guardrail_config=guardrail_config, - ) - - try: - response = client.converse_stream(**kwargs) - except Exception as exc: - retry_kwargs = recover_from_cache_point_rejection(exc, kwargs) - if retry_kwargs is not None: - return normalize_converse_stream_events( - client.converse_stream(**retry_kwargs) - ) - if is_streaming_access_denied_error(exc): - # IAM allows bedrock:InvokeModel but not - # InvokeModelWithResponseStream — permanent for this session. - # Fall back to the non-streaming converse() path. - logger.info( - "bedrock: converse_stream denied by IAM on (region=%s, model=%s) — " - "falling back to non-streaming converse().", - region, model, - ) - return normalize_converse_response(client.converse(**kwargs)) - if is_stale_connection_error(exc): - logger.warning( - "bedrock: stale-connection error on converse_stream(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_stream_events(response) - - -# --------------------------------------------------------------------------- -# Model discovery -# --------------------------------------------------------------------------- +# --- Model discovery --- _discovery_cache: Dict[str, Any] = {} _DISCOVERY_CACHE_TTL_SECONDS = 3600 @@ -1685,24 +1186,74 @@ def reset_discovery_cache(): _discovery_cache.clear() -def discover_bedrock_models( - region: str, - provider_filter: Optional[List[str]] = None, -) -> List[Dict[str, Any]]: - """Discover available Bedrock foundation models and inference profiles. +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 + if filter_set: + provider_name = (summary.get("providerName") or "").lower() + model_prefix = model_id.split(".")[0].lower() if "." in model_id else "" + if provider_name not in filter_set and model_prefix not in filter_set: + continue + if summary.get("modelLifecycle", {}).get("status", "").upper() != "ACTIVE": + continue + if not summary.get("responseStreamingSupported", False): + continue + output_mods = summary.get("outputModalities", []) + if "TEXT" not in output_mods: + continue + models.append({ + "id": model_id, + "name": (summary.get("modelName") or model_id).strip(), + "provider": (summary.get("providerName") or "").strip(), + "input_modalities": summary.get("inputModalities", []), + "output_modalities": output_mods, + "streaming": True, + }) - Returns a list of model info dicts with keys: - - ``id``: Model ID (e.g. "anthropic.claude-sonnet-4-6-20250514-v1:0") - - ``name``: Human-readable name - - ``provider``: Model provider (e.g. "Anthropic", "Amazon", "Meta") - - ``input_modalities``: List of input types (e.g. ["TEXT", "IMAGE"]) - - ``output_modalities``: List of output types - - ``streaming``: Whether streaming is supported - Caches results for 1 hour per region to avoid repeated API calls. +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: + kwargs = {"nextToken": next_token} if next_token else {} + response = client.list_inference_profiles(**kwargs) + profiles.extend(response.get("inferenceProfileSummaries", [])) + next_token = response.get("nextToken") + if not next_token: + break - Mirrors OpenClaw's ``discoverBedrockModels()`` in - ``extensions/amazon-bedrock/discovery.ts``. + 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({ + "id": profile_id, + "name": (profile.get("inferenceProfileName") or profile_id).strip(), + "provider": "inference-profile", + "input_modalities": ["TEXT"], + "output_modalities": ["TEXT"], + "streaming": True, + }) + seen_ids.add(profile_id.lower()) + + +def discover_bedrock_models(region: str, provider_filter: Optional[List[str]] = None) -> List[Dict[str, Any]]: + """Discover Bedrock foundation models + inference profiles (cached 1h per region/filter). + + Each entry has ``id``, ``name``, ``provider`` ("inference-profile" for + profiles), ``input_modalities``, ``output_modalities``, ``streaming``. + Sorted with ``global.`` cross-region profiles first, then by name. + Returns [] when the client cannot be built. """ import time @@ -1718,188 +1269,43 @@ def discover_bedrock_models( return [] models = [] - seen_ids = set() filter_set = {f.lower() for f in (provider_filter or [])} - # 1. Discover foundation models try: - response = client.list_foundation_models() - for summary in response.get("modelSummaries", []): - model_id = (summary.get("modelId") or "").strip() - if not model_id: - continue - - # Apply provider filter - if filter_set: - provider_name = (summary.get("providerName") or "").lower() - model_prefix = model_id.split(".")[0].lower() if "." in model_id else "" - if provider_name not in filter_set and model_prefix not in filter_set: - continue - - # Only include active, streaming-capable, text-output models - lifecycle = summary.get("modelLifecycle", {}) - if lifecycle.get("status", "").upper() != "ACTIVE": - continue - if not summary.get("responseStreamingSupported", False): - continue - output_mods = summary.get("outputModalities", []) - if "TEXT" not in output_mods: - continue - - models.append({ - "id": model_id, - "name": (summary.get("modelName") or model_id).strip(), - "provider": (summary.get("providerName") or "").strip(), - "input_modalities": summary.get("inputModalities", []), - "output_modalities": output_mods, - "streaming": True, - }) - seen_ids.add(model_id.lower()) + _list_foundation_models(client, filter_set, models) except Exception as e: logger.warning("Failed to list Bedrock foundation models: %s", e) - # 2. Discover inference profiles (cross-region, better capacity) try: - profiles = [] - next_token = None - while True: - kwargs = {} - if next_token: - kwargs["nextToken"] = next_token - response = client.list_inference_profiles(**kwargs) - for profile in response.get("inferenceProfileSummaries", []): - profiles.append(profile) - next_token = response.get("nextToken") - if not next_token: - break - - for profile in profiles: - profile_id = (profile.get("inferenceProfileId") or "").strip() - if not profile_id: - continue - if profile.get("status") != "ACTIVE": - continue - if profile_id.lower() in seen_ids: - continue - - # Apply provider filter to underlying models - if filter_set: - profile_models = profile.get("models", []) - matches = any( - _extract_provider_from_arn(m.get("modelArn", "")).lower() in filter_set - for m in profile_models - ) - if not matches: - continue - - models.append({ - "id": profile_id, - "name": (profile.get("inferenceProfileName") or profile_id).strip(), - "provider": "inference-profile", - "input_modalities": ["TEXT"], - "output_modalities": ["TEXT"], - "streaming": True, - }) - seen_ids.add(profile_id.lower()) + _list_inference_profiles(client, filter_set, models) except Exception as e: logger.debug("Skipping inference profile discovery: %s", e) - # Sort: global cross-region profiles first (recommended), then alphabetical 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, - } + _discovery_cache[cache_key] = {"timestamp": time.time(), "models": models} return models def _extract_provider_from_arn(arn: str) -> str: - """Extract the model provider from a Bedrock model ARN. - - Example: "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-v2" - → "anthropic" - """ + """``arn:aws:bedrock:...:foundation-model/anthropic.claude-v2`` → ``"anthropic"``.""" match = re.search(r"foundation-model/([^.]+)", arn) return match.group(1) if match else "" -# --------------------------------------------------------------------------- -# Error classification — Bedrock-specific exceptions -# --------------------------------------------------------------------------- -# Mirrors OpenClaw's classifyFailoverReason() and matchesContextOverflowError() -# in extensions/amazon-bedrock/register.sync.runtime.ts. - -# Patterns that indicate the input context exceeded the model's token limit. -# Used by run_agent.py to trigger context compression instead of retrying. -CONTEXT_OVERFLOW_PATTERNS = [ - re.compile(r"ValidationException.*(?:input is too long|max input token|input token.*exceed)", re.IGNORECASE), - re.compile(r"ValidationException.*(?:exceeds? the (?:maximum|max) (?:number of )?(?:input )?tokens)", re.IGNORECASE), - re.compile(r"ModelStreamErrorException.*(?:Input is too long|too many input tokens)", re.IGNORECASE), -] - -# Patterns for throttling / rate limit errors — should trigger backoff + retry. -THROTTLE_PATTERNS = [ - re.compile(r"ThrottlingException", re.IGNORECASE), - re.compile(r"Too many concurrent requests", re.IGNORECASE), - re.compile(r"ServiceQuotaExceededException", re.IGNORECASE), -] - -# Patterns for transient overload — model is temporarily unavailable. -OVERLOAD_PATTERNS = [ - re.compile(r"ModelNotReadyException", re.IGNORECASE), - re.compile(r"ModelTimeoutException", re.IGNORECASE), - re.compile(r"InternalServerException", re.IGNORECASE), -] -def is_context_overflow_error(error_message: str) -> bool: - """Return True if the error indicates the input context was too large. - - When this returns True, the agent should compress context and retry - rather than treating it as a fatal error. - """ - return any(p.search(error_message) for p in CONTEXT_OVERFLOW_PATTERNS) - - -def classify_bedrock_error(error_message: str) -> str: - """Classify a Bedrock error for retry/failover decisions. - - Returns: - - ``"context_overflow"`` — input too long, compress and retry - - ``"rate_limit"`` — throttled, backoff and retry - - ``"overloaded"`` — model temporarily unavailable, retry with delay - - ``"unknown"`` — unclassified error - """ - if is_context_overflow_error(error_message): - return "context_overflow" - if any(p.search(error_message) for p in THROTTLE_PATTERNS): - return "rate_limit" - if any(p.search(error_message) for p in OVERLOAD_PATTERNS): - return "overloaded" - return "unknown" - - -# --------------------------------------------------------------------------- -# Bedrock model context lengths -# --------------------------------------------------------------------------- -# Static fallback table for models where the Bedrock API doesn't expose -# context window sizes. Used by agent/model_metadata.py when dynamic -# detection is unavailable. +# --- Bedrock model context lengths --- +# Static fallback table for when the live probe is unavailable; used by +# agent/model_metadata.py. Keys are matched by longest substring, so versioned +# entries (opus-4-6/4-7/4-8) win over the generic "anthropic.claude-opus-4". BEDROCK_CONTEXT_LENGTHS: Dict[str, int] = { - # Anthropic Claude models on Bedrock. - # Context windows per Anthropic's official models comparison - # (https://platform.claude.com/docs/en/about-claude/models/overview). - # Fable / Sonnet 5 / Opus 4.8 / 4.7 / 4.6 / Sonnet 4.6 have 1M generally - # available (no beta header required as of April 2026). Sonnet 4.5 and - # Sonnet 4 had their `context-1m-2025-08-07` beta retired on - # April 30, 2026, so they are standard 200K; Haiku 4.5 is 200K. - # These 1M entries must match agent/model_metadata.py - # DEFAULT_CONTEXT_LENGTHS or the agent compresses context prematurely. - # Keys are matched by longest-substring, so the versioned 4-6/4-7/4-8 - # entries win over the generic "anthropic.claude-opus-4" fallback. + # Anthropic Claude (per Anthropic's models overview). Fable / Sonnet 5 / + # Opus 4.8-4.6 / Sonnet 4.6 are 1M GA; Sonnet 4.5 / Sonnet 4 lost their 1M + # beta and are 200K; Haiku 4.5 is 200K. The 1M entries must match + # agent/model_metadata.py DEFAULT_CONTEXT_LENGTHS or context compresses early. "anthropic.claude-fable-5": 1_000_000, "anthropic.claude-fable": 1_000_000, "anthropic.claude-sonnet-5": 1_000_000, @@ -1939,59 +1345,32 @@ BEDROCK_CONTEXT_LENGTHS: Dict[str, int] = { # Default for unknown Bedrock models BEDROCK_DEFAULT_CONTEXT_LENGTH = 128_000 -# Probe tiers (in tokens). We send a request padded just past each tier and -# read the real window from Bedrock's length-validation error. Two reasons -# this is tiered rather than one giant request: -# 1. A wildly oversized payload (e.g. 5M tokens) makes Bedrock return an -# opaque InternalServerException after retries instead of a clean -# ValidationException — so we must stay within a sane overage. -# 2. Stepping up lets us discover larger windows (2M+) without over-padding -# smaller ones. -# Each tier value is the *padding target*; the error reports the true maximum, -# which is what we actually return. +# Probe padding targets (tokens). Tiered rather than one giant request because +# a wildly oversized payload (5M tokens) yields an opaque InternalServerException +# instead of a clean ValidationException, and stepping up discovers 2M+ +# windows without over-padding smaller ones. The error reports the true maximum. _BEDROCK_PROBE_TIERS = (1_300_000, 2_200_000) _WORDS_PER_TOKEN = 0.9 # conservative: ensures the padded prompt clears the tier def _static_bedrock_context_length(model_id: str) -> int: - """Longest-substring-match lookup against the static fallback table. - - Uses substring matching so versioned IDs like - ``anthropic.claude-sonnet-4-6-20250514-v1:0`` resolve correctly. - """ + """Longest-substring match against the static table (versioned IDs resolve correctly).""" model_lower = model_id.lower() - best_key = "" - best_val = BEDROCK_DEFAULT_CONTEXT_LENGTH - for key, val in BEDROCK_CONTEXT_LENGTHS.items(): - if key in model_lower and len(key) > len(best_key): - best_key = key - best_val = val - return best_val + matches = [key for key in BEDROCK_CONTEXT_LENGTHS if key in model_lower] + return BEDROCK_CONTEXT_LENGTHS[max(matches, key=len)] if matches else BEDROCK_DEFAULT_CONTEXT_LENGTH def probe_bedrock_context_length(model_id: str, region: str) -> Optional[int]: - """Discover a Bedrock model's real context window by provoking a length error. + """Discover a model's real context window by provoking a length error. - Bedrock does not expose the context window via any metadata API - (``get-foundation-model`` omits it, ``Converse`` metrics omit it, - ``CountTokens`` is unsupported on several models). The only authoritative - source is the ``ValidationException`` raised when a prompt exceeds the - window: + No Bedrock metadata API exposes the window; the only authoritative source + is the ValidationException for an oversized prompt ("prompt is too long: + 1300032 tokens > 1000000 maximum"). Length validation happens before + inference, so the probe is rejected immediately and costs nothing. If a + tier is accepted, that tier is returned as a safe lower bound. - "The model returned the following errors: prompt is too long: - 1300032 tokens > 1000000 maximum" - - Length validation happens *before* inference, so an oversized request is - rejected immediately and cheaply — no tokens are generated and no input is - actually processed. We pad a request just past each tier in - ``_BEDROCK_PROBE_TIERS`` and parse the reported ``maximum``. Tiers exist - because (a) a *wildly* oversized payload makes Bedrock fail with an opaque - InternalServerException instead of a clean length error, and (b) stepping - up discovers larger windows without over-padding smaller ones. - - Returns the detected window, or ``None`` if the probe could not run - (missing credentials, network error, or no parseable limit) so the caller - can fall back to the static table. + Returns None if the probe could not run (no credentials, network error, + no parseable limit) so the caller falls back to the static table. """ try: from agent.model_metadata import parse_context_limit_from_error @@ -2006,36 +1385,28 @@ def probe_bedrock_context_length(model_id: str, region: str) -> Optional[int]: last_error = "" for tier_tokens in _BEDROCK_PROBE_TIERS: - pad_words = int(tier_tokens / _WORDS_PER_TOKEN) - oversized = "data " * pad_words + oversized = "data " * int(tier_tokens / _WORDS_PER_TOKEN) try: client.converse( modelId=model_id, messages=[{"role": "user", "content": [{"text": oversized}]}], inferenceConfig={"maxTokens": 8}, ) - # Accepted a prompt this large → the window is at least this tier. - # Returning the tier as a lower bound is safe and avoids inventing - # a number we can't confirm. 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: - msg = str(exc) - last_error = msg - limit = parse_context_limit_from_error(msg) + 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 - # No parseable limit at this tier (opaque server error, auth, - # throttle). Try the next, smaller-overage strategy is N/A here — - # tiers ascend — so just continue; if all fail we return None. - continue + # Opaque server error / auth / throttle at this tier — try the next. logger.debug( "Bedrock context probe for %s returned no parseable limit: %s", @@ -2045,22 +1416,12 @@ def probe_bedrock_context_length(model_id: str, region: str) -> Optional[int]: def get_bedrock_context_length(model_id: str, region: str = "", probe: bool = True) -> int: - """Resolve the context window for a Bedrock model. + """Context window: live probe (if ``probe`` and ``region``) → static table → default. - Resolution order: - 1. Live probe against Bedrock (authoritative; cached by the caller). - 2. Static fallback table (longest-substring match). - 3. Conservative default. - - The static table is intentionally a *fallback*, not the primary source: - AWS ships new model versions (opus-4-7, opus-4-8, ...) faster than the - table can track, and a stale entry silently caps the window (e.g. a - 1M-token Opus pinned to 200K via an ``opus-4`` substring match). The - probe asks Bedrock directly so every model — current or future — gets its - real window with no table maintenance. - - ``probe=False`` (or an empty ``region``) skips the network call and uses - the static table only — used by pure-offline/display code paths. + The static table is a fallback only: AWS ships new versions faster than + the table tracks, and a stale substring match silently caps the window + (e.g. a 1M Opus pinned to 200K via "opus-4"). ``probe=False`` or an empty + region skips the network call for offline/display paths. """ if probe and region: probed = probe_bedrock_context_length(model_id, region) diff --git a/tests/agent/test_bedrock_adapter.py b/tests/agent/test_bedrock_adapter.py index 7e51aed9df..7d5653edc3 100644 --- a/tests/agent/test_bedrock_adapter.py +++ b/tests/agent/test_bedrock_adapter.py @@ -574,35 +574,6 @@ class TestBuildConverseKwargs: ) assert "inferenceConfig" not in kwargs - def test_call_converse_stream_omits_cap_for_none(self): - """The streaming entry point funnels through the same builder — pin - that max_tokens=None omits the cap there too.""" - from unittest.mock import MagicMock, patch as mock_patch - from agent.bedrock_adapter import call_converse_stream - boto3_client = MagicMock() - boto3_client.converse_stream.return_value = {"stream": []} - with mock_patch( - "agent.bedrock_adapter._get_bedrock_runtime_client", - return_value=boto3_client, - ): - call_converse_stream( - region="us-east-1", - model="test-model", - messages=[{"role": "user", "content": "Hi"}], - max_tokens=None, - temperature=0.2, - ) - wire_kwargs = boto3_client.converse_stream.call_args.kwargs - assert "maxTokens" not in wire_kwargs.get("inferenceConfig", {}) - - - - - - - - - def test_cache_point_added_for_supported_model(self): """Claude and Nova on the Converse path get cachePoint markers on system, tools, and the message before the newest turn.""" @@ -1033,27 +1004,6 @@ class TestGuardrailConfig: assert "guardrailConfig" not in kwargs -# --------------------------------------------------------------------------- -# Error classification -# --------------------------------------------------------------------------- - -class TestBedrockErrorClassification: - """Test Bedrock-specific error classification.""" - - def test_context_overflow_validation_exception(self): - from agent.bedrock_adapter import classify_bedrock_error - assert classify_bedrock_error( - "ValidationException: input is too long for model" - ) == "context_overflow" - - - - - - - - - class TestBedrockContextLength: """Test Bedrock model context length lookup.""" @@ -1277,36 +1227,11 @@ class TestIsStaleConnectionError: class TestCallConverseInvalidatesOnStaleError: - """call_converse / call_converse_stream evict the cached client when the + """call_converse evicts the cached client when the boto3 call raises a stale-connection error — so the next invocation reconnects instead of reusing the dead socket.""" - def test_converse_stream_evicts_client_on_stale_error(self): - pytest.importorskip("botocore.exceptions", reason="botocore (with working exceptions module) required") - from agent.bedrock_adapter import ( - _bedrock_runtime_client_cache, - call_converse_stream, - reset_client_cache, - ) - from botocore.exceptions import ConnectionClosedError - - reset_client_cache() - dead_client = MagicMock() - dead_client.converse_stream.side_effect = ConnectionClosedError( - endpoint_url="https://bedrock.example", - ) - _bedrock_runtime_client_cache["us-east-1"] = dead_client - - with pytest.raises(ConnectionClosedError): - call_converse_stream( - region="us-east-1", - model="anthropic.claude-3-sonnet-20240229-v1:0", - messages=[{"role": "user", "content": "hi"}], - ) - - assert "us-east-1" not in _bedrock_runtime_client_cache - def test_converse_does_not_evict_on_non_stale_error(self): """Non-stale errors (e.g. ValidationException) leave the client cache alone.""" pytest.importorskip("botocore.exceptions", reason="botocore (with working exceptions module) required") @@ -1376,52 +1301,6 @@ class TestStreamingAccessDeniedDetection: ) is False -class TestCallConverseStreamIamFallback: - """call_converse_stream() falls back to converse() when IAM denies the - streaming action — InvokeModel-only policies keep working.""" - - def test_falls_back_to_converse_on_streaming_denial(self): - pytest.importorskip("botocore.exceptions", reason="botocore (with working exceptions module) required") - from agent.bedrock_adapter import ( - _bedrock_runtime_client_cache, - call_converse_stream, - reset_client_cache, - ) - from botocore.exceptions import ClientError - - reset_client_cache() - client = MagicMock() - client.converse_stream.side_effect = ClientError( - error_response={ - "Error": { - "Code": "AccessDeniedException", - "Message": ( - "User is not authorized to perform: " - "bedrock:InvokeModelWithResponseStream" - ), - } - }, - operation_name="ConverseStream", - ) - client.converse.return_value = { - "output": {"message": {"role": "assistant", "content": [{"text": "hi"}]}}, - "stopReason": "end_turn", - "usage": {"inputTokens": 1, "outputTokens": 1, "totalTokens": 2}, - } - _bedrock_runtime_client_cache["us-east-1"] = client - - result = call_converse_stream( - region="us-east-1", - model="anthropic.claude-3-sonnet-20240229-v1:0", - messages=[{"role": "user", "content": "hi"}], - ) - - client.converse.assert_called_once() - assert result.choices[0].message.content == "hi" - # Not a stale connection — client stays cached. - assert _bedrock_runtime_client_cache.get("us-east-1") is client - - # --------------------------------------------------------------------------- # boto3 version check # ---------------------------------------------------------------------------