Wrap streaming callback properly in C++

sdk_v2/cpp/docs/CppPortGuide.md

@@ -152,11 +152,13 @@ See the "Session caching" TODO in `State of the Branch.md` for current status.
 **C#:** Pull-based via `IAsyncEnumerable<ChatCompletionCreateResponse>`. The caller drives
 iteration with `await foreach`.
 
-**C++:** Push-based via `StreamingCallbackFn` (a `std::function<int(flStreamingCallbackData, void*)>`).
-Each callback delivers an `flStreamingCallbackData` containing an `flItemQueue*` — one item
-is pushed to the queue per callback invocation. The return value is 0 to continue or
-non-zero to cancel. `StreamingThreadTracker` manages the lifecycle of the background
-threads that drive generation to ensure clean shutdown.
+**C++:** Push-based via `Session::SetStreamingCallback()` (a `std::function<int(Item)>`).
+Each callback receives one owning `Item`; the wrapper pops it from the C ABI's queue and
+releases it automatically after the callback unless the caller moves it elsewhere. The
+return value is 0 to continue or non-zero to cancel. `StreamingThreadTracker` manages the
+lifecycle of the background threads that drive generation to ensure clean shutdown.
+The former `std::function<int(flStreamingCallbackData)>` overload remains available but is
+deprecated; new code should use the item-oriented overload.
 
 **Why:** C++ has no equivalent of `IAsyncEnumerable`. Callbacks are the natural C++
 pattern and map cleanly to the C ABI's function-pointer-based streaming interface.
@@ -265,12 +267,12 @@ item in a `Request` or as the delivery mechanism in a streaming callback. Owners
 item transfers through the queue: the consumer that pops an item is responsible for
 deleting it.
 
-**Streaming output:** The `flStreamingCallbackData` passed to the streaming callback
-contains an `flItemQueue*`. Each callback invocation pushes one item to the queue. This
-guarantees ordering — items arrive in generation order regardless of threading. The queue
-carries the same item types as the overall `Response` (e.g., `MessageItem` for text
-tokens, `ToolCallItem` for tool invocations), so streaming output is a per-item view of
-the final response.
+**Streaming output:** At the C ABI, `flStreamingCallbackData` contains an `flItemQueue*`,
+with one item pushed per callback invocation. The public C++ wrapper pops that item and
+passes it to the callback as an owning `Item`. This guarantees ordering — items arrive in
+generation order regardless of threading. Streaming uses incremental item types (e.g., `TextItem`
+for text tokens and `ToolCallItem` for tool invocations), while the final `Response` may
+aggregate them into higher-level items such as `MessageItem`.
 
 **Streaming input:** A `Request` can contain an `ItemQueue` as one of its input items.
 This enables scenarios like realtime audio:

sdk_v2/cpp/examples/basic_chat/main.cc

@@ -55,7 +55,7 @@ void BasicChat(IModel& model) {
   // 5. Check finish reason and usage.
   if (response.GetFinishReason() == FOUNDRY_LOCAL_FINISH_STOP) {
     flUsage usage = response.GetUsage();
-    std::cout << "Tokens — prompt: " << usage.prompt_tokens
+    std::cout << "Tokens - prompt: " << usage.prompt_tokens
               << ", completion: " << usage.completion_tokens
               << ", total: " << usage.total_tokens << "\n";
   }
@@ -85,28 +85,17 @@ void StreamingChat(IModel& model) {
   // 1. Create a session for multi-turn chat.
   ChatSession session(model);
 
-  // 2. Set a streaming callback. Each invocation receives one item in the queue.
-  //    We pop the item from the queue and print any message content incrementally.
+  // 2. Set a streaming callback. Each invocation receives one owning Item.
+  //    We print any message content incrementally; the Item releases itself when the callback returns.
 
   // lambda can capture user data if needed (equivalent to the additional `void* user_data` parameter in the C API).
-  session.SetStreamingCallback([/*user_data*/](flStreamingCallbackData event) -> int {
-    const auto* item_api = detail::item_api();
-
-    flItem* raw_item = nullptr;
-    if (item_api->ItemQueue_TryPop(event.item_queue, &raw_item)) {
-      Item item(*raw_item);
-      // in this example we're streaming each token generated as simple text.
-      // what is streamed is flexible though given flItem is being used
-      if (item.GetType() == FOUNDRY_LOCAL_ITEM_TEXT) {
-        std::cout << item.GetText().text << std::flush;
-      } else {
-        std::cerr << "Unexpected item type" << std::endl;
-      }
-
-      // ~Item calls release.
+  session.SetStreamingCallback([/*user_data*/](Item item) -> int {
+    // in this example we're streaming each token generated as simple text.
+    // what is streamed is flexible though given Item is being used
+    if (item.GetType() == FOUNDRY_LOCAL_ITEM_TEXT) {
+      std::cout << item.GetText().text << std::flush;
     } else {
-      // should never happen. adding item to queue to callback should be 1:1.
-      std::cerr << "Callback invoked but no item in queue" << std::endl;
+      std::cerr << "Unexpected item type" << std::endl;
     }
 
     return 0;  // return non-zero to cancel
@@ -123,7 +112,7 @@ void StreamingChat(IModel& model) {
   // 4. The full response is still available after streaming completes.
   if (response.GetFinishReason() == FOUNDRY_LOCAL_FINISH_STOP) {
     flUsage usage = response.GetUsage();
-    std::cout << "Tokens — prompt: " << usage.prompt_tokens
+    std::cout << "Tokens - prompt: " << usage.prompt_tokens
               << ", completion: " << usage.completion_tokens
               << ", total: " << usage.total_tokens << "\n";
   }

sdk_v2/cpp/examples/realtime_audio/main.cc

@@ -37,27 +37,16 @@ void RealtimeAudioChat(IModel& model, const std::string& audio_path) {
   AudioSession session(model);
 
   // 2. Set a streaming callback to receive incremental text as words are generated.
-  //    Each callback invocation delivers one item in the queue.
+  //    Each callback invocation delivers one owning Item.
 
   // lambda can capture user data if needed (equivalent to the additional `void* user_data` parameter in the C API).
-  session.SetStreamingCallback([/*user_data*/](flStreamingCallbackData event) -> int {
-    const auto* item_api = detail::item_api();
-
-    flItem* raw_item = nullptr;
-    if (item_api->ItemQueue_TryPop(event.item_queue, &raw_item)) {
-      Item item(*raw_item);
-      if (item.GetType() == FOUNDRY_LOCAL_ITEM_SPEECH_SEGMENT) {
-        auto seg = item.GetSpeechSegment();
-        std::cout.write(seg.text.data(), seg.text.size());
-        std::cout.flush();
-      } else {
-        std::cerr << "Unexpected item type" << std::endl;
-      }
-
-      // ~Item calls release.
+  session.SetStreamingCallback([/*user_data*/](Item item) -> int {
+    if (item.GetType() == FOUNDRY_LOCAL_ITEM_SPEECH_SEGMENT) {
+      auto seg = item.GetSpeechSegment();
+      std::cout.write(seg.text.data(), seg.text.size());
+      std::cout.flush();
     } else {
-      // should never happen. adding item to queue to callback should be 1:1.
-      std::cerr << "Callback invoked but no item in queue" << std::endl;
+      std::cerr << "Unexpected item type" << std::endl;
     }
 
     return 0;  // return non-zero to cancel

sdk_v2/cpp/examples/tool_calling/main.cc

@@ -120,17 +120,7 @@ void StreamingToolCalling(IModel& model) {
   // Tokens that belong inside a tool-call block are buffered internally and emitted
   // as a single ToolCallItem once the block closes, so the callback never sees raw
   // `<tool_call>...</tool_call>` markers as text.
-  session.SetStreamingCallback([](flStreamingCallbackData event) -> int {
-    const auto* item_api = detail::item_api();
-
-    flItem* raw_item = nullptr;
-    if (!item_api->ItemQueue_TryPop(event.item_queue, &raw_item)) {
-      std::cerr << "Callback invoked but no item in queue\n";
-      return 0;
-    }
-
-    Item item(*raw_item);  // takes ownership; ~Item releases.
-
+  session.SetStreamingCallback([](Item item) -> int {
     switch (item.GetType()) {
       case FOUNDRY_LOCAL_ITEM_TEXT:
         std::cout << item.GetText().text << std::flush;

sdk_v2/cpp/include/foundry_local/foundry_local_cpp.h

@@ -1034,10 +1034,19 @@ class Session {
   /// Options to apply to all requests on this session.
   Session& SetOptions(const RequestOptions& options);
 
-  /// Set the streaming callback using a std::function.
+  /// Set a callback that receives each streamed item as an owning RAII wrapper.
+  /// The callback may move the item elsewhere to extend its lifetime.
   /// Return 0 to continue, non-zero to cancel.
+  Session& SetStreamingCallback(std::function<int(Item)> callback);
+
+  /// Set the legacy callback that receives the C streaming event and must pop its item manually.
+  /// Return 0 to continue, non-zero to cancel.
+  [[deprecated("Use SetStreamingCallback(std::function<int(Item)>) instead")]]
   Session& SetStreamingCallback(std::function<int(flStreamingCallbackData)> callback);
 
+  /// Clear the streaming callback.
+  Session& SetStreamingCallback(std::nullptr_t);
+
   /// Process the request.
   /// Populates the response with output items, finish reason, and usage.
   Response ProcessRequest(const Request& request);
@@ -1096,17 +1105,30 @@ namespace detail {
 
 /// Adapts a std::function streaming callback for use with the C API's function-pointer + void* interface.
 struct StreamingCallbackHelper {
-  using CallbackFn = std::function<int(flStreamingCallbackData)>;
+  using ItemCallbackFn = std::function<int(Item)>;
+  using LegacyCallbackFn = std::function<int(flStreamingCallbackData)>;
 
-  explicit StreamingCallbackHelper(CallbackFn callback) : callback_(std::move(callback)) {}
+  explicit StreamingCallbackHelper(ItemCallbackFn callback) : item_callback_(std::move(callback)) {}
+  explicit StreamingCallbackHelper(LegacyCallbackFn callback) : legacy_callback_(std::move(callback)) {}
 
   static int CCallback(flStreamingCallbackData data, void* user_data) {
     auto* helper = static_cast<StreamingCallbackHelper*>(user_data);
-    return helper->callback_(data);
+    if (helper->legacy_callback_) {
+      return helper->legacy_callback_(data);
+    }
+
+    flItem* raw_item = nullptr;
+    if (!detail::item_api()->ItemQueue_TryPop(data.item_queue, &raw_item) || !raw_item) {
+      // should never happen. adding item to queue to callback should be 1:1.
+      return 1;
+    }
+
+    return helper->item_callback_(Item(*raw_item));
   }
 
  private:
-  CallbackFn callback_;
+  ItemCallbackFn item_callback_;
+  LegacyCallbackFn legacy_callback_;
 };
 
 /// Adapts a std::function deleter for use with C API data struct deleters.

sdk_v2/cpp/include/foundry_local/foundry_local_cpp.inline.h

@@ -1202,6 +1202,22 @@ inline Session& Session::SetOptions(const RequestOptions& options) {
   return *this;
 }
 
+inline Session& Session::SetStreamingCallback(std::function<int(Item)> callback) {
+  flStreamingCallback callback_fn = nullptr;
+  void* user_data = nullptr;
+
+  if (callback) {
+    streaming_callback_helper_ = std::make_unique<detail::StreamingCallbackHelper>(std::move(callback));
+    callback_fn = &detail::StreamingCallbackHelper::CCallback;
+    user_data = streaming_callback_helper_.get();
+  } else {
+    streaming_callback_helper_.reset();
+  }
+
+  Check(detail::inference_api()->Session_SetStreamingCallback(handle_.get_mutable(), callback_fn, user_data));
+  return *this;
+}
+
 inline Session& Session::SetStreamingCallback(std::function<int(flStreamingCallbackData)> callback) {
   flStreamingCallback callback_fn = nullptr;
   void* user_data = nullptr;
@@ -1218,6 +1234,10 @@ inline Session& Session::SetStreamingCallback(std::function<int(flStreamingCallb
   return *this;
 }
 
+inline Session& Session::SetStreamingCallback(std::nullptr_t) {
+  return SetStreamingCallback(std::function<int(Item)>{});
+}
+
 inline flSession* detail::CreateSession(IModel& model) {
   flSession* session = nullptr;
   Check(detail::inference_api()->Session_Create(

sdk_v2/cpp/test/sdk_api/chat_session_test.cc

@@ -501,13 +501,7 @@ TEST_F(ToolCallFixture, ToolCallStreamingWithRequired) {
   std::vector<StreamedToolCall> collected_tool_calls;
   std::string collected_text;
 
-  session.SetStreamingCallback([&](flStreamingCallbackData data) -> int {
-    flItem* raw_item = nullptr;
-    if (!detail::item_api()->ItemQueue_TryPop(data.item_queue, &raw_item) || !raw_item) {
-      return 0;
-    }
-
-    Item item(*raw_item);
+  session.SetStreamingCallback([&](Item item) -> int {
     ++callback_count;
 
     if (item.GetType() == FOUNDRY_LOCAL_ITEM_TOOL_CALL) {

sdk_v2/cpp/test/sdk_api/streaming_audio_test.cc

@@ -250,16 +250,7 @@ TEST_F(StreamingAudioFixture, StreamingCallbackReceivesTokens) {
   std::mutex text_mutex;
   std::string streamed_text;
 
-  session.SetStreamingCallback([&](flStreamingCallbackData data) {
-    // Pop the item from the queue — this is the established contract.
-    flItem* raw_item = nullptr;
-    if (!detail::item_api()->ItemQueue_TryPop(data.item_queue, &raw_item) || !raw_item) {
-      return 0;
-    }
-
-    // Wrap in Item for RAII release and checked accessors.
-    Item item(*raw_item);
-
+  session.SetStreamingCallback([&](Item item) {
     // Audio output is always a SpeechSegmentItem per token.
     EXPECT_EQ(item.GetType(), FOUNDRY_LOCAL_ITEM_SPEECH_SEGMENT);
     auto seg = item.GetSpeechSegment();

sdk_v2/js/native/src/session.cc

@@ -192,25 +192,19 @@ class StreamWorker : public Napi::AsyncWorker {
     try {
       auto tsfn = tsfn_;
       auto* ctx = ctx_;
-      sess_->SetStreamingCallback([tsfn, ctx](flStreamingCallbackData data) -> int {
+      sess_->SetStreamingCallback([tsfn, ctx](foundry_local::Item item) -> int {
         (void)ctx;
-        if (data.item_queue == nullptr) return 0;
-        flItem* raw = nullptr;
-        while (foundry_local::detail::item_api()->ItemQueue_TryPop(data.item_queue, &raw)) {
-          if (raw == nullptr) break;
-          auto* item = new foundry_local::Item(*raw);
-          napi_status status = tsfn.BlockingCall(
-              item, [](Napi::Env env, Napi::Function jsCb, foundry_local::Item* it) {
-                Napi::HandleScope scope(env);
-                Napi::Value js_item = ItemToJs(env, *it);
-                delete it;
-                jsCb.Call({js_item});
-              });
-          if (status != napi_ok) {
-            delete item;
-            return 1;
-          }
-          raw = nullptr;
+        auto* streamed_item = new foundry_local::Item(std::move(item));
+        napi_status status = tsfn.BlockingCall(
+            streamed_item, [](Napi::Env env, Napi::Function jsCb, foundry_local::Item* item) {
+              Napi::HandleScope scope(env);
+              Napi::Value js_item = ItemToJs(env, *item);
+              delete item;
+              jsCb.Call({js_item});
+            });
+        if (status != napi_ok) {
+          delete streamed_item;
+          return 1;
         }
         return 0;
       });