"""Tests for the AWS Bedrock Converse API adapter. Covers: - AWS credential detection and region resolution - Message format conversion (OpenAI → Converse and back) - Tool definition conversion - Response normalization (non-streaming and streaming) - Model discovery with caching - Edge cases: empty messages, consecutive roles, image content """ import json from contextlib import contextmanager from types import ModuleType from unittest.mock import MagicMock, patch import pytest # --------------------------------------------------------------------------- # botocore import hygiene (anti-flake, Aug 2026) # # Several tests in this file plant fake ``botocore`` modules in sys.modules # (to run without the real package / without touching the AWS credential # chain). The real package's ``botocore.exceptions`` lazily executes # ``from botocore.vendored import requests`` on FIRST import — if that first # import happens while a fake parent is (or was) installed, the chain # resolves against a module with no real __path__ and the whole file's # exception tests die with ``No module named 'botocore.vendored'`` — but # only in interpreter states where nothing imported it earlier (the exact # CI-vs-local flake on PR #92617). # # Two defenses, both required: # 1. Import the real exception types HERE, at module scope, before any # test can stub sys.modules. Once cached, later ``from # botocore.exceptions import X`` is a dict hit and can never # re-execute the vendored import under a poisoned parent. # 2. An autouse fixture snapshots every boto* sys.modules entry before # each test and restores it after, so no stub window can leak state # into a later test regardless of ordering. # --------------------------------------------------------------------------- try: # pragma: no cover - exercised implicitly by every exception test from botocore.exceptions import ( # noqa: F401 ClientError as _RealClientError, ConnectionClosedError as _RealConnectionClosedError, ) except Exception: # botocore genuinely not installed / torn — tests skip _RealClientError = _RealConnectionClosedError = None _BOTO_PREFIXES = ("botocore", "boto3") @pytest.fixture(autouse=True) def _boto_sys_modules_hygiene(): """Restore every boto* sys.modules entry after each test (see above).""" import sys as _sys saved = { name: mod for name, mod in _sys.modules.items() if name.split(".", 1)[0] in _BOTO_PREFIXES } yield for name in [ n for n in _sys.modules if n.split(".", 1)[0] in _BOTO_PREFIXES ]: _sys.modules.pop(name, None) _sys.modules.update(saved) @contextmanager def _mock_botocore_session(*, return_value=None, side_effect=None): """Patch botocore.session even when botocore is not installed.""" botocore_mod = ModuleType("botocore") session_mod = ModuleType("botocore.session") session_mod.get_session = MagicMock(return_value=return_value, side_effect=side_effect) botocore_mod.session = session_mod with patch.dict("sys.modules", {"botocore": botocore_mod, "botocore.session": session_mod}): yield session_mod.get_session # --------------------------------------------------------------------------- # AWS credential detection # --------------------------------------------------------------------------- class TestResolveAwsAuthEnvVar: """Test AWS credential environment variable detection. Mirrors OpenClaw's resolveAwsSdkEnvVarName() priority order. """ def test_requires_both_access_key_and_secret(self): from agent.bedrock_adapter import resolve_aws_auth_env_var # Only access key, no secret → should not match env = {"AWS_ACCESS_KEY_ID": "AKIA..."} assert resolve_aws_auth_env_var(env) != "AWS_ACCESS_KEY_ID" def test_returns_none_when_no_aws_auth(self): from agent.bedrock_adapter import resolve_aws_auth_env_var # Mock botocore to return no credentials (covers EC2 IMDS fallback) mock_session = MagicMock() mock_session.get_credentials.return_value = None with patch.dict("sys.modules", {"botocore": MagicMock(), "botocore.session": MagicMock()}): import botocore.session as _bs _bs.get_session = MagicMock(return_value=mock_session) assert resolve_aws_auth_env_var({}) is None class TestHasAwsCredentials: def test_true_with_profile(self): from agent.bedrock_adapter import has_aws_credentials assert has_aws_credentials({"AWS_PROFILE": "default"}) is True def test_false_with_empty_env(self): from agent.bedrock_adapter import has_aws_credentials mock_session = MagicMock() mock_session.get_credentials.return_value = None with patch.dict("sys.modules", {"botocore": MagicMock(), "botocore.session": MagicMock()}): import botocore.session as _bs _bs.get_session = MagicMock(return_value=mock_session) assert has_aws_credentials({}) is False class TestScopedAwsSessionKwargs: """A served multiplex profile never signs with the launch profile's ambient AWS chain (#116313).""" def test_two_homes_multiplex_refuses_ambient_chain_and_keeps_standalone(self, tmp_path, monkeypatch): """A -> B -> A over two real homes: A (own AWS_* in .env) gets its key pair, cred-less B is refused at the production client seam BEFORE boto3 is touched (``{}`` would let ``boto3.Session()`` sign as A) and its bearer read is scoped, A again is unaffected. Control: a standalone run keeps the ambient chain (``{}`` + process-env bearer).""" from agent import bedrock_adapter, secret_scope from agent.bedrock_adapter import _cached_client, resolve_bedrock_bearer_token, scoped_aws_session_kwargs from hermes_constants import reset_hermes_home_override, set_hermes_home_override home_a, home_b = tmp_path / "home-A", tmp_path / "home-B" for home in (home_a, home_b): home.mkdir() (home_a / ".env").write_text( "AWS_ACCESS_KEY_ID=AKIA-A\nAWS_SECRET_ACCESS_KEY=secret-A\nAWS_BEARER_TOKEN_BEDROCK=bearer-A\n" ) (home_b / ".env").write_text("") monkeypatch.setenv("AWS_ACCESS_KEY_ID", "AKIA-A") monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "secret-A") monkeypatch.setenv("AWS_BEARER_TOKEN_BEDROCK", "bearer-A") def boom(): raise AssertionError("boto3 must not be imported for a refused profile") monkeypatch.setattr(bedrock_adapter, "_require_boto3", boom) def in_scope(home, fn): h_tok = set_hermes_home_override(str(home)) s_tok = secret_scope.set_secret_scope(secret_scope.build_profile_secret_scope(home)) try: return fn() finally: secret_scope.reset_secret_scope(s_tok) reset_hermes_home_override(h_tok) # Control: standalone keeps the ambient chain. assert scoped_aws_session_kwargs() == {} assert resolve_bedrock_bearer_token() == "bearer-A" monkeypatch.setattr(secret_scope, "_MULTIPLEX_ACTIVE", True) a_kwargs = {"aws_access_key_id": "AKIA-A", "aws_secret_access_key": "secret-A"} assert in_scope(home_a, scoped_aws_session_kwargs) == a_kwargs with pytest.raises(RuntimeError, match="refused for this profile"): in_scope(home_b, lambda: _cached_client({}, "bedrock-runtime", "us-east-1")) assert in_scope(home_b, resolve_bedrock_bearer_token) == "" assert in_scope(home_a, scoped_aws_session_kwargs) == a_kwargs assert in_scope(home_a, resolve_bedrock_bearer_token) == "bearer-A" class TestResolveBedrocRegion: def test_prefers_aws_region(self): from agent.bedrock_adapter import resolve_bedrock_region env = {"AWS_REGION": "eu-west-1", "AWS_DEFAULT_REGION": "us-west-2"} assert resolve_bedrock_region(env) == "eu-west-1" def test_defaults_to_us_east_1(self): from agent.bedrock_adapter import resolve_bedrock_region from unittest.mock import MagicMock mock_session = MagicMock() mock_session.get_config_variable.return_value = None with _mock_botocore_session(return_value=mock_session): assert resolve_bedrock_region({}) == "us-east-1" # --------------------------------------------------------------------------- # Tool conversion # --------------------------------------------------------------------------- class TestConvertToolsToConverse: """Test OpenAI → Bedrock Converse tool definition conversion.""" def test_converts_single_tool(self): from agent.bedrock_adapter import convert_tools_to_converse tools = [{ "type": "function", "function": { "name": "read_file", "description": "Read a file from disk", "parameters": { "type": "object", "properties": { "path": {"type": "string", "description": "File path"}, }, "required": ["path"], }, }, }] result = convert_tools_to_converse(tools) assert len(result) == 1 spec = result[0]["toolSpec"] assert spec["name"] == "read_file" assert spec["description"] == "Read a file from disk" assert spec["inputSchema"]["json"]["type"] == "object" assert "path" in spec["inputSchema"]["json"]["properties"] def test_empty_tools(self): from agent.bedrock_adapter import convert_tools_to_converse assert convert_tools_to_converse([]) == [] assert convert_tools_to_converse(None) == [] # --------------------------------------------------------------------------- # Message conversion: OpenAI → Converse # --------------------------------------------------------------------------- class TestConvertMessagesToConverse: """Test OpenAI message format → Bedrock Converse format conversion.""" def test_extracts_system_prompt(self): from agent.bedrock_adapter import convert_messages_to_converse messages = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hello"}, ] system, msgs = convert_messages_to_converse(messages) assert system is not None assert len(system) == 1 assert system[0]["text"] == "You are a helpful assistant." assert len(msgs) == 1 assert msgs[0]["role"] == "user" def test_assistant_with_tool_calls(self): from agent.bedrock_adapter import convert_messages_to_converse messages = [ {"role": "user", "content": "Read the file"}, { "role": "assistant", "content": "I'll read that file.", "tool_calls": [{ "id": "call_123", "type": "function", "function": { "name": "read_file", "arguments": '{"path": "/tmp/test.txt"}', }, }], }, ] system, msgs = convert_messages_to_converse(messages) # 3 messages: user, assistant, trailing user (Converse requires last=user) assert len(msgs) == 3 assistant_content = msgs[1]["content"] # Should have text block + toolUse block assert any("text" in b for b in assistant_content) tool_use_blocks = [b for b in assistant_content if "toolUse" in b] assert len(tool_use_blocks) == 1 assert tool_use_blocks[0]["toolUse"]["name"] == "read_file" assert tool_use_blocks[0]["toolUse"]["toolUseId"] == "call_123" assert tool_use_blocks[0]["toolUse"]["input"] == {"path": "/tmp/test.txt"} def test_tool_result_becomes_user_message(self): from agent.bedrock_adapter import convert_messages_to_converse messages = [ {"role": "user", "content": "Read it"}, {"role": "assistant", "content": None, "tool_calls": [{ "id": "call_1", "type": "function", "function": {"name": "read_file", "arguments": "{}"}, }]}, {"role": "tool", "tool_call_id": "call_1", "content": "file contents here"}, ] system, msgs = convert_messages_to_converse(messages) # Tool result should be in a user-role message tool_result_msg = [m for m in msgs if m["role"] == "user" and any( "toolResult" in b for b in m["content"] )] assert len(tool_result_msg) == 1 tr = [b for b in tool_result_msg[0]["content"] if "toolResult" in b][0] assert tr["toolResult"]["toolUseId"] == "call_1" assert tr["toolResult"]["content"][0]["text"] == "file contents here" # --------------------------------------------------------------------------- # Response normalization: Converse → OpenAI # --------------------------------------------------------------------------- class TestNormalizeConverseResponse: """Test Bedrock Converse response → OpenAI format conversion.""" def test_text_response(self): from agent.bedrock_adapter import normalize_converse_response response = { "output": { "message": { "role": "assistant", "content": [{"text": "Hello, world!"}], }, }, "stopReason": "end_turn", "usage": {"inputTokens": 10, "outputTokens": 5}, } result = normalize_converse_response(response) assert result.choices[0].message.content == "Hello, world!" assert result.choices[0].message.tool_calls is None assert result.choices[0].finish_reason == "stop" assert result.usage.prompt_tokens == 10 assert result.usage.completion_tokens == 5 assert result.usage.total_tokens == 15 def test_cache_tokens_folded_into_prompt_tokens(self): """Converse's inputTokens excludes cache read/write tokens (unlike OpenAI's prompt_tokens). normalize_converse_response must add them back into prompt_tokens/total_tokens and surface the Anthropic-named fields so normalize_usage() picks them up via its existing fallback.""" from agent.bedrock_adapter import normalize_converse_response response = { "output": {"message": {"role": "assistant", "content": [{"text": "hi"}]}}, "stopReason": "end_turn", "usage": { "inputTokens": 50, "outputTokens": 20, "cacheReadInputTokens": 900, "cacheWriteInputTokens": 300, }, } result = normalize_converse_response(response) assert result.usage.prompt_tokens == 50 + 900 + 300 assert result.usage.completion_tokens == 20 assert result.usage.total_tokens == 50 + 900 + 300 + 20 assert result.usage.cache_read_input_tokens == 900 assert result.usage.cache_creation_input_tokens == 300 def test_tool_use_response(self): from agent.bedrock_adapter import normalize_converse_response response = { "output": { "message": { "role": "assistant", "content": [ {"text": "I'll read that file."}, { "toolUse": { "toolUseId": "call_abc", "name": "read_file", "input": {"path": "/tmp/test.txt"}, }, }, ], }, }, "stopReason": "tool_use", "usage": {"inputTokens": 20, "outputTokens": 15}, } result = normalize_converse_response(response) assert result.choices[0].message.content == "I'll read that file." assert result.choices[0].finish_reason == "tool_calls" tool_calls = result.choices[0].message.tool_calls assert len(tool_calls) == 1 assert tool_calls[0].id == "call_abc" assert tool_calls[0].function.name == "read_file" assert json.loads(tool_calls[0].function.arguments) == {"path": "/tmp/test.txt"} def test_redacted_reasoning_is_preserved_for_replay(self): from agent.bedrock_adapter import normalize_converse_response response = { "output": { "message": { "role": "assistant", "content": [ {"reasoningContent": {"redactedContent": b"opaque-bedrock-bytes"}}, {"toolUse": {"toolUseId": "call_1", "name": "read_file", "input": {}}}, ], }, }, "stopReason": "tool_use", "usage": {"inputTokens": 1, "outputTokens": 2}, } result = normalize_converse_response(response) details = result.choices[0].message.reasoning_details assert details == [{ "type": "redacted_thinking", "data": "b3BhcXVlLWJlZHJvY2stYnl0ZXM=", }] def test_redacted_reasoning_replays_as_bedrock_content_block(self): from agent.bedrock_adapter import convert_messages_to_converse _system, messages = convert_messages_to_converse([ { "role": "assistant", "content": None, "reasoning_details": [{ "type": "redacted_thinking", "data": "b3BhcXVlLWJlZHJvY2stYnl0ZXM=", }], "tool_calls": [{ "id": "call_1", "function": {"name": "read_file", "arguments": "{}"}, }], }, {"role": "tool", "tool_call_id": "call_1", "content": "ok"}, ]) assistant = next(m for m in messages if m["role"] == "assistant") assert assistant["content"][0] == { "reasoningContent": {"redactedContent": b"opaque-bedrock-bytes"} } def test_interleaved_reasoning_and_tools_keep_exact_order(self): from agent.bedrock_adapter import convert_messages_to_converse, normalize_converse_response normalized = normalize_converse_response({ "output": {"message": {"role": "assistant", "content": [ {"reasoningContent": {"redactedContent": b"r1"}}, {"toolUse": {"toolUseId": "t1", "name": "one", "input": {"n": 1}}}, {"reasoningContent": {"redactedContent": b"r2"}}, {"toolUse": {"toolUseId": "t2", "name": "two", "input": {"n": 2}}}, ]}}, "stopReason": "tool_use", }) msg = normalized.choices[0].message _system, replay = convert_messages_to_converse([{ "role": "user", "content": "go", }, { "role": "assistant", "content": msg.content, "tool_calls": [{ "id": tc.id, "type": "function", "function": {"name": tc.function.name, "arguments": tc.function.arguments}, } for tc in msg.tool_calls], "reasoning_details": msg.reasoning_details, "bedrock_content_blocks": msg.bedrock_content_blocks, }]) blocks = replay[1]["content"] assert [next(iter(block)) for block in blocks] == [ "reasoningContent", "toolUse", "reasoningContent", "toolUse" ] assert blocks[0]["reasoningContent"]["redactedContent"] == b"r1" assert blocks[2]["reasoningContent"]["redactedContent"] == b"r2" # --------------------------------------------------------------------------- # Streaming response normalization # --------------------------------------------------------------------------- class TestNormalizeConverseStreamEvents: """Test Bedrock ConverseStream event → OpenAI format conversion.""" def test_text_stream(self): from agent.bedrock_adapter import normalize_converse_stream_events events = {"stream": [ {"messageStart": {"role": "assistant"}}, {"contentBlockStart": {"contentBlockIndex": 0, "start": {}}}, {"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": "Hello"}}}, {"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": ", world!"}}}, {"contentBlockStop": {"contentBlockIndex": 0}}, {"messageStop": {"stopReason": "end_turn"}}, {"metadata": {"usage": {"inputTokens": 5, "outputTokens": 3}}}, ]} result = normalize_converse_stream_events(events) assert result.choices[0].message.content == "Hello, world!" assert result.choices[0].finish_reason == "stop" assert result.usage.prompt_tokens == 5 assert result.usage.completion_tokens == 3 def test_redacted_reasoning_stream_is_preserved(self): from agent.bedrock_adapter import normalize_converse_stream_events events = {"stream": [ {"messageStart": {"role": "assistant"}}, {"contentBlockStart": {"contentBlockIndex": 0, "start": {}}}, {"contentBlockDelta": { "contentBlockIndex": 0, "delta": {"reasoningContent": {"redactedContent": b"stream-secret"}}, }}, {"contentBlockStop": {"contentBlockIndex": 0}}, {"messageStop": {"stopReason": "end_turn"}}, {"metadata": {"usage": {"inputTokens": 2, "outputTokens": 3}}}, ]} result = normalize_converse_stream_events(events) assert result.choices[0].message.reasoning_details == [{ "type": "redacted_thinking", "data": "c3RyZWFtLXNlY3JldA==", }] def test_tool_use_stream(self): from agent.bedrock_adapter import normalize_converse_stream_events events = {"stream": [ {"messageStart": {"role": "assistant"}}, {"contentBlockStart": {"contentBlockIndex": 0, "start": { "toolUse": {"toolUseId": "call_1", "name": "read_file"}, }}}, {"contentBlockDelta": {"contentBlockIndex": 0, "delta": { "toolUse": {"input": '{"path":'}, }}}, {"contentBlockDelta": {"contentBlockIndex": 0, "delta": { "toolUse": {"input": '"/tmp/f"}'}, }}}, {"contentBlockStop": {"contentBlockIndex": 0}}, {"messageStop": {"stopReason": "tool_use"}}, {"metadata": {"usage": {"inputTokens": 10, "outputTokens": 8}}}, ]} result = normalize_converse_stream_events(events) assert result.choices[0].finish_reason == "tool_calls" tc = result.choices[0].message.tool_calls assert len(tc) == 1 assert tc[0].id == "call_1" assert tc[0].function.name == "read_file" assert json.loads(tc[0].function.arguments) == {"path": "/tmp/f"} # Real ConverseStream wire shape (captured from global.anthropic.claude-opus-5): a text block gets NO # contentBlockStart, only deltas stamped contentBlockIndex=0; the toolUse block then starts at index 1. _LIVE_TEXT_THEN_TOOL_EVENTS = [ {"messageStart": {"role": "assistant"}}, {"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": "I"}}}, {"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": "'ll echo "}}}, {"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": "banana"}}}, {"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": " now."}}}, {"contentBlockStop": {"contentBlockIndex": 0}}, {"contentBlockStart": {"contentBlockIndex": 1, "start": {"toolUse": {"toolUseId": "tooluse_1", "name": "echo"}}}}, {"contentBlockDelta": {"contentBlockIndex": 1, "delta": {"toolUse": {"input": ""}}}}, {"contentBlockDelta": {"contentBlockIndex": 1, "delta": {"toolUse": {"input": '{"s": "banana"}'}}}}, {"contentBlockStop": {"contentBlockIndex": 1}}, {"messageStop": {"stopReason": "tool_use"}}, {"metadata": {"usage": {"inputTokens": 10, "outputTokens": 8}}}, ] def test_text_deltas_without_content_block_start_stay_one_block_ahead_of_tool_use(self): """Regression: keying text deltas by a running counter instead of their contentBlockIndex shredded the text into one block per delta and let the toolUse start (index 1) overwrite the second fragment and sort into the middle — a ``[text, toolUse, text...]`` sidecar Claude 5 on Bedrock rejects on replay as "does not support assistant message prefill".""" from agent.bedrock_adapter import normalize_converse_stream_events result = normalize_converse_stream_events({"stream": list(self._LIVE_TEXT_THEN_TOOL_EVENTS)}) msg = result.choices[0].message assert msg.content == "I'll echo banana now." assert msg.bedrock_content_blocks == [ {"text": "I'll echo banana now."}, {"toolUse": {"toolUseId": "tooluse_1", "name": "echo", "input": {"s": "banana"}}}, ] assert [tc.function.name for tc in msg.tool_calls] == ["echo"] def test_events_without_content_block_index_fall_back_to_arrival_order(self): """Proxies/test doubles may omit contentBlockIndex: deltas continue the current block, a start opens a new one, so text still lands before the toolUse instead of being overwritten by it.""" from agent.bedrock_adapter import normalize_converse_stream_events events = [{k: {kk: vv for kk, vv in v.items() if kk != "contentBlockIndex"} for k, v in e.items()} for e in self._LIVE_TEXT_THEN_TOOL_EVENTS[:-2]] # Text after the tool's stop must open its own slot, not land on the closed tool block. events += [{"contentBlockDelta": {"delta": {"text": "done"}}}, {"contentBlockStop": {}}, {"messageStop": {"stopReason": "tool_use"}}] msg = normalize_converse_stream_events({"stream": events}).choices[0].message assert msg.content == "I'll echo banana now.\ndone" assert [list(b) for b in msg.bedrock_content_blocks] == [["text"], ["toolUse"], ["text"]] assert msg.bedrock_content_blocks[1]["toolUse"]["input"] == {"s": "banana"} # --------------------------------------------------------------------------- # build_converse_kwargs # --------------------------------------------------------------------------- class TestBuildConverseKwargs: """Test the high-level kwargs builder for Converse API calls.""" def test_basic_kwargs(self): from agent.bedrock_adapter import build_converse_kwargs messages = [ {"role": "system", "content": "Be helpful."}, {"role": "user", "content": "Hi"}, ] kwargs = build_converse_kwargs( model="anthropic.claude-sonnet-4-6-20250514-v1:0", messages=messages, max_tokens=1024, ) assert kwargs["modelId"] == "anthropic.claude-sonnet-4-6-20250514-v1:0" assert kwargs["inferenceConfig"]["maxTokens"] == 1024 assert kwargs["system"] is not None assert len(kwargs["messages"]) >= 1 def test_includes_tools(self): from agent.bedrock_adapter import build_converse_kwargs tools = [{"type": "function", "function": { "name": "test", "description": "Test", "parameters": {}, }}] kwargs = build_converse_kwargs( model="test-model", messages=[{"role": "user", "content": "Hi"}], tools=tools, ) assert "toolConfig" in kwargs assert len(kwargs["toolConfig"]["tools"]) == 1 def test_max_tokens_none_omits_cap(self): """max_tokens=None omits inferenceConfig.maxTokens so Bedrock uses the model's maximum allowed output (the Converse field is optional).""" from agent.bedrock_adapter import build_converse_kwargs kwargs = build_converse_kwargs( model="test-model", messages=[{"role": "user", "content": "Hi"}], max_tokens=None, temperature=0.1, ) assert "maxTokens" not in kwargs["inferenceConfig"] # Other inference params still flow through. assert kwargs["inferenceConfig"]["temperature"] == 0.1 def test_max_tokens_none_and_no_sampling_drops_empty_inference_config(self): """When every inference param is absent, don't send an empty inferenceConfig object on the wire.""" from agent.bedrock_adapter import build_converse_kwargs kwargs = build_converse_kwargs( model="test-model", messages=[{"role": "user", "content": "Hi"}], max_tokens=None, ) assert "inferenceConfig" not in kwargs def test_bedrock_xai_grok_models_never_receive_sampling_params(self): """Bedrock-hosted xAI Grok rejects temperature/topP in Converse with a hard 400 (ValidationException: "This model doesn't support the temperature field"); the _forbids_sampling_params guard is Claude-only, so Grok needs its own denylist. Sibling Bedrock models keep receiving sampling params.""" from agent.bedrock_adapter import build_converse_kwargs msgs = [{"role": "user", "content": "Hi"}] for model in ("us.xai.grok-4.6", "global.xai.grok-4.6"): cfg = build_converse_kwargs( model=model, messages=msgs, temperature=0.3, top_p=0.9 )["inferenceConfig"] assert "temperature" not in cfg and "topP" not in cfg, model for model in ("test-model", "qwen.qwen3-vl-235b-a22b"): cfg = build_converse_kwargs( model=model, messages=msgs, temperature=0.3, top_p=0.9 )["inferenceConfig"] assert cfg["temperature"] == 0.3 and cfg["topP"] == 0.9, model 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.""" from agent.bedrock_adapter import build_converse_kwargs tools = [{"type": "function", "function": { "name": "test", "description": "Test", "parameters": {}, }}] messages = [ {"role": "system", "content": "Be helpful."}, {"role": "user", "content": "First"}, {"role": "assistant", "content": "Reply"}, {"role": "user", "content": "Second"}, ] kwargs = build_converse_kwargs( model="anthropic.claude-sonnet-4-6-20250514-v1:0", messages=messages, tools=tools, ) assert kwargs["system"][-1] == {"cachePoint": {"type": "default"}} assert kwargs["toolConfig"]["tools"][-1] == {"cachePoint": {"type": "default"}} # Second-to-last converse message (the assistant "Reply" turn) carries # the checkpoint; the newest "Second" turn does not. marked = kwargs["messages"][-2]["content"] assert marked[-1] == {"cachePoint": {"type": "default"}} assert kwargs["messages"][-1]["content"][-1] != {"cachePoint": {"type": "default"}} def test_no_cache_point_for_unsupported_model(self): from agent.bedrock_adapter import build_converse_kwargs messages = [ {"role": "system", "content": "Be helpful."}, {"role": "user", "content": "First"}, {"role": "assistant", "content": "Reply"}, {"role": "user", "content": "Second"}, ] kwargs = build_converse_kwargs(model="meta.llama3-70b-instruct-v1:0", messages=messages) assert {"cachePoint": {"type": "default"}} not in kwargs["system"] for m in kwargs["messages"]: assert {"cachePoint": {"type": "default"}} not in m["content"] # --------------------------------------------------------------------------- # cachePoint rejection self-heal (#97281) # --------------------------------------------------------------------------- CACHE_POINT = {"cachePoint": {"type": "default"}} NOVA_TOOLS_REJECTION = ( "An error occurred (ValidationException) when calling the ConverseStream " "operation: The model returned the following errors: Malformed input " "request: #/toolConfig/tools/18: extraneous key [cachePoint] is not " "permitted, please reformat your input and try again." ) @pytest.fixture(autouse=True) def _clean_cache_point_rejections(): """Rejections are process-wide; keep them from leaking between tests.""" from agent.bedrock_adapter import reset_cache_point_rejections reset_cache_point_rejections() yield reset_cache_point_rejections() class TestCachePointRejectionRecovery: """Bedrock placement rules are per-family and per-field: Nova accepts cachePoint in system/messages but rejects it inside toolConfig.tools, failing 100% of tool-enabled turns (#97281). The server verdict is the authority - record it, drop that one marker, and retry.""" def _nova_kwargs(self): from agent.bedrock_adapter import build_converse_kwargs return build_converse_kwargs( model="us.amazon.nova-pro-v1:0", messages=[ {"role": "system", "content": "Be helpful."}, {"role": "user", "content": "First"}, {"role": "assistant", "content": "Reply"}, {"role": "user", "content": "Second"}, ], tools=[{"type": "function", "function": { "name": "test", "description": "Test", "parameters": {}, }}], ) def test_classifies_tools_rejection(self): from agent.bedrock_adapter import cache_point_rejection_placement assert cache_point_rejection_placement( Exception(NOVA_TOOLS_REJECTION) ) == "tools" def test_classifies_system_and_messages_rejections(self): from agent.bedrock_adapter import cache_point_rejection_placement assert cache_point_rejection_placement(Exception( "Malformed input request: #/system/1: extraneous key [cachePoint] " "is not permitted" )) == "system" assert cache_point_rejection_placement(Exception( "Malformed input request: #/messages/2/content/3: extraneous key " "[cachePoint] is not permitted" )) == "messages" def test_ignores_unrelated_errors(self): from agent.bedrock_adapter import cache_point_rejection_placement assert cache_point_rejection_placement( Exception("ThrottlingException: Too many requests") ) is None assert cache_point_rejection_placement(Exception( "Malformed input request: #/toolConfig/tools/0: extraneous key " "[toolChoice] is not permitted" )) is None def test_strip_removes_only_the_rejected_placement(self): from agent.bedrock_adapter import strip_cache_points kwargs = self._nova_kwargs() assert CACHE_POINT in kwargs["toolConfig"]["tools"] stripped = strip_cache_points(kwargs, "tools") assert CACHE_POINT not in stripped["toolConfig"]["tools"] # system and messages markers survive - Nova accepts those. assert stripped["system"][-1] == CACHE_POINT assert stripped["messages"][-2]["content"][-1] == CACHE_POINT # Original kwargs are untouched (no in-place mutation). assert CACHE_POINT in kwargs["toolConfig"]["tools"] def test_strip_is_identity_when_marker_absent(self): """No marker to remove -> same object, so callers know a retry is futile.""" from agent.bedrock_adapter import strip_cache_points kwargs = {"modelId": "x", "toolConfig": {"tools": [{"toolSpec": {}}]}} assert strip_cache_points(kwargs, "tools") is kwargs def test_recovery_records_verdict_so_later_turns_omit_the_marker(self): from agent.bedrock_adapter import recover_from_cache_point_rejection kwargs = self._nova_kwargs() retry = recover_from_cache_point_rejection( Exception(NOVA_TOOLS_REJECTION), kwargs ) assert retry is not None assert CACHE_POINT not in retry["toolConfig"]["tools"] # Next turn is built clean without another round-trip failure. rebuilt = self._nova_kwargs() assert CACHE_POINT not in rebuilt["toolConfig"]["tools"] assert rebuilt["system"][-1] == CACHE_POINT assert rebuilt["messages"][-2]["content"][-1] == CACHE_POINT def test_verdict_is_scoped_to_the_rejecting_model(self): from agent.bedrock_adapter import ( build_converse_kwargs, recover_from_cache_point_rejection, ) recover_from_cache_point_rejection( Exception(NOVA_TOOLS_REJECTION), self._nova_kwargs() ) claude = build_converse_kwargs( model="anthropic.claude-sonnet-4-6-20250514-v1:0", messages=[{"role": "user", "content": "hi"}], tools=[{"type": "function", "function": { "name": "test", "description": "Test", "parameters": {}, }}], ) assert claude["toolConfig"]["tools"][-1] == CACHE_POINT def test_recovery_declines_when_nothing_can_be_stripped(self): """A cachePoint rejection with no marker present must re-raise, not loop.""" from agent.bedrock_adapter import recover_from_cache_point_rejection kwargs = {"modelId": "us.amazon.nova-pro-v1:0", "toolConfig": {"tools": [{"toolSpec": {}}]}} assert recover_from_cache_point_rejection( Exception(NOVA_TOOLS_REJECTION), kwargs ) is None def test_call_converse_retries_without_the_marker(self): from agent.bedrock_adapter import call_converse client = MagicMock() client.converse.side_effect = [ Exception(NOVA_TOOLS_REJECTION), {"output": {"message": {"role": "assistant", "content": [{"text": "ok"}]}}, "stopReason": "end_turn", "usage": {"inputTokens": 1, "outputTokens": 1, "totalTokens": 2}}, ] with patch("agent.bedrock_adapter._get_bedrock_runtime_client", return_value=client): response = call_converse( region="us-east-1", model="us.amazon.nova-pro-v1:0", messages=[{"role": "user", "content": "hi"}], tools=[{"type": "function", "function": { "name": "test", "description": "Test", "parameters": {}, }}], ) assert response.choices[0].message.content == "ok" assert client.converse.call_count == 2 first, second = client.converse.call_args_list assert CACHE_POINT in first.kwargs["toolConfig"]["tools"] assert CACHE_POINT not in second.kwargs["toolConfig"]["tools"] def test_call_converse_reraises_unrelated_errors(self): from agent.bedrock_adapter import call_converse client = MagicMock() client.converse.side_effect = Exception("ThrottlingException") with patch("agent.bedrock_adapter._get_bedrock_runtime_client", return_value=client): with pytest.raises(Exception, match="ThrottlingException"): call_converse( region="us-east-1", model="us.amazon.nova-pro-v1:0", messages=[{"role": "user", "content": "hi"}], ) assert client.converse.call_count == 1 # --------------------------------------------------------------------------- # Model discovery # --------------------------------------------------------------------------- class TestDiscoverBedrockModels: """Test Bedrock model discovery with mocked AWS API calls.""" def test_provider_filter(self): from agent.bedrock_adapter import discover_bedrock_models, reset_discovery_cache reset_discovery_cache() mock_client = MagicMock() mock_client.list_foundation_models.return_value = { "modelSummaries": [ { "modelId": "anthropic.claude-v2", "modelName": "Claude v2", "providerName": "Anthropic", "inputModalities": ["TEXT"], "outputModalities": ["TEXT"], "responseStreamingSupported": True, "modelLifecycle": {"status": "ACTIVE"}, }, { "modelId": "amazon.titan-text", "modelName": "Titan", "providerName": "Amazon", "inputModalities": ["TEXT"], "outputModalities": ["TEXT"], "responseStreamingSupported": True, "modelLifecycle": {"status": "ACTIVE"}, }, ], } mock_client.list_inference_profiles.return_value = {"inferenceProfileSummaries": []} with patch("agent.bedrock_adapter._get_bedrock_control_client", return_value=mock_client): models = discover_bedrock_models("us-east-1", provider_filter=["anthropic"]) assert len(models) == 1 assert models[0]["id"] == "anthropic.claude-v2" def test_caches_results(self): from agent.bedrock_adapter import discover_bedrock_models, reset_discovery_cache reset_discovery_cache() mock_client = MagicMock() mock_client.list_foundation_models.return_value = { "modelSummaries": [{ "modelId": "test-model", "modelName": "Test", "providerName": "Test", "inputModalities": ["TEXT"], "outputModalities": ["TEXT"], "responseStreamingSupported": True, "modelLifecycle": {"status": "ACTIVE"}, }], } mock_client.list_inference_profiles.return_value = {"inferenceProfileSummaries": []} with patch("agent.bedrock_adapter._get_bedrock_control_client", return_value=mock_client): first = discover_bedrock_models("us-east-1") second = discover_bedrock_models("us-east-1") # Should only call the API once (second call uses cache) assert mock_client.list_foundation_models.call_count == 1 assert first == second def test_handles_api_error_gracefully(self): from agent.bedrock_adapter import discover_bedrock_models, reset_discovery_cache reset_discovery_cache() with patch("agent.bedrock_adapter._get_bedrock_control_client", side_effect=Exception("No creds")): models = discover_bedrock_models("us-east-1") assert models == [] class TestExtractProviderFromArn: def test_extracts_anthropic(self): from agent.bedrock_adapter import _extract_provider_from_arn arn = "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-sonnet-4-6" assert _extract_provider_from_arn(arn) == "anthropic" def test_returns_empty_for_invalid_arn(self): from agent.bedrock_adapter import _extract_provider_from_arn assert _extract_provider_from_arn("not-an-arn") == "" assert _extract_provider_from_arn("") == "" # --------------------------------------------------------------------------- # Client cache management # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Streaming with callbacks # --------------------------------------------------------------------------- class TestStreamConverseWithCallbacks: """Test real-time streaming with delta callbacks.""" def test_cache_tokens_folded_into_prompt_tokens(self): """The streaming path must fold cacheRead/WriteInputTokens into prompt_tokens the same way the non-streaming path does (see TestNormalizeConverseResponse.test_cache_tokens_folded_into_prompt_tokens).""" from agent.bedrock_adapter import stream_converse_with_callbacks events = {"stream": [ {"messageStart": {"role": "assistant"}}, {"contentBlockStart": {"contentBlockIndex": 0, "start": {}}}, {"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": "hi"}}}, {"contentBlockStop": {"contentBlockIndex": 0}}, {"messageStop": {"stopReason": "end_turn"}}, {"metadata": {"usage": { "inputTokens": 50, "outputTokens": 20, "cacheReadInputTokens": 900, "cacheWriteInputTokens": 300, }}}, ]} result = stream_converse_with_callbacks(events) assert result.usage.prompt_tokens == 50 + 900 + 300 assert result.usage.total_tokens == 50 + 900 + 300 + 20 assert result.usage.cache_read_input_tokens == 900 assert result.usage.cache_creation_input_tokens == 300 def test_text_deltas_suppressed_when_tool_use_present(self): """Text deltas should NOT fire when tool_use blocks are present.""" from agent.bedrock_adapter import stream_converse_with_callbacks deltas = [] events = {"stream": [ {"messageStart": {"role": "assistant"}}, {"contentBlockStart": {"contentBlockIndex": 0, "start": {}}}, {"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": "Let me check."}}}, {"contentBlockStop": {"contentBlockIndex": 0}}, {"contentBlockStart": {"contentBlockIndex": 1, "start": { "toolUse": {"toolUseId": "c1", "name": "search"}, }}}, {"contentBlockDelta": {"contentBlockIndex": 1, "delta": { "toolUse": {"input": '{"q":"test"}'}, }}}, {"contentBlockStop": {"contentBlockIndex": 1}}, {"messageStop": {"stopReason": "tool_use"}}, {"metadata": {"usage": {"inputTokens": 0, "outputTokens": 0}}}, ]} result = stream_converse_with_callbacks( events, on_text_delta=lambda t: deltas.append(t), ) # Text delta for "Let me check." should fire (before tool_use was seen) assert "Let me check." in deltas # But the result should still have both text and tool calls assert result.choices[0].message.content == "Let me check." assert len(result.choices[0].message.tool_calls) == 1 # --------------------------------------------------------------------------- # Guardrail config in build_converse_kwargs # --------------------------------------------------------------------------- class TestGuardrailConfig: """Test that guardrail configuration is correctly passed through.""" def test_guardrail_included_in_kwargs(self): from agent.bedrock_adapter import build_converse_kwargs guardrail = { "guardrailIdentifier": "gr-abc123", "guardrailVersion": "1", "streamProcessingMode": "async", "trace": "enabled", } kwargs = build_converse_kwargs( model="test-model", messages=[{"role": "user", "content": "Hi"}], guardrail_config=guardrail, ) assert kwargs["guardrailConfig"] == guardrail def test_no_guardrail_when_empty_dict(self): from agent.bedrock_adapter import build_converse_kwargs kwargs = build_converse_kwargs( model="test-model", messages=[{"role": "user", "content": "Hi"}], guardrail_config={}, ) # Empty dict is falsy, should not be included assert "guardrailConfig" not in kwargs # --------------------------------------------------------------------------- # Error classification # --------------------------------------------------------------------------- class TestBedrockContextLength: """Test Bedrock model context length lookup.""" def test_unknown_model_gets_default(self): from agent.bedrock_adapter import get_bedrock_context_length, BEDROCK_DEFAULT_CONTEXT_LENGTH assert get_bedrock_context_length("unknown.model-v1:0") == BEDROCK_DEFAULT_CONTEXT_LENGTH def test_no_region_skips_probe_uses_table(self): # Default call (no region) must NOT hit the network — returns the # static table value. Guards backward compatibility for callers that # still invoke get_bedrock_context_length(model_id) with one arg. from agent.bedrock_adapter import get_bedrock_context_length with patch("agent.bedrock_adapter.probe_bedrock_context_length") as mock_probe: assert get_bedrock_context_length("anthropic.claude-opus-4-6") == 1_000_000 mock_probe.assert_not_called() def test_static_catalog_claude_ids_match_their_anthropic_window(self): """Every Claude id in the Bedrock static fallback gets the same offline window as its bare Anthropic id, so a model added to the picker can't silently land on the 128K default.""" from agent.bedrock_adapter import get_bedrock_context_length from agent.model_metadata import DEFAULT_CONTEXT_LENGTHS, _longest_key_match from hermes_cli.models_catalog_static import _PROVIDER_MODELS mismatched = [] for model_id in _PROVIDER_MODELS["bedrock"]: _, sep, bare = model_id.partition("anthropic.") if not sep: continue hit = _longest_key_match(DEFAULT_CONTEXT_LENGTHS, bare) expected = hit[1] if hit else None actual = get_bedrock_context_length(model_id, probe=False) if actual != expected: mismatched.append((model_id, expected, actual)) assert not mismatched, f"Bedrock static Claude ids drift from DEFAULT_CONTEXT_LENGTHS: {mismatched}" def test_million_token_claude_entries_match_model_metadata(self): """BEDROCK_CONTEXT_LENGTHS must not drift from DEFAULT_CONTEXT_LENGTHS. The table's own comment requires the pairing, but nothing enforced it, so ``claude-opus-5`` reached one table and not the other and silently fell through to BEDROCK_DEFAULT_CONTEXT_LENGTH (#74263). Assert the relationship, not a snapshot of today's catalog. DEFAULT_CONTEXT_LENGTHS spells revisions with dots (``claude-opus-4.8``) while Bedrock IDs use hyphens — normalize rather than skip, so a future 1M model that only ever gets a dotted alias cannot escape the check. """ from agent.bedrock_adapter import get_bedrock_context_length from agent.model_metadata import DEFAULT_CONTEXT_LENGTHS mismatched = [] with patch("agent.bedrock_adapter.probe_bedrock_context_length") as mock_probe: for name, expected in DEFAULT_CONTEXT_LENGTHS.items(): if not name.startswith("claude-") or expected < 1_000_000: continue actual = get_bedrock_context_length(f"anthropic.{name.replace('.', '-')}", probe=False) if actual != expected: mismatched.append((name, expected, actual)) mock_probe.assert_not_called() assert not mismatched, f"1M Claude entries missing from BEDROCK_CONTEXT_LENGTHS: {mismatched}" class TestInferenceProfileContextLength: """Application-inference-profile ARNs name no model, so the window must come from the model the profile wraps via GetInferenceProfile — on the production call shape (no region, probe=False).""" ARN = "arn:aws:bedrock:us-west-2:123456789012:application-inference-profile/abcdef123456" def setup_method(self): from agent import bedrock_adapter bedrock_adapter._inference_profile_model_cache.clear() def test_arn_resolves_wrapped_model_window_in_the_arn_region(self): # No region passed (agent/model_metadata.py::_resolve_bedrock_context_length passes none) and # AWS_REGION elsewhere: the lookup must still run, in the ARN's own region, with the # parameter name botocore actually validates (inferenceProfileIdentifier). from agent.bedrock_adapter import get_bedrock_context_length client = MagicMock() client.get_inference_profile.return_value = {"models": [ {"modelArn": "arn:aws:bedrock:us-west-2::foundation-model/anthropic.claude-sonnet-4-6"}]} with patch("agent.bedrock_adapter._get_bedrock_control_client", return_value=client) as factory, \ patch.dict("os.environ", {"AWS_REGION": "us-east-1"}): assert get_bedrock_context_length(self.ARN, probe=False) == 1_000_000 assert get_bedrock_context_length(self.ARN, probe=False) == 1_000_000 # cached per process factory.assert_called_once_with("us-west-2") client.get_inference_profile.assert_called_once_with(inferenceProfileIdentifier=self.ARN) def test_resolution_denied_falls_back_to_default_with_warning(self, caplog): # Without bedrock:GetInferenceProfile the default window applies and the silence is broken # with a WARNING naming the profile and the explicit-config escape hatch. from agent.bedrock_adapter import get_bedrock_context_length, BEDROCK_DEFAULT_CONTEXT_LENGTH client = MagicMock() client.get_inference_profile.side_effect = Exception("AccessDeniedException") with patch("agent.bedrock_adapter._get_bedrock_control_client", return_value=client), \ caplog.at_level("WARNING", logger="agent.bedrock_adapter"): assert get_bedrock_context_length(self.ARN, probe=False) == BEDROCK_DEFAULT_CONTEXT_LENGTH warnings = [r for r in caplog.records if r.levelname == "WARNING"] assert len(warnings) == 1 and self.ARN in warnings[0].getMessage() assert "GetInferenceProfile" in warnings[0].getMessage() def test_profile_wrapping_claude_keeps_prompt_cache_markers(self): # build_converse_kwargs gates cachePoint on the model id; the opaque profile ARN must be # resolved to the wrapped Claude (cached lookup) or the profile silently loses prompt caching. from agent.bedrock_adapter import build_converse_kwargs client = MagicMock() client.get_inference_profile.return_value = {"models": [ {"modelArn": "arn:aws:bedrock:us-west-2::foundation-model/anthropic.claude-sonnet-4-6"}]} messages = [{"role": "system", "content": "Be helpful."}, {"role": "user", "content": "Hi"}] with patch("agent.bedrock_adapter._get_bedrock_control_client", return_value=client): kwargs = build_converse_kwargs(model=self.ARN, messages=messages) assert kwargs["modelId"] == self.ARN # the request still targets the profile assert kwargs["system"][-1] == {"cachePoint": {"type": "default"}} class TestBedrockContextProbe: """Test the live context-window probe that reads the real window from Bedrock's 'prompt is too long' validation error.""" def _client_raising(self, message): client = MagicMock() client.converse.side_effect = Exception(message) return client def test_probe_returns_none_when_client_unavailable(self): from agent.bedrock_adapter import probe_bedrock_context_length with patch("agent.bedrock_adapter._get_bedrock_runtime_client", side_effect=RuntimeError("boto3 missing")): assert probe_bedrock_context_length("any.model", "eu-central-1") is None def test_probe_result_beats_static_table(self): # A successful probe (1M) must override the stale table value (200K # via the 'anthropic.claude-opus-4' substring match). from agent.bedrock_adapter import get_bedrock_context_length err = "prompt is too long: 5000032 tokens > 1000000 maximum" with patch("agent.bedrock_adapter._get_bedrock_runtime_client", return_value=self._client_raising(err)): assert get_bedrock_context_length( "eu.anthropic.claude-opus-4-8", region="eu-central-1") == 1_000_000 # --------------------------------------------------------------------------- # Tool-calling capability detection # --------------------------------------------------------------------------- class TestBuildConverseKwargsToolStripping: """Test that tools are stripped for non-tool-calling models.""" def test_tools_stripped_for_deepseek_r1(self): from agent.bedrock_adapter import build_converse_kwargs tools = [{"type": "function", "function": {"name": "test", "description": "t", "parameters": {}}}] kwargs = build_converse_kwargs( model="us.deepseek.r1-v1:0", messages=[{"role": "user", "content": "Hi"}], tools=tools, ) assert "toolConfig" not in kwargs # --------------------------------------------------------------------------- # Dual-path model routing # --------------------------------------------------------------------------- class TestIsAnthropicBedrockModel: """Test Claude model detection for dual-path routing.""" def test_us_claude_sonnet(self): from agent.bedrock_adapter import is_anthropic_bedrock_model assert is_anthropic_bedrock_model("us.anthropic.claude-sonnet-4-6") is True def test_nova_is_not_anthropic(self): from agent.bedrock_adapter import is_anthropic_bedrock_model assert is_anthropic_bedrock_model("us.amazon.nova-pro-v1:0") is False def test_au_inference_profile(self): from agent.bedrock_adapter import is_anthropic_bedrock_model assert is_anthropic_bedrock_model("au.anthropic.claude-haiku-4-5-20251001-v1:0") is True assert is_anthropic_bedrock_model("au.anthropic.claude-sonnet-4-6") is True class TestEmptyTextBlockFix: """Test that empty/whitespace-only text blocks are replaced with a non-whitespace placeholder (not a literal space, which is itself whitespace and gets rejected by the same Bedrock validation rule).""" def test_none_content_gets_placeholder(self): from agent.bedrock_adapter import _convert_content_to_converse, _EMPTY_TEXT_PLACEHOLDER blocks = _convert_content_to_converse(None) assert blocks[0]["text"] == _EMPTY_TEXT_PLACEHOLDER assert blocks[0]["text"].strip() # --------------------------------------------------------------------------- # Stale-connection detection and per-region client invalidation # --------------------------------------------------------------------------- class TestInvalidateRuntimeClient: """Per-region eviction used to discard dead/stale bedrock-runtime clients.""" def test_evicts_only_the_target_region(self): from agent.bedrock_adapter import ( _bedrock_runtime_client_cache, invalidate_runtime_client, reset_client_cache, ) reset_client_cache() _bedrock_runtime_client_cache["us-east-1"] = "dead-client" _bedrock_runtime_client_cache["us-west-2"] = "live-client" evicted = invalidate_runtime_client("us-east-1") assert evicted is True assert "us-east-1" not in _bedrock_runtime_client_cache assert _bedrock_runtime_client_cache["us-west-2"] == "live-client" def test_returns_false_when_region_not_cached(self): from agent.bedrock_adapter import invalidate_runtime_client, reset_client_cache reset_client_cache() assert invalidate_runtime_client("eu-west-1") is False class TestIsStaleConnectionError: """Classifier that decides whether an exception warrants client eviction.""" def test_detects_botocore_read_timeout(self): pytest.importorskip("botocore.exceptions", reason="botocore (with working exceptions module) required") from agent.bedrock_adapter import is_stale_connection_error from botocore.exceptions import ReadTimeoutError exc = ReadTimeoutError(endpoint_url="https://bedrock.example") assert is_stale_connection_error(exc) is True def test_detects_library_internal_assertion_error(self): """A bare AssertionError raised from inside urllib3/botocore signals a corrupted connection-pool invariant and should trigger eviction.""" from agent.bedrock_adapter import is_stale_connection_error # Fabricate an AssertionError whose traceback's last frame belongs # to a module named "urllib3.connectionpool". We do this by exec'ing # a tiny `assert False` under a fake globals dict — the resulting # frame's ``f_globals["__name__"]`` is what the classifier inspects. fake_globals = {"__name__": "urllib3.connectionpool"} try: exec("def _boom():\n assert False\n_boom()", fake_globals) except AssertionError as exc: assert is_stale_connection_error(exc) is True else: pytest.fail("AssertionError not raised") def test_ignores_unrelated_exceptions(self): from agent.bedrock_adapter import is_stale_connection_error assert is_stale_connection_error(ValueError("bad input")) is False assert is_stale_connection_error(KeyError("missing")) is False class TestCallConverseInvalidatesOnStaleError: """call_converse evicts the cached client only on a stale-connection error — so the next invocation reconnects instead of reusing the dead socket (the agent's streaming path is pinned in ``TestAgentBedrockStreamRecovery``).""" 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") from agent.bedrock_adapter import ( _bedrock_runtime_client_cache, call_converse, reset_client_cache, ) from botocore.exceptions import ClientError reset_client_cache() live_client = MagicMock() live_client.converse.side_effect = ClientError( error_response={"Error": {"Code": "ValidationException", "Message": "bad"}}, operation_name="Converse", ) _bedrock_runtime_client_cache["us-east-1"] = live_client with pytest.raises(ClientError): call_converse( region="us-east-1", model="anthropic.claude-3-sonnet-20240229-v1:0", messages=[{"role": "user", "content": "hi"}], ) assert _bedrock_runtime_client_cache.get("us-east-1") is live_client, ( "validation errors do not indicate a dead connection — keep the client" ) class TestStreamingAccessDeniedDetection: """is_streaming_access_denied_error() recognizes IAM denials of bedrock:InvokeModelWithResponseStream (InvokeModel-only policies).""" def _denied_client_error(self): from botocore.exceptions import ClientError return ClientError( error_response={ "Error": { "Code": "AccessDeniedException", "Message": ( "User: arn:aws:iam::123456789012:user/x is not " "authorized to perform: " "bedrock:InvokeModelWithResponseStream on resource: " "arn:aws:bedrock:us-east-1::foundation-model/" "anthropic.claude-3-sonnet-20240229-v1:0" ), } }, operation_name="ConverseStream", ) def test_matches_access_denied_client_error(self): pytest.importorskip("botocore.exceptions", reason="botocore (with working exceptions module) required") from agent.bedrock_adapter import is_streaming_access_denied_error assert is_streaming_access_denied_error(self._denied_client_error()) is True def test_ignores_unrelated_errors(self): from agent.bedrock_adapter import is_streaming_access_denied_error assert is_streaming_access_denied_error(ValueError("boom")) is False assert is_streaming_access_denied_error( RuntimeError("stream not supported") ) is False class TestAgentBedrockStreamRecovery: """The agent loop streams through ``chat_completion_helpers._bedrock_converse_call``; pin the recovery ladder on that live path: IAM streaming denial → ``_BedrockStream._fall_back_to_converse`` (non-streaming converse, streaming disabled for the session, client kept), stale connection → cached client evicted so the outer retry reconnects.""" _KW = {"__bedrock_region__": "us-east-1", "modelId": "anthropic.claude-3-sonnet-20240229-v1:0", "messages": [{"role": "user", "content": [{"text": "hi"}]}]} def test_streaming_denial_falls_back_to_converse_via_bedrock_stream(self): pytest.importorskip("botocore.exceptions", reason="botocore (with working exceptions module) required") from types import SimpleNamespace from agent.bedrock_adapter import _bedrock_runtime_client_cache, reset_client_cache from agent.chat_completion_helpers import _BedrockStream 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 agent = SimpleNamespace(_disable_streaming=False, _safe_print=MagicMock(), model="m", provider="bedrock") stream = _BedrockStream(agent, dict(self._KW), on_first_delta=None) result = stream._open_stream(dict(self._KW)) client.converse.assert_called_once() assert "__bedrock_region__" not in client.converse.call_args.kwargs assert result.choices[0].message.content == "hi" assert agent._disable_streaming is True assert "InvokeModelWithResponseStream" in agent._safe_print.call_args.args[0] # Not a stale connection — client stays cached. assert _bedrock_runtime_client_cache.get("us-east-1") is client def test_stale_connection_evicts_client_on_agent_stream_path(self): pytest.importorskip("botocore.exceptions", reason="botocore (with working exceptions module) required") from agent.bedrock_adapter import _bedrock_runtime_client_cache, reset_client_cache from agent.chat_completion_helpers import _bedrock_converse_call 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 denied = MagicMock() with pytest.raises(ConnectionClosedError): _bedrock_converse_call(dict(self._KW), stream=True, on_stream_denied=denied) denied.assert_not_called() assert "us-east-1" not in _bedrock_runtime_client_cache # --------------------------------------------------------------------------- # boto3 version check # --------------------------------------------------------------------------- class TestRequireBoto3VersionCheck: """Test that _require_boto3() rejects boto3 versions older than 1.34.59.""" def test_raises_runtime_error_when_boto3_too_old(self): """boto3 < 1.34.59 should raise RuntimeError with upgrade instructions.""" from agent.bedrock_adapter import _require_boto3 fake_boto3 = MagicMock() fake_boto3.__version__ = "1.34.46" with patch.dict("sys.modules", {"boto3": fake_boto3}): with pytest.raises(RuntimeError, match="does not support converse_stream"): _require_boto3() def test_missing_boto3_error_reports_why_the_lazy_install_did_not_land(self, monkeypatch): """A completed install that needs a restart must not tell the user to install it again.""" import pm from agent.bedrock_adapter import _require_boto3 from pm.package import InstallError restart = InstallError("venv", "bedrock installed; restart Hermes to activate the new dependency environment") def ensure_import(extra): raise restart monkeypatch.setattr(pm, "ensure_import", ensure_import) with patch.dict("sys.modules", {"boto3": None}): with pytest.raises(ImportError) as excinfo: _require_boto3() assert str(restart) in str(excinfo.value) assert pm.install_hint("bedrock") not in str(excinfo.value) class TestImageBase64Decoding: """Image data URLs must be decoded to raw bytes before passing to Converse API. boto3 re-encodes at the wire layer, so passing the base64 string directly results in double-encoding. Bedrock rejects with 'Failed to sanitize image'. Ref: #33317. """ def test_data_url_decoded_to_bytes(self): from agent.bedrock_adapter import _convert_content_to_converse import base64 # A tiny 1x1 red PNG raw_png = base64.b64decode( "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8/5+hHgAHggJ/PchI7wAAAABJRU5ErkJggg==" ) data_url = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8/5+hHgAHggJ/PchI7wAAAABJRU5ErkJggg==" content = [{"type": "image_url", "image_url": {"url": data_url}}] blocks = _convert_content_to_converse(content) assert len(blocks) == 1 img_block = blocks[0]["image"] assert img_block["format"] == "png" # Must be raw bytes, not a base64 string assert isinstance(img_block["source"]["bytes"], bytes) assert img_block["source"]["bytes"] == raw_png def test_invalid_base64_falls_back_to_encode(self): from agent.bedrock_adapter import _convert_content_to_converse data_url = "data:image/jpeg;base64,NOT_VALID_BASE64!!!" content = [{"type": "image_url", "image_url": {"url": data_url}}] blocks = _convert_content_to_converse(content) # Should not crash — falls back to encoding the string as bytes assert len(blocks) == 1 assert isinstance(blocks[0]["image"]["source"]["bytes"], bytes) class TestBearerTokenRoutesToConverse: """Bearer Token users must go through Converse API, not AnthropicBedrock SDK. The AnthropicBedrock SDK only supports SigV4 signing — it cannot use AWS_BEARER_TOKEN_BEDROCK. Ref: #28156. """ def _resolve(self, monkeypatch, *, bearer: bool): import os from hermes_cli import runtime_provider as rp if bearer: monkeypatch.setenv("AWS_BEARER_TOKEN_BEDROCK", "test-bearer-token-123") else: monkeypatch.delenv("AWS_BEARER_TOKEN_BEDROCK", raising=False) monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1") assert "AWS_BEARER_TOKEN_BEDROCK" in os.environ or not bearer monkeypatch.setattr( rp, "_get_model_config", lambda: { "default": "us.anthropic.claude-sonnet-4-6", "provider": "bedrock", }, ) monkeypatch.setattr(rp, "load_config", lambda: {"bedrock": {}}) return rp.resolve_runtime_provider(requested="bedrock") def test_bearer_token_forces_converse_for_claude(self, monkeypatch): """Claude model + Bearer Token → bedrock_converse, not anthropic_messages.""" runtime = self._resolve(monkeypatch, bearer=True) assert runtime["api_mode"] == "bedrock_converse" assert "bedrock_anthropic" not in runtime def test_sigv4_claude_still_uses_anthropic_bedrock_sdk(self, monkeypatch): """Without a bearer token, Claude keeps the AnthropicBedrock SDK path.""" runtime = self._resolve(monkeypatch, bearer=False) assert runtime["api_mode"] == "anthropic_messages" assert runtime.get("bedrock_anthropic") is True # --------------------------------------------------------------------------- # Reasoning replay through the Converse tagged union + sealed-blob resend-once (#115865) # --------------------------------------------------------------------------- CROSS_REGION_REJECTION = ( "An error occurred (ValidationException) when calling the ConverseStream operation: The model returned " 'the following errors: {"error":{"code":"validation_error","message":"Encrypted content cannot be used in a ' 'different region from the one that created it.","param":null,"type":"invalid_request_error"}}' ) def _kimi_turn_history(): """Turn 1 as the ConverseStream path captures it: signed thinking, a sealed blob, then a tool call.""" from agent.bedrock_adapter import normalize_converse_stream_events events = [ {"messageStart": {"role": "assistant"}}, {"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"reasoningContent": {"text": "let me think"}}}}, {"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"reasoningContent": {"signature": "sig-1"}}}}, {"contentBlockStop": {"contentBlockIndex": 0}}, {"contentBlockDelta": {"contentBlockIndex": 1, "delta": {"reasoningContent": {"redactedContent": b"sealed"}}}}, {"contentBlockStop": {"contentBlockIndex": 1}}, {"contentBlockStart": {"contentBlockIndex": 2, "start": {"toolUse": {"toolUseId": "t1", "name": "read_file"}}}}, {"contentBlockDelta": {"contentBlockIndex": 2, "delta": {"toolUse": {"input": "{}"}}}}, {"contentBlockStop": {"contentBlockIndex": 2}}, {"messageStop": {"stopReason": "tool_use"}}, {"metadata": {"usage": {"inputTokens": 1, "outputTokens": 1}}}, ] msg = normalize_converse_stream_events({"stream": events}).choices[0].message return [ {"role": "user", "content": "go"}, {"role": "assistant", "content": None, "reasoning_content": msg.reasoning_content, "reasoning_details": msg.reasoning_details, "bedrock_content_blocks": msg.bedrock_content_blocks, "tool_calls": [{"id": "t1", "type": "function", "function": {"name": "read_file", "arguments": "{}"}}]}, {"role": "tool", "tool_call_id": "t1", "content": "ok"}, ] def _ok_converse_response(): return {"output": {"message": {"role": "assistant", "content": [{"text": "done"}]}}, "stopReason": "end_turn", "usage": {"inputTokens": 1, "outputTokens": 1}} class TestReasoningReplaySchema: """Converse ``reasoningContent`` is a tagged union (``reasoningText{text,signature}`` | ``redactedContent``); replaying captured thinking as a bare ``text`` key dies client-side with ParamValidationError (#115865).""" def test_call_converse_replays_thinking_botocore_accepts(self): pytest.importorskip("botocore.session", reason="botocore (bedrock extra) required") import botocore.session from botocore.validate import validate_parameters from agent.bedrock_adapter import call_converse shape = botocore.session.get_session().get_service_model("bedrock-runtime").operation_model("Converse").input_shape client = MagicMock() def converse(**kwargs): validate_parameters(kwargs, shape) # the real client's client-side validation return _ok_converse_response() client.converse.side_effect = converse with patch("agent.bedrock_adapter._get_bedrock_runtime_client", return_value=client): response = call_converse(region="us-east-1", model="global.moonshotai.kimi-k3", messages=_kimi_turn_history()) assert response.choices[0].message.content == "done" replayed = client.converse.call_args.kwargs["messages"][1]["content"] assert replayed[0] == {"reasoningContent": {"reasoningText": {"text": "let me think", "signature": "sig-1"}}} assert replayed[1] == {"reasoningContent": {"redactedContent": b"sealed"}} assert "toolUse" in replayed[2] def test_sync_response_reads_nested_reasoning_text(self): from agent.bedrock_adapter import normalize_converse_response msg = normalize_converse_response({ "output": {"message": {"role": "assistant", "content": [ {"reasoningContent": {"reasoningText": {"text": "hmm", "signature": "s"}}}, {"text": "hi"}]}}, "stopReason": "end_turn", "usage": {"inputTokens": 1, "outputTokens": 1}, }).choices[0].message assert msg.reasoning_content == "hmm" assert msg.bedrock_content_blocks[0] == {"reasoningContent": {"text": "hmm", "signature": "s"}} class TestSealedReasoningResendOnce: """A redacted blob is sealed to the region/model that minted it; a ``global.*`` profile routed elsewhere rejects it with ValidationException. Drop the sealed blocks, keep everything else, resend once (#115865).""" def test_call_converse_strips_sealed_blocks_and_keeps_other_turns_intact(self): from agent.bedrock_adapter import call_converse client = MagicMock() client.converse.side_effect = [Exception(CROSS_REGION_REJECTION), _ok_converse_response()] with patch("agent.bedrock_adapter._get_bedrock_runtime_client", return_value=client): response = call_converse(region="us-east-1", model="global.moonshotai.kimi-k3", messages=_kimi_turn_history()) assert response.choices[0].message.content == "done" assert client.converse.call_count == 2 first, resent = (c.kwargs["messages"] for c in client.converse.call_args_list) assert resent[1]["content"] == [b for b in first[1]["content"] if "redactedContent" not in b.get("reasoningContent", {})] assert resent[0] == first[0] and resent[2] == first[2] # untouched turns are replayed verbatim def test_call_converse_reraises_when_nothing_sealed_remains(self): from agent.bedrock_adapter import call_converse client = MagicMock() client.converse.side_effect = Exception(CROSS_REGION_REJECTION) with patch("agent.bedrock_adapter._get_bedrock_runtime_client", return_value=client): with pytest.raises(Exception, match="ValidationException"): call_converse(region="us-east-1", model="m", messages=[{"role": "user", "content": "hi"}]) assert client.converse.call_count == 1