handleStream(response, stream, responseProps) {
    const { uuid = uuidv4(), sources = [] } = responseProps;
    let hasUsageMetrics = false;
    let usage = { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 };

    return new Promise(async (resolve) => {
      let fullText = "";
      let reasoningText = "";

      // Abort handler for client closing connection
      const handleAbort = () => {
        this.#log(`Client closed connection for stream ${uuid}. Aborting.`);
        stream?.endMeasurement(usage); // Finalize metrics
        clientAbortedHandler(resolve, fullText); // Resolve with partial text
      };
      response.on("close", handleAbort);

      try {
        // Process stream chunks
        for await (const chunk of stream.stream) {
          if (!chunk) {
            this.#log("Stream returned null/undefined chunk.");
            continue;
          }
          const action = Object.keys(chunk)[0];

          switch (action) {
            case "metadata": // Contains usage metrics at the end
              if (chunk.metadata?.usage) {
                hasUsageMetrics = true;
                usage = {
                  // Overwrite with final metrics
                  prompt_tokens: chunk.metadata.usage.inputTokens ?? 0,
                  completion_tokens: chunk.metadata.usage.outputTokens ?? 0,
                  total_tokens: chunk.metadata.usage.totalTokens ?? 0,
                };
              }
              break;
            case "contentBlockDelta": {
              // Contains text or reasoning deltas
              const delta = chunk.contentBlockDelta?.delta;
              if (!delta) break;
              const token = delta.text;
              const reasoningToken = delta.reasoningContent?.text;

              if (reasoningToken) {
                // Handle reasoning text
                if (reasoningText.length === 0) {
                  // Start of reasoning block
                  const startTag = "<think>";
                  writeResponseChunk(response, {
                    uuid,
                    sources,
                    type: "textResponseChunk",
                    textResponse: startTag + reasoningToken,
                    close: false,
                    error: false,
                  });
                  reasoningText += startTag + reasoningToken;
                } else {
                  // Continuation of reasoning block
                  writeResponseChunk(response, {
                    uuid,
                    sources,
                    type: "textResponseChunk",
                    textResponse: reasoningToken,
                    close: false,
                    error: false,
                  });
                  reasoningText += reasoningToken;
                }
              } else if (token) {
                // Handle regular text
                if (reasoningText.length > 0) {
                  // If reasoning was just output, close the tag
                  const endTag = "</think>";
                  writeResponseChunk(response, {
                    uuid,
                    sources,
                    type: "textResponseChunk",
                    textResponse: endTag,
                    close: false,
                    error: false,
                  });
                  fullText += reasoningText + endTag; // Add completed reasoning to final text
                  reasoningText = ""; // Reset reasoning buffer
                }
                fullText += token; // Append regular text
                if (!hasUsageMetrics) usage.completion_tokens++; // Estimate usage if no metrics yet
                writeResponseChunk(response, {
                  uuid,
                  sources,
                  type: "textResponseChunk",
                  textResponse: token,
                  close: false,
                  error: false,
                });
              }
              break;
            }
            case "messageStop": // End of message event
              if (chunk.messageStop?.usage) {
                // Check for final metrics here too
                hasUsageMetrics = true;
                usage = {
                  // Overwrite with final metrics if available
                  prompt_tokens:
                    chunk.messageStop.usage.inputTokens ?? usage.prompt_tokens,
                  completion_tokens:
                    chunk.messageStop.usage.outputTokens ??
                    usage.completion_tokens,
                  total_tokens:
                    chunk.messageStop.usage.totalTokens ?? usage.total_tokens,
                };
              }
              // Ensure reasoning tag is closed if message stops mid-reasoning
              if (reasoningText.length > 0) {
                const endTag = "</think>";
                writeResponseChunk(response, {
                  uuid,
                  sources,
                  type: "textResponseChunk",
                  textResponse: endTag,
                  close: false,
                  error: false,
                });
                fullText += reasoningText + endTag;
                reasoningText = "";
              }
              break;
            // Ignore other event types for now
            case "messageStart":
            case "contentBlockStart":
            case "contentBlockStop":
              break;
            default:
              this.#log(`Unhandled stream action: ${action}`, chunk);
          }
        } // End for await loop

        // Final cleanup for reasoning tag in case stream ended abruptly
        if (reasoningText.length > 0 && !fullText.endsWith("</think>")) {
          const endTag = "</think>";
          if (!response.writableEnded) {
            writeResponseChunk(response, {
              uuid,
              sources,
              type: "textResponseChunk",
              textResponse: endTag,
              close: false,
              error: false,
            });
          }
          fullText += reasoningText + endTag;
        }

        // Send final closing chunk to signal end of stream
        if (!response.writableEnded) {
          writeResponseChunk(response, {
            uuid,
            sources,
            type: "textResponseChunk",
            textResponse: "",
            close: true,
            error: false,
          });
        }
      } catch (error) {
        // Handle errors during stream processing
        this.#log(
          `\x1b[43m\x1b[34m[STREAMING ERROR]\x1b[0m ${error.message}`,
          error
        );
        if (response && !response.writableEnded) {
          writeResponseChunk(response, {
            uuid,
            type: "abort",
            textResponse: null,
            sources,
            close: true,
            error: `AWSBedrock:streaming - error. ${
              error?.message ?? "Unknown error"
            }`,
          });
        }
      } finally {
        response.removeListener("close", handleAbort);
        stream?.endMeasurement(usage);
        resolve(fullText); // Resolve with the accumulated text
      }
    });
  }
Handles the stream response from the AWS Bedrock API ConverseStreamCommand.
Parses chunks, handles reasoning tags, and estimates token usage if not provided.
@param {object} response - The HTTP response object to write chunks to.
@param {import('../../helpers/chat/LLMPerformanceMonitor').MonitoredStream} stream - The monitored stream object from streamGetChatCompletion.
@param {object} responseProps - Additional properties for the response chunks.
@param {string} responseProps.uuid - Unique ID for the response.
@param {Array} responseProps.sources - Source documents used (if any).
@returns {Promise<string>} A promise that resolves with the complete text response when the stream ends.