.

src/Runner.Common/RunServer.cs

@@ -0,0 +1,76 @@
+using System;
+using System.Collections.Generic;
+using System.Runtime.Serialization;
+using System.Threading;
+using System.Threading.Tasks;
+using GitHub.DistributedTask.Pipelines;
+using GitHub.DistributedTask.WebApi;
+using GitHub.Runner.Common.Util;
+using GitHub.Runner.Sdk;
+using GitHub.Services.Common;
+using GitHub.Services.WebApi;
+
+namespace GitHub.Runner.Common
+{
+    [ServiceLocator(Default = typeof(RunServer))]
+    public interface IRunServer : IRunnerService
+    {
+        Task ConnectAsync(Uri serverUrl, VssCredentials credentials);
+
+        Task<AgentJobRequestMessage> GetJobMessageAsync(string id);
+    }
+
+    public sealed class RunServer : RunnerService, IRunServer
+    {
+        private bool _hasConnection;
+        private VssConnection _connection;
+        private TaskAgentHttpClient _taskAgentClient;
+
+        public async Task ConnectAsync(Uri serverUrl, VssCredentials credentials)
+        {
+            _connection = await EstablishVssConnection(serverUrl, credentials, TimeSpan.FromSeconds(100));
+            _taskAgentClient = _connection.GetClient<TaskAgentHttpClient>();
+            _hasConnection = true;
+        }
+
+        private async Task<VssConnection> EstablishVssConnection(Uri serverUrl, VssCredentials credentials, TimeSpan timeout)
+        {
+            Trace.Info($"EstablishVssConnection");
+            Trace.Info($"Establish connection with {timeout.TotalSeconds} seconds timeout.");
+            int attemptCount = 5;
+            while (attemptCount-- > 0)
+            {
+                var connection = VssUtil.CreateConnection(serverUrl, credentials, timeout: timeout);
+                try
+                {
+                    await connection.ConnectAsync();
+                    return connection;
+                }
+                catch (Exception ex) when (attemptCount > 0)
+                {
+                    Trace.Info($"Catch exception during connect. {attemptCount} attempt left.");
+                    Trace.Error(ex);
+
+                    await HostContext.Delay(TimeSpan.FromMilliseconds(100), CancellationToken.None);
+                }
+            }
+
+            // should never reach here.
+            throw new InvalidOperationException(nameof(EstablishVssConnection));
+        }
+
+        private void CheckConnection()
+        {
+            if (!_hasConnection)
+            {
+                throw new InvalidOperationException($"SetConnection");
+            }
+        }
+
+        public Task<AgentJobRequestMessage> GetJobMessageAsync(string id)
+        {
+            CheckConnection();
+            return _taskAgentClient.GetJobMessageAsync(id);
+        }
+    }
+}

src/Runner.Listener/Runner.cs

@@ -13,6 +13,10 @@ using GitHub.Runner.Sdk;
 using System.Linq;
 using GitHub.Runner.Listener.Check;
 using System.Collections.Generic;
+using System.Runtime.Serialization;
+using System.Net;
+using System.Net.Http;
+using System.Net.Http.Headers;
 
 namespace GitHub.Runner.Listener
 {
@@ -457,6 +461,35 @@ namespace GitHub.Runner.Listener
                                     }
                                 }
                             }
+                            // Broker flow
+                            else if (string.Equals(message.MessageType, JobRequestMessageTypes.RunnerJobRequest, StringComparison.OrdinalIgnoreCase))
+                            {
+                                if (autoUpdateInProgress || runOnceJobReceived)
+                                {
+                                    skipMessageDeletion = true;
+                                    Trace.Info($"Skip message deletion for job request message '{message.MessageId}'.");
+                                }
+                                else
+                                {
+                                    var messageRef = StringUtil.ConvertFromJson<RunnerJobRequestRef>(message.Body);
+
+                                    // Create connection
+                                    var credMgr = HostContext.GetService<ICredentialManager>();
+                                    var creds = credMgr.LoadCredentials();
+
+                                    // todo: add retries
+                                    var runServer = HostContext.CreateService<IRunServer>();
+                                    await runServer.ConnectAsync(new Uri(settings.ServerUrl), creds);
+                                    var jobMessage = await runServer.GetJobMessageAsync(messageRef.RunnerRequestId);
+
+                                    jobDispatcher.Run(jobMessage, runOnce);
+                                    if (runOnce)
+                                    {
+                                        Trace.Info("One time used runner received job message.");
+                                        runOnceJobReceived = true;
+                                    }
+                                }
+                            }
                             else if (string.Equals(message.MessageType, JobCancelMessage.MessageType, StringComparison.OrdinalIgnoreCase))
                             {
                                 var cancelJobMessage = JsonUtility.FromString<JobCancelMessage>(message.Body);

src/Runner.Listener/RunnerJobRequestRef.cs

@@ -0,0 +1,13 @@
+using System.Runtime.Serialization;
+
+namespace GitHub.Runner.Listener
+{
+    [DataContract]
+    public sealed class RunnerJobRequestRef
+    {
+        [DataMember(Name = "id")]
+        public string Id { get; set; }
+        [DataMember(Name = "runner_request_id")]
+        public string RunnerRequestId { get; set; }
+    }
+}

src/Runner.Sdk/Util/VssUtil.cs

@@ -57,6 +57,7 @@ namespace GitHub.Runner.Sdk
                 settings.SendTimeout = TimeSpan.FromSeconds(Math.Min(Math.Max(httpRequestTimeoutSeconds, 100), 1200));
             }
 
+            settings.AllowAutoRedirect = true;
 
             // Remove Invariant from the list of accepted languages.
             //

src/Sdk/DTGenerated/Generated/TaskAgentHttpClientBase.cs

@@ -27,6 +27,7 @@ using System.Net.Http.Headers;
 using System.Net.Http.Formatting;
 using System.Threading;
 using System.Threading.Tasks;
+using GitHub.DistributedTask.Pipelines;
 using GitHub.Services.Common;
 using GitHub.Services.WebApi;
 

src/Sdk/DTWebApi/WebApi/JobRequestMessageTypes.cs

@@ -5,5 +5,6 @@ namespace GitHub.DistributedTask.WebApi
     public static class JobRequestMessageTypes
     {
         public const String PipelineAgentJobRequest = "PipelineAgentJobRequest";
+        public const String RunnerJobRequest = "RunnerJobRequest";
     }
 }

src/Sdk/DTWebApi/WebApi/TaskAgentHttpClient.cs

@@ -141,6 +141,24 @@ namespace GitHub.DistributedTask.WebApi
             return ReplaceAgentAsync(poolId, agent.Id, agent, userState, cancellationToken);
         }
 
+        public Task<Pipelines.AgentJobRequestMessage> GetJobMessageAsync(
+            string messageId,
+            object userState = null,
+            CancellationToken cancellationToken = default)
+        {
+            HttpMethod httpMethod = new HttpMethod("GET");
+            Guid locationId = new Guid("25adab70-1379-4186-be8e-b643061ebe3a");
+            object routeValues = new { messageId = messageId };
+
+            return SendAsync<Pipelines.AgentJobRequestMessage>(
+                httpMethod,
+                locationId,
+                routeValues: routeValues,
+                version: new ApiResourceVersion(1.0, 1),
+                userState: userState,
+                cancellationToken: cancellationToken);
+        }
+
         protected Task<T> SendAsync<T>(
             HttpMethod method,
             Guid locationId,