fix(bedrock): recover from server-side cachePoint rejections per placement
Bedrock's cachePoint rules are per-model-family AND per-field. Amazon Nova
accepts a cachePoint block in `system` and `messages` but rejects it inside
`toolConfig.tools`, failing the whole request with
ValidationException: Malformed input request: #/toolConfig/tools/18:
extraneous key [cachePoint] is not permitted
so every tool-enabled Nova turn fails, with no retry path and no way for the
user to turn cache markers off (#97281).
The adapter decided placement from one static allowlist that answers only
"does this model cache at all", never "in which section". Any family whose
placement rules differ breaks 100% of turns until someone edits the table and
ships a release — the same maintenance trap the `_NON_TOOL_CALLING_PATTERNS`
comment already admits to ("if a model fails with a tool-related
ValidationException, add it here").
Make Bedrock's own verdict authoritative alongside the table: classify the
rejection by the JSON pointer AWS returns, drop the marker for that one
section, retry the request once, and remember the verdict for the rest of the
process so later turns are built clean. The other sections keep their cache
markers, so Nova still gets system/messages caching instead of losing prompt
caching wholesale. This mirrors the module's existing self-heal idiom
(`is_streaming_access_denied_error` → non-streaming `converse()`).
Applied at all four boto3 call sites: `call_converse`, `call_converse_stream`,
and both Bedrock dispatch sites in `chat_completion_helpers` (the streaming
one is the path in the report). A rejection with no marker to strip returns
None so the caller re-raises instead of looping.
Fixes #97281
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Gcoy6nLTg5R6FHHhjcLZEC
This commit is contained in:
joaomarcos
+171
-3
@@ -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<path>[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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user