[codex] Scope console logs to workflow instances (#7535)

* Avoid null endpoint DTO metadata in tests

* Enforce console logs hub read permission

* Remove unused console logs hub import

* Support mapped endpoint metadata in auth tests

* Reduce console log capture throughput impact

* Address Copilot console logs review

* Refactor task scheduling to support tenant-level background work and enhance logging functionality.

* Introduce ConsoleStreamHook for stdout/stderr tee and enhance logging validation. Adjust test cases and startup warnings for distributed lock provider usage.

* Refactor console logging pipeline with capture optimization and new ConsoleLogsHost; update tests accordingly.

* Add Ansi SGR parser for console logs and associated unit tests

* Remove ANSI color renderings and parsers; integrate ConsoleLogScopeAccessor for improved logging context with workflow instance ID support.

* Address console logs code quality feedback

* Address PR review feedback

* Preserve console logs extension points

* Stabilize console logs host lifecycle

* Address final automated review comments

* Tighten console log capture shutdown

* Address console log review feedback

* Address follow-up review feedback

* Cover final review feedback

* Avoid recursive console provider initialization

* Guard console host lease shutdown

* Preserve console log scope and provider lifetime

* Correlate console log scope fallback

* Tighten console scope correlation

* Expose host services during provider construction

* Redact ANSI-normalized console lines

doc/wiki/workflow-runtime.md

@@ -172,14 +172,15 @@ Distributed runtime support lives in [Elsa.Workflows.Runtime.Distributed](../../
 
 ### Distributed Lock Provider Safety
 
-The default workflow runtime lock provider is file-system based and writes under `App_Data/locks`. That provider is useful for single-host development and tests, but it is not safe for clustered deployments where nodes have separate file systems. When `UseDistributedRuntime()` is enabled, startup fails if Elsa detects the default file-system provider or the no-op provider unless the host explicitly opts in to local-only lock semantics:
+The default workflow runtime lock provider is file-system based and writes under `App_Data/locks`. That provider is useful for single-host development and tests, but it is not safe for clustered deployments where nodes have separate file systems. When `UseDistributedRuntime()` is enabled, Elsa logs a startup warning if it detects the default file-system provider or the no-op provider unless the host explicitly acknowledges local-only lock semantics:
 
 ```csharp
 elsa.UseWorkflowRuntime(runtime =>
 {
     runtime.UseDistributedRuntime();
 
-    // Single-host development/test only. Do not use this for clustered production deployments.
+    // Single-host development/test only. Suppresses the startup warning.
+    // Do not use this for clustered production deployments.
     runtime.DistributedLockingOptions = options => options.AllowLocalLockProviderInDistributedRuntime = true;
 });
 ```

src/apps/Elsa.ModularServer.Web/Program.cs

@@ -1,6 +1,8 @@
 using CShells.AspNetCore.Configuration;
 using CShells.AspNetCore.Extensions;
 using CShells.DependencyInjection;
+using Elsa.Diagnostics.ConsoleLogs.Extensions;
+using Elsa.Diagnostics.ConsoleLogs.Services;
 using Elsa.ModularServer.Web;
 using Elsa.ModularServer.Web.Catalog;
 using Elsa.ShellFeatures;
@@ -13,10 +15,19 @@ using Nuplane;
 using Nuplane.Loading.Hosting.Builder;
 using Nuplane.Sources.Directory.Configuration;
 
+// Install the console stream tee as early as possible — before WebApplication.CreateBuilder and
+// any logger provider construction — so that downstream loggers (Microsoft.Extensions.Logging.Console,
+// Kestrel/Hosting loggers, etc.) capture our tee writer instead of the raw Console.Out / Console.Error.
+ConsoleStreamHook.Install();
+
 var builder = WebApplication.CreateBuilder(args);
 var services = builder.Services;
 var configuration = builder.Configuration;
 
+// Console output is a single process-wide resource; the capture pipeline is owned by the root host so
+// every shell's diagnostics endpoints (REST + SignalR) read from the same in-memory ring buffer.
+services.AddConsoleLogsHost();
+
 var nuplaneConfiguration = configuration.GetSection("Nuplane");
 
 services.AddNuplane(nuplaneConfiguration, nuplane =>

src/apps/Elsa.ModularServer.Web/appsettings.json

@@ -5,6 +5,11 @@
       "Microsoft.AspNetCore": "Warning",
       "Microsoft.EntityFrameworkCore": "Warning",
       "Nuplane": "Warning"
+    },
+    "Console": {
+      "FormatterOptions": {
+        "ColorBehavior": "Enabled"
+      }
     }
   },
   "AllowedHosts": "*",

src/apps/Elsa.Server.Web/Program.cs

@@ -91,7 +91,7 @@ services
                 runtime.UseEntityFrameworkCore(ef => ef.UseSqlite());
                 runtime.UseCache();
                 runtime.UseDistributedRuntime();
-                // This sample host uses single-host file-system locks only for development or explicit single-host opt-in.
+                // This sample host acknowledges single-host file-system locks in development or explicit single-host opt-in.
                 runtime.DistributedLockingOptions = options => options.AllowLocalLockProviderInDistributedRuntime = allowLocalDistributedRuntimeLockProvider;
             })
             .UseWorkflowsApi()

src/modules/Elsa.Common/Contracts/BackgroundTask.cs

@@ -5,4 +5,4 @@ public abstract class BackgroundTask : IBackgroundTask
     public virtual Task ExecuteAsync(CancellationToken cancellationToken) => Task.CompletedTask;
     public virtual Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
     public virtual Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
-}
+}

src/modules/Elsa.Common/DistributedHosting/Options/DistributedLockingOptions.cs

@@ -14,8 +14,8 @@ public class DistributedLockingOptions
     public TimeSpan LockAcquisitionTimeout { get; set; } = TimeSpan.FromMinutes(10);
 
     /// <summary>
-    /// Allows the distributed workflow runtime to use a lock provider that only coordinates within this node or local file system.
+    /// Acknowledges that the distributed workflow runtime is using a lock provider that only coordinates within this node or local file system.
-    /// Enable this only for single-host development or tests. Clustered deployments should configure a cross-node provider such as Redis, SQL Server, or PostgreSQL.
+    /// Enable this for single-host development or tests to suppress the startup warning. Clustered deployments should configure a cross-node provider such as Redis, SQL Server, or PostgreSQL.
     /// </summary>
     public bool AllowLocalLockProviderInDistributedRuntime { get; set; }
 }

src/modules/Elsa.Common/Features/MultitenancyFeature.cs

@@ -2,6 +2,7 @@ using Elsa.Common.Multitenancy;
 using Elsa.Common.Multitenancy.EventHandlers;
 using Elsa.Common.Multitenancy.HostedServices;
 using Elsa.Common.RecurringTasks;
+using Elsa.Extensions;
 using Elsa.Features.Abstractions;
 using Elsa.Features.Services;
 using Microsoft.Extensions.DependencyInjection;
@@ -50,8 +51,10 @@ public class MultitenancyFeature(IModule module) : FeatureBase(module)
             .AddSingleton<TenantEventsManager>()
             .AddScoped<DefaultTenantsProvider>()
             .AddScoped<DefaultTenantResolver>()
+            .AddScoped<ITenantBackgroundWorkQueue, TenantBackgroundWorkQueue>()
             .AddScoped<ITaskExecutor, TaskExecutor>()
             .AddScoped<IBackgroundTaskStarter, TaskExecutor>()
+            .AddBackgroundTask<TenantBackgroundWorkQueueWorker>()
             .AddScoped(_tenantsProviderFactory);
     }
-}
+}

src/modules/Elsa.Common/Multitenancy/Contracts/ITenantBackgroundWorkQueue.cs

@@ -0,0 +1,9 @@
+namespace Elsa.Common.Multitenancy;
+
+public delegate Task TenantBackgroundWorkItem(IServiceProvider serviceProvider, CancellationToken cancellationToken);
+
+public interface ITenantBackgroundWorkQueue
+{
+    ValueTask EnqueueAsync(TenantBackgroundWorkItem workItem, CancellationToken cancellationToken = default);
+    IAsyncEnumerable<TenantBackgroundWorkItem> DequeueAllAsync(CancellationToken cancellationToken = default);
+}

src/modules/Elsa.Common/Multitenancy/Implementations/TenantBackgroundWorkQueue.cs

@@ -0,0 +1,23 @@
+using System.Threading.Channels;
+
+namespace Elsa.Common.Multitenancy;
+
+public class TenantBackgroundWorkQueue : ITenantBackgroundWorkQueue
+{
+    private readonly Channel<TenantBackgroundWorkItem> _channel = Channel.CreateUnbounded<TenantBackgroundWorkItem>(new UnboundedChannelOptions
+    {
+        SingleReader = true,
+        SingleWriter = false
+    });
+
+    public ValueTask EnqueueAsync(TenantBackgroundWorkItem workItem, CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(workItem);
+        return _channel.Writer.WriteAsync(workItem, cancellationToken);
+    }
+
+    public IAsyncEnumerable<TenantBackgroundWorkItem> DequeueAllAsync(CancellationToken cancellationToken = default)
+    {
+        return _channel.Reader.ReadAllAsync(cancellationToken);
+    }
+}

src/modules/Elsa.Common/Multitenancy/Implementations/TenantBackgroundWorkQueueWorker.cs

@@ -0,0 +1,37 @@
+using Elsa.Common.Helpers;
+using Microsoft.Extensions.Logging;
+
+namespace Elsa.Common.Multitenancy;
+
+public class TenantBackgroundWorkQueueWorker(
+    ITenantBackgroundWorkQueue workQueue,
+    IServiceProvider serviceProvider,
+    ILogger<TenantBackgroundWorkQueueWorker> logger) : BackgroundTask
+{
+    public override async Task ExecuteAsync(CancellationToken cancellationToken)
+    {
+        try
+        {
+            await using var enumerator = workQueue.DequeueAllAsync(cancellationToken).GetAsyncEnumerator(cancellationToken);
+            while (await enumerator.MoveNextAsync())
+            {
+                try
+                {
+                    await enumerator.Current(serviceProvider, cancellationToken);
+                }
+                catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+                {
+                    return;
+                }
+                catch (Exception e) when (!e.IsFatal())
+                {
+                    logger.LogError(e, "A tenant background work item failed.");
+                }
+            }
+        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            logger.LogDebug("Tenant background work queue worker stopped because cancellation was requested.");
+        }
+    }
+}

src/modules/Elsa.Common/Multitenancy/Implementations/TenantTaskLifecycleCoordinator.cs

@@ -92,27 +92,38 @@ public class TenantTaskLifecycleCoordinator(RecurringTaskScheduleManager schedul
             await taskExecutor.ExecuteTaskAsync(task, cancellationToken);
     }
 
-    private Task RunBackgroundTasksAsync(ITenantScope tenantScope, ITaskExecutor taskExecutor, TenantRuntimeState state, CancellationToken cancellationToken)
+    private async Task RunBackgroundTasksAsync(ITenantScope tenantScope, ITaskExecutor taskExecutor, TenantRuntimeState state, CancellationToken cancellationToken)
     {
-        var backgroundTasks = tenantScope.ServiceProvider.GetServices<IBackgroundTask>();
+        var backgroundTasks = tenantScope.ServiceProvider.GetServices<IBackgroundTask>().ToList();
         var backgroundTaskStarter = tenantScope.ServiceProvider.GetRequiredService<IBackgroundTaskStarter>();
         var tenantCancellationToken = state.CancellationTokenSource?.Token ?? cancellationToken;
 
         foreach (var backgroundTask in backgroundTasks)
         {
-            var task = backgroundTaskStarter
+            await backgroundTaskStarter.StartAsync(backgroundTask, tenantCancellationToken);
-                .StartAsync(backgroundTask, tenantCancellationToken)
+            state.BackgroundTasks.Add(new(backgroundTask, backgroundTaskStarter));
-                .ContinueWith(_ => taskExecutor.ExecuteTaskAsync(backgroundTask, tenantCancellationToken),
+
-                    cancellationToken,
+            var task = RunBackgroundTaskAsync(backgroundTask, taskExecutor, tenantCancellationToken);
-                    TaskContinuationOptions.RunContinuationsAsynchronously,
-                    TaskScheduler.Default)
-                .Unwrap();
 
             if (!task.IsCompleted)
                 state.RunningBackgroundTasks.Add(task);
         }
+    }
 
-        return Task.CompletedTask;
+    private async Task RunBackgroundTaskAsync(IBackgroundTask backgroundTask, ITaskExecutor taskExecutor, CancellationToken cancellationToken)
+    {
+        try
+        {
+            await taskExecutor.ExecuteTaskAsync(backgroundTask, cancellationToken);
+        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            logger.LogInformation("Background task {TaskType} was cancelled", backgroundTask.GetType().Name);
+        }
+        catch (Exception e) when (!e.IsFatal())
+        {
+            logger.LogError(e, "Background task {TaskType} failed with an error", backgroundTask.GetType().Name);
+        }
     }
 
     private async Task StartRecurringTasksAsync(ITenantScope tenantScope, ITaskExecutor taskExecutor, TenantRuntimeState state, CancellationToken cancellationToken)
@@ -225,6 +236,23 @@ public class TenantTaskLifecycleCoordinator(RecurringTaskScheduleManager schedul
         }
         state.ScheduledTimers.Clear();
 
+        foreach (var backgroundTask in state.BackgroundTasks)
+        {
+            try
+            {
+                await backgroundTask.Starter.StopAsync(backgroundTask.Task, cancellationToken);
+            }
+            catch (OperationCanceledException)
+            {
+                // Expected if caller requested cancellation during tenant deactivation.
+            }
+            catch (Exception e) when (!e.IsFatal())
+            {
+                logger.LogError(e, "Failed to stop background task {TaskType}", backgroundTask.Task.GetType().Name);
+            }
+        }
+        state.BackgroundTasks.Clear();
+
         if (state.RunningBackgroundTasks.Count > 0)
         {
             try
@@ -290,8 +318,11 @@ public class TenantTaskLifecycleCoordinator(RecurringTaskScheduleManager schedul
         // Since we exclusively use WaitAsync(), no kernel handle is ever created and disposal is a no-op.
         public SemaphoreSlim Gate { get; } = new(1, 1);
         public List<Task> RunningBackgroundTasks { get; } = [];
+        public List<BackgroundTaskRegistration> BackgroundTasks { get; } = [];
         public List<ScheduledTimer> ScheduledTimers { get; } = [];
         public List<IRecurringTask> RecurringTasks { get; } = [];
         public CancellationTokenSource? CancellationTokenSource { get; set; }
     }
+
+    private record BackgroundTaskRegistration(IBackgroundTask Task, IBackgroundTaskStarter Starter);
 }

src/modules/Elsa.Common/ShellFeatures/MultitenancyFeature.cs

@@ -5,6 +5,7 @@ using Elsa.Common.Multitenancy.EventHandlers;
 using Elsa.Common.Multitenancy.HostedServices;
 using Elsa.Common.RecurringTasks;
 using Elsa.Common.ShellHandlers;
+using Elsa.Extensions;
 using Microsoft.Extensions.DependencyInjection;
 
 namespace Elsa.Common.ShellFeatures;
@@ -35,8 +36,10 @@ public class MultitenancyFeature : IShellFeature
             .AddSingleton<TenantEventsManager>()
             .AddScoped<DefaultTenantsProvider>()
             .AddScoped<DefaultTenantResolver>()
+            .AddScoped<ITenantBackgroundWorkQueue, TenantBackgroundWorkQueue>()
             .AddScoped<ITaskExecutor, TaskExecutor>()
             .AddScoped<IBackgroundTaskStarter, TaskExecutor>()
+            .AddBackgroundTask<TenantBackgroundWorkQueueWorker>()
             .AddScoped(_tenantsProviderFactory)
             
             // Transient per CShells 0.0.15 convention — the registry resolves IEnumerable<IShellInitializer> and

src/modules/Elsa.Diagnostics.ConsoleLogs/Contracts/IConsoleLogCapture.cs

@@ -1,10 +1,9 @@
 namespace Elsa.Diagnostics.ConsoleLogs.Contracts;
 
+[Obsolete("Console log capture is host-owned. Use ConsoleLogsHost.Capture for host-level operations.")]
 public interface IConsoleLogCapture : IAsyncDisposable
 {
     ValueTask StartAsync(CancellationToken cancellationToken = default);
-
     ValueTask StopAsync(CancellationToken cancellationToken = default);
-
     ValueTask FlushIdleAsync(CancellationToken cancellationToken = default);
 }

src/modules/Elsa.Diagnostics.ConsoleLogs/Extensions/HostServiceCollectionExtensions.cs

@@ -0,0 +1,39 @@
+using Elsa.Diagnostics.ConsoleLogs.Contracts;
+using Elsa.Diagnostics.ConsoleLogs.Services;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+using Microsoft.Extensions.Logging;
+
+namespace Elsa.Diagnostics.ConsoleLogs.Extensions;
+
+public static class HostServiceCollectionExtensions
+{
+    /// <summary>
+    /// Registers the host-level console capture pipeline. Call this from <c>Program.cs</c> (ideally right
+    /// after <c>ConsoleStreamHook.Install()</c>) so that the capture singleton, the in-memory ring buffer,
+    /// the source registry, and the idle flush loop are owned by the root host rather than by any
+    /// individual shell.
+    ///
+    /// <para>
+    /// Shell-level features that surface console logs (REST endpoints, SignalR hub) still call
+    /// <c>AddConsoleLogsServices</c> on their own service collection — those registrations resolve to the
+    /// same host-owned singletons exposed by <see cref="ConsoleLogsHost"/>.
+    /// </para>
+    /// </summary>
+    public static IServiceCollection AddConsoleLogsHost(this IServiceCollection services, Action<ConsoleLogsOptions>? configure = null)
+    {
+        ConsoleStreamHook.Install();
+
+        if (configure != null)
+            ConsoleLogsHost.Configure(configure);
+
+        services.TryAddSingleton(_ => ConsoleLogsHost.ScopeAccessor);
+        services.AddSingleton<ILoggerProvider>(sp => sp.GetRequiredService<ConsoleLogScopeAccessor>());
+#pragma warning disable CS0618
+        services.TryAddSingleton<IConsoleLogCapture, ConsoleLogCaptureAdapter>();
+#pragma warning restore CS0618
+        services.TryAddEnumerable(ServiceDescriptor.Singleton<Microsoft.Extensions.Hosting.IHostedService, ConsoleLogsHostedService>());
+        return services;
+    }
+}
+

src/modules/Elsa.Diagnostics.ConsoleLogs/Extensions/ServiceCollectionExtensions.cs

@@ -1,27 +1,94 @@
-using Elsa.Diagnostics.ConsoleLogs.Providers.InMemory;
+using Elsa.Diagnostics.ConsoleLogs.Contracts;
 using Elsa.Diagnostics.ConsoleLogs.RealTime;
 using Elsa.Diagnostics.ConsoleLogs.Services;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
 
 namespace Elsa.Diagnostics.ConsoleLogs.Extensions;
 
 public static class ServiceCollectionExtensions
 {
+    /// <summary>
+    /// Registers shell-level console-log consumers (SignalR, subscription manager) and exposes the
+    /// process-wide capture pipeline owned by <see cref="ConsoleLogsHost"/> through DI. Configuration
+    /// supplied here applies only if the host has not yet been initialized (first-wins). Prefer calling
+    /// <c>AddConsoleLogsHost</c> from <c>Program.cs</c> for deterministic host-level configuration.
+    /// </summary>
     public static IServiceCollection AddConsoleLogsServices(this IServiceCollection services, Action<ConsoleLogsOptions>? configureOptions = null)
     {
+        ConsoleStreamHook.Install();
+        var defaultProviderDescriptor = ServiceDescriptor.Singleton<IConsoleLogProvider>(_ => ConsoleLogsHost.Provider);
+
         if (configureOptions != null)
-            services.Configure(configureOptions);
+            ConsoleLogsHost.Configure(configureOptions);
+
+        services.TryAdd(defaultProviderDescriptor);
+        services.TryAddSingleton(new ConsoleLogsProviderRegistration(services, defaultProviderDescriptor));
+
+        ConsoleLogsProviderRegistration GetRegistration(IServiceProvider serviceProvider) =>
+            serviceProvider.GetRequiredService<ConsoleLogsProviderRegistration>();
+
+        void ConfigureCustomProvider(IServiceProvider serviceProvider) =>
+            GetRegistration(serviceProvider).ConfigureProvider(serviceProvider);
 
         services.AddSignalR();
-        services.AddOptions<ConsoleLogsOptions>();
+
-        services.TryAddSingleton<IConsoleLogSourceRegistry, ConsoleLogSourceRegistry>();
+        // Every shell resolves the same process-wide singletons — console output is a shared OS resource.
-        services.TryAddSingleton<IConsoleLogRedactor, ConsoleLogRedactor>();
+        services.TryAddSingleton<IOptions<ConsoleLogsOptions>>(sp =>
-        services.TryAddSingleton<ConsoleLineFormatter>();
+        {
-        services.TryAddSingleton<IConsoleLogProvider, InMemoryConsoleLogProvider>();
+            if (GetRegistration(sp).TryGetHostOptions(out var options))
+                return options;
+
+            ConfigureCustomProvider(sp);
+            return ConsoleLogsHost.Options;
+        });
+        services.TryAddSingleton<IConsoleLogSourceRegistry>(sp =>
+        {
+            if (GetRegistration(sp).TryGetHostSourceRegistry(out var sourceRegistry))
+                return sourceRegistry;
+
+            ConfigureCustomProvider(sp);
+            return ConsoleLogsHost.SourceRegistry;
+        });
+        services.TryAddSingleton<IConsoleLogRedactor>(sp =>
+        {
+            if (GetRegistration(sp).TryGetHostRedactor(out var redactor))
+                return redactor;
+
+            ConfigureCustomProvider(sp);
+            return ConsoleLogsHost.Redactor;
+        });
+        services.TryAddSingleton(sp =>
+        {
+            if (GetRegistration(sp).TryGetHostFormatter(out var formatter))
+                return formatter;
+
+            ConfigureCustomProvider(sp);
+            return ConsoleLogsHost.Formatter;
+        });
+        services.TryAddSingleton(sp =>
+        {
+            if (GetRegistration(sp).TryGetHostScopeAccessor(out var scopeAccessor))
+                return scopeAccessor;
+
+            ConfigureCustomProvider(sp);
+            return ConsoleLogsHost.ScopeAccessor;
+        });
+        services.AddSingleton<ILoggerProvider>(sp =>
+        {
+            ConfigureCustomProvider(sp);
+            return ConsoleLogsHost.ScopeAccessor;
+        });
+#pragma warning disable CS0618
+        services.TryAddSingleton<IConsoleLogCapture, ConsoleLogCaptureAdapter>();
+#pragma warning restore CS0618
+
+        // The subscription manager wraps a per-shell SignalR hub context, so it must be per-shell.
         services.TryAddSingleton<ConsoleLogSubscriptionManager>();
-        services.TryAddSingleton<IConsoleLogCapture, ConsoleCaptureTee>();
+        services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService, ConsoleLogsHostedService>());
-        services.AddHostedService<ConsoleLogCaptureHostedService>();
 
         return services;
     }

src/modules/Elsa.Diagnostics.ConsoleLogs/Models/ConsoleLogFilter.cs

@@ -5,6 +5,7 @@ public record ConsoleLogFilter
     public string? SourceId { get; init; }
     public ConsoleLogStream? Stream { get; init; }
     public string? Query { get; init; }
+    public string? WorkflowInstanceId { get; init; }
     public DateTimeOffset? From { get; init; }
     public DateTimeOffset? To { get; init; }
     public int? Limit { get; init; }

src/modules/Elsa.Diagnostics.ConsoleLogs/Models/ConsoleLogLine.cs

@@ -9,6 +9,7 @@ public record ConsoleLogLine
     public ConsoleLogStream Stream { get; init; }
     public string Text { get; init; } = string.Empty;
     public ConsoleLogSource Source { get; init; } = default!;
+    public string? WorkflowInstanceId { get; init; }
     public bool Truncated { get; init; }
     public ConsoleLogDroppedSummary? Dropped { get; init; }
 }

src/modules/Elsa.Diagnostics.ConsoleLogs/Options/ConsoleLogsOptions.cs

@@ -8,7 +8,13 @@ public class ConsoleLogsOptions
     public int MaxRecentQuerySize { get; set; } = 1_000;
     public int MaxLineLength { get; set; } = 16_384;
     public TimeSpan IdleFlushTimeout { get; set; } = TimeSpan.FromSeconds(1);
-    public bool StripAnsiEscapeSequences { get; set; } = true;
+
+    /// <summary>
+    /// When true, ANSI escape sequences (colors, cursor commands, etc.) are removed before the captured
+    /// line is published. The default is <c>false</c> so the wire payload preserves the original bytes the
+    /// process wrote; consumers (e.g., the Studio) decide whether to render or strip them client-side.
+    /// </summary>
+    public bool StripAnsiEscapeSequences { get; set; } = false;
     public TimeSpan SourceHeartbeatTimeout { get; set; } = TimeSpan.FromSeconds(30);
     public bool IncludeConsoleLogsInternalLogs { get; set; }
     public string RedactionReplacement { get; set; } = "[Redacted]";

src/modules/Elsa.Diagnostics.ConsoleLogs/README.md

@@ -22,7 +22,7 @@ services.AddElsa(elsa =>
         options.SubscriberChannelCapacity = 1_000;
         options.MaxRecentQuerySize = 1_000;
         options.MaxLineLength = 16_384;
-        options.StripAnsiEscapeSequences = true;
+        options.StripAnsiEscapeSequences = false; // pass colors through to consumers
     });
 });
 ```
@@ -33,6 +33,31 @@ Map the live hub:
 app.UseConsoleLogs();
 ```
 
+## Capturing Colors
+
+The capture passes raw stdout/stderr bytes through to consumers. ANSI escape sequences (colors, cursor
+moves) are preserved by default; consumers (e.g. the Studio's `Raw ANSI` toggle) decide whether to render
+them or strip them on display.
+
+Whether colors actually arrive at the capture depends on whether the .NET console logger emits them.
+`Microsoft.Extensions.Logging.Console`'s `SimpleConsoleFormatter` defaults to `LoggerColorBehavior.Default`,
+which suppresses colors whenever `Console.IsOutputRedirected` is `true` — that includes most Docker/Kubernetes
+deployments, IDE-hosted runs (JetBrains Rider, VS Code), and any piped stdout. To force colors regardless,
+add the following to `appsettings.json` (no module dependency required):
+
+```json
+"Logging": {
+  "Console": {
+    "FormatterOptions": {
+      "ColorBehavior": "Enabled"
+    }
+  }
+}
+```
+
+With `ColorBehavior = Enabled`, the logger writes ANSI sequences such as `\x1b[32m` and `\x1b[0m` into the
+tee, the capture publishes them verbatim, and the Studio renders them.
+
 ## Contracts
 
 - Recent lines: `POST /diagnostics/console-logs/recent`
@@ -45,7 +70,7 @@ Recent, source, and hub access all require the same permission.
 ## Safety Boundaries
 
 - Redaction runs before recent buffering, live streaming, endpoint responses, and provider storage.
-- ANSI escape sequences are stripped by default.
+- ANSI escape sequences are preserved by default; set `StripAnsiEscapeSequences = true` to strip them server-side.
 - Partial writes are buffered until newline, max line length, or idle flush.
 - Oversized lines are truncated to one event and marked as truncated.
 - Providers receive only redacted line text and redacted source metadata.

src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleAnsiEscapeSequences.cs

@@ -0,0 +1,10 @@
+using System.Text.RegularExpressions;
+
+namespace Elsa.Diagnostics.ConsoleLogs.Services;
+
+internal static class ConsoleAnsiEscapeSequences
+{
+    private static readonly Regex AnsiRegex = new(@"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])", RegexOptions.Compiled | RegexOptions.CultureInvariant);
+
+    public static string Strip(string text) => AnsiRegex.Replace(text, string.Empty);
+}

src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleCaptureTee.cs

@@ -1,151 +1,179 @@
-using System.Text;
 using System.Threading.Channels;
 using Microsoft.Extensions.Options;
 
 namespace Elsa.Diagnostics.ConsoleLogs.Services;
 
-public class ConsoleCaptureTee(
+/// <summary>
-    IConsoleLogProvider provider,
+/// Process-wide adapter that turns raw stdout/stderr writes captured by <see cref="ConsoleStreamHook"/>
-    IConsoleLogSourceRegistry sourceRegistry,
+/// into fully formed <see cref="ConsoleLogLine"/> records, then publishes them to the configured
-    IConsoleLogRedactor redactor,
+/// <see cref="IConsoleLogProvider"/> on a single background task.
-    ConsoleLineFormatter formatter,
+///
-    IOptions<ConsoleLogsOptions> options) : TextWriter, IConsoleLogCapture
+/// <para>
+/// Owned by <see cref="ConsoleLogsHost"/> — there is one instance per process. Capture is wired up in the
+/// constructor so that lines emitted during host startup — before any hosted service has run — are not
+/// lost. The capture pipeline can never block the caller of <c>Console.Write*</c>: when the publish
+/// channel is full, the current line is reported as dropped to the provider.
+/// </para>
+///
+/// <para>
+/// To prevent the publish path itself from feeding new lines back into capture (e.g. when the in-memory
+/// provider's SignalR fan-out logs internally), <see cref="ConsoleStreamHook.SuppressCapture"/> is set on
+/// the publisher's async context for the lifetime of each <see cref="IConsoleLogProvider.PublishAsync"/>
+/// invocation.
+/// </para>
+/// </summary>
+public sealed class ConsoleCaptureTee : IAsyncDisposable, IDisposable
 {
-    private readonly object _lock = new();
+    private readonly IConsoleLogProvider _provider;
-    private readonly ConsoleLogsOptions _options = options.Value;
+    private readonly IConsoleLogSourceRegistry _sourceRegistry;
-    private readonly ConsoleLineBuffer _stdoutBuffer = new(options);
+    private readonly IConsoleLogRedactor _redactor;
-    private readonly ConsoleLineBuffer _stderrBuffer = new(options);
+    private readonly ConsoleLineFormatter _formatter;
-    private TextWriter? _originalOut;
+    private readonly ConsoleLogScopeAccessor _scopeAccessor;
-    private TextWriter? _originalError;
+    private readonly ConsoleLineBuffer _stdoutBuffer;
-    private Channel<ConsoleLogLine>? _publishChannel;
+    private readonly ConsoleLineBuffer _stderrBuffer;
-    private CancellationTokenSource? _publishCancellation;
+    private readonly object _bufferLock = new();
-    private Task? _publishTask;
+    private readonly Channel<ConsoleLogLine> _publishChannel;
-    private int _startCount;
+    private readonly CancellationTokenSource _shutdownCts = new();
+    private readonly Task _publishTask;
+    private readonly IDisposable _subscription;
     private long _sequence;
+    private int _disposed;
 
-    public override Encoding Encoding => _originalOut?.Encoding ?? Encoding.UTF8;
+    public ConsoleCaptureTee(
-
+        IConsoleLogProvider provider,
-    public ValueTask StartAsync(CancellationToken cancellationToken = default)
+        IConsoleLogSourceRegistry sourceRegistry,
+        IConsoleLogRedactor redactor,
+        ConsoleLineFormatter formatter,
+        ConsoleLogScopeAccessor scopeAccessor,
+        IOptions<ConsoleLogsOptions> options)
     {
-        lock (_lock)
+        _provider = provider;
+        _sourceRegistry = sourceRegistry;
+        _redactor = redactor;
+        _formatter = formatter;
+        _scopeAccessor = scopeAccessor;
+        _stdoutBuffer = new(options);
+        _stderrBuffer = new(options);
+
+        _publishChannel = Channel.CreateBounded<ConsoleLogLine>(new BoundedChannelOptions(Math.Max(1, options.Value.CaptureChannelCapacity))
         {
-            if (_startCount++ > 0)
+            FullMode = BoundedChannelFullMode.Wait,
-                return ValueTask.CompletedTask;
+            SingleReader = true,
+            SingleWriter = false
+        });
 
-            _publishCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+        _publishTask = Task.Run(() => PublishQueuedLinesAsync(_publishChannel.Reader, _shutdownCts.Token));
-            _publishChannel = Channel.CreateBounded<ConsoleLogLine>(new BoundedChannelOptions(Math.Max(1, _options.CaptureChannelCapacity))
-            {
-                FullMode = BoundedChannelFullMode.Wait,
-                SingleReader = true,
-                SingleWriter = false
-            });
-            _publishTask = PublishQueuedLinesAsync(_publishChannel.Reader, _publishCancellation.Token);
-            _originalOut = Console.Out;
-            _originalError = Console.Error;
-            Console.SetOut(new TeeWriter(_originalOut, this, ConsoleLogStream.Stdout));
-            Console.SetError(new TeeWriter(_originalError, this, ConsoleLogStream.Stderr));
-        }
 
-        return ValueTask.CompletedTask;
+        // Subscribe eagerly so that any output produced between this constructor and StartAsync is captured
-    }
+        // (the hosted service starts the flush loop, not the capture itself).
-
+        _subscription = ConsoleStreamHook.Subscribe(OnChunkCaptured);
-    public ValueTask StopAsync(CancellationToken cancellationToken = default)
-    {
-        Task? publishTask;
-        CancellationTokenSource? publishCancellation;
-
-        lock (_lock)
-        {
-            if (_startCount == 0)
-                return ValueTask.CompletedTask;
-
-            if (--_startCount > 0)
-                return ValueTask.CompletedTask;
-
-            if (_originalOut != null)
-                Console.SetOut(_originalOut);
-
-            if (_originalError != null)
-                Console.SetError(_originalError);
-
-            FlushRemaining(ConsoleLogStream.Stdout);
-            FlushRemaining(ConsoleLogStream.Stderr);
-            _publishChannel?.Writer.TryComplete();
-            publishTask = _publishTask;
-            publishCancellation = _publishCancellation;
-            _originalOut = null;
-            _originalError = null;
-            _publishChannel = null;
-            _publishCancellation = null;
-            _publishTask = null;
-        }
-
-        return AwaitPublisherAsync(publishTask, publishCancellation, cancellationToken);
-    }
-
-    public override void Write(char value)
-    {
-        Capture(ConsoleLogStream.Stdout, value.ToString());
-    }
-
-    public override ValueTask DisposeAsync()
-    {
-        return StopAsync();
     }
 
     public ValueTask FlushIdleAsync(CancellationToken cancellationToken = default)
     {
-        foreach (var stream in new[] { ConsoleLogStream.Stdout, ConsoleLogStream.Stderr })
+        var now = DateTimeOffset.UtcNow;
-        {
-            string? line;
-            lock (_lock)
-                line = GetBuffer(stream).FlushIfIdle(DateTimeOffset.UtcNow);
 
-            if (line != null)
+        Publish(ConsoleLogStream.Stdout, FlushIfIdle(_stdoutBuffer, now), now);
-                Publish(stream, line);
+        Publish(ConsoleLogStream.Stderr, FlushIfIdle(_stderrBuffer, now), now);
-        }
 
         return ValueTask.CompletedTask;
     }
 
-    private void Capture(ConsoleLogStream stream, string value)
+    public async ValueTask DisposeAsync()
     {
-        IReadOnlyCollection<string> lines;
+        await DisposeAsync(CancellationToken.None).ConfigureAwait(false);
-        lock (_lock)
+    }
-            lines = GetBuffer(stream).Append(value, DateTimeOffset.UtcNow);
+
+    public async ValueTask DisposeAsync(CancellationToken cancellationToken)
+    {
+        if (Interlocked.Exchange(ref _disposed, 1) != 0)
+            return;
+
+        _subscription.Dispose();
+
+        // Drain any pending buffered text before completing the channel.
+        FlushRemaining(ConsoleLogStream.Stdout);
+        FlushRemaining(ConsoleLogStream.Stderr);
+
+        _publishChannel.Writer.TryComplete();
+
+        try
+        {
+            await _publishTask.WaitAsync(cancellationToken).ConfigureAwait(false);
+        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            _shutdownCts.Cancel();
+        }
+        catch (OperationCanceledException)
+        {
+            // Expected during shutdown.
+        }
+
+        _shutdownCts.Dispose();
+    }
+
+    void IDisposable.Dispose()
+    {
+        // MS DI requires IDisposable on registered singletons even when the service exposes IAsyncDisposable.
+        // Bridge to DisposeAsync so the publish task is drained instead of orphaned.
+        DisposeAsync().AsTask().GetAwaiter().GetResult();
+    }
+
+    private void OnChunkCaptured(CapturedChunk chunk)
+    {
+        IReadOnlyCollection<BufferedConsoleLine> lines;
+
+        var workflowInstanceId = _scopeAccessor.GetWorkflowInstanceId();
+
+        lock (_bufferLock)
+            lines = GetBuffer(chunk.Stream).Append(chunk.Text, chunk.TimestampUtc, workflowInstanceId);
 
         foreach (var line in lines)
-            Publish(stream, line);
+            Publish(chunk.Stream, line, chunk.TimestampUtc);
+    }
+
+    private BufferedConsoleLine? FlushIfIdle(ConsoleLineBuffer buffer, DateTimeOffset now)
+    {
+        lock (_bufferLock)
+            return buffer.FlushIfIdle(now);
     }
 
     private void FlushRemaining(ConsoleLogStream stream)
     {
-        var line = GetBuffer(stream).Flush();
+        BufferedConsoleLine? line;
-        if (line != null)
+        lock (_bufferLock)
-            Publish(stream, line);
+            line = GetBuffer(stream).Flush();
+
+        Publish(stream, line, DateTimeOffset.UtcNow);
     }
 
-    private ConsoleLineBuffer GetBuffer(ConsoleLogStream stream) => stream == ConsoleLogStream.Stdout ? _stdoutBuffer : _stderrBuffer;
+    private ConsoleLineBuffer GetBuffer(ConsoleLogStream stream) =>
+        stream == ConsoleLogStream.Stdout ? _stdoutBuffer : _stderrBuffer;
 
-    private void Publish(ConsoleLogStream stream, string text)
+    private void Publish(ConsoleLogStream stream, BufferedConsoleLine? capturedLine, DateTimeOffset timestamp)
     {
-        var formatted = formatter.Format(text);
+        if (capturedLine == null)
-        var now = DateTimeOffset.UtcNow;
+            return;
+
+        var text = capturedLine.Value.Text;
+        var workflowInstanceId = capturedLine.Value.WorkflowInstanceId ?? _scopeAccessor.TakeLoggedWorkflowInstanceId(text, timestamp);
+        var formatted = _formatter.Format(text);
         var line = new ConsoleLogLine
         {
-            Timestamp = now,
+            Timestamp = timestamp,
-            ReceivedAt = now,
+            ReceivedAt = timestamp,
             Sequence = Interlocked.Increment(ref _sequence),
             Stream = stream,
             Text = formatted.Text,
-            Source = sourceRegistry.Current,
+            Source = _sourceRegistry.Current,
+            WorkflowInstanceId = workflowInstanceId,
             Truncated = formatted.Truncated
         };
 
-        if (_publishChannel?.Writer.TryWrite(line) != false)
+        // The capture path must not block console writers. When the bounded channel is full, drop this line
-            return;
+        // and surface a summary so the UI can warn the operator.
-
+        if (!_publishChannel.Writer.TryWrite(line) && _provider is IConsoleLogDroppedLineReporter reporter)
-        if (provider is IConsoleLogDroppedLineReporter reporter)
             reporter.ReportDropped(new ConsoleLogDroppedSummary(line.Source.Id, stream, "CaptureChannelFull", 1));
     }
 
@@ -153,11 +181,14 @@ public class ConsoleCaptureTee(
     {
         try
         {
-            await foreach (var line in reader.ReadAllAsync(cancellationToken))
+            await foreach (var line in reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
             {
+                // Suppress capture on this async context so that any writes the provider performs
+                // (directly or via SignalR fan-out it invokes synchronously) do not loop back into us.
+                ConsoleStreamHook.SuppressCapture = true;
                 try
                 {
-                    await provider.PublishAsync(redactor.Redact(line), cancellationToken);
+                    await _provider.PublishAsync(_redactor.Redact(line), cancellationToken).ConfigureAwait(false);
                 }
                 catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
                 {
@@ -165,54 +196,17 @@ public class ConsoleCaptureTee(
                 }
                 catch (Exception)
                 {
-                    // Avoid logging from the console capture path; doing so would recurse through the same tee.
+                    // Never log from the publish path — that would route right back through the same tee.
+                }
+                finally
+                {
+                    ConsoleStreamHook.SuppressCapture = false;
                 }
             }
         }
         catch (OperationCanceledException)
         {
-            // Normal shutdown completes or cancels the publish pump.
+            // Normal shutdown.
-        }
-    }
-
-    private static async ValueTask AwaitPublisherAsync(Task? publishTask, CancellationTokenSource? publishCancellation, CancellationToken cancellationToken)
-    {
-        if (publishTask == null)
-            return;
-
-        using var cancellation = publishCancellation;
-
-        try
-        {
-            await publishTask.WaitAsync(cancellationToken);
-        }
-        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
-        {
-            cancellation?.Cancel();
-            throw;
-        }
-    }
-
-    private sealed class TeeWriter(TextWriter original, ConsoleCaptureTee capture, ConsoleLogStream stream) : TextWriter
-    {
-        public override Encoding Encoding => original.Encoding;
-
-        public override void Write(char value)
-        {
-            original.Write(value);
-            capture.Capture(stream, value.ToString());
-        }
-
-        public override void Write(string? value)
-        {
-            original.Write(value);
-            if (value != null)
-                capture.Capture(stream, value);
-        }
-
-        public override void Flush()
-        {
-            original.Flush();
         }
     }
 }

src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLineBuffer.cs

@@ -3,37 +3,56 @@ using Microsoft.Extensions.Options;
 
 namespace Elsa.Diagnostics.ConsoleLogs.Services;
 
+/// <summary>
+/// Reassembles arbitrary stdout/stderr write chunks into complete lines. Lines are emitted whenever a
+/// terminator (<c>\n</c>) is encountered, when the buffer reaches <see cref="ConsoleLogsOptions.MaxLineLength"/>,
+/// or when <see cref="FlushIfIdle"/> is called after the configured idle timeout has elapsed.
+/// Empty lines (a bare <c>\n</c> with no preceding content on the current logical line) are preserved.
+/// </summary>
 public class ConsoleLineBuffer(IOptions<ConsoleLogsOptions> options)
 {
     private readonly StringBuilder _buffer = new();
     private readonly ConsoleLogsOptions _options = options.Value;
     private DateTimeOffset? _lastWriteAt;
+    private bool _logicalLineHasContent;
+    private bool _workflowInstanceIdCaptured;
+    private string? _workflowInstanceId;
 
-    public IReadOnlyCollection<string> Append(string value, DateTimeOffset now)
+    public IReadOnlyCollection<BufferedConsoleLine> Append(string value, DateTimeOffset now, string? workflowInstanceId = null)
     {
-        _lastWriteAt = now;
+        var lines = new List<BufferedConsoleLine>();
-        var lines = new List<string>();
 
-        foreach (var ch in value.Where(x => x != '\r'))
+        foreach (var ch in value)
         {
+            if (ch == '\r')
+                continue;
+
             if (ch == '\n')
             {
                 if (_buffer.Length > 0)
                     lines.Add(FlushBuffer());
+                else if (!_logicalLineHasContent)
+                    lines.Add(new(string.Empty, NormalizeWorkflowInstanceId(workflowInstanceId)));
 
+                _logicalLineHasContent = false;
                 continue;
             }
 
+            CaptureScope(workflowInstanceId);
             _buffer.Append(ch);
+            _logicalLineHasContent = true;
 
             if (_buffer.Length >= _options.MaxLineLength)
                 lines.Add(FlushBuffer());
         }
 
+        // Track last activity only when characters remain pending so that FlushIfIdle has a deadline to act on.
+        _lastWriteAt = _buffer.Length > 0 ? now : null;
+
         return lines;
     }
 
-    public string? FlushIfIdle(DateTimeOffset now)
+    public BufferedConsoleLine? FlushIfIdle(DateTimeOffset now)
     {
         if (_buffer.Length == 0 || _lastWriteAt == null)
             return null;
@@ -41,16 +60,32 @@ public class ConsoleLineBuffer(IOptions<ConsoleLogsOptions> options)
         return now - _lastWriteAt >= _options.IdleFlushTimeout ? FlushBuffer() : null;
     }
 
-    public string? Flush()
+    public BufferedConsoleLine? Flush()
     {
         return _buffer.Length == 0 ? null : FlushBuffer();
     }
 
-    private string FlushBuffer()
+    private void CaptureScope(string? workflowInstanceId)
+    {
+        if (_workflowInstanceIdCaptured)
+            return;
+
+        _workflowInstanceIdCaptured = true;
+        _workflowInstanceId = NormalizeWorkflowInstanceId(workflowInstanceId);
+    }
+
+    private static string? NormalizeWorkflowInstanceId(string? workflowInstanceId) =>
+        string.IsNullOrWhiteSpace(workflowInstanceId) ? null : workflowInstanceId;
+
+    private BufferedConsoleLine FlushBuffer()
     {
         var line = _buffer.ToString();
+        var workflowInstanceId = _workflowInstanceId;
         _buffer.Clear();
-        _lastWriteAt = null;
+        _workflowInstanceIdCaptured = false;
-        return line;
+        _workflowInstanceId = null;
+        return new(line, workflowInstanceId);
     }
 }
+
+public readonly record struct BufferedConsoleLine(string Text, string? WorkflowInstanceId);

src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLineFormatter.cs

@@ -1,17 +1,15 @@
-using System.Text.RegularExpressions;
 using Microsoft.Extensions.Options;
 
 namespace Elsa.Diagnostics.ConsoleLogs.Services;
 
 public class ConsoleLineFormatter(IOptions<ConsoleLogsOptions> options)
 {
-    private static readonly Regex AnsiRegex = new(@"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])", RegexOptions.Compiled | RegexOptions.CultureInvariant);
     private readonly ConsoleLogsOptions _options = options.Value;
 
     public FormattedConsoleLine Format(string text)
     {
         if (_options.StripAnsiEscapeSequences)
-            text = AnsiRegex.Replace(text, string.Empty);
+            text = ConsoleAnsiEscapeSequences.Strip(text);
 
         var truncated = text.Length > _options.MaxLineLength;
         if (truncated)

src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogCaptureAdapter.cs

@@ -0,0 +1,22 @@
+using Elsa.Diagnostics.ConsoleLogs.Contracts;
+
+namespace Elsa.Diagnostics.ConsoleLogs.Services;
+
+#pragma warning disable CS0618
+[Obsolete("Console log capture is host-owned. Use ConsoleLogsHost.Capture for host-level operations.")]
+public sealed class ConsoleLogCaptureAdapter : IConsoleLogCapture
+#pragma warning restore CS0618
+{
+    public ValueTask StartAsync(CancellationToken cancellationToken = default)
+    {
+        ConsoleLogsHost.EnsureInitialized();
+        return ValueTask.CompletedTask;
+    }
+
+    public ValueTask StopAsync(CancellationToken cancellationToken = default) => ValueTask.CompletedTask;
+
+    public ValueTask FlushIdleAsync(CancellationToken cancellationToken = default) =>
+        ConsoleLogsHost.Capture.FlushIdleAsync(cancellationToken);
+
+    public ValueTask DisposeAsync() => ValueTask.CompletedTask;
+}

src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogCaptureHostedService.cs

@@ -1,30 +0,0 @@
-using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Options;
-
-namespace Elsa.Diagnostics.ConsoleLogs.Services;
-
-public class ConsoleLogCaptureHostedService(IConsoleLogCapture capture, IOptions<ConsoleLogsOptions> options) : BackgroundService
-{
-    public override async Task StartAsync(CancellationToken cancellationToken)
-    {
-        await capture.StartAsync(cancellationToken);
-        await base.StartAsync(cancellationToken);
-    }
-
-    public override async Task StopAsync(CancellationToken cancellationToken)
-    {
-        await base.StopAsync(cancellationToken);
-        await capture.StopAsync(cancellationToken);
-    }
-
-    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
-    {
-        var interval = TimeSpan.FromMilliseconds(Math.Max(100, options.Value.IdleFlushTimeout.TotalMilliseconds / 2));
-
-        while (!stoppingToken.IsCancellationRequested)
-        {
-            await Task.Delay(interval, stoppingToken);
-            await capture.FlushIdleAsync(stoppingToken);
-        }
-    }
-}

src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogFilterEvaluator.cs

@@ -13,6 +13,9 @@ public static class ConsoleLogFilterEvaluator
         if (!ContainsText(line, filter.Query))
             return false;
 
+        if (!EqualsFilter(line.WorkflowInstanceId, filter.WorkflowInstanceId))
+            return false;
+
         if (filter.From is { } from && line.ReceivedAt < from)
             return false;
 

src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogRedactor.cs

@@ -49,8 +49,22 @@ public class ConsoleLogRedactor(IOptions<ConsoleLogsOptions> options) : IConsole
         if (IsSensitiveName(name))
             return _options.RedactionReplacement;
 
-        return _sensitiveTextPatterns.Aggregate(value, (current, pattern) => pattern.Replace(current, _options.RedactionReplacement));
+        return RedactSensitiveText(value);
     }
 
+    private string RedactSensitiveText(string value)
+    {
+        var redacted = ApplySensitiveTextPatterns(value);
+        var normalized = ConsoleAnsiEscapeSequences.Strip(value);
+        if (normalized == value)
+            return redacted;
+
+        var normalizedRedacted = ApplySensitiveTextPatterns(normalized);
+        return normalizedRedacted == normalized ? redacted : normalizedRedacted;
+    }
+
+    private string ApplySensitiveTextPatterns(string value) =>
+        _sensitiveTextPatterns.Aggregate(value, (current, pattern) => pattern.Replace(current, _options.RedactionReplacement));
+
     private bool IsSensitiveName(string name) => _sensitiveNames.Any(sensitiveName => name.Contains(sensitiveName, StringComparison.OrdinalIgnoreCase));
 }

src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogScopeAccessor.cs

@@ -0,0 +1,217 @@
+using System.Collections;
+using System.Linq;
+using Microsoft.Extensions.Logging;
+
+namespace Elsa.Diagnostics.ConsoleLogs.Services;
+
+public sealed class ConsoleLogScopeAccessor : ILoggerProvider, ISupportExternalScope
+{
+    private const string WorkflowInstanceIdKey = "WorkflowInstanceId";
+    private const int PendingWorkflowLogsCapacity = 1024;
+    private static readonly TimeSpan PendingWorkflowLogTtl = TimeSpan.FromSeconds(30);
+    private readonly IExternalScopeProvider _defaultScopeProvider = new LoggerExternalScopeProvider();
+    private readonly object _lock = new();
+    private readonly object _pendingWorkflowLogsLock = new();
+    private readonly List<PendingWorkflowLog> _pendingWorkflowLogs = [];
+    private WeakReference<IExternalScopeProvider>[] _scopeProviders;
+
+    public ConsoleLogScopeAccessor()
+    {
+        _scopeProviders = [new(_defaultScopeProvider)];
+    }
+
+    public ILogger CreateLogger(string categoryName) => new ScopeCapturingLogger(this);
+    public void Dispose()
+    {
+    }
+
+    public void SetScopeProvider(IExternalScopeProvider scopeProvider)
+    {
+        lock (_lock)
+        {
+            var scopeProviders = GetLiveScopeProviders();
+            if (scopeProviders.Any(x => ReferenceEquals(x, scopeProvider)))
+                return;
+
+            _scopeProviders = [.. _scopeProviders, new WeakReference<IExternalScopeProvider>(scopeProvider)];
+        }
+    }
+
+    public string? GetWorkflowInstanceId()
+    {
+        var scopeProviders = GetLiveScopeProviders();
+        string? value = null;
+
+        foreach (var scopeProvider in scopeProviders)
+        {
+            var state = new ScopeSearchState(WorkflowInstanceIdKey);
+            scopeProvider.ForEachScope((scope, searchState) => searchState.Capture(scope), state);
+            if (state.Value != null)
+                value = state.Value;
+        }
+
+        return value;
+    }
+
+    internal string? TakeLoggedWorkflowInstanceId(string text, DateTimeOffset capturedAt)
+    {
+        if (string.IsNullOrWhiteSpace(text))
+            return null;
+
+        lock (_pendingWorkflowLogsLock)
+        {
+            PruneExpiredPendingWorkflowLogs(capturedAt);
+
+            var index = _pendingWorkflowLogs.FindIndex(x => IsCorrelated(text, x.Message));
+            if (index < 0)
+                return null;
+
+            var workflowInstanceId = _pendingWorkflowLogs[index].WorkflowInstanceId;
+            _pendingWorkflowLogs.RemoveAt(index);
+            return workflowInstanceId;
+        }
+    }
+
+    private IExternalScopeProvider[] GetLiveScopeProviders()
+    {
+        lock (_lock)
+        {
+            var liveScopeProviders = _scopeProviders
+                .Select(reference => (Reference: reference, Provider: reference.TryGetTarget(out var provider) ? provider : null))
+                .Where(x => x.Provider != null)
+                .ToArray();
+
+            if (liveScopeProviders.Length != _scopeProviders.Length)
+                _scopeProviders = [.. liveScopeProviders.Select(x => x.Reference)];
+
+            return [.. liveScopeProviders.Select(x => x.Provider!)];
+        }
+    }
+
+    private sealed class ScopeSearchState(string key)
+    {
+        public string? Value { get; private set; }
+
+        public void Capture(object? scope)
+        {
+            var value = TryGetScopeValue(scope, key);
+            if (value != null)
+                Value = value;
+        }
+    }
+
+    private static string? TryGetScopeValue(object? scope, string key)
+    {
+        if (scope is not IEnumerable values)
+            return null;
+
+        foreach (var value in values)
+        {
+            if (value == null)
+                continue;
+
+            if (value is KeyValuePair<string, object?> kvp)
+            {
+                if (string.Equals(kvp.Key, key, StringComparison.OrdinalIgnoreCase))
+                    return kvp.Value?.ToString();
+
+                continue;
+            }
+
+            var type = value.GetType();
+            if (!type.IsGenericType || type.GetGenericTypeDefinition() != typeof(KeyValuePair<,>))
+                continue;
+
+            var itemKey = type.GetProperty(nameof(KeyValuePair<string, object?>.Key))?.GetValue(value) as string;
+            if (!string.Equals(itemKey, key, StringComparison.OrdinalIgnoreCase))
+                continue;
+
+            return type.GetProperty(nameof(KeyValuePair<string, object?>.Value))?.GetValue(value)?.ToString();
+        }
+
+        return null;
+    }
+
+    private void CaptureLoggedWorkflowInstanceId<TState>(
+        TState state,
+        Exception? exception,
+        Func<TState, Exception?, string> formatter)
+    {
+        var workflowInstanceId = GetWorkflowInstanceId();
+        if (string.IsNullOrWhiteSpace(workflowInstanceId))
+            return;
+
+        string message;
+        try
+        {
+            message = formatter(state, exception);
+        }
+        catch (Exception)
+        {
+            return;
+        }
+
+        if (string.IsNullOrWhiteSpace(message))
+            return;
+
+        var now = DateTimeOffset.UtcNow;
+
+        lock (_pendingWorkflowLogsLock)
+        {
+            PruneExpiredPendingWorkflowLogs(now);
+
+            while (_pendingWorkflowLogs.Count >= PendingWorkflowLogsCapacity)
+                _pendingWorkflowLogs.RemoveAt(0);
+
+            _pendingWorkflowLogs.Add(new PendingWorkflowLog(message, workflowInstanceId, now));
+        }
+    }
+
+    private void PruneExpiredPendingWorkflowLogs(DateTimeOffset now)
+    {
+        for (var i = _pendingWorkflowLogs.Count - 1; i >= 0; i--)
+        {
+            if (now - _pendingWorkflowLogs[i].CapturedAt <= PendingWorkflowLogTtl)
+                continue;
+
+            _pendingWorkflowLogs.RemoveAt(i);
+        }
+    }
+
+    private static bool IsCorrelated(string capturedLine, string loggedMessage)
+    {
+        var trimmedLine = capturedLine.Trim();
+        var trimmedMessage = loggedMessage.Trim();
+
+        return trimmedLine.Equals(trimmedMessage, StringComparison.Ordinal);
+    }
+
+    private sealed class ScopeCapturingLogger(ConsoleLogScopeAccessor accessor) : ILogger
+    {
+        public IDisposable? BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;
+
+        public bool IsEnabled(LogLevel logLevel) => logLevel != LogLevel.None;
+
+        public void Log<TState>(
+            LogLevel logLevel,
+            EventId eventId,
+            TState state,
+            Exception? exception,
+            Func<TState, Exception?, string> formatter)
+        {
+            if (IsEnabled(logLevel))
+                accessor.CaptureLoggedWorkflowInstanceId(state, exception, formatter);
+        }
+    }
+
+    private sealed record PendingWorkflowLog(string Message, string WorkflowInstanceId, DateTimeOffset CapturedAt);
+
+    private sealed class NullScope : IDisposable
+    {
+        public static NullScope Instance { get; } = new();
+
+        public void Dispose()
+        {
+        }
+    }
+}

src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsHost.cs

@@ -0,0 +1,194 @@
+using Elsa.Diagnostics.ConsoleLogs.Providers.InMemory;
+using Microsoft.Extensions.Options;
+
+namespace Elsa.Diagnostics.ConsoleLogs.Services;
+
+/// <summary>
+/// Process-wide owner of the console capture pipeline. Console output is, by nature, a single shared OS
+/// resource — there is exactly one stdout and one stderr per process — so the capture, the in-memory ring
+/// buffer, and the source registry are likewise single instances regardless of how many DI containers
+/// (shells, tenants) want to consume them. This class is the seam that makes that fact explicit.
+///
+/// <para>
+/// Configuration follows a first-wins model: <see cref="Configure"/> calls made before the host is first
+/// accessed are applied to a single <see cref="ConsoleLogsOptions"/> instance. Subsequent calls are
+/// ignored because the configuration would not apply to anything already captured. Call
+/// <c>AddConsoleLogsHost</c> from <c>Program.cs</c> to set host-wide options deterministically.
+/// </para>
+///
+/// <para>
+/// Initialization is lazy and thread-safe: the first access from any thread builds the pipeline. The host
+/// also auto-installs <see cref="ConsoleStreamHook"/>, so simply touching <see cref="Provider"/>,
+/// <see cref="SourceRegistry"/>, or <see cref="Capture"/> is enough to start capturing.
+/// </para>
+/// </summary>
+public static class ConsoleLogsHost
+{
+    private static readonly object Lock = new();
+    private static Action<ConsoleLogsOptions>? _pendingConfigure;
+    private static Func<ConsoleLogsProviderContext, IConsoleLogProvider>? _providerFactory;
+    private static Lazy<HostState> _state = CreateLazy();
+    private static int _leases;
+
+    /// <summary>
+    /// Applies a configuration delegate to the host options if the host has not yet been initialized.
+    /// Calls after initialization are ignored to preserve consistency between configuration and captured
+    /// data. Returns <c>true</c> when the delegate was accepted, <c>false</c> when the host had already
+    /// initialized.
+    /// </summary>
+    public static bool Configure(Action<ConsoleLogsOptions> configure)
+    {
+        ArgumentNullException.ThrowIfNull(configure);
+
+        lock (Lock)
+        {
+            if (_state.IsValueCreated)
+                return false;
+
+            _pendingConfigure = (Action<ConsoleLogsOptions>?)Delegate.Combine(_pendingConfigure, configure);
+            return true;
+        }
+    }
+
+    /// <summary>
+    /// Configures the provider used by the process-wide capture pipeline. The factory must be registered
+    /// before the host is initialized.
+    /// </summary>
+    public static bool ConfigureProvider(Func<IOptions<ConsoleLogsOptions>, IConsoleLogSourceRegistry, IConsoleLogProvider> providerFactory)
+    {
+        ArgumentNullException.ThrowIfNull(providerFactory);
+        return ConfigureProvider(context => providerFactory(context.Options, context.SourceRegistry));
+    }
+
+    internal static bool ConfigureProvider(Func<ConsoleLogsProviderContext, IConsoleLogProvider> providerFactory)
+    {
+        ArgumentNullException.ThrowIfNull(providerFactory);
+
+        lock (Lock)
+        {
+            if (_state.IsValueCreated || _providerFactory != null)
+                return false;
+
+            _providerFactory = providerFactory;
+            return true;
+        }
+    }
+
+    public static IOptions<ConsoleLogsOptions> Options => _state.Value.Options;
+    public static IConsoleLogSourceRegistry SourceRegistry => _state.Value.SourceRegistry;
+    public static IConsoleLogRedactor Redactor => _state.Value.Redactor;
+    public static ConsoleLineFormatter Formatter => _state.Value.Formatter;
+    public static ConsoleLogScopeAccessor ScopeAccessor => _state.Value.ScopeAccessor;
+    public static IConsoleLogProvider Provider => _state.Value.Provider;
+    public static ConsoleCaptureTee Capture => _state.Value.Capture;
+
+    /// <summary>
+    /// Triggers initialization without resolving any specific component. Returns <c>true</c> when this call
+    /// performed the initialization, <c>false</c> when the host was already initialized.
+    /// </summary>
+    public static bool EnsureInitialized()
+    {
+        var wasCreated = _state.IsValueCreated;
+        _ = _state.Value;
+        return !wasCreated;
+    }
+
+    public static void AddReference()
+    {
+        lock (Lock)
+            _leases++;
+    }
+
+    public static ValueTask ReleaseReferenceAsync(CancellationToken cancellationToken = default)
+    {
+        Lazy<HostState> state;
+
+        lock (Lock)
+        {
+            if (_leases <= 0)
+                return ValueTask.CompletedTask;
+
+            _leases--;
+            if (_leases > 0)
+                return ValueTask.CompletedTask;
+
+            state = ResetState();
+        }
+
+        return DisposeStateAsync(state, cancellationToken);
+    }
+
+    /// <summary>
+    /// Tears down the host pipeline. Intended for tests and for orderly host shutdown.
+    /// </summary>
+    public static async ValueTask ShutdownAsync(CancellationToken cancellationToken = default)
+    {
+        Lazy<HostState> state;
+
+        lock (Lock)
+        {
+            state = ResetState();
+            _leases = 0;
+        }
+
+        await DisposeStateAsync(state, cancellationToken).ConfigureAwait(false);
+    }
+
+    private static Lazy<HostState> CreateLazy() => new(BuildState, LazyThreadSafetyMode.ExecutionAndPublication);
+
+    private static Lazy<HostState> ResetState()
+    {
+        var state = _state;
+        _state = CreateLazy();
+        _pendingConfigure = null;
+        _providerFactory = null;
+        return state;
+    }
+
+    private static async ValueTask DisposeStateAsync(Lazy<HostState> state, CancellationToken cancellationToken)
+    {
+        if (state.IsValueCreated)
+            await state.Value.Capture.DisposeAsync(cancellationToken).ConfigureAwait(false);
+    }
+
+    private static HostState BuildState()
+    {
+        var options = new ConsoleLogsOptions();
+
+        Func<ConsoleLogsProviderContext, IConsoleLogProvider>? providerFactory;
+
+        lock (Lock)
+        {
+            _pendingConfigure?.Invoke(options);
+            providerFactory = _providerFactory;
+        }
+
+        var wrappedOptions = Microsoft.Extensions.Options.Options.Create(options);
+        var sourceRegistry = new ConsoleLogSourceRegistry(wrappedOptions);
+        var redactor = new ConsoleLogRedactor(wrappedOptions);
+        var formatter = new ConsoleLineFormatter(wrappedOptions);
+        var scopeAccessor = new ConsoleLogScopeAccessor();
+        var providerContext = new ConsoleLogsProviderContext(wrappedOptions, sourceRegistry, redactor, formatter, scopeAccessor);
+        var provider = providerFactory?.Invoke(providerContext) ?? new InMemoryConsoleLogProvider(wrappedOptions, sourceRegistry);
+        var capture = new ConsoleCaptureTee(provider, sourceRegistry, redactor, formatter, scopeAccessor, wrappedOptions);
+
+        return new HostState(wrappedOptions, sourceRegistry, redactor, formatter, scopeAccessor, provider, capture);
+    }
+
+    private sealed record HostState(
+        IOptions<ConsoleLogsOptions> Options,
+        IConsoleLogSourceRegistry SourceRegistry,
+        IConsoleLogRedactor Redactor,
+        ConsoleLineFormatter Formatter,
+        ConsoleLogScopeAccessor ScopeAccessor,
+        IConsoleLogProvider Provider,
+        ConsoleCaptureTee Capture);
+}
+
+internal sealed record ConsoleLogsProviderContext(
+    IOptions<ConsoleLogsOptions> Options,
+    IConsoleLogSourceRegistry SourceRegistry,
+    IConsoleLogRedactor Redactor,
+    ConsoleLineFormatter Formatter,
+    ConsoleLogScopeAccessor ScopeAccessor);
+

src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsHostedService.cs

@@ -0,0 +1,57 @@
+using Microsoft.Extensions.Hosting;
+
+namespace Elsa.Diagnostics.ConsoleLogs.Services;
+
+/// <summary>
+/// Host-level hosted service that ensures the console capture is initialized at host build time,
+/// periodically flushes idle partial lines, and drains pending writes during host shutdown.
+/// </summary>
+public sealed class ConsoleLogsHostedService : BackgroundService
+{
+    private readonly Action? _configureHost;
+
+    public ConsoleLogsHostedService() : this((Action?)null)
+    {
+    }
+
+    public ConsoleLogsHostedService(IServiceProvider serviceProvider) : this(() =>
+    {
+        if (serviceProvider.GetService(typeof(ConsoleLogsProviderRegistration)) is ConsoleLogsProviderRegistration registration)
+            registration.ConfigureProvider(serviceProvider);
+    })
+    {
+    }
+
+    public ConsoleLogsHostedService(Action? configureHost)
+    {
+        _configureHost = configureHost;
+    }
+
+    public override Task StartAsync(CancellationToken cancellationToken)
+    {
+        _configureHost?.Invoke();
+
+        // Touch the host so initialization happens before the first request, not lazily on first capture.
+        ConsoleLogsHost.AddReference();
+        ConsoleLogsHost.EnsureInitialized();
+        return base.StartAsync(cancellationToken);
+    }
+
+    public override async Task StopAsync(CancellationToken cancellationToken)
+    {
+        await base.StopAsync(cancellationToken).ConfigureAwait(false);
+        await ConsoleLogsHost.ReleaseReferenceAsync(cancellationToken).ConfigureAwait(false);
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        var interval = TimeSpan.FromMilliseconds(Math.Max(100, ConsoleLogsHost.Options.Value.IdleFlushTimeout.TotalMilliseconds / 2));
+
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            await Task.Delay(interval, stoppingToken).ConfigureAwait(false);
+            await ConsoleLogsHost.Capture.FlushIdleAsync(stoppingToken).ConfigureAwait(false);
+        }
+    }
+}
+

src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsProviderRegistration.cs

@@ -0,0 +1,85 @@
+using System.Diagnostics.CodeAnalysis;
+using Elsa.Diagnostics.ConsoleLogs.Contracts;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Options;
+
+namespace Elsa.Diagnostics.ConsoleLogs.Services;
+
+internal sealed class ConsoleLogsProviderRegistration(IServiceCollection services, ServiceDescriptor defaultProviderDescriptor)
+{
+    private readonly object _lock = new();
+    private ConsoleLogsProviderContext? _hostContext;
+
+    public void ConfigureProvider(IServiceProvider serviceProvider)
+    {
+        var descriptor = GetCustomProviderDescriptor();
+        if (descriptor == null)
+            return;
+
+        ConsoleLogsHost.ConfigureProvider(context => ResolveProvider(serviceProvider, context));
+    }
+
+    public bool TryGetHostOptions([NotNullWhen(true)] out IOptions<ConsoleLogsOptions>? options)
+    {
+        lock (_lock)
+        {
+            options = _hostContext?.Options;
+            return options != null;
+        }
+    }
+
+    public bool TryGetHostSourceRegistry([NotNullWhen(true)] out IConsoleLogSourceRegistry? sourceRegistry)
+    {
+        lock (_lock)
+        {
+            sourceRegistry = _hostContext?.SourceRegistry;
+            return sourceRegistry != null;
+        }
+    }
+
+    public bool TryGetHostRedactor([NotNullWhen(true)] out IConsoleLogRedactor? redactor)
+    {
+        lock (_lock)
+        {
+            redactor = _hostContext?.Redactor;
+            return redactor != null;
+        }
+    }
+
+    public bool TryGetHostFormatter([NotNullWhen(true)] out ConsoleLineFormatter? formatter)
+    {
+        lock (_lock)
+        {
+            formatter = _hostContext?.Formatter;
+            return formatter != null;
+        }
+    }
+
+    public bool TryGetHostScopeAccessor([NotNullWhen(true)] out ConsoleLogScopeAccessor? scopeAccessor)
+    {
+        lock (_lock)
+        {
+            scopeAccessor = _hostContext?.ScopeAccessor;
+            return scopeAccessor != null;
+        }
+    }
+
+    private ServiceDescriptor? GetCustomProviderDescriptor() =>
+        services.LastOrDefault(x => x.ServiceType == typeof(IConsoleLogProvider) && !ReferenceEquals(x, defaultProviderDescriptor));
+
+    private IConsoleLogProvider ResolveProvider(IServiceProvider serviceProvider, ConsoleLogsProviderContext context)
+    {
+        lock (_lock)
+            _hostContext = context;
+
+        try
+        {
+            return serviceProvider.GetRequiredService<IConsoleLogProvider>();
+        }
+        finally
+        {
+            lock (_lock)
+                _hostContext = null;
+        }
+    }
+}

src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleStreamHook.cs

@@ -0,0 +1,207 @@
+namespace Elsa.Diagnostics.ConsoleLogs.Services;
+
+/// <summary>
+/// Installs a process-wide tee on <see cref="Console.Out"/> and <see cref="Console.Error"/> so that every
+/// captured write is fanned out to all currently registered subscribers, in addition to the original writers.
+///
+/// <para>
+/// Downstream logger providers (e.g. <c>Microsoft.Extensions.Logging.Console</c>) capture <c>Console.Out</c>
+/// once at construction. Therefore the tee must be installed BEFORE any such provider is constructed —
+/// ideally as the very first line of <c>Program.cs</c>. Calling <see cref="Subscribe"/> implicitly installs
+/// the tee, so a host can also rely on subscribers attaching early in the pipeline.
+/// </para>
+///
+/// <para>
+/// Writes captured before any subscriber attaches are kept in a bounded replay buffer and delivered to the
+/// first subscriber (when <c>replayPreviousWrites</c> is true), preserving original timestamps. This bridges
+/// the gap between <see cref="Install"/> and the first <see cref="Subscribe"/> call, which typically happens
+/// when the DI container is built and the capture singleton is resolved.
+/// </para>
+///
+/// <para>
+/// Subscribers that themselves write to the console while handling a capture would cause infinite recursion.
+/// Set <see cref="SuppressCapture"/> to true on the publish path's <see cref="AsyncLocal{T}"/> context to
+/// short-circuit dispatch for the duration of that asynchronous flow.
+/// </para>
+/// </summary>
+public static class ConsoleStreamHook
+{
+    private static readonly object Lock = new();
+    private static readonly AsyncLocal<bool> SuppressFlag = new();
+    private static volatile Subscription[] _subscriptions = [];
+    private static readonly ReplayBuffer Replay = new(capacity: 4096);
+    private static volatile bool _installed;
+    private static TextWriter? _originalOut;
+    private static TextWriter? _originalError;
+
+    /// <summary>
+    /// While true on the current async context, captured writes will not be dispatched to subscribers.
+    /// Subscribers must set this to true while running their publish pipeline so that any logging triggered
+    /// by that pipeline does not feed back into capture.
+    /// </summary>
+    public static bool SuppressCapture
+    {
+        get => SuppressFlag.Value;
+        set => SuppressFlag.Value = value;
+    }
+
+    /// <summary>
+    /// Installs the tee if not already installed. Safe to call multiple times.
+    /// </summary>
+    public static void Install()
+    {
+        if (_installed)
+            return;
+
+        lock (Lock)
+        {
+            if (_installed)
+                return;
+
+            _originalOut = Console.Out;
+            _originalError = Console.Error;
+            Console.SetOut(new ConsoleStreamTeeWriter(_originalOut, ConsoleLogStream.Stdout, Dispatch));
+            Console.SetError(new ConsoleStreamTeeWriter(_originalError, ConsoleLogStream.Stderr, Dispatch));
+            _installed = true;
+        }
+    }
+
+    /// <summary>
+    /// Restores the original writers and clears all subscribers. Intended for tests.
+    /// </summary>
+    public static void Uninstall()
+    {
+        lock (Lock)
+        {
+            if (!_installed)
+                return;
+
+            if (_originalOut != null)
+                Console.SetOut(_originalOut);
+
+            if (_originalError != null)
+                Console.SetError(_originalError);
+
+            _originalOut = null;
+            _originalError = null;
+            _subscriptions = [];
+            Replay.Clear();
+            _installed = false;
+        }
+    }
+
+    /// <summary>
+    /// Registers a subscriber that receives every captured write. Disposing the returned token removes the
+    /// subscription. Calling Subscribe auto-installs the tee. When <paramref name="replayPreviousWrites"/>
+    /// is true, the buffered writes captured since install are replayed to this subscriber synchronously.
+    /// </summary>
+    public static IDisposable Subscribe(Action<CapturedChunk> handler, bool replayPreviousWrites = true)
+    {
+        ArgumentNullException.ThrowIfNull(handler);
+
+        Install();
+        var subscription = new Subscription(handler);
+
+        lock (Lock)
+        {
+            _subscriptions = [.._subscriptions, subscription];
+
+            if (replayPreviousWrites)
+            {
+                foreach (var chunk in Replay.Snapshot())
+                    InvokeSafely(subscription, chunk);
+
+                Replay.Clear();
+            }
+        }
+
+        return subscription;
+    }
+
+    internal static void Remove(Subscription subscription)
+    {
+        lock (Lock)
+        {
+            _subscriptions = _subscriptions.Where(x => !ReferenceEquals(x, subscription)).ToArray();
+        }
+    }
+
+    private static void Dispatch(ConsoleLogStream stream, string value)
+    {
+        if (SuppressFlag.Value || value.Length == 0)
+            return;
+
+        var chunk = new CapturedChunk(stream, value, DateTimeOffset.UtcNow);
+        var subscriptions = _subscriptions;
+
+        if (subscriptions.Length == 0)
+        {
+            // No subscriber yet — retain for replay to the first one that attaches.
+            Replay.Add(chunk);
+            return;
+        }
+
+        foreach (var subscription in subscriptions)
+            InvokeSafely(subscription, chunk);
+    }
+
+    private static void InvokeSafely(Subscription subscription, CapturedChunk chunk)
+    {
+        try
+        {
+            subscription.Handler(chunk);
+        }
+        catch (Exception e) when (!IsCriticalException(e))
+        {
+            // Never let a subscriber take down the writer. Subscribers are responsible for their own resilience.
+        }
+    }
+
+    private static bool IsCriticalException(Exception exception) =>
+        exception is OutOfMemoryException
+            or StackOverflowException
+            or AccessViolationException
+            or AppDomainUnloadedException
+            or BadImageFormatException
+            or CannotUnloadAppDomainException;
+
+    internal sealed class Subscription(Action<CapturedChunk> handler) : IDisposable
+    {
+        public Action<CapturedChunk> Handler { get; } = handler;
+        public void Dispose() => Remove(this);
+    }
+
+    private sealed class ReplayBuffer(int capacity)
+    {
+        private readonly object _lock = new();
+        private readonly Queue<CapturedChunk> _items = new(capacity);
+
+        public void Add(CapturedChunk chunk)
+        {
+            lock (_lock)
+            {
+                if (_items.Count == capacity)
+                    _items.Dequeue();
+
+                _items.Enqueue(chunk);
+            }
+        }
+
+        public IReadOnlyList<CapturedChunk> Snapshot()
+        {
+            lock (_lock)
+                return _items.ToArray();
+        }
+
+        public void Clear()
+        {
+            lock (_lock)
+                _items.Clear();
+        }
+    }
+}
+
+/// <summary>
+/// A single chunk of text captured from stdout/stderr together with the moment of capture.
+/// </summary>
+public readonly record struct CapturedChunk(ConsoleLogStream Stream, string Text, DateTimeOffset TimestampUtc);

src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleStreamTeeWriter.cs

@@ -0,0 +1,212 @@
+using System.Diagnostics.CodeAnalysis;
+using System.Text;
+
+namespace Elsa.Diagnostics.ConsoleLogs.Services;
+
+/// <summary>
+/// A <see cref="TextWriter"/> that forwards every write to the original underlying writer and a configurable capture sink.
+/// </summary>
+internal sealed class ConsoleStreamTeeWriter(TextWriter original, ConsoleLogStream stream, Action<ConsoleLogStream, string> capture) : TextWriter
+{
+    public TextWriter Original => original;
+
+    public override Encoding Encoding => original.Encoding;
+    public override IFormatProvider FormatProvider => original.FormatProvider;
+    [AllowNull]
+    public override string NewLine { get => original.NewLine; set => original.NewLine = value; }
+
+    // ── Write overloads ──
+
+    public override void Write(char value)
+    {
+        original.Write(value);
+        capture(stream, value.ToString());
+    }
+
+    public override void Write(char[]? buffer)
+    {
+        original.Write(buffer);
+        if (buffer is { Length: > 0 })
+            capture(stream, new string(buffer));
+    }
+
+    public override void Write(char[] buffer, int index, int count)
+    {
+        original.Write(buffer, index, count);
+        if (count > 0)
+            capture(stream, new string(buffer, index, count));
+    }
+
+    public override void Write(ReadOnlySpan<char> buffer)
+    {
+        original.Write(buffer);
+        if (!buffer.IsEmpty)
+            capture(stream, buffer.ToString());
+    }
+
+    public override void Write(bool value) { original.Write(value); capture(stream, value ? "True" : "False"); }
+    public override void Write(int value) { original.Write(value); capture(stream, value.ToString(FormatProvider)); }
+    public override void Write(uint value) { original.Write(value); capture(stream, value.ToString(FormatProvider)); }
+    public override void Write(long value) { original.Write(value); capture(stream, value.ToString(FormatProvider)); }
+    public override void Write(ulong value) { original.Write(value); capture(stream, value.ToString(FormatProvider)); }
+    public override void Write(float value) { original.Write(value); capture(stream, value.ToString(FormatProvider)); }
+    public override void Write(double value) { original.Write(value); capture(stream, value.ToString(FormatProvider)); }
+    public override void Write(decimal value) { original.Write(value); capture(stream, value.ToString(FormatProvider)); }
+
+    public override void Write(string? value)
+    {
+        original.Write(value);
+        if (value != null)
+            capture(stream, value);
+    }
+
+    public override void Write(object? value)
+    {
+        original.Write(value);
+        var text = value?.ToString();
+        if (text != null)
+            capture(stream, text);
+    }
+
+    public override void Write(StringBuilder? value)
+    {
+        original.Write(value);
+        if (value is { Length: > 0 })
+            capture(stream, value.ToString());
+    }
+
+    public override void Write(string format, object? arg0) { var text = string.Format(FormatProvider, format, arg0); original.Write(text); capture(stream, text); }
+    public override void Write(string format, object? arg0, object? arg1) { var text = string.Format(FormatProvider, format, arg0, arg1); original.Write(text); capture(stream, text); }
+    public override void Write(string format, object? arg0, object? arg1, object? arg2) { var text = string.Format(FormatProvider, format, arg0, arg1, arg2); original.Write(text); capture(stream, text); }
+    public override void Write(string format, params object?[] arg) { var text = string.Format(FormatProvider, format, arg); original.Write(text); capture(stream, text); }
+#if NET9_0_OR_GREATER
+    public override void Write(string format, params ReadOnlySpan<object?> arg) { var text = string.Format(FormatProvider, format, arg); original.Write(text); capture(stream, text); }
+#endif
+
+    // ── WriteLine overloads ──
+
+    public override void WriteLine() { original.WriteLine(); capture(stream, "\n"); }
+    public override void WriteLine(char value) { original.WriteLine(value); capture(stream, value + "\n"); }
+
+    public override void WriteLine(char[]? buffer)
+    {
+        original.WriteLine(buffer);
+        capture(stream, (buffer is { Length: > 0 } ? new string(buffer) : "") + "\n");
+    }
+
+    public override void WriteLine(char[] buffer, int index, int count)
+    {
+        original.WriteLine(buffer, index, count);
+        capture(stream, new string(buffer, index, count) + "\n");
+    }
+
+    public override void WriteLine(ReadOnlySpan<char> buffer)
+    {
+        original.WriteLine(buffer);
+        capture(stream, string.Concat(buffer, "\n"));
+    }
+
+    public override void WriteLine(bool value) { original.WriteLine(value); capture(stream, (value ? "True" : "False") + "\n"); }
+    public override void WriteLine(int value) { original.WriteLine(value); capture(stream, value.ToString(FormatProvider) + "\n"); }
+    public override void WriteLine(uint value) { original.WriteLine(value); capture(stream, value.ToString(FormatProvider) + "\n"); }
+    public override void WriteLine(long value) { original.WriteLine(value); capture(stream, value.ToString(FormatProvider) + "\n"); }
+    public override void WriteLine(ulong value) { original.WriteLine(value); capture(stream, value.ToString(FormatProvider) + "\n"); }
+    public override void WriteLine(float value) { original.WriteLine(value); capture(stream, value.ToString(FormatProvider) + "\n"); }
+    public override void WriteLine(double value) { original.WriteLine(value); capture(stream, value.ToString(FormatProvider) + "\n"); }
+    public override void WriteLine(decimal value) { original.WriteLine(value); capture(stream, value.ToString(FormatProvider) + "\n"); }
+
+    public override void WriteLine(string? value)
+    {
+        original.WriteLine(value);
+        capture(stream, (value ?? "") + "\n");
+    }
+
+    public override void WriteLine(object? value)
+    {
+        original.WriteLine(value);
+        capture(stream, (value?.ToString() ?? "") + "\n");
+    }
+
+    public override void WriteLine(StringBuilder? value)
+    {
+        original.WriteLine(value);
+        capture(stream, (value is { Length: > 0 } ? value.ToString() : "") + "\n");
+    }
+
+    public override void WriteLine(string format, object? arg0) { var text = string.Format(FormatProvider, format, arg0); original.WriteLine(text); capture(stream, text + "\n"); }
+    public override void WriteLine(string format, object? arg0, object? arg1) { var text = string.Format(FormatProvider, format, arg0, arg1); original.WriteLine(text); capture(stream, text + "\n"); }
+    public override void WriteLine(string format, object? arg0, object? arg1, object? arg2) { var text = string.Format(FormatProvider, format, arg0, arg1, arg2); original.WriteLine(text); capture(stream, text + "\n"); }
+    public override void WriteLine(string format, params object?[] arg) { var text = string.Format(FormatProvider, format, arg); original.WriteLine(text); capture(stream, text + "\n"); }
+#if NET9_0_OR_GREATER
+    public override void WriteLine(string format, params ReadOnlySpan<object?> arg) { var text = string.Format(FormatProvider, format, arg); original.WriteLine(text); capture(stream, text + "\n"); }
+#endif
+
+    // ── Async Write overloads ──
+
+    public override async Task WriteAsync(char value) { await original.WriteAsync(value); capture(stream, value.ToString()); }
+
+    public override async Task WriteAsync(string? value)
+    {
+        await original.WriteAsync(value);
+        if (value != null)
+            capture(stream, value);
+    }
+
+    public override async Task WriteAsync(char[] buffer, int index, int count)
+    {
+        await original.WriteAsync(buffer, index, count);
+        if (count > 0)
+            capture(stream, new string(buffer, index, count));
+    }
+
+    public override async Task WriteAsync(ReadOnlyMemory<char> buffer, CancellationToken cancellationToken = default)
+    {
+        await original.WriteAsync(buffer, cancellationToken);
+        if (!buffer.IsEmpty)
+            capture(stream, buffer.ToString());
+    }
+
+    public override async Task WriteAsync(StringBuilder? value, CancellationToken cancellationToken = default)
+    {
+        await original.WriteAsync(value, cancellationToken);
+        if (value is { Length: > 0 })
+            capture(stream, value.ToString());
+    }
+
+    // ── Async WriteLine overloads ──
+
+    public override async Task WriteLineAsync() { await original.WriteLineAsync(); capture(stream, "\n"); }
+    public override async Task WriteLineAsync(char value) { await original.WriteLineAsync(value); capture(stream, value + "\n"); }
+
+    public override async Task WriteLineAsync(string? value)
+    {
+        await original.WriteLineAsync(value);
+        capture(stream, (value ?? "") + "\n");
+    }
+
+    public override async Task WriteLineAsync(char[] buffer, int index, int count)
+    {
+        await original.WriteLineAsync(buffer, index, count);
+        capture(stream, new string(buffer, index, count) + "\n");
+    }
+
+    public override async Task WriteLineAsync(ReadOnlyMemory<char> buffer, CancellationToken cancellationToken = default)
+    {
+        await original.WriteLineAsync(buffer, cancellationToken);
+        capture(stream, string.Concat(buffer.Span, "\n"));
+    }
+
+    public override async Task WriteLineAsync(StringBuilder? value, CancellationToken cancellationToken = default)
+    {
+        await original.WriteLineAsync(value, cancellationToken);
+        capture(stream, (value is { Length: > 0 } ? value.ToString() : "") + "\n");
+    }
+
+    // ── Lifecycle ──
+
+    public override void Flush() => original.Flush();
+    public override Task FlushAsync() => original.FlushAsync();
+    public override Task FlushAsync(CancellationToken cancellationToken) => original.FlushAsync(cancellationToken);
+    public override void Close() => original.Close();
+}
+

src/modules/Elsa.Identity/HostedServices/AdminUserInitializer.cs

@@ -20,7 +20,7 @@ public class AdminUserInitializer(
     ILogger<AdminUserInitializer> logger)
     : BackgroundTask
 {
-    public override async Task StartAsync(CancellationToken cancellationToken)
+    public override async Task ExecuteAsync(CancellationToken cancellationToken)
     {
         var adminUserName = options.Value.AdminUserName;
         var adminPassword = options.Value.AdminPassword;

src/modules/Elsa.Scheduling/Features/SchedulingFeature.cs

@@ -7,8 +7,8 @@ using Elsa.Features.Attributes;
 using Elsa.Features.Services;
 using Elsa.Scheduling.Bookmarks;
 using Elsa.Scheduling.Handlers;
-using Elsa.Scheduling.HostedServices;
 using Elsa.Scheduling.Services;
+using Elsa.Scheduling.StartupTasks;
 using Elsa.Scheduling.TriggerPayloadValidators;
 using Elsa.Workflows.Management.Features;
 using Microsoft.Extensions.DependencyInjection;
@@ -41,7 +41,6 @@ public class SchedulingFeature : FeatureBase
     {
         Services
             .AddSingleton<UpdateTenantSchedules>()
-            .AddSingleton<ITenantActivatedEvent>(sp => sp.GetRequiredService<UpdateTenantSchedules>())
             .AddSingleton<ITenantDeletedEvent>(sp => sp.GetRequiredService<UpdateTenantSchedules>())
             .AddSingleton<IScheduler, LocalScheduler>()
             .AddSingleton<CronosCronParser>()
@@ -50,7 +49,7 @@ public class SchedulingFeature : FeatureBase
             .AddScoped<IBookmarkScheduler, DefaultBookmarkScheduler>()
             .AddScoped<DefaultWorkflowScheduler>()
             .AddScoped(WorkflowScheduler)
-            .AddBackgroundTask<CreateSchedulesBackgroundTask>()
+            .AddStartupTask<CreateSchedulesStartupTask>()
             .AddHandlersFrom<ScheduleWorkflows>()
 
             //Trigger payload validators.

src/modules/Elsa.Scheduling/Handlers/UpdateTenantSchedules.cs

@@ -9,7 +9,7 @@ using Timer = Elsa.Scheduling.Activities.Timer;
 
 namespace Elsa.Scheduling.Handlers;
 
-public class UpdateTenantSchedules : ITenantActivatedEvent, ITenantDeletedEvent
+public class UpdateTenantSchedules : ITenantDeletedEvent
 {
     private static readonly string[] ActivityTypeNames =
     [
@@ -18,19 +18,6 @@ public class UpdateTenantSchedules : ITenantActivatedEvent, ITenantDeletedEvent
         ActivityTypeNameHelper.GenerateTypeName<Delay>()
     ];
 
-    public async Task TenantActivatedAsync(TenantActivatedEventArgs args)
-    {
-        var serviceProvider = args.TenantScope.ServiceProvider;
-        var cancellationToken = args.CancellationToken;
-        var triggerScheduler = args.TenantScope.ServiceProvider.GetRequiredService<ITriggerScheduler>();
-        var bookmarkScheduler = args.TenantScope.ServiceProvider.GetRequiredService<IBookmarkScheduler>();
-        var triggers = (await GetTriggersAsync(serviceProvider, cancellationToken)).ToList();
-        var bookmarks = (await GetBookmarksAsync(serviceProvider, cancellationToken)).ToList();
-
-        await triggerScheduler.ScheduleAsync(triggers, args.CancellationToken);
-        await bookmarkScheduler.ScheduleAsync(bookmarks, args.CancellationToken);
-    }
-
     public async Task TenantDeletedAsync(TenantDeletedEventArgs args)
     {
         var serviceProvider = args.TenantScope.ServiceProvider;
@@ -57,4 +44,4 @@ public class UpdateTenantSchedules : ITenantActivatedEvent, ITenantDeletedEvent
         var filter = BookmarkFilter.ByActivityTypeNames(ActivityTypeNames);
         return await bookmarkStore.FindManyAsync(filter, cancellationToken);
     }
-}
+}

src/modules/Elsa.Scheduling/HostedServices/CreateSchedulesBackgroundTask.cs

@@ -1,35 +0,0 @@
-using Elsa.Common;
-using Elsa.Workflows.Runtime;
-using Elsa.Workflows.Runtime.Filters;
-using JetBrains.Annotations;
-
-namespace Elsa.Scheduling.HostedServices;
-
-/// <summary>
-/// Creates new schedules when using the default scheduler (which doesn't have its own persistence layer like Quartz or Hangfire).
-/// </summary>
-[UsedImplicitly]
-public class CreateSchedulesBackgroundTask(ITriggerStore triggerStore, IBookmarkStore bookmarkStore, ITriggerScheduler triggerScheduler, IBookmarkScheduler bookmarkScheduler) : BackgroundTask
-{
-    /// <inheritdoc />
-    public override async Task ExecuteAsync(CancellationToken stoppingToken)
-    {
-        var stimulusNames = new[]
-        {
-            SchedulingStimulusNames.Cron, SchedulingStimulusNames.Timer, SchedulingStimulusNames.StartAt, SchedulingStimulusNames.Delay,
-        };
-        var triggerFilter = new TriggerFilter
-        {
-            Names = stimulusNames
-        };
-        var bookmarkFilter = new BookmarkFilter
-        {
-            Names = stimulusNames
-        };
-        var triggers = (await triggerStore.FindManyAsync(triggerFilter, stoppingToken)).ToList();
-        var bookmarks = (await bookmarkStore.FindManyAsync(bookmarkFilter, stoppingToken)).ToList();
-
-        await triggerScheduler.ScheduleAsync(triggers, stoppingToken);
-        await bookmarkScheduler.ScheduleAsync(bookmarks, stoppingToken);
-    }
-}

src/modules/Elsa.Scheduling/ShellFeatures/SchedulingFeature.cs

@@ -4,8 +4,8 @@ using Elsa.Common.ShellFeatures;
 using Elsa.Extensions;
 using Elsa.Scheduling.Bookmarks;
 using Elsa.Scheduling.Handlers;
-using Elsa.Scheduling.HostedServices;
 using Elsa.Scheduling.Services;
+using Elsa.Scheduling.StartupTasks;
 using Elsa.Scheduling.TriggerPayloadValidators;
 using Elsa.Workflows.Management.Extensions;
 using JetBrains.Annotations;
@@ -37,7 +37,6 @@ public class SchedulingFeature : IShellFeature
     {
         services
             .AddSingleton<UpdateTenantSchedules>()
-            .AddSingleton<ITenantActivatedEvent>(sp => sp.GetRequiredService<UpdateTenantSchedules>())
             .AddSingleton<ITenantDeletedEvent>(sp => sp.GetRequiredService<UpdateTenantSchedules>())
             .AddSingleton<IScheduler, LocalScheduler>()
             .AddSingleton<CronosCronParser>()
@@ -46,10 +45,9 @@ public class SchedulingFeature : IShellFeature
             .AddScoped<IBookmarkScheduler, DefaultBookmarkScheduler>()
             .AddScoped<DefaultWorkflowScheduler>()
             .AddScoped(WorkflowScheduler)
-            .AddBackgroundTask<CreateSchedulesBackgroundTask>()
+            .AddStartupTask<CreateSchedulesStartupTask>()
             .AddHandlersFrom<ScheduleWorkflows>()
             .AddTriggerPayloadValidator<CronTriggerPayloadValidator, CronTriggerPayload>()
             .AddActivitiesFrom<SchedulingFeature>();
     }
 }
-

src/modules/Elsa.Scheduling/StartupTasks/CreateSchedulesStartupTask.cs

@@ -0,0 +1,50 @@
+using Elsa.Common;
+using Elsa.Common.Multitenancy;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Filters;
+using Elsa.Workflows.Runtime.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Elsa.Scheduling.StartupTasks;
+
+/// <summary>
+/// Enqueues schedule creation when using the default scheduler, which doesn't have its own persistence layer like Quartz or Hangfire.
+/// </summary>
+[TaskDependency(typeof(PopulateRegistriesStartupTask))]
+public class CreateSchedulesStartupTask(IServiceProvider serviceProvider) : IStartupTask
+{
+    public async Task ExecuteAsync(CancellationToken cancellationToken)
+    {
+        var workQueue = serviceProvider.GetService<ITenantBackgroundWorkQueue>();
+
+        if (workQueue != null)
+            await workQueue.EnqueueAsync(CreateSchedulesAsync, cancellationToken);
+        else
+            await CreateSchedulesAsync(serviceProvider, cancellationToken);
+    }
+
+    private static async Task CreateSchedulesAsync(IServiceProvider serviceProvider, CancellationToken cancellationToken)
+    {
+        var triggerStore = serviceProvider.GetRequiredService<ITriggerStore>();
+        var bookmarkStore = serviceProvider.GetRequiredService<IBookmarkStore>();
+        var triggerScheduler = serviceProvider.GetRequiredService<ITriggerScheduler>();
+        var bookmarkScheduler = serviceProvider.GetRequiredService<IBookmarkScheduler>();
+        var stimulusNames = new[]
+        {
+            SchedulingStimulusNames.Cron, SchedulingStimulusNames.Timer, SchedulingStimulusNames.StartAt, SchedulingStimulusNames.Delay,
+        };
+        var triggerFilter = new TriggerFilter
+        {
+            Names = stimulusNames
+        };
+        var bookmarkFilter = new BookmarkFilter
+        {
+            Names = stimulusNames
+        };
+        var triggers = (await triggerStore.FindManyAsync(triggerFilter, cancellationToken)).ToList();
+        var bookmarks = (await bookmarkStore.FindManyAsync(bookmarkFilter, cancellationToken)).ToList();
+
+        await triggerScheduler.ScheduleAsync(triggers, cancellationToken);
+        await bookmarkScheduler.ScheduleAsync(bookmarks, cancellationToken);
+    }
+}

src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedRuntimeLockProviderValidator.cs

@@ -10,7 +10,7 @@ using Microsoft.Extensions.Options;
 namespace Elsa.Workflows.Runtime.Distributed;
 
 /// <summary>
-/// Validates that the distributed runtime is not accidentally backed by a local-only lock provider.
+/// Logs when the distributed runtime is backed by a local-only lock provider.
 /// </summary>
 public class DistributedRuntimeLockProviderValidator(
     IDistributedLockProvider distributedLockProvider,
@@ -18,7 +18,7 @@ public class DistributedRuntimeLockProviderValidator(
     ILogger<DistributedRuntimeLockProviderValidator> logger)
 {
     /// <summary>
-    /// Validates the configured provider.
+    /// Checks the configured provider.
     /// </summary>
     public void Validate()
     {
@@ -37,19 +37,17 @@ public class DistributedRuntimeLockProviderValidator(
 
         if (options.Value.AllowLocalLockProviderInDistributedRuntime)
         {
-            logger.LogWarning(
+            logger.LogDebug(
-                "Distributed workflow runtime is using local-only lock provider {LocalDistributedLockProviderType} through configured provider {DistributedLockProviderType}. This is safe only for single-host deployments.",
+                "Distributed workflow runtime is using acknowledged local-only lock provider {LocalDistributedLockProviderType} through configured provider {DistributedLockProviderType}.",
                 localProviderTypeName,
                 configuredProviderTypeName);
             return;
         }
 
-        var message =
+        logger.LogWarning(
-            $"The distributed workflow runtime is configured with local-only distributed lock provider '{localProviderTypeName}' through '{configuredProviderTypeName}'. " +
+            "Distributed workflow runtime is using local-only lock provider {LocalDistributedLockProviderType} through configured provider {DistributedLockProviderType}. This provider does not coordinate across application nodes and can allow concurrent workflow processing in clustered deployments. Configure a cross-node IDistributedLockProvider such as Redis, SQL Server, or PostgreSQL for clustered deployments, or set DistributedLockingOptions.AllowLocalLockProviderInDistributedRuntime to true to acknowledge single-host development/test usage.",
-            "This provider does not coordinate across application nodes and can allow concurrent workflow processing in clustered deployments. " +
+            localProviderTypeName,
-            "Configure a cross-node IDistributedLockProvider such as Redis, SQL Server, or PostgreSQL, or explicitly set DistributedLockingOptions.AllowLocalLockProviderInDistributedRuntime to true for single-host development/test deployments.";
+            configuredProviderTypeName);
-
-        throw new InvalidOperationException(message);
     }
 
     private Type? FindLocalProviderType(IDistributedLockProvider provider, ISet<IDistributedLockProvider>? visited = null)

src/modules/Elsa.Workflows.Runtime.Distributed/StartupTasks/ValidateDistributedRuntimeLockProviderStartupTask.cs

@@ -5,7 +5,7 @@ using JetBrains.Annotations;
 namespace Elsa.Workflows.Runtime.Distributed.StartupTasks;
 
 /// <summary>
-/// Fails startup when distributed runtime uses a local-only lock provider without explicit opt-in.
+/// Checks the distributed runtime lock provider during startup.
 /// </summary>
 [UsedImplicitly]
 [Order(-1000)]

test/unit/Elsa.Common.UnitTests/Multitenancy/TenantTaskLifecycleCoordinatorTests.cs

@@ -1,6 +1,7 @@
 using Elsa.Common.Multitenancy;
 using Elsa.Common.RecurringTasks;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
 using NSubstitute;
 
@@ -18,6 +19,40 @@ public class TenantTaskLifecycleCoordinatorTests : IAsyncDisposable
 
     public ValueTask DisposeAsync() => _coordinator.DisposeAsync();
 
+    [Fact]
+    public async Task ActivateAsync_WithQueuedWork_ReturnsBeforeWorkItemCompletes()
+    {
+        QueueBlockingWorkStartupTask? startupTask = null;
+        var serviceProvider = BuildTenantServiceProvider(
+            startupTasks: [sp => startupTask = ActivatorUtilities.CreateInstance<QueueBlockingWorkStartupTask>(sp)],
+            backgroundTasks: [sp => ActivatorUtilities.CreateInstance<TenantBackgroundWorkQueueWorker>(sp)]);
+        var activationTask = _coordinator.ActivateTenantAsync(new TenantActivatedEventArgs(_tenant, CreateTenantScope(_tenant, serviceProvider), CancellationToken.None));
+
+        var completedTask = await Task.WhenAny(activationTask, Task.Delay(TimeSpan.FromSeconds(1)));
+
+        Assert.Same(activationTask, completedTask);
+        await activationTask;
+        await startupTask!.WorkStarted.Task.WaitAsync(TimeSpan.FromSeconds(1));
+        Assert.False(startupTask.WorkCompleted.Task.IsCompleted);
+
+        startupTask.ReleaseWork();
+        await startupTask.WorkCompleted.Task.WaitAsync(TimeSpan.FromSeconds(1));
+    }
+
+    [Fact]
+    public async Task DeactivateAsync_WithRunningBackgroundTask_StopsTask()
+    {
+        var backgroundTask = new TrackingBackgroundTask();
+        var serviceProvider = BuildTenantServiceProvider(backgroundTasks: [_ => backgroundTask]);
+
+        await _coordinator.ActivateTenantAsync(new TenantActivatedEventArgs(_tenant, CreateTenantScope(_tenant, serviceProvider), CancellationToken.None));
+        await backgroundTask.Started.Task.WaitAsync(TimeSpan.FromSeconds(1));
+        await _coordinator.DeactivateTenantAsync(DeactivationArgs(serviceProvider));
+
+        Assert.True(backgroundTask.WasStopCalled);
+        Assert.True(backgroundTask.WasCancelled);
+    }
+
     /// <summary>
     /// Regression test: before the fix, <c>TryRemove</c> was called before <c>WaitAsync</c>,
     /// so a cancelled <c>WaitAsync</c> would orphan the state (removed from the dictionary but
@@ -80,6 +115,9 @@ public class TenantTaskLifecycleCoordinatorTests : IAsyncDisposable
     private TenantDeactivatedEventArgs DeactivationArgs(CancellationToken cancellationToken = default) =>
         DeactivationArgs(_tenant, cancellationToken);
 
+    private TenantDeactivatedEventArgs DeactivationArgs(IServiceProvider serviceProvider, CancellationToken cancellationToken = default) =>
+        new(_tenant, CreateTenantScope(_tenant, serviceProvider), cancellationToken);
+
     private TenantDeactivatedEventArgs DeactivationArgs(Tenant tenant, CancellationToken cancellationToken = default) =>
         new(tenant, CreateTenantScope(tenant, _serviceProvider), cancellationToken);
 
@@ -91,13 +129,22 @@ public class TenantTaskLifecycleCoordinatorTests : IAsyncDisposable
         return new TenantTaskLifecycleCoordinator(scheduleManager, NullLogger<TenantTaskLifecycleCoordinator>.Instance);
     }
 
-    private static IServiceProvider BuildTenantServiceProvider(IEnumerable<IRecurringTask>? recurringTasks = null)
+    private static IServiceProvider BuildTenantServiceProvider(
+        IEnumerable<IRecurringTask>? recurringTasks = null,
+        IEnumerable<Func<IServiceProvider, IStartupTask>>? startupTasks = null,
+        IEnumerable<Func<IServiceProvider, IBackgroundTask>>? backgroundTasks = null)
     {
         var services = new ServiceCollection();
-        services.AddSingleton<ITaskExecutor>(new InstantTaskExecutor());
+        services.AddSingleton<ITaskExecutor>(new DirectTaskExecutor());
-        services.AddSingleton(Substitute.For<IBackgroundTaskStarter>());
+        services.AddSingleton<IBackgroundTaskStarter, DirectBackgroundTaskStarter>();
+        services.AddSingleton<ITenantBackgroundWorkQueue, TenantBackgroundWorkQueue>();
+        services.AddSingleton<ILogger<TenantBackgroundWorkQueueWorker>>(NullLogger<TenantBackgroundWorkQueueWorker>.Instance);
+        foreach (var taskFactory in startupTasks ?? [])
+            services.AddSingleton<IStartupTask>(taskFactory);
         foreach (var task in recurringTasks ?? [])
             services.AddSingleton<IRecurringTask>(task);
+        foreach (var taskFactory in backgroundTasks ?? [])
+            services.AddSingleton<IBackgroundTask>(taskFactory);
         return services.BuildServiceProvider();
     }
 
@@ -126,8 +173,61 @@ public class TenantTaskLifecycleCoordinatorTests : IAsyncDisposable
         }
     }
 
-    private class InstantTaskExecutor : ITaskExecutor
+    private class QueueBlockingWorkStartupTask(ITenantBackgroundWorkQueue workQueue) : IStartupTask
     {
-        public Task ExecuteTaskAsync(ITask task, CancellationToken cancellationToken) => Task.CompletedTask;
+        private readonly TaskCompletionSource _releaseWork = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        public TaskCompletionSource WorkStarted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        public TaskCompletionSource WorkCompleted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        public async Task ExecuteAsync(CancellationToken cancellationToken)
+        {
+            await workQueue.EnqueueAsync(async (_, ct) =>
+            {
+                WorkStarted.SetResult();
+                await _releaseWork.Task.WaitAsync(ct);
+                WorkCompleted.SetResult();
+            }, cancellationToken);
+        }
+
+        public void ReleaseWork() => _releaseWork.SetResult();
+    }
+
+    private class TrackingBackgroundTask : BackgroundTask
+    {
+        public TaskCompletionSource Started { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        public bool WasCancelled { get; private set; }
+        public bool WasStopCalled { get; private set; }
+
+        public override async Task ExecuteAsync(CancellationToken cancellationToken)
+        {
+            Started.SetResult();
+
+            try
+            {
+                await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken);
+            }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                WasCancelled = true;
+            }
+        }
+
+        public override Task StopAsync(CancellationToken cancellationToken)
+        {
+            WasStopCalled = true;
+            return Task.CompletedTask;
+        }
+    }
+
+    private class DirectTaskExecutor : ITaskExecutor
+    {
+        public Task ExecuteTaskAsync(ITask task, CancellationToken cancellationToken) => task.ExecuteAsync(cancellationToken);
+    }
+
+    private class DirectBackgroundTaskStarter : IBackgroundTaskStarter
+    {
+        public Task StartAsync(IBackgroundTask task, CancellationToken cancellationToken) => task.StartAsync(cancellationToken);
+        public Task StopAsync(IBackgroundTask task, CancellationToken cancellationToken) => task.StopAsync(cancellationToken);
     }
 }

test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleCaptureTeeTests.cs

@@ -1,162 +1,292 @@
 using Elsa.Diagnostics.ConsoleLogs.Contracts;
 using Elsa.Diagnostics.ConsoleLogs.Services;
+using Microsoft.Extensions.Logging;
 
 namespace Elsa.Diagnostics.ConsoleLogs.UnitTests.Capture;
 
-public class ConsoleCaptureTeeTests
+[Collection(ConsoleHostStateCollection.Name)]
+public class ConsoleCaptureTeeTests : IDisposable
 {
-    [Fact]
+    private readonly TextWriter _originalOut = Console.Out;
-    public async Task StartAsync_PreservesOriginalConsoleOutputAndPublishesLine()
+    private readonly TextWriter _originalError = Console.Error;
+    private readonly StringWriter _consoleOutput = new();
+    private readonly CapturingProvider _provider = new();
+    private readonly IOptions<ConsoleLogsOptions> _options = Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions());
+    private readonly ConsoleLogScopeAccessor _scopeAccessor = new();
+
+    public ConsoleCaptureTeeTests()
     {
-        var originalOut = Console.Out;
+        // Tear down any process-wide host instance so its hook subscription doesn't double-publish.
-        var originalError = Console.Error;
+        ConsoleLogsHost.ShutdownAsync().AsTask().GetAwaiter().GetResult();
-        using var consoleOutput = new StringWriter();
+        ConsoleStreamHook.Uninstall();
-        var provider = new CapturingProvider();
+        Console.SetOut(_consoleOutput);
-        var registry = new ConsoleLogSourceRegistry(Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions()));
+    }
-        var options = Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions());
-        var capture = new ConsoleCaptureTee(provider, registry, new ConsoleLogRedactor(options), new ConsoleLineFormatter(options), options);
 
-        try
+    public void Dispose()
-        {
+    {
-            Console.SetOut(consoleOutput);
+        ConsoleStreamHook.Uninstall();
-            await capture.StartAsync();
+        ConsoleLogsHost.ShutdownAsync().AsTask().GetAwaiter().GetResult();
+        Console.SetOut(_originalOut);
+        Console.SetError(_originalError);
+        _consoleOutput.Dispose();
+    }
 
-            Console.WriteLine("hello");
+    private ConsoleCaptureTee CreateCapture() => new(
-            await WaitForLineAsync(provider);
+        _provider,
+        new ConsoleLogSourceRegistry(_options),
+        new ConsoleLogRedactor(_options),
+        new ConsoleLineFormatter(_options),
+        _scopeAccessor,
+        _options);
 
-            Assert.Equal($"hello{Environment.NewLine}", consoleOutput.ToString());
+    [Fact]
-            Assert.Single(provider.Lines);
+    public async Task Capture_PreservesOriginalConsoleOutputAndPublishesLine()
-            Assert.Equal("hello", provider.Lines[0].Text);
+    {
-            Assert.Equal(ConsoleLogStream.Stdout, provider.Lines[0].Stream);
+        await using var capture = CreateCapture();
-        }
+
-        finally
+        Console.WriteLine("hello");
-        {
+        await WaitForLinesAsync(_provider, 1);
-            await capture.StopAsync();
+
-            Console.SetOut(originalOut);
+        Assert.Equal($"hello{Environment.NewLine}", _consoleOutput.ToString());
-            Console.SetError(originalError);
+        var line = Assert.Single(_provider.Lines);
-        }
+        Assert.Equal("hello", line.Text);
+        Assert.Equal(ConsoleLogStream.Stdout, line.Stream);
     }
 
     [Fact]
-    public async Task StartAsync_WhenAlreadyStarted_DoesNotWrapConsoleTwice()
+    public async Task Capture_RedactsBeforePublishingToProvider()
     {
-        var originalOut = Console.Out;
+        await using var capture = CreateCapture();
-        var originalError = Console.Error;
-        using var consoleOutput = new StringWriter();
-        var provider = new CapturingProvider();
-        var registry = new ConsoleLogSourceRegistry(Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions()));
-        var options = Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions());
-        var capture = new ConsoleCaptureTee(provider, registry, new ConsoleLogRedactor(options), new ConsoleLineFormatter(options), options);
 
-        try
+        Console.WriteLine("token=secret-value");
-        {
+        await WaitForLinesAsync(_provider, 1);
-            Console.SetOut(consoleOutput);
-            await capture.StartAsync();
-            await capture.StartAsync();
 
-            Console.WriteLine("hello");
+        Assert.Equal("[Redacted]", _provider.Lines[0].Text);
-            await WaitForLineAsync(provider);
-
-            Assert.Equal($"hello{Environment.NewLine}", consoleOutput.ToString());
-            Assert.Single(provider.Lines);
-        }
-        finally
-        {
-            await capture.StopAsync();
-            await capture.StopAsync();
-            Console.SetOut(originalOut);
-            Console.SetError(originalError);
-        }
     }
 
     [Fact]
-    public async Task StartAsync_RedactsBeforePublishingToProvider()
+    public async Task Capture_AfterDispose_StopsPublishing()
     {
-        var originalOut = Console.Out;
+        var capture = CreateCapture();
-        var originalError = Console.Error;
-        using var consoleOutput = new StringWriter();
-        var provider = new CapturingProvider();
-        var registry = new ConsoleLogSourceRegistry(Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions()));
-        var options = Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions());
-        var capture = new ConsoleCaptureTee(provider, registry, new ConsoleLogRedactor(options), new ConsoleLineFormatter(options), options);
 
-        try
+        Console.WriteLine("first");
-        {
+        await WaitForLinesAsync(_provider, 1);
-            Console.SetOut(consoleOutput);
-            await capture.StartAsync();
 
-            Console.WriteLine("token=secret-value");
+        await capture.DisposeAsync();
-            await WaitForLineAsync(provider);
 
-            Assert.Equal("[Redacted]", provider.Lines[0].Text);
+        Console.WriteLine("second");
-        }
+        await Task.Delay(50);
-        finally
+
-        {
+        Assert.Single(_provider.Lines);
-            await capture.StopAsync();
+        Assert.Equal("first", _provider.Lines[0].Text);
-            Console.SetOut(originalOut);
-            Console.SetError(originalError);
-        }
     }
 
     [Fact]
-    public async Task StopAsync_WhenStartedTwice_KeepsCaptureActiveUntilSecondStop()
+    public async Task Capture_FansOutToMultipleConcurrentSubscribers()
     {
-        var originalOut = Console.Out;
+        var secondProvider = new CapturingProvider();
-        var originalError = Console.Error;
+        await using var first = CreateCapture();
-        using var consoleOutput = new StringWriter();
+        await using var second = new ConsoleCaptureTee(
-        var provider = new CapturingProvider();
+            secondProvider,
-        var registry = new ConsoleLogSourceRegistry(Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions()));
+            new ConsoleLogSourceRegistry(_options),
-        var options = Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions());
+            new ConsoleLogRedactor(_options),
-        var capture = new ConsoleCaptureTee(provider, registry, new ConsoleLogRedactor(options), new ConsoleLineFormatter(options), options);
+            new ConsoleLineFormatter(_options),
+            _scopeAccessor,
+            _options);
 
-        try
+        Console.WriteLine("broadcast");
-        {
+        await WaitForLinesAsync(_provider, 1);
-            Console.SetOut(consoleOutput);
+        await WaitForLinesAsync(secondProvider, 1);
-            await capture.StartAsync();
-            await capture.StartAsync();
 
-            await capture.StopAsync();
+        Assert.Equal("broadcast", _provider.Lines[^1].Text);
-            Console.WriteLine("still captured");
+        Assert.Equal("broadcast", secondProvider.Lines[^1].Text);
-            await WaitForLineAsync(provider);
-
-            Assert.Single(provider.Lines);
-            Assert.Equal("still captured", provider.Lines[0].Text);
-
-            await capture.StopAsync();
-            Console.WriteLine("not captured");
-            await Task.Delay(50);
-
-            Assert.Single(provider.Lines);
-        }
-        finally
-        {
-            await capture.StopAsync();
-            Console.SetOut(originalOut);
-            Console.SetError(originalError);
-        }
     }
 
-    private static async Task WaitForLineAsync(CapturingProvider provider)
+    [Fact]
+    public async Task Capture_ReplaysStartupWritesOnlyOnce()
     {
-        var timeout = DateTimeOffset.UtcNow.AddSeconds(2);
+        ConsoleStreamHook.Install();
-        while (provider.Lines.Count == 0 && DateTimeOffset.UtcNow < timeout)
+        Console.WriteLine("startup");
+
+        var secondProvider = new CapturingProvider();
+        await using var first = CreateCapture();
+        await WaitForLinesAsync(_provider, 1);
+
+        await using var second = new ConsoleCaptureTee(
+            secondProvider,
+            new ConsoleLogSourceRegistry(_options),
+            new ConsoleLogRedactor(_options),
+            new ConsoleLineFormatter(_options),
+            _scopeAccessor,
+            _options);
+        await Task.Delay(50);
+
+        Assert.Equal("startup", _provider.Lines[0].Text);
+        Assert.Empty(secondProvider.Lines);
+    }
+
+    [Fact]
+    public async Task Capture_SuppressesPublishPathReentrancy()
+    {
+        var loggingProvider = new ReentrantProvider();
+        await using var capture = new ConsoleCaptureTee(
+            loggingProvider,
+            new ConsoleLogSourceRegistry(_options),
+            new ConsoleLogRedactor(_options),
+            new ConsoleLineFormatter(_options),
+            _scopeAccessor,
+            _options);
+
+        Console.WriteLine("trigger");
+        await WaitForLinesAsync(loggingProvider, 1);
+        await Task.Delay(50); // Give any reentrant publish a chance to misbehave.
+
+        Assert.Single(loggingProvider.Lines);
+        Assert.Equal("trigger", loggingProvider.Lines[0].Text);
+    }
+
+    [Fact]
+    public async Task Capture_ContinuesAfterProviderOperationCanceledException()
+    {
+        var provider = new OperationCanceledOnceProvider();
+        await using var capture = new ConsoleCaptureTee(
+            provider,
+            new ConsoleLogSourceRegistry(_options),
+            new ConsoleLogRedactor(_options),
+            new ConsoleLineFormatter(_options),
+            _scopeAccessor,
+            _options);
+
+        Console.WriteLine("first");
+        Console.WriteLine("second");
+        await WaitForLinesAsync(provider, 1);
+
+        var line = Assert.Single(provider.Lines);
+        Assert.Equal("second", line.Text);
+    }
+
+    [Fact]
+    public async Task Capture_AttachesCurrentWorkflowInstanceId()
+    {
+        using var loggerFactory = LoggerFactory.Create(builder => builder.AddProvider(_scopeAccessor));
+        var logger = loggerFactory.CreateLogger("workflow");
+        await using var capture = CreateCapture();
+        using var scope = logger.BeginScope(new Dictionary<string, object>
+        {
+            ["WorkflowInstanceId"] = "workflow-instance-a"
+        });
+
+        Console.WriteLine("scoped");
+        await WaitForLinesAsync(_provider, 1);
+
+        Assert.Equal("workflow-instance-a", _provider.Lines[0].WorkflowInstanceId);
+    }
+
+    [Fact]
+    public async Task Capture_AttachesInnermostWorkflowInstanceId()
+    {
+        using var loggerFactory = LoggerFactory.Create(builder => builder.AddProvider(_scopeAccessor));
+        var logger = loggerFactory.CreateLogger("workflow");
+        await using var capture = CreateCapture();
+        using var outerScope = logger.BeginScope(new Dictionary<string, object>
+        {
+            ["WorkflowInstanceId"] = "workflow-instance-a"
+        });
+        using var innerScope = logger.BeginScope(new Dictionary<string, object>
+        {
+            ["WorkflowInstanceId"] = "workflow-instance-b"
+        });
+
+        Console.WriteLine("scoped");
+        await WaitForLinesAsync(_provider, 1);
+
+        Assert.Equal("workflow-instance-b", _provider.Lines[0].WorkflowInstanceId);
+    }
+
+    [Fact]
+    public async Task Capture_PreservesScopeAcrossMultipleLoggerFactories()
+    {
+        using var firstLoggerFactory = LoggerFactory.Create(builder => builder.AddProvider(_scopeAccessor));
+        using var secondLoggerFactory = LoggerFactory.Create(builder => builder.AddProvider(_scopeAccessor));
+        var logger = firstLoggerFactory.CreateLogger("workflow");
+        await using var capture = CreateCapture();
+        using var scope = logger.BeginScope(new Dictionary<string, object>
+        {
+            ["WorkflowInstanceId"] = "workflow-instance-a"
+        });
+
+        _ = secondLoggerFactory.CreateLogger("other");
+        Console.WriteLine("scoped");
+        await WaitForLinesAsync(_provider, 1);
+
+        Assert.Equal("workflow-instance-a", _provider.Lines[0].WorkflowInstanceId);
+    }
+
+    [Fact]
+    public async Task Capture_AttachesWorkflowInstanceIdCapturedBeforeAsyncConsoleWrite()
+    {
+        using var loggerFactory = LoggerFactory.Create(builder => builder.AddProvider(_scopeAccessor));
+        var logger = loggerFactory.CreateLogger("workflow");
+        await using var capture = CreateCapture();
+
+        using (logger.BeginScope(new Dictionary<string, object>
+        {
+            ["WorkflowInstanceId"] = "workflow-instance-a"
+        }))
+        {
+            logger.LogInformation("scoped");
+        }
+
+        await Task.Run(() => Console.WriteLine("scoped"));
+        await WaitForLinesAsync(_provider, 1);
+
+        Assert.Equal("workflow-instance-a", _provider.Lines[0].WorkflowInstanceId);
+    }
+
+    [Fact]
+    public async Task Capture_DoesNotAttachCapturedWorkflowInstanceIdToUnrelatedConsoleWrite()
+    {
+        using var loggerFactory = LoggerFactory.Create(builder => builder.AddProvider(_scopeAccessor));
+        var logger = loggerFactory.CreateLogger("workflow");
+        await using var capture = CreateCapture();
+
+        using (logger.BeginScope(new Dictionary<string, object>
+        {
+            ["WorkflowInstanceId"] = "workflow-instance-a"
+        }))
+        {
+            logger.LogInformation("scoped");
+        }
+
+        await Task.Run(() => Console.WriteLine("not scoped"));
+        await Task.Run(() => Console.WriteLine("scoped"));
+        await WaitForLinesAsync(_provider, 2);
+
+        Assert.Null(_provider.Lines[0].WorkflowInstanceId);
+        Assert.Equal("workflow-instance-a", _provider.Lines[1].WorkflowInstanceId);
+    }
+
+    private static async Task WaitForLinesAsync(CapturingProvider provider, int count)
+    {
+        var deadline = DateTimeOffset.UtcNow.AddSeconds(2);
+        while (provider.Lines.Count < count && DateTimeOffset.UtcNow < deadline)
             await Task.Delay(10);
     }
 
-    private sealed class CapturingProvider : IConsoleLogProvider
+    private class CapturingProvider : IConsoleLogProvider
     {
         public List<ConsoleLogLine> Lines { get; } = [];
 
-        public ValueTask PublishAsync(ConsoleLogLine line, CancellationToken cancellationToken = default)
+        public virtual ValueTask PublishAsync(ConsoleLogLine line, CancellationToken cancellationToken = default)
         {
-            Lines.Add(line);
+            lock (Lines)
+                Lines.Add(line);
+
             return ValueTask.CompletedTask;
         }
 
-        public ValueTask<RecentConsoleLogsResult> GetRecentAsync(ConsoleLogFilter filter, CancellationToken cancellationToken = default)
+        public ValueTask<RecentConsoleLogsResult> GetRecentAsync(ConsoleLogFilter filter, CancellationToken cancellationToken = default) =>
-        {
+            ValueTask.FromResult(new RecentConsoleLogsResult(Lines));
-            return ValueTask.FromResult(new RecentConsoleLogsResult(Lines));
-        }
 
         public async IAsyncEnumerable<ConsoleLogStreamItem> SubscribeAsync(ConsoleLogFilter filter, [EnumeratorCancellation] CancellationToken cancellationToken = default)
         {
@@ -165,9 +295,31 @@ public class ConsoleCaptureTeeTests
                 yield return ConsoleLogStreamItem.FromLine(line);
         }
 
-        public ValueTask<IReadOnlyCollection<ConsoleLogSource>> ListSourcesAsync(CancellationToken cancellationToken = default)
+        public ValueTask<IReadOnlyCollection<ConsoleLogSource>> ListSourcesAsync(CancellationToken cancellationToken = default) =>
+            ValueTask.FromResult<IReadOnlyCollection<ConsoleLogSource>>([]);
+    }
+
+    private sealed class ReentrantProvider : CapturingProvider
+    {
+        public override ValueTask PublishAsync(ConsoleLogLine line, CancellationToken cancellationToken = default)
         {
-            return ValueTask.FromResult<IReadOnlyCollection<ConsoleLogSource>>([]);
+            // Simulate a provider whose internals write to the console (e.g., SignalR diagnostics).
+            // With SuppressCapture set on this async context, this MUST NOT recurse.
+            Console.WriteLine("[provider-internal-log]");
+            return base.PublishAsync(line, cancellationToken);
+        }
+    }
+
+    private sealed class OperationCanceledOnceProvider : CapturingProvider
+    {
+        private int _throwCount;
+
+        public override ValueTask PublishAsync(ConsoleLogLine line, CancellationToken cancellationToken = default)
+        {
+            if (Interlocked.Increment(ref _throwCount) == 1)
+                throw new OperationCanceledException();
+
+            return base.PublishAsync(line, cancellationToken);
         }
     }
 }

test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleLineBufferTests.cs

@@ -13,7 +13,7 @@ public class ConsoleLineBufferTests
 
         var lines = _buffer.Append("lo\n", DateTimeOffset.UtcNow);
 
-        Assert.Equal(["hello"], lines);
+        Assert.Equal(["hello"], lines.Select(x => x.Text));
     }
 
     [Fact]
@@ -21,8 +21,8 @@ public class ConsoleLineBufferTests
     {
         var lines = _buffer.Append("hello!", DateTimeOffset.UtcNow);
 
-        Assert.Equal(["hello"], lines);
+        Assert.Equal(["hello"], lines.Select(x => x.Text));
-        Assert.Equal("!", _buffer.Flush());
+        Assert.Equal("!", _buffer.Flush()?.Text);
     }
 
     [Fact]
@@ -33,6 +33,36 @@ public class ConsoleLineBufferTests
         _buffer.Append("tail", now);
 
         Assert.Null(_buffer.FlushIfIdle(now.AddMilliseconds(500)));
-        Assert.Equal("tail", _buffer.FlushIfIdle(now.AddSeconds(2)));
+        Assert.Equal("tail", _buffer.FlushIfIdle(now.AddSeconds(2))?.Text);
+    }
+
+    [Fact]
+    public void Append_PreservesEmptyLinesBetweenWrites()
+    {
+        var lines = _buffer.Append("a\n\nb\n", DateTimeOffset.UtcNow);
+
+        Assert.Equal(["a", "", "b"], lines.Select(x => x.Text));
+    }
+
+    [Fact]
+    public void Append_PreservesWorkflowInstanceIdAcrossPartialWrites()
+    {
+        _buffer.Append("hel", DateTimeOffset.UtcNow, "workflow-instance-a");
+
+        var line = Assert.Single(_buffer.Append("lo\n", DateTimeOffset.UtcNow));
+
+        Assert.Equal("hello", line.Text);
+        Assert.Equal("workflow-instance-a", line.WorkflowInstanceId);
+    }
+
+    [Fact]
+    public void Append_DoesNotAssignWorkflowInstanceIdAfterUnscopedPartialWrite()
+    {
+        _buffer.Append("hel", DateTimeOffset.UtcNow);
+
+        var line = Assert.Single(_buffer.Append("lo\n", DateTimeOffset.UtcNow, "workflow-instance-a"));
+
+        Assert.Equal("hello", line.Text);
+        Assert.Null(line.WorkflowInstanceId);
     }
 }

test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleLineFormatterTests.cs

@@ -5,16 +5,26 @@ namespace Elsa.Diagnostics.ConsoleLogs.UnitTests.Capture;
 public class ConsoleLineFormatterTests
 {
     [Fact]
-    public void Format_StripsAnsiByDefault()
+    public void Format_PreservesAnsiByDefault()
     {
         var formatter = new ConsoleLineFormatter(Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions()));
 
         var result = formatter.Format("\u001b[31mred\u001b[0m");
 
-        Assert.Equal("red", result.Text);
+        Assert.Equal("\u001b[31mred\u001b[0m", result.Text);
         Assert.False(result.Truncated);
     }
 
+    [Fact]
+    public void Format_StripsAnsiWhenConfigured()
+    {
+        var formatter = new ConsoleLineFormatter(Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions { StripAnsiEscapeSequences = true }));
+
+        var result = formatter.Format("\u001b[31mred\u001b[0m");
+
+        Assert.Equal("red", result.Text);
+    }
+
     [Fact]
     public void Format_PreservesAnsiWhenConfigured()
     {

test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/ConsoleHostStateCollection.cs

@@ -0,0 +1,15 @@
+using Elsa.Diagnostics.ConsoleLogs.Services;
+
+namespace Elsa.Diagnostics.ConsoleLogs.UnitTests;
+
+/// <summary>
+/// xUnit collection used by every test class that mutates the process-wide <see cref="ConsoleStreamHook"/>
+/// or <see cref="ConsoleLogsHost"/>. Members of this collection run sequentially, preventing parallel
+/// tests from racing on the same static state.
+/// </summary>
+[CollectionDefinition(Name)]
+public sealed class ConsoleHostStateCollection
+{
+    public const string Name = "ConsoleHostState";
+}
+

test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/ConsoleLogsRegistrationTests.cs

@@ -4,32 +4,209 @@ using Elsa.Diagnostics.ConsoleLogs.Providers.InMemory;
 using Elsa.Diagnostics.ConsoleLogs.Services;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
 
 namespace Elsa.Diagnostics.ConsoleLogs.UnitTests;
 
-public class ConsoleLogsRegistrationTests
+[Collection(ConsoleHostStateCollection.Name)]
+public class ConsoleLogsRegistrationTests : IAsyncLifetime
 {
+    public async Task InitializeAsync()
+    {
+        await ConsoleLogsHost.ShutdownAsync();
+        ConsoleStreamHook.Uninstall();
+    }
+
+    public async Task DisposeAsync()
+    {
+        await ConsoleLogsHost.ShutdownAsync();
+        ConsoleStreamHook.Uninstall();
+    }
+
     [Fact]
-    public void AddConsoleLogsServices_WhenNoProviderIsConfigured_UsesInMemoryProvider()
+    public async Task AddConsoleLogsServices_ResolvesHostOwnedSingletons()
     {
         var services = new ServiceCollection();
-
         services.AddConsoleLogsServices();
 
-        using var serviceProvider = services.BuildServiceProvider();
+        await using var serviceProvider = services.BuildServiceProvider();
+
         Assert.IsType<InMemoryConsoleLogProvider>(serviceProvider.GetRequiredService<IConsoleLogProvider>());
         Assert.IsType<ConsoleLogSourceRegistry>(serviceProvider.GetRequiredService<IConsoleLogSourceRegistry>());
         Assert.IsType<ConsoleLogRedactor>(serviceProvider.GetRequiredService<IConsoleLogRedactor>());
+        Assert.Same(ConsoleLogsHost.ScopeAccessor, serviceProvider.GetRequiredService<ConsoleLogScopeAccessor>());
+        Assert.Contains(serviceProvider.GetServices<ILoggerProvider>(), x => ReferenceEquals(x, ConsoleLogsHost.ScopeAccessor));
+        Assert.Contains(serviceProvider.GetServices<IHostedService>(), x => x.GetType() == typeof(ConsoleLogsHostedService));
+#pragma warning disable CS0618
+        Assert.IsType<ConsoleLogCaptureAdapter>(serviceProvider.GetRequiredService<IConsoleLogCapture>());
+#pragma warning restore CS0618
+
+        // Same instance must be visible across separate shell containers — console output is process-wide.
+        var services2 = new ServiceCollection();
+        services2.AddConsoleLogsServices();
+        await using var serviceProvider2 = services2.BuildServiceProvider();
+
+        Assert.Same(serviceProvider.GetRequiredService<IConsoleLogProvider>(), serviceProvider2.GetRequiredService<IConsoleLogProvider>());
+        Assert.Same(serviceProvider.GetRequiredService<IConsoleLogSourceRegistry>(), serviceProvider2.GetRequiredService<IConsoleLogSourceRegistry>());
     }
 
     [Fact]
-    public void AddConsoleLogsServices_RegistersHostedCapture()
+    public async Task AddConsoleLogsServices_UsesCustomProviderForCapturePipeline()
     {
+        var provider = new CustomProvider();
         var services = new ServiceCollection();
-
+        services.AddSingleton<IConsoleLogProvider>(provider);
         services.AddConsoleLogsServices();
 
-        using var serviceProvider = services.BuildServiceProvider();
+        await using var serviceProvider = services.BuildServiceProvider();
-        Assert.Contains(serviceProvider.GetServices<IHostedService>(), x => x.GetType() == typeof(ConsoleLogCaptureHostedService));
+
+        _ = serviceProvider.GetRequiredService<ConsoleLogScopeAccessor>();
+
+        foreach (var hostedService in serviceProvider.GetServices<IHostedService>())
+            await hostedService.StartAsync(CancellationToken.None);
+
+        Assert.Same(provider, serviceProvider.GetRequiredService<IConsoleLogProvider>());
+        Assert.Same(provider, ConsoleLogsHost.Provider);
+
+        foreach (var hostedService in serviceProvider.GetServices<IHostedService>())
+            await hostedService.StopAsync(CancellationToken.None);
+    }
+
+    [Fact]
+    public async Task AddConsoleLogsServices_UsesCustomProviderRegisteredAfterConsoleLogsServices()
+    {
+        var provider = new CustomProvider();
+        var services = new ServiceCollection();
+        services.AddConsoleLogsServices();
+        services.AddSingleton<IConsoleLogProvider>(provider);
+
+        await using var serviceProvider = services.BuildServiceProvider();
+
+        foreach (var hostedService in serviceProvider.GetServices<IHostedService>())
+            await hostedService.StartAsync(CancellationToken.None);
+
+        Assert.Same(provider, serviceProvider.GetRequiredService<IConsoleLogProvider>());
+        Assert.Same(provider, ConsoleLogsHost.Provider);
+
+        foreach (var hostedService in serviceProvider.GetServices<IHostedService>())
+            await hostedService.StopAsync(CancellationToken.None);
+    }
+
+    [Fact]
+    public async Task AddConsoleLogsServices_CreatesCustomProviderWithHostOwnedDependencies()
+    {
+        var services = new ServiceCollection();
+        services.AddConsoleLogsServices();
+        services.AddSingleton<IConsoleLogProvider, ProviderWithHostDependencies>();
+
+        await using var serviceProvider = services.BuildServiceProvider();
+
+        foreach (var hostedService in serviceProvider.GetServices<IHostedService>())
+            await hostedService.StartAsync(CancellationToken.None);
+
+        var provider = Assert.IsType<ProviderWithHostDependencies>(ConsoleLogsHost.Provider);
+        Assert.Same(serviceProvider.GetRequiredService<IConsoleLogProvider>(), provider);
+        Assert.Same(ConsoleLogsHost.Options, provider.Options);
+        Assert.Same(ConsoleLogsHost.SourceRegistry, provider.SourceRegistry);
+        Assert.Same(ConsoleLogsHost.Redactor, provider.Redactor);
+        Assert.Same(ConsoleLogsHost.Formatter, provider.Formatter);
+        Assert.Same(ConsoleLogsHost.ScopeAccessor, provider.ScopeAccessor);
+
+        foreach (var hostedService in serviceProvider.GetServices<IHostedService>())
+            await hostedService.StopAsync(CancellationToken.None);
+    }
+
+    [Fact]
+    public async Task ShutdownAsync_ResetsReferenceLeases()
+    {
+        ConsoleLogsHost.AddReference();
+        ConsoleLogsHost.EnsureInitialized();
+
+        await ConsoleLogsHost.ShutdownAsync();
+
+        ConsoleLogsHost.AddReference();
+        ConsoleLogsHost.EnsureInitialized();
+        await ConsoleLogsHost.ReleaseReferenceAsync();
+
+        var provider = new CustomProvider();
+        Assert.True(ConsoleLogsHost.ConfigureProvider((_, _) => provider));
+        Assert.Same(provider, ConsoleLogsHost.Provider);
+    }
+
+    [Fact]
+    public async Task AddConsoleLogsHost_RegistersHostedFlushService()
+    {
+        var services = new ServiceCollection();
+        services.AddConsoleLogsHost();
+
+        await using var serviceProvider = services.BuildServiceProvider();
+
+        Assert.Contains(serviceProvider.GetServices<IHostedService>(), x => x.GetType() == typeof(ConsoleLogsHostedService));
+        Assert.Contains(serviceProvider.GetServices<ILoggerProvider>(), x => ReferenceEquals(x, ConsoleLogsHost.ScopeAccessor));
+    }
+
+    [Fact]
+    public void AddConsoleLogsHost_AppliesConfigurationWhenCalledBeforeFirstAccess()
+    {
+        var services = new ServiceCollection();
+        services.AddConsoleLogsHost(options => options.RecentLogCapacity = 17);
+
+        Assert.Equal(17, ConsoleLogsHost.Options.Value.RecentLogCapacity);
+    }
+
+    [Fact]
+    public void AddConsoleLogsHost_DoesNotInitializeHostDuringRegistration()
+    {
+        var services = new ServiceCollection();
+        services.AddConsoleLogsHost();
+
+        Assert.True(ConsoleLogsHost.Configure(options => options.RecentLogCapacity = 23));
+        Assert.Equal(23, ConsoleLogsHost.Options.Value.RecentLogCapacity);
+    }
+
+    private sealed class CustomProvider : IConsoleLogProvider
+    {
+        public ValueTask PublishAsync(ConsoleLogLine line, CancellationToken cancellationToken = default) => ValueTask.CompletedTask;
+
+        public ValueTask<RecentConsoleLogsResult> GetRecentAsync(ConsoleLogFilter filter, CancellationToken cancellationToken = default) =>
+            ValueTask.FromResult(new RecentConsoleLogsResult([]));
+
+        public async IAsyncEnumerable<ConsoleLogStreamItem> SubscribeAsync(ConsoleLogFilter filter, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+        {
+            await Task.CompletedTask;
+            yield break;
+        }
+
+        public ValueTask<IReadOnlyCollection<ConsoleLogSource>> ListSourcesAsync(CancellationToken cancellationToken = default) =>
+            ValueTask.FromResult<IReadOnlyCollection<ConsoleLogSource>>([]);
+    }
+
+    private sealed class ProviderWithHostDependencies(
+        IOptions<ConsoleLogsOptions> options,
+        IConsoleLogSourceRegistry sourceRegistry,
+        IConsoleLogRedactor redactor,
+        ConsoleLineFormatter formatter,
+        ConsoleLogScopeAccessor scopeAccessor) : IConsoleLogProvider
+    {
+        public IOptions<ConsoleLogsOptions> Options { get; } = options;
+        public IConsoleLogSourceRegistry SourceRegistry { get; } = sourceRegistry;
+        public IConsoleLogRedactor Redactor { get; } = redactor;
+        public ConsoleLineFormatter Formatter { get; } = formatter;
+        public ConsoleLogScopeAccessor ScopeAccessor { get; } = scopeAccessor;
+
+        public ValueTask PublishAsync(ConsoleLogLine line, CancellationToken cancellationToken = default) => ValueTask.CompletedTask;
+
+        public ValueTask<RecentConsoleLogsResult> GetRecentAsync(ConsoleLogFilter filter, CancellationToken cancellationToken = default) =>
+            ValueTask.FromResult(new RecentConsoleLogsResult([]));
+
+        public async IAsyncEnumerable<ConsoleLogStreamItem> SubscribeAsync(ConsoleLogFilter filter, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+        {
+            await Task.CompletedTask;
+            yield break;
+        }
+
+        public ValueTask<IReadOnlyCollection<ConsoleLogSource>> ListSourcesAsync(CancellationToken cancellationToken = default) =>
+            ValueTask.FromResult<IReadOnlyCollection<ConsoleLogSource>>([]);
     }
 }

test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Filtering/ConsoleLogFilterTests.cs

@@ -9,6 +9,7 @@ public class ConsoleLogFilterTests
         Text = "workflow started",
         Stream = ConsoleLogStream.Stdout,
         Source = new ConsoleLogSource { Id = "source-a", DisplayName = "Source A", MachineName = "machine" },
+        WorkflowInstanceId = "workflow-instance-a",
         ReceivedAt = DateTimeOffset.Parse("2026-05-18T10:00:00Z")
     };
 
@@ -33,4 +34,11 @@ public class ConsoleLogFilterTests
         Assert.False(ConsoleLogFilterEvaluator.Matches(_line, new ConsoleLogFilter { Query = "missing" }));
         Assert.False(ConsoleLogFilterEvaluator.Matches(_line, new ConsoleLogFilter { From = _line.ReceivedAt.AddSeconds(1) }));
     }
+
+    [Fact]
+    public void Matches_FiltersByWorkflowInstanceId()
+    {
+        Assert.True(ConsoleLogFilterEvaluator.Matches(_line, new ConsoleLogFilter { WorkflowInstanceId = "workflow-instance-a" }));
+        Assert.False(ConsoleLogFilterEvaluator.Matches(_line, new ConsoleLogFilter { WorkflowInstanceId = "workflow-instance-b" }));
+    }
 }

test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Redaction/ConsoleLogProviderRedactionTests.cs

@@ -3,39 +3,68 @@ using Elsa.Diagnostics.ConsoleLogs.Services;
 
 namespace Elsa.Diagnostics.ConsoleLogs.UnitTests.Redaction;
 
-public class ConsoleLogProviderRedactionTests
+[Collection(ConsoleHostStateCollection.Name)]
+public class ConsoleLogProviderRedactionTests : IAsyncLifetime
 {
+    public async Task InitializeAsync()
+    {
+        await ConsoleLogsHost.ShutdownAsync();
+        ConsoleStreamHook.Uninstall();
+    }
+
+    public async Task DisposeAsync()
+    {
+        await ConsoleLogsHost.ShutdownAsync();
+        ConsoleStreamHook.Uninstall();
+    }
+
     [Fact]
     public async Task CaptureTee_PublishesRedactedLinesToProvider()
     {
         var originalOut = Console.Out;
         var provider = new CapturingProvider();
-        var registry = new ConsoleLogSourceRegistry(Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions()));
         var options = Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions());
-        var capture = new ConsoleCaptureTee(provider, registry, new ConsoleLogRedactor(options), new ConsoleLineFormatter(options), options);
 
         try
         {
+            ConsoleStreamHook.Uninstall();
             Console.SetOut(TextWriter.Null);
-            await capture.StartAsync();
+            await using var capture = new ConsoleCaptureTee(
+                provider,
+                new ConsoleLogSourceRegistry(options),
+                new ConsoleLogRedactor(options),
+                new ConsoleLineFormatter(options),
+                new ConsoleLogScopeAccessor(),
+                options);
+
             Console.WriteLine(string.Concat("pass", "word", "=", "sample-value"));
-            await capture.StopAsync();
+            await WaitForLineAsync(provider);
+
             Assert.Equal("[Redacted]", Assert.Single(provider.Lines).Text);
         }
         finally
         {
-            await capture.StopAsync();
+            ConsoleStreamHook.Uninstall();
             Console.SetOut(originalOut);
         }
     }
 
+    private static async Task WaitForLineAsync(CapturingProvider provider)
+    {
+        var deadline = DateTimeOffset.UtcNow.AddSeconds(2);
+        while (provider.Lines.Count == 0 && DateTimeOffset.UtcNow < deadline)
+            await Task.Delay(10);
+    }
+
     private sealed class CapturingProvider : IConsoleLogProvider
     {
         public List<ConsoleLogLine> Lines { get; } = [];
 
         public ValueTask PublishAsync(ConsoleLogLine line, CancellationToken cancellationToken = default)
         {
-            Lines.Add(line);
+            lock (Lines)
+                Lines.Add(line);
+
             return ValueTask.CompletedTask;
         }
 

test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Redaction/ConsoleLogRedactorTests.cs

@@ -16,6 +16,26 @@ public class ConsoleLogRedactorTests
         Assert.Equal("Authorization: [Redacted]", redacted.Text);
     }
 
+    [Fact]
+    public void Redact_MasksSensitiveLineTextSplitByAnsi()
+    {
+        var line = CreateLine("to\u001b[31mken=sample-token");
+
+        var redacted = _redactor.Redact(line);
+
+        Assert.Equal("[Redacted]", redacted.Text);
+    }
+
+    [Fact]
+    public void Redact_PreservesAnsiWhenLineTextIsNotSensitive()
+    {
+        var line = CreateLine("\u001b[31mhello\u001b[0m");
+
+        var redacted = _redactor.Redact(line);
+
+        Assert.Equal("\u001b[31mhello\u001b[0m", redacted.Text);
+    }
+
     [Fact]
     public void Redact_MasksSensitiveSourceMetadata()
     {

test/unit/Elsa.Scheduling.UnitTests/StartupTasks/CreateSchedulesStartupTaskTests.cs

@@ -0,0 +1,78 @@
+using Elsa.Common;
+using Elsa.Common.Multitenancy;
+using Elsa.Scheduling.StartupTasks;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Entities;
+using Elsa.Workflows.Runtime.Filters;
+using Elsa.Workflows.Runtime.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using NSubstitute;
+
+namespace Elsa.Scheduling.UnitTests.StartupTasks;
+
+public class CreateSchedulesStartupTaskTests
+{
+    private readonly StoredTrigger[] _triggers = [new() { WorkflowDefinitionId = "definition", WorkflowDefinitionVersionId = "version", ActivityId = "activity" }];
+    private readonly StoredBookmark[] _bookmarks = [new() { Hash = "hash", WorkflowInstanceId = "instance" }];
+    private readonly ITriggerStore _triggerStore = Substitute.For<ITriggerStore>();
+    private readonly IBookmarkStore _bookmarkStore = Substitute.For<IBookmarkStore>();
+    private readonly ITriggerScheduler _triggerScheduler = Substitute.For<ITriggerScheduler>();
+    private readonly IBookmarkScheduler _bookmarkScheduler = Substitute.For<IBookmarkScheduler>();
+
+    public CreateSchedulesStartupTaskTests()
+    {
+        _triggerStore.FindManyAsync(Arg.Any<TriggerFilter>(), Arg.Any<CancellationToken>()).Returns(_triggers);
+        _bookmarkStore.FindManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<CancellationToken>()).Returns(_bookmarks);
+    }
+
+    [Fact]
+    public void Task_DependsOnPopulateRegistriesStartupTask()
+    {
+        var dependency = Assert.Single(typeof(CreateSchedulesStartupTask).GetCustomAttributes(typeof(TaskDependencyAttribute), false).Cast<TaskDependencyAttribute>());
+
+        Assert.Equal(typeof(PopulateRegistriesStartupTask), dependency.DependencyTaskType);
+    }
+
+    [Fact]
+    public async Task ExecuteAsync_WithoutTenantBackgroundQueue_SchedulesImmediately()
+    {
+        var task = new CreateSchedulesStartupTask(CreateServiceProvider());
+
+        await task.ExecuteAsync(CancellationToken.None);
+
+        await _triggerScheduler.Received(1).ScheduleAsync(Arg.Is<IEnumerable<StoredTrigger>>(x => x.SequenceEqual(_triggers)), Arg.Any<CancellationToken>());
+        await _bookmarkScheduler.Received(1).ScheduleAsync(Arg.Is<IEnumerable<StoredBookmark>>(x => x.SequenceEqual(_bookmarks)), Arg.Any<CancellationToken>());
+    }
+
+    [Fact]
+    public async Task ExecuteAsync_WithTenantBackgroundQueue_EnqueuesScheduleCreation()
+    {
+        TenantBackgroundWorkItem? workItem = null;
+        var workQueue = Substitute.For<ITenantBackgroundWorkQueue>();
+        workQueue.EnqueueAsync(Arg.Do<TenantBackgroundWorkItem>(x => workItem = x), Arg.Any<CancellationToken>()).Returns(ValueTask.CompletedTask);
+        var serviceProvider = CreateServiceProvider(services => services.AddSingleton(workQueue));
+        var task = new CreateSchedulesStartupTask(serviceProvider);
+
+        await task.ExecuteAsync(CancellationToken.None);
+
+        await workQueue.Received(1).EnqueueAsync(Arg.Any<TenantBackgroundWorkItem>(), Arg.Any<CancellationToken>());
+        await _triggerScheduler.DidNotReceive().ScheduleAsync(Arg.Any<IEnumerable<StoredTrigger>>(), Arg.Any<CancellationToken>());
+
+        Assert.NotNull(workItem);
+        await workItem(serviceProvider, CancellationToken.None);
+
+        await _triggerScheduler.Received(1).ScheduleAsync(Arg.Is<IEnumerable<StoredTrigger>>(x => x.SequenceEqual(_triggers)), Arg.Any<CancellationToken>());
+        await _bookmarkScheduler.Received(1).ScheduleAsync(Arg.Is<IEnumerable<StoredBookmark>>(x => x.SequenceEqual(_bookmarks)), Arg.Any<CancellationToken>());
+    }
+
+    private ServiceProvider CreateServiceProvider(Action<IServiceCollection>? configureServices = null)
+    {
+        var services = new ServiceCollection();
+        services.AddSingleton(_triggerStore);
+        services.AddSingleton(_bookmarkStore);
+        services.AddSingleton(_triggerScheduler);
+        services.AddSingleton(_bookmarkScheduler);
+        configureServices?.Invoke(services);
+        return services.BuildServiceProvider();
+    }
+}

test/unit/Elsa.Workflows.Runtime.UnitTests/Distributed/DistributedRuntimeLockProviderValidatorTests.cs

@@ -3,6 +3,7 @@ using Elsa.Common.DistributedHosting.DistributedLocks;
 using Elsa.Workflows.Runtime.Distributed;
 using Medallion.Threading;
 using Medallion.Threading.FileSystem;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
 
 namespace Elsa.Workflows.Runtime.UnitTests.Distributed;
@@ -19,14 +20,24 @@ public class DistributedRuntimeLockProviderValidatorTests : IDisposable
     }
 
     [Fact]
-    public void Validate_Throws_WhenFileSystemProviderIsUsedWithoutOptIn()
+    public void Validate_DoesNotThrow_WhenFileSystemProviderIsUsedWithoutOptIn()
     {
         var validator = CreateValidator(_fileProvider);
 
-        var exception = Assert.Throws<InvalidOperationException>(validator.Validate);
+        validator.Validate();
+    }
 
-        Assert.Contains(nameof(FileDistributedSynchronizationProvider), exception.Message);
+    [Fact]
-        Assert.Contains(nameof(DistributedLockingOptions.AllowLocalLockProviderInDistributedRuntime), exception.Message);
+    public void Validate_LogsWarning_WhenFileSystemProviderIsUsedWithoutOptIn()
+    {
+        var logger = new CapturingLogger<DistributedRuntimeLockProviderValidator>();
+        var validator = CreateValidator(_fileProvider, logger: logger);
+
+        validator.Validate();
+
+        Assert.Contains(logger.Entries, entry =>
+            entry.Level == LogLevel.Warning &&
+            entry.Message.Contains("local-only lock provider", StringComparison.OrdinalIgnoreCase));
     }
 
     [Fact]
@@ -38,48 +49,49 @@ public class DistributedRuntimeLockProviderValidatorTests : IDisposable
     }
 
     [Fact]
-    public void Validate_Throws_WhenWrappedProviderUsesFileSystemProvider()
+    public void Validate_DoesNotThrow_WhenWrappedProviderUsesFileSystemProvider()
     {
         var validator = CreateValidator(new WrappedDistributedLockProvider(_fileProvider));
 
-        var exception = Assert.Throws<InvalidOperationException>(validator.Validate);
+        validator.Validate();
-
-        Assert.Contains(nameof(FileDistributedSynchronizationProvider), exception.Message);
-        Assert.Contains(nameof(WrappedDistributedLockProvider), exception.Message);
     }
 
     [Fact]
-    public void Validate_Throws_WhenProviderUsesFileSystemProviderThroughCustomProperty()
+    public void Validate_LogsWarning_WhenWrappedProviderUsesFileSystemProvider()
+    {
+        var logger = new CapturingLogger<DistributedRuntimeLockProviderValidator>();
+        var validator = CreateValidator(new WrappedDistributedLockProvider(_fileProvider), logger: logger);
+
+        validator.Validate();
+
+        Assert.Contains(logger.Entries, entry =>
+            entry.Level == LogLevel.Warning &&
+            entry.Message.Contains(nameof(WrappedDistributedLockProvider), StringComparison.Ordinal));
+    }
+
+    [Fact]
+    public void Validate_DoesNotThrow_WhenProviderUsesFileSystemProviderThroughCustomProperty()
     {
         var validator = CreateValidator(new CustomWrappedDistributedLockProvider(_fileProvider));
 
-        var exception = Assert.Throws<InvalidOperationException>(validator.Validate);
+        validator.Validate();
-
-        Assert.Contains(nameof(FileDistributedSynchronizationProvider), exception.Message);
-        Assert.Contains(nameof(CustomWrappedDistributedLockProvider), exception.Message);
     }
 
     [Fact]
-    public void Validate_Throws_WhenProviderUsesFileSystemProviderThroughProviderCollection()
+    public void Validate_DoesNotThrow_WhenProviderUsesFileSystemProviderThroughProviderCollection()
     {
         var validator = CreateValidator(new CompositeDistributedLockProvider([new CustomDistributedLockProvider(), _fileProvider]));
 
-        var exception = Assert.Throws<InvalidOperationException>(validator.Validate);
+        validator.Validate();
-
-        Assert.Contains(nameof(FileDistributedSynchronizationProvider), exception.Message);
-        Assert.Contains(nameof(CompositeDistributedLockProvider), exception.Message);
     }
 
     [Fact]
-    public void Validate_Throws_WhenProviderUsesFileSystemProviderThroughNonGenericProviderCollection()
+    public void Validate_DoesNotThrow_WhenProviderUsesFileSystemProviderThroughNonGenericProviderCollection()
     {
         System.Collections.IEnumerable providers = new IDistributedLockProvider[] { new CustomDistributedLockProvider(), _fileProvider };
         var validator = CreateValidator(new NonGenericCompositeDistributedLockProvider(providers));
 
-        var exception = Assert.Throws<InvalidOperationException>(validator.Validate);
+        validator.Validate();
-
-        Assert.Contains(nameof(FileDistributedSynchronizationProvider), exception.Message);
-        Assert.Contains(nameof(NonGenericCompositeDistributedLockProvider), exception.Message);
     }
 
     [Fact]
@@ -99,14 +111,11 @@ public class DistributedRuntimeLockProviderValidatorTests : IDisposable
     }
 
     [Fact]
-    public void Validate_Throws_WhenNullableProviderCollectionContainsFileSystemProvider()
+    public void Validate_DoesNotThrow_WhenNullableProviderCollectionContainsFileSystemProvider()
     {
         var validator = CreateValidator(new NullableCompositeDistributedLockProvider([new CustomDistributedLockProvider(), null, _fileProvider]));
 
-        var exception = Assert.Throws<InvalidOperationException>(validator.Validate);
+        validator.Validate();
-
-        Assert.Contains(nameof(FileDistributedSynchronizationProvider), exception.Message);
-        Assert.Contains(nameof(NullableCompositeDistributedLockProvider), exception.Message);
     }
 
     [Fact]
@@ -141,20 +150,15 @@ public class DistributedRuntimeLockProviderValidatorTests : IDisposable
             new ValueEqualDistributedLockProvider("same", _fileProvider)
         ]));
 
-        var exception = Assert.Throws<InvalidOperationException>(validator.Validate);
+        validator.Validate();
-
-        Assert.Contains(nameof(FileDistributedSynchronizationProvider), exception.Message);
-        Assert.Contains(nameof(CompositeDistributedLockProvider), exception.Message);
     }
 
     [Fact]
-    public void Validate_Throws_WhenNoopProviderIsUsedWithoutOptIn()
+    public void Validate_DoesNotThrow_WhenNoopProviderIsUsedWithoutOptIn()
     {
         var validator = CreateValidator(new NoopDistributedSynchronizationProvider());
 
-        var exception = Assert.Throws<InvalidOperationException>(validator.Validate);
+        validator.Validate();
-
-        Assert.Contains(nameof(NoopDistributedSynchronizationProvider), exception.Message);
     }
 
     [Fact]
@@ -173,12 +177,15 @@ public class DistributedRuntimeLockProviderValidatorTests : IDisposable
         validator.Validate();
     }
 
-    private static DistributedRuntimeLockProviderValidator CreateValidator(IDistributedLockProvider provider, Action<DistributedLockingOptions>? configure = null)
+    private static DistributedRuntimeLockProviderValidator CreateValidator(
+        IDistributedLockProvider provider,
+        Action<DistributedLockingOptions>? configure = null,
+        ILogger<DistributedRuntimeLockProviderValidator>? logger = null)
     {
         var options = new DistributedLockingOptions();
         configure?.Invoke(options);
 
-        return new(provider, Microsoft.Extensions.Options.Options.Create(options), NullLogger<DistributedRuntimeLockProviderValidator>.Instance);
+        return new(provider, Microsoft.Extensions.Options.Options.Create(options), logger ?? NullLogger<DistributedRuntimeLockProviderValidator>.Instance);
     }
 
     public void Dispose()
@@ -298,4 +305,32 @@ public class DistributedRuntimeLockProviderValidatorTests : IDisposable
 
         public ValueTask<IDistributedSynchronizationHandle?> TryAcquireAsync(TimeSpan timeout = default, CancellationToken cancellationToken = default) => throw new NotSupportedException();
     }
+
+    private sealed class CapturingLogger<T> : ILogger<T>
+    {
+        public List<LogEntry> Entries { get; } = [];
+
+        public IDisposable BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;
+        public bool IsEnabled(LogLevel logLevel) => true;
+
+        public void Log<TState>(
+            LogLevel logLevel,
+            EventId eventId,
+            TState state,
+            Exception? exception,
+            Func<TState, Exception?, string> formatter)
+        {
+            Entries.Add(new(logLevel, formatter(state, exception)));
+        }
+    }
+
+    private sealed record LogEntry(LogLevel Level, string Message);
+
+    private sealed class NullScope : IDisposable
+    {
+        public static readonly NullScope Instance = new();
+        public void Dispose()
+        {
+        }
+    }
 }