diff --git a/agent/bedrock_adapter.py b/agent/bedrock_adapter.py index 5e2c082080..359a9441f4 100644 --- a/agent/bedrock_adapter.py +++ b/agent/bedrock_adapter.py @@ -645,6 +645,162 @@ def _model_supports_prompt_cache(model_id: str) -> bool: 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()). + +CACHE_POINT_PLACEMENTS = ("tools", "system", "messages") + +# model_id (lowercased) → placements Bedrock has rejected this process. +_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 +) + + +def cache_point_rejection_placement(exc: BaseException) -> Optional[str]: + """Return the Converse section whose cachePoint block Bedrock refused. + + 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. + """ + msg = str(exc) + lowered = msg.lower() + if "cachepoint" not in lowered: + return None + if "not permitted" not in lowered and "extraneous" not in lowered: + return None + match = _CACHE_POINT_PATH_PATTERN.search(msg) + path = (match.group("path") if match else "").lower() + if "toolconfig" in path or "tools" in path: + return "tools" + if "system" in path: + 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" + + +def note_cache_point_rejection(model_id: str, placement: str) -> None: + """Record that ``model_id`` refuses cachePoint blocks in ``placement``.""" + if placement not in CACHE_POINT_PLACEMENTS: + return + _CACHE_POINT_REJECTIONS.setdefault(model_id.lower(), set()).add(placement) + + +def cache_point_allowed(model_id: str, placement: str) -> bool: + """Return False once Bedrock has refused this placement for this model.""" + return placement not in _CACHE_POINT_REJECTIONS.get(model_id.lower(), ()) + + +def reset_cache_point_rejections() -> None: + """Clear recorded cachePoint rejections. Used in tests.""" + _CACHE_POINT_REJECTIONS.clear() + + +def _is_cache_point_block(block: Any) -> bool: + return isinstance(block, dict) and set(block.keys()) == {"cachePoint"} + + +def strip_cache_points(kwargs: Dict[str, Any], placement: str) -> Dict[str, Any]: + """Return a 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} + + 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): + return kwargs + return {**kwargs, "toolConfig": {**tool_config, "tools": cleaned}} + + if placement == "messages": + messages = kwargs.get("messages") + if not isinstance(messages, list): + return kwargs + 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): + changed = True + cleaned_messages.append({ + **msg, + "content": [b for b in content if not _is_cache_point_block(b)], + }) + else: + cleaned_messages.append(msg) + if not changed: + return kwargs + return {**kwargs, "messages": cleaned_messages} + + return kwargs + + +def recover_from_cache_point_rejection( + exc: BaseException, kwargs: Dict[str, Any] +) -> Optional[Dict[str, Any]]: + """Record Bedrock's cachePoint verdict and return retry kwargs, or None. + + 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. + """ + placement = cache_point_rejection_placement(exc) + if placement is None: + return None + retry_kwargs = strip_cache_points(kwargs, placement) + if retry_kwargs is kwargs: + return None + model_id = str(kwargs.get("modelId", "")) + note_cache_point_rejection(model_id, placement) + logger.warning( + "bedrock: %s rejected a cachePoint block in %s — dropping that cache " + "marker for this model and retrying. Prompt caching stays active for " + "the remaining sections.", + model_id or "model", placement, + ) + return retry_kwargs + + def is_anthropic_bedrock_model(model_id: str) -> bool: """Return True if the model is an Anthropic Claude model on Bedrock. @@ -1238,7 +1394,7 @@ def build_converse_kwargs( } if system_prompt: - if cache_enabled: + if cache_enabled and cache_point_allowed(model, "system"): system_prompt = system_prompt + [{"cachePoint": {"type": "default"}}] kwargs["system"] = system_prompt @@ -1263,7 +1419,7 @@ def build_converse_kwargs( # Strip tools for known non-tool-calling models and warn the user. # Ref: PR #7920 feedback from @ptlally, pattern from PR #4346. if _model_supports_tool_use(model): - if cache_enabled: + if cache_enabled and cache_point_allowed(model, "tools"): converse_tools = converse_tools + [{"cachePoint": {"type": "default"}}] kwargs["toolConfig"] = {"tools": converse_tools} else: @@ -1272,7 +1428,11 @@ def build_converse_kwargs( "The agent will operate in text-only mode.", model ) - if cache_enabled and len(converse_messages) >= 2: + if ( + cache_enabled + 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. @@ -1320,6 +1480,9 @@ def call_converse( try: response = client.converse(**kwargs) except Exception as exc: + retry_kwargs = recover_from_cache_point_rejection(exc, kwargs) + if retry_kwargs is not None: + return normalize_converse_response(client.converse(**retry_kwargs)) if is_stale_connection_error(exc): logger.warning( "bedrock: stale-connection error on converse(region=%s, model=%s): " @@ -1362,6 +1525,11 @@ def call_converse_stream( 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. diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 936d1d7463..b6c46c85d4 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -958,6 +958,7 @@ def _dispatch_nonstreaming_api_request(agent, api_kwargs: dict, *, make_client): invalidate_runtime_client, is_stale_connection_error, normalize_converse_response, + recover_from_cache_point_rejection, ) region = api_kwargs.pop("__bedrock_region__", "us-east-1") api_kwargs.pop("__bedrock_converse__", None) @@ -965,6 +966,15 @@ def _dispatch_nonstreaming_api_request(agent, api_kwargs: dict, *, make_client): try: raw_response = client.converse(**api_kwargs) except Exception as _bedrock_exc: + # A model that refuses cachePoint in one section (Nova rejects it + # inside toolConfig.tools, #97281) fails every turn otherwise — + # drop that marker and resend before surfacing the error. + _retry_kwargs = recover_from_cache_point_rejection( + _bedrock_exc, api_kwargs + ) + if _retry_kwargs is not None: + raw_response = client.converse(**_retry_kwargs) + return normalize_converse_response(raw_response) # Evict the cached client on stale-connection failures # so the outer retry loop builds a fresh client/pool. if is_stale_connection_error(_bedrock_exc): @@ -3438,6 +3448,7 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= is_stale_connection_error, is_streaming_access_denied_error, normalize_converse_response, + recover_from_cache_point_rejection, stream_converse_with_callbacks, ) intercepted_events = [] @@ -3451,6 +3462,17 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= try: raw_response = client.converse_stream(**final_kwargs) except Exception as _bedrock_exc: + # Bedrock refuses a cachePoint block in one section for + # some families (Nova: toolConfig.tools, #97281) and + # fails the whole request. Drop that marker and reopen + # the stream inside the same Relay attempt. + _retry_kwargs = recover_from_cache_point_rejection( + _bedrock_exc, final_kwargs + ) + if _retry_kwargs is not None: + return client.converse_stream(**_retry_kwargs).get( + "stream", [] + ) # InvokeModel-only policies cannot open a stream. Keep # the fallback inside the same managed Relay attempt so # the real provider request and terminal response still diff --git a/tests/agent/test_bedrock_adapter.py b/tests/agent/test_bedrock_adapter.py index ff6a8b1329..0218bb9bc4 100644 --- a/tests/agent/test_bedrock_adapter.py +++ b/tests/agent/test_bedrock_adapter.py @@ -543,6 +543,176 @@ class TestBuildConverseKwargs: 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 # ---------------------------------------------------------------------------