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

using System.Text.Json;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using NSubstitute;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.InboundApi;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.InboundAPI;
using ZB.MOM.WW.ScadaBridge.ManagementService.FlowDebug;

namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests;

/// <summary>
/// Flow Debugger design 2026-09-05 §7.1: the runner behind the three FlowDebug* commands —
/// standby gate, schema validation before any routing, synthesized inbound row + sink events
/// on a Live run, the value-change poll loop, and Replay from the repository.
/// </summary>
public class FlowDebugRunnerTests
{
    private static readonly AuthenticatedUser Deployer = new("ops", "Ops", new[] { "Deployer" }, Array.Empty<string>());

    private readonly ServiceCollection _services = new();
    private readonly IInboundApiRepository _apiRepo = Substitute.For<IInboundApiRepository>();
    private readonly IInstanceLocator _locator = Substitute.For<IInstanceLocator>();
    private readonly IInstanceRouter _router = Substitute.For<IInstanceRouter>();
    private readonly List<AuditEvent> _central = new();
    private readonly IActiveNodeGate _gate = Substitute.For<IActiveNodeGate>();
    private readonly InboundScriptExecutor _executor = new(NullLogger<InboundScriptExecutor>.Instance, Substitute.For<IServiceProvider>());
    private DateTimeOffset _clock = new(2026, 9, 5, 12, 0, 0, TimeSpan.Zero);

    public FlowDebugRunnerTests()
    {
        _gate.IsActiveNode.Returns(true);
        var writer = Substitute.For<ICentralAuditWriter>();
        writer.WriteAsync(Arg.Do<AuditEvent>(_central.Add), Arg.Any<CancellationToken>()).Returns(Task.CompletedTask);
        _services.AddSingleton(_apiRepo);
        _services.AddSingleton(_gate);
        _services.AddSingleton(writer);
        _services.AddSingleton(_locator);
        _services.AddSingleton(_router);
        _services.AddSingleton(_executor);
        _services.AddScoped(sp => new RouteHelper(_locator, _router, sp.GetService<ICentralAuditWriter>(), null));
    }

    private FlowDebugRunner Runner() => new(
        NullLogger<FlowDebugRunner>.Instance,
        (d, _) => { _clock += d; return Task.CompletedTask; },
        () => _clock);

    private static JsonElement Json(string s) => JsonDocument.Parse(s).RootElement.Clone();

    [Fact]
    public async Task RunMethod_OnStandby_RefusesWithStandbyCode_BeforeTouchingTheMethod()
    {
        _gate.IsActiveNode.Returns(false);
        var ex = await Assert.ThrowsAsync<ManagementCommandException>(() =>
            Runner().RunMethodAsync(_services.BuildServiceProvider(), new FlowDebugRunMethodCommand(1, null), Deployer, CancellationToken.None));
        Assert.Equal(FlowDebugRunner.StandbyNodeCode, ex.Code);
        Assert.Contains("ACTIVE central node", ex.Message);
        await _apiRepo.DidNotReceiveWithAnyArgs().GetApiMethodByIdAsync(default, default);
    }

    [Fact]
    public async Task RunMethod_InvalidParameters_FailsBeforeAnyRouteCall()
    {
        var method = new ApiMethod("M", "return 1;")
        {
            Id = 3, TimeoutSeconds = 5,
            ParameterDefinitions = "{\"type\":\"object\",\"properties\":{\"n\":{\"type\":\"integer\"}},\"required\":[\"n\"]}",
        };
        _apiRepo.GetApiMethodByIdAsync(3, Arg.Any<CancellationToken>()).Returns(method);

        var ex = await Assert.ThrowsAsync<ManagementCommandException>(() =>
            Runner().RunMethodAsync(_services.BuildServiceProvider(), new FlowDebugRunMethodCommand(3, Json("{}")), Deployer, CancellationToken.None));
        Assert.Equal(FlowDebugRunner.ValidationFailedCode, ex.Code);
        await _router.DidNotReceiveWithAnyArgs().RouteToCallAsync(default!, default!, default);
        Assert.Empty(_central);
    }

    [Fact]
    public async Task RunMethod_HappyPath_ReturnsInboundRow_ThenSinkEvents_Ordered_AndWritesTheInboundRow()
    {
        var method = new ApiMethod("M", "x") { Id = 4, TimeoutSeconds = 5 };
        _apiRepo.GetApiMethodByIdAsync(4, Arg.Any<CancellationToken>()).Returns(method);
        _locator.GetSiteIdForInstanceAsync("inst", Arg.Any<CancellationToken>()).Returns("site-a");
        var child = Guid.NewGuid();
        var siteRun = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
            occurredAtUtc: DateTime.UtcNow.AddSeconds(1), executionId: child, target: "work");
        RouteToCallRequest? captured = null;
        _router.RouteToCallAsync("site-a", Arg.Do<RouteToCallRequest>(r => captured = r), Arg.Any<CancellationToken>())
            .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 7, null, DateTimeOffset.UtcNow, child, [siteRun]));
        _executor.RegisterHandler("M", ctx => ctx.Route.To("inst").Call("work"));

        var result = await Runner().RunMethodAsync(_services.BuildServiceProvider(),
            new FlowDebugRunMethodCommand(4, Json("{\"a\":1}")), Deployer, CancellationToken.None);

        Assert.True(result.Completed, result.Error);
        Assert.Null(result.ErrorCode);
        Assert.True(captured!.Trace);
        Assert.Equal(result.RootExecutionId, captured.ParentExecutionId);

        // inbound (synthesized) → RouteCall (sink) → site ScriptRun (sink, from the reply)
        Assert.Equal(3, result.Events.Count);
        var inbound = result.Events[0].AsRow();
        Assert.Equal(AuditKind.InboundRequest, inbound.Kind);
        Assert.Equal("ops", result.Events[0].Actor);
        Assert.Equal("M", result.Events[0].Target);
        Assert.Equal(result.RootExecutionId, inbound.ExecutionId);
        Assert.Equal("flow-debugger", ScriptTraceExtra.TryParse(inbound.Extra)!.Source);
        Assert.Contains("\"a\":1", inbound.RequestSummary);
        Assert.Equal(AuditKind.RouteCall, result.Events[1].AsRow().Kind);
        Assert.Same(siteRun, result.Events[2]);

        // Central writer saw the inbound row AND the always-on RouteCall row.
        Assert.Contains(_central, e => e.AsRow().Kind == AuditKind.InboundRequest);
        Assert.Contains(_central, e => e.AsRow().Kind == AuditKind.RouteCall);
    }

    [Fact]
    public async Task RunMethod_SiteClockBehindCentral_StillReturnsTheSinkEventsInEmissionOrder()
    {
        // Arch-review F5: the site's rows carry the SITE clock. A site 2 s behind central used to
        // re-sort ahead of the always-on central RouteCall row.
        var method = new ApiMethod("M", "x") { Id = 6, TimeoutSeconds = 5 };
        _apiRepo.GetApiMethodByIdAsync(6, Arg.Any<CancellationToken>()).Returns(method);
        _locator.GetSiteIdForInstanceAsync("inst", Arg.Any<CancellationToken>()).Returns("site-a");
        var child = Guid.NewGuid();
        var siteRun = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
            occurredAtUtc: DateTime.UtcNow.AddSeconds(-2), executionId: child, target: "work");
        _router.RouteToCallAsync("site-a", Arg.Any<RouteToCallRequest>(), Arg.Any<CancellationToken>())
            .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 7, null, DateTimeOffset.UtcNow, child, [siteRun]));
        _executor.RegisterHandler("M", ctx => ctx.Route.To("inst").Call("work"));

        var result = await Runner().RunMethodAsync(_services.BuildServiceProvider(),
            new FlowDebugRunMethodCommand(6, null), Deployer, CancellationToken.None);

        Assert.True(result.Completed, result.Error);
        Assert.Equal(3, result.Events.Count);
        Assert.Equal(AuditKind.InboundRequest, result.Events[0].AsRow().Kind);
        Assert.Equal(AuditKind.RouteCall, result.Events[1].AsRow().Kind);
        Assert.Same(siteRun, result.Events[2]);
    }

    [Fact]
    public async Task RunMethod_UnknownMethod_IsNotFound()
    {
        var ex = await Assert.ThrowsAsync<ManagementCommandException>(() =>
            Runner().RunMethodAsync(_services.BuildServiceProvider(), new FlowDebugRunMethodCommand(99, null), Deployer, CancellationToken.None));
        Assert.Equal(FlowDebugRunner.NotFoundCode, ex.Code);
    }

    private (ITemplateEngineRepository Templates, ISiteRepository Sites, IPullTraceEventsClient Puller) WireValueChange()
    {
        var templates = Substitute.For<ITemplateEngineRepository>();
        var sites = Substitute.For<ISiteRepository>();
        var puller = Substitute.For<IPullTraceEventsClient>();
        templates.GetInstanceByIdAsync(5, Arg.Any<CancellationToken>()).Returns(new Instance("Pump1") { Id = 5, SiteId = 2 });
        sites.GetSiteByIdAsync(2, Arg.Any<CancellationToken>()).Returns(new Site("Site A", "site-a") { Id = 2 });
        _router.RouteToSetAttributesAsync("site-a", Arg.Any<RouteToSetAttributesRequest>(), Arg.Any<CancellationToken>())
            .Returns(ci => new RouteToSetAttributesResponse(((RouteToSetAttributesRequest)ci[1]).CorrelationId, true, null, DateTimeOffset.UtcNow));
        _services.AddSingleton(templates);
        _services.AddSingleton(sites);
        _services.AddSingleton(puller);
        return (templates, sites, puller);
    }

    [Fact]
    public async Task RunValueChange_TracedWrite_ThenPollsUntilSpawnedRunSettles()
    {
        var (_, _, puller) = WireValueChange();
        RouteToSetAttributesRequest? write = null;
        _router.RouteToSetAttributesAsync("site-a", Arg.Do<RouteToSetAttributesRequest>(r => write = r), Arg.Any<CancellationToken>())
            .Returns(ci => new RouteToSetAttributesResponse(((RouteToSetAttributesRequest)ci[1]).CorrelationId, true, null, DateTimeOffset.UtcNow));
        var calls = 0;
        puller.PullAsync("site-a", Arg.Any<IReadOnlyCollection<Guid>>(), Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(ci =>
            {
                calls++;
                var ids = (IReadOnlyCollection<Guid>)ci[1];
                var root = ids.First();
                // Second poll onwards: a ScriptRun parented to the root exists.
                return calls >= 2
                    ? (IReadOnlyList<AuditEvent>)[ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
                        eventId: Guid.Parse("aaaaaaaa-0000-0000-0000-000000000001"), executionId: Guid.NewGuid(), parentExecutionId: root, target: "OnStatus")]
                    : Array.Empty<AuditEvent>();
            });

        var result = await Runner().RunValueChangeAsync(_services.BuildServiceProvider(),
            new FlowDebugRunValueChangeCommand(5, "Status", "Running", WaitSeconds: 30), Deployer, CancellationToken.None);

        Assert.True(result.Completed, result.Error);
        Assert.True(write!.Trace);
        Assert.Equal(result.RootExecutionId, write.ParentExecutionId);
        Assert.Equal("Pump1", write.InstanceUniqueName);
        Assert.Equal("Running", write.AttributeValues["Status"]);
        Assert.Equal(2, result.Events.Count);
        Assert.Equal(AuditKind.AttributeWrite, result.Events[0].AsRow().Kind);
        Assert.Equal("flow-debugger", ScriptTraceExtra.TryParse(result.Events[0].AsRow().Extra)!.Source);
        Assert.Equal(AuditKind.ScriptRun, result.Events[1].AsRow().Kind);
        // Settled: at least one quiet poll (2 s window) after the last new row, then stopped.
        Assert.True(calls >= 3, $"expected settle polls, got {calls}");
    }

    [Fact]
    public async Task RunValueChange_NoSpawnWithinWindow_ReportsNoTriggerCode()
    {
        var (_, _, puller) = WireValueChange();
        puller.PullAsync(default!, default!, default, default, default).ReturnsForAnyArgs(Array.Empty<AuditEvent>());

        var result = await Runner().RunValueChangeAsync(_services.BuildServiceProvider(),
            new FlowDebugRunValueChangeCommand(5, "Status", "Running", WaitSeconds: 3), Deployer, CancellationToken.None);

        Assert.False(result.Completed);
        Assert.Equal(FlowDebugRunner.NoTriggerCode, result.ErrorCode);
        Assert.Single(result.Events); // just the synthesized write row
    }

    [Fact]
    public async Task RunValueChange_OutOfScopeSite_IsRefused()
    {
        WireValueChange();
        var scoped = new AuthenticatedUser("ops", "Ops", new[] { "Deployer" }, new[] { "9" });
        await Assert.ThrowsAsync<SiteScopeViolationException>(() => Runner().RunValueChangeAsync(
            _services.BuildServiceProvider(), new FlowDebugRunValueChangeCommand(5, "Status", "x"), scoped, CancellationToken.None));
        await _router.DidNotReceiveWithAnyArgs().RouteToSetAttributesAsync(default!, default!, default);
    }

    [Fact]
    public async Task Replay_ReturnsRepositoryRows_WithRootFromTheFirstRow()
    {
        var repo = Substitute.For<IAuditLogRepository>();
        var root = Guid.NewGuid();
        var child = Guid.NewGuid();
        var rows = new List<AuditEvent>
        {
            ScadaBridgeAuditEventFactory.Create(AuditChannel.ApiInbound, AuditKind.InboundRequest, AuditStatus.Delivered, executionId: root),
            ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered, executionId: child, parentExecutionId: root),
        };
        repo.GetTraceEventsAsync(child, Arg.Any<CancellationToken>()).Returns(rows);
        _services.AddSingleton(repo);

        var result = await Runner().ReplayAsync(_services.BuildServiceProvider(), new FlowDebugReplayCommand(child), CancellationToken.None);

        Assert.True(result.Completed);
        Assert.Equal(root, result.RootExecutionId);
        Assert.Equal(rows, result.Events);

        var empty = await Runner().ReplayAsync(_services.BuildServiceProvider(), new FlowDebugReplayCommand(Guid.NewGuid()), CancellationToken.None);
        Assert.False(empty.Completed);
        Assert.Equal(FlowDebugRunner.NotFoundCode, empty.ErrorCode);
    }

    // --- Arch-review F2: a method-entry Live run is bound by the operator's site scope ---
    //
    // RunValueChange already called EnforceSiteScope on the target instance; a method
    // Live run has no single target site up front, so the scope is pushed into the
    // RouteHelper and enforced at route time (RouteTarget.ResolveSiteAsync), before any
    // site traffic. The denial comes back as SITE_SCOPE_DENIED and the runner converts it.

    /// <summary>Wires a method whose script routes to <paramref name="instanceCode"/>, resident at <paramref name="siteIdentifier"/>.</summary>
    private ISiteRepository WireRoutedMethod(int methodId, string instanceCode, string siteIdentifier)
    {
        var method = new ApiMethod("M", "x") { Id = methodId, TimeoutSeconds = 5 };
        _apiRepo.GetApiMethodByIdAsync(methodId, Arg.Any<CancellationToken>()).Returns(method);
        _locator.GetSiteIdForInstanceAsync(instanceCode, Arg.Any<CancellationToken>()).Returns(siteIdentifier);
        _router.RouteToCallAsync(siteIdentifier, Arg.Any<RouteToCallRequest>(), Arg.Any<CancellationToken>())
            .Returns(ci => new RouteToCallResponse(
                ((RouteToCallRequest)ci[1]).CorrelationId, true, 7, null, DateTimeOffset.UtcNow));
        _executor.RegisterHandler("M", ctx => ctx.Route.To(instanceCode).Call("work"));

        // The user's PermittedSiteIds are NUMERIC ids; the locator speaks site IDENTIFIERS,
        // so the runner resolves one to the other through the repository.
        var sites = Substitute.For<ISiteRepository>();
        sites.GetSiteByIdAsync(1, Arg.Any<CancellationToken>()).Returns(new Site("Site A", "site-a") { Id = 1 });
        _services.AddSingleton(sites);
        return sites;
    }

    [Fact]
    public async Task RunMethod_ScopedDeployer_RouteToOtherSite_IsUnauthorized_AndInboundRowIsWritten()
    {
        WireRoutedMethod(11, "inst", "site-b");
        var scoped = new AuthenticatedUser("ops", "Ops", new[] { "Deployer" }, new[] { "1" });

        await Assert.ThrowsAsync<SiteScopeViolationException>(() => Runner().RunMethodAsync(
            _services.BuildServiceProvider(), new FlowDebugRunMethodCommand(11, null), scoped, CancellationToken.None));

        // The site was never contacted...
        await _router.DidNotReceiveWithAnyArgs().RouteToCallAsync(default!, default!, default);
        // ...but the denied attempt IS recorded against the operator's name.
        Assert.Contains(_central, e => e.AsRow().Kind == AuditKind.InboundRequest && e.Actor == "ops");
    }

    [Fact]
    public async Task RunMethod_ScopedDeployer_RouteWithinScope_Runs()
    {
        // The same scoped user, routing to a site their numeric id DOES resolve to.
        WireRoutedMethod(12, "inst", "site-a");
        var scoped = new AuthenticatedUser("ops", "Ops", new[] { "Deployer" }, new[] { "1" });

        var result = await Runner().RunMethodAsync(
            _services.BuildServiceProvider(), new FlowDebugRunMethodCommand(12, null), scoped, CancellationToken.None);

        Assert.True(result.Completed, result.Error);
        await _router.ReceivedWithAnyArgs(1).RouteToCallAsync(default!, default!, default);
    }

    [Fact]
    public async Task RunMethod_Administrator_IsNeverScoped()
    {
        WireRoutedMethod(13, "inst", "site-b");
        var admin = new AuthenticatedUser("boss", "Boss", new[] { "Deployer", "Administrator" }, new[] { "1" });

        var result = await Runner().RunMethodAsync(
            _services.BuildServiceProvider(), new FlowDebugRunMethodCommand(13, null), admin, CancellationToken.None);

        Assert.True(result.Completed, result.Error);
        await _router.ReceivedWithAnyArgs(1).RouteToCallAsync(default!, default!, default);
    }

    [Fact]
    public async Task RunMethod_UnscopedUser_IsNeverScoped()
    {
        WireRoutedMethod(14, "inst", "site-b");

        // Deployer with an EMPTY PermittedSiteIds = global scope, as EnforceSiteScope treats it.
        var result = await Runner().RunMethodAsync(
            _services.BuildServiceProvider(), new FlowDebugRunMethodCommand(14, null), Deployer, CancellationToken.None);

        Assert.True(result.Completed, result.Error);
        await _router.ReceivedWithAnyArgs(1).RouteToCallAsync(default!, default!, default);
    }

    [Fact]
    public void RoleGate_TheTwoRunCommands_AreDeployerOnly()
    {
        // arch-review F3: a Live run executes scripts and writes attributes, so it stays Deployer-only.
        foreach (var cmd in new object[] { new FlowDebugRunMethodCommand(1, null), new FlowDebugRunValueChangeCommand(1, "a", "b") })
            Assert.Equal(new[] { "Deployer" }, ManagementActor.GetRequiredRoles(cmd));
    }

    [Fact]
    public void RoleGate_Replay_AlsoAcceptsTheOperationalAuditRoles()
    {
        // arch-review F3: Replay only reads the audit chain, and the Audit Log / Execution
        // Chain drill-ins must work for an auditor who holds no Deployer role.
        var actual = ManagementActor.GetRequiredRoles(new FlowDebugReplayCommand(Guid.NewGuid()));
        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);
        // Review follow-up: the row the RUNNER synthesizes is redacted too — the central writer
        // only redacts its own private copy, so the caller's copy would otherwise be the raw one.
        var written = Assert.Single(_central, e => e.AsRow().Kind == AuditKind.InboundRequest);
        Assert.Equal(written.DetailsJson!.ToUpperInvariant(), result.Events[0].DetailsJson);
        Assert.NotEqual(written.DetailsJson, result.Events[0].DetailsJson); // the raw row still reaches the writer
    }

    [Fact]
    public async Task RunValueChange_SynthesizedRootRow_IsRedacted_WhenARedactorIsRegistered()
    {
        var (_, _, puller) = WireValueChange();
        puller.PullAsync(default!, default!, default, default, default).ReturnsForAnyArgs(Array.Empty<AuditEvent>());
        _services.AddSingleton<IAuditRedactor>(new UpperRedactor());

        var result = await Runner().RunValueChangeAsync(_services.BuildServiceProvider(),
            new FlowDebugRunValueChangeCommand(5, "Status", "Running", WaitSeconds: 1), Deployer, CancellationToken.None);

        var root = Assert.Single(result.Events);
        var written = Assert.Single(_central);
        Assert.Equal(written.DetailsJson!.ToUpperInvariant(), root.DetailsJson);
        Assert.NotEqual(written.DetailsJson, root.DetailsJson);
    }

    [Fact]
    public async Task RunValueChange_FailedWrite_ReturnsTheRedactedRootRow()
    {
        WireValueChange();
        _router.RouteToSetAttributesAsync("site-a", Arg.Any<RouteToSetAttributesRequest>(), Arg.Any<CancellationToken>())
            .Returns(ci => new RouteToSetAttributesResponse(((RouteToSetAttributesRequest)ci[1]).CorrelationId, false, "no such attribute", DateTimeOffset.UtcNow));
        _services.AddSingleton<IAuditRedactor>(new UpperRedactor());

        var result = await Runner().RunValueChangeAsync(_services.BuildServiceProvider(),
            new FlowDebugRunValueChangeCommand(5, "Status", "Running", WaitSeconds: 1), Deployer, CancellationToken.None);

        Assert.False(result.Completed);
        Assert.Equal("WRITE_FAILED", result.ErrorCode);
        var root = Assert.Single(result.Events);
        var written = Assert.Single(_central);
        Assert.Equal(written.DetailsJson!.ToUpperInvariant(), root.DetailsJson);
        Assert.NotEqual(written.DetailsJson, root.DetailsJson);
    }

    [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() };
    }
}