perf(serve): append streamed output without copying accumulated text

edge0-infer/tests/unit/cli/serve_protocol_test.cpp

@@ -3,6 +3,36 @@
 #include <fstream>
 #include <iostream>
 #include <set>
+#include <algorithm>
+#include <cstdlib>
+#include <new>
+namespace allocation_probe {
+bool active=false;
+std::size_t maximum=0,large=0;
+constexpr std::size_t threshold=32768;
+// Keep the deallocator out of line so GCC does not diagnose an inlined free
+// against the matching replacement operator new used by this test.
+#if defined(__GNUC__)
+__attribute__((noinline))
+#elif defined(_MSC_VER)
+__declspec(noinline)
+#endif
+void release(void* p) noexcept {std::free(p);}
+}
+void* operator new(std::size_t bytes) {
+    if (allocation_probe::active) {
+        allocation_probe::maximum=std::max(allocation_probe::maximum,bytes);
+        if (bytes>=allocation_probe::threshold) ++allocation_probe::large;
+    }
+    if (auto* p=std::malloc(bytes?bytes:1)) return p;
+    throw std::bad_alloc();
+}
+void operator delete(void* p) noexcept {allocation_probe::release(p);}
+void operator delete(void* p,std::size_t) noexcept {allocation_probe::release(p);}
+void* operator new[](std::size_t bytes) {return ::operator new(bytes);}
+void operator delete[](void* p) noexcept {allocation_probe::release(p);}
+void operator delete[](void* p,std::size_t) noexcept {allocation_probe::release(p);}
+
 using portal::Json;
 using portal::Protocol;
 void rejects(const Json& j,Protocol p,int status=400) {
@@ -55,6 +85,25 @@ void metrics_contract() {
     }
 }
 int main(int argc,char** argv) {
+    for (auto kind:{edge0::OutputKind::answer,edge0::OutputKind::reasoning,edge0::OutputKind::tool_call_arguments}) {
+        portal::Request request;request.protocol=Protocol::responses;
+        portal::Encoder encoder(request,"test","1");
+        edge0::OutputEvent event;
+        if (kind==edge0::OutputKind::tool_call_arguments) {
+            event.kind=edge0::OutputKind::tool_call_begin;event.tool_call={"call_a","f",""};encoder.accept(event);
+        }
+        event.kind=kind;event.text=std::string(1024,'x');
+        allocation_probe::large=0;allocation_probe::active=true;
+        for (int i=0;i<256;++i) encoder.accept(event);
+        allocation_probe::active=false;
+        std::cout << "large_response_allocations=" << allocation_probe::large << "\n";
+        CHECK(allocation_probe::large<=8);
+        edge0::GenerationResult result;result.stop_reason=edge0::StopReason::end_of_generation;
+        encoder.finish(result);
+        const auto full=encoder.complete(result);
+        const auto actual=kind==edge0::OutputKind::tool_call_arguments?full["output"][0]["arguments"]:full["output"][0]["content"][0]["text"];
+        CHECK(actual==std::string(256*1024,'x'));
+    }
     metrics_contract();
     CHECK(argc==2);
     std::ifstream input(argv[1]);Json fixtures;input>>fixtures;

edge0-infer/tools/cli/serve_protocol.cpp

@@ -328,13 +328,13 @@ std::vector<std::string> Encoder::accept(const edge0::OutputEvent& e) {
     if(e.kind==K::tool_call_begin) {open(records,e);call_indices_[e.index]=std::size_t(active_);return records;}
     if(e.kind==K::tool_call_arguments) {
         auto it=call_indices_.find(e.index);if(it==call_indices_.end())throw std::logic_error("Tool arguments without begin");
-        auto& item=output_[it->second];item["arguments"]=item["arguments"].get<std::string>()+e.text;
+        auto& item=output_[it->second];item["arguments"].get_ref<std::string&>().append(e.text);
         records.push_back(event("response.function_call_arguments.delta",{{"output_index",it->second},{"item_id",item["id"]},{"delta",e.text}}));return records;
     }
     if(e.kind==K::answer || e.kind==K::reasoning) {
         const std::string type=e.kind==K::answer?"message":"reasoning";
         if(active_<0 || output_[active_]["type"]!=type)open(records,e);
-        auto& item=output_[active_];auto& part=item["content"][0];part["text"]=part["text"].get<std::string>()+e.text;
+        auto& item=output_[active_];auto& part=item["content"][0];part["text"].get_ref<std::string&>().append(e.text);
         Json value={{"output_index",active_},{"item_id",item["id"]},{"content_index",0},{"delta",e.text}};
         if(e.kind==K::answer) {value["logprobs"]=pending_probs_;for(const auto& p:pending_probs_)part["logprobs"].push_back(p);pending_probs_=Json::array();}
         records.push_back(event(e.kind==K::answer?"response.output_text.delta":"response.reasoning_text.delta",value));