diff --git a/sdk_v2/cpp/include/foundry_local/foundry_local_c.h b/sdk_v2/cpp/include/foundry_local/foundry_local_c.h index 4bce94459..1c26056d0 100644 --- a/sdk_v2/cpp/include/foundry_local/foundry_local_c.h +++ b/sdk_v2/cpp/include/foundry_local/foundry_local_c.h @@ -886,7 +886,7 @@ struct flInferenceApi { /// Values are string representations; the implementation parses them for the appropriate type. /// The request copies the data — the caller may release the pairs after this call. FL_API_STATUS(Request_SetOptions, _In_ flRequest* request, _In_ const flKeyValuePairs* options); - /// Cancel an in-progress request. + /// Cancel the in-flight invocation of a request. This is a no-op while idle or after completion. FL_API_STATUS(Request_Cancel, _In_ flRequest* request); /* Response */ diff --git a/sdk_v2/cpp/include/foundry_local/foundry_local_cpp.h b/sdk_v2/cpp/include/foundry_local/foundry_local_cpp.h index 3a0e0c16c..9a2f2eeeb 100644 --- a/sdk_v2/cpp/include/foundry_local/foundry_local_cpp.h +++ b/sdk_v2/cpp/include/foundry_local/foundry_local_cpp.h @@ -1028,7 +1028,7 @@ class Request { /// Options for this request. Overrides session options for the duration of this request. Request& SetOptions(const RequestOptions& options); - /// Cancel the current request. Inferencing will stop as soon as possible. + /// Cancel this request's in-flight invocation. This is a no-op while idle or after completion. void Cancel(); const flRequest* native_handle() const noexcept { return handle_.get(); } diff --git a/sdk_v2/cpp/src/c_api.cc b/sdk_v2/cpp/src/c_api.cc index 92ebc0556..04d41b7a2 100644 --- a/sdk_v2/cpp/src/c_api.cc +++ b/sdk_v2/cpp/src/c_api.cc @@ -1727,7 +1727,7 @@ FL_API_STATUS_IMPL(Request_CancelImpl, flRequest* request) { if (!request) { return MakeStatus(FOUNDRY_LOCAL_ERROR_INVALID_ARGUMENT, "null argument"); } - AsImpl(request)->canceled = true; + AsImpl(request)->Cancel(); return nullptr; API_IMPL_END } diff --git a/sdk_v2/cpp/src/inferencing/generative/audio/audio_session.cc b/sdk_v2/cpp/src/inferencing/generative/audio/audio_session.cc index 896309417..4039cfcdf 100644 --- a/sdk_v2/cpp/src/inferencing/generative/audio/audio_session.cc +++ b/sdk_v2/cpp/src/inferencing/generative/audio/audio_session.cc @@ -240,7 +240,7 @@ void AudioSession::ProcessRequestImpl(const Request& request, Response& response int prompt_tokens = generator->PromptTokenCount(); // Token-by-token generation with optional streaming. - // Check request.canceled each iteration — a streaming callback returning + // Check request cancellation each iteration — a streaming callback returning // non-zero sets this flag asynchronously via CallbackHandler. std::vector token_texts; token_texts.reserve(kInitialTokenCapacity); @@ -248,7 +248,7 @@ void AudioSession::ProcessRequestImpl(const Request& request, Response& response std::vector> segments; segments.reserve(kInitialTokenCapacity); - while (!generator->IsDone() && !request.canceled) { + while (!generator->IsDone() && !request.IsCancellationRequested()) { generator->GenerateNextToken(); std::string token = generator->Decode(); @@ -262,7 +262,7 @@ void AudioSession::ProcessRequestImpl(const Request& request, Response& response token_texts.push_back(std::move(token)); } - if (request.canceled) { + if (request.IsCancellationRequested()) { generator->Cancel(); } } @@ -275,7 +275,7 @@ void AudioSession::ProcessRequestImpl(const Request& request, Response& response response.items.push_back(BuildSpeechResult(std::move(text), std::move(segments))); // Set finish reason - if (request.canceled) { + if (request.IsCancellationRequested()) { response.finish_reason = FOUNDRY_LOCAL_FINISH_NONE; } else { response.finish_reason = FOUNDRY_LOCAL_FINISH_STOP; @@ -358,7 +358,7 @@ void AudioSession::ProcessStreamingAudio(const AudioItem& format_item, ItemQueue } // 4. Read from queue until finished or cancelled - while (!request.canceled) { + while (!request.IsCancellationRequested()) { auto item = queue.WaitAndPop(std::chrono::milliseconds(100)); if (!item) { @@ -383,7 +383,7 @@ void AudioSession::ProcessStreamingAudio(const AudioItem& format_item, ItemQueue } // 5. Flush remaining buffered audio - if (!request.canceled) { + if (!request.IsCancellationRequested()) { auto flush_tensors = processor->Flush(); if (flush_tensors) { @@ -398,7 +398,7 @@ void AudioSession::ProcessStreamingAudio(const AudioItem& format_item, ItemQueue const size_t full_text_size = full_text.size(); response.items.push_back(BuildSpeechResult(std::move(full_text), std::move(segments))); - if (request.canceled) { + if (request.IsCancellationRequested()) { response.finish_reason = FOUNDRY_LOCAL_FINISH_NONE; } else { response.finish_reason = FOUNDRY_LOCAL_FINISH_STOP; @@ -434,7 +434,7 @@ void AudioSession::DecodeTokens(OgaGenerator& generator, OgaTokenizerStream& tok const std::unique_ptr& callback, const Request& request, int& completion_tokens) { - while (!generator.IsDone() && !generator.IsSessionTerminated() && !request.canceled) { + while (!generator.IsDone() && !generator.IsSessionTerminated() && !request.IsCancellationRequested()) { generator.GenerateNextToken(); auto next_tokens = generator.GetNextTokens(); @@ -508,7 +508,7 @@ void AudioSession::ProcessAudioTranscriptionJson(const std::string& request_json // Generate token-by-token std::string text; - while (!generator->IsDone() && !original_request.canceled) { + while (!generator->IsDone() && !original_request.IsCancellationRequested()) { generator->GenerateNextToken(); std::string token = generator->Decode(); @@ -525,7 +525,7 @@ void AudioSession::ProcessAudioTranscriptionJson(const std::string& request_json } } - if (original_request.canceled) { + if (original_request.IsCancellationRequested()) { generator->Cancel(); } } @@ -534,7 +534,7 @@ void AudioSession::ProcessAudioTranscriptionJson(const std::string& request_json int completion_tokens = total_tokens - prompt_tokens; // Set finish reason - if (original_request.canceled) { + if (original_request.IsCancellationRequested()) { response.finish_reason = FOUNDRY_LOCAL_FINISH_NONE; } else { response.finish_reason = FOUNDRY_LOCAL_FINISH_STOP; @@ -589,7 +589,8 @@ 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.IsCancellationRequested()) { generator.GenerateNextToken(); auto next_tokens = generator.GetNextTokens(); if (next_tokens.empty()) { @@ -624,7 +625,7 @@ void AudioSession::RunNemotronDecodePass(std::unique_ptr tensor const std::unique_ptr& streaming_callback, const std::string& response_id, const Request& original_request, int& completion_tokens) const { - if (!tensors || original_request.canceled) { + if (!tensors || original_request.IsCancellationRequested()) { return; } @@ -636,7 +637,7 @@ void AudioSession::RunNemotronDecodePass(std::unique_ptr tensor void AudioSession::ProcessNemotronFileTranscription(const AudioTranscriptionRequest& req, const Request& original_request, Response& response) { - if (original_request.canceled) { + if (original_request.IsCancellationRequested()) { response.finish_reason = FOUNDRY_LOCAL_FINISH_NONE; return; } @@ -673,18 +674,20 @@ 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.IsCancellationRequested(); offset += kNemotronSamplesPerChunk) { size_t count = std::min(kNemotronSamplesPerChunk, samples.size() - offset); RunNemotronDecodePass(processor->Process(samples.data() + offset, count), *generator, *tokenizer_stream, text, streaming_callback, response_id, original_request, completion_tokens); } - if (!original_request.canceled) { + if (!original_request.IsCancellationRequested()) { RunNemotronDecodePass(processor->Flush(), *generator, *tokenizer_stream, text, streaming_callback, response_id, original_request, completion_tokens); } - response.finish_reason = original_request.canceled ? FOUNDRY_LOCAL_FINISH_NONE : FOUNDRY_LOCAL_FINISH_STOP; + response.finish_reason = + original_request.IsCancellationRequested() ? FOUNDRY_LOCAL_FINISH_NONE : FOUNDRY_LOCAL_FINISH_STOP; // Nemotron file-transcription path feeds audio tensors directly and does not expose prompt token accounting. response.usage.prompt_tokens = 0; response.usage.completion_tokens = completion_tokens; diff --git a/sdk_v2/cpp/src/inferencing/generative/chat/chat_session.cc b/sdk_v2/cpp/src/inferencing/generative/chat/chat_session.cc index 7450fa289..b59432321 100644 --- a/sdk_v2/cpp/src/inferencing/generative/chat/chat_session.cc +++ b/sdk_v2/cpp/src/inferencing/generative/chat/chat_session.cc @@ -587,17 +587,12 @@ bool IsNaturalToolOutputEnd(bool canceled, return backend_termination == BackendTerminationCause::kNaturalEnd; } -flFinishReason ResolveGeneratedFinishReason(bool canceled, - bool has_tool_calls, +flFinishReason ResolveGeneratedFinishReason(bool has_tool_calls, bool stop_sequence_matched, bool host_output_limit_reached, std::optional backend_finish_reason, int completion_tokens, std::optional max_output_tokens) { - if (canceled) { - return FOUNDRY_LOCAL_FINISH_NONE; - } - if (has_tool_calls) { return FOUNDRY_LOCAL_FINISH_TOOL_CALLS; } @@ -691,7 +686,7 @@ GuidedEngineRetryResult RunGuidedEngineToolRetry( const SearchOptions& options, GenAIModelInstance& model, ToolCallContext tool_ctx, - const std::atomic& canceled, + const Request& request, bool use_full_context, const TextChatGeneratorFactory& factory) { tool_ctx.text_output = false; @@ -716,8 +711,7 @@ GuidedEngineRetryResult RunGuidedEngineToolRetry( StopStringFilter stop_filter(options.stop_sequences); auto* active_stop_filter = options.stop_sequences.empty() ? nullptr : &stop_filter; bool stop_sequence_matched = false; - while (!generator->IsDone() && - !canceled.load(std::memory_order_relaxed)) { + while (!generator->IsDone() && !request.IsCancellationRequested()) { generator->GenerateNextToken(); const auto token_id = generator->CurrentTokenId(); auto token = generator->Decode(); @@ -728,14 +722,14 @@ GuidedEngineRetryResult RunGuidedEngineToolRetry( } } - result.canceled = canceled.load(std::memory_order_relaxed); + result.canceled = request.IsCancellationRequested(); if (result.canceled) { generator->Cancel(); } chat_session_internal::FlushDecodedStream(active_stop_filter, splitter, process_segments); - if (!result.canceled && canceled.load(std::memory_order_relaxed)) { + if (!result.canceled && request.IsCancellationRequested()) { result.canceled = true; generator->Cancel(); } @@ -750,8 +744,7 @@ GuidedEngineRetryResult RunGuidedEngineToolRetry( termination = usage->termination_cause; } - const bool canceled_after_usage = - result.canceled || canceled.load(std::memory_order_relaxed); + const bool canceled_after_usage = result.canceled || request.IsCancellationRequested(); if (!result.canceled && canceled_after_usage) { generator->Cancel(); } @@ -1008,7 +1001,6 @@ ToolCallContext ChatSession::BuildToolCallContext(const Request& request, void ChatSession::ProcessGeneratedOutput(std::vector events, const ToolCallContext& tool_ctx, const SearchOptions& effective_options, - bool canceled, bool stop_sequence_matched, bool host_output_limit_reached, Response& response, @@ -1064,8 +1056,8 @@ void ChatSession::ProcessGeneratedOutput(std::vector event flush_segments(); response.finish_reason = chat_session_internal::ResolveGeneratedFinishReason( - canceled, has_tool_calls, stop_sequence_matched, host_output_limit_reached, backend_finish_reason, - completion_tokens, effective_options.max_output_tokens); + has_tool_calls, stop_sequence_matched, host_output_limit_reached, backend_finish_reason, completion_tokens, + effective_options.max_output_tokens); response.usage.prompt_tokens = prompt_tokens; response.usage.completion_tokens = completion_tokens; @@ -1259,7 +1251,7 @@ void ChatSession::ProcessRequestImpl(const Request& request, Response& response) } // Generate token-by-token with optional streaming. - // Check request.canceled each iteration — a streaming callback returning + // Check request cancellation each iteration — a streaming callback returning // non-zero sets this flag asynchronously via CallbackHandler. auto streaming_callback = CreateCallbackHandler(request); int output_tokens = 0; @@ -1371,7 +1363,7 @@ void ChatSession::ProcessRequestImpl(const Request& request, Response& response) active_raw_detector, tool_accumulator, natural_end)); }; - while (!cached_generator_->IsDone() && !request.canceled && !turn_guard.TurnEnded()) { + while (!cached_generator_->IsDone() && !request.IsCancellationRequested() && !turn_guard.TurnEnded()) { cached_generator_->GenerateNextToken(); const auto token_id = cached_generator_->CurrentTokenId(); std::string token = cached_generator_->Decode(); @@ -1401,7 +1393,7 @@ void ChatSession::ProcessRequestImpl(const Request& request, Response& response) streaming_callback->DrainPending(); } - const bool canceled_after_drain = request.canceled.load(std::memory_order_relaxed); + const bool canceled_after_drain = request.IsCancellationRequested(); if (canceled_after_drain) { cached_generator_->Cancel(); } @@ -1417,13 +1409,24 @@ void ChatSession::ProcessRequestImpl(const Request& request, Response& response) } const bool natural_tool_output_end = chat_session_internal::IsNaturalToolOutputEnd( - request.canceled, stop_sequence_matched, host_output_limit_reached, backend_termination); + request.IsCancellationRequested(), stop_sequence_matched, host_output_limit_reached, backend_termination); flush_accumulator(natural_tool_output_end); int accepted_reasoning_tokens = splitter.ReasoningTokenCount(); + if (streaming_callback) { + streaming_callback->DrainPending(); + } + + if (request.IsCancellationRequested()) { + if (!canceled_after_drain && cached_generator_) { + cached_generator_->Cancel(); + } + return; + } + bool recovered_tool_output = false; ToolCallContext completed_tool_ctx; - if (malformed_tool_output_seen && !request.canceled) { + if (malformed_tool_output_seen) { const bool recovery_eligible = IsGuidedEngineToolRetryEligible( backend_kind, natural_tool_output_end, semantic_output_seen, effective_options, cached_tool_ctx_); @@ -1443,7 +1446,7 @@ void ChatSession::ProcessRequestImpl(const Request& request, Response& response) GuidedEngineRetryResult retry; try { retry = RunGuidedEngineToolRetry( - prepared_messages, effective_options, Model(), std::move(retry_tool_ctx), request.canceled, + prepared_messages, effective_options, Model(), std::move(retry_tool_ctx), request, /*use_full_context=*/true, text_generator_factory_); } catch (...) { if (streaming_callback) { @@ -1483,39 +1486,34 @@ void ChatSession::ProcessRequestImpl(const Request& request, Response& response) recovered_tool_output = true; } - if (request.canceled && !canceled_after_drain) { - if (cached_generator_) { - cached_generator_->Cancel(); - } - } else if (cached_generator_ && (stop_sequence_matched || host_output_limit_reached) && - !cached_generator_->IsDone()) { + if (cached_generator_ && (stop_sequence_matched || host_output_limit_reached) && + !cached_generator_->IsDone()) { cached_generator_->Cancel(); } if (streaming_callback) { - streaming_callback->Drain(); + streaming_callback->DrainPending(); + } + + if (request.IsCancellationRequested()) { + if (cached_generator_) { + cached_generator_->Cancel(); + } + return; } const auto& output_tool_ctx = recovered_tool_output ? completed_tool_ctx : cached_tool_ctx_; auto assistant_message = MakeAssistantMessage(generated_events, output_tool_ctx, logger_); const bool generated_tool_calls = assistant_message.HasToolCalls(); - if (!request.canceled) { - // Reject a generation whose calls cannot be correlated before it reaches the caller — a committed turn must never - // leave the outstanding-call set inconsistent. - transcript_.ValidateGeneratedOutput(assistant_message); - } + // Reject a generation whose calls cannot be correlated before it reaches the caller — a committed turn must never + // leave the outstanding-call set inconsistent. + transcript_.ValidateGeneratedOutput(assistant_message); - ProcessGeneratedOutput(std::move(generated_events), output_tool_ctx, effective_options, request.canceled, + ProcessGeneratedOutput(std::move(generated_events), output_tool_ctx, effective_options, stop_sequence_matched, host_output_limit_reached, response, prompt_tokens, total_tokens, accepted_reasoning_tokens, backend_finish_reason); - if (request.canceled) { - // Cancel is permanent for classic generators, and Engine cannot rewind. The scope guard therefore discards every - // canceled generator so the next request rebuilds from the committed transcript. - return; - } - // LARK grammar (tool-call-only mode) is a single-shot finite parse. If generation was truncated while grammar was // active, the parser is in an unrecoverable state. Additionally, a completed grammar signals EOS — IsDone() would // return true on the next turn. Invalidate after any grammar-guided generation so the next turn rebuilds. @@ -1542,6 +1540,10 @@ void ChatSession::ProcessRequestImpl(const Request& request, Response& response) // The reply may only merge into an input message from the last replay segment — this request's own input. Merging // into an earlier hop's assistant message would glue two recorded turns together. + if (!request.TryComplete()) { + return; + } + transcript_.CommitTurn(std::move(inputs), std::move(assistant_message), {pre_turn_token_count, total_tokens}, ingest.last_segment_start); turn_committed = true; @@ -1774,7 +1776,7 @@ void ChatSession::ProcessChatCompletionsJson(const std::string& request_json, co StopStringFilter stop_filter(options.stop_sequences); auto* active_stop_filter = options.stop_sequences.empty() ? nullptr : &stop_filter; bool stop_sequence_matched = false; - while (!generator->IsDone() && !original_request.canceled && !turn_guard.TurnEnded()) { + while (!generator->IsDone() && !original_request.IsCancellationRequested() && !turn_guard.TurnEnded()) { generator->GenerateNextToken(); const auto token_id = generator->CurrentTokenId(); std::string token = generator->Decode(); @@ -1796,8 +1798,7 @@ void ChatSession::ProcessChatCompletionsJson(const std::string& request_json, co streaming_callback->DrainPending(); } - const bool canceled_after_drain = - original_request.canceled.load(std::memory_order_relaxed); + const bool canceled_after_drain = original_request.IsCancellationRequested(); if (canceled_after_drain) { generator->Cancel(); } @@ -1813,13 +1814,24 @@ void ChatSession::ProcessChatCompletionsJson(const std::string& request_json, co } const bool natural_tool_output_end = chat_session_internal::IsNaturalToolOutputEnd( - original_request.canceled, stop_sequence_matched, /*host_output_limit_reached=*/false, + original_request.IsCancellationRequested(), stop_sequence_matched, /*host_output_limit_reached=*/false, backend_termination); process_tool_output(chat_session_internal::FlushToolOutput( active_raw_detector, tool_accumulator, natural_tool_output_end)); int accepted_reasoning_tokens = splitter.ReasoningTokenCount(); - if (malformed_tool_output_seen && !original_request.canceled) { + if (streaming_callback) { + streaming_callback->DrainPending(); + } + + if (original_request.IsCancellationRequested()) { + if (!canceled_after_drain) { + generator->Cancel(); + } + return; + } + + if (malformed_tool_output_seen) { const bool recovery_eligible = IsGuidedEngineToolRetryEligible( Model().GetGenAIConfig().GetChatBackendKind(), natural_tool_output_end, semantic_output_seen, options, tool_ctx); @@ -1836,7 +1848,7 @@ void ChatSession::ProcessChatCompletionsJson(const std::string& request_json, co GuidedEngineRetryResult retry; try { retry = RunGuidedEngineToolRetry( - prepared_messages, options, Model(), tool_ctx, original_request.canceled, + prepared_messages, options, Model(), tool_ctx, original_request, /*use_full_context=*/false, text_generator_factory_); } catch (...) { if (streaming_callback) { @@ -1872,11 +1884,7 @@ void ChatSession::ProcessChatCompletionsJson(const std::string& request_json, co stop_sequence_matched = false; } - if (original_request.canceled && !canceled_after_drain) { - if (generator) { - generator->Cancel(); - } - } else if (generator && stop_sequence_matched && !generator->IsDone()) { + if (generator && stop_sequence_matched && !generator->IsDone()) { generator->Cancel(); } @@ -1884,11 +1892,18 @@ void ChatSession::ProcessChatCompletionsJson(const std::string& request_json, co streaming_callback->DrainPending(); } - ProcessGeneratedOutput(std::move(generated_events), tool_ctx, options, original_request.canceled, + if (original_request.IsCancellationRequested()) { + if (generator) { + generator->Cancel(); + } + return; + } + + ProcessGeneratedOutput(std::move(generated_events), tool_ctx, options, stop_sequence_matched, /*host_output_limit_reached=*/false, response, prompt_tokens, total_tokens, accepted_reasoning_tokens, backend_finish_reason); - if (original_request.canceled) { + if (!original_request.TryComplete()) { return; } diff --git a/sdk_v2/cpp/src/inferencing/generative/chat/chat_session.h b/sdk_v2/cpp/src/inferencing/generative/chat/chat_session.h index 5b2050b61..e56d8e265 100644 --- a/sdk_v2/cpp/src/inferencing/generative/chat/chat_session.h +++ b/sdk_v2/cpp/src/inferencing/generative/chat/chat_session.h @@ -99,8 +99,7 @@ void FlushDecodedStream(StopStringFilter* stop_filter, process_segments(splitter.Flush()); } -flFinishReason ResolveGeneratedFinishReason(bool canceled, - bool has_tool_calls, +flFinishReason ResolveGeneratedFinishReason(bool has_tool_calls, bool stop_sequence_matched, bool host_output_limit_reached, std::optional backend_finish_reason, @@ -214,7 +213,6 @@ class ChatSession : public Session { void ProcessGeneratedOutput(std::vector events, const ToolCallContext& tool_ctx, const SearchOptions& effective_options, - bool canceled, bool stop_sequence_matched, bool host_output_limit_reached, Response& response, diff --git a/sdk_v2/cpp/src/inferencing/generative/chat/onnx_engine_chat_stream.cc b/sdk_v2/cpp/src/inferencing/generative/chat/onnx_engine_chat_stream.cc index 28a9a44e4..65e1e2367 100644 --- a/sdk_v2/cpp/src/inferencing/generative/chat/onnx_engine_chat_stream.cc +++ b/sdk_v2/cpp/src/inferencing/generative/chat/onnx_engine_chat_stream.cc @@ -23,10 +23,10 @@ std::optional MapFinishReason(OgaFinishReason reason) { case OgaFinishReason_MaxGeneratedTokens: case OgaFinishReason_MaxSessionTokens: return FOUNDRY_LOCAL_FINISH_LENGTH; - case OgaFinishReason_Cancelled: - return FOUNDRY_LOCAL_FINISH_NONE; case OgaFinishReason_Failed: return FOUNDRY_LOCAL_FINISH_ERROR; + case OgaFinishReason_Cancelled: + return std::nullopt; default: return std::nullopt; } diff --git a/sdk_v2/cpp/src/inferencing/session/callback_handler.h b/sdk_v2/cpp/src/inferencing/session/callback_handler.h index 1d73b2237..8ca47634b 100644 --- a/sdk_v2/cpp/src/inferencing/session/callback_handler.h +++ b/sdk_v2/cpp/src/inferencing/session/callback_handler.h @@ -28,10 +28,14 @@ namespace fl { /// user callback. This decouples token generation from callback speed /// while guaranteeing delivery order. /// +/// On cancellation, a backlog within the normal 64-item backpressure window is delivered. A larger backlog is +/// discarded so cancellation cannot be delayed indefinitely by a stalled consumer. +/// /// The Request reference is bound at construction — no need to pass it per push. /// Destruction drains the queue and joins the worker thread (RAII). struct CallbackHandler { using CallbackFn = std::function; + static constexpr size_t kMaxCancellationDrainItems = 64; CallbackHandler(const Request& request, CallbackFn callback_fn, ILogger& logger, void* user_data = nullptr) @@ -56,7 +60,8 @@ struct CallbackHandler { /// Push an item into the queue and wake the worker. /// Called from the generator thread — returns immediately. void PushItem(std::unique_ptr item) { - if (request_.canceled) { + std::lock_guard lock(callback_mutex_); + if (disabled_after_exception_ || request_.IsCancellationRequested()) { return; } @@ -90,24 +95,27 @@ struct CallbackHandler { // Fire the callback for each available item. // The callback pops from the queue — that is the established contract. while (queue_->Size() > 0) { + if (request_.IsCancellationRequested() && queue_->Size() > kMaxCancellationDrainItems) { + DropPendingItemsAndNotify(false); + break; + } + SetCallbackInProgress(true); try { if (fn_(data_, user_data_) != 0) { - request_.canceled = true; + request_.Cancel(Request::CancellationReason::StreamingCallback); } } catch (const std::exception& e) { logger_.Log(LogLevel::Warning, fmt::format("streaming callback threw an exception; cancelling request: {}", e.what())); - DisableAfterException(); - SetCallbackInProgress(false); + DisableAfterException(e.what()); return; } catch (...) { logger_.Log(LogLevel::Warning, "streaming callback threw a non-std exception; cancelling request"); - DisableAfterException(); - SetCallbackInProgress(false); + DisableAfterException("non-standard exception"); return; } @@ -124,10 +132,21 @@ struct CallbackHandler { /// Called from the worker thread after the user callback throws. Marks the request /// cancelled (so PushItem becomes a no-op and the generator loop stops feeding work) /// and drops any items still queued so the destructor can join cleanly. - void DisableAfterException() { - request_.canceled = true; - while (queue_->TryPop()) { + void DisableAfterException(std::string_view detail) { + request_.CancelFromStreamingCallbackException(detail); + DropPendingItemsAndNotify(true); + } + + void DropPendingItemsAndNotify(bool disable_after_exception) { + { + std::lock_guard lock(callback_mutex_); + disabled_after_exception_ = disabled_after_exception_ || disable_after_exception; + while (queue_->TryPop()) { + } + callback_in_progress_ = false; } + + callback_cv_.notify_all(); } void SetCallbackInProgress(bool value) { @@ -151,6 +170,7 @@ struct CallbackHandler { std::mutex callback_mutex_; std::condition_variable callback_cv_; bool callback_in_progress_ = false; + bool disabled_after_exception_ = false; std::thread worker_; }; diff --git a/sdk_v2/cpp/src/inferencing/session/request.h b/sdk_v2/cpp/src/inferencing/session/request.h index 0ae4f3e07..c1011e370 100644 --- a/sdk_v2/cpp/src/inferencing/session/request.h +++ b/sdk_v2/cpp/src/inferencing/session/request.h @@ -8,9 +8,12 @@ #include #include +#include #include +#include #include #include +#include #include namespace fl { @@ -26,6 +29,25 @@ inline constexpr const char* kSystemPromptOption = "system_prompt"; /// Generic inference request — pure input data. /// Items are stored as borrowed pointers. Owned items are kept alive in owned_items. struct Request { + enum class CancellationReason : uint8_t { + None, + Caller, + StreamingCallback, + StreamingCallbackException, + SessionShutdown, + }; + + enum class State : uint8_t { + Ready, + Running, + CanceledByCaller, + CanceledByStreamingCallback, + CanceledByStreamingCallbackException, + CanceledBySessionShutdown, + Completing, + Completed, + }; + std::vector items; // all items (borrowed pointers) KeyValuePairs options; /// Request-local definitions already validated by an HTTP adapter. Used only to carry Chat @@ -49,12 +71,6 @@ struct Request { /// ascending and consistent with `items`. std::vector item_segment_starts; - /// Cancellation flag — set by the C API or streaming callback handler to cancel - /// an in-flight request. Checked in generation loops. Atomic because it is written - /// by one thread (callback worker or C API) and read by another (generator loop). - /// Uses relaxed ordering since it is a one-way flag and exact timing doesn't matter. - mutable std::atomic canceled{false}; - Request() = default; Request(Request&& other) noexcept @@ -64,7 +80,8 @@ struct Request { forced_tool_choice(std::move(other.forced_tool_choice)), raw_envelope_descriptor(std::move(other.raw_envelope_descriptor)), item_segment_starts(std::move(other.item_segment_starts)), - canceled(other.canceled.load(std::memory_order_relaxed)), + state_(other.state_.load(std::memory_order_relaxed)), + cancellation_detail_(std::move(other.cancellation_detail_)), owned_items(std::move(other.owned_items)) {} Request& operator=(Request&& other) noexcept { @@ -74,7 +91,8 @@ struct Request { forced_tool_choice = std::move(other.forced_tool_choice); raw_envelope_descriptor = std::move(other.raw_envelope_descriptor); item_segment_starts = std::move(other.item_segment_starts); - canceled.store(other.canceled.load(std::memory_order_relaxed), std::memory_order_relaxed); + state_.store(other.state_.load(std::memory_order_relaxed), std::memory_order_relaxed); + cancellation_detail_ = std::move(other.cancellation_detail_); owned_items = std::move(other.owned_items); return *this; } @@ -99,7 +117,134 @@ struct Request { item_segment_starts.push_back(items.size()); } + /// Cancels only the invocation currently being processed. Idle and completed requests are unchanged. + bool Cancel(CancellationReason reason = CancellationReason::Caller) const noexcept { + const auto canceled_state = CanceledState(reason); + auto expected = State::Running; + if (state_.compare_exchange_strong(expected, canceled_state, + std::memory_order_acq_rel, + std::memory_order_acquire)) { + return true; + } + + return IsCanceledState(expected); + } + + bool CancelFromStreamingCallbackException(std::string_view detail) const noexcept { + try { + std::lock_guard lock(cancellation_detail_mutex_); + auto expected = state_.load(std::memory_order_acquire); + if (expected != State::Running) { + return IsCanceledState(expected); + } + + cancellation_detail_ = detail; + if (state_.compare_exchange_strong(expected, State::CanceledByStreamingCallbackException, + std::memory_order_acq_rel, + std::memory_order_acquire)) { + return true; + } + + return IsCanceledState(expected); + } catch (...) { + // Cancellation itself must remain reliable if preserving diagnostic text runs out of memory. + } + + return Cancel(CancellationReason::StreamingCallbackException); + } + + bool IsCancellationRequested() const noexcept { + return IsCanceledState(state_.load(std::memory_order_acquire)); + } + + CancellationReason GetCancellationReason() const noexcept { + return ReasonFromState(state_.load(std::memory_order_acquire)); + } + + std::string CancellationDetail() const { + std::lock_guard lock(cancellation_detail_mutex_); + return cancellation_detail_; + } + + /// Starts first-time processing or reuses a request whose previous operation completed. + bool TryBegin() const { + std::lock_guard lock(cancellation_detail_mutex_); + auto state = state_.load(std::memory_order_acquire); + while (state == State::Ready || state == State::Completed) { + cancellation_detail_.clear(); + if (state_.compare_exchange_weak(state, State::Running, + std::memory_order_acq_rel, + std::memory_order_acquire)) { + return true; + } + } + + return false; + } + + /// Atomically claims the normal or error terminal boundary. Returns false when cancellation won first. + bool TryComplete() const noexcept { + auto expected = State::Running; + if (state_.compare_exchange_strong(expected, State::Completing, + std::memory_order_acq_rel, + std::memory_order_acquire)) { + return true; + } + + return expected == State::Completing; + } + + /// Makes a claimed completion reusable after the outer request call has finished publishing its result. + void PublishCompletion() const noexcept { + state_.store(State::Completed, std::memory_order_release); + } + + bool IsCompleted() const noexcept { + return state_.load(std::memory_order_acquire) == State::Completed; + } + private: + static State CanceledState(CancellationReason reason) noexcept { + switch (reason) { + case CancellationReason::StreamingCallback: + return State::CanceledByStreamingCallback; + case CancellationReason::StreamingCallbackException: + return State::CanceledByStreamingCallbackException; + case CancellationReason::SessionShutdown: + return State::CanceledBySessionShutdown; + case CancellationReason::None: + // Cancel() always means cancellation; defensively classify a missing or unknown reason as caller-initiated. + case CancellationReason::Caller: + default: + return State::CanceledByCaller; + } + } + + static bool IsCanceledState(State state) noexcept { + return state == State::CanceledByCaller || + state == State::CanceledByStreamingCallback || + state == State::CanceledByStreamingCallbackException || + state == State::CanceledBySessionShutdown; + } + + static CancellationReason ReasonFromState(State state) noexcept { + switch (state) { + case State::CanceledByCaller: + return CancellationReason::Caller; + case State::CanceledByStreamingCallback: + return CancellationReason::StreamingCallback; + case State::CanceledByStreamingCallbackException: + return CancellationReason::StreamingCallbackException; + case State::CanceledBySessionShutdown: + return CancellationReason::SessionShutdown; + default: + return CancellationReason::None; + } + } + + mutable std::atomic state_{State::Ready}; + mutable std::mutex cancellation_detail_mutex_; + mutable std::string cancellation_detail_; std::vector> owned_items; // owned items (lifetime) }; diff --git a/sdk_v2/cpp/src/inferencing/session/session.cc b/sdk_v2/cpp/src/inferencing/session/session.cc index d2f34bb51..8d686ed4e 100644 --- a/sdk_v2/cpp/src/inferencing/session/session.cc +++ b/sdk_v2/cpp/src/inferencing/session/session.cc @@ -13,6 +13,7 @@ #include "model.h" #include "telemetry/telemetry.h" #include "telemetry/telemetry_action_tracker.h" +#include "util/scope_guard.h" #include "utils.h" #include @@ -22,6 +23,29 @@ namespace fl { +namespace { + +[[noreturn]] void ThrowCancellation(const Request& request) { + switch (request.GetCancellationReason()) { + case Request::CancellationReason::StreamingCallback: + FL_THROW(FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED, "request cancelled by streaming callback"); + case Request::CancellationReason::StreamingCallbackException: { + const auto detail = request.CancellationDetail(); + FL_THROW(FOUNDRY_LOCAL_ERROR_INTERNAL, + detail.empty() ? "streaming callback threw an exception" + : fmt::format("streaming callback threw an exception: {}", detail)); + } + case Request::CancellationReason::SessionShutdown: + FL_THROW(FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED, "request cancelled because the session is shutting down"); + case Request::CancellationReason::None: + case Request::CancellationReason::Caller: + default: + FL_THROW(FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED, "request cancelled by caller"); + } +} + +} // namespace + Session::Session(const fl::Model& catalog_model, ILogger& logger, ITelemetry& telemetry, bool allow_concurrent_requests) : catalog_model_(catalog_model), @@ -145,50 +169,103 @@ void Session::ProcessRequest(const Request& request, Response& response) { lock.lock(); } + bool admitted = false; + bool registered = false; + auto finish_lifecycle = [&]() noexcept { + if (!admitted) { + return; + } + + if (registered) { + std::lock_guard active_lock(*active_requests_mutex_); + active_requests_.erase(&request); + registered = false; + } + + request.PublishCompletion(); + admitted = false; + }; + ScopeGuard lifecycle_guard([&]() noexcept { finish_lifecycle(); }); + { std::lock_guard active_lock(*active_requests_mutex_); - active_requests_.insert(&request); + if (!request.TryBegin()) { + FL_THROW(FOUNDRY_LOCAL_ERROR_INVALID_USAGE, "request is already being processed"); + } + admitted = true; - // If Cancel() already ran (shutdown began before this request was admitted), stamp it now so the - // generation loop exits at its first poll instead of running an uncanceled turn. + // A late shutdown admission must first claim the request, then be canceled under the same lock that protects + // active registration. This keeps TryBegin as the sole admission gate while preventing backend entry. if (session_canceled_) { - request.canceled.store(true, std::memory_order_relaxed); + request.Cancel(Request::CancellationReason::SessionShutdown); } - } - // RAII: deregister the request even if ProcessRequestImpl throws, so Cancel() never - // dereferences a dangling Request after this call unwinds. - struct ActiveRequestGuard { - Session& session; - const Request& request; - ~ActiveRequestGuard() { - std::lock_guard active_lock(*session.active_requests_mutex_); - session.active_requests_.erase(&request); - } - } active_guard{*this, request}; + active_requests_.insert(&request); + registered = true; + } ActionTracker tracker(Action::kSessionProcessRequest, telemetry_); tracker.SetModelId(CatalogModel().Id()); + Response staged_response; try { + if (request.IsCancellationRequested()) { + ThrowCancellation(request); + } + ValidateRequestItems(request); - ProcessRequestImpl(request, response); + ProcessRequestImpl(request, staged_response); + if (!request.TryComplete()) { + ThrowCancellation(request); + } + + response = std::move(staged_response); + finish_lifecycle(); + lifecycle_guard.Dismiss(); tracker.SetStatus(ActionStatus::kSuccess); } catch (const std::exception& ex) { - tracker.RecordException(ex); - throw; + if (request.TryComplete()) { + finish_lifecycle(); + lifecycle_guard.Dismiss(); + tracker.RecordException(ex); + throw; + } + + try { + ThrowCancellation(request); + } catch (const std::exception& cancellation) { + finish_lifecycle(); + lifecycle_guard.Dismiss(); + tracker.RecordException(cancellation); + throw; + } + } catch (...) { + if (request.TryComplete()) { + finish_lifecycle(); + lifecycle_guard.Dismiss(); + throw; + } + + try { + ThrowCancellation(request); + } catch (const std::exception& cancellation) { + finish_lifecycle(); + lifecycle_guard.Dismiss(); + tracker.RecordException(cancellation); + throw; + } } } void Session::Cancel() { - // Only flip cancel flags — never block or join — so this is safe to call while the + // Only update request lifecycle state — never block or join — so this is safe to call while the // SessionManager holds its own lock during shutdown. Generation loops poll the flag. std::lock_guard lock(*active_requests_mutex_); session_canceled_ = true; for (const Request* r : active_requests_) { - r->canceled.store(true, std::memory_order_relaxed); + r->Cancel(Request::CancellationReason::SessionShutdown); } } diff --git a/sdk_v2/cpp/src/inferencing/session/session.h b/sdk_v2/cpp/src/inferencing/session/session.h index f2db93c5c..cbeea7dfc 100644 --- a/sdk_v2/cpp/src/inferencing/session/session.h +++ b/sdk_v2/cpp/src/inferencing/session/session.h @@ -54,9 +54,9 @@ class Session { /// in-flight callbacks and ensures the Response is fully populated on return. void ProcessRequest(const Request& request, Response& response); - /// Signal every in-flight request on this session to cancel. Only sets each request's atomic - /// cancel flag — never blocks and never joins — so it is safe to call from a shutdown path while - /// another thread holds a manager lock. Generation loops poll the flag and stop within ~one token. + /// Signal every in-flight request on this session to cancel. Only updates each request's atomic + /// lifecycle — never blocks and never joins — so it is safe to call from a shutdown path while + /// another thread holds a manager lock. Generation loops poll the request between backend scheduling quanta. void Cancel(); /// Add a tool definition to this session. Names are case-sensitive and unique across kinds. @@ -167,9 +167,9 @@ class Session { const bool allow_concurrent_requests_; mutable std::unique_ptr request_mutex_ = std::make_unique(); - // In-flight requests tracked so Cancel() can flip their cancel flags from another thread. Guarded - // by its own mutex (not request_mutex_) because concurrent sessions (e.g. audio) may hold several - // at once, and Cancel() must run without waiting on an active generation holding request_mutex_. + // In-flight requests tracked so Cancel() can transition their lifecycle state from another thread. Guarded by its + // own mutex (not request_mutex_) because concurrent sessions (e.g. audio) may hold several at once, and Cancel() + // must run without waiting on an active generation holding request_mutex_. // unique_ptr keeps Session movable (std::mutex is not movable), matching request_mutex_. std::unordered_set active_requests_; mutable std::unique_ptr active_requests_mutex_ = std::make_unique(); diff --git a/sdk_v2/cpp/src/inferencing/session/session_manager.cc b/sdk_v2/cpp/src/inferencing/session/session_manager.cc index 3e35eb45e..4d3e58ee9 100644 --- a/sdk_v2/cpp/src/inferencing/session/session_manager.cc +++ b/sdk_v2/cpp/src/inferencing/session/session_manager.cc @@ -68,7 +68,7 @@ void SessionManager::CancelAll() { logger_.Log(LogLevel::Information, fmt::format("SessionManager: cancelling all sessions ({} active)", sessions_.size())); - // Signal every in-flight request to stop. Cancel() only sets atomic flags — no joins, no + // Signal every in-flight request to stop. Cancel() only updates atomic lifecycle state — no joins, no // re-entrancy into SessionManager — so calling it while holding mutex_ cannot deadlock. for (Session* s : sessions_) { s->Cancel(); diff --git a/sdk_v2/cpp/src/manager.cc b/sdk_v2/cpp/src/manager.cc index af189b641..1a9304717 100644 --- a/sdk_v2/cpp/src/manager.cc +++ b/sdk_v2/cpp/src/manager.cc @@ -542,7 +542,7 @@ void Manager::Shutdown() { // Order matters: // 1. Reject new loads so callers gated on IsShutdownRequested can stop early. // 2. Cancel in-flight generations BEFORE stopping the web service. StopWebService() hard-joins - // streaming threads; a generation grinding in ORT GenAI only stops when request.canceled is + // streaming threads; a generation grinding in ORT GenAI only stops when request cancellation is // set, so cancelling first is what lets JoinAll() return promptly instead of deadlocking // process shutdown. // 3. Stop the web service (JoinAll now unblocks), then drain HTTP-tracked sessions. diff --git a/sdk_v2/cpp/test/internal_api/c_api_test.cc b/sdk_v2/cpp/test/internal_api/c_api_test.cc index 150266bb9..11836f669 100644 --- a/sdk_v2/cpp/test/internal_api/c_api_test.cc +++ b/sdk_v2/cpp/test/internal_api/c_api_test.cc @@ -1712,7 +1712,7 @@ TEST(CApiTest, ItemQueuePushPopAndFinish) { // Inference API — Request_Cancel // ======================================================================== -TEST(CApiTest, RequestCancelOnIdleRequest) { +TEST(CApiTest, RequestCancelWhileIdleSucceeds) { const flApi* api = GetApi(); const flInferenceApi* inf_api = api->GetInferenceApi(); @@ -1720,7 +1720,7 @@ TEST(CApiTest, RequestCancelOnIdleRequest) { ASSERT_TRUE(IsOk(inf_api->Request_Create(&req))); ASSERT_NE(req, nullptr); - // Cancel on an idle request should succeed (no-op) + // Idle cancellation is a successful no-op. EXPECT_TRUE(IsOk(inf_api->Request_Cancel(req))); inf_api->Request_Release(req); diff --git a/sdk_v2/cpp/test/internal_api/callback_handler_test.cc b/sdk_v2/cpp/test/internal_api/callback_handler_test.cc index 872cdb45b..566fa276f 100644 --- a/sdk_v2/cpp/test/internal_api/callback_handler_test.cc +++ b/sdk_v2/cpp/test/internal_api/callback_handler_test.cc @@ -9,6 +9,8 @@ #include #include +#include +#include #include #include @@ -31,6 +33,7 @@ CallbackHandler::CallbackFn MakeThrowingCallback(std::atomic& invocations) TEST(CallbackHandlerTest, StdExceptionFromCallbackDoesNotTerminate) { Request request; + ASSERT_TRUE(request.TryBegin()); std::atomic invocations{0}; { @@ -42,11 +45,14 @@ TEST(CallbackHandlerTest, StdExceptionFromCallbackDoesNotTerminate) { } EXPECT_GE(invocations.load(), 1); - EXPECT_TRUE(request.canceled.load()); + EXPECT_TRUE(request.IsCancellationRequested()); + EXPECT_EQ(request.GetCancellationReason(), Request::CancellationReason::StreamingCallbackException); + EXPECT_EQ(request.CancellationDetail(), "boom"); } TEST(CallbackHandlerTest, NonStdExceptionFromCallbackDoesNotTerminate) { Request request; + ASSERT_TRUE(request.TryBegin()); std::atomic invocations{0}; auto fn = [&invocations](flStreamingCallbackData, void*) -> int { @@ -61,29 +67,33 @@ TEST(CallbackHandlerTest, NonStdExceptionFromCallbackDoesNotTerminate) { } EXPECT_GE(invocations.load(), 1); - EXPECT_TRUE(request.canceled.load()); + EXPECT_TRUE(request.IsCancellationRequested()); + EXPECT_EQ(request.GetCancellationReason(), Request::CancellationReason::StreamingCallbackException); + EXPECT_EQ(request.CancellationDetail(), "non-standard exception"); } TEST(CallbackHandlerTest, FurtherPushesAfterExceptionAreNoOps) { Request request; + ASSERT_TRUE(request.TryBegin()); std::atomic invocations{0}; CallbackHandler handler(request, MakeThrowingCallback(invocations), fl::test::NullLog()); handler.PushItem(std::make_unique("first")); // Wait until the worker has cancelled the request after catching the throw. - for (int i = 0; i < 200 && !request.canceled.load(); ++i) { + for (int i = 0; i < 200 && !request.IsCancellationRequested(); ++i) { std::this_thread::sleep_for(std::chrono::milliseconds(5)); } - ASSERT_TRUE(request.canceled.load()); + ASSERT_TRUE(request.IsCancellationRequested()); const int invocations_after_first = invocations.load(); - // Subsequent pushes must be dropped (canceled is set, so PushItem skips). + // Subsequent pushes must be dropped once the request's cancellation state is set. handler.PushItem(std::make_unique("second")); handler.PushItem(std::make_unique("third")); + handler.DrainPending(); handler.Drain(); // Worker exited after the throw — no further callback invocations. @@ -92,6 +102,7 @@ TEST(CallbackHandlerTest, FurtherPushesAfterExceptionAreNoOps) { TEST(CallbackHandlerTest, NormalCallbackCancelsViaReturnValue) { Request request; + ASSERT_TRUE(request.TryBegin()); std::atomic invocations{0}; auto fn = [&invocations, &request](flStreamingCallbackData data, void*) -> int { @@ -109,11 +120,13 @@ TEST(CallbackHandlerTest, NormalCallbackCancelsViaReturnValue) { handler.Drain(); EXPECT_EQ(invocations.load(), 1); - EXPECT_TRUE(request.canceled.load()); + EXPECT_TRUE(request.IsCancellationRequested()); + EXPECT_EQ(request.GetCancellationReason(), Request::CancellationReason::StreamingCallback); } TEST(CallbackHandlerTest, DrainPendingWaitsForDeliveryWithoutClosingTheQueue) { Request request; + ASSERT_TRUE(request.TryBegin()); std::atomic invocations{0}; auto fn = [&invocations](flStreamingCallbackData data, void*) -> int { @@ -132,3 +145,92 @@ TEST(CallbackHandlerTest, DrainPendingWaitsForDeliveryWithoutClosingTheQueue) { handler.Drain(); EXPECT_EQ(invocations.load(), 2); } + +TEST(CallbackHandlerTest, CancellationDrainsSmallBufferedBacklog) { + Request request; + ASSERT_TRUE(request.TryBegin()); + std::atomic invocations{0}; + std::mutex mutex; + std::condition_variable cv; + bool callback_started = false; + bool release_callback = false; + + auto fn = [&](flStreamingCallbackData data, void*) -> int { + auto* queue = reinterpret_cast(data.item_queue); + (void)queue->TryPop(); + ++invocations; + + std::unique_lock lock(mutex); + callback_started = true; + cv.notify_all(); + cv.wait(lock, [&] { return release_callback; }); + return 0; + }; + + CallbackHandler handler(request, fn, fl::test::NullLog()); + handler.PushItem(std::make_unique("first")); + bool started = false; + { + std::unique_lock lock(mutex); + started = cv.wait_for(lock, std::chrono::seconds(2), [&] { return callback_started; }); + } + EXPECT_TRUE(started); + + for (size_t i = 0; i < 3; ++i) { + handler.PushItem(std::make_unique("buffered")); + } + request.Cancel(); + { + std::lock_guard lock(mutex); + release_callback = true; + } + cv.notify_all(); + + handler.Drain(); + EXPECT_EQ(invocations.load(), 4); +} + +TEST(CallbackHandlerTest, CancellationDropsLargeBufferedBacklog) { + Request request; + ASSERT_TRUE(request.TryBegin()); + std::atomic invocations{0}; + std::mutex mutex; + std::condition_variable cv; + bool callback_started = false; + bool release_callback = false; + + auto fn = [&](flStreamingCallbackData data, void*) -> int { + auto* queue = reinterpret_cast(data.item_queue); + (void)queue->TryPop(); + ++invocations; + + std::unique_lock lock(mutex); + callback_started = true; + cv.notify_all(); + cv.wait(lock, [&] { return release_callback; }); + return 0; + }; + + CallbackHandler handler(request, fn, fl::test::NullLog()); + handler.PushItem(std::make_unique("first")); + bool started = false; + { + std::unique_lock lock(mutex); + started = cv.wait_for(lock, std::chrono::seconds(2), [&] { return callback_started; }); + } + EXPECT_TRUE(started); + + for (size_t i = 0; i <= CallbackHandler::kMaxCancellationDrainItems; ++i) { + handler.PushItem(std::make_unique("buffered")); + } + request.Cancel(); + { + std::lock_guard lock(mutex); + release_callback = true; + } + cv.notify_all(); + + handler.DrainPending(); + handler.Drain(); + EXPECT_EQ(invocations.load(), 1); +} diff --git a/sdk_v2/cpp/test/internal_api/chat/chat_session_test.cc b/sdk_v2/cpp/test/internal_api/chat/chat_session_test.cc index 83dd5a506..60c6bf5f6 100644 --- a/sdk_v2/cpp/test/internal_api/chat/chat_session_test.cc +++ b/sdk_v2/cpp/test/internal_api/chat/chat_session_test.cc @@ -235,6 +235,16 @@ void AppendSegments(std::vector& destination, const std::vector +void ExpectOperationCancelled(Action&& action) { + try { + action(); + FAIL() << "Expected operation cancellation"; + } catch (const fl::Exception& error) { + EXPECT_EQ(error.code(), FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED); + } +} + std::vector RunToolOutput( const std::vector& chunks, const std::string& tools = {}, const std::unordered_map& kinds = {}) { @@ -1068,31 +1078,28 @@ TEST(ChatSessionDecisionTest, FinishReasonPrecedenceCoversEveryTerminalSource) { struct TestCase { const char* name; - bool canceled; bool has_tool_calls; bool stop_sequence_matched; bool host_output_limit_reached; std::optional backend_finish_reason; flFinishReason expected; }; - const std::vector cases = { - {"cancellation wins", true, true, true, true, FOUNDRY_LOCAL_FINISH_NONE, FOUNDRY_LOCAL_FINISH_NONE}, - {"tool calls win over stop", false, true, true, false, FOUNDRY_LOCAL_FINISH_STOP, + {"tool calls win over stop", true, true, false, FOUNDRY_LOCAL_FINISH_STOP, FOUNDRY_LOCAL_FINISH_TOOL_CALLS}, - {"tool calls win over host limit", false, true, true, true, FOUNDRY_LOCAL_FINISH_NONE, + {"tool calls win over host limit", true, true, true, FOUNDRY_LOCAL_FINISH_NONE, FOUNDRY_LOCAL_FINISH_TOOL_CALLS}, - {"stop wins over host limit", false, false, true, true, FOUNDRY_LOCAL_FINISH_NONE, + {"stop wins over host limit", false, true, true, FOUNDRY_LOCAL_FINISH_NONE, FOUNDRY_LOCAL_FINISH_STOP}, - {"host limit produces length", false, false, false, true, FOUNDRY_LOCAL_FINISH_NONE, + {"host limit produces length", false, false, true, FOUNDRY_LOCAL_FINISH_NONE, FOUNDRY_LOCAL_FINISH_LENGTH}, - {"backend reason survives natural completion", false, false, false, false, FOUNDRY_LOCAL_FINISH_STOP, + {"backend reason survives natural completion", false, false, false, FOUNDRY_LOCAL_FINISH_STOP, FOUNDRY_LOCAL_FINISH_STOP}, }; for (const auto& test : cases) { SCOPED_TRACE(test.name); - EXPECT_EQ(ResolveGeneratedFinishReason(test.canceled, test.has_tool_calls, test.stop_sequence_matched, + EXPECT_EQ(ResolveGeneratedFinishReason(test.has_tool_calls, test.stop_sequence_matched, test.host_output_limit_reached, test.backend_finish_reason, /*completion_tokens=*/32, /*max_output_tokens=*/32), test.expected); @@ -1849,7 +1856,7 @@ TEST_F(QwenNativeProductionIntegrationTest, std::function hook; if (index == 1) { hook = [&] { - active_request->canceled.store(true, std::memory_order_relaxed); + active_request->Cancel(); }; } @@ -1879,9 +1886,10 @@ TEST_F(QwenNativeProductionIntegrationTest, auto request = MakeStatefulRequest("route this"); active_request = &request; Response response; - EXPECT_NO_THROW(session.ProcessRequest(request, response)); + ExpectOperationCancelled([&] { session.ProcessRequest(request, response); }); - EXPECT_TRUE(request.canceled.load(std::memory_order_relaxed)); + EXPECT_TRUE(request.IsCompleted()); + EXPECT_FALSE(request.IsCancellationRequested()); EXPECT_EQ(counters->created, 2); EXPECT_EQ(counters->closed, 1); EXPECT_EQ(counters->canceled, 1); @@ -1918,17 +1926,15 @@ TEST_F(QwenNativeProductionIntegrationTest, auto request = MakeStatefulRequest("route this"); Response response; - EXPECT_NO_THROW(session.ProcessRequest(request, response)); + ExpectOperationCancelled([&] { session.ProcessRequest(request, response); }); - EXPECT_TRUE(request.canceled.load(std::memory_order_relaxed)); + EXPECT_TRUE(request.IsCompleted()); + EXPECT_FALSE(request.IsCancellationRequested()); EXPECT_EQ(counters->created, 1); EXPECT_EQ(counters->canceled, 1); EXPECT_EQ(counters->usage_after_cancel, 1); EXPECT_EQ(streamed_text, "safe prefix"); - ASSERT_EQ(response.items.size(), 1u); - ASSERT_EQ(response.items.front()->type, FOUNDRY_LOCAL_ITEM_MESSAGE); - EXPECT_EQ(static_cast(*response.items.front()).GetSimpleText(), - output); + EXPECT_TRUE(response.items.empty()); EXPECT_EQ(response.finish_reason, FOUNDRY_LOCAL_FINISH_NONE); EXPECT_EQ(session.TurnCount(), 0u); EXPECT_TRUE(session.Transcript().Empty()); @@ -1979,9 +1985,10 @@ TEST_F(QwenNativeProductionIntegrationTest, request.AddOwnedItem(std::make_unique( body.dump(), FOUNDRY_LOCAL_TEXT_ITEM_TYPE_OPENAI_JSON)); Response response; - EXPECT_NO_THROW(session.ProcessRequest(request, response)); + ExpectOperationCancelled([&] { session.ProcessRequest(request, response); }); - EXPECT_TRUE(request.canceled.load(std::memory_order_relaxed)); + EXPECT_TRUE(request.IsCompleted()); + EXPECT_FALSE(request.IsCancellationRequested()); EXPECT_EQ(counters->created, 1); EXPECT_EQ(counters->canceled, 1); EXPECT_EQ(counters->usage_after_cancel, 1); @@ -1996,10 +2003,7 @@ TEST_F(QwenNativeProductionIntegrationTest, EXPECT_EQ(streamed_content, "safe prefix"); EXPECT_EQ(streamed_content.find(""), std::string::npos); EXPECT_EQ(streamed_content.find("type, FOUNDRY_LOCAL_ITEM_MESSAGE); - EXPECT_EQ(static_cast(*response.items.front()).GetSimpleText(), - output); + EXPECT_TRUE(response.items.empty()); EXPECT_EQ(response.finish_reason, FOUNDRY_LOCAL_FINISH_NONE); EXPECT_TRUE(session.Transcript().Empty()); } @@ -3621,12 +3625,12 @@ TEST_F(ChatSessionTest, AppendedClassicGeneratorIsDiscardedAfterStreamingCancell session.SetStreamingCallback(callback_fn); Response response; - session.ProcessRequest(request, response); + ExpectOperationCancelled([&] { session.ProcessRequest(request, response); }); - // Should have stopped early + // The canceled response is not published. EXPECT_EQ(response.finish_reason, FOUNDRY_LOCAL_FINISH_NONE); - // we check cancellation at the start of each loop and we don't use std::atomic to make processing cheaper - // so allow for a couple of extra tokens to come through after the cancellation condition is met + EXPECT_TRUE(response.items.empty()); + // Cancellation is observed between token steps, so allow for a couple of extra tokens after the callback. EXPECT_LE(tokens_received, 6); // The appended canceled turn commits nothing; only the seed turn remains. @@ -3672,9 +3676,10 @@ TEST_F(ChatSessionTest, CancellationFromTheLastQueuedCallbackPreventsCommit) { }); Response response; - session.ProcessRequest(request, response); + ExpectOperationCancelled([&] { session.ProcessRequest(request, response); }); EXPECT_EQ(response.finish_reason, FOUNDRY_LOCAL_FINISH_NONE); + EXPECT_TRUE(response.items.empty()); EXPECT_EQ(session.MessageCount(), 0u); EXPECT_EQ(session.TurnCount(), 0u); } @@ -3719,9 +3724,10 @@ TEST_F(ChatSessionTest, OpenAIJsonCancellationFromLastContentCallbackPublishesNo }); Response response; - session.ProcessRequest(request, response); + ExpectOperationCancelled([&] { session.ProcessRequest(request, response); }); EXPECT_EQ(response.finish_reason, FOUNDRY_LOCAL_FINISH_NONE); + EXPECT_TRUE(response.items.empty()); ASSERT_EQ(delivered.size(), 2u); EXPECT_EQ(delivered[0]["choices"][0]["delta"]["role"], "assistant"); EXPECT_FALSE(delivered[0]["choices"][0]["finish_reason"].is_string()); diff --git a/sdk_v2/cpp/test/internal_api/chat/dynamic_engine_chat_test.cc b/sdk_v2/cpp/test/internal_api/chat/dynamic_engine_chat_test.cc index 81839d3c8..ee66e5903 100644 --- a/sdk_v2/cpp/test/internal_api/chat/dynamic_engine_chat_test.cc +++ b/sdk_v2/cpp/test/internal_api/chat/dynamic_engine_chat_test.cc @@ -482,6 +482,23 @@ TEST_F(DynamicEngineChatTest, LateCancelAfterCompletedConversationRemovalIsNoOp) EXPECT_EQ(result_after_cancel.finish_reason, result.finish_reason); } +TEST_F(DynamicEngineChatTest, BackendCancellationIsNotAnApiErrorByItself) { + SearchOptions options; + options.max_output_tokens = 512; + options.temperature = 0.0f; + ToolCallContext tool_context; + std::vector messages = { + {FOUNDRY_LOCAL_ROLE_USER, "Write a long essay about mathematics."}}; + auto stream = OnnxEngineChatStream::Create(messages, options, ModelInstance(), tool_context); + + stream->GenerateNextToken(); + stream->Cancel(); + + const auto usage = stream->GetTurnUsage(); + ASSERT_TRUE(usage.has_value()); + EXPECT_FALSE(usage->finish_reason.has_value()); +} + TEST_F(DynamicEngineChatTest, CancellationRebuildsCommittedHistoryWithinBudget) { ChatSession session(CatalogModel(), ModelInstance(), *logger_, telemetry_); @@ -499,8 +516,14 @@ TEST_F(DynamicEngineChatTest, CancellationRebuildsCommittedHistoryWithinBudget) auto canceled = MakeRequest("Write a long essay about mathematics.", 512); Response canceled_response; - session.ProcessRequest(canceled, canceled_response); + try { + session.ProcessRequest(canceled, canceled_response); + FAIL() << "Expected operation cancellation"; + } catch (const fl::Exception& error) { + EXPECT_EQ(error.code(), FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED); + } EXPECT_EQ(canceled_response.finish_reason, FOUNDRY_LOCAL_FINISH_NONE); + EXPECT_TRUE(canceled_response.items.empty()); EXPECT_EQ(session.TurnCount(), 1u); EXPECT_GE(streamed_tokens, 3); diff --git a/sdk_v2/cpp/test/internal_api/item_test.cc b/sdk_v2/cpp/test/internal_api/item_test.cc index 02d5284ec..b25f746ff 100644 --- a/sdk_v2/cpp/test/internal_api/item_test.cc +++ b/sdk_v2/cpp/test/internal_api/item_test.cc @@ -683,12 +683,59 @@ TEST(RequestTest, MixedOwnedAndBorrowedItems) { EXPECT_TRUE(req.items[1]->type == FOUNDRY_LOCAL_ITEM_MESSAGE); } -TEST(RequestTest, CancellationFlag) { +TEST(RequestTest, ReadyCancellationIsNoOpAndTryBeginSucceeds) { Request req; - EXPECT_FALSE(req.canceled); + EXPECT_FALSE(req.IsCancellationRequested()); - req.canceled = true; - EXPECT_TRUE(req.canceled); + EXPECT_FALSE(req.Cancel()); + EXPECT_FALSE(req.IsCancellationRequested()); + EXPECT_TRUE(req.TryBegin()); +} + +TEST(RequestTest, CancellationWinsBeforeCompletion) { + Request req; + ASSERT_TRUE(req.TryBegin()); + + EXPECT_TRUE(req.Cancel()); + EXPECT_TRUE(req.Cancel(Request::CancellationReason::SessionShutdown)); + EXPECT_TRUE(req.IsCancellationRequested()); + EXPECT_EQ(req.GetCancellationReason(), Request::CancellationReason::Caller); + EXPECT_FALSE(req.TryComplete()); +} + +TEST(RequestTest, CancellationPreservesActionableReason) { + Request callback_request; + ASSERT_TRUE(callback_request.TryBegin()); + EXPECT_TRUE(callback_request.Cancel(Request::CancellationReason::StreamingCallback)); + EXPECT_EQ(callback_request.GetCancellationReason(), Request::CancellationReason::StreamingCallback); + + Request exception_request; + ASSERT_TRUE(exception_request.TryBegin()); + EXPECT_TRUE(exception_request.CancelFromStreamingCallbackException("callback exploded")); + EXPECT_TRUE(exception_request.CancelFromStreamingCallbackException("replacement detail")); + EXPECT_EQ(exception_request.GetCancellationReason(), Request::CancellationReason::StreamingCallbackException); + EXPECT_EQ(exception_request.CancellationDetail(), "callback exploded"); + + Request shutdown_request; + ASSERT_TRUE(shutdown_request.TryBegin()); + EXPECT_TRUE(shutdown_request.Cancel(Request::CancellationReason::SessionShutdown)); + EXPECT_EQ(shutdown_request.GetCancellationReason(), Request::CancellationReason::SessionShutdown); +} + +TEST(RequestTest, CompletedCancellationIsNoOpAndRequestCanBeReused) { + Request req; + + ASSERT_TRUE(req.TryBegin()); + EXPECT_TRUE(req.TryComplete()); + EXPECT_FALSE(req.Cancel()); + EXPECT_FALSE(req.IsCancellationRequested()); + EXPECT_FALSE(req.IsCompleted()); + EXPECT_FALSE(req.TryBegin()); + req.PublishCompletion(); + EXPECT_TRUE(req.IsCompleted()); + EXPECT_FALSE(req.Cancel()); + EXPECT_FALSE(req.IsCancellationRequested()); + EXPECT_TRUE(req.TryBegin()); } // ======================================================================== diff --git a/sdk_v2/cpp/test/internal_api/session_manager_test.cc b/sdk_v2/cpp/test/internal_api/session_manager_test.cc index a7b29eab9..1e465514c 100644 --- a/sdk_v2/cpp/test/internal_api/session_manager_test.cc +++ b/sdk_v2/cpp/test/internal_api/session_manager_test.cc @@ -399,7 +399,7 @@ TEST_F(SessionManagerTest, CheckedOutSessionNotAffectedByCheckIn) { namespace { -/// Test-only Session that blocks inside ProcessRequestImpl until its request's cancel flag is +/// Test-only Session that blocks inside ProcessRequestImpl until its request's cancellation is /// observed. Lets a unit test verify SessionManager::CancelAll() propagates cancellation to every /// registered session without loading a model. Polls the atomic exactly like the real generation /// loop, with a safety deadline so a broken Cancel() fails the test instead of hanging the suite. @@ -417,7 +417,7 @@ class BlockingCancelSession : public Session { in_flight_.store(true, std::memory_order_release); const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); - while (!request.canceled.load(std::memory_order_relaxed)) { + while (!request.IsCancellationRequested()) { if (std::chrono::steady_clock::now() >= deadline) { return; // safety net: a broken Cancel() must not hang the test suite } @@ -430,6 +430,165 @@ class BlockingCancelSession : public Session { std::atomic in_flight_{false}; }; +class CompletingSession : public Session { + public: + CompletingSession(const Model& model, ILogger& logger, ITelemetry& telemetry) + : Session(model, logger, telemetry) {} + + SessionType Type() const override { return SessionType::kChat; } + size_t ProcessCount() const { return process_count_; } + + protected: + void ProcessRequestImpl(const Request& /*request*/, Response& response) override { + ++process_count_; + response.finish_reason = FOUNDRY_LOCAL_FINISH_STOP; + } + + private: + size_t process_count_ = 0; +}; + +class ControlledCancelSession : public Session { + public: + ControlledCancelSession(const Model& model, ILogger& logger, ITelemetry& telemetry) + : Session(model, logger, telemetry, /*allow_concurrent_requests=*/true) {} + + SessionType Type() const override { return SessionType::kChat; } + bool FirstInvocationEntered() const { return first_invocation_entered_.load(std::memory_order_acquire); } + size_t ProcessCount() const { return process_count_.load(std::memory_order_acquire); } + + void ReleaseFirstInvocation() { + release_first_invocation_.store(true, std::memory_order_release); + } + + protected: + void ProcessRequestImpl(const Request& request, Response& response) override { + const auto invocation = process_count_.fetch_add(1, std::memory_order_acq_rel); + if (invocation == 0) { + first_invocation_entered_.store(true, std::memory_order_release); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (!request.IsCancellationRequested() || !release_first_invocation_.load(std::memory_order_acquire)) { + if (std::chrono::steady_clock::now() >= deadline) { + return; + } + + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + } + + response.finish_reason = FOUNDRY_LOCAL_FINISH_STOP; + } + + private: + std::atomic process_count_{0}; + std::atomic first_invocation_entered_{false}; + std::atomic release_first_invocation_{false}; +}; + +class CallbackExceptionSession : public Session { + public: + CallbackExceptionSession(const Model& model, ILogger& logger, ITelemetry& telemetry) + : Session(model, logger, telemetry) {} + + SessionType Type() const override { return SessionType::kChat; } + size_t ProcessCount() const { return process_count_; } + + protected: + void ProcessRequestImpl(const Request& request, Response& response) override { + ++process_count_; + if (process_count_ == 1) { + request.CancelFromStreamingCallbackException("callback exploded"); + return; + } + + response.finish_reason = FOUNDRY_LOCAL_FINISH_STOP; + } + + private: + size_t process_count_ = 0; +}; + +class ThrowThenCompleteSession : public Session { + public: + ThrowThenCompleteSession(const Model& model, ILogger& logger, ITelemetry& telemetry) + : Session(model, logger, telemetry) {} + + SessionType Type() const override { return SessionType::kChat; } + size_t ProcessCount() const { return process_count_; } + + protected: + void ProcessRequestImpl(const Request& /*request*/, Response& response) override { + ++process_count_; + if (process_count_ == 1) { + throw std::runtime_error("backend exploded"); + } + + response.finish_reason = FOUNDRY_LOCAL_FINISH_STOP; + } + + private: + size_t process_count_ = 0; +}; + +class NonStdThrowThenCompleteSession : public Session { + public: + NonStdThrowThenCompleteSession(const Model& model, ILogger& logger, ITelemetry& telemetry) + : Session(model, logger, telemetry) {} + + SessionType Type() const override { return SessionType::kChat; } + size_t ProcessCount() const { return process_count_; } + + protected: + void ProcessRequestImpl(const Request& /*request*/, Response& response) override { + ++process_count_; + if (process_count_ == 1) { + throw 42; + } + + response.finish_reason = FOUNDRY_LOCAL_FINISH_STOP; + } + + private: + size_t process_count_ = 0; +}; + +class NonStdThrowAfterCancelSession : public Session { + public: + NonStdThrowAfterCancelSession(const Model& model, ILogger& logger, ITelemetry& telemetry) + : Session(model, logger, telemetry) {} + + SessionType Type() const override { return SessionType::kChat; } + bool InFlight() const { return in_flight_.load(std::memory_order_acquire); } + + protected: + void ProcessRequestImpl(const Request& request, Response& /*response*/) override { + in_flight_.store(true, std::memory_order_release); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (!request.IsCancellationRequested()) { + if (std::chrono::steady_clock::now() >= deadline) { + throw 42; + } + + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + + throw 42; + } + + private: + std::atomic in_flight_{false}; +}; + +flErrorCode ProcessAndGetCode(Session& session, const Request& request) { + try { + Response response; + session.ProcessRequest(request, response); + return FOUNDRY_LOCAL_OK; + } catch (const Exception& error) { + return error.code(); + } +} + /// Spin until `pred` is true or the timeout elapses. Returns pred's final value. template bool WaitUntil(Pred pred, std::chrono::milliseconds timeout) { @@ -463,25 +622,23 @@ TEST(SessionManagerCancelTest, CancelAllCancelsInFlightRequestsOnEverySession) { // Drive each session's blocking ProcessRequest on its own worker so both requests are in-flight // (registered in active_requests_) at the same time — exercising "every registered session". - auto f1 = std::async(std::launch::async, [&] { - Response resp; - s1.ProcessRequest(req1, resp); - }); - auto f2 = std::async(std::launch::async, [&] { - Response resp; - s2.ProcessRequest(req2, resp); - }); + auto f1 = std::async(std::launch::async, [&] { return ProcessAndGetCode(s1, req1); }); + auto f2 = std::async(std::launch::async, [&] { return ProcessAndGetCode(s2, req2); }); ASSERT_TRUE(WaitUntil([&] { return s1.InFlight() && s2.InFlight(); }, std::chrono::seconds(2))) << "worker requests never became in-flight"; mgr.CancelAll(); - // CancelAll set each request's flag; the blocked workers observe it and return promptly. + // CancelAll canceled each request; the blocked workers observe it and return promptly. EXPECT_EQ(f1.wait_for(std::chrono::seconds(2)), std::future_status::ready); EXPECT_EQ(f2.wait_for(std::chrono::seconds(2)), std::future_status::ready); - EXPECT_TRUE(req1.canceled.load(std::memory_order_relaxed)); - EXPECT_TRUE(req2.canceled.load(std::memory_order_relaxed)); + EXPECT_EQ(f1.get(), FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED); + EXPECT_EQ(f2.get(), FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED); + EXPECT_TRUE(req1.IsCompleted()); + EXPECT_TRUE(req2.IsCompleted()); + EXPECT_FALSE(req1.IsCancellationRequested()); + EXPECT_FALSE(req2.IsCancellationRequested()); } TEST(SessionManagerCancelTest, RequestAdmittedAfterCancelIsStampedCanceled) { @@ -505,12 +662,177 @@ TEST(SessionManagerCancelTest, RequestAdmittedAfterCancelIsStampedCanceled) { // Now drive the request. It is admitted after Cancel() ran, so ProcessRequest must stamp it on // insert and the blocking loop must observe cancellation at its first poll. - auto f = std::async(std::launch::async, [&] { - Response resp; - s.ProcessRequest(req, resp); - }); + auto f = std::async(std::launch::async, [&] { return ProcessAndGetCode(s, req); }); EXPECT_EQ(f.wait_for(std::chrono::seconds(2)), std::future_status::ready) << "late-admitted request ran uncanceled — the session_canceled_ latch did not stamp it"; - EXPECT_TRUE(req.canceled.load(std::memory_order_relaxed)); + EXPECT_EQ(f.get(), FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED); + EXPECT_TRUE(req.IsCompleted()); + EXPECT_FALSE(req.IsCancellationRequested()); + EXPECT_FALSE(s.InFlight()); +} + +TEST(SessionRequestLifecycleTest, IdleCancelIsNoOpAndRequestReachesBackend) { + fl::test::FakeServiceBindings svc; + Model catalog_model = Model::FromModelInfo(ModelInfo{}, "", svc.download_manager, svc.model_load_manager); + TelemetryLogger telemetry{"test", fl::test::NullLog()}; + CompletingSession session(catalog_model, fl::test::NullLog(), telemetry); + Request request; + ASSERT_FALSE(request.Cancel()); + + EXPECT_EQ(ProcessAndGetCode(session, request), FOUNDRY_LOCAL_OK); + EXPECT_EQ(session.ProcessCount(), 1u); +} + +TEST(SessionRequestLifecycleTest, PublishedCompletionMakesLateCancellationNoOp) { + fl::test::FakeServiceBindings svc; + Model catalog_model = Model::FromModelInfo(ModelInfo{}, "", svc.download_manager, svc.model_load_manager); + TelemetryLogger telemetry{"test", fl::test::NullLog()}; + CompletingSession session(catalog_model, fl::test::NullLog(), telemetry); + Request request; + Response response; + + session.ProcessRequest(request, response); + + EXPECT_EQ(response.finish_reason, FOUNDRY_LOCAL_FINISH_STOP); + EXPECT_TRUE(request.IsCompleted()); + EXPECT_FALSE(request.Cancel()); + EXPECT_FALSE(request.IsCancellationRequested()); +} + +TEST(SessionRequestLifecycleTest, ReusedRequestAfterSessionCancelNeverReachesBackend) { + fl::test::FakeServiceBindings svc; + Model catalog_model = Model::FromModelInfo(ModelInfo{}, "", svc.download_manager, svc.model_load_manager); + TelemetryLogger telemetry{"test", fl::test::NullLog()}; + CompletingSession session(catalog_model, fl::test::NullLog(), telemetry); + Request request; + Response response; + + session.ProcessRequest(request, response); + ASSERT_EQ(session.ProcessCount(), 1u); + ASSERT_TRUE(request.IsCompleted()); + + session.Cancel(); + + EXPECT_EQ(ProcessAndGetCode(session, request), FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED); + EXPECT_TRUE(request.IsCompleted()); + EXPECT_EQ(request.GetCancellationReason(), Request::CancellationReason::None); + EXPECT_EQ(session.ProcessCount(), 1u); +} + +TEST(SessionRequestLifecycleTest, InFlightCancellationAllowsRequestReuse) { + fl::test::FakeServiceBindings svc; + Model catalog_model = Model::FromModelInfo(ModelInfo{}, "", svc.download_manager, svc.model_load_manager); + TelemetryLogger telemetry{"test", fl::test::NullLog()}; + ControlledCancelSession session(catalog_model, fl::test::NullLog(), telemetry); + Request request; + + auto first = std::async(std::launch::async, [&] { return ProcessAndGetCode(session, request); }); + ASSERT_TRUE(WaitUntil([&] { return session.FirstInvocationEntered(); }, std::chrono::seconds(2))); + ASSERT_TRUE(request.Cancel()); + session.ReleaseFirstInvocation(); + + EXPECT_EQ(first.get(), FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED); + EXPECT_EQ(request.GetCancellationReason(), Request::CancellationReason::None); + EXPECT_TRUE(request.IsCompleted()); + + EXPECT_EQ(ProcessAndGetCode(session, request), FOUNDRY_LOCAL_OK); + EXPECT_EQ(session.ProcessCount(), 2u); +} + +TEST(SessionRequestLifecycleTest, ConcurrentReuseOfCanceledActiveRequestIsRejected) { + fl::test::FakeServiceBindings svc; + Model catalog_model = Model::FromModelInfo(ModelInfo{}, "", svc.download_manager, svc.model_load_manager); + TelemetryLogger telemetry{"test", fl::test::NullLog()}; + ControlledCancelSession session(catalog_model, fl::test::NullLog(), telemetry); + Request request; + + auto first = std::async(std::launch::async, [&] { return ProcessAndGetCode(session, request); }); + ASSERT_TRUE(WaitUntil([&] { return session.FirstInvocationEntered(); }, std::chrono::seconds(2))); + ASSERT_TRUE(request.Cancel()); + + auto concurrent = std::async(std::launch::async, [&] { return ProcessAndGetCode(session, request); }); + EXPECT_EQ(concurrent.wait_for(std::chrono::seconds(2)), std::future_status::ready); + EXPECT_EQ(concurrent.get(), FOUNDRY_LOCAL_ERROR_INVALID_USAGE); + EXPECT_EQ(session.ProcessCount(), 1u); + + session.ReleaseFirstInvocation(); + EXPECT_EQ(first.get(), FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED); +} + +TEST(SessionRequestLifecycleTest, OrdinaryExceptionAllowsRequestReuse) { + fl::test::FakeServiceBindings svc; + Model catalog_model = Model::FromModelInfo(ModelInfo{}, "", svc.download_manager, svc.model_load_manager); + TelemetryLogger telemetry{"test", fl::test::NullLog()}; + ThrowThenCompleteSession session(catalog_model, fl::test::NullLog(), telemetry); + Request request; + + EXPECT_THROW( + { + Response response; + session.ProcessRequest(request, response); + }, + std::runtime_error); + EXPECT_TRUE(request.IsCompleted()); + + EXPECT_EQ(ProcessAndGetCode(session, request), FOUNDRY_LOCAL_OK); + EXPECT_EQ(session.ProcessCount(), 2u); +} + +TEST(SessionRequestLifecycleTest, NonStdExceptionAllowsRequestReuse) { + fl::test::FakeServiceBindings svc; + Model catalog_model = Model::FromModelInfo(ModelInfo{}, "", svc.download_manager, svc.model_load_manager); + TelemetryLogger telemetry{"test", fl::test::NullLog()}; + NonStdThrowThenCompleteSession session(catalog_model, fl::test::NullLog(), telemetry); + Request request; + + try { + Response response; + session.ProcessRequest(request, response); + FAIL() << "expected non-standard exception"; + } catch (int value) { + EXPECT_EQ(value, 42); + } + EXPECT_TRUE(request.IsCompleted()); + + EXPECT_EQ(ProcessAndGetCode(session, request), FOUNDRY_LOCAL_OK); + EXPECT_EQ(session.ProcessCount(), 2u); +} + +TEST(SessionRequestLifecycleTest, CancellationWinsOverNonStdException) { + fl::test::FakeServiceBindings svc; + Model catalog_model = Model::FromModelInfo(ModelInfo{}, "", svc.download_manager, svc.model_load_manager); + TelemetryLogger telemetry{"test", fl::test::NullLog()}; + NonStdThrowAfterCancelSession session(catalog_model, fl::test::NullLog(), telemetry); + Request request; + + auto processing = std::async(std::launch::async, [&] { return ProcessAndGetCode(session, request); }); + ASSERT_TRUE(WaitUntil([&] { return session.InFlight(); }, std::chrono::seconds(2))); + ASSERT_TRUE(request.Cancel()); + + EXPECT_EQ(processing.get(), FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED); + EXPECT_TRUE(request.IsCompleted()); +} + +TEST(SessionRequestLifecycleTest, CallbackExceptionSurfacesOriginalCauseAndAllowsReuse) { + fl::test::FakeServiceBindings svc; + Model catalog_model = Model::FromModelInfo(ModelInfo{}, "", svc.download_manager, svc.model_load_manager); + TelemetryLogger telemetry{"test", fl::test::NullLog()}; + CallbackExceptionSession session(catalog_model, fl::test::NullLog(), telemetry); + Request request; + + try { + Response response; + session.ProcessRequest(request, response); + FAIL() << "expected callback failure"; + } catch (const Exception& error) { + EXPECT_EQ(error.code(), FOUNDRY_LOCAL_ERROR_INTERNAL); + EXPECT_NE(std::string(error.what()).find("streaming callback threw an exception: callback exploded"), + std::string::npos); + } + + EXPECT_TRUE(request.IsCompleted()); + EXPECT_EQ(request.GetCancellationReason(), Request::CancellationReason::None); + EXPECT_EQ(ProcessAndGetCode(session, request), FOUNDRY_LOCAL_OK); + EXPECT_EQ(session.ProcessCount(), 2u); } diff --git a/sdk_v2/cpp/test/sdk_api/cpp_api_test.cc b/sdk_v2/cpp/test/sdk_api/cpp_api_test.cc index 8dc7e67cb..316df305d 100644 --- a/sdk_v2/cpp/test/sdk_api/cpp_api_test.cc +++ b/sdk_v2/cpp/test/sdk_api/cpp_api_test.cc @@ -284,9 +284,9 @@ TEST(CppApiTest, RequestSetOptions) { request.SetOptions(opts); } -TEST(CppApiTest, RequestCancelOnIdleRequest) { +TEST(CppApiTest, RequestCancelWhileIdleSucceeds) { foundry_local::Request request; - // Cancel on an idle request should succeed (no-op) + // Idle cancellation is a successful no-op. EXPECT_NO_THROW(request.Cancel()); } diff --git a/sdk_v2/cpp/test/sdk_api/streaming_audio_test.cc b/sdk_v2/cpp/test/sdk_api/streaming_audio_test.cc index 3594de61f..60a90d6ec 100644 --- a/sdk_v2/cpp/test/sdk_api/streaming_audio_test.cc +++ b/sdk_v2/cpp/test/sdk_api/streaming_audio_test.cc @@ -9,12 +9,14 @@ #include "utils/string_utils.h" #include +#include #include #include #include #include #include #include +#include #include using fl::test::ToLower; @@ -202,6 +204,7 @@ TEST_F(StreamingAudioFixture, CancellationMidStream) { auto pcm = LoadPcm(); auto chunks = SplitIntoChunks(pcm, 3200); + ASSERT_GT(chunks.size(), 1u); auto audio = Item::AudioFromData("pcm", nullptr, 0, /*sample_rate=*/16000, /*channels=*/1); ItemQueue queue; @@ -221,13 +224,30 @@ TEST_F(StreamingAudioFixture, CancellationMidStream) { queue.Push(Item::Bytes(FOUNDRY_LOCAL_ITEM_BYTES, chunks[i].data(), chunks[i].size())); } + const auto admission_deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + while (queue.Size() == half && std::chrono::steady_clock::now() < admission_deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + + if (queue.Size() == half) { + request.Cancel(); + queue.MarkFinished(); + future.wait(); + FAIL() << "Audio session did not consume input before the cancellation deadline"; + } + + ASSERT_EQ(future.wait_for(std::chrono::milliseconds(0)), std::future_status::timeout) + << "Audio session completed before cancellation"; + request.Cancel(); queue.MarkFinished(); - Response response = future.get(); - - EXPECT_EQ(response.GetFinishReason(), FOUNDRY_LOCAL_FINISH_NONE) - << "Cancelled request should have NONE finish reason"; + try { + (void)future.get(); + FAIL() << "Expected operation cancellation"; + } catch (const foundry_local::Error& error) { + EXPECT_EQ(error.Code(), FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED); + } } TEST_F(StreamingAudioFixture, StreamingCallbackReceivesTokens) { diff --git a/sdk_v2/cs/src/Request.cs b/sdk_v2/cs/src/Request.cs index 7761f0e61..4ece717a6 100644 --- a/sdk_v2/cs/src/Request.cs +++ b/sdk_v2/cs/src/Request.cs @@ -92,6 +92,9 @@ internal Request SetOptions(IntPtr options) return this; } + /// + /// Cancels this request's in-flight invocation. This has no effect while the request is idle or after completion. + /// public void Cancel() { Api.CheckStatus(Api.Inference.RequestCancel(Ptr)); diff --git a/sdk_v2/cs/src/StreamingResponse.cs b/sdk_v2/cs/src/StreamingResponse.cs index b8f2dd682..b349b251d 100644 --- a/sdk_v2/cs/src/StreamingResponse.cs +++ b/sdk_v2/cs/src/StreamingResponse.cs @@ -146,8 +146,8 @@ private async IAsyncEnumerable EnumerateAsync([EnumeratorCancellation] Can // TryComplete (channel completion is always preceded by ProcessRequest returning). The // native request has therefore already produced its full output, so cleanup MUST NOT // signal abort — doing so races the native streaming-callback worker thread and can - // flip request.canceled on a still-finalizing request, truncating the output. Only an - // early break / dispose / external-token cancellation (which skip this line) should abort. + // transition a still-finalizing request to canceled, truncating the output. Only an early + // break / dispose / external-token cancellation (which skip this line) should abort. Interlocked.Exchange(ref _drainedNaturally, 1); } finally diff --git a/sdk_v2/js/docs/PortJsToSdkV2.md b/sdk_v2/js/docs/PortJsToSdkV2.md index 439def37b..30369ee41 100644 --- a/sdk_v2/js/docs/PortJsToSdkV2.md +++ b/sdk_v2/js/docs/PortJsToSdkV2.md @@ -115,13 +115,12 @@ underlying C ABI call is a memory copy, not I/O. ## Streaming -- `Session.processRequestStreaming` returns an `AsyncIterable`. Each - native streaming-callback push lands on a `Napi::ThreadSafeFunction` - acquired in the session's constructor and released when the iterable is - closed. -- Cancellation: each async API accepts an `AbortSignal`. The signal is - bound to `Request::Cancel()`, which the C++ wrapper translates into a - cancellation signal observed by the streaming callback. +- `Session.processStreamingRequest` returns an `AsyncIterable`. Each request owns a worker whose + `Napi::ThreadSafeFunction` forwards native streaming-callback items and is released after the worker completes and + its queued callbacks drain. +- Cancellation: streaming APIs accept an `AbortSignal`. A signal already aborted at call time rejects before native + work is submitted. Once submitted, the signal calls `Request::Cancel()`, which affects only an invocation currently + inside native `Session::ProcessRequest`; work still waiting in the addon's per-session queue is not canceled. - Live PCM input (audio transcription with chunks arriving over time) is expressed by adding an `AudioItem` descriptor to the `Request` and pushing PCM bytes through a paired `ItemQueue`. The session consumes the diff --git a/sdk_v2/js/native/src/manager.cc b/sdk_v2/js/native/src/manager.cc index e94f29803..d78e67262 100644 --- a/sdk_v2/js/native/src/manager.cc +++ b/sdk_v2/js/native/src/manager.cc @@ -5,7 +5,6 @@ #include "addon_data.h" #include "catalog.h" #include "errors.h" -#include "promise_worker.h" #include #include diff --git a/sdk_v2/js/native/src/promise_worker.h b/sdk_v2/js/native/src/promise_worker.h index b0a338229..a495d0b66 100644 --- a/sdk_v2/js/native/src/promise_worker.h +++ b/sdk_v2/js/native/src/promise_worker.h @@ -1,7 +1,7 @@ // Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. // -// Used by model download/load/unload and session inference. +// Used by asynchronous catalog queries and model load/unload operations. // // PromiseWorker — generic Napi::AsyncWorker that runs a std::function // on a libuv worker thread and resolves / rejects a JS Promise with the diff --git a/sdk_v2/js/native/src/session.cc b/sdk_v2/js/native/src/session.cc index 5e92dbc9a..0b40197db 100644 --- a/sdk_v2/js/native/src/session.cc +++ b/sdk_v2/js/native/src/session.cc @@ -6,13 +6,14 @@ #include "errors.h" #include "items.h" #include "model.h" -#include "promise_worker.h" #include "request.h" #include "request_options.h" #include #include +#include +#include #include #include #include @@ -20,6 +21,31 @@ namespace foundry_local_node { +void SessionScheduler::Enqueue(std::function start) { + pending_.push_back(std::move(start)); + StartNext(); +} + +void SessionScheduler::Complete() { + if (!running_) { + return; + } + + running_ = false; + StartNext(); +} + +void SessionScheduler::StartNext() { + if (running_ || pending_.empty()) { + return; + } + + running_ = true; + auto start = std::move(pending_.front()); + pending_.pop_front(); + start(); +} + namespace { const char* FinishReasonToString(flFinishReason r) { @@ -93,30 +119,145 @@ foundry_local::Request* UnwrapRequest(Napi::Env env, const Napi::Value& v) { return req->native(); } -// Process a Request on the worker thread, converting to JS in the resolver. -// Pins both the Manager (so the Model handle the Session holds stays alive) -// and the Request (so the C++ Request the worker reads stays alive). +// Process a Request on a worker thread, converting to JS after completion. Chat/audio workers are queued by their +// per-session scheduler before consuming a libuv worker; embeddings workers start immediately. template -Napi::Value ProcessRequestOn(Napi::Env env, SessT* sess, const Napi::Value& request_arg, - Napi::ObjectReference manager_ref) { +class SessionPromiseWorker : public Napi::AsyncWorker { + public: + using Result = std::shared_ptr; + + static Napi::Promise Run(Napi::Env env, std::shared_ptr sess, foundry_local::Request* req, + Napi::ObjectReference manager, Napi::ObjectReference request, + std::shared_ptr scheduler) { + auto* worker = new SessionPromiseWorker(env, std::move(sess), req, std::move(manager), std::move(request), + std::move(scheduler)); + Napi::Promise promise = worker->deferred_.Promise(); + try { + if (worker->scheduler_ != nullptr) { + worker->scheduler_->Enqueue([worker] { worker->Start(); }); + } else { + worker->Start(); + } + } catch (const std::exception& e) { + worker->FailBeforeQueue(e.what()); + } catch (...) { + worker->FailBeforeQueue("Failed to schedule native request"); + } + + return promise; + } + + void Execute() override { + try { + response_ = std::make_shared(sess_->ProcessRequest(*req_)); + } catch (const foundry_local::Error& e) { + err_code_ = static_cast(e.Code()); + err_msg_ = e.what(); + tagged_ = true; + SetError(err_msg_); + } catch (const std::exception& e) { + err_msg_ = e.what(); + SetError(err_msg_); + } catch (...) { + err_msg_ = "Unknown native exception"; + SetError(err_msg_); + } + } + + void OnOK() override { + Napi::Env env = Env(); + Napi::HandleScope scope(env); + try { + deferred_.Resolve(ResponseToJs(env, *response_)); + } catch (const Napi::Error& e) { + deferred_.Reject(e.Value()); + } catch (const std::exception& e) { + deferred_.Reject(Napi::Error::New(env, e.what()).Value()); + } catch (...) { + deferred_.Reject(Napi::Error::New(env, "Failed to convert native response").Value()); + } + + CompleteScheduler(); + } + + void OnError(const Napi::Error& /*unused*/) override { + Napi::Env env = Env(); + Napi::HandleScope scope(env); + if (tagged_) { + Napi::Error err = Napi::Error::New(env, err_msg_); + Napi::Object value = err.Value(); + value.Set("name", Napi::String::New(env, "FoundryLocalError")); + value.Set("code", Napi::Number::New(env, err_code_)); + deferred_.Reject(value); + } else { + deferred_.Reject(Napi::Error::New(env, err_msg_).Value()); + } + + CompleteScheduler(); + } + + private: + SessionPromiseWorker(Napi::Env env, std::shared_ptr sess, foundry_local::Request* req, + Napi::ObjectReference manager, Napi::ObjectReference request, + std::shared_ptr scheduler) + : Napi::AsyncWorker(env), + deferred_(Napi::Promise::Deferred::New(env)), + sess_(std::move(sess)), + req_(req), + manager_(std::move(manager)), + request_(std::move(request)), + scheduler_(std::move(scheduler)) {} + + void Start() { + try { + Queue(); + } catch (const std::exception& e) { + deferred_.Reject(Napi::Error::New(Env(), e.what()).Value()); + CompleteScheduler(); + delete this; + } catch (...) { + deferred_.Reject(Napi::Error::New(Env(), "Failed to queue native request").Value()); + CompleteScheduler(); + delete this; + } + } + + void CompleteScheduler() { + if (!scheduler_completed_ && scheduler_ != nullptr) { + scheduler_completed_ = true; + auto scheduler = std::move(scheduler_); + scheduler->Complete(); + } + } + + void FailBeforeQueue(const std::string& message) { + scheduler_.reset(); + deferred_.Reject(Napi::Error::New(Env(), message).Value()); + delete this; + } + + Napi::Promise::Deferred deferred_; + std::shared_ptr sess_; + foundry_local::Request* req_; + Napi::ObjectReference manager_; + Napi::ObjectReference request_; + std::shared_ptr scheduler_; + Result response_; + std::string err_msg_; + int err_code_ = 0; + bool tagged_ = false; + bool scheduler_completed_ = false; +}; + +template +Napi::Value ProcessRequestOn(Napi::Env env, std::shared_ptr sess, const Napi::Value& request_arg, + Napi::ObjectReference manager_ref, std::shared_ptr scheduler) { foundry_local::Request* req = UnwrapRequest(env, request_arg); if (req == nullptr) return env.Undefined(); // pending exception Napi::ObjectReference req_pin = Napi::Reference::New(request_arg.As(), 1); - using Result = std::shared_ptr; - struct Pins { - Napi::ObjectReference manager; - Napi::ObjectReference request; - }; - auto pins = std::make_shared(Pins{std::move(manager_ref), std::move(req_pin)}); - - return PromiseWorker::Run( - env, - [sess, req, pins]() -> Result { - (void)pins; // keepalive captured by reference count - return std::make_shared(sess->ProcessRequest(*req)); - }, - [](Napi::Env env, Result& resp) -> Napi::Value { return ResponseToJs(env, *resp); }); + return SessionPromiseWorker::Run(env, std::move(sess), req, std::move(manager_ref), std::move(req_pin), + std::move(scheduler)); } // ────────────────────────────────────────────────────────────────────────── @@ -148,6 +289,7 @@ struct StreamCtx { Napi::Promise::Deferred deferred; Napi::ObjectReference manager; Napi::ObjectReference request; + std::shared_ptr scheduler; std::shared_ptr response; std::string err_msg; int err_code = 0; @@ -155,69 +297,122 @@ struct StreamCtx { bool errored = false; }; +class StreamFinalizeGuard { + public: + explicit StreamFinalizeGuard(StreamCtx* ctx) : ctx_(ctx) {} + + ~StreamFinalizeGuard() { + if (ctx_->scheduler != nullptr) { + auto scheduler = std::move(ctx_->scheduler); + scheduler->Complete(); + } + delete ctx_; + } + + private: + StreamCtx* ctx_; +}; + void FinalizeStream(Napi::Env env, void* /*data*/, StreamCtx* ctx) { + StreamFinalizeGuard finalize_guard(ctx); Napi::HandleScope scope(env); - if (ctx->errored) { - if (ctx->tagged) { - Napi::Error err = Napi::Error::New(env, ctx->err_msg); - Napi::Object v = err.Value(); - v.Set("name", Napi::String::New(env, "FoundryLocalError")); - v.Set("code", Napi::Number::New(env, ctx->err_code)); - ctx->deferred.Reject(v); + try { + if (ctx->errored) { + if (ctx->tagged) { + Napi::Error err = Napi::Error::New(env, ctx->err_msg); + Napi::Object v = err.Value(); + v.Set("name", Napi::String::New(env, "FoundryLocalError")); + v.Set("code", Napi::Number::New(env, ctx->err_code)); + ctx->deferred.Reject(v); + } else { + ctx->deferred.Reject(Napi::Error::New(env, ctx->err_msg).Value()); + } + } else if (ctx->response != nullptr) { + ctx->deferred.Resolve(ResponseToJs(env, *ctx->response)); } else { - ctx->deferred.Reject(Napi::Error::New(env, ctx->err_msg).Value()); + // Should not happen: successful path always captures a Response. Guard anyway so we never leave the deferred + // pending. + ctx->deferred.Resolve(env.Undefined()); } - } else if (ctx->response != nullptr) { - ctx->deferred.Resolve(ResponseToJs(env, *ctx->response)); - } else { - // Should not happen: successful path always captures a Response. Guard - // anyway so we never leave the deferred pending. - ctx->deferred.Resolve(env.Undefined()); + } catch (const Napi::Error& e) { + ctx->deferred.Reject(e.Value()); + } catch (const std::exception& e) { + ctx->deferred.Reject(Napi::Error::New(env, e.what()).Value()); + } catch (...) { + ctx->deferred.Reject(Napi::Error::New(env, "Failed to settle native streaming response").Value()); } - delete ctx; } template class StreamWorker : public Napi::AsyncWorker { public: - static Napi::Promise Run(Napi::Env env, SessT* sess, foundry_local::Request* req, - Napi::Function jsCallback, StreamCtx* ctx) { - auto* w = new StreamWorker(env, sess, req, jsCallback, ctx); + static Napi::Promise Run(Napi::Env env, std::shared_ptr sess, foundry_local::Request* req, + Napi::Function jsCallback, StreamCtx* ctx, + std::shared_ptr scheduler) { Napi::Promise p = ctx->deferred.Promise(); - w->Queue(); + StreamWorker* w = nullptr; + try { + w = new StreamWorker(env, std::move(sess), req, jsCallback, ctx, std::move(scheduler)); + } catch (const Napi::Error& e) { + ctx->scheduler.reset(); + ctx->deferred.Reject(e.Value()); + delete ctx; + return p; + } catch (const std::exception& e) { + ctx->scheduler.reset(); + ctx->deferred.Reject(Napi::Error::New(env, e.what()).Value()); + delete ctx; + return p; + } catch (...) { + ctx->scheduler.reset(); + ctx->deferred.Reject(Napi::Error::New(env, "Failed to initialize native streaming request").Value()); + delete ctx; + return p; + } + + try { + if (w->scheduler_ != nullptr) { + w->scheduler_->Enqueue([w] { w->Start(); }); + } else { + w->Start(); + } + } catch (const std::exception& e) { + w->FailBeforeQueue(e.what()); + } catch (...) { + w->FailBeforeQueue("Failed to schedule native streaming request"); + } + return p; } void Execute() override { + bool callback_installed = false; try { auto tsfn = tsfn_; - auto* ctx = ctx_; - sess_->SetStreamingCallback([tsfn, ctx](flStreamingCallbackData data) -> int { - (void)ctx; + sess_->SetStreamingCallback([tsfn](flStreamingCallbackData data) -> int { 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; + if (!foundry_local::detail::item_api()->ItemQueue_TryPop(data.item_queue, &raw) || raw == nullptr) { + return 0; } + + 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; + } + return 0; }); + callback_installed = true; ctx_->response = std::make_shared(sess_->ProcessRequest(*req_)); - // Drop the callback so any stale shared state in the lambda is released - // before the Session is re-used for a follow-up request. - sess_->SetStreamingCallback(nullptr); } catch (const foundry_local::Error& e) { ctx_->errored = true; ctx_->err_code = static_cast(e.Code()); @@ -230,35 +425,94 @@ class StreamWorker : public Napi::AsyncWorker { ctx_->errored = true; ctx_->err_msg = "Unknown native exception"; } + + if (callback_installed) { + try { + sess_->SetStreamingCallback(nullptr); + } catch (const foundry_local::Error& e) { + if (!ctx_->errored) { + ctx_->errored = true; + ctx_->err_code = static_cast(e.Code()); + ctx_->err_msg = e.what(); + ctx_->tagged = true; + } + } catch (const std::exception& e) { + if (!ctx_->errored) { + ctx_->errored = true; + ctx_->err_msg = e.what(); + } + } catch (...) { + if (!ctx_->errored) { + ctx_->errored = true; + ctx_->err_msg = "Failed to clear native streaming callback"; + } + } + } } // Promise resolution happens in FinalizeStream — overriding OnOK/OnError // here only releases the TSFN so its finalizer can run on the JS thread // once all queued item callbacks have drained. - void OnOK() override { tsfn_.Release(); } - void OnError(const Napi::Error& /*unused*/) override { tsfn_.Release(); } + void OnOK() override { ReleaseTsfn(); } + void OnError(const Napi::Error& /*unused*/) override { ReleaseTsfn(); } private: - StreamWorker(Napi::Env env, SessT* sess, foundry_local::Request* req, - Napi::Function jsCallback, StreamCtx* ctx) + StreamWorker(Napi::Env env, std::shared_ptr sess, foundry_local::Request* req, + Napi::Function jsCallback, StreamCtx* ctx, std::shared_ptr scheduler) : Napi::AsyncWorker(env), - sess_(sess), + sess_(std::move(sess)), req_(req), ctx_(ctx), + scheduler_(std::move(scheduler)), tsfn_(Napi::ThreadSafeFunction::New(env, jsCallback, "foundry_local_stream", /*max_queue=*/64, /*threads=*/1, ctx, FinalizeStream, static_cast(nullptr))) {} - SessT* sess_; + void Start() { + try { + Queue(); + } catch (const std::exception& e) { + ctx_->errored = true; + ctx_->err_msg = e.what(); + ReleaseTsfn(); + delete this; + } catch (...) { + ctx_->errored = true; + ctx_->err_msg = "Failed to queue native streaming request"; + ReleaseTsfn(); + delete this; + } + } + + void ReleaseTsfn() { + if (!tsfn_released_) { + tsfn_released_ = true; + tsfn_.Release(); + } + } + + void FailBeforeQueue(const std::string& message) { + scheduler_.reset(); + ctx_->scheduler.reset(); + ctx_->errored = true; + ctx_->err_msg = message; + ReleaseTsfn(); + delete this; + } + + std::shared_ptr sess_; foundry_local::Request* req_; StreamCtx* ctx_; + std::shared_ptr scheduler_; Napi::ThreadSafeFunction tsfn_; + bool tsfn_released_ = false; }; template -Napi::Value ProcessStreamingRequestOn(Napi::Env env, SessT* sess, const Napi::CallbackInfo& info, - Napi::ObjectReference manager_ref) { +Napi::Value ProcessStreamingRequestOn(Napi::Env env, std::shared_ptr sess, const Napi::CallbackInfo& info, + Napi::ObjectReference manager_ref, + std::shared_ptr scheduler) { if (info.Length() < 2 || !info[1].IsFunction()) { Napi::TypeError::New(env, "processStreamingRequest(request: Request, onItem: (item) => void)") .ThrowAsJavaScriptException(); @@ -272,12 +526,14 @@ Napi::Value ProcessStreamingRequestOn(Napi::Env env, SessT* sess, const Napi::Ca auto* ctx = new StreamCtx{Napi::Promise::Deferred::New(env), std::move(manager_ref), std::move(req_pin), + scheduler, nullptr, "", 0, false, false}; - return StreamWorker::Run(env, sess, req, info[1].As(), ctx); + return StreamWorker::Run(env, std::move(sess), req, info[1].As(), ctx, + std::move(scheduler)); } } // namespace @@ -303,7 +559,8 @@ Napi::Function ChatSession::Init(Napi::Env env) { }); } -ChatSession::ChatSession(const Napi::CallbackInfo& info) : Napi::ObjectWrap(info) { +ChatSession::ChatSession(const Napi::CallbackInfo& info) + : Napi::ObjectWrap(info), scheduler_(std::make_shared()) { Napi::Env env = info.Env(); auto* data = env.GetInstanceData(); if (info.Length() != 1 || !info[0].IsObject() || @@ -322,7 +579,7 @@ ChatSession::ChatSession(const Napi::CallbackInfo& info) : Napi::ObjectWrap(*native); + impl_ = std::make_shared(*native); } catch (const foundry_local::Error& e) { ThrowFoundryLocalError(env, static_cast(e.Code()), e.what()); return; @@ -350,14 +607,18 @@ Napi::Value ChatSession::ProcessRequest(const Napi::CallbackInfo& info) { return env.Undefined(); } Napi::ObjectReference owner = Napi::Reference::New(manager_.Value(), 1); - return ProcessRequestOn(env, impl_.get(), info[0], std::move(owner)); + auto impl = impl_; + auto scheduler = scheduler_; + return ProcessRequestOn(env, std::move(impl), info[0], std::move(owner), std::move(scheduler)); } Napi::Value ChatSession::ProcessStreamingRequest(const Napi::CallbackInfo& info) { Napi::Env env = info.Env(); if (ThrowIfDisposed(env)) return env.Undefined(); Napi::ObjectReference owner = Napi::Reference::New(manager_.Value(), 1); - return ProcessStreamingRequestOn(env, impl_.get(), info, std::move(owner)); + auto impl = impl_; + auto scheduler = scheduler_; + return ProcessStreamingRequestOn(env, std::move(impl), info, std::move(owner), std::move(scheduler)); } Napi::Value ChatSession::SetOptions(const Napi::CallbackInfo& info) { @@ -504,7 +765,7 @@ EmbeddingsSession::EmbeddingsSession(const Napi::CallbackInfo& info) return; } try { - impl_ = std::make_unique(*native); + impl_ = std::make_shared(*native); } catch (const foundry_local::Error& e) { ThrowFoundryLocalError(env, static_cast(e.Code()), e.what()); return; @@ -532,7 +793,8 @@ Napi::Value EmbeddingsSession::ProcessRequest(const Napi::CallbackInfo& info) { return env.Undefined(); } Napi::ObjectReference owner = Napi::Reference::New(manager_.Value(), 1); - return ProcessRequestOn(env, impl_.get(), info[0], std::move(owner)); + auto impl = impl_; + return ProcessRequestOn(env, std::move(impl), info[0], std::move(owner), nullptr); } Napi::Value EmbeddingsSession::SetOptions(const Napi::CallbackInfo& info) { @@ -579,7 +841,7 @@ Napi::Function AudioSession::Init(Napi::Env env) { } AudioSession::AudioSession(const Napi::CallbackInfo& info) - : Napi::ObjectWrap(info) { + : Napi::ObjectWrap(info), scheduler_(std::make_shared()) { Napi::Env env = info.Env(); auto* data = env.GetInstanceData(); if (info.Length() != 1 || !info[0].IsObject() || @@ -598,7 +860,7 @@ AudioSession::AudioSession(const Napi::CallbackInfo& info) return; } try { - impl_ = std::make_unique(*native); + impl_ = std::make_shared(*native); } catch (const foundry_local::Error& e) { ThrowFoundryLocalError(env, static_cast(e.Code()), e.what()); return; @@ -626,14 +888,18 @@ Napi::Value AudioSession::ProcessRequest(const Napi::CallbackInfo& info) { return env.Undefined(); } Napi::ObjectReference owner = Napi::Reference::New(manager_.Value(), 1); - return ProcessRequestOn(env, impl_.get(), info[0], std::move(owner)); + auto impl = impl_; + auto scheduler = scheduler_; + return ProcessRequestOn(env, std::move(impl), info[0], std::move(owner), std::move(scheduler)); } Napi::Value AudioSession::ProcessStreamingRequest(const Napi::CallbackInfo& info) { Napi::Env env = info.Env(); if (ThrowIfDisposed(env)) return env.Undefined(); Napi::ObjectReference owner = Napi::Reference::New(manager_.Value(), 1); - return ProcessStreamingRequestOn(env, impl_.get(), info, std::move(owner)); + auto impl = impl_; + auto scheduler = scheduler_; + return ProcessStreamingRequestOn(env, std::move(impl), info, std::move(owner), std::move(scheduler)); } Napi::Value AudioSession::SetOptions(const Napi::CallbackInfo& info) { diff --git a/sdk_v2/js/native/src/session.h b/sdk_v2/js/native/src/session.h index 2b1db7b86..b8a4f0051 100644 --- a/sdk_v2/js/native/src/session.h +++ b/sdk_v2/js/native/src/session.h @@ -6,9 +6,9 @@ // Surface: // * new ChatSession(model) — sync construction; underlying // flSession_Create is fast. -// * session.processRequest(request) -> Promise (PromiseWorker) +// * session.processRequest(request) -> Promise — scheduled on a per-session worker queue. // * session.processStreamingRequest(request, onItem) -> Promise — streaming bridge via -// Napi::ThreadSafeFunction; resolves with the terminal Response after every item callback drains. +// Napi::ThreadSafeFunction; resolves with the terminal Response after every queued JS callback runs. // The JS layer wraps this in an AsyncIterable whose `.response` promise carries the resolved value. // * session.setOptions(kvp) — session-level options applied to subsequent sends. // * ChatSession adds: turnCount(), undoTurns(count), addToolDefinition({...}). @@ -21,17 +21,33 @@ // // Lifetime: the ChatSession pins the parent Manager via an ObjectReference so // the underlying foundry_local::Model the C++ Session captured can't be -// released out from under it. +// released out from under it. Session implementations are shared with queued +// workers so dispose() can detach the wrapper without blocking the JS thread +// or invalidating work that was already accepted. #pragma once #include #include +#include +#include #include namespace foundry_local_node { +class SessionScheduler { + public: + void Enqueue(std::function start); + void Complete(); + + private: + void StartNext(); + + std::deque> pending_; + bool running_ = false; +}; + class ChatSession : public Napi::ObjectWrap { public: static Napi::Function Init(Napi::Env env); @@ -51,8 +67,9 @@ class ChatSession : public Napi::ObjectWrap { bool ThrowIfDisposed(Napi::Env env); - std::unique_ptr impl_; + std::shared_ptr impl_; Napi::ObjectReference manager_; + std::shared_ptr scheduler_; }; // Napi::ObjectWrap over foundry_local::EmbeddingsSession. @@ -82,7 +99,7 @@ class EmbeddingsSession : public Napi::ObjectWrap { bool ThrowIfDisposed(Napi::Env env); - std::unique_ptr impl_; + std::shared_ptr impl_; Napi::ObjectReference manager_; }; @@ -111,8 +128,9 @@ class AudioSession : public Napi::ObjectWrap { bool ThrowIfDisposed(Napi::Env env); - std::unique_ptr impl_; + std::shared_ptr impl_; Napi::ObjectReference manager_; + std::shared_ptr scheduler_; }; } // namespace foundry_local_node diff --git a/sdk_v2/js/src/request.ts b/sdk_v2/js/src/request.ts index 801f0399c..77e7fe16a 100644 --- a/sdk_v2/js/src/request.ts +++ b/sdk_v2/js/src/request.ts @@ -74,10 +74,9 @@ export class Request { } /** - * Cancel an in-flight request. Safe to call at any time — if the request - * is not currently being processed by a session, this is a no-op. - * Cancellation makes the matching `Session.processRequest()` reject with a - * `FoundryLocalError` whose `code === FlErrorCode.OperationCancelled`. + * Cancel this request's invocation only while it is inside native `Session::ProcessRequest`. Calling this while the + * request is idle, waiting in the addon's per-session queue, or already completed has no effect. Active cancellation + * makes the matching `Session.processRequest()` reject with `code === FlErrorCode.OperationCancelled`. */ cancel(): void { this.#native.cancel(); diff --git a/sdk_v2/js/src/session.ts b/sdk_v2/js/src/session.ts index 932c6638d..09053de20 100644 --- a/sdk_v2/js/src/session.ts +++ b/sdk_v2/js/src/session.ts @@ -32,9 +32,10 @@ import type { Response } from "./response.js"; /** Options accepted by streaming Session APIs. */ export interface StreamOptions { /** - * Optional cancellation signal. When the signal aborts, the underlying - * `Request` is cancelled and the async iterator rejects with an `Error` - * whose `name === "AbortError"` (mirroring the Web/Node standard). + * Optional cancellation signal. Once native processing is active, aborting cancels the `Request` and the iterator + * rejects with an `Error` whose `name === "AbortError"`. An abort while work is only waiting in the native session + * FIFO does not prevent that work from running. A signal already aborted when this method is called rejects before + * native work is submitted. */ readonly signal?: AbortSignal; } @@ -45,10 +46,10 @@ export interface StreamOptions { * once the native call completes — carrying stop reason, usage, and any * non-streamed items (e.g. the final aggregated text item). * - * `response` settles after the iterator finishes draining. It rejects with - * the same error the iterator would throw (including `AbortError` when the - * stream is cancelled, and `OperationCancelled` when the consumer breaks - * early without an `AbortSignal`). + * `response` settles after the native call completes and all queued item callbacks have run. Breaking iteration early + * requests cancellation of an active native invocation, but native completion can win that race. A request still + * waiting in the addon's per-session queue is unaffected. When cancellation wins, `response` rejects with + * `AbortError` for an aborted signal or `OperationCancelled` for an early break. */ export interface StreamingResponse extends AsyncIterable { readonly response: Promise; @@ -111,12 +112,12 @@ function modelToNativeAudioSession(model: IModel): NativeAudioSession { * Drive a native streaming session and yield each item to the consumer. * Handles backpressure (the JS-side queue grows; the native TSFN backpressure * caps producer-side queueing), abort signal wiring, error mapping, and - * deterministic cleanup on early break. + * cleanup on early break. * * The native call starts eagerly so the returned `response` promise is * meaningful even if the caller never iterates (e.g. awaits `.response` - * directly). The promise settles only after the consumer has fully drained - * the iterator, mirroring native finalize-on-drain semantics. + * directly). The promise settles after the native call and queued item callbacks complete, independently of whether the + * consumer drains the JS iterator. */ function streamItems(native: NativeSession, request: Request, signal: AbortSignal | undefined): StreamingResponse { const queue: Item[] = []; @@ -276,10 +277,11 @@ export abstract class Session { * `Response` (stop reason, usage, aggregate text item, etc.) once the * native call completes. * - * Cancellation: pass `{ signal }`; aborting the signal cancels the native - * request and causes the iterator to throw an `Error` with - * `name === "AbortError"`. Breaking out of the `for await` loop also - * cancels the underlying request. + * Cancellation: pass `{ signal }`; aborting while the invocation is active cancels the native request and causes the + * iterator to throw an `Error` with `name === "AbortError"`. Aborting while work is only waiting in the addon's + * per-session queue does not prevent execution. A signal already aborted at call time rejects before submission. + * Breaking out of the `for await` loop similarly requests active cancellation. If native completion wins the race, + * `response` resolves normally; otherwise it rejects with `OperationCancelled`. * * Non-cancellation failures throw a `FoundryLocalError`. */ diff --git a/sdk_v2/js/test/items.test.ts b/sdk_v2/js/test/items.test.ts index 53e0003d0..2ffa6f1b0 100644 --- a/sdk_v2/js/test/items.test.ts +++ b/sdk_v2/js/test/items.test.ts @@ -180,7 +180,7 @@ describeIfBuilt("Request round-trip through the native layer", () => { expect(req.itemCount).toBe(3); }); - it("cancel on an unattached request is a no-op", () => { + it("cancel while idle is an accepted no-op", () => { const req = new Request(); expect(() => req.cancel()).not.toThrow(); }); diff --git a/sdk_v2/js/test/streaming.test.ts b/sdk_v2/js/test/streaming.test.ts index dff5127e0..58d9e688e 100644 --- a/sdk_v2/js/test/streaming.test.ts +++ b/sdk_v2/js/test/streaming.test.ts @@ -47,6 +47,7 @@ function countUkTokens(text: string): number { // Used by the multi-turn streaming test: a context-dependent follow-up // ("What is the capital of each?") should mention the UK capitals. const UK_CAPITAL_TOKENS = ["london", "edinburgh", "cardiff", "belfast"] as const; +const PRIMARY_COLOR_TOKENS = ["red", "blue", "yellow"] as const; function countUkCapitalTokens(text: string): number { const lower = text.toLowerCase(); @@ -108,27 +109,28 @@ describe.skipIf(!haveTestModelCache)("ChatSession.processStreamingRequest (real ); it( - "concatenated streamed text contains the expected answer content", - async () => { - if (session === undefined) throw new Error("fixture missing"); - let text = ""; - for await (const item of session.processStreamingRequest(buildPrompt())) { - text += extractText(item); - } - expect(countUkTokens(text)).toBeGreaterThanOrEqual(2); - }, - 2 * 60_000, - ); - - it( - "early break cancels the stream cleanly and the session remains usable", + "early break requests cancellation while permitting prior native completion", async () => { if (session === undefined) throw new Error("fixture missing"); + const stream = session.processStreamingRequest(buildPrompt()); let count = 0; - for await (const _item of session.processStreamingRequest(buildPrompt())) { + for await (const _item of stream) { count++; if (count >= 1) break; } + const outcome = await stream.response.then( + (response) => ({ response, error: null }), + (error: unknown) => ({ response: null, error }), + ); + if (outcome.error !== null) { + expect(outcome.error).toMatchObject({ + name: "FoundryLocalError", + code: FlErrorCode.OperationCancelled, + }); + } else { + expect(outcome.response.finishReason).not.toBe("none"); + } + // After the break the session should accept a follow-up send. const resp = await session.processRequest( new Request() @@ -237,9 +239,7 @@ describe.skipIf(!haveTestModelCache)("ChatSession.processStreamingRequest (real expect(["stop", "length", "toolCalls", "error", "none"]).toContain(resp.finishReason); expect(resp.usage.promptTokens).toBeGreaterThan(0); expect(resp.usage.completionTokens).toBeGreaterThan(0); - expect(resp.usage.totalTokens).toBeGreaterThanOrEqual( - resp.usage.promptTokens + resp.usage.completionTokens, - ); + expect(resp.usage.totalTokens).toBeGreaterThanOrEqual(resp.usage.promptTokens + resp.usage.completionTokens); // The Response's text should match what we accumulated from the stream // (modulo possible model post-processing — assert non-empty overlap on // the boundary tokens rather than strict equality). @@ -250,6 +250,75 @@ describe.skipIf(!haveTestModelCache)("ChatSession.processStreamingRequest (real 3 * 60_000, ); + it( + "serializes overlapping streams without replacing either callback", + async () => { + if (session === undefined) throw new Error("fixture missing"); + const first = session.processStreamingRequest(buildPrompt()); + const second = session.processStreamingRequest( + new Request() + .addItem(Item.userMessage("Name three primary colors.")) + .setOptions({ search: { maxOutputTokens: 64, temperature: 0 } }), + ); + + const collect = async (stream: AsyncIterable): Promise => { + const items: Item[] = []; + for await (const item of stream) items.push(item); + return items; + }; + const [firstItems, secondItems, firstResponse, secondResponse] = await Promise.all([ + collect(first), + collect(second), + first.response, + second.response, + ]); + + expect(firstItems.length).toBeGreaterThan(0); + expect(secondItems.length).toBeGreaterThan(0); + expect(countUkTokens(firstItems.map(extractText).join(""))).toBeGreaterThanOrEqual(2); + expect( + PRIMARY_COLOR_TOKENS.filter((token) => secondItems.map(extractText).join("").toLowerCase().includes(token)) + .length, + ).toBeGreaterThanOrEqual(2); + expect(firstResponse.finishReason).not.toBe("none"); + expect(secondResponse.finishReason).not.toBe("none"); + }, + 4 * 60_000, + ); + + it( + "finishes accepted queued work after dispose and rejects future work", + async () => { + if (session === undefined) throw new Error("fixture missing"); + const active = session.processStreamingRequest(buildPrompt()); + const queued = session.processRequest( + new Request() + .addItem(Item.userMessage("Reply with the single word 'ok'.")) + .setOptions({ search: { maxOutputTokens: 4, temperature: 0 } }), + ); + + session.dispose(); + expect(session.disposed).toBe(true); + + await expect( + session.processRequest(new Request().addItem(Item.userMessage("This must not be accepted."))), + ).rejects.toMatchObject({ + name: "FoundryLocalError", + code: FlErrorCode.InvalidUsage, + }); + + const activeItems: Item[] = []; + for await (const item of active) activeItems.push(item); + const [activeResponse, queuedResponse] = await Promise.all([active.response, queued]); + + expect(activeItems.length).toBeGreaterThan(0); + expect(activeResponse.finishReason).not.toBe("none"); + expect(queuedResponse.finishReason).not.toBe("none"); + expect(queuedResponse.output.map(extractText).join("").toLowerCase()).toContain("ok"); + }, + 4 * 60_000, + ); + it( "stream.response resolves without iteration (eager native start)", async () => { @@ -264,47 +333,56 @@ describe.skipIf(!haveTestModelCache)("ChatSession.processStreamingRequest (real 3 * 60_000, ); - it( - "stream.response rejects with AbortError when pre-aborted", - async () => { - if (session === undefined) throw new Error("fixture missing"); - const ctrl = new AbortController(); - ctrl.abort(); - const stream = session.processStreamingRequest(buildPrompt(), { signal: ctrl.signal }); - await expect(stream.response).rejects.toMatchObject({ name: "AbortError" }); - }, - 60_000, - ); + it("stream.response rejects with AbortError when pre-aborted", async () => { + if (session === undefined) throw new Error("fixture missing"); + const ctrl = new AbortController(); + ctrl.abort(); + const stream = session.processStreamingRequest(buildPrompt(), { signal: ctrl.signal }); + await expect(stream.response).rejects.toMatchObject({ name: "AbortError" }); + }, 60_000); it( - "stream.response resolves with finishReason='none' when request.cancel() is called mid-stream", + "request.cancel yields OperationCancelled unless native completion wins", async () => { if (session === undefined) throw new Error("fixture missing"); - // Native ChatSession::ProcessRequestImpl treats Request::Cancel as a - // graceful early-exit: the generation loop breaks, the generator is - // rewound, and ProcessGeneratedOutput sets finish_reason=NONE. The - // call returns a normal Response — it does NOT throw OperationCancelled - // (that exception is only raised on the pre-call path). The JS layer - // must surface that same contract: `.response` resolves with a - // FinishReason of "none". const req = new Request() .addItem(Item.systemMessage("You are verbose.")) .addItem(Item.userMessage("Write a 500-word essay about the history of bread.")) .setOptions({ search: { maxOutputTokens: 1024, temperature: 0 } }); const stream = session.processStreamingRequest(req); - let observed = 0; - for await (const _item of stream) { - observed++; - if (observed >= 1) { - req.cancel(); - break; + const iteration = async (): Promise => { + let observed = 0; + for await (const _item of stream) { + if (++observed >= 1) { + req.cancel(); + } } + }; + + const iterationOutcome = await iteration().then( + () => null, + (error: unknown) => error, + ); + const responseOutcome = await stream.response.then( + (response) => ({ response, error: null }), + (error: unknown) => ({ response: null, error }), + ); + + if (responseOutcome.error !== null) { + expect(iterationOutcome).toMatchObject({ + name: "FoundryLocalError", + code: FlErrorCode.OperationCancelled, + }); + expect(responseOutcome.error).toMatchObject({ + name: "FoundryLocalError", + code: FlErrorCode.OperationCancelled, + }); + expect(session.turnCount).toBe(0); + } else { + expect(iterationOutcome).toBeNull(); + expect(responseOutcome.response.finishReason).not.toBe("none"); + expect(session.turnCount).toBe(1); } - const resp = await stream.response; - expect(resp.finishReason).toBe("none"); - // History must NOT be committed on cancel — CommitTurn is skipped - // when request.canceled is true (see ChatSession::ProcessRequestImpl). - expect(session.turnCount).toBe(0); }, 3 * 60_000, ); @@ -334,9 +412,11 @@ describe.skipIf(!haveTestModelCache)("ChatSession.processStreamingRequest (real }); const req = new Request() - .addItem(Item.systemMessage( - "You are a helpful AI assistant. If necessary, you can use any provided tools to answer the question.", - )) + .addItem( + Item.systemMessage( + "You are a helpful AI assistant. If necessary, you can use any provided tools to answer the question.", + ), + ) .addItem(Item.userMessage("What is the answer to 7 multiplied by 6?")) .setOptions({ search: { temperature: 0, maxOutputTokens: 256 }, @@ -362,7 +442,8 @@ describe.skipIf(!haveTestModelCache)("ChatSession.processStreamingRequest (real expect(itemCount).toBeGreaterThan(0); expect(streamedToolCalls.length).toBeGreaterThanOrEqual(1); - const streamed = streamedToolCalls[0]!; + const streamed = streamedToolCalls[0]; + if (streamed === undefined) throw new Error("expected a streamed tool call"); expect(streamed.name).toBe("multiply_numbers"); expect(streamed.arguments.length).toBeGreaterThan(0); expect(streamed.callId.length).toBeGreaterThan(0); @@ -373,13 +454,11 @@ describe.skipIf(!haveTestModelCache)("ChatSession.processStreamingRequest (real const resp = await stream.response; expect(resp.finishReason).toBe("toolCalls"); - const finalToolCall = resp.output.find((it): it is Extract => - it.type === "toolCall", - ); - expect(finalToolCall).toBeDefined(); - expect(finalToolCall!.name).toBe(streamed.name); - expect(finalToolCall!.arguments).toBe(streamed.arguments); - expect(finalToolCall!.callId).toBe(streamed.callId); + const finalToolCall = resp.output.find((it): it is Extract => it.type === "toolCall"); + if (finalToolCall === undefined) throw new Error("expected a final tool call"); + expect(finalToolCall.name).toBe(streamed.name); + expect(finalToolCall.arguments).toBe(streamed.arguments); + expect(finalToolCall.callId).toBe(streamed.callId); }, 3 * 60_000, ); diff --git a/sdk_v2/python/src/foundry_local_sdk/request.py b/sdk_v2/python/src/foundry_local_sdk/request.py index fe6879e81..c7d8d9a62 100644 --- a/sdk_v2/python/src/foundry_local_sdk/request.py +++ b/sdk_v2/python/src/foundry_local_sdk/request.py @@ -103,7 +103,11 @@ def set_options(self, options: "RequestOptions") -> "Request": return self def cancel(self) -> None: - """Signal cancellation for an in-flight request.""" + """Cancel this request. + + Only an invocation currently being processed is affected. Calling this + while the request is idle or after completion has no effect. + """ self._check_open() from foundry_local_sdk._native.api import api