BareGit

Implement streaming LLM completions

Translate SSE frames into typed events while HTTP remains active. Buffer tool
arguments until completion and cover streaming behavior with mock tests.
Author: MetroWind <chris.corsair@gmail.com>
Date: Tue Jul 21 18:56:55 2026 -0700
Commit: 14361bdfb9d34584be56fd09adf8d12b3c915595

Changes

diff --git a/include/agent_smith/agent_smith.hpp b/include/agent_smith/agent_smith.hpp
index 152b11b..7548b5d 100644
--- a/include/agent_smith/agent_smith.hpp
+++ b/include/agent_smith/agent_smith.hpp
@@ -5,4 +5,5 @@
 #include "llm_client.hpp"
 #include "message.hpp"
 #include "runtime.hpp"
+#include "stream_event.hpp"
 #include "version.hpp"
diff --git a/include/agent_smith/llm_client.hpp b/include/agent_smith/llm_client.hpp
index 996b5b9..f8f17bc 100644
--- a/include/agent_smith/llm_client.hpp
+++ b/include/agent_smith/llm_client.hpp
@@ -12,6 +12,7 @@
 
 #include "message.hpp"
 #include "runtime.hpp"
+#include "stream_event.hpp"
 
 namespace agent_smith
 {
@@ -91,6 +92,11 @@ public:
     /// Send a non-streaming request to `/v1/chat/completions`.
     mw::Task<mw::E<LlmResponse>> complete(LlmRequest request);
 
+    /// Send a streaming request and synchronously deliver typed stream events.
+    mw::Task<mw::E<LlmResponse>>
+    completeStream(LlmRequest request,
+                   std::function<void(const StreamEvent&)> on_event);
+
 private:
     Runtime* runtime_;
     mw::HTTPSessionAsyncInterface* http_;
diff --git a/include/agent_smith/stream_event.hpp b/include/agent_smith/stream_event.hpp
new file mode 100644
index 0000000..598bc38
--- /dev/null
+++ b/include/agent_smith/stream_event.hpp
@@ -0,0 +1,58 @@
+#pragma once
+
+#include <optional>
+#include <string>
+#include <variant>
+
+#include <nlohmann/json.hpp>
+
+#include "message.hpp"
+
+namespace agent_smith
+{
+
+/// A text fragment received from an LLM streaming response.
+struct StreamTokenDelta
+{
+    /// The newly generated text fragment.
+    std::string text;
+};
+
+/// A complete tool-call identity received from an LLM streaming response.
+struct StreamToolCallStarted
+{
+    /// Server-generated identifier for this tool invocation.
+    std::string id;
+
+    /// Name of the requested tool.
+    std::string name;
+
+    /// Position of the tool call in the streamed response.
+    int index = 0;
+};
+
+/// A tool call whose buffered arguments have become valid JSON.
+struct StreamToolCallCompleted
+{
+    /// Position of the tool call in the streamed response.
+    int index = 0;
+
+    /// Fully parsed function arguments.
+    nlohmann::json arguments;
+};
+
+/// Marks the final event from an LLM streaming response.
+struct StreamFinished
+{
+    /// Provider-supplied reason the response finished.
+    std::string finish_reason;
+
+    /// Token accounting emitted by the provider, when available.
+    std::optional<TokenUsage> usage;
+};
+
+/// One typed event emitted while an LLM streaming response is active.
+using StreamEvent = std::variant<StreamTokenDelta, StreamToolCallStarted,
+                                 StreamToolCallCompleted, StreamFinished>;
+
+} // namespace agent_smith
diff --git a/src/llm_client.cpp b/src/llm_client.cpp
index 0c78892..29d9154 100644
--- a/src/llm_client.cpp
+++ b/src/llm_client.cpp
@@ -6,11 +6,19 @@
 #include <charconv>
 #include <chrono>
 #include <cstdint>
+#include <exception>
+#include <iterator>
+#include <limits>
 #include <random>
 #include <ranges>
+#include <span>
 #include <string_view>
 #include <utility>
 
+#include "agent_smith/detail/async_queue.hpp"
+#include "agent_smith/detail/sse_parser.hpp"
+#include "agent_smith/detail/started_task.hpp"
+
 namespace agent_smith
 {
 namespace
@@ -77,7 +85,8 @@ mw::E<void> mergeExtra(nlohmann::json& target, const nlohmann::json& extra)
 }
 
 mw::E<nlohmann::json> buildRequestBody(const LlmRequest& request,
-                                       const LlmClientConfig& config)
+                                       const LlmClientConfig& config,
+                                       bool streaming)
 {
     SamplingParams sampling = config.default_sampling;
     applySamplingOverrides(sampling, request.sampling);
@@ -85,7 +94,7 @@ mw::E<nlohmann::json> buildRequestBody(const LlmRequest& request,
     nlohmann::json body{
         {"model", config.model},
         {"messages", nlohmann::json::array()},
-        {"stream", false},
+        {"stream", streaming},
     };
     for(const Message& message : request.messages)
     {
@@ -125,6 +134,430 @@ mw::E<nlohmann::json> buildRequestBody(const LlmRequest& request,
     return body;
 }
 
+mw::E<std::uint64_t> tokenCount(const nlohmann::json& json,
+                                std::string_view key)
+{
+    const auto found = json.find(key);
+    if(found == json.end())
+    {
+        return std::unexpected(
+            error::parse("expected unsigned token count: " + std::string(key)));
+    }
+    if(found->is_number_unsigned())
+    {
+        return found->get<std::uint64_t>();
+    }
+    if(found->is_number_integer())
+    {
+        const std::int64_t value = found->get<std::int64_t>();
+        if(value >= 0)
+        {
+            return static_cast<std::uint64_t>(value);
+        }
+    }
+    return std::unexpected(
+        error::parse("expected unsigned token count: " + std::string(key)));
+}
+
+mw::E<TokenUsage> usageFromOpenAIJson(const nlohmann::json& json)
+{
+    if(!json.is_object())
+    {
+        return std::unexpected(error::parse("expected usage object"));
+    }
+
+    auto prompt = tokenCount(json, "prompt_tokens");
+    if(!prompt)
+    {
+        return std::unexpected(prompt.error());
+    }
+    auto completion = tokenCount(json, "completion_tokens");
+    if(!completion)
+    {
+        return std::unexpected(completion.error());
+    }
+    auto total = tokenCount(json, "total_tokens");
+    if(!total)
+    {
+        return std::unexpected(total.error());
+    }
+    return TokenUsage{
+        .prompt = *prompt,
+        .completion = *completion,
+        .total = *total,
+    };
+}
+
+mw::E<int> toolCallIndex(const nlohmann::json& json)
+{
+    if(json.is_number_unsigned())
+    {
+        const std::uint64_t value = json.get<std::uint64_t>();
+        if(value <= static_cast<std::uint64_t>(std::numeric_limits<int>::max()))
+        {
+            return static_cast<int>(value);
+        }
+    }
+    else if(json.is_number_integer())
+    {
+        const std::int64_t value = json.get<std::int64_t>();
+        if(value >= 0 && value <= std::numeric_limits<int>::max())
+        {
+            return static_cast<int>(value);
+        }
+    }
+    return std::unexpected(
+        error::parse("expected non-negative tool call index"));
+}
+
+class StreamDeltaParser
+{
+public:
+    mw::E<std::vector<StreamEvent>> feed(std::string_view payload)
+    {
+        const nlohmann::json json =
+            nlohmann::json::parse(payload, nullptr, false);
+        if(json.is_discarded() || !json.is_object())
+        {
+            return std::unexpected(error::parse("expected JSON stream chunk"));
+        }
+
+        const auto usage = json.find("usage");
+        if(usage != json.end() && !usage->is_null())
+        {
+            auto parsed_usage = usageFromOpenAIJson(*usage);
+            if(!parsed_usage)
+            {
+                return std::unexpected(parsed_usage.error());
+            }
+            message_.usage = std::move(*parsed_usage);
+        }
+
+        const auto choices = json.find("choices");
+        if(choices == json.end())
+        {
+            if(usage != json.end() && !usage->is_null())
+            {
+                return std::vector<StreamEvent>{};
+            }
+            return std::unexpected(
+                error::parse("expected stream choices array"));
+        }
+        if(!choices->is_array())
+        {
+            return std::unexpected(
+                error::parse("expected stream choices array"));
+        }
+        if(choices->empty())
+        {
+            return std::vector<StreamEvent>{};
+        }
+
+        const nlohmann::json& choice = choices->front();
+        if(!choice.is_object())
+        {
+            return std::unexpected(
+                error::parse("expected stream choice object"));
+        }
+
+        std::vector<StreamEvent> events;
+        const auto delta = choice.find("delta");
+        if(delta != choice.end() && !delta->is_null())
+        {
+            if(!delta->is_object())
+            {
+                return std::unexpected(
+                    error::parse("expected stream delta object"));
+            }
+            auto delta_events = parseDelta(*delta);
+            if(!delta_events)
+            {
+                return std::unexpected(delta_events.error());
+            }
+            events = std::move(*delta_events);
+        }
+
+        const auto finish_reason = choice.find("finish_reason");
+        if(finish_reason != choice.end() && !finish_reason->is_null())
+        {
+            if(!finish_reason->is_string())
+            {
+                return std::unexpected(
+                    error::parse("expected stream finish reason string"));
+            }
+            finish_reason_ = finish_reason->get<std::string>();
+        }
+        return events;
+    }
+
+    mw::E<std::vector<StreamEvent>> finish()
+    {
+        if(finished_)
+        {
+            return std::vector<StreamEvent>{};
+        }
+
+        std::vector<StreamEvent> events;
+        for(std::size_t index = 0; index < tool_calls_.size(); ++index)
+        {
+            PartialToolCall& partial = tool_calls_[index];
+            if(!partial.seen)
+            {
+                continue;
+            }
+            if(partial.id.empty() || partial.name.empty())
+            {
+                return std::unexpected(
+                    error::parse("incomplete streamed tool call identity"));
+            }
+
+            nlohmann::json arguments =
+                nlohmann::json::parse(partial.arguments_buffer, nullptr, false);
+            if(arguments.is_discarded() || !arguments.is_object())
+            {
+                return std::unexpected(error::parse(
+                    "expected streamed tool arguments JSON object"));
+            }
+            message_.tool_calls.push_back(ToolCall{
+                .id = std::move(partial.id),
+                .name = std::move(partial.name),
+                .arguments = arguments,
+            });
+            events.push_back(StreamToolCallCompleted{
+                .index = static_cast<int>(index),
+                .arguments = std::move(arguments),
+            });
+        }
+        events.push_back(StreamFinished{
+            .finish_reason = finish_reason_.value_or(""),
+            .usage = message_.usage,
+        });
+        finished_ = true;
+        return events;
+    }
+
+    LlmResponse takeResponse()
+    {
+        return LlmResponse{
+            .message = std::move(message_),
+        };
+    }
+
+private:
+    struct PartialToolCall
+    {
+        std::string id;
+        std::string name;
+        std::string arguments_buffer;
+        bool seen = false;
+        bool started = false;
+    };
+
+    mw::E<std::vector<StreamEvent>> parseDelta(const nlohmann::json& delta)
+    {
+        std::vector<StreamEvent> events;
+        const auto content = delta.find("content");
+        if(content != delta.end() && !content->is_null())
+        {
+            if(!content->is_string())
+            {
+                return std::unexpected(
+                    error::parse("expected stream content string"));
+            }
+            std::string text = content->get<std::string>();
+            message_.content.append(text);
+            if(!text.empty())
+            {
+                events.push_back(StreamTokenDelta{
+                    .text = std::move(text),
+                });
+            }
+        }
+
+        const auto tool_calls = delta.find("tool_calls");
+        if(tool_calls == delta.end() || tool_calls->is_null())
+        {
+            return events;
+        }
+        if(!tool_calls->is_array())
+        {
+            return std::unexpected(
+                error::parse("expected stream tool_calls array"));
+        }
+
+        for(const nlohmann::json& tool_call : *tool_calls)
+        {
+            auto tool_events = parseToolCall(tool_call);
+            if(!tool_events)
+            {
+                return std::unexpected(tool_events.error());
+            }
+            events.insert(events.end(),
+                          std::make_move_iterator(tool_events->begin()),
+                          std::make_move_iterator(tool_events->end()));
+        }
+        return events;
+    }
+
+    mw::E<std::vector<StreamEvent>>
+    parseToolCall(const nlohmann::json& tool_call)
+    {
+        if(!tool_call.is_object())
+        {
+            return std::unexpected(
+                error::parse("expected stream tool call object"));
+        }
+        const auto index = tool_call.find("index");
+        if(index == tool_call.end())
+        {
+            return std::unexpected(
+                error::parse("expected stream tool call index"));
+        }
+        auto parsed_index = toolCallIndex(*index);
+        if(!parsed_index)
+        {
+            return std::unexpected(parsed_index.error());
+        }
+        constexpr int MAX_TOOL_CALL_INDEX = 1024;
+        if(*parsed_index > MAX_TOOL_CALL_INDEX)
+        {
+            return std::unexpected(
+                error::parse("stream tool call index too large"));
+        }
+
+        if(tool_calls_.size() <= static_cast<std::size_t>(*parsed_index))
+        {
+            tool_calls_.resize(static_cast<std::size_t>(*parsed_index) + 1);
+        }
+        PartialToolCall& partial = tool_calls_[*parsed_index];
+        partial.seen = true;
+
+        const auto type = tool_call.find("type");
+        if(type != tool_call.end() &&
+           (!type->is_string() || *type != "function"))
+        {
+            return std::unexpected(
+                error::parse("expected stream function tool call"));
+        }
+        const auto id = tool_call.find("id");
+        if(id != tool_call.end() && !id->is_null())
+        {
+            if(!id->is_string())
+            {
+                return std::unexpected(error::parse("expected stream tool id"));
+            }
+            partial.id = id->get<std::string>();
+        }
+
+        const auto function = tool_call.find("function");
+        if(function != tool_call.end() && !function->is_null())
+        {
+            if(!function->is_object())
+            {
+                return std::unexpected(
+                    error::parse("expected stream tool function object"));
+            }
+            const auto name = function->find("name");
+            if(name != function->end() && !name->is_null())
+            {
+                if(!name->is_string())
+                {
+                    return std::unexpected(
+                        error::parse("expected stream tool name"));
+                }
+                partial.name = name->get<std::string>();
+            }
+            const auto arguments = function->find("arguments");
+            if(arguments != function->end() && !arguments->is_null())
+            {
+                if(!arguments->is_string())
+                {
+                    return std::unexpected(
+                        error::parse("expected stream tool arguments string"));
+                }
+                partial.arguments_buffer.append(arguments->get<std::string>());
+            }
+        }
+
+        std::vector<StreamEvent> events;
+        if(!partial.started && !partial.id.empty() && !partial.name.empty())
+        {
+            partial.started = true;
+            events.push_back(StreamToolCallStarted{
+                .id = partial.id,
+                .name = partial.name,
+                .index = *parsed_index,
+            });
+        }
+        return events;
+    }
+
+    AssistantMessage message_;
+    std::vector<PartialToolCall> tool_calls_;
+    std::optional<std::string> finish_reason_;
+    bool finished_ = false;
+};
+
+mw::Task<mw::E<mw::HTTPResponse>>
+streamRequest(mw::HTTPSessionAsyncInterface& http, mw::HTTPRequest request,
+              detail::SseFrameParser& frame_parser,
+              StreamDeltaParser& delta_parser,
+              detail::AsyncQueue<StreamEvent>& queue)
+{
+    std::optional<mw::Error> callback_error;
+    auto response = co_await http.postStream(
+        std::move(request),
+        [&](std::span<const std::byte> chunk)
+        {
+            if(callback_error)
+            {
+                return false;
+            }
+
+            const auto bytes = std::string_view(
+                reinterpret_cast<const char*>(chunk.data()), chunk.size());
+            const auto payloads = frame_parser.feed(bytes);
+            for(const std::string& payload : payloads)
+            {
+                auto events = delta_parser.feed(payload);
+                if(!events)
+                {
+                    callback_error = events.error();
+                    return false;
+                }
+                for(StreamEvent& event : *events)
+                {
+                    queue.push(std::move(event));
+                }
+            }
+            return true;
+        });
+
+    if(callback_error)
+    {
+        queue.close();
+        co_return std::unexpected(std::move(*callback_error));
+    }
+    if(!response)
+    {
+        queue.close();
+        co_return std::unexpected(response.error());
+    }
+
+    auto final_events = delta_parser.finish();
+    if(!final_events)
+    {
+        queue.close();
+        co_return std::unexpected(final_events.error());
+    }
+    for(StreamEvent& event : *final_events)
+    {
+        queue.push(std::move(event));
+    }
+    queue.close();
+    co_return std::move(*response);
+}
+
 std::chrono::milliseconds retryDelay(int retry_number)
 {
     const int exponent = std::clamp(retry_number - 1, 0, 4);
@@ -262,7 +695,7 @@ LlmClient::LlmClient(Runtime& runtime, mw::HTTPSessionAsyncInterface& http,
 
 mw::Task<mw::E<LlmResponse>> LlmClient::complete(LlmRequest request)
 {
-    auto body = buildRequestBody(request, config_);
+    auto body = buildRequestBody(request, config_, false);
     if(!body)
     {
         co_return std::unexpected(body.error());
@@ -315,4 +748,66 @@ mw::Task<mw::E<LlmResponse>> LlmClient::complete(LlmRequest request)
     }
 }
 
+mw::Task<mw::E<LlmResponse>>
+LlmClient::completeStream(LlmRequest request,
+                          std::function<void(const StreamEvent&)> on_event)
+{
+    auto body = buildRequestBody(request, config_, true);
+    if(!body)
+    {
+        co_return std::unexpected(body.error());
+    }
+    if(!config_.endpoint.valid())
+    {
+        co_return std::unexpected(error::parse("expected valid LLM endpoint"));
+    }
+
+    auto endpoint = config_.endpoint.resolve("/v1/chat/completions");
+    if(!endpoint)
+    {
+        co_return std::unexpected(endpoint.error());
+    }
+    LlmClientConfig request_config = config_;
+    request_config.endpoint = std::move(*endpoint);
+
+    detail::SseFrameParser frame_parser;
+    StreamDeltaParser delta_parser;
+    detail::AsyncQueue<StreamEvent> queue;
+    auto request_task = detail::StartedTask<mw::HTTPResponse>::start(
+        streamRequest(*http_, buildHttpRequest(*body, request_config),
+                      frame_parser, delta_parser, queue));
+
+    std::optional<mw::Error> callback_error;
+    while(auto event = co_await queue.pop())
+    {
+        if(callback_error)
+        {
+            continue;
+        }
+        try
+        {
+            on_event(*event);
+        }
+        catch(const std::exception& error)
+        {
+            callback_error = mw::runtimeError(error.what());
+        }
+        catch(...)
+        {
+            callback_error = mw::runtimeError("stream event callback threw");
+        }
+    }
+
+    auto response = co_await std::move(request_task);
+    if(!response)
+    {
+        co_return std::unexpected(response.error());
+    }
+    if(callback_error)
+    {
+        co_return std::unexpected(std::move(*callback_error));
+    }
+    co_return delta_parser.takeResponse();
+}
+
 } // namespace agent_smith
diff --git a/tests/test_llm_client.cpp b/tests/test_llm_client.cpp
index 7715592..4a0a008 100644
--- a/tests/test_llm_client.cpp
+++ b/tests/test_llm_client.cpp
@@ -1,10 +1,16 @@
+#include <atomic>
+#include <chrono>
 #include <condition_variable>
 #include <coroutine>
+#include <cstddef>
 #include <exception>
 #include <mutex>
 #include <optional>
+#include <span>
 #include <string>
+#include <string_view>
 #include <utility>
+#include <vector>
 
 #include <agent_smith/llm_client.hpp>
 #include <gmock/gmock.h>
@@ -149,6 +155,38 @@ LlmClientConfig testConfig()
     };
 }
 
+bool sendChunk(const mw::ChunkCallback& callback, std::string_view text)
+{
+    return callback(std::as_bytes(std::span(text.data(), text.size())));
+}
+
+mw::Task<mw::E<mw::HTTPResponse>>
+streamChunks(mw::ChunkCallback callback, Runtime& runtime,
+             std::vector<std::string> chunks,
+             std::atomic<bool>& second_chunk_sent)
+{
+    if(chunks.empty() || !sendChunk(callback, chunks.front()))
+    {
+        co_return std::unexpected(mw::runtimeError("stream callback aborted"));
+    }
+
+    auto slept = co_await runtime.sleepFor(std::chrono::milliseconds(10));
+    if(!slept)
+    {
+        co_return std::unexpected(slept.error());
+    }
+    second_chunk_sent.store(true);
+    for(std::size_t index = 1; index < chunks.size(); ++index)
+    {
+        if(!sendChunk(callback, chunks[index]))
+        {
+            co_return std::unexpected(
+                mw::runtimeError("stream callback aborted"));
+        }
+    }
+    co_return mw::HTTPResponse(200, "");
+}
+
 TEST(LlmClientTest, SendsOpenAIRequestAndParsesAssistantResponse)
 {
     Runtime runtime;
@@ -295,5 +333,104 @@ TEST(LlmClientTest, RetriesRetryableHttpFailures)
     EXPECT_EQ(result->message.content, "recovered");
 }
 
+TEST(LlmClientTest, StreamsEventsWhileHttpRequestIsActive)
+{
+    Runtime runtime;
+    mw::HTTPSessionAsyncMock http;
+    LlmClient client(runtime, http, testConfig());
+    std::atomic<bool> second_chunk_sent = false;
+    mw::HTTPRequest captured;
+
+    const std::vector<std::string> chunks{
+        "data: {\"choices\":[{\"delta\":{\"content\":\"Hello \"},"
+        "\"finish_reason\":null}]}\n\n",
+        R"(data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"weather","arguments":"{\"city\":"}}]},"finish_reason":null}]}
+
+data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\"Portland\"}"}}]},"finish_reason":"tool_calls"}]}
+
+data: {"usage":{"prompt_tokens":3,"completion_tokens":2,"total_tokens":5}}
+
+data: [DONE]
+
+)",
+    };
+
+    EXPECT_CALL(http, postStream(_, _))
+        .WillOnce(Invoke(
+            [&](mw::HTTPRequest request, mw::ChunkCallback callback)
+            {
+                captured = std::move(request);
+                return streamChunks(std::move(callback), runtime, chunks,
+                                    second_chunk_sent);
+            }));
+
+    std::vector<StreamEvent> events;
+    bool first_hook_before_second_chunk = false;
+    auto result = syncWait(client.completeStream(
+        {},
+        [&](const StreamEvent& event)
+        {
+            if(std::holds_alternative<StreamTokenDelta>(event))
+            {
+                first_hook_before_second_chunk = !second_chunk_sent.load();
+            }
+            events.push_back(event);
+        }));
+
+    ASSERT_TRUE(result.has_value());
+    EXPECT_TRUE(first_hook_before_second_chunk);
+    EXPECT_TRUE(nlohmann::json::parse(captured.request_data).at("stream"));
+    ASSERT_EQ(events.size(), 4U);
+    const auto* token = std::get_if<StreamTokenDelta>(&events[0]);
+    ASSERT_NE(token, nullptr);
+    EXPECT_EQ(token->text, "Hello ");
+    const auto* started = std::get_if<StreamToolCallStarted>(&events[1]);
+    ASSERT_NE(started, nullptr);
+    EXPECT_EQ(started->id, "call_1");
+    EXPECT_EQ(started->name, "weather");
+    const auto* completed = std::get_if<StreamToolCallCompleted>(&events[2]);
+    ASSERT_NE(completed, nullptr);
+    EXPECT_EQ(completed->arguments, nlohmann::json({{"city", "Portland"}}));
+    const auto* finished = std::get_if<StreamFinished>(&events[3]);
+    ASSERT_NE(finished, nullptr);
+    EXPECT_EQ(finished->finish_reason, "tool_calls");
+    ASSERT_TRUE(finished->usage.has_value());
+    EXPECT_EQ(finished->usage->total, 5U);
+
+    EXPECT_EQ(result->message.content, "Hello ");
+    ASSERT_EQ(result->message.tool_calls.size(), 1U);
+    EXPECT_EQ(result->message.tool_calls[0].id, "call_1");
+    EXPECT_EQ(result->message.tool_calls[0].arguments,
+              nlohmann::json({{"city", "Portland"}}));
+}
+
+TEST(LlmClientTest, RejectsMalformedStreamToolArguments)
+{
+    Runtime runtime;
+    mw::HTTPSessionAsyncMock http;
+    LlmClient client(runtime, http, testConfig());
+    std::atomic<bool> second_chunk_sent = false;
+
+    EXPECT_CALL(http, postStream(_, _))
+        .WillOnce(Invoke(
+            [&](mw::HTTPRequest, mw::ChunkCallback callback)
+            {
+                return streamChunks(
+                    std::move(callback), runtime,
+                    {R"(data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"weather","arguments":"not JSON"}}]},"finish_reason":"tool_calls"}]}
+
+data: [DONE]
+
+)"},
+                    second_chunk_sent);
+            }));
+
+    auto result =
+        syncWait(client.completeStream({}, [](const StreamEvent&) {}));
+
+    ASSERT_FALSE(result.has_value());
+    EXPECT_NE(result.error().as<ParseError>(), nullptr);
+}
+
 } // namespace
 } // namespace agent_smith