From 197257effd496dfb3818990a360d6fac9b3b9e1d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Albert=20Castellana=20LLu=C3=ADs?= Date: Thu, 24 Sep 2026 20:48:11 +0200 Subject: [PATCH 1/2] Fix Bedrock parallel tool results and empty tool arguments --- provider_adapters/bedrock.py | 23 ++++++++++++++--------- tests/test_native_providers.py | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 9 deletions(-) diff --git a/provider_adapters/bedrock.py b/provider_adapters/bedrock.py index 3da6187..c9a8226 100644 --- a/provider_adapters/bedrock.py +++ b/provider_adapters/bedrock.py @@ -79,15 +79,16 @@ def _openai_messages_to_bedrock( tool_use_id = msg.get("tool_call_id") or msg.get("id") if not tool_use_id: continue - out.append({ - "role": "user", - "content": [{ - "toolResult": { - "toolUseId": tool_use_id, - "content": [{"text": text or ""}], - }, - }], - }) + block = {"toolResult": { + "toolUseId": tool_use_id, + "content": [{"text": text or ""}], + }} + # Converse requires parallel results in one following user turn. + if (out and out[-1]["role"] == "user" + and all("toolResult" in b for b in out[-1]["content"])): + out[-1]["content"].append(block) + else: + out.append({"role": "user", "content": [block]}) continue if role == "assistant": @@ -366,6 +367,10 @@ def _tool_call(idx: int) -> dict: return _err(_classify_bedrock_error(exc), 0, _latency(), str(exc)[:500]) tool_calls = [tool_calls_acc[i] for i in sorted(tool_calls_acc)] or None + for call in tool_calls or []: + # A completed no-argument tool may emit no input delta at all. + if call["function"]["arguments"] == "": + call["function"]["arguments"] = "{}" text = "".join(text_parts) if not text.strip() and not tool_calls: return _err("bad_response", 200, _latency(), "empty assistant content") diff --git a/tests/test_native_providers.py b/tests/test_native_providers.py index e1f383f..a39f447 100644 --- a/tests/test_native_providers.py +++ b/tests/test_native_providers.py @@ -656,3 +656,37 @@ async def test_google_native_never_leaks_key_in_url_or_errors(): assert result["ok"] is False assert result["error_kind"] == "timeout" assert "super-secret-key" not in result["error_message"] + + +@pytest.mark.asyncio +async def test_bedrock_parallel_no_argument_tools_roundtrip(): + client = FakeBedrockClient({"stream": [ + {"contentBlockStart": {"contentBlockIndex": i, + "start": {"toolUse": {"toolUseId": f"call_{i}", "name": name}}}} + for i, name in enumerate(["read_context", "learning_query"]) + ] + [{"messageStop": {"stopReason": "tool_use"}}]}) + + async def emit(_): + pass + + result = await stream_bedrock({ + "api_kind": "bedrock", "served_model_id": "sonnet", + "messages": [{"role": "user", "content": "Review evidence"}], + }, emit, env_get={"AWS_REGION": "us-east-1"}.get, client=client) + calls = result["response"]["tool_calls"] + assert [c["function"]["arguments"] for c in calls] == ["{}", "{}"] + from provider_adapters.bedrock import _openai_messages_to_bedrock + messages, _ = _openai_messages_to_bedrock([ + {"role": "user", "content": "Review evidence"}, + {"role": "assistant", "tool_calls": calls}, + {"role": "tool", "tool_call_id": "call_0", "content": "context"}, + {"role": "tool", "tool_call_id": "call_1", "content": "memory"}, + {"role": "assistant", "content": "Done"}, + {"role": "user", "content": "Next turn"}, + ]) + assert len(messages) == 5 + assert messages[2]["role"] == "user" + assert [b["toolResult"]["toolUseId"] for b in messages[2]["content"]] == ["call_0", "call_1"] + assert [b["toolResult"]["content"][0]["text"] for b in messages[2]["content"]] == ["context", "memory"] + assert messages[3]["role"] == "assistant" + assert messages[4]["content"] == [{"text": "Next turn"}] From 2d2b5fdae3ed05d632b10ea4dbcc05e075819d14 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Albert=20Castellana=20LLu=C3=ADs?= Date: Thu, 24 Sep 2026 21:07:10 +0200 Subject: [PATCH 2/2] Reject incomplete Bedrock tool calls and validate result turns --- docs/qualification/BEDROCK-TOOLS-20260924.md | 51 +++++++ provider_adapters/bedrock.py | 68 ++++++++-- tests/test_native_providers.py | 132 ++++++++++++++++++- 3 files changed, 240 insertions(+), 11 deletions(-) create mode 100644 docs/qualification/BEDROCK-TOOLS-20260924.md diff --git a/docs/qualification/BEDROCK-TOOLS-20260924.md b/docs/qualification/BEDROCK-TOOLS-20260924.md new file mode 100644 index 0000000..04265eb --- /dev/null +++ b/docs/qualification/BEDROCK-TOOLS-20260924.md @@ -0,0 +1,51 @@ +# Bedrock tool protocol review — 24 September 2026 + +PR #123 is a general OpenAI-to-Converse adapter fix. No agent name, trading rule, +wallet, or application-specific branch exists in the adapter. + +## Independent model review + +Unhardcoded reported `openrouter_market`, served model +`anthropic/claude-opus-5.5`. The first non-streaming request returned 504 without +review text. A streaming review was truncated (`finish_reason: length`); it +identified adjacent user turns after tool results. A second streaming review +completed (`finish_reason: stop`) and recommended changes. Reported costs for +the two successful requests: $0.106366 and $0.105636. The failed request's cost +is unknown; conservative reservations remain in TraderSwarm's ledger. + +The reviewer received the complete adapter and diff. Its second review preceded +the final corrections below; this is independent review evidence, not model +approval of the final commit or live AWS validation. + +Accepted and implemented: + +- Merge consecutive user blocks, with tool results preceding ordinary text. +- Reject results without a matching preceding assistant tool use, in both call paths. +- Return tool calls only after `tool_use` and closure of every tool block. +- Normalize absent arguments only for completed calls; reject malformed, + non-object, and non-finite JSON. +- Report `stream_interrupted` when text was already emitted before an incomplete + tool response; return no executable tool calls from the failed batch. + +Not adopted: defaulting an absent contentBlockStop index to zero. The API requires +that index; malformed input should fail closed rather than invent a block identity. + +Pre-existing, outside this PR: full non-streaming response validation, duplicate +model-generated tool IDs, and preservation of reasoning blocks/signatures. +These are not proved correct by this patch. + +## Validation + +56 native-provider, streaming and provider tests pass. Eight negative cases fail +against the initial PR and pass after hardening. Tests include parallel results +in different orders, result/text mixtures, an orphan result, incomplete parallel +batches, missing completion, token limits, invalid JSON and emitted text before +failure. Provider calls are mocked; the corrected adapter is not yet deployed or +verified with live Bedrock. + +References: + +- https://docs.aws.amazon.com/bedrock/latest/userguide/bedrock-runtime_example_bedrock-runtime_Scenario_ToolUse_section.html +- https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_MessageStopEvent.html +- https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_ContentBlockStopEvent.html +- https://platform.claude.com/docs/en/agents-and-tools/tool-use/handle-tool-calls diff --git a/provider_adapters/bedrock.py b/provider_adapters/bedrock.py index c9a8226..7326bcf 100644 --- a/provider_adapters/bedrock.py +++ b/provider_adapters/bedrock.py @@ -61,6 +61,10 @@ def _bedrock_model_id(request: dict) -> str: or request["served_model_id"]) +def _reject_json_constant(value: str): + raise ValueError(f"non-finite JSON constant: {value}") + + def _openai_messages_to_bedrock( messages: list[dict], ) -> tuple[list[dict], list[dict] | None]: @@ -79,14 +83,25 @@ def _openai_messages_to_bedrock( tool_use_id = msg.get("tool_call_id") or msg.get("id") if not tool_use_id: continue + previous = out[-1] if out else {} + if previous.get("role") == "user": + previous = out[-2] if len(out) > 1 else {} + expected = { + b["toolUse"]["toolUseId"] for b in previous.get("content", []) + if "toolUse" in b + } if previous.get("role") == "assistant" else set() + if tool_use_id not in expected: + raise ValueError("tool result has no matching preceding tool use") block = {"toolResult": { "toolUseId": tool_use_id, "content": [{"text": text or ""}], }} # Converse requires parallel results in one following user turn. - if (out and out[-1]["role"] == "user" - and all("toolResult" in b for b in out[-1]["content"])): - out[-1]["content"].append(block) + if out and out[-1]["role"] == "user": + content = out[-1]["content"] + # Claude-backed Converse requires results before ordinary text. + position = next((i for i, b in enumerate(content) if "toolResult" not in b), len(content)) + content.insert(position, block) else: out.append({"role": "user", "content": [block]}) continue @@ -112,7 +127,10 @@ def _openai_messages_to_bedrock( continue if text: - out.append({"role": "user", "content": [{"text": text}]}) + if out and out[-1]["role"] == "user": + out[-1]["content"].append({"text": text}) + else: + out.append({"role": "user", "content": [{"text": text}]}) system = [{"text": "\n\n".join(system_parts)}] if system_parts else None return out, system @@ -266,12 +284,16 @@ async def stream_bedrock( region = _aws_region(request, _env_get) bedrock = client or (client_factory(region) if client_factory else _bedrock_client(region, timeout_s, _env_get)) - body = _bedrock_request(request) + try: + body = _bedrock_request(request) + except ValueError as exc: + return _err("bad_request", 400, 0, str(exc)) t0 = time.monotonic() saw_output = False emitted = False text_parts: list[str] = [] tool_calls_acc: dict[int, dict] = {} + closed_tool_blocks: set[int] = set() finish_reason = None usage: dict = {} first_timeout_s = first_token_timeout_s(request) @@ -315,6 +337,11 @@ def _tool_call(idx: int) -> dict: finish_reason = (event["messageStop"] or {}).get("stopReason") \ or finish_reason continue + if "contentBlockStop" in event: + idx = (event["contentBlockStop"] or {}).get("contentBlockIndex") + if idx in tool_calls_acc: + closed_tool_blocks.add(idx) + continue if "metadata" in event: usage = (event["metadata"] or {}).get("usage") or usage continue @@ -367,10 +394,28 @@ def _tool_call(idx: int) -> dict: return _err(_classify_bedrock_error(exc), 0, _latency(), str(exc)[:500]) tool_calls = [tool_calls_acc[i] for i in sorted(tool_calls_acc)] or None - for call in tool_calls or []: - # A completed no-argument tool may emit no input delta at all. - if call["function"]["arguments"] == "": - call["function"]["arguments"] = "{}" + if tool_calls: + failure_kind = "stream_interrupted" if emitted else "bad_response" + # Never turn an interrupted or token-limited tool request into an action. + if finish_reason != "tool_use" or set(tool_calls_acc) != closed_tool_blocks: + return _err(failure_kind, 200, _latency(), + "incomplete Bedrock tool-use response") + for call in tool_calls: + if not call["id"] or not call["function"]["name"]: + return _err(failure_kind, 200, _latency(), + "missing Bedrock tool identity") + # A closed no-argument tool may emit no input delta at all. + arguments = call["function"]["arguments"] + if arguments == "": + arguments = "{}" + try: + parsed = json.loads(arguments, parse_constant=_reject_json_constant) + if not isinstance(parsed, dict): + raise ValueError("tool arguments must be an object") + except (ValueError, TypeError): + return _err(failure_kind, 200, _latency(), + "invalid Bedrock tool arguments JSON") + call["function"]["arguments"] = arguments text = "".join(text_parts) if not text.strip() and not tool_calls: return _err("bad_response", 200, _latency(), "empty assistant content") @@ -416,7 +461,10 @@ async def call(request: dict) -> dict: request, ignore_delta, env_get=_env_get, timeout_s=timeout_s, client=client, client_factory=client_factory) - body = _bedrock_request(request) + try: + body = _bedrock_request(request) + except ValueError as exc: + return _err("bad_request", 400, 0, str(exc)) region = _aws_region(request, _env_get) bedrock = _client(region) diff --git a/tests/test_native_providers.py b/tests/test_native_providers.py index a39f447..9064ae6 100644 --- a/tests/test_native_providers.py +++ b/tests/test_native_providers.py @@ -408,6 +408,7 @@ async def test_bedrock_native_stream_emits_and_aggregates(): "delta": {"toolUse": { "input": '{"id": "abc"}', }}}}, + {"contentBlockStop": {"contentBlockIndex": 1}}, {"messageStop": {"stopReason": "tool_use"}}, {"metadata": {"usage": {"inputTokens": 11, "outputTokens": 7, "totalTokens": 18}}}, @@ -664,7 +665,8 @@ async def test_bedrock_parallel_no_argument_tools_roundtrip(): {"contentBlockStart": {"contentBlockIndex": i, "start": {"toolUse": {"toolUseId": f"call_{i}", "name": name}}}} for i, name in enumerate(["read_context", "learning_query"]) - ] + [{"messageStop": {"stopReason": "tool_use"}}]}) + ] + [{"contentBlockStop": {"contentBlockIndex": i}} for i in range(2)] + + [{"messageStop": {"stopReason": "tool_use"}}]}) async def emit(_): pass @@ -690,3 +692,131 @@ async def emit(_): assert [b["toolResult"]["content"][0]["text"] for b in messages[2]["content"]] == ["context", "memory"] assert messages[3]["role"] == "assistant" assert messages[4]["content"] == [{"text": "Next turn"}] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("stop,closed,args,valid", [ + (None, False, "", False), + (None, True, "", False), + ("max_tokens", True, "", False), + ("max_tokens", True, "{}", False), + ("end_turn", True, "{}", False), + ("tool_use", False, "{}", False), + ("tool_use", True, '{"city":', False), + ("tool_use", True, " ", False), + ("tool_use", True, "[]", False), + ("tool_use", True, "null", False), + ("tool_use", True, "42", False), + ("tool_use", True, '{"value": NaN}', False), + ("tool_use", True, '{"value": Infinity}', False), + ("tool_use", True, "", True), + ("tool_use", True, '{"city":"Madrid"}', True), +]) +async def test_bedrock_only_complete_tool_calls_are_returned(stop, closed, args, valid): + events = [{"contentBlockStart": {"contentBlockIndex": 0, "start": { + "toolUse": {"toolUseId": "weather_1", "name": "weather"}}}}] + if args: + # Preserve actual fragment concatenation; do not repair partial JSON. + for fragment in [args[:len(args)//2], args[len(args)//2:]]: + events.append({"contentBlockDelta": {"contentBlockIndex": 0, + "delta": {"toolUse": {"input": fragment}}}}) + if closed: + events.append({"contentBlockStop": {"contentBlockIndex": 0}}) + if stop: + events.append({"messageStop": {"stopReason": stop}}) + async def emit(_): + pass + result = await stream_bedrock({ + "api_kind": "bedrock", "served_model_id": "test", + "messages": [{"role": "user", "content": "Weather?"}], + }, emit, client=FakeBedrockClient({"stream": events})) + assert result["ok"] is valid + if valid: + assert result["response"]["tool_calls"][0]["function"]["arguments"] == (args or "{}") + else: + assert result["error_kind"] == "bad_response" + assert "response" not in result + + +def test_bedrock_parallel_results_preserve_ids_order_and_turn_boundaries(): + from provider_adapters.bedrock import _openai_messages_to_bedrock + calls = [{"id": str(i), "function": {"name": "lookup", "arguments": "{}"}} + for i in range(3)] + messages, _ = _openai_messages_to_bedrock([ + {"role": "assistant", "tool_calls": calls}, + *[{"role": "tool", "tool_call_id": str(i), "content": str(i)} for i in [2, 0, 1]], + {"role": "user", "content": "A separate user message"}, + {"role": "assistant", "tool_calls": [calls[0]]}, + {"role": "tool", "tool_call_id": "0", "content": "next result"}, + ]) + assert len(messages) == 4 + assert [b["toolResult"]["toolUseId"] for b in messages[1]["content"][:3]] == ["2", "0", "1"] + assert messages[1]["content"][3] == {"text": "A separate user message"} + assert messages[2]["role"] == "assistant" + assert messages[3]["content"][0]["toolResult"]["content"] == [{"text": "next result"}] + + +@pytest.mark.asyncio +async def test_bedrock_incomplete_parallel_call_rejects_entire_batch(): + events = [] + for i in range(2): + events.append({"contentBlockStart": {"contentBlockIndex": i, + "start": {"toolUse": {"toolUseId": str(i), "name": "lookup"}}}}) + events.extend([{"contentBlockStop": {"contentBlockIndex": 0}}, + {"messageStop": {"stopReason": "tool_use"}}]) + async def emit(_): + pass + result = await stream_bedrock({ + "api_kind": "bedrock", "served_model_id": "test", + "messages": [{"role": "user", "content": "Lookup"}], + }, emit, client=FakeBedrockClient({"stream": events})) + assert result["ok"] is False + assert "response" not in result + + +def test_bedrock_user_text_before_results_keeps_block_order(): + from provider_adapters.bedrock import _openai_messages_to_bedrock + messages, _ = _openai_messages_to_bedrock([ + {"role": "assistant", "tool_calls": [{"id": "id1", "function": {"name": "lookup", "arguments": "{}"}}]}, + {"role": "user", "content": "context"}, + {"role": "tool", "tool_call_id": "id1", "content": "result"}, + {"role": "user", "content": "reminder"}, + ]) + assert len(messages) == 2 + assert messages[1]["content"][0]["toolResult"]["toolUseId"] == "id1" + assert messages[1]["content"][1] == {"text": "context"} + assert messages[1]["content"][2] == {"text": "reminder"} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streaming", [False, True]) +async def test_bedrock_orphan_tool_result_fails_before_provider(streaming): + client = FakeBedrockClient({}) + call = make_bedrock_async_call_provider(client=client) + request = {"api_kind": "bedrock", "served_model_id": "test", "messages": [ + {"role": "user", "content": "hello"}, + {"role": "tool", "tool_call_id": "orphan", "content": "result"}]} + if streaming: + request["first_token_timeout_ms"] = 1000 + result = await call(request) + assert result["ok"] is False + assert result["error_kind"] == "bad_request" + assert client.requests == [] + + +@pytest.mark.asyncio +async def test_bedrock_text_then_incomplete_tool_is_interrupted(): + client = FakeBedrockClient({"stream": [ + {"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": "Checking"}}}, + {"contentBlockStart": {"contentBlockIndex": 1, "start": { + "toolUse": {"toolUseId": "id1", "name": "lookup"}}}}, + ]}) + emitted = [] + async def emit(text): + emitted.append(text) + result = await stream_bedrock({"api_kind": "bedrock", "served_model_id": "test", + "messages": [{"role": "user", "content": "hi"}]}, + emit, client=client) + assert emitted == ["Checking"] + assert result["error_kind"] == "stream_interrupted" + assert "response" not in result