V3

src/ConductorSharp.Client/ConductorSharp.Client.csproj

@@ -6,7 +6,7 @@
 		<Authors>Codaxy</Authors>
 		<Company>Codaxy</Company>
 		<PackageId>ConductorSharp.Client</PackageId>
-		<Version>3.7.2</Version>
+		<Version>3.8.1</Version>
 		<Description>Client library for Netflix Conductor, with some additional quality of life features.</Description>
 		<RepositoryUrl>https://github.com/codaxy/conductor-sharp</RepositoryUrl>
 		<PackageTags>netflix;conductor</PackageTags>

src/ConductorSharp.Engine/AssemblyInfo.cs

@@ -0,0 +1,3 @@
+using System.Runtime.CompilerServices;
+
+[assembly: InternalsVisibleTo("ConductorSharp.Engine.Tests")]

src/ConductorSharp.Engine/ConductorSharp.Engine.csproj

@@ -6,7 +6,7 @@
     <Authors>Codaxy</Authors>
 	<Company>Codaxy</Company>
 	<PackageId>ConductorSharp.Engine</PackageId>
-	<Version>3.7.2</Version>
+	<Version>3.8.1</Version>
     <Description>Client library for Netflix Conductor, with some additional quality of life features.</Description>
 	<RepositoryUrl>https://github.com/codaxy/conductor-sharp</RepositoryUrl>
 	<PackageTags>netflix;conductor</PackageTags>

src/ConductorSharp.Engine/Exceptions/StructuredErrorException.cs

@@ -0,0 +1,75 @@
+using System;
+
+namespace ConductorSharp.Engine.Exceptions
+{
+    /// <summary>
+    /// Thrown by a worker to attach a structured, sanitized error classification to the failed task's output.
+    /// When caught by the execution manager, the <see cref="Code"/>/<see cref="Reason"/>/<see cref="ReferenceError"/>
+    /// and the diagnostic message are serialized under the <c>structured_error</c> output key (see
+    /// <see cref="ConductorSharp.Engine.Util.StructuredErrorSerializer"/>), in addition to the plain
+    /// <c>error_message</c>, so downstream consumers can read a stable classification without parsing free-text
+    /// reasons. Plain exceptions are unaffected and keep producing only <c>error_message</c>.
+    /// </summary>
+    /// <remarks>
+    /// There is deliberately no <c>Message</c> property here: the diagnostic message is carried by the inherited
+    /// <see cref="Exception.Message"/>, which the message-taking constructors set. When no message is supplied it
+    /// falls back to <see cref="Reason"/>, matching the behaviour of the original constructors.
+    /// </remarks>
+    public class StructuredErrorException : Exception
+    {
+        /// <summary>Stable, opaque classification code. Consumers map this to a failure response.</summary>
+        public string Code { get; }
+
+        /// <summary>Short, stable, sanitized reason. Safe to surface across a layer boundary.</summary>
+        public string Reason { get; }
+
+        /// <summary>Optional URI pointing at the entity where the failure originated (drill-down link).</summary>
+        public string ReferenceError { get; }
+
+        public StructuredErrorException(string code, string reason, string referenceError = null)
+            : base(reason)
+        {
+            Code = code;
+            Reason = reason;
+            ReferenceError = referenceError;
+        }
+
+        public StructuredErrorException(string code, string reason, string referenceError, Exception innerException)
+            : base(reason, innerException)
+        {
+            Code = code;
+            Reason = reason;
+            ReferenceError = referenceError;
+        }
+
+        /// <summary>
+        /// Declares a diagnostic <paramref name="message"/> distinct from the short, stable
+        /// <paramref name="reason"/>. Pass <c>null</c> for <paramref name="message"/> to fall back to the reason,
+        /// and <c>null</c> for <paramref name="referenceError"/> when there is no entity to drill down into.
+        /// </summary>
+        /// <remarks>
+        /// <paramref name="message"/> trails <paramref name="referenceError"/> rather than following
+        /// <paramref name="reason"/> on purpose. Overload resolution cannot choose between
+        /// this constructor and the <see cref="Exception"/> one when the fourth argument is an untyped <c>null</c>,
+        /// so the nullable parameter is placed third, where it is typed the same either way. The only call this
+        /// leaves ambiguous is <c>(code, reason, referenceError, null)</c> — a declared-but-null inner exception,
+        /// which the three-argument constructor already expresses. Disambiguate with a named argument if needed.
+        /// </remarks>
+        public StructuredErrorException(string code, string reason, string referenceError, string message)
+            : base(message ?? reason)
+        {
+            Code = code;
+            Reason = reason;
+            ReferenceError = referenceError;
+        }
+
+        /// <inheritdoc cref="StructuredErrorException(string, string, string, string)"/>
+        public StructuredErrorException(string code, string reason, string referenceError, string message, Exception innerException)
+            : base(message ?? reason, innerException)
+        {
+            Code = code;
+            Reason = reason;
+            ReferenceError = referenceError;
+        }
+    }
+}

src/ConductorSharp.Engine/ExecutionManager.cs

@@ -8,9 +8,11 @@
 using ConductorSharp.Client.Generated;
 using ConductorSharp.Client.Service;
 using ConductorSharp.Client.Util;
+using ConductorSharp.Engine.Exceptions;
 using ConductorSharp.Engine.Interface;
 using ConductorSharp.Engine.Model;
 using ConductorSharp.Engine.Polling;
+using ConductorSharp.Engine.Service;
 using ConductorSharp.Engine.Util;
 using MediatR;
 using Microsoft.Extensions.DependencyInjection;
@@ -32,6 +34,7 @@ internal class ExecutionManager : IExecutionManager
         private readonly IPollTimingStrategy _pollTimingStrategy;
         private readonly IPollOrderStrategy _pollOrderStrategy;
         private readonly ICancellationNotifier _cancellationNotifier;
+        private readonly TaskQueuePollingService _taskQueuePollingService;
 
         public ExecutionManager(
             WorkerSetConfig options,
@@ -42,7 +45,8 @@ public ExecutionManager(
             IServiceScopeFactory lifetimeScope,
             IPollTimingStrategy pollTimingStrategy,
             IPollOrderStrategy pollOrderStrategy,
-            ICancellationNotifier cancellationNotifier
+            ICancellationNotifier cancellationNotifier,
+            TaskQueuePollingService taskQueuePollingService
         )
         {
             _configuration = options;
@@ -55,6 +59,7 @@ ICancellationNotifier cancellationNotifier
             _pollOrderStrategy = pollOrderStrategy;
             _cancellationNotifier = cancellationNotifier;
             _externalPayloadService = externalPayloadService;
+            _taskQueuePollingService = taskQueuePollingService;
         }
 
         public async Task StartAsync(CancellationToken cancellationToken)
@@ -63,7 +68,7 @@ public async Task StartAsync(CancellationToken cancellationToken)
 
             while (!cancellationToken.IsCancellationRequested)
             {
-                var queuedTasks = (await _taskManager.ListQueuesAsync(cancellationToken))
+                var queuedTasks = (await _taskQueuePollingService.ListQueuesAsync(cancellationToken))
                     .Where(a => a.Value > 0)
                     .ToDictionary(a => a.Key, a => a.Value);
 
@@ -242,7 +247,9 @@ await _taskManager.UpdateAsync(
                     pollResponse.WorkflowInstanceId
                 );
 
-                var errorMessage = new ErrorOutput { ErrorMessage = exception.Message };
+                // A worker may throw a StructuredErrorException to attach a sanitized, stable classification to the
+                // failed task's output. Plain exceptions keep producing only error_message, preserving backward compatibility.
+                var errorMessage = new ErrorOutput { ErrorMessage = exception.Message, StructuredError = StructuredError.FromException(exception) };
 
                 // TODO: We should verify that this is alright, it is possible that when executed concurrently,
                 // the updates caused by LogAsync will be discarded because the call of UpdateAsync(TaskResult...)

src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs

@@ -17,7 +17,7 @@
 {
     public class ConductorSharpBuilder(IServiceCollection builder) : IConductorSharpBuilder, IExecutionManagerBuilder
     {
         public IServiceCollection Builder { get; set; } = builder;
 
         public IExecutionManagerBuilder AddExecutionManager(
             int maxConcurrentWorkers,
@@ -43,7 +43,9 @@
 
             Builder.AddTransient<ModuleDeployment>();
 
-            Builder.AddSingleton<IExecutionManager,ExecutionManager>();
+            Builder.AddSingleton<TaskQueuePollingService>();
+
+            Builder.AddSingleton<IExecutionManager, ExecutionManager>();
 
             Builder.AddScoped<ConductorSharpExecutionContext>();
 
@@ -62,10 +64,10 @@
 
         public IExecutionManagerBuilder UseBetaExecutionManager()
         {
-            Builder.AddSingleton<IExecutionManager,TypePollSpreadingExecutionManager>();
+            Builder.AddSingleton<IExecutionManager, TypePollSpreadingExecutionManager>();
             return this;
         }
-        
+
         public IExecutionManagerBuilder AddPipelines(Action<IPipelineBuilder> behaviorBuilder)
         {
             var pipelineBuilder = new PipelineBuilder(Builder);

src/ConductorSharp.Engine/Model/ErrorOutput.cs

@@ -7,5 +7,12 @@ namespace ConductorSharp.Engine.Model
     public class ErrorOutput
     {
         public string ErrorMessage { get; set; }
+
+        /// <summary>
+        /// Optional structured error classification. Null for plain (unclassified) failures, in which case it is
+        /// omitted from serialized output (NullValueHandling.Ignore), preserving backward compatibility with
+        /// consumers that only read <see cref="ErrorMessage"/>.
+        /// </summary>
+        public StructuredError StructuredError { get; set; }
     }
 }

src/ConductorSharp.Engine/Model/StructuredError.cs

@@ -0,0 +1,57 @@
+using System;
+using ConductorSharp.Engine.Exceptions;
+
+namespace ConductorSharp.Engine.Model
+{
+    /// <summary>
+    /// Sanitized, structured error classification transported across the <c>structured_error</c> task-output key.
+    /// The field shape is versioned via <see cref="Version"/> so it can evolve without silent misparses.
+    /// </summary>
+    public class StructuredError
+    {
+        /// <summary>Current structured-error payload shape version.</summary>
+        public const int CurrentVersion = 1;
+
+        /// <summary>Stable, opaque classification code (e.g. an implementation-defined code, or <c>UNCLASSIFIED</c>).</summary>
+        public string Code { get; set; }
+
+        /// <summary>Short, stable, sanitized reason. Consumers may key off this text, so keep it terse.</summary>
+        public string Reason { get; set; }
+
+        /// <summary>
+        /// Optional diagnostic detail, longer and more specific than <see cref="Reason"/> — the explanation an
+        /// operator needs, kept out of <see cref="Reason"/> so that stays short and stable. Null when the producer
+        /// supplied nothing distinct from the reason, in which case it is omitted from serialized output
+        /// (NullValueHandling.Ignore) and the payload is unchanged from before this field existed.
+        /// </summary>
+        public string Message { get; set; }
+
+        /// <summary>Optional URI pointing at the entity where the failure originated (drill-down link).</summary>
+        public string ReferenceError { get; set; }
+
+        /// <summary>Payload shape version marker. Defaults to <see cref="CurrentVersion"/>.</summary>
+        public int Version { get; set; } = CurrentVersion;
+
+        /// <summary>
+        /// Maps a thrown exception onto the payload, returning <c>null</c> for anything that is not a
+        /// <see cref="StructuredErrorException"/> so plain exceptions keep producing only <c>error_message</c>.
+        /// This is the single exception-to-payload mapping: both execution managers and the contract tests call it,
+        /// so the two poll strategies cannot drift apart as the shape evolves.
+        /// </summary>
+        public static StructuredError FromException(Exception exception)
+        {
+            if (exception is not StructuredErrorException structuredException)
+                return null;
+
+            return new StructuredError
+            {
+                Code = structuredException.Code,
+                Reason = structuredException.Reason,
+                // Exception.Message falls back to Reason when the thrower supplied no distinct detail, so only
+                // carry it when it actually adds something. Existing call sites keep their exact payload.
+                Message = structuredException.Message == structuredException.Reason ? null : structuredException.Message,
+                ReferenceError = structuredException.ReferenceError
+            };
+        }
+    }
+}

src/ConductorSharp.Engine/Service/TaskQueuePollingService.cs

@@ -0,0 +1,97 @@
+using System;
+using System.Collections.Generic;
+using System.Net.Http;
+using System.Threading;
+using System.Threading.Tasks;
+using ConductorSharp.Client.Generated;
+using ConductorSharp.Client.Service;
+using Microsoft.Extensions.Logging;
+using Task = System.Threading.Tasks.Task;
+
+namespace ConductorSharp.Engine.Service
+{
+    internal class TaskQueuePollingService
+    {
+        private const int DefaultMaxAttempts = 5;
+        private static readonly TimeSpan DefaultInitialRetryDelay = TimeSpan.FromSeconds(1);
+        private static readonly TimeSpan DefaultMaxRetryDelay = TimeSpan.FromSeconds(30);
+        private static readonly TimeSpan DefaultMaxJitter = TimeSpan.FromMilliseconds(500);
+
+        private readonly ITaskService _taskService;
+        private readonly ILogger<TaskQueuePollingService> _logger;
+        private readonly int _maxAttempts;
+        private readonly TimeSpan _initialRetryDelay;
+        private readonly TimeSpan _maxRetryDelay;
+        private readonly TimeSpan _maxJitter;
+        private readonly Func<TimeSpan, CancellationToken, Task> _delay;
+        private readonly Func<double> _jitter;
+
+        public TaskQueuePollingService(ITaskService taskService, ILogger<TaskQueuePollingService> logger)
+            : this(
+                taskService,
+                logger,
+                DefaultMaxAttempts,
+                DefaultInitialRetryDelay,
+                DefaultMaxRetryDelay,
+                DefaultMaxJitter,
+                Task.Delay,
+                Random.Shared.NextDouble
+            ) { }
+
+        internal TaskQueuePollingService(
+            ITaskService taskService,
+            ILogger<TaskQueuePollingService> logger,
+            int maxAttempts,
+            TimeSpan initialRetryDelay,
+            TimeSpan maxRetryDelay,
+            TimeSpan maxJitter,
+            Func<TimeSpan, CancellationToken, Task> delay,
+            Func<double> jitter
+        )
+        {
+            _taskService = taskService;
+            _logger = logger;
+            _maxAttempts = maxAttempts;
+            _initialRetryDelay = initialRetryDelay;
+            _maxRetryDelay = maxRetryDelay;
+            _maxJitter = maxJitter;
+            _delay = delay;
+            _jitter = jitter;
+        }
+
+        public async Task<IDictionary<string, long>> ListQueuesAsync(CancellationToken cancellationToken)
+        {
+            var retryDelay = _initialRetryDelay;
+
+            for (var attempt = 1; ; attempt++)
+            {
+                try
+                {
+                    return await _taskService.ListQueuesAsync(cancellationToken);
+                }
+                catch (Exception exception) when (!cancellationToken.IsCancellationRequested && IsTransient(exception) && attempt < _maxAttempts)
+                {
+                    var delay = retryDelay + TimeSpan.FromMilliseconds(_jitter() * _maxJitter.TotalMilliseconds);
+
+                    _logger.LogWarning(
+                        exception,
+                        "Failed to read Conductor task queues. Attempt {Attempt}/{MaxAttempts}; retrying in {RetryDelay}",
+                        attempt,
+                        _maxAttempts,
+                        delay
+                    );
+
+                    await _delay(delay, cancellationToken);
+                    retryDelay = TimeSpan.FromMilliseconds(Math.Min(retryDelay.TotalMilliseconds * 2, _maxRetryDelay.TotalMilliseconds));
+                }
+            }
+        }
+
+        private static bool IsTransient(Exception exception)
+        {
+            return exception is HttpRequestException
+                || exception is TaskCanceledException
+                || exception is ApiException apiException && (apiException.StatusCode is 408 or 429 || apiException.StatusCode >= 500);
+        }
+    }
+}