Fix Bedrock parallel tool results and empty arguments

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

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,15 +83,27 @@ 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 ""}],
-                    },
-                }],
-            })
+            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":
+                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
 
         if role == "assistant":
@@ -111,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
@@ -265,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)
@@ -314,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
@@ -366,6 +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
+    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")
@@ -411,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)

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}}},
@@ -656,3 +657,166 @@ 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"])
+    ] + [{"contentBlockStop": {"contentBlockIndex": i}} for i in range(2)]
+      + [{"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"}]
+
+
+@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