Fix: post exec action fixes

Services/Orchestration/OrchestratorService.cs

@@ -427,7 +427,8 @@ private async Task RunTaskAsync(RunHeader header, int seq, string taskId, int at
         _logger.LogDebug("[Scheduler] Task completed: {TaskId}", taskId);
     }
 
-    private async Task RunPostExecutionAsync(RunHeader header, JobDescriptor descriptor, CancellationToken ct)
+    private async Task RunPostExecutionAsync(RunHeader header, JobDescriptor descriptor, CancellationToken ct,
+        PowerShellWorker? worker = null)
     {
         var postExecScript = Script(_settings.Orchestrator.PostExecFunction);
         if (postExecScript == null)
@@ -455,7 +456,7 @@ private async Task RunPostExecutionAsync(RunHeader header, JobDescriptor descrip
             if (await _store.GetPostExecParametersAsync(header.RunKey, ct) is { Length: > 0 } postParameters)
                 parameters["ParametersJson"] = postParameters;
 
-            await RunScriptAsync(postExecScript, parameters, captureOutput: false);
+            await RunScriptAsync(postExecScript, parameters, captureOutput: false, worker);
             await OrchestratorBridge.DrainPendingAsync();
             _logger.LogInformation("[Orchestrator] PostExecution Push-{Function} completed for run {Name}",
                 header.PostExecFunctionName, header.Name);
@@ -521,13 +522,6 @@ void StepDone(string status, string? error, WorkStore.ClaimedTask? next) =>
                         await _store.CancelPendingAsync(header.RunKey, ct: jobCt);
                         break;
                     }
-                    if (step.Seq == WorkStore.AggregateSeq)
-                    {
-                        var aggregate = new JobDescriptor(header.Name, step.TaskId, header.Priority) { RunKey = header.RunKey, Seq = step.Seq, Attempt = step.Attempt };
-                        await RunPostExecutionAsync(current, aggregate, jobCt);
-                        break;
-                    }
-
                     if (worker is { IsHung: true })
                     {
                         // A step timed out on a worker that could not be stopped. Hand it back for replacement.
@@ -536,6 +530,14 @@ void StepDone(string status, string? error, WorkStore.ClaimedTask? next) =>
                         worker = CheckoutSequentialWorker(jobCt);
                     }
 
+                    if (step.Seq == WorkStore.AggregateSeq)
+                    {
+                        var aggregate = new JobDescriptor(header.Name, step.TaskId, header.Priority) { RunKey = header.RunKey, Seq = step.Seq, Attempt = step.Attempt };
+                        // On the pinned worker: asking the pool for another while holding this one deadlocks a full pool.
+                        await RunPostExecutionAsync(current, aggregate, jobCt, worker);
+                        break;
+                    }
+
                     var parameters = nextPayload ?? await _store.GetPayloadAsync(header.RunKey, step.Seq, jobCt);
                     nextPayload = null;
                     WorkStore.Finish finish;

Services/PowerShellHost/PowerShellWorker.cs

@@ -621,24 +621,38 @@ public void Warmup(CraftSettings settings)
         if (settings.Worker.WarmupScripts.Count == 0) return;
 
         var combinedScript = string.Join("\n", settings.Worker.WarmupScripts);
-        _pwsh.AddScript($@"
+
+        // Log records as they are emitted so warmup lines keep their real time and order.
+        EventHandler<DataAddedEventArgs> onWarning = (s, e) =>
+            _logger?.LogWarning("[Warmup] {Message}", ((PSDataCollection<WarningRecord>)s!)[e.Index].Message);
+        EventHandler<DataAddedEventArgs> onInfo = (s, e) =>
+            _logger?.LogInformation("[Warmup] {Message}", ((PSDataCollection<InformationRecord>)s!)[e.Index].MessageData);
+        EventHandler<DataAddedEventArgs> onError = (s, e) =>
+        {
+            var err = ((PSDataCollection<ErrorRecord>)s!)[e.Index];
+            _logger?.LogError("[Warmup] {Error}", err.Exception?.Message ?? err.ToString());
+        };
+        _pwsh.Streams.Warning.DataAdded += onWarning;
+        _pwsh.Streams.Information.DataAdded += onInfo;
+        _pwsh.Streams.Error.DataAdded += onError;
+        try
+        {
+            _pwsh.AddScript($@"
 try {{
     {combinedScript}
 }} catch {{
     Write-Warning ""Warmup failed: $_""
 }}
 ").Invoke();
-
-        // Surface warmup diagnostics before clearing streams — RunScript discards them.
-        foreach (var warn in _pwsh.Streams.Warning)
-            _logger?.LogWarning("[Warmup] {Message}", warn.Message);
-        foreach (var info in _pwsh.Streams.Information)
-            _logger?.LogInformation("[Warmup] {Message}", info.MessageData);
-        foreach (var err in _pwsh.Streams.Error)
-            _logger?.LogError("[Warmup] {Error}", err.Exception?.Message ?? err.ToString());
-
-        _pwsh.Commands.Clear();
-        _pwsh.Streams.ClearStreams();
+        }
+        finally
+        {
+            _pwsh.Streams.Warning.DataAdded -= onWarning;
+            _pwsh.Streams.Information.DataAdded -= onInfo;
+            _pwsh.Streams.Error.DataAdded -= onError;
+            _pwsh.Commands.Clear();
+            _pwsh.Streams.ClearStreams();
+        }
     }
 
     public void Dispose()

tests/Craft.Tests/HungWorkerRecycleTests.cs

@@ -186,6 +186,20 @@ public async Task ASequentialRunSwapsOutAWorkerThatHung_AndRunsTheRemainingSteps
         Assert.Equal((1, 3), (run.Failed, run.Done));
     }
 
+    [Fact]
+    public async Task ASequentialRunsPostExecutionRunsOnItsPinnedWorker()
+    {
+        // The driver holds its worker until the post-execution returns, so asking the pool for another deadlocks a full pool.
+        await using var h = await OrchestrationHarness.CreateAsync();
+        using var pinned = NewWorker(TimeSpan.FromSeconds(5));
+        h.Svc.NextWorker = () => pinned;
+        Assert.True(await h.Start("SeqPost", OrchestrationHarness.Batch(2, "s"), postExec: "Done", sequential: true));
+
+        Assert.True(await h.DriveUntilFinished("SeqPost"));
+        Assert.Equal(new PowerShellWorker?[] { pinned }, h.Svc.PostExecOn.ToArray());
+        Assert.Equal(1, h.Svc.Checkouts);
+    }
+
     // ── The pool, through the runner ────────────────────────────────────────────────────────────
 
     private static (PowerShellWorkerPool Pool, PowerShellRunnerService Runner) StartPool()

tests/Craft.Tests/OrchestrationHarness.cs

@@ -52,6 +52,7 @@ internal override async Task<string> RunScriptAsync(string path, Dictionary<stri
         if (path == PostExecScript)
         {
             PostExecs.Enqueue((parameters, File.ReadAllLines((string)parameters["ResultsPath"])));
+            PostExecOn.Enqueue(worker);
             if (PostExecBody != null) await PostExecBody();
             return string.Empty;
         }
@@ -80,6 +81,7 @@ internal override async Task<string> RunScriptAsync(string path, Dictionary<stri
     /// <summary>The worker a sequential driver checks out; null unless a test supplies one.</summary>
     public Func<PowerShellWorker?>? NextWorker;
     public readonly ConcurrentQueue<PowerShellWorker?> RanOn = new();
+    public readonly ConcurrentQueue<PowerShellWorker?> PostExecOn = new();
     public readonly ConcurrentQueue<(PowerShellWorker? Worker, bool Faulted)> Reclaimed = new();
 
     internal override PowerShellWorker? CheckoutSequentialWorker(CancellationToken ct)