@smithy/eventstream-codec Error: "Reported message length does not match received message length" when decoding converseStream API response in node

[stephmilovic]

Description:

I'm working with LangChain + Bedrock and trying to switch from the Invoke APIs to the Converse APIs. I am hitting the Bedrock APIs directly with axios and handing the result back to LangChain's NodeHttpHandler and BedrockRuntimeClient to handle the results.

The issue I'm having is in is a different spot in my app where I need to decode the raw stream from the API in order to get the usage object. When using the InvokeWithResponseStream API, I'm able to use the @smithy/eventstream-codec to decode the final response chunk to get the usage object. However, with the ConverseStream API an issue occurs intermittently, with some chunks decoding successfully and others failing with the error:

Error: Reported message length does not match received message length
    at splitMessage (/node_modules/@smithy/eventstream-codec/dist-cjs/index.js:276:11)
    at _EventStreamCodec.decode (/node_modules/@smithy/eventstream-codec/dist-cjs/index.js:367:31)

The same decoding logic works reliably with the InvokeWithResponseStream API, so the issue appears specific to ConverseStream. I suspect there may be special encoding applied to ConverseStream, but I could not find documentation about how to handle or decode these streams properly.

---

Environment:

• Node.js version: v20.15.1
• @smithy/eventstream-codec version: 3.1.9
• Bedrock model: anthropic.claude-3-5-sonnet-20240620-v1:0

---

Steps to Reproduce:

1. Use the converseStream API in AWS Bedrock to receive a streaming response.
2. Attempt to decode the stream using the following code:

   import { EventStreamCodec } from "@smithy/eventstream-codec";
   import { finished } from "stream/promises";
   import { toUtf8, fromUtf8 } from "@aws-sdk/util-utf8-node";
   
   export const parseBedrockConverseStream = async (responseStream) => {
     const responseBuffer = [];
     responseStream.on("data", (chunk) => {
       responseBuffer.push(chunk);
     });
     await finished(responseStream);
   
     const awsDecoder = new EventStreamCodec(toUtf8, fromUtf8);
     const finalChunk = responseBuffer[responseBuffer.length - 1];
   
     try {
       console.log("debugGoHere finalChunk", finalChunk);
       const event = awsDecoder.decode(finalChunk);
       console.log("debugGoHere event", event);
       const decoded = new TextDecoder().decode(event.body);
       console.log("debugGoHere decoded", decoded);
       const usage = JSON.parse(decoded).usage;
       return usage
         ? {
             total_tokens: usage.totalTokens,
             prompt_tokens: usage.inputTokens,
             completion_tokens: usage.outputTokens,
           }
         : null;
     } catch (e) {
       const regularDecoded = new TextDecoder().decode(finalChunk);
       console.log("debugGoHere error regularDecoded", regularDecoded);
       console.log("debugGoHere error", e);
       return null;
     }
   };

---

Observed Behavior:

• Successful Decoding:

  debugGoHere finalChunk <Buffer 00 00 00 c9 00 00 00 4e ... 151 more bytes>
  debugGoHere event {
   headers: {
     ':event-type': { type: 'string', value: 'metadata' },
     ':content-type': { type: 'string', value: 'application/json' },
     ':message-type': { type: 'string', value: 'event' }
   },
   body: Uint8Array(107) [
     123,  34, 109, 101, 116, 114, 105,  99, 115,  34,  58, 123,
     ... 
   ]
  }
  debugGoHere decoded {"metrics":{"latencyMs":1694},"p":"abcd","usage:{"inputTokens":1560,"outputTokens":47,"totalTokens":1607}}
  debugGoHere usage { inputTokens: 1560, outputTokens: 47, totalTokens: 1607 }
  

• Failed Decoding:

  debugGoHere finalChunk <Buffer 00 00 00 be 00 00 00 56 48  ... 504 more bytes>
  debugGoHere error regularDecoded �VHݝ�
  :message-typeevent{"contentBlockIndex":0,"p":"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ012345"}씐��Q�)�=
                                                                                                                    :event-type
  :message-typeevent{"p":"abcdefghijklmnopqrstuvwxyzABCDEFGHIJK","stopReason":"end_turn"}]9�N���                                messageStop
  :message-typeevent{"metrics":{"latencyMs":949},"p":"abcde","usage":{"inputTokens":78,"outputTokens":47,"totalTokens":125}}�#�4
  debugGoHere error Error: Reported message length does not match received message length
     at splitMessage (/Users/stephmilovic/dev/kibana/node_modules/@smithy/eventstream-codec/dist-cjs/index.js:276:11)
     at _EventStreamCodec.decode (/Users/stephmilovic/dev/kibana/node_modules/@smithy/eventstream-codec/dist-cjs/index.js:367:31)
     at parseBedrockConverseStream
  

---

Troubleshooting Steps Taken:

1. Concatenated Chunks: Combined all chunks into a single Uint8Array to ensure message continuity.
2. Logging: Examined raw chunks and attempted manual decoding using TextDecoder to verify structure. Can see in the failed decoded chunks, the usage object exists but I cannot parse it out of these strings which appear to be a mix of binary data and JSON-like structures.
3. Alternate API: Confirmed the same code works reliably for invokeStream, suggesting the issue is specific to converseStream.

---

Questions:

1. Is there any special encoding or format used for the converseStream API that differs from invokeStream?
2. Is there documentation on how to properly decode the converseStream API response using @smithy/eventstream-codec?
3. Are there known issues with fragmented or malformed messages in the converseStream API?

---

Request:

Please provide guidance on decoding the converseStream API response or clarify if there are specific differences in encoding for this API. Let me know if additional logs, raw data, or debugging information is needed.

[pagood]

Running into same issue, did you figure it out?