diff --git a/CMakeLists.txt b/CMakeLists.txt index 2bd8521..1849e29 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -87,6 +87,7 @@ target_sources(ai-sdk-cpp-core # HTTP and base provider infrastructure src/http/http_request_handler.cpp src/providers/base_provider_client.cpp + src/providers/http_sse_stream.cpp # Tool calling implementations src/tools/tool_executor.cpp diff --git a/include/ai/types/stream_event.h b/include/ai/types/stream_event.h index 6becd03..8646af3 100644 --- a/include/ai/types/stream_event.h +++ b/include/ai/types/stream_event.h @@ -1,6 +1,7 @@ #pragma once #include "enums.h" +#include "tool.h" #include "usage.h" #include @@ -20,12 +21,20 @@ struct StreamEvent { FinishReason reason) : type(event_type), usage(usage_stats), finish_reason(reason) {} + StreamEvent(StreamEventType event_type, FinishReason reason) + : type(event_type), finish_reason(reason) {} + explicit StreamEvent(StreamEventType event_type) : type(event_type) {} + explicit StreamEvent(ToolCall call) + : type(kStreamEventTypeToolCall), tool_call(std::move(call)) {} + bool is_text_delta() const { return type == kStreamEventTypeTextDelta; } bool is_error() const { return type == kStreamEventTypeError; } + bool is_tool_call() const { return type == kStreamEventTypeToolCall; } + bool is_finish() const { return type == kStreamEventTypeFinish; } StreamEventType type; @@ -33,7 +42,8 @@ struct StreamEvent { std::optional error; std::optional usage; std::optional finish_reason; + std::optional tool_call; std::optional metadata; }; -} // namespace ai \ No newline at end of file +} // namespace ai diff --git a/src/providers/anthropic/anthropic_stream.cpp b/src/providers/anthropic/anthropic_stream.cpp index 049d7f4..5a3b6a6 100644 --- a/src/providers/anthropic/anthropic_stream.cpp +++ b/src/providers/anthropic/anthropic_stream.cpp @@ -2,168 +2,72 @@ #include "ai/logger.h" -#include -#include -#include - namespace { -constexpr auto kEventTimeout = static_cast(30); -constexpr auto kSleepInterval = std::chrono::milliseconds(1); +// Null-tolerant JSON field accessors: gateways may deliver fields as JSON +// null, which nlohmann's value() would throw on (discarding the whole event). +int int_or_zero(const nlohmann::json& obj, const char* key) { + const auto it = obj.find(key); + return it != obj.end() && it->is_number() ? it->get() : 0; +} + +std::string string_or_empty(const nlohmann::json& obj, const char* key) { + const auto it = obj.find(key); + return it != obj.end() && it->is_string() ? it->get() + : std::string{}; +} + +std::size_t index_or_zero(const nlohmann::json& obj) { + const auto it = obj.find("index"); + return it != obj.end() && it->is_number_unsigned() ? it->get() + : std::size_t{0}; +} } // namespace namespace ai { namespace anthropic { AnthropicStreamImpl::~AnthropicStreamImpl() { + // Join before this class's state is destroyed; the stream thread calls + // process_sse_line. stop_stream(); } -void AnthropicStreamImpl::start_stream(const std::string& url, - const httplib::Headers& headers, - const nlohmann::json& request_body) { - ai::logger::log_debug("Starting Anthropic stream to URL: {}", url); - - // Start streaming in a separate thread - stream_thread_ = std::thread([this, url, headers, request_body]() { - try { - run_stream(url, headers, request_body); - } catch (const std::exception& e) { - ai::logger::log_error("Stream thread exception: {}", e.what()); - StreamEvent error_event(kStreamEventTypeError, - std::string("Stream error: ") + e.what()); - push_event(error_event); - mark_complete(); - } - }); +void AnthropicStreamImpl::reset_stream_state() { + event_data_.clear(); } -StreamEvent AnthropicStreamImpl::get_next_event() { - StreamEvent event(""); - auto start_time = std::chrono::steady_clock::now(); - - while (!event_queue_.try_dequeue(event)) { - if (stream_complete_ && event_queue_.size_approx() == 0) { - // Stream is complete and queue is empty - ai::logger::log_debug( - "Stream complete and queue empty, returning empty event"); - return StreamEvent(""); +void AnthropicStreamImpl::process_sse_line(std::string_view line) { + if (line.empty()) { + // Blank line terminates the SSE event + if (!event_data_.empty()) { + process_sse_event(event_data_); + event_data_.clear(); } - - // Check for timeout - if (std::chrono::steady_clock::now() - start_time > kEventTimeout) { - ai::logger::log_error( - "Timeout waiting for next stream event after {} seconds", - kEventTimeout.count()); - return StreamEvent(kStreamEventTypeError, - "Timeout waiting for next event"); + } else if (line.starts_with("data:")) { + auto data = line.substr(5); + if (!data.empty() && data.front() == ' ') { + data.remove_prefix(1); } - - std::this_thread::sleep_for(kSleepInterval); - } - - ai::logger::log_debug("Dequeued event type: {}", - static_cast(event.type)); - return event; -} - -bool AnthropicStreamImpl::has_more_events() const { - return event_queue_.size_approx() > 0 || !stream_complete_; -} - -void AnthropicStreamImpl::stop_stream() { - ai::logger::log_debug("Stopping Anthropic stream"); - stop_requested_ = true; - if (stream_thread_.joinable()) { - stream_thread_.join(); + if (!event_data_.empty()) { + event_data_ += '\n'; + } + event_data_.append(data); } } -void AnthropicStreamImpl::run_stream(const std::string& url, - const httplib::Headers& headers, - const nlohmann::json& request_body) { - ai::logger::log_debug("Performing stream request"); - - // Parse URL to extract host and path - std::string host, path; - bool use_ssl = true; - - if (url.starts_with("https://")) { - host = url.substr(8); - use_ssl = true; - } else if (url.starts_with("http://")) { - host = url.substr(7); - use_ssl = false; - } - - if (auto pos = host.find('/'); pos != std::string::npos) { - path = host.substr(pos); - host = host.substr(0, pos); - } else { - path = "/v1/messages"; - } - - ai::logger::log_debug("Stream host: {}, path: {}, SSL: {}", host, path, - use_ssl); - - try { - if (use_ssl) { - httplib::SSLClient client(host); - client.enable_server_certificate_verification(true); - client.set_connection_timeout(30, 0); - client.set_read_timeout(120, 0); - - auto result = - client.Post(path, headers, request_body.dump(), "application/json"); - - if (result && result->status == 200) { - parse_sse_response(result->body); - } else { - handle_stream_error(result ? result->status : 0, - result ? result->body : "Connection failed"); - } - } else { - httplib::Client client(host); - client.set_connection_timeout(30, 0); - client.set_read_timeout(120, 0); - - auto result = - client.Post(path, headers, request_body.dump(), "application/json"); - - if (result && result->status == 200) { - parse_sse_response(result->body); - } else { - handle_stream_error(result ? result->status : 0, - result ? result->body : "Connection failed"); - } - } - } catch (const std::exception& e) { - ai::logger::log_error("Stream request exception: {}", e.what()); - handle_stream_error(0, std::string("Request failed: ") + e.what()); +void AnthropicStreamImpl::finalize_stream() { + // Dispatch an event whose terminating blank line never arrived. + if (!event_data_.empty()) { + process_sse_event(event_data_); + event_data_.clear(); } - - mark_complete(); } -void AnthropicStreamImpl::parse_sse_response(const std::string& response) { - ai::logger::log_debug("Processing SSE response, size: {}", response.size()); - - std::istringstream stream(response); - std::string line; - std::string event_data; - - while (std::getline(stream, line) && !stop_requested_) { - if (line.empty()) { - // Empty line signals end of event - if (!event_data.empty()) { - process_sse_event(event_data); - event_data.clear(); - } - } else if (line.starts_with("data: ")) { - event_data = line.substr(6); - } +Usage& AnthropicStreamImpl::ensure_usage() { + if (!usage_) { + usage_ = Usage{}; } - - ai::logger::log_debug("SSE processing complete"); + return *usage_; } void AnthropicStreamImpl::process_sse_event(const std::string& data) { @@ -174,72 +78,91 @@ void AnthropicStreamImpl::process_sse_event(const std::string& data) { try { auto json_event = nlohmann::json::parse(data); - std::string event_type = json_event.value("type", ""); + std::string event_type = string_or_empty(json_event, "type"); ai::logger::log_debug("Processing SSE event type: {}", event_type); if (event_type == "message_start") { - // Start of message - could extract metadata here + if (json_event.contains("message") && json_event["message"].is_object()) { + const auto& message = json_event["message"]; + if (message.contains("usage") && message["usage"].is_object()) { + auto& usage = ensure_usage(); + usage.prompt_tokens = int_or_zero(message["usage"], "input_tokens"); + usage.total_tokens = usage.prompt_tokens + usage.completion_tokens; + } + } return; } else if (event_type == "content_block_start") { - // Start of content block + if (json_event.contains("content_block") && + json_event["content_block"].is_object() && + string_or_empty(json_event["content_block"], "type") == "tool_use") { + const auto index = index_or_zero(json_event); + const auto& block = json_event["content_block"]; + auto& pending = pending_tool_calls_[index]; + pending.id = string_or_empty(block, "id"); + pending.name = string_or_empty(block, "name"); + if (block.contains("input") && block["input"].is_object() && + !block["input"].empty()) { + // Keep separately from the delta accumulator: appending deltas to + // an already complete JSON object would produce invalid JSON. + pending.initial_input = block["input"].dump(); + } + } return; } else if (event_type == "content_block_delta") { // Text delta - this is what we want to stream - if (json_event.contains("delta") && - json_event["delta"].contains("text")) { + if (json_event.contains("delta") && json_event["delta"].is_object() && + json_event["delta"].contains("text") && + json_event["delta"]["text"].is_string()) { std::string text = json_event["delta"]["text"]; - StreamEvent event(text); - push_event(event); + push_event(StreamEvent(text)); ai::logger::log_debug("Enqueued text delta: '{}'", text); + } else if (json_event.contains("delta") && + json_event["delta"].is_object() && + string_or_empty(json_event["delta"], "type") == + "input_json_delta") { + const auto index = index_or_zero(json_event); + pending_tool_calls_[index].arguments += + string_or_empty(json_event["delta"], "partial_json"); } } else if (event_type == "content_block_stop") { - // End of content block + flush_pending_tool_call(index_or_zero(json_event)); return; } else if (event_type == "message_delta") { - // Message-level delta (could contain stop reason) + if (json_event.contains("delta") && json_event["delta"].is_object()) { + const auto stop_reason = + string_or_empty(json_event["delta"], "stop_reason"); + if (stop_reason == "max_tokens") { + finish_reason_ = kFinishReasonLength; + } else if (stop_reason == "tool_use") { + finish_reason_ = kFinishReasonToolCalls; + } else if (!stop_reason.empty()) { + finish_reason_ = kFinishReasonStop; + } + } + if (json_event.contains("usage") && json_event["usage"].is_object()) { + auto& usage = ensure_usage(); + usage.completion_tokens = + int_or_zero(json_event["usage"], "output_tokens"); + usage.total_tokens = usage.prompt_tokens + usage.completion_tokens; + } return; } else if (event_type == "message_stop") { - // End of message - StreamEvent event(kStreamEventTypeFinish, Usage{}, kFinishReasonStop); - push_event(event); - - ai::logger::log_debug("Enqueued finish event"); + // The stream is semantically finished; complete immediately so + // consumers are not left waiting for the connection to close. + mark_complete(); + } else if (event_type == "error") { + const auto message = json_event.value("error", nlohmann::json::object()) + .value("message", "Anthropic stream error"); + note_stream_error(); + push_event(create_error_event(message)); } } catch (const std::exception& e) { ai::logger::log_error("Failed to parse SSE event: {}", e.what()); } } -void AnthropicStreamImpl::push_event(const StreamEvent& event) { - event_queue_.enqueue(event); -} - -void AnthropicStreamImpl::mark_complete() { - stream_complete_ = true; - - // Push final finish event if not already done - StreamEvent finish_event(kStreamEventTypeFinish); - push_event(finish_event); -} - -StreamEvent AnthropicStreamImpl::create_error_event( - const std::string& message) { - return StreamEvent(kStreamEventTypeError, message); -} - -void AnthropicStreamImpl::handle_stream_error(int status_code, - const std::string& error_body) { - ai::logger::log_error("Stream error - status: {}, body: {}", status_code, - error_body); - - StreamEvent error_event( - kStreamEventTypeError, - "Stream error (" + std::to_string(status_code) + "): " + error_body); - push_event(error_event); -} - } // namespace anthropic -} // namespace ai \ No newline at end of file +} // namespace ai diff --git a/src/providers/anthropic/anthropic_stream.h b/src/providers/anthropic/anthropic_stream.h index c4c2ae4..74d6eb1 100644 --- a/src/providers/anthropic/anthropic_stream.h +++ b/src/providers/anthropic/anthropic_stream.h @@ -1,55 +1,33 @@ #pragma once -#include "ai/types/stream_result.h" +#include "providers/http_sse_stream.h" -#include -#include -#include -#include -#include - -#include +#include +#include namespace ai { namespace anthropic { -class AnthropicStreamImpl : public internal::StreamResultImpl { +class AnthropicStreamImpl : public providers::streaming::HttpSseStream { public: - AnthropicStreamImpl() = default; - ~AnthropicStreamImpl(); - - // Non-copyable, non-movable for thread safety - AnthropicStreamImpl(const AnthropicStreamImpl&) = delete; - AnthropicStreamImpl& operator=(const AnthropicStreamImpl&) = delete; - AnthropicStreamImpl(AnthropicStreamImpl&&) = delete; - AnthropicStreamImpl& operator=(AnthropicStreamImpl&&) = delete; - - void start_stream(const std::string& url, - const httplib::Headers& headers, - const nlohmann::json& request_body); - - StreamEvent get_next_event() override; - bool has_more_events() const override; - void stop_stream() override; + AnthropicStreamImpl() + : HttpSseStream({.default_path = "/v1/messages", + .connection_timeout_seconds = 30, + .read_timeout_seconds = 120}) {} + ~AnthropicStreamImpl() override; private: - void run_stream(const std::string& url, - const httplib::Headers& headers, - const nlohmann::json& request_body); - void parse_sse_response(const std::string& response); - void process_sse_event(const std::string& data); - void push_event(const StreamEvent& event); - void mark_complete(); + void reset_stream_state() override; + void process_sse_line(std::string_view line) override; + void finalize_stream() override; - // Helper functions - StreamEvent create_error_event(const std::string& message); - void handle_stream_error(int status_code, const std::string& error_body); + void process_sse_event(const std::string& data); + Usage& ensure_usage(); - moodycamel::ConcurrentQueue event_queue_; - std::thread stream_thread_; - std::atomic stop_requested_{false}; - std::atomic stream_complete_{false}; + // Anthropic SSE events span multiple "data:" lines; accumulate them here + // until the blank line that terminates the event. + std::string event_data_; }; } // namespace anthropic -} // namespace ai \ No newline at end of file +} // namespace ai diff --git a/src/providers/http_sse_stream.cpp b/src/providers/http_sse_stream.cpp new file mode 100644 index 0000000..292ef80 --- /dev/null +++ b/src/providers/http_sse_stream.cpp @@ -0,0 +1,268 @@ +#include "http_sse_stream.h" + +#include "ai/logger.h" + +#include +#include + +namespace { +constexpr auto kEventTimeout = static_cast(30); +constexpr auto kSleepInterval = std::chrono::milliseconds(1); +} // namespace + +namespace ai { +namespace providers { +namespace streaming { + +HttpSseStream::HttpSseStream(Options options) : options_(std::move(options)) {} + +HttpSseStream::~HttpSseStream() { + // Concrete classes join in their own destructor (their state is still alive + // for the stream thread there); this is a backstop for the base state. + stop_stream(); +} + +void HttpSseStream::start_stream(const std::string& url, + const httplib::Headers& headers, + const nlohmann::json& request_body) { + ai::logger::log_debug("Starting stream to URL: {}", url); + + std::lock_guard lock(thread_mutex_); + + if (stream_thread_.joinable()) { + ai::logger::log_debug( + "Stream thread already running, not starting new one"); + return; // Already running + } + + // Reset state for new stream + should_stop_ = false; + is_complete_ = false; + finish_event_pushed_ = false; + stream_errored_ = false; + line_buffer_.clear(); + pending_tool_calls_.clear(); + finish_reason_ = kFinishReasonStop; + usage_.reset(); + reset_stream_state(); + + ai::logger::log_info("Launching stream thread"); + + stream_thread_ = std::thread([this, url, headers, request_body]() { + run_stream(url, headers, request_body); + }); +} + +StreamEvent HttpSseStream::get_next_event() { + StreamEvent event(""); + auto start_time = std::chrono::steady_clock::now(); + + while (!event_queue_.try_dequeue(event)) { + if (is_complete_ && event_queue_.size_approx() == 0) { + // Stream is complete and queue is empty + ai::logger::log_debug( + "Stream complete and queue empty, returning empty event"); + return StreamEvent(""); + } + + // Check for timeout + if (std::chrono::steady_clock::now() - start_time > kEventTimeout) { + ai::logger::log_error( + "Timeout waiting for next stream event after {} seconds", + kEventTimeout.count()); + return StreamEvent(kStreamEventTypeError, + "Timeout waiting for next event"); + } + + std::this_thread::sleep_for(kSleepInterval); + } + + ai::logger::log_debug("Dequeued event type: {}", + static_cast(event.type)); + return event; +} + +bool HttpSseStream::has_more_events() const { + // No locks needed - these are atomic operations + return event_queue_.size_approx() > 0 || !is_complete_; +} + +void HttpSseStream::stop_stream() { + ai::logger::log_debug("Stopping stream"); + + should_stop_ = true; // Atomic write + + std::lock_guard lock(thread_mutex_); + if (stream_thread_.joinable()) { + ai::logger::log_debug("Waiting for stream thread to finish"); + stream_thread_.join(); + ai::logger::log_info("Stream stopped successfully"); + } +} + +void HttpSseStream::run_stream(const std::string& url, + const httplib::Headers& headers, + const nlohmann::json& request_body) { + // Extract origin and path from URL. Using httplib::Client with the scheme + // preserved supports both HTTPS provider endpoints and HTTP test/gateway + // endpoints. + const auto [origin, path] = split_origin_and_path(url, options_.default_path); + + ai::logger::log_debug( + "Stream thread started - connecting to {} with path: {}", origin, path); + + try { + httplib::Client client(origin); + client.enable_server_certificate_verification(true); + client.set_connection_timeout(options_.connection_timeout_seconds, 0); + client.set_read_timeout(options_.read_timeout_seconds, 0); + + int response_status = 0; + std::string error_body; + + httplib::Request request; + request.method = "POST"; + request.path = path; + request.headers = headers; + request.body = request_body.dump(); + request.set_header("Content-Type", "application/json"); + + ai::logger::log_debug( + "Stream request prepared - path: {}, body size: {} bytes", path, + request.body.length()); + + // Capture the status before the body arrives so error bodies can be + // preserved (httplib routes the body of every status through the content + // receiver, leaving response.body empty). + request.response_handler = + [&response_status](const httplib::Response& response) { + response_status = response.status; + return true; + }; + + request.content_receiver = [this, &response_status, &error_body]( + const char* data, size_t length, + uint64_t /*offset*/, + uint64_t /*total_length*/) { + if (should_stop_) { + return false; + } + if (response_status != 200) { + // Error responses are not SSE; keep the body for diagnostics. + error_body.append(data, length); + return true; + } + consume_chunk(data, length); + return !should_stop_; + }; + + httplib::Response response; + httplib::Error error; + + ai::logger::log_info("Sending stream request"); + + if (!client.send(request, response, error)) { + // A deliberate stop makes the receiver return false (Error::Canceled); + // do not report that as a network failure. + if (!should_stop_) { + std::string message = "Network error: " + httplib::to_string(error); + ai::logger::log_error("Failed to send stream request: {}", message); + note_stream_error(); + push_event(create_error_event(message)); + } + } else if (response.status != 200) { + ai::logger::log_error("Stream API returned status {} - body: {}", + response.status, error_body); + note_stream_error(); + push_event(create_error_event("HTTP " + std::to_string(response.status) + + " error: " + error_body)); + } else { + if (!line_buffer_.empty()) { + // Process a trailing line that arrived without a final newline. + consume_chunk("\n", 1); + } + if (!stream_errored_) { + finalize_stream(); + // Tool calls whose terminating event never arrived (some gateways + // close the connection early) must not be silently dropped. + flush_all_pending_tool_calls(); + } + ai::logger::log_info("Stream completed successfully"); + } + } catch (const std::exception& e) { + ai::logger::log_error("Exception in stream thread: {}", e.what()); + note_stream_error(); + push_event(create_error_event(e.what())); + } + + mark_complete(); + ai::logger::log_debug("Stream thread exiting"); +} + +void HttpSseStream::consume_chunk(const char* data, std::size_t length) { + line_buffer_.append(data, length); + + std::size_t start = 0; + std::size_t newline = 0; + while ((newline = line_buffer_.find('\n', start)) != std::string::npos) { + std::string_view line(line_buffer_.data() + start, newline - start); + start = newline + 1; + if (!line.empty() && line.back() == '\r') { + line.remove_suffix(1); + } + process_sse_line(line); + } + line_buffer_.erase(0, start); +} + +void HttpSseStream::push_event(StreamEvent event) { + event_queue_.enqueue(std::move(event)); +} + +StreamEvent HttpSseStream::create_error_event(const std::string& message) { + ai::logger::log_debug("Creating error event: {}", message); + return StreamEvent(kStreamEventTypeError, message); +} + +void HttpSseStream::push_finish_event_if_needed() { + bool expected = false; + if (!finish_event_pushed_.compare_exchange_strong(expected, true)) { + ai::logger::log_debug("Finish event already pushed, skipping"); + return; + } + + ai::logger::log_debug("Pushing finish event to queue"); + if (stream_errored_) { + // Do not stamp a failed stream with a clean finish reason or usage. + push_event(StreamEvent(kStreamEventTypeFinish, kFinishReasonError)); + } else if (usage_) { + push_event(StreamEvent(kStreamEventTypeFinish, *usage_, finish_reason_)); + } else { + push_event(StreamEvent(kStreamEventTypeFinish, finish_reason_)); + } +} + +void HttpSseStream::mark_complete() { + push_finish_event_if_needed(); + is_complete_ = true; // Atomic write +} + +void HttpSseStream::flush_all_pending_tool_calls() { + for (const auto& entry : pending_tool_calls_) { + push_event(make_tool_call_event(entry.second)); + } + pending_tool_calls_.clear(); +} + +void HttpSseStream::flush_pending_tool_call(std::size_t index) { + const auto pending_it = pending_tool_calls_.find(index); + if (pending_it == pending_tool_calls_.end()) { + return; + } + push_event(make_tool_call_event(pending_it->second)); + pending_tool_calls_.erase(pending_it); +} + +} // namespace streaming +} // namespace providers +} // namespace ai diff --git a/src/providers/http_sse_stream.h b/src/providers/http_sse_stream.h new file mode 100644 index 0000000..6b9bb7d --- /dev/null +++ b/src/providers/http_sse_stream.h @@ -0,0 +1,121 @@ +#pragma once + +#include "ai/types/stream_result.h" +#include "providers/stream_utils.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace ai { +namespace providers { +namespace streaming { + +/// Threaded HTTP SSE stream shared by the provider stream implementations. +/// Owns the event queue, the stream thread, and the HTTP request plus SSE +/// line-splitting plumbing; derived classes only translate provider-specific +/// SSE lines into stream events. +/// +/// Concrete classes must call stop_stream() in their destructor so the stream +/// thread is joined before their members are destroyed. +class HttpSseStream : public internal::StreamResultImpl { + public: + ~HttpSseStream() override; + + // Non-copyable, non-movable for thread safety + HttpSseStream(const HttpSseStream&) = delete; + HttpSseStream& operator=(const HttpSseStream&) = delete; + HttpSseStream(HttpSseStream&&) = delete; + HttpSseStream& operator=(HttpSseStream&&) = delete; + + void start_stream(const std::string& url, + const httplib::Headers& headers, + const nlohmann::json& request_body); + + StreamEvent get_next_event() override; + bool has_more_events() const override; + void stop_stream() override; + +#ifdef AI_SDK_TESTING + void process_sse_chunk_for_testing(const std::string& chunk) { + consume_chunk(chunk.data(), chunk.size()); + } + void process_sse_line_for_testing(const std::string& line) { + process_sse_line(line); + } + std::size_t queued_event_count_for_testing() const { + return event_queue_.size_approx(); + } +#endif + + protected: + struct Options { + std::string default_path; // Used when the URL carries no path. + time_t connection_timeout_seconds{30}; + time_t read_timeout_seconds{120}; + }; + + explicit HttpSseStream(Options options); + + /// Reset provider-specific parse state before a new stream starts. + virtual void reset_stream_state() {} + + /// Handle one decoded SSE line (trailing CR already stripped). + virtual void process_sse_line(std::string_view line) = 0; + + /// Called after a successful response body has been fully consumed, before + /// pending tool calls are flushed and the finish event is pushed; flush any + /// provider-side leftovers here. + virtual void finalize_stream() {} + + void push_event(StreamEvent event); + static StreamEvent create_error_event(const std::string& message); + /// Push the terminal finish event (at most once per stream). + void push_finish_event_if_needed(); + /// Push the terminal finish event if still needed and mark the stream + /// complete so consumers stop waiting for events. + void mark_complete(); + /// Record that the stream failed; the finish event will carry + /// kFinishReasonError and no usage. + void note_stream_error() { stream_errored_ = true; } + + /// Emit all accumulated tool calls (or parse-error events) and clear them. + void flush_all_pending_tool_calls(); + /// Emit one accumulated tool call by content-block index, if present. + void flush_pending_tool_call(std::size_t index); + + // Terminal-finish payload and tool-call accumulator, maintained by the + // derived parser and consumed when the finish event is pushed. + FinishReason finish_reason_{kFinishReasonStop}; + std::optional usage_; + std::map pending_tool_calls_; + + private: + void run_stream(const std::string& url, + const httplib::Headers& headers, + const nlohmann::json& request_body); + void consume_chunk(const char* data, std::size_t length); + + Options options_; + moodycamel::ConcurrentQueue event_queue_; + std::thread stream_thread_; + std::mutex thread_mutex_; + std::string line_buffer_; + std::atomic is_complete_{false}; + std::atomic should_stop_{false}; + std::atomic finish_event_pushed_{false}; + std::atomic stream_errored_{false}; +}; + +} // namespace streaming +} // namespace providers +} // namespace ai diff --git a/src/providers/openai/openai_client.cpp b/src/providers/openai/openai_client.cpp index a938642..ef52005 100644 --- a/src/providers/openai/openai_client.cpp +++ b/src/providers/openai/openai_client.cpp @@ -58,6 +58,9 @@ StreamResult OpenAIClient::stream_text(const StreamOptions& options) { // Build request with stream: true auto request_json = request_builder_->build_request_json(options); request_json["stream"] = true; + // Request the terminal usage chunk; without this OpenAI omits usage from + // streaming responses and the finish event would never carry token counts. + request_json["stream_options"] = {{"include_usage", true}}; ai::logger::log_debug("Stream request JSON built with stream=true"); // Create headers diff --git a/src/providers/openai/openai_stream.cpp b/src/providers/openai/openai_stream.cpp index bc82b13..6347b3a 100644 --- a/src/providers/openai/openai_stream.cpp +++ b/src/providers/openai/openai_stream.cpp @@ -1,282 +1,99 @@ #include "openai_stream.h" #include "ai/logger.h" -#include "http/http_request_handler.h" - -#include -#include - -namespace { -constexpr auto kEventTimeout = static_cast(30); -constexpr auto kSleepInterval = std::chrono::milliseconds(1); -constexpr auto kConnectionTimeout = 30; // seconds -constexpr auto kReadTimeout = 300; // 5 minutes for long generations -} // namespace namespace ai { namespace openai { OpenAIStreamImpl::~OpenAIStreamImpl() { + // Join before this class's state is destroyed; the stream thread calls + // process_sse_line. stop_stream(); } -void OpenAIStreamImpl::start_stream(const std::string& url, - const httplib::Headers& headers, - const nlohmann::json& request_body) { - ai::logger::log_debug("Starting OpenAI stream - URL: {}", url); - - std::lock_guard lock(thread_mutex_); - - if (stream_thread_.joinable()) { - ai::logger::log_debug( - "Stream thread already running, not starting new one"); - return; // Already running - } - - // Reset state for new stream - should_stop_ = false; - is_complete_ = false; - finish_event_pushed_ = false; - - ai::logger::log_info("Launching stream thread for OpenAI API"); - - stream_thread_ = std::thread([this, url, headers, request_body]() { - run_stream(url, headers, request_body); - }); -} - -StreamEvent OpenAIStreamImpl::get_next_event() { - StreamEvent event(""); - auto start_time = std::chrono::steady_clock::now(); - - while (!event_queue_.try_dequeue(event)) { - if (is_complete_ && event_queue_.size_approx() == 0) { - // Stream is complete and queue is empty - ai::logger::log_debug( - "Stream complete and queue empty, returning empty event"); - return StreamEvent(""); - } - - // Check for timeout - if (std::chrono::steady_clock::now() - start_time > kEventTimeout) { - ai::logger::log_error( - "Timeout waiting for next stream event after {} seconds", - kEventTimeout.count()); - return StreamEvent(kStreamEventTypeError, - "Timeout waiting for next event"); +void OpenAIStreamImpl::process_sse_line(std::string_view line) { + if (!line.starts_with("data: ")) { + if (!line.empty()) { + ai::logger::log_debug("Ignoring non-data SSE line: {}", line); } - - std::this_thread::sleep_for(kSleepInterval); + return; } - ai::logger::log_debug("Dequeued event type: {}", - static_cast(event.type)); - return event; -} - -bool OpenAIStreamImpl::has_more_events() const { - // No locks needed - these are atomic operations - return event_queue_.size_approx() > 0 || !is_complete_; -} - -void OpenAIStreamImpl::stop_stream() { - ai::logger::log_debug("Stopping OpenAI stream"); - - should_stop_ = true; // Atomic write - - std::lock_guard lock(thread_mutex_); - if (stream_thread_.joinable()) { - ai::logger::log_debug("Waiting for stream thread to finish"); - stream_thread_.join(); - ai::logger::log_info("OpenAI stream stopped successfully"); - } -} + auto data = line.substr(6); -void OpenAIStreamImpl::run_stream(const std::string& url, - const httplib::Headers& headers, - const nlohmann::json& request_body) { - // Extract host and path from URL - std::string_view url_view(url); + ai::logger::log_debug("Processing SSE line - data length: {}", data.length()); - // Skip protocol - if (auto pos = url_view.find("://"); pos != std::string_view::npos) { - url_view.remove_prefix(pos + 3); + if (data == "[DONE]") { + ai::logger::log_debug("Received [DONE] signal, stream ending"); + flush_all_pending_tool_calls(); + mark_complete(); + return; } - // Split host and path - auto slash_pos = url_view.find('/'); - std::string host(url_view.substr(0, slash_pos)); - std::string path = (slash_pos != std::string_view::npos) - ? std::string(url_view.substr(slash_pos)) - : "/v1/chat/completions"; - - ai::logger::log_debug( - "Stream thread started - connecting to {} with path: {}", host, path); - try { - httplib::SSLClient client(host); - client.enable_server_certificate_verification(true); - client.set_connection_timeout(kConnectionTimeout); - client.set_read_timeout(kReadTimeout); - - ai::logger::log_debug( - "SSL client created with connection_timeout: {}s, read_timeout: {}s", - kConnectionTimeout, kReadTimeout); - - std::string accumulated_data; - - // Create request - httplib::Request req; - req.method = "POST"; - req.path = path; - req.headers = headers; - req.body = request_body.dump(); - req.set_header("Content-Type", "application/json"); - - ai::logger::log_debug( - "Stream request prepared - path: {}, body size: {} bytes", path, - req.body.length()); - - // Set content receiver for streaming response - req.content_receiver = [this, &accumulated_data]( - const char* data, size_t data_length, - uint64_t /*offset*/, uint64_t /*total_length*/) { - // Accumulate data and process complete lines - accumulated_data.append(data, data_length); - - ai::logger::log_debug("Received {} bytes of stream data", data_length); - - // Process complete lines - size_t pos = 0; - while ((pos = accumulated_data.find('\n')) != std::string::npos) { - std::string line = accumulated_data.substr(0, pos); - accumulated_data.erase(0, pos + 1); - - if (!line.empty() && line.back() == '\r') { - line.pop_back(); - } + auto json = nlohmann::json::parse(data); + auto& choices = json["choices"]; + + if (!choices.empty() && choices[0].contains("delta")) { + auto& delta = choices[0]["delta"]; + if (delta.contains("content") && !delta["content"].is_null()) { + std::string content = delta["content"].get(); + ai::logger::log_debug("Received content chunk - length: {}", + content.length()); + push_event(StreamEvent(content)); + } - // Check if we should stop - atomic read, no lock needed - if (should_stop_) { - ai::logger::log_debug( - "Stream stop requested, ending content receiver"); - return false; + if (delta.contains("tool_calls") && delta["tool_calls"].is_array()) { + for (const auto& fragment : delta["tool_calls"]) { + const auto index = fragment.value("index", std::size_t{0}); + auto& pending = pending_tool_calls_[index]; + + if (fragment.contains("id") && fragment["id"].is_string()) { + pending.id = fragment["id"].get(); + } + if (fragment.contains("function") && + fragment["function"].is_object()) { + const auto& function = fragment["function"]; + if (function.contains("name") && function["name"].is_string()) { + pending.name = function["name"].get(); + } + if (function.contains("arguments") && + function["arguments"].is_string()) { + pending.arguments += function["arguments"].get(); + } + } } - - parse_sse_line(line); } - - return true; // Continue receiving - }; - - httplib::Response res; - httplib::Error error; - - ai::logger::log_info("Sending stream request to OpenAI API"); - - if (!client.send(req, res, error)) { - std::string error_msg = "Network error: " + httplib::to_string(error); - ai::logger::log_error("Failed to send stream request: {}", error_msg); - push_event(create_error_event(error_msg)); - } else if (res.status != 200) { - ai::logger::log_error("OpenAI stream API returned status {} - body: {}", - res.status, res.body); - push_event(create_error_event("HTTP " + std::to_string(res.status) + - " error: " + res.body)); - } else { - ai::logger::log_info("Stream completed successfully"); - } - } catch (const std::exception& e) { - ai::logger::log_error("Exception in stream thread: {}", e.what()); - push_event(create_error_event(e.what())); - } - - mark_complete(); - ai::logger::log_debug("Stream thread exiting"); -} - -void OpenAIStreamImpl::parse_sse_line(const std::string& line) { - if (line.starts_with("data: ")) { - auto data = line.substr(6); - - ai::logger::log_debug("Processing SSE line - data length: {}", - data.length()); - - if (data == "[DONE]") { - ai::logger::log_debug("Received [DONE] signal, stream ending"); - push_finish_event_if_needed(); - mark_complete(); - return; } - try { - auto json = nlohmann::json::parse(data); - auto& choices = json["choices"]; + // Check for finish_reason + if (!choices.empty() && choices[0].contains("finish_reason") && + !choices[0]["finish_reason"].is_null()) { + auto finish_reason_str = choices[0]["finish_reason"].get(); + finish_reason_ = parse_finish_reason(finish_reason_str); - if (!choices.empty() && choices[0].contains("delta")) { - auto& delta = choices[0]["delta"]; - if (delta.contains("content") && !delta["content"].is_null()) { - std::string content = delta["content"].get(); - ai::logger::log_debug("Received content chunk - length: {}", - content.length()); - push_event(StreamEvent(content)); - } + if (finish_reason_ == kFinishReasonToolCalls) { + flush_all_pending_tool_calls(); } - // Check for finish_reason - if (!choices.empty() && choices[0].contains("finish_reason") && - !choices[0]["finish_reason"].is_null()) { - auto finish_reason_str = choices[0]["finish_reason"].get(); - auto finish_reason = parse_finish_reason(finish_reason_str); - - ai::logger::log_debug("Stream finished with reason: {}", - finish_reason_str); - - finish_event_pushed_ = true; - - if (json.contains("usage")) { - auto usage = parse_usage(json["usage"]); - ai::logger::log_info( - "Stream completed - tokens used: {} prompt, {} completion, {} " - "total", - usage.prompt_tokens, usage.completion_tokens, usage.total_tokens); - push_event(StreamEvent(kStreamEventTypeFinish, usage, finish_reason)); - } else { - push_event(StreamEvent(kStreamEventTypeFinish)); - } - } - } catch (const std::exception& e) { - ai::logger::log_error("Failed to parse SSE line: {} - Line content: {}", - e.what(), data); + ai::logger::log_debug("Stream finished with reason: {}", + finish_reason_str); } - } else if (!line.empty()) { - ai::logger::log_debug("Ignoring non-data SSE line: {}", line); - } -} -void OpenAIStreamImpl::push_event(StreamEvent event) { - event_queue_.enqueue(std::move(event)); -} - -void OpenAIStreamImpl::push_finish_event_if_needed() { - bool expected = false; - if (finish_event_pushed_.compare_exchange_strong(expected, true)) { - ai::logger::log_debug("Pushing finish event to queue"); - event_queue_.enqueue(StreamEvent(kStreamEventTypeFinish)); - } else { - ai::logger::log_debug("Finish event already pushed, skipping"); + if (json.contains("usage") && !json["usage"].is_null()) { + usage_ = parse_usage(json["usage"]); + ai::logger::log_info( + "Stream completed - tokens used: {} prompt, {} completion, {} " + "total", + usage_->prompt_tokens, usage_->completion_tokens, + usage_->total_tokens); + } + } catch (const std::exception& e) { + ai::logger::log_error("Failed to parse SSE line: {} - Line content: {}", + e.what(), data); } } -void OpenAIStreamImpl::mark_complete() { - is_complete_ = true; // Atomic write -} - -StreamEvent OpenAIStreamImpl::create_error_event(const std::string& message) { - ai::logger::log_debug("Creating error event: {}", message); - return StreamEvent(kStreamEventTypeError, message); -} - FinishReason OpenAIStreamImpl::parse_finish_reason( const std::string& reason_str) { if (reason_str == "stop") { @@ -285,6 +102,8 @@ FinishReason OpenAIStreamImpl::parse_finish_reason( return kFinishReasonLength; } else if (reason_str == "content_filter") { return kFinishReasonContentFilter; + } else if (reason_str == "tool_calls") { + return kFinishReasonToolCalls; } return kFinishReasonStop; } diff --git a/src/providers/openai/openai_stream.h b/src/providers/openai/openai_stream.h index 4c15c91..d2b514a 100644 --- a/src/providers/openai/openai_stream.h +++ b/src/providers/openai/openai_stream.h @@ -1,58 +1,31 @@ #pragma once -#include "ai/types/stream_result.h" +#include "providers/http_sse_stream.h" -#include -#include -#include -#include -#include +#include +#include #include namespace ai { namespace openai { -class OpenAIStreamImpl : public internal::StreamResultImpl { +class OpenAIStreamImpl : public providers::streaming::HttpSseStream { public: - OpenAIStreamImpl() = default; - ~OpenAIStreamImpl(); - - // Non-copyable, non-movable for thread safety - OpenAIStreamImpl(const OpenAIStreamImpl&) = delete; - OpenAIStreamImpl& operator=(const OpenAIStreamImpl&) = delete; - OpenAIStreamImpl(OpenAIStreamImpl&&) = delete; - OpenAIStreamImpl& operator=(OpenAIStreamImpl&&) = delete; - - void start_stream(const std::string& url, - const httplib::Headers& headers, - const nlohmann::json& request_body); - - StreamEvent get_next_event() override; - bool has_more_events() const override; - void stop_stream() override; + OpenAIStreamImpl() + : HttpSseStream({.default_path = "/v1/chat/completions", + .connection_timeout_seconds = 30, + // 5 minutes for long generations + .read_timeout_seconds = 300}) {} + ~OpenAIStreamImpl() override; private: - void run_stream(const std::string& url, - const httplib::Headers& headers, - const nlohmann::json& request_body); - void parse_sse_line(const std::string& line); - void push_event(StreamEvent event); - void push_finish_event_if_needed(); - void mark_complete(); + void process_sse_line(std::string_view line) override; // Helper functions - StreamEvent create_error_event(const std::string& message); - FinishReason parse_finish_reason(const std::string& reason_str); - Usage parse_usage(const nlohmann::json& usage_json); - - moodycamel::ConcurrentQueue event_queue_; - std::thread stream_thread_; - std::mutex thread_mutex_; - std::atomic is_complete_{false}; - std::atomic should_stop_{false}; - std::atomic finish_event_pushed_{false}; + static FinishReason parse_finish_reason(const std::string& reason_str); + static Usage parse_usage(const nlohmann::json& usage_json); }; } // namespace openai -} // namespace ai \ No newline at end of file +} // namespace ai diff --git a/src/providers/stream_utils.h b/src/providers/stream_utils.h new file mode 100644 index 0000000..a798b80 --- /dev/null +++ b/src/providers/stream_utils.h @@ -0,0 +1,59 @@ +#pragma once + +#include "ai/types/stream_event.h" + +#include +#include +#include + +#include + +namespace ai { +namespace providers { +namespace streaming { + +/// One partially received tool call, accumulated across SSE fragments. +struct PendingToolCall { + std::string id; + std::string name; + std::string arguments; // Accumulated argument fragments (JSON text). + std::string initial_input; // Complete input object sent up front, if any. +}; + +/// Convert an accumulated tool call into a StreamEvent. Returns a tool-call +/// event on success or an error event when the accumulated arguments are not +/// valid JSON. +inline StreamEvent make_tool_call_event(const PendingToolCall& pending) { + const std::string& raw = + pending.arguments.empty() ? pending.initial_input : pending.arguments; + try { + auto arguments = + raw.empty() ? nlohmann::json::object() : nlohmann::json::parse(raw); + return StreamEvent( + ToolCall(pending.id, pending.name, std::move(arguments))); + } catch (const nlohmann::json::exception& e) { + return StreamEvent(kStreamEventTypeError, + "Failed to parse streamed tool-call arguments: " + + std::string(e.what())); + } +} + +/// Split a URL into origin ("scheme://host[:port]") and path, defaulting the +/// path when the URL has none. +inline std::pair split_origin_and_path( + std::string_view url, + std::string_view default_path) { + const auto scheme_pos = url.find("://"); + const auto authority_start = + scheme_pos == std::string_view::npos ? 0 : scheme_pos + 3; + const auto slash_pos = url.find('/', authority_start); + std::string origin(url.substr(0, slash_pos)); + std::string path = slash_pos == std::string_view::npos + ? std::string(default_path) + : std::string(url.substr(slash_pos)); + return {std::move(origin), std::move(path)}; +} + +} // namespace streaming +} // namespace providers +} // namespace ai diff --git a/tests/unit/anthropic_stream_test.cpp b/tests/unit/anthropic_stream_test.cpp index 518eeef..ded11ba 100644 --- a/tests/unit/anthropic_stream_test.cpp +++ b/tests/unit/anthropic_stream_test.cpp @@ -1,3 +1,5 @@ +#include "providers/anthropic/anthropic_stream.h" + #include "../utils/mock_anthropic_client.h" #include "../utils/test_fixtures.h" #include "ai/types/stream_event.h" @@ -98,6 +100,64 @@ TEST_F(AnthropicStreamEventTest, ParseErrorEvent) { EXPECT_THAT(error_data, testing::HasSubstr("invalid_request_error")); } +TEST_F(AnthropicStreamEventTest, ParsesFragmentedResponseWithUsage) { + anthropic::AnthropicStreamImpl stream; + stream.process_sse_chunk_for_testing( + "data: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_" + "tokens\":12}}}\n\n" + "data: {\"type\":\"content_block_delta\",\"delta\":{\"text\":\"Hel"); + stream.process_sse_chunk_for_testing( + "lo\"}}\n\n" + "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"max_" + "tokens\"},\"usage\":{\"output_tokens\":7}}\n\n" + "data: {\"type\":\"message_stop\"}\n\n"); + + auto text_event = stream.get_next_event(); + ASSERT_TRUE(text_event.is_text_delta()); + EXPECT_EQ(text_event.text_delta, "Hello"); + + auto finish_event = stream.get_next_event(); + ASSERT_TRUE(finish_event.is_finish()); + ASSERT_TRUE(finish_event.usage.has_value()); + EXPECT_EQ(finish_event.usage->prompt_tokens, 12); + EXPECT_EQ(finish_event.usage->completion_tokens, 7); + EXPECT_EQ(finish_event.usage->total_tokens, 19); + EXPECT_EQ(finish_event.finish_reason, kFinishReasonLength); +} + +TEST_F(AnthropicStreamEventTest, EmitsOnlyOneFinishEvent) { + anthropic::AnthropicStreamImpl stream; + stream.process_sse_chunk_for_testing( + "data: {\"type\":\"message_stop\"}\n\n" + "data: {\"type\":\"message_stop\"}\n\n"); + + EXPECT_EQ(stream.queued_event_count_for_testing(), 1); + EXPECT_TRUE(stream.get_next_event().is_finish()); +} + +TEST_F(AnthropicStreamEventTest, ParsesFragmentedToolCall) { + anthropic::AnthropicStreamImpl stream; + stream.process_sse_chunk_for_testing( + R"(data: {"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"tool_123","name":"get_weather","input":{}}} + +data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"loc"}} + +)"); + stream.process_sse_chunk_for_testing( + R"(data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"ation\":\"Chicago\"}"}} + +data: {"type":"content_block_stop","index":1} + +)"); + + auto event = stream.get_next_event(); + ASSERT_TRUE(event.is_tool_call()); + ASSERT_TRUE(event.tool_call.has_value()); + EXPECT_EQ(event.tool_call->id, "tool_123"); + EXPECT_EQ(event.tool_call->tool_name, "get_weather"); + EXPECT_EQ(event.tool_call->arguments["location"], "Chicago"); +} + // Mock Stream Implementation Tests class MockAnthropicStreamImpl { public: @@ -391,4 +451,4 @@ TEST_F(AnthropicStreamEventTypesTest, MessageStopEvent) { } } // namespace test -} // namespace ai \ No newline at end of file +} // namespace ai diff --git a/tests/unit/openai_stream_test.cpp b/tests/unit/openai_stream_test.cpp index 616bfa1..c16ca16 100644 --- a/tests/unit/openai_stream_test.cpp +++ b/tests/unit/openai_stream_test.cpp @@ -1,3 +1,5 @@ +#include "providers/openai/openai_stream.h" + #include "../utils/mock_openai_client.h" #include "../utils/test_fixtures.h" #include "ai/types/stream_event.h" @@ -14,9 +16,9 @@ class OpenAIStreamTest : public OpenAITestFixture {}; // StreamOptions Tests TEST_F(OpenAIStreamTest, StreamOptionsBasicConstructor) { - StreamOptions options(GenerateOptions("gpt-4o", "Hello, world!")); + StreamOptions options(GenerateOptions("test-model", "Hello, world!")); - EXPECT_EQ(options.model, "gpt-4o"); + EXPECT_EQ(options.model, "test-model"); EXPECT_EQ(options.prompt, "Hello, world!"); EXPECT_TRUE(options.system.empty()); EXPECT_TRUE(options.messages.empty()); @@ -24,30 +26,30 @@ TEST_F(OpenAIStreamTest, StreamOptionsBasicConstructor) { TEST_F(OpenAIStreamTest, StreamOptionsWithSystemPrompt) { StreamOptions options( - GenerateOptions("gpt-4o", "System prompt", "User prompt")); + GenerateOptions("test-model", "System prompt", "User prompt")); - EXPECT_EQ(options.model, "gpt-4o"); + EXPECT_EQ(options.model, "test-model"); EXPECT_EQ(options.system, "System prompt"); EXPECT_EQ(options.prompt, "User prompt"); } TEST_F(OpenAIStreamTest, StreamOptionsWithMessages) { Messages messages = createSampleConversation(); - StreamOptions options(GenerateOptions("gpt-4o", std::move(messages))); + StreamOptions options(GenerateOptions("test-model", std::move(messages))); - EXPECT_EQ(options.model, "gpt-4o"); + EXPECT_EQ(options.model, "test-model"); EXPECT_FALSE(options.messages.empty()); EXPECT_TRUE(options.has_messages()); } TEST_F(OpenAIStreamTest, StreamOptionsValidation) { - StreamOptions valid_options(GenerateOptions("gpt-4o", "Valid prompt")); + StreamOptions valid_options(GenerateOptions("test-model", "Valid prompt")); EXPECT_TRUE(valid_options.is_valid()); StreamOptions invalid_model(GenerateOptions("", "Valid prompt")); EXPECT_FALSE(invalid_model.is_valid()); - StreamOptions invalid_prompt(GenerateOptions("gpt-4o", "")); + StreamOptions invalid_prompt(GenerateOptions("test-model", "")); EXPECT_FALSE(invalid_prompt.is_valid()); } @@ -90,6 +92,43 @@ TEST_F(StreamEventTest, ParseErrorEvent) { EXPECT_THAT(error_data, testing::HasSubstr("Stream error")); } +TEST_F(StreamEventTest, ParsesFragmentedToolCall) { + openai::OpenAIStreamImpl stream; + stream.process_sse_line_for_testing( + R"(data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_123","type":"function","function":{"name":"get_weather","arguments":"{\"loc"}}]}}]})"); + stream.process_sse_line_for_testing( + R"(data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"ation\":\"Chicago\"}"}}]},"finish_reason":"tool_calls"}]})"); + stream.process_sse_line_for_testing("data: [DONE]"); + + auto tool_event = stream.get_next_event(); + ASSERT_TRUE(tool_event.is_tool_call()); + ASSERT_TRUE(tool_event.tool_call.has_value()); + EXPECT_EQ(tool_event.tool_call->id, "call_123"); + EXPECT_EQ(tool_event.tool_call->tool_name, "get_weather"); + EXPECT_EQ(tool_event.tool_call->arguments["location"], "Chicago"); + + auto finish_event = stream.get_next_event(); + ASSERT_TRUE(finish_event.is_finish()); + ASSERT_TRUE(finish_event.finish_reason.has_value()); + EXPECT_EQ(*finish_event.finish_reason, kFinishReasonToolCalls); + EXPECT_FALSE(finish_event.usage.has_value()); +} + +TEST_F(StreamEventTest, EmitsOneFinishWithTerminalUsage) { + openai::OpenAIStreamImpl stream; + stream.process_sse_line_for_testing( + R"(data: {"choices":[{"delta":{},"finish_reason":"stop"}]})"); + stream.process_sse_line_for_testing( + R"(data: {"choices":[{"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":4,"completion_tokens":3,"total_tokens":7}})"); + stream.process_sse_line_for_testing("data: [DONE]"); + + ASSERT_EQ(stream.queued_event_count_for_testing(), 1); + auto finish_event = stream.get_next_event(); + ASSERT_TRUE(finish_event.is_finish()); + ASSERT_TRUE(finish_event.usage.has_value()); + EXPECT_EQ(finish_event.usage->total_tokens, 7); +} + // Mock Stream Implementation Tests class MockStreamImpl { public: @@ -180,7 +219,7 @@ class StreamErrorTest : public OpenAITestFixture {}; TEST_F(StreamErrorTest, HandleStreamConnectionError) { ControllableOpenAIClient client(kTestApiKey); - StreamOptions options(GenerateOptions("gpt-4o", "Test prompt")); + StreamOptions options(GenerateOptions("test-model", "Test prompt")); client.setShouldFail(true); @@ -193,7 +232,7 @@ TEST_F(StreamErrorTest, HandleStreamConnectionError) { TEST_F(StreamErrorTest, HandleStreamTimeout) { ControllableOpenAIClient client(kTestApiKey); - StreamOptions options(GenerateOptions("gpt-4o", "Test prompt")); + StreamOptions options(GenerateOptions("test-model", "Test prompt")); client.setShouldTimeout(true); @@ -273,13 +312,13 @@ TEST_F(StreamIntegrationTest, StreamWithClientConfiguration) { // Verify client supports streaming EXPECT_TRUE(client.is_valid()); - StreamOptions options(GenerateOptions("gpt-4o", "Stream test")); + StreamOptions options(GenerateOptions("test-model", "Stream test")); auto result = client.stream_text(options); EXPECT_EQ(client.getCallCount(), 1); auto last_options = client.getLastStreamOptions(); - EXPECT_EQ(last_options.model, "gpt-4o"); + EXPECT_EQ(last_options.model, "test-model"); EXPECT_EQ(last_options.prompt, "Stream test"); } @@ -328,4 +367,4 @@ TEST_F(StreamEdgeCaseTest, UnicodeInStream) { } } // namespace test -} // namespace ai \ No newline at end of file +} // namespace ai