Add cancellation and deadline support to Session.process_request

sdk_v2/cpp/CMakeLists.txt

@@ -215,6 +215,8 @@ set(FOUNDRY_LOCAL_SOURCES
     src/inferencing/generative/chat/chat_generator.cc
     src/inferencing/session/session.cc
     src/inferencing/session/session_manager.cc
+    src/inferencing/session/oga_generator_cancellable.cc
+    src/inferencing/session/live_session_registry.cc
     src/inferencing/generative/chat/chat_session.cc
     src/inferencing/generative/chat/chat_template.cc
     src/configuration.cc

sdk_v2/cpp/include/foundry_local/foundry_local_c.h

@@ -184,6 +184,9 @@ typedef enum flErrorCode {
   FOUNDRY_LOCAL_ERROR_INVALID_USAGE = 4,
   FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED = 5,
   FOUNDRY_LOCAL_ERROR_NETWORK = 6,
+  /// An operation exceeded its caller-supplied deadline. Distinct from
+  /// FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED, which means an explicit cancel.
+  FOUNDRY_LOCAL_ERROR_TIMEOUT = 7,
 } flErrorCode;
 
 typedef enum flLogLevel {
@@ -892,6 +895,24 @@ struct flInferenceApi {
   /// If all turns are undone, the cached generator is destroyed.
   FL_API_STATUS(Session_UndoTurns, _In_ flSession* session, size_t count);
 
+  /// Set a wall-clock budget for the request, in milliseconds. 0 (the default) means no
+  /// deadline. The clock starts when Session_ProcessRequest is called and is re-armed on
+  /// every call, so a Request may be reused.
+  ///
+  /// On expiry the in-flight generation is interrupted — including mid-compute, not just
+  /// at token boundaries — the session's model reference is released, and
+  /// Session_ProcessRequest returns FOUNDRY_LOCAL_ERROR_TIMEOUT.
+  FL_API_STATUS(Request_SetTimeoutMs, _In_ flRequest* request, uint64_t timeout_ms);
+
+  /// Cancel the request currently in flight on this session, and make subsequent
+  /// Session_ProcessRequest calls fail fast with FOUNDRY_LOCAL_ERROR_INVALID_USAGE.
+  ///
+  /// Unlike Request_Cancel — which only reaches generation loops that poll the request —
+  /// this interrupts the underlying engine, so it also stops a non-streaming request that
+  /// is blocked inside a long prefill or decode. Safe to call from any thread; idempotent.
+  /// This is the supported way to release a session that is pinning a model.
+  FL_API_STATUS(Session_Cancel, _In_ flSession* session);
+
   // End V1
 };
 

sdk_v2/cpp/include/foundry_local/foundry_local_cpp.h

@@ -20,6 +20,7 @@
 #include "foundry_local/foundry_local_c.h"
 
 #include <cassert>
+#include <chrono>
 #include <cstddef>
 #include <cstdint>
 #include <cstdlib>
@@ -955,6 +956,15 @@ class Request {
   /// Cancel the current request. Inferencing will stop as soon as possible.
   void Cancel();
 
+  /// Set a wall-clock deadline for the request, in milliseconds. Pass 0 to disable.
+  ///
+  /// The deadline covers the whole ProcessRequest call, including prefill, and applies to
+  /// streaming and non-streaming alike. On expiry the run is interrupted and
+  /// ProcessRequest throws with FOUNDRY_LOCAL_ERROR_TIMEOUT.
+  ///
+  /// The deadline is re-armed on each ProcessRequest call, so a Request may be reused.
+  Request& SetTimeout(std::chrono::milliseconds timeout);
+
   const flRequest* native_handle() const noexcept { return handle_.get(); }
 
  private:
@@ -1001,6 +1011,17 @@ class Session {
   /// Populates the response with output items, finish reason, and usage.
   Response ProcessRequest(const Request& request);
 
+  /// Interrupt any requests currently running on this session.
+  ///
+  /// Thread-safe and callable from any thread — this is the point: it is meant to be
+  /// invoked from a different thread than the one blocked in ProcessRequest. It
+  /// interrupts inferencing mid-compute rather than only between tokens, so a
+  /// non-terminating generation cannot keep the session's reference to the model alive.
+  ///
+  /// The interrupted ProcessRequest returns promptly with a canceled finish reason.
+  /// Idempotent; safe to call when no request is running.
+  void Cancel();
+
  protected:
   detail::Base<flSession> handle_;
 

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

@@ -1102,6 +1102,12 @@ inline void Request::Cancel() {
   Check(detail::inference_api()->Request_Cancel(handle_.get_mutable()));
 }
 
+inline Request& Request::SetTimeout(std::chrono::milliseconds timeout) {
+  const auto ms = timeout.count() > 0 ? static_cast<uint64_t>(timeout.count()) : uint64_t{0};
+  Check(detail::inference_api()->Request_SetTimeoutMs(handle_.get_mutable(), ms));
+  return *this;
+}
+
 inline flRequest* detail::CreateRequest() {
   flRequest* req = nullptr;
   Check(detail::inference_api()->Request_Create(&req));
@@ -1155,6 +1161,10 @@ inline Response Session::ProcessRequest(const Request& request) {
   return Response(response);
 }
 
+inline void Session::Cancel() {
+  Check(detail::inference_api()->Session_Cancel(handle_.get_mutable()));
+}
+
 inline Session& Session::SetOptions(const RequestOptions& options) {
   KeyValuePairs kvp = detail::ToKeyValuePairs(options);
   Check(detail::inference_api()->Session_SetOptions(handle_.get_mutable(), kvp.native_handle()));

sdk_v2/cpp/src/c_api.cc

@@ -22,6 +22,7 @@
 #include "manager.h"
 #include "ep_detection/ep_bootstrapper.h"
 
+#include <chrono>
 #include <functional>
 #include <map>
 #include <memory>
@@ -1625,6 +1626,17 @@ FL_API_STATUS_IMPL(Request_CancelImpl, flRequest* request) {
   API_IMPL_END
 }
 
+FL_API_STATUS_IMPL(Request_SetTimeoutMsImpl, flRequest* request, uint64_t timeout_ms) {
+  API_IMPL_BEGIN
+  if (!request) {
+    return MakeStatus(FOUNDRY_LOCAL_ERROR_INVALID_ARGUMENT, "null argument");
+  }
+
+  AsImpl(request)->SetTimeout(std::chrono::milliseconds(timeout_ms));
+  return nullptr;
+  API_IMPL_END
+}
+
 // --- Response ---
 
 FL_API_STATUS_IMPL(Response_CreateImpl, flResponse** out_response) {
@@ -1746,9 +1758,19 @@ FL_API_STATUS_IMPL(Session_SetOptionsImpl, flSession* session, const flKeyValueP
   API_IMPL_END
 }
 
-FL_API_STATUS_IMPL(Session_ProcessRequestImpl, flSession* session, const flRequest* request,
-                   flResponse** response) {
+FL_API_STATUS_IMPL(Session_CancelImpl, flSession* session) {
   API_IMPL_BEGIN
+  if (!session) {
+    return MakeStatus(FOUNDRY_LOCAL_ERROR_INVALID_ARGUMENT, "null session");
+  }
+
+  AsImpl(session)->Cancel();
+  return nullptr;
+  API_IMPL_END
+}
+
+FL_API_STATUS_IMPL(Session_ProcessRequestImpl, flSession* session, const flRequest* request,
+                   flResponse** response) {  API_IMPL_BEGIN
   if (!session || !request || !response) {
     return MakeStatus(FOUNDRY_LOCAL_ERROR_INVALID_ARGUMENT, "null argument");
   }
@@ -1816,6 +1838,8 @@ static const flInferenceApi g_inference_api = {
     Session_RemoveToolDefinitionImpl,
     Session_GetTurnCountImpl,
     Session_UndoTurnsImpl,
+    Request_SetTimeoutMsImpl,
+    Session_CancelImpl,
 };
 
 // ========================================================================

sdk_v2/cpp/src/inferencing/generative/audio/audio_generator.h

@@ -2,6 +2,8 @@
 // Licensed under the MIT License.
 #pragma once
 
+#include "inferencing/session/cancellable.h"
+
 #include <string>
 
 namespace fl {
@@ -10,7 +12,7 @@ namespace fl {
 /// One generator per request — not reusable, not thread-safe.
 /// Same pull-based iterator pattern as ChatGenerator:
 ///   while (!IsDone()) { GenerateNextToken(); text += Decode(); }
-class AudioGenerator {
+class AudioGenerator : public ICancellable {
  public:
   virtual ~AudioGenerator() = default;
 
@@ -39,7 +41,7 @@ class AudioGenerator {
 
   /// Request cancellation of generation. Thread-safe — can be called from another thread.
   /// After cancellation, IsDone() should return true on the next check.
-  virtual void Cancel() = 0;
+  void Cancel() override = 0;
 
  protected:
   AudioGenerator() = default;

sdk_v2/cpp/src/inferencing/generative/audio/audio_session.cc

@@ -2,6 +2,7 @@
 // Licensed under the MIT License.
 
 #include "inferencing/generative/audio/audio_session.h"
+#include "inferencing/session/oga_generator_cancellable.h"
 
 #include "contracts/audio_transcriptions.h"
 #include "inferencing/generative/audio/onnx_audio_generator.h"
@@ -248,22 +249,28 @@ void AudioSession::ProcessRequestImpl(const Request& request, Response& response
   std::vector<std::unique_ptr<SpeechSegmentItem>> segments;
   segments.reserve(kInitialTokenCapacity);
 
-  while (!generator->IsDone() && !request.canceled) {
-    generator->GenerateNextToken();
-    std::string token = generator->Decode();
+  // Publish the generator so Session::Cancel() and the deadline watchdog can interrupt
+  // an in-flight compute, not just the loop between tokens.
+  {
+    ActiveGenerator active(*this, *generator);
 
-    if (!token.empty()) {
-      segments.push_back(MakeNoneSegment(token));
+    while (!generator->IsDone() && !request.ShouldStop()) {
+      generator->GenerateNextToken();
+      std::string token = generator->Decode();
 
-      if (streaming_callback) {
-        streaming_callback->PushItem(MakeNoneSegment(token));
-      }
+      if (!token.empty()) {
+        segments.push_back(MakeNoneSegment(token));
 
-      token_texts.push_back(std::move(token));
-    }
+        if (streaming_callback) {
+          streaming_callback->PushItem(MakeNoneSegment(token));
+        }
 
-    if (request.canceled) {
-      generator->Cancel();
+        token_texts.push_back(std::move(token));
+      }
+
+      if (request.canceled) {
+        generator->Cancel();
+      }
     }
   }
 
@@ -334,6 +341,11 @@ void AudioSession::ProcessStreamingAudio(const AudioItem& format_item, ItemQueue
   auto generator = OgaGenerator::Create(oga_model, *gen_params);
   auto tokenizer_stream = OgaTokenizerStream::Create(Model().Tokenizer().Oga());
 
+  // Publish the raw generator so Session::Cancel() and the deadline watchdog can interrupt
+  // an in-flight encoder/decoder pass. Spans steps 3-5 below, which all drive this generator.
+  OgaGeneratorCancellable cancellable(*generator);
+  ActiveGenerator active(*this, cancellable);
+
   auto streaming_callback = CreateCallbackHandler(request);
   std::vector<std::string> token_texts;
   token_texts.reserve(kInitialTokenCapacity);
@@ -352,7 +364,7 @@ void AudioSession::ProcessStreamingAudio(const AudioItem& format_item, ItemQueue
   }
 
   // 4. Read from queue until finished or cancelled
-  while (!request.canceled) {
+  while (!request.ShouldStop()) {
     auto item = queue.WaitAndPop(std::chrono::milliseconds(100));
 
     if (!item) {
@@ -428,7 +440,7 @@ void AudioSession::DecodeTokens(OgaGenerator& generator, OgaTokenizerStream& tok
                                 const std::unique_ptr<CallbackHandler>& callback,
                                 const Request& request,
                                 int& completion_tokens) {
-  while (!generator.IsDone() && !generator.IsSessionTerminated() && !request.canceled) {
+  while (!generator.IsDone() && !generator.IsSessionTerminated() && !request.ShouldStop()) {
     generator.GenerateNextToken();
     auto next_tokens = generator.GetNextTokens();
 
@@ -502,25 +514,31 @@ void AudioSession::ProcessAudioTranscriptionJson(const std::string& request_json
 
   // Generate token-by-token
   std::string text;
-  while (!generator->IsDone() && !original_request.canceled) {
-    generator->GenerateNextToken();
-    std::string token = generator->Decode();
-
-    if (!token.empty()) {
-      text += token;
-
-      if (is_streaming) {
-        // Emit streaming chunk as an OPENAI_JSON-tagged TextItem wrapping AudioTranscriptionResponse.
-        AudioTranscriptionResponse chunk;
-        chunk.id = response_id;
-        chunk.text = token;
-        streaming_callback->PushItem(std::make_unique<TextItem>(nlohmann::json(chunk).dump(),
-                                                                FOUNDRY_LOCAL_TEXT_ITEM_TYPE_OPENAI_JSON));
+  {
+    // Publish the generator so Session::Cancel() and the deadline watchdog can interrupt
+    // an in-flight compute, not just the loop between tokens.
+    ActiveGenerator active(*this, *generator);
+
+    while (!generator->IsDone() && !original_request.ShouldStop()) {
+      generator->GenerateNextToken();
+      std::string token = generator->Decode();
+
+      if (!token.empty()) {
+        text += token;
+
+        if (is_streaming) {
+          // Emit streaming chunk as an OPENAI_JSON-tagged TextItem wrapping AudioTranscriptionResponse.
+          AudioTranscriptionResponse chunk;
+          chunk.id = response_id;
+          chunk.text = token;
+          streaming_callback->PushItem(std::make_unique<TextItem>(nlohmann::json(chunk).dump(),
+                                                                  FOUNDRY_LOCAL_TEXT_ITEM_TYPE_OPENAI_JSON));
+        }
       }
-    }
 
-    if (original_request.canceled) {
-      generator->Cancel();
+      if (original_request.canceled) {
+        generator->Cancel();
+      }
     }
   }
 
@@ -583,7 +601,7 @@ void AudioSession::DecodeNemotronTokens(OgaGenerator& generator, OgaTokenizerStr
                                         int& completion_tokens) const {
   const bool is_streaming = (streaming_callback != nullptr);
 
-  while (!generator.IsDone() && !generator.IsSessionTerminated() && !original_request.canceled) {
+  while (!generator.IsDone() && !generator.IsSessionTerminated() && !original_request.ShouldStop()) {
     generator.GenerateNextToken();
     auto next_tokens = generator.GetNextTokens();
     if (next_tokens.empty()) {
@@ -618,7 +636,7 @@ void AudioSession::RunNemotronDecodePass(std::unique_ptr<OgaNamedTensors> tensor
                                          const std::unique_ptr<CallbackHandler>& streaming_callback,
                                          const std::string& response_id, const Request& original_request,
                                          int& completion_tokens) const {
-  if (!tensors || original_request.canceled) {
+  if (!tensors || original_request.ShouldStop()) {
     return;
   }
 
@@ -659,6 +677,11 @@ void AudioSession::ProcessNemotronFileTranscription(const AudioTranscriptionRequ
   auto generator = OgaGenerator::Create(oga_model, *generator_params);
   TryNemotronLanguageId(*generator, language);
 
+  // Publish the raw generator so cancellation/deadline can interrupt an in-flight
+  // encoder or decode pass rather than only stopping between chunks.
+  OgaGeneratorCancellable cancellable(*generator);
+  ActiveGenerator active(*this, cancellable);
+
   auto streaming_callback = CreateCallbackHandler(original_request);
   std::string response_id = ResponseConverter::GenerateId("audio");
 
@@ -667,7 +690,7 @@ void AudioSession::ProcessNemotronFileTranscription(const AudioTranscriptionRequ
   int completion_tokens = 0;
 
   constexpr size_t kNemotronSamplesPerChunk = 1600;  // 100ms at 16kHz
-  for (size_t offset = 0; offset < samples.size() && !original_request.canceled;
+  for (size_t offset = 0; offset < samples.size() && !original_request.ShouldStop();
        offset += kNemotronSamplesPerChunk) {
     size_t count = std::min(kNemotronSamplesPerChunk, samples.size() - offset);
     RunNemotronDecodePass(processor->Process(samples.data() + offset, count), *generator, *tokenizer_stream, text,

sdk_v2/cpp/src/inferencing/generative/chat/chat_generator.h

@@ -2,6 +2,8 @@
 // Licensed under the MIT License.
 #pragma once
 
+#include "inferencing/session/cancellable.h"
+
 #include <string>
 
 namespace fl {
@@ -10,13 +12,12 @@ namespace fl {
 /// One generator per request — not reusable, not thread-safe.
 /// Follows the classic pull-based iterator pattern:
 ///   while (!IsDone()) { GenerateNextToken(); text += Decode(); }
-class ChatGenerator {
+class ChatGenerator : public ICancellable {
  public:
   virtual ~ChatGenerator() = default;
 
   ChatGenerator(const ChatGenerator&) = delete;
   ChatGenerator& operator=(const ChatGenerator&) = delete;
-
   /// Returns true when generation is complete (EOS token, max_length, or stop condition).
   virtual bool IsDone() const = 0;
 
@@ -39,7 +40,7 @@ class ChatGenerator {
 
   /// Request cancellation of generation. Thread-safe — can be called from another thread.
   /// After cancellation, IsDone() should return true on the next check.
-  virtual void Cancel() = 0;
+  void Cancel() override = 0;
 
  protected:
   ChatGenerator() = default;