[codex] Fix ForEach completion from nested flowchart (#7702)

* Fix ForEach completion from nested flowchart

* Assert ForEach complete output regression

* Use break flag helper consistently

* Cover parent flowchart ForEach completion

src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.Counters.cs

@@ -380,10 +380,15 @@ public partial class Flowchart
         var flowchartContext = context.ReceiverActivityExecutionContext;
         await CompleteIfNoPendingWorkAsync(flowchartContext);
         var flowchart = (Flowchart)flowchartContext.Activity;
+        var canceledActivity = context.SenderActivityExecutionContext.Activity;
+
+        if (!flowchart.Activities.Contains(canceledActivity))
+            return;
+
         var flowGraph = flowchartContext.GetFlowGraph();
         var flowScope = flowchart.GetFlowScope(flowchartContext);
 
         // Propagate canceled connections visited count by scheduling with Outcomes.Empty
-        await MaybeScheduleOutboundActivitiesAsync(flowGraph, flowScope, flowchartContext, context.SenderActivityExecutionContext.Activity, context.SenderActivityExecutionContext, Outcomes.Empty, OnChildCompletedAsync);
+        await MaybeScheduleOutboundActivitiesAsync(flowGraph, flowScope, flowchartContext, canceledActivity, context.SenderActivityExecutionContext, Outcomes.Empty, OnChildCompletedAsync);
     }
-}
+}

src/modules/Elsa.Workflows.Core/Behaviors/BreakBehavior.cs

@@ -21,6 +21,8 @@ public class BreakBehavior : Behavior
 
     private async ValueTask OnCompleteCompositeAsync(CompleteCompositeSignal signal, SignalContext context)
     {
+        context.ReceiverActivityExecutionContext.SetIsBreaking();
+
         // Cancel each descendant to clear bookmarks and cancel jobs etc.
         await CancelDescendantsAsync(context);
 
@@ -34,11 +36,11 @@ public class BreakBehavior : Behavior
         context.StopPropagation();
         
         // Set the IsBreaking property to true.
-        context.ReceiverActivityExecutionContext.SetProperty("IsBreaking", true);
+        context.ReceiverActivityExecutionContext.SetIsBreaking();
     }
     
     private async Task CancelDescendantsAsync(SignalContext context)
     {
         await context.ReceiverActivityExecutionContext.CancelActivityAsync();
     }
-}
+}

test/integration/Elsa.Activities.IntegrationTests/ForEachTests.cs

@@ -2,7 +2,12 @@ using Elsa.Extensions;
 using Elsa.Testing.Shared;
 using Elsa.Workflows;
 using Elsa.Workflows.Activities;
+using Elsa.Workflows.Activities.Flowchart.Activities;
+using Elsa.Workflows.Activities.Flowchart.Extensions;
+using Elsa.Workflows.Activities.Flowchart.Models;
+using Elsa.Workflows.Management.Activities.SetOutput;
 using Elsa.Workflows.Models;
+using Elsa.Workflows.Options;
 using Xunit.Abstractions;
 
 namespace Elsa.Activities.IntegrationTests;
@@ -247,6 +252,61 @@ public class ForEachTests(ITestOutputHelper testOutputHelper)
         Assert.Equal(ActivityStatus.Running, forEachContext.Status);
         Assert.Equal(1, forEachContext.AggregateFaultCount);
     }
+
+    [Fact(DisplayName = "ForEach completes when a nested flowchart completes the composite")]
+    public async Task ForEach_Completes_WhenNestedFlowchartCompletesComposite()
+    {
+        var dataSource = new[]
+        {
+            "a", "b", "c"
+        };
+        var writeLine = WriteCurrentValue();
+        var decision = new FlowDecision(context => context.GetVariable<string>(CurrentValueVar) == "b");
+        var setOutput = new SetOutput
+        {
+            OutputName = new("Output"),
+            OutputValue = new(context => context.GetVariable<string>(CurrentValueVar))
+        };
+        var complete = new Complete(["True"]);
+        var forEach = new ForEach<string>(dataSource)
+        {
+            Body = new Flowchart
+            {
+                Activities =
+                {
+                    writeLine,
+                    decision,
+                    setOutput,
+                    complete
+                },
+                Connections =
+                {
+                    new() { Source = new(writeLine, "Done"), Target = new(decision) },
+                    new() { Source = new(decision, "True"), Target = new(setOutput) },
+                    new() { Source = new(setOutput, "Done"), Target = new(complete) }
+                }
+            }
+        };
+        var outerComplete = new Complete(["False"]);
+        var outerFlowchart = new Flowchart
+        {
+            Activities =
+            {
+                forEach,
+                outerComplete
+            },
+            Connections =
+            {
+                new() { Source = new(forEach, "Done"), Target = new(outerComplete) }
+            }
+        };
+        var options = new RunWorkflowOptions().WithCounterBasedFlowchart();
+
+        var result = await _fixture.RunActivityAsync(outerFlowchart, options);
+
+        Assert.Equal(new[] { "a", "b" }, _fixture.CapturingTextWriter.Lines);
+        Assert.Equal("b", result.WorkflowState.Output["Output"]);
+    }
     
     private static WriteLine WriteCurrentValue() => new(context => context.GetVariable<string>(CurrentValueVar));
     
@@ -260,4 +320,4 @@ public class ForEachTests(ITestOutputHelper testOutputHelper)
 record Foo(string Bar = "Baz")
 {
     public override string ToString() => Bar;
-}
+}