fix(management): Flow Debugger Live result is the redacted, capped trace copy (arch-review F1, central half)

RunMethodAsync collected sink events into a plain ConcurrentQueue and always
reported Truncated: false, so the Live result carried RAW audit events — the
per-target SQL-parameter overrides, header scrubbing and 8 KB / 64 KB payload
caps that Replay reads back never applied to what the operator saw, and the
reply had no upper bound at all.

It now collects through the same FlowTraceCollector the site launcher uses,
constructed with the host's IAuditRedactor (AddAuditLog's singleton; a host
without one collects raw but is still capped by count and bytes), and reports
Truncated from the collector. Complete() is called as soon as the run returns,
so a late un-awaited call cannot mutate the snapshot the reply is built from.

Claude-Session: https://claude.ai/code/session_01NcMG6xXdsJbmD2wYAgkK29

src/ZB.MOM.WW.ScadaBridge.ManagementService/FlowDebug/FlowDebugRunner.cs

@@ -1,4 +1,3 @@
-using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.Text.Json;
 using Microsoft.Extensions.DependencyInjection;
@@ -83,6 +82,14 @@ public sealed class FlowDebugRunner
     /// Live run of an inbound API method (design §7.1): active-node gate, schema validation,
     /// synthesized inbound row (actor = operator), traced execution through the real
     /// <see cref="InboundScriptExecutor"/> with a trace-sink <see cref="RouteHelper"/>.
+    ///
+    /// <para>
+    /// Arch-review F1: the sink is a <see cref="FlowTraceCollector"/> — the same collector the
+    /// site launcher fills — so the returned copy is REDACTED and capped by count and bytes,
+    /// and a Live result shows exactly the filtered rows a later Replay reads back. The
+    /// redactor is <c>AddAuditLog</c>'s <see cref="IAuditRedactor"/> singleton; a host that
+    /// registered none (a test composition root) collects raw but is still capped.
+    /// </para>
     /// </summary>
     public async Task<FlowDebugRunResult> RunMethodAsync(
         IServiceProvider sp, FlowDebugRunMethodCommand cmd, AuthenticatedUser user, CancellationToken ct)
@@ -106,7 +113,7 @@ public sealed class FlowDebugRunner
         }
 
         var root = Guid.NewGuid();
-        var events = new ConcurrentQueue<AuditEvent>();
+        var collector = new FlowTraceCollector(redactor: sp.GetService<IAuditRedactor>());
         var startedAt = DateTimeOffset.UtcNow;
         var sw = Stopwatch.StartNew();
         var timeout = TimeSpan.FromSeconds(Math.Max(1, method.TimeoutSeconds));
@@ -116,7 +123,7 @@ public sealed class FlowDebugRunner
         // only known once the script runs. So the operator's scope is pushed into the
         // RouteHelper and enforced at route time, before any site traffic.
         var scope = await ResolvePermittedSiteIdentifiersAsync(sp, user, ct).ConfigureAwait(false);
-        var route = sp.GetRequiredService<RouteHelper>().WithTraceSink(events.Enqueue).WithSiteScope(scope);
+        var route = sp.GetRequiredService<RouteHelper>().WithTraceSink(collector.Add).WithSiteScope(scope);
         var executor = sp.GetRequiredService<InboundScriptExecutor>();
 
         InboundScriptResult result;
@@ -137,6 +144,9 @@ public sealed class FlowDebugRunner
         }
 
         var finishedAt = DateTimeOffset.UtcNow;
+        // Close the sink here, not at the point the list is built: an un-awaited call finishing
+        // after the run returned must not mutate the snapshot the reply is assembled from.
+        var events = collector.Complete();
         var requestJson = cmd.Parameters is { } p ? p.GetRawText() : null;
         var inboundRow = ScadaBridgeAuditEventFactory.Create(
             AuditChannel.ApiInbound, AuditKind.InboundRequest,
@@ -168,7 +178,7 @@ public sealed class FlowDebugRunner
 
         return new FlowDebugRunResult(
             root, result.Success, result.ErrorMessage, result.Success ? null : result.ErrorCode,
-            ordered, Truncated: false, startedAt, finishedAt);
+            ordered, Truncated: collector.Truncated, startedAt, finishedAt);
     }
 
     /// <summary>

tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/FlowDebugRunnerTests.cs

@@ -331,4 +331,62 @@ public class FlowDebugRunnerTests
         Assert.NotNull(actual);
         Assert.Equal(new[] { "Administrator", "Deployer", "Viewer" }, actual!.OrderBy(r => r, StringComparer.Ordinal));
     }
+
+    // --- Arch-review F1: the Live result is the REDACTED, capped trace copy ---
+    //
+    // The runner collects sink events through the same FlowTraceCollector the site launcher
+    // uses, so central and site filter identically and a Live run shows exactly the rows a
+    // later Replay reads back out of the audit store.
+
+    /// <summary>Wires a method whose handler pushes <paramref name="events"/> straight at the trace sink.</summary>
+    private void WireSinkPushingMethod(int methodId, IEnumerable<AuditEvent> events)
+    {
+        var method = new ApiMethod("M", "x") { Id = methodId, TimeoutSeconds = 5 };
+        _apiRepo.GetApiMethodByIdAsync(methodId, Arg.Any<CancellationToken>()).Returns(method);
+        _executor.RegisterHandler("M", ctx =>
+        {
+            foreach (var e in events) ctx.Route.TraceSink!(e);
+            return Task.FromResult<object?>(null);
+        });
+    }
+
+    private static AuditEvent TraceRow(string? detailsJson) =>
+        ScadaBridgeAuditEventFactory.Create(
+            AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
+            executionId: Guid.NewGuid(), target: "work") with
+        { DetailsJson = detailsJson };
+
+    [Fact]
+    public async Task RunMethod_SinkEvents_AreRedacted_WhenARedactorIsRegistered()
+    {
+        WireSinkPushingMethod(31, [TraceRow("{\"secret\":\"abc\"}")]);
+        _services.AddSingleton<IAuditRedactor>(new UpperRedactor());
+
+        var result = await Runner().RunMethodAsync(_services.BuildServiceProvider(),
+            new FlowDebugRunMethodCommand(31, null), Deployer, CancellationToken.None);
+
+        Assert.True(result.Completed, result.Error);
+        Assert.Equal(2, result.Events.Count); // inbound row + the one sink event
+        Assert.Equal("{\"SECRET\":\"ABC\"}", result.Events[1].DetailsJson);
+        Assert.False(result.Truncated);
+    }
+
+    [Fact]
+    public async Task RunMethod_OverCap_ReportsTruncated()
+    {
+        WireSinkPushingMethod(32, Enumerable.Range(0, 501).Select(_ => TraceRow("{}")).ToList());
+
+        var result = await Runner().RunMethodAsync(_services.BuildServiceProvider(),
+            new FlowDebugRunMethodCommand(32, null), Deployer, CancellationToken.None);
+
+        Assert.True(result.Truncated);
+        Assert.Equal(501, result.Events.Count); // inbound row + the 500 kept
+    }
+
+    /// <summary>Upper-cases <c>DetailsJson</c> so a test can see the filtered copy land.</summary>
+    private sealed class UpperRedactor : IAuditRedactor
+    {
+        public AuditEvent Apply(AuditEvent rawEvent)
+            => rawEvent with { DetailsJson = rawEvent.DetailsJson?.ToUpperInvariant() };
+    }
 }