pr feedback

Author: danielgerlag

src/WorkflowCore.DSL/Services/DefinitionLoader.cs

@@ -153,7 +153,7 @@ private WorkflowStepCollection ConvertSteps(ICollection<StepSourceV1> source, Ty
                     }
                     catch (Exception ex) when (ex is System.Linq.Dynamic.Core.Exceptions.ParseException || ex is InvalidOperationException)
                     {
-                        throw new WorkflowDefinitionLoadException($"Error parsing cancel condition expression '{nextStep.CancelCondition}' for step '{nextStep.Id}': {ex.Message}");
+                        throw new WorkflowDefinitionLoadException($"Error parsing cancel condition expression '{nextStep.CancelCondition}' for step '{nextStep.Id}': {ex.Message}", ex);
                     }
                 }
 
@@ -283,7 +283,7 @@ private void AttachOutputs(StepSourceV1 source, Type dataType, Type stepType, Wo
                 }
                 catch (Exception ex) when (ex is System.Linq.Dynamic.Core.Exceptions.ParseException || ex is InvalidOperationException)
                 {
-                    throw new WorkflowDefinitionLoadException($"Error parsing output expression '{output.Value}': {ex.Message}");
+                    throw new WorkflowDefinitionLoadException($"Error parsing output expression '{output.Value}': {ex.Message}", ex);
                 }
 
                 var dataParameter = Expression.Parameter(dataType, "data");
@@ -442,7 +442,7 @@ private void AttachOutcomes(StepSourceV1 source, Type dataType, WorkflowStep ste
                 }
                 catch (Exception ex) when (ex is System.Linq.Dynamic.Core.Exceptions.ParseException || ex is InvalidOperationException)
                 {
-                    throw new WorkflowDefinitionLoadException($"Error parsing select next step expression '{nextStep.Value}': {ex.Message}");
+                    throw new WorkflowDefinitionLoadException($"Error parsing select next step expression '{nextStep.Value}': {ex.Message}", ex);
                 }
                 Expression<Func<object, object, bool>> sourceExpr = (data, outcome) => System.Convert.ToBoolean(sourceDelegate.DynamicInvoke(data, outcome));
                 step.Outcomes.Add(new ExpressionOutcome<object>(sourceExpr)
@@ -467,7 +467,7 @@ private static Action<IStepBody, object, IStepExecutionContext> BuildScalarInput
             }
             catch (Exception ex) when (ex is System.Linq.Dynamic.Core.Exceptions.ParseException || ex is InvalidOperationException)
             {
-                throw new WorkflowDefinitionLoadException($"Error parsing input expression '{expr}' for property '{input.Key}': {ex.Message}");
+                throw new WorkflowDefinitionLoadException($"Error parsing input expression '{expr}' for property '{input.Key}': {ex.Message}", ex);
             }
 
             void acn(IStepBody pStep, object pData, IStepExecutionContext pContext)

src/WorkflowCore/Exceptions/WorkflowDefinitionLoadException.cs

@@ -8,5 +8,10 @@ public WorkflowDefinitionLoadException(string message)
             : base (message)
         {            
         }
+
+        public WorkflowDefinitionLoadException(string message, Exception innerException)
+            : base(message, innerException)
+        {
+        }
     }
 }

src/WorkflowCore/Services/ActivityController.cs

@@ -55,7 +55,7 @@ public async Task<PendingActivity> GetPendingActivity(string activityName, strin
                     Token = token.Encode(),
                     ActivityName = subscription.EventKey,
                     Parameters = subscription.SubscriptionData,
-                    TokenExpiry = DateTime.MaxValue
+                    TokenExpiry = DateTime.SpecifyKind(DateTime.MaxValue, DateTimeKind.Utc)
                 };
 
                 if (!await _subscriptionRepository.SetSubscriptionToken(subscription.Id, result.Token, workerId, result.TokenExpiry))

src/WorkflowCore/Services/DefaultProviders/MemoryPersistenceProvider.cs

@@ -178,6 +178,8 @@ public Task<bool> SetSubscriptionToken(string eventSubscriptionId, string token,
             lock (_subscriptions)
             {
                 var sub = _subscriptions.SingleOrDefault(x => x.Id == eventSubscriptionId);
+                if (sub == null)
+                    return Task.FromResult(false);
                 sub.ExternalToken = token;
                 sub.ExternalWorkerId = workerId;
                 sub.ExternalTokenExpiry = expiry;
@@ -191,6 +193,8 @@ public Task ClearSubscriptionToken(string eventSubscriptionId, string token, Can
             lock (_subscriptions)
             {
                 var sub = _subscriptions.SingleOrDefault(x => x.Id == eventSubscriptionId);
+                if (sub == null)
+                    throw new InvalidOperationException($"Subscription {eventSubscriptionId} not found.");
                 if (sub.ExternalToken != token)
                     throw new InvalidOperationException();
                 sub.ExternalToken = null;

src/WorkflowCore/Services/DefaultProviders/SingleNodeEventHub.cs

@@ -18,9 +18,9 @@ public SingleNodeEventHub(ILoggerFactory loggerFactory)
             _logger = loggerFactory.CreateLogger<SingleNodeEventHub>();
         }
 
-        public async Task PublishNotification(LifeCycleEvent evt)
+        public Task PublishNotification(LifeCycleEvent evt)
         {
-            await Task.Run(() =>
+            Task.Run(() =>
             {
                 foreach (var subscriber in _subscribers.ToArray())
                 {
@@ -34,6 +34,7 @@ await Task.Run(() =>
                     }
                 }
             });
+            return Task.CompletedTask;
         }
 
         public void Subscribe(Action<LifeCycleEvent> action)

src/extensions/WorkflowCore.AI.AzureFoundry/Services/InMemoryConversationStore.cs

@@ -38,7 +38,22 @@ public Task<ConversationThread> GetOrCreateThreadAsync(string workflowInstanceId
                 return thread.Id;
             });
 
-            return Task.FromResult(_threads[threadId]);
+            if (_threads.TryGetValue(threadId, out var existingThread))
+            {
+                return Task.FromResult(existingThread);
+            }
+
+            // The thread was removed or the mapping is stale; recreate and update the mapping.
+            var newThread = new ConversationThread
+            {
+                WorkflowInstanceId = workflowInstanceId,
+                ExecutionPointerId = executionPointerId
+            };
+
+            _threads[newThread.Id] = newThread;
+            _workflowThreadMap[key] = newThread.Id;
+
+            return Task.FromResult(newThread);
         }
 
         public Task SaveThreadAsync(ConversationThread thread)

src/extensions/WorkflowCore.WebAPI/Controllers/WorkflowsController.cs

@@ -56,7 +56,7 @@ public async Task<IActionResult> Post(string id, int? version, string reference,
             string workflowId = null;            
             var def = _registry.GetDefinition(id, version);
             if (def == null)
-                return NotFound(String.Format("Workflow defintion of {0} not found", id));
+                return NotFound(String.Format("Workflow definition of {0} not found", id));
             if ((data != null) && (def.DataType != null))
             {
                 var dataStr = JsonConvert.SerializeObject(data);

src/providers/WorkflowCore.Persistence.EntityFramework/Services/EntityFrameworkPersistenceProvider.cs

@@ -145,7 +145,7 @@ public async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken c
                     .ThenInclude(ep => ep.ExtensionAttributes)
                     .Include(wf => wf.ExecutionPointers)
                     .AsTracking()
-                    .FirstOrDefaultAsync(cancellationToken);
+                    .FirstAsync(cancellationToken);
 
                 var persistable = workflow.ToPersistable(existingEntity);
                 await db.SaveChangesAsync(cancellationToken);
@@ -163,7 +163,7 @@ public async Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscript
                     .ThenInclude(ep => ep.ExtensionAttributes)
                     .Include(wf => wf.ExecutionPointers)
                     .AsTracking()
-                    .FirstOrDefaultAsync(cancellationToken);
+                    .FirstAsync(cancellationToken);
 
                 var workflowPersistable = workflow.ToPersistable(existingEntity);
 
@@ -183,7 +183,7 @@ public async Task TerminateSubscription(string eventSubscriptionId, Cancellation
             using (var db = ConstructDbContext())
             {
                 var uid = new Guid(eventSubscriptionId);
-                var existing = await db.Set<PersistedSubscription>().FirstOrDefaultAsync(x => x.SubscriptionId == uid, cancellationToken);
+                var existing = await db.Set<PersistedSubscription>().FirstAsync(x => x.SubscriptionId == uid, cancellationToken);
                 db.Set<PersistedSubscription>().Remove(existing);
                 await db.SaveChangesAsync(cancellationToken);
             }
@@ -273,6 +273,9 @@ public async Task MarkEventProcessed(string id, CancellationToken cancellationTo
                     .AsTracking()
                     .FirstOrDefaultAsync(cancellationToken);
 
+                if (existingEntity == null)
+                    return;
+
                 existingEntity.IsProcessed = true;
                 await db.SaveChangesAsync(cancellationToken);
             }
@@ -307,6 +310,9 @@ public async Task MarkEventUnprocessed(string id, CancellationToken cancellation
                     .AsTracking()
                     .FirstOrDefaultAsync(cancellationToken);
 
+                if (existingEntity == null)
+                    return;
+
                 existingEntity.IsProcessed = false;
                 await db.SaveChangesAsync(cancellationToken);
             }
@@ -363,7 +369,7 @@ public async Task<bool> SetSubscriptionToken(string eventSubscriptionId, string
                 var existingEntity = await db.Set<PersistedSubscription>()
                     .Where(x => x.SubscriptionId == uid)
                     .AsTracking()
-                    .FirstOrDefaultAsync(cancellationToken);
+                    .FirstAsync(cancellationToken);
 
                 existingEntity.ExternalToken = token;
                 existingEntity.ExternalWorkerId = workerId;
@@ -382,7 +388,7 @@ public async Task ClearSubscriptionToken(string eventSubscriptionId, string toke
                 var existingEntity = await db.Set<PersistedSubscription>()
                     .Where(x => x.SubscriptionId == uid)
                     .AsTracking()
-                    .FirstOrDefaultAsync(cancellationToken);
+                    .FirstAsync(cancellationToken);
 
                 if (existingEntity.ExternalToken != token)
                     throw new InvalidOperationException();

src/providers/WorkflowCore.QueueProviders.SqlServer/Services/SqlServerQueueProvider.cs

@@ -128,9 +128,8 @@ private static string SanitizeIdentifier(string name)
         {
             if (string.IsNullOrWhiteSpace(name))
                 throw new ArgumentException("Queue name cannot be null or empty.", nameof(name));
-            if (!System.Text.RegularExpressions.Regex.IsMatch(name, @"^[a-zA-Z_][a-zA-Z0-9_/]*$"))
-                throw new ArgumentException($"Queue name '{name}' contains invalid characters.", nameof(name));
-            return name;
+            // Escape any ']' characters to prevent breaking out of the delimited identifier
+            return name.Replace("]", "]]");
         }
     }
 }