Changes
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 1130f30..e08fb5a 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -75,6 +75,7 @@ add_library(agent_smith ${AGENT_SMITH_LIBRARY_TYPE}
src/detail/timer_service.cpp
src/message.cpp
src/runtime.cpp
+ src/sse_parser.cpp
)
add_library(agent_smith::agent_smith ALIAS agent_smith)
diff --git a/include/agent_smith/detail/sse_parser.hpp b/include/agent_smith/detail/sse_parser.hpp
new file mode 100644
index 0000000..0423d41
--- /dev/null
+++ b/include/agent_smith/detail/sse_parser.hpp
@@ -0,0 +1,25 @@
+#pragma once
+
+#include <string>
+#include <string_view>
+#include <vector>
+
+namespace agent_smith::detail
+{
+
+/// Incrementally extracts SSE data payloads from arbitrary byte chunks.
+class SseFrameParser
+{
+public:
+ /// Consume a chunk and return every complete non-terminal data payload.
+ std::vector<std::string> feed(std::string_view bytes);
+
+ /// Return whether the terminal SSE payload `[DONE]` has been received.
+ bool finished() const noexcept;
+
+private:
+ std::string buffer_;
+ bool finished_ = false;
+};
+
+} // namespace agent_smith::detail
diff --git a/src/sse_parser.cpp b/src/sse_parser.cpp
new file mode 100644
index 0000000..405ab99
--- /dev/null
+++ b/src/sse_parser.cpp
@@ -0,0 +1,133 @@
+#include "agent_smith/detail/sse_parser.hpp"
+
+#include <optional>
+#include <string_view>
+#include <utility>
+
+namespace agent_smith::detail
+{
+namespace
+{
+
+struct FrameBoundary
+{
+ std::size_t position;
+ std::size_t length;
+};
+
+std::optional<FrameBoundary> findFrameBoundary(std::string_view bytes)
+{
+ const std::size_t lf_boundary = bytes.find("\n\n");
+ const std::size_t crlf_boundary = bytes.find("\r\n\r\n");
+
+ if(lf_boundary == std::string_view::npos &&
+ crlf_boundary == std::string_view::npos)
+ {
+ return std::nullopt;
+ }
+ if(crlf_boundary == std::string_view::npos ||
+ (lf_boundary != std::string_view::npos && lf_boundary < crlf_boundary))
+ {
+ return FrameBoundary{
+ .position = lf_boundary,
+ .length = 2,
+ };
+ }
+ return FrameBoundary{
+ .position = crlf_boundary,
+ .length = 4,
+ };
+}
+
+std::optional<std::string> dataPayload(std::string_view frame)
+{
+ std::string payload;
+ bool has_data = false;
+ std::size_t line_start = 0;
+
+ while(line_start <= frame.size())
+ {
+ const std::size_t line_end = frame.find('\n', line_start);
+ const std::size_t length = line_end == std::string_view::npos
+ ? frame.size() - line_start
+ : line_end - line_start;
+ std::string_view line = frame.substr(line_start, length);
+ if(!line.empty() && line.back() == '\r')
+ {
+ line.remove_suffix(1);
+ }
+
+ if(line.starts_with("data:"))
+ {
+ std::string_view value = line.substr(5);
+ if(value.starts_with(' '))
+ {
+ value.remove_prefix(1);
+ }
+ if(has_data)
+ {
+ payload.push_back('\n');
+ }
+ payload.append(value);
+ has_data = true;
+ }
+
+ if(line_end == std::string_view::npos)
+ {
+ break;
+ }
+ line_start = line_end + 1;
+ }
+
+ if(!has_data)
+ {
+ return std::nullopt;
+ }
+ return payload;
+}
+
+} // namespace
+
+std::vector<std::string> SseFrameParser::feed(std::string_view bytes)
+{
+ if(finished_)
+ {
+ return {};
+ }
+
+ buffer_.append(bytes);
+ std::vector<std::string> payloads;
+
+ while(true)
+ {
+ const auto boundary = findFrameBoundary(buffer_);
+ if(!boundary)
+ {
+ break;
+ }
+
+ std::string frame = buffer_.substr(0, boundary->position);
+ buffer_.erase(0, boundary->position + boundary->length);
+ auto payload = dataPayload(frame);
+ if(!payload)
+ {
+ continue;
+ }
+ if(*payload == "[DONE]")
+ {
+ finished_ = true;
+ buffer_.clear();
+ break;
+ }
+ payloads.push_back(std::move(*payload));
+ }
+
+ return payloads;
+}
+
+bool SseFrameParser::finished() const noexcept
+{
+ return finished_;
+}
+
+} // namespace agent_smith::detail
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 6e38f7b..50ce20a 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -1,5 +1,6 @@
add_executable(agent_smith_test_skeleton
test_message.cpp
+ test_sse_parser.cpp
test_skeleton.cpp
)
target_link_libraries(agent_smith_test_skeleton
diff --git a/tests/test_sse_parser.cpp b/tests/test_sse_parser.cpp
new file mode 100644
index 0000000..3adc718
--- /dev/null
+++ b/tests/test_sse_parser.cpp
@@ -0,0 +1,98 @@
+#include <string>
+#include <vector>
+
+#include <agent_smith/detail/sse_parser.hpp>
+#include <gtest/gtest.h>
+
+namespace agent_smith::detail
+{
+namespace
+{
+
+TEST(SseFrameParserTest, EmitsDataPayloadsAndIgnoresOtherFields)
+{
+ SseFrameParser parser;
+
+ const auto payloads = parser.feed("event: completion\n"
+ "id: 7\n"
+ "data: {\"first\":true}\n"
+ "\n"
+ ": keepalive\n"
+ "data: {\"second\":true}\n"
+ "\n");
+
+ EXPECT_EQ(payloads, (std::vector<std::string>{"{\"first\":true}",
+ "{\"second\":true}"}));
+ EXPECT_FALSE(parser.finished());
+}
+
+TEST(SseFrameParserTest, RetainsIncompleteFramesAcrossChunks)
+{
+ SseFrameParser parser;
+
+ EXPECT_TRUE(parser.feed("data: {\"choi").empty());
+ const auto payloads = parser.feed("ces\":[1]}\n\n");
+
+ EXPECT_EQ(payloads, (std::vector<std::string>{"{\"choices\":[1]}"}));
+}
+
+TEST(SseFrameParserTest, JoinsMultipleDataLinesWithNewlines)
+{
+ SseFrameParser parser;
+
+ const auto payloads = parser.feed("data: first\n"
+ "data: second\n"
+ "\n");
+
+ EXPECT_EQ(payloads, (std::vector<std::string>{"first\nsecond"}));
+}
+
+TEST(SseFrameParserTest, HandlesCarriageReturnLineEndings)
+{
+ SseFrameParser parser;
+
+ const auto payloads = parser.feed("data: one\r\n\r\n");
+
+ EXPECT_EQ(payloads, (std::vector<std::string>{"one"}));
+}
+
+TEST(SseFrameParserTest, StopsAtDoneAndDiscardsLaterPayloads)
+{
+ SseFrameParser parser;
+
+ const auto payloads = parser.feed("data: before\n\n"
+ "data: [DONE]\n\n"
+ "data: after\n\n");
+
+ EXPECT_EQ(payloads, (std::vector<std::string>{"before"}));
+ EXPECT_TRUE(parser.finished());
+ EXPECT_TRUE(parser.feed("data: ignored\n\n").empty());
+}
+
+TEST(SseFrameParserTest, ProducesIdenticalPayloadsAtEveryByteSplit)
+{
+ const std::string stream = "event: completion\n"
+ "data: {\"choices\":[1]}\n"
+ "\n"
+ "data: {\"usage\":{\"total_tokens\":4}}\n"
+ "\n";
+ const std::vector<std::string> expected{
+ "{\"choices\":[1]}",
+ "{\"usage\":{\"total_tokens\":4}}",
+ };
+
+ for(std::size_t split = 0; split <= stream.size(); ++split)
+ {
+ SseFrameParser parser;
+ std::vector<std::string> payloads =
+ parser.feed(stream.substr(0, split));
+ const auto suffix_payloads = parser.feed(stream.substr(split));
+ payloads.insert(payloads.end(), suffix_payloads.begin(),
+ suffix_payloads.end());
+
+ EXPECT_EQ(payloads, expected) << "split at byte " << split;
+ }
+}
+
+} // namespace
+} // namespace agent_smith::detail