Add ability to stream chat responses

What does this MR do and why?

This adds the capabilities to stream messages for GitLab Duo chat.

The streamed chunks get send over aiCompletionResponse but the order is not guaranteed on the client. Each chunk has a chunkId which has a guaranteed order, starting with 1.

The client_subscription_id is meant for the client to identify which mutation belongs to which subscription response. However, chat is a special feature where we want to sync responses between multiple chat instances of the same user.

This means we want to have two different subscriptions for the same request: 1. User-specific to sync between multiple clients 2. Request-specific for streaming

By not streaming the response when a client_subscription_id is not present, we also make sure that older clients still work when the stream_gitlab_duo is enabled. Otherwise, they would receive a streamed response on a subscription on user_id and resource_id.

NOTE: Related FE MR by @dmishunov !130347 (merged)

Screenshots or screen recordings

The following only works with the FE and graphql changes. There is an MR that combines both: !130415 (closed)

Without FF	With FF
streaming-example-no-ff	streaming-example

Before	After

How to set up and validate locally

1. Have GitLab Duo and all AI features set up
2. Enable the FF stream_gitlab_duo
3. Checkout !130415 (closed) as it combines this MR with the FE changes and updates the GraphQL subscriptions

MR acceptance checklist

This checklist encourages us to confirm any changes have been analyzed to reduce risks in quality, performance, reliability, security, and maintainability.

• I have evaluated the MR acceptance checklist for this MR.

ee/lib/gitlab/llm/chain/agents/zero_shot/executor.rb

@@ -24,18 +24,21 @@ class Executor
             # @param [String] user_input - a question from a user
             # @param [Array<Tool>] tools - an array of Tools defined in the tools module.
             # @param [GitlabContext] context - Gitlab context containing useful context information
-            def initialize(user_input:, tools:, context:, response_handler:)
+            # @param [ResponseService] response_handler - Handles returning the response to the client
+            # @param [ResponseService] stream_response_handler - Handles streaming chunks to the client
+            def initialize(user_input:, tools:, context:, response_handler:, stream_response_handler: nil)
               @user_input = user_input
               @tools = tools
               @context = context
               @iterations = 0
               @logger = Gitlab::Llm::Logger.build
               @response_handler = response_handler
+              @stream_response_handler = stream_response_handler
             end
 
             def execute
               MAX_ITERATIONS.times do
-                thought = if Feature.enabled?(:stream_gitlab_duo, context.current_user)
+                thought = if stream_response_handler && Feature.enabled?(:stream_gitlab_duo, context.current_user)
                             execute_streamed_request
                           else
                             request
@@ -85,7 +88,7 @@ def execute_streamed_request
                 chunk = streamed_answer.next_chunk(content)
 
                 if chunk
-                  response_handler.execute(
+                  stream_response_handler.execute(
                     response: Gitlab::Llm::Chain::PlainResponseModifier.new(content),
                     options: {
                       cache_response: false,
@@ -101,7 +104,7 @@ def tools_cycle?
               context.tools_used.size != context.tools_used.uniq.size
             end
 
-            attr_reader :logger
+            attr_reader :logger, :stream_response_handler
 
             # This method should not be memoized because the input variables change over time
             def base_prompt