using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging.Abstractions;
using Moq;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Instance;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.ScriptExecution;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
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 IAuditWriter = ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services.IAuditWriter;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
using ZB.MOM.WW.ScadaBridge.StoreAndForward;
using ZB.MOM.WW.ScadaBridge.TestSupport;

namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Scripts;

/// <summary>
/// Flow Debugger design 2026-09-05 §5.1/§5.2: a traced <see cref="ScriptRuntimeContext"/>
/// emits <c>ScriptTrace</c> rows for attribute writes and waits (into the inline collector
/// AND the audit writer), stamps <c>Trace</c> on the commands it sends, and inherits the flag
/// into nested calls. An untraced context is byte-for-byte the pre-feature behaviour.
/// Outbound-calls addendum (plan 2026-09-06 Task 3): the <c>ApiOutbound</c> /
/// <c>DbOutbound</c> / <c>Notification</c> rows a run emits ride the same collector — the
/// SAME <see cref="AuditEvent"/> instance the writer / telemetry forwarder receives — and
/// nothing lands after the collector is <see cref="FlowTraceCollector.Complete"/>d.
/// </summary>
public class ScriptRuntimeContextTraceEmissionTests : TestKit
{
    private const string InstanceName = "Plant.Pump42";

    private sealed class CapturingAuditWriter : IAuditWriter
    {
        public List<AuditEvent> Events { get; } = new();
        public Task WriteAsync(AuditEvent evt, CancellationToken ct = default)
        {
            lock (Events) Events.Add(evt);
            return Task.CompletedTask;
        }
    }

    private sealed class CapturingForwarder : ICachedCallTelemetryForwarder
    {
        public List<CachedCallTelemetry> Telemetry { get; } = new();
        public Task ForwardAsync(CachedCallTelemetry telemetry, CancellationToken ct = default)
        {
            Telemetry.Add(telemetry);
            return Task.CompletedTask;
        }
    }

    /// <summary>A real S&amp;F service over a temp LocalDb file — <c>Notify.Send</c> needs the
    /// concrete <see cref="StoreAndForwardService"/>, there is no interface to stub.</summary>
    private sealed class TempStoreAndForward : IDisposable
    {
        private readonly TestLocalDb _localDb;
        public StoreAndForwardService Service { get; }

        private TempStoreAndForward(TestLocalDb localDb, StoreAndForwardService service)
        {
            _localDb = localDb;
            Service = service;
        }

        public static async Task<TempStoreAndForward> CreateAsync()
        {
            var localDb = TestLocalDb.CreateTemp("TraceEmissionNotify");
            var storage = new StoreAndForwardStorage(localDb.Db, NullLogger<StoreAndForwardStorage>.Instance);
            await storage.InitializeAsync();
            var options = new StoreAndForwardOptions
            {
                DefaultRetryInterval = TimeSpan.Zero,
                DefaultMaxRetries = 3,
                RetryTimerInterval = TimeSpan.FromMinutes(10),
            };
            return new TempStoreAndForward(localDb,
                new StoreAndForwardService(storage, options, NullLogger<StoreAndForwardService>.Instance));
        }

        public void Dispose()
        {
            _localDb.Dispose();
            TestLocalDb.DeleteFiles(_localDb.Path);
        }
    }

    private static Mock<IExternalSystemClient> SyncClient(ExternalCallResult result)
    {
        var client = new Mock<IExternalSystemClient>();
        client
            .Setup(c => c.CallAsync("ERP", "GetOrder", It.IsAny<IReadOnlyDictionary<string, object?>?>(), It.IsAny<CancellationToken>()))
            .ReturnsAsync(result);
        return client;
    }

    private static Mock<IExternalSystemClient> CachedClient(ExternalCallResult result)
    {
        var client = new Mock<IExternalSystemClient>();
        client
            .Setup(c => c.CachedCallAsync(
                "ERP", "GetOrder",
                It.IsAny<IReadOnlyDictionary<string, object?>?>(),
                InstanceName,
                It.IsAny<CancellationToken>(),
                It.IsAny<TrackedOperationId?>(),
                It.IsAny<Guid?>(), It.IsAny<string?>(), It.IsAny<Guid?>()))
            .ReturnsAsync(result);
        return client;
    }

    /// <summary>In-memory SQLite handed out by a stub gateway; the keep-alive root must outlive
    /// the auditing wrapper so the database survives the wrapper's dispose.</summary>
    private static Mock<IDatabaseGateway> SqliteGateway(out SqliteConnection keepAlive)
    {
        var connStr = $"Data Source=trace-{Guid.NewGuid():N};Mode=Memory;Cache=Shared";
        keepAlive = new SqliteConnection(connStr);
        keepAlive.Open();
        using (var seed = keepAlive.CreateCommand())
        {
            seed.CommandText = "CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT NOT NULL);";
            seed.ExecuteNonQuery();
        }
        var live = new SqliteConnection(connStr);
        live.Open();
        var gateway = new Mock<IDatabaseGateway>();
        gateway
            .Setup(g => g.GetConnectionAsync("machineData", It.IsAny<CancellationToken>()))
            .ReturnsAsync(live);
        return gateway;
    }

    private static AuditRowProjection.AuditRowValues Row(AuditEvent evt) => evt.AsRow();

    private static SharedScriptLibrary NewLibrary() =>
        new(new ScriptCompilationService(NullLogger<ScriptCompilationService>.Instance),
            NullLogger<SharedScriptLibrary>.Instance);

    private static ScriptRuntimeContext CreateContext(
        IActorRef instanceActor, IAuditWriter? writer, bool traced, FlowTraceCollector? collector,
        Guid? executionId = null, Guid? parentExecutionId = null,
        IExternalSystemClient? externalSystemClient = null,
        IDatabaseGateway? databaseGateway = null,
        ICachedCallTelemetryForwarder? cachedForwarder = null,
        StoreAndForwardService? storeAndForward = null)
        => new(
            instanceActor, NewLibrary(),
            currentCallDepth: 0, maxCallDepth: 10, askTimeout: TimeSpan.FromSeconds(5),
            instanceName: InstanceName, logger: NullLogger.Instance,
            externalSystemClient: externalSystemClient, databaseGateway: databaseGateway,
            storeAndForward: storeAndForward,
            siteId: "site-a", sourceScript: "ScriptActor:A", auditWriter: writer,
            cachedForwarder: cachedForwarder,
            executionId: executionId, parentExecutionId: parentExecutionId, sourceNode: "node-a",
            isTraced: traced, traceCollector: collector);

    [Fact]
    public async Task SetAttribute_WhenTraced_EmitsAttributeWriteRow_AndPassesTraceOnCommand()
    {
        var probe = CreateTestProbe();
        var writer = new CapturingAuditWriter();
        var collector = new FlowTraceCollector();
        var exec = Guid.NewGuid();
        var parent = Guid.NewGuid();
        var ctx = CreateContext(probe.Ref, writer, traced: true, collector, exec, parent);

        var call = ctx.SetAttribute("Setpoint", "42");
        var cmd = probe.ExpectMsg<SetStaticAttributeCommand>(TimeSpan.FromSeconds(5));
        Assert.True(cmd.Trace);
        Assert.Equal(exec, cmd.SourceExecutionId);
        probe.Reply(new SetStaticAttributeResponse(cmd.CorrelationId, InstanceName, "Setpoint", true, null, DateTimeOffset.UtcNow));
        await call;

        var evt = Assert.Single(writer.Events);
        var row = evt.AsRow();
        Assert.Equal("ScriptTrace", evt.Category);
        Assert.Equal("ScriptTrace.AttributeWrite", evt.Action);
        Assert.Equal(exec, row.ExecutionId);
        Assert.Equal(parent, row.ParentExecutionId);
        Assert.Equal("site-a", row.SourceSiteId);
        Assert.Equal("node-a", evt.SourceNode);
        Assert.Equal("Setpoint", evt.Target);
        Assert.Equal(AuditOutcome.Success, evt.Outcome);
        var extra = ScriptTraceExtra.TryParse(row.Extra);
        var write = Assert.Single(extra!.Writes!);
        Assert.Equal(("Setpoint", "42", true), (write.Attribute, write.Written, write.Success));
        Assert.Contains("\"value\":\"42\"", row.RequestSummary);

        Assert.Same(evt, Assert.Single(collector.Snapshot()));
    }

    [Fact]
    public async Task SetAttribute_WhenNotTraced_EmitsNothing_AndCommandTraceFalse()
    {
        var probe = CreateTestProbe();
        var writer = new CapturingAuditWriter();
        var ctx = CreateContext(probe.Ref, writer, traced: false, collector: null);

        var call = ctx.SetAttribute("Setpoint", "42");
        var cmd = probe.ExpectMsg<SetStaticAttributeCommand>(TimeSpan.FromSeconds(5));
        Assert.False(cmd.Trace);
        probe.Reply(new SetStaticAttributeResponse(cmd.CorrelationId, InstanceName, "Setpoint", true, null, DateTimeOffset.UtcNow));
        await call;

        Assert.Empty(writer.Events);
        Assert.False(ctx.IsTraced);
    }

    /// <summary>
    /// Arch-review F1: the collector applies the host redactor, so the inline copy the reply
    /// carries is the FILTERED record — while the writer still receives the raw event and
    /// redacts it itself (redaction is idempotent, so the stored row is identical).
    /// </summary>
    [Fact]
    public async Task SetAttribute_WhenTraced_InlineCopyIsRedacted_WriterStillSeesTheRawEvent()
    {
        var probe = CreateTestProbe();
        var writer = new CapturingAuditWriter();
        var collector = new FlowTraceCollector(redactor: new UpperRedactor());
        var ctx = CreateContext(probe.Ref, writer, traced: true, collector);

        var call = ctx.SetAttribute("Setpoint", "42");
        var cmd = probe.ExpectMsg<SetStaticAttributeCommand>(TimeSpan.FromSeconds(5));
        probe.Reply(new SetStaticAttributeResponse(cmd.CorrelationId, InstanceName, "Setpoint", true, null, DateTimeOffset.UtcNow));
        await call;

        var written = Assert.Single(writer.Events);
        var collected = Assert.Single(collector.Complete());
        Assert.NotSame(written, collected);
        Assert.Contains("\"channel\":\"ScriptTrace\"", written.DetailsJson);
        Assert.NotEqual(written.DetailsJson, collected.DetailsJson);
        Assert.Equal(written.DetailsJson!.ToUpperInvariant(), collected.DetailsJson);
    }

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

    [Fact]
    public async Task SetAttribute_WhenTraced_FailedWrite_EmitsFailedRow_ThenThrows()
    {
        var probe = CreateTestProbe();
        var writer = new CapturingAuditWriter();
        var ctx = CreateContext(probe.Ref, writer, traced: true, new FlowTraceCollector());

        var call = ctx.SetAttribute("Setpoint", "42");
        var cmd = probe.ExpectMsg<SetStaticAttributeCommand>(TimeSpan.FromSeconds(5));
        probe.Reply(new SetStaticAttributeResponse(cmd.CorrelationId, InstanceName, "Setpoint", false, "device rejected", DateTimeOffset.UtcNow));

        await Assert.ThrowsAsync<InvalidOperationException>(() => call);
        var row = Assert.Single(writer.Events).AsRow();
        Assert.Equal("Failed", row.Status.ToString());
        Assert.Equal("device rejected", row.ErrorMessage);
    }

    [Fact]
    public async Task WaitInternal_WhenTraced_EmitsAttributeWait_WithOutcomeSatisfied()
    {
        var probe = CreateTestProbe();
        var writer = new CapturingAuditWriter();
        var collector = new FlowTraceCollector();
        var ctx = CreateContext(probe.Ref, writer, traced: true, collector);

        var wait = ctx.WaitAttributeFull("Processed", "true", null, TimeSpan.FromSeconds(3));
        var req = probe.ExpectMsg<WaitForAttributeRequest>(TimeSpan.FromSeconds(5));
        probe.Reply(new WaitForAttributeResponse(req.CorrelationId, Matched: true, true, "Good", TimedOut: false));
        var result = await wait;

        Assert.True(result.Matched);
        var evt = Assert.Single(writer.Events);
        Assert.Equal("ScriptTrace.AttributeWait", evt.Action);
        Assert.Equal("Processed", evt.Target);
        var extra = ScriptTraceExtra.TryParse(evt.AsRow().Extra)!;
        Assert.Equal("satisfied", extra.WaitOutcome);
        Assert.Equal("Processed", extra.WaitAttribute);
        Assert.Equal("true", extra.WaitTarget);
        Assert.Equal(3000, extra.WaitTimeoutMs);
        Assert.Equal("true", extra.ObservedValue, ignoreCase: true);
        Assert.Single(collector.Snapshot());
    }

    [Fact]
    public async Task WaitInternal_WhenTraced_Timeout_EmitsFailedRow_WithTimeoutOutcome()
    {
        var probe = CreateTestProbe();
        var writer = new CapturingAuditWriter();
        var ctx = CreateContext(probe.Ref, writer, traced: true, new FlowTraceCollector());

        var wait = ctx.WaitAttribute("Processed", "true", null, TimeSpan.FromSeconds(1));
        var req = probe.ExpectMsg<WaitForAttributeRequest>(TimeSpan.FromSeconds(5));
        probe.Reply(new WaitForAttributeResponse(req.CorrelationId, Matched: false, null, null, TimedOut: true));

        Assert.False(await wait);
        var row = Assert.Single(writer.Events).AsRow();
        Assert.Equal("Failed", row.Status.ToString());
        Assert.Contains("timed out", row.ErrorMessage);
        Assert.Equal("timeout", ScriptTraceExtra.TryParse(row.Extra)!.WaitOutcome);
    }

    [Fact]
    public async Task WaitInternal_WhenNotTraced_EmitsNothing()
    {
        var probe = CreateTestProbe();
        var writer = new CapturingAuditWriter();
        var ctx = CreateContext(probe.Ref, writer, traced: false, collector: null);

        var wait = ctx.WaitAttribute("Processed", "true", null, TimeSpan.FromSeconds(1));
        var req = probe.ExpectMsg<WaitForAttributeRequest>(TimeSpan.FromSeconds(5));
        probe.Reply(new WaitForAttributeResponse(req.CorrelationId, Matched: true, true, "Good", TimedOut: false));
        await wait;

        Assert.Empty(writer.Events);
    }

    [Fact]
    public async Task CallScript_WhenTraced_PropagatesTraceOnScriptCallRequest()
    {
        var probe = CreateTestProbe();
        var ctx = CreateContext(probe.Ref, null, traced: true, new FlowTraceCollector());

        var call = ctx.CallScript("B");
        var req = probe.ExpectMsg<ScriptCallRequest>(TimeSpan.FromSeconds(5));
        Assert.True(req.Trace);
        probe.Reply(new ScriptCallResult(req.CorrelationId, true, null, null));
        await call;

        // The inline shared-script child context inherits the flag too.
        Assert.True(ctx.CreateChildContextForSharedScript(1).IsTraced);
        Assert.False(CreateContext(probe.Ref, null, traced: false, null).CreateChildContextForSharedScript(1).IsTraced);
    }

    [Fact]
    public async Task CallScript_WhenNotTraced_RequestTraceIsFalse()
    {
        var probe = CreateTestProbe();
        var ctx = CreateContext(probe.Ref, null, traced: false, null);

        var call = ctx.CallScript("B");
        var req = probe.ExpectMsg<ScriptCallRequest>(TimeSpan.FromSeconds(5));
        Assert.False(req.Trace);
        probe.Reply(new ScriptCallResult(req.CorrelationId, true, null, null));
        await call;
    }

    // ---- Outbound rows ride the collector (plan 2026-09-06 Task 3) -------------------------

    [Fact]
    public async Task ExternalSystemCall_WhenTraced_CollectsTheSameApiCallRow_TheWriterReceives()
    {
        var writer = new CapturingAuditWriter();
        var collector = new FlowTraceCollector();
        var exec = Guid.NewGuid();
        var ctx = CreateContext(CreateTestProbe().Ref, writer, traced: true, collector, exec,
            externalSystemClient: SyncClient(new ExternalCallResult(true, "{}", null)).Object);

        var result = await ctx.ExternalSystem.Call("ERP", "GetOrder");

        Assert.True(result.Success);
        var written = Assert.Single(writer.Events);
        var collected = Assert.Single(collector.Snapshot());
        Assert.Same(written, collected);
        Assert.Equal(AuditChannel.ApiOutbound, Row(collected).Channel);
        Assert.Equal(AuditKind.ApiCall, Row(collected).Kind);
        Assert.Equal(exec, Row(collected).ExecutionId);
    }

    [Fact]
    public async Task ExternalSystemCachedCall_Buffered_CollectsCachedSubmitOnly_SameInstanceAsTelemetry()
    {
        var forwarder = new CapturingForwarder();
        var collector = new FlowTraceCollector();
        var ctx = CreateContext(CreateTestProbe().Ref, new CapturingAuditWriter(), traced: true, collector,
            externalSystemClient: CachedClient(new ExternalCallResult(true, null, null, WasBuffered: true)).Object,
            cachedForwarder: forwarder);

        var trackedId = await ctx.ExternalSystem.CachedCall("ERP", "GetOrder");

        var packet = Assert.Single(forwarder.Telemetry);
        var collected = Assert.Single(collector.Snapshot());
        Assert.Same(packet.Audit, collected);
        Assert.Equal(AuditKind.CachedSubmit, Row(collected).Kind);
        Assert.Equal(trackedId.Value, Row(collected).CorrelationId);
    }

    [Fact]
    public async Task ExternalSystemCachedCall_ImmediateCompletion_CollectsSubmitAttemptedResolve_WhileRunIsOpen()
    {
        var forwarder = new CapturingForwarder();
        var collector = new FlowTraceCollector();
        var ctx = CreateContext(CreateTestProbe().Ref, new CapturingAuditWriter(), traced: true, collector,
            externalSystemClient: CachedClient(new ExternalCallResult(true, "{\"ok\":true}", null, WasBuffered: false)).Object,
            cachedForwarder: forwarder);

        await ctx.ExternalSystem.CachedCall("ERP", "GetOrder");

        Assert.Equal(3, forwarder.Telemetry.Count);
        var snap = collector.Snapshot();
        Assert.Equal(3, snap.Count);
        for (var i = 0; i < 3; i++) Assert.Same(forwarder.Telemetry[i].Audit, snap[i]);
        Assert.Equal(
            new[] { AuditKind.CachedSubmit, AuditKind.ApiCallCached, AuditKind.CachedResolve },
            snap.Select(e => Row(e).Kind));
    }

    [Fact]
    public async Task OutboundRows_AfterCollectorCompleted_DoNotTouchTheCollector_ButStillReachTelemetry()
    {
        // The S&F forwarder's later Forwarded/Attempted/Resolve rows run after the launcher
        // has snapshotted TraceEvents. Complete() is that snapshot; anything emitted after it
        // (here: a fast inline resolve on a context the run already returned from) is ignored
        // by the collector while the writer/forwarder still receive every row.
        var forwarder = new CapturingForwarder();
        var writer = new CapturingAuditWriter();
        var collector = new FlowTraceCollector();
        var ctx = CreateContext(CreateTestProbe().Ref, writer, traced: true, collector,
            externalSystemClient: CachedClient(new ExternalCallResult(true, null, null, WasBuffered: false)).Object,
            cachedForwarder: forwarder);

        var first = await ctx.ExternalSystem.CachedCall("ERP", "GetOrder");
        var snapshot = collector.Complete();

        Assert.True(collector.Completed);
        Assert.Equal(3, snapshot.Count);
        Assert.Equal(first.Value, Row(snapshot[0]).CorrelationId);

        await ctx.ExternalSystem.CachedCall("ERP", "GetOrder");

        Assert.Equal(6, forwarder.Telemetry.Count);
        Assert.Equal(3, collector.Count);
        Assert.False(collector.Truncated);
        Assert.Equal(snapshot.Select(e => e.EventId), collector.Snapshot().Select(e => e.EventId));
    }

    [Fact]
    public async Task DatabaseExecute_WhenTraced_CollectsTheSameDbWriteRow_TheWriterReceives()
    {
        var gateway = SqliteGateway(out var keepAlive);
        using var _ = keepAlive;
        var writer = new CapturingAuditWriter();
        var collector = new FlowTraceCollector();
        var ctx = CreateContext(CreateTestProbe().Ref, writer, traced: true, collector,
            databaseGateway: gateway.Object);

        await using (var conn = await ctx.Database.Connection("machineData"))
        await using (var cmd = conn.CreateCommand())
        {
            cmd.CommandText = "INSERT INTO t (id, name) VALUES (1, 'alpha')";
            Assert.Equal(1, await cmd.ExecuteNonQueryAsync());
        }

        var written = Assert.Single(writer.Events);
        var collected = Assert.Single(collector.Snapshot());
        Assert.Same(written, collected);
        Assert.Equal(AuditChannel.DbOutbound, Row(collected).Channel);
        Assert.Equal(AuditKind.DbWrite, Row(collected).Kind);
        Assert.Contains("\"rowsAffected\":1", Row(collected).Extra);
    }

    [Fact]
    public async Task DatabaseCachedWrite_Buffered_CollectsCachedSubmit_SameInstanceAsTelemetry()
    {
        var gateway = new Mock<IDatabaseGateway>();
        gateway
            .Setup(g => g.CachedWriteAsync(
                "machineData", "UPDATE t SET name = 'x'",
                It.IsAny<IReadOnlyDictionary<string, object?>?>(), InstanceName,
                It.IsAny<CancellationToken>(), It.IsAny<TrackedOperationId?>(),
                It.IsAny<Guid?>(), It.IsAny<string?>(), It.IsAny<Guid?>()))
            .ReturnsAsync(new ExternalCallResult(true, null, null, WasBuffered: true));
        var forwarder = new CapturingForwarder();
        var collector = new FlowTraceCollector();
        var ctx = CreateContext(CreateTestProbe().Ref, new CapturingAuditWriter(), traced: true, collector,
            databaseGateway: gateway.Object, cachedForwarder: forwarder);

        var trackedId = await ctx.Database.CachedWrite("machineData", "UPDATE t SET name = 'x'");

        var packet = Assert.Single(forwarder.Telemetry);
        var collected = Assert.Single(collector.Snapshot());
        Assert.Same(packet.Audit, collected);
        Assert.Equal(AuditChannel.DbOutbound, Row(collected).Channel);
        Assert.Equal(AuditKind.CachedSubmit, Row(collected).Kind);
        Assert.Equal(trackedId.Value, Row(collected).CorrelationId);
    }

    [Fact]
    public async Task NotifySend_WhenTraced_CollectsTheSameNotifySendRow_TheWriterReceives()
    {
        using var saf = await TempStoreAndForward.CreateAsync();
        var writer = new CapturingAuditWriter();
        var collector = new FlowTraceCollector();
        var ctx = CreateContext(CreateTestProbe().Ref, writer, traced: true, collector,
            storeAndForward: saf.Service);

        var notificationId = await ctx.Notify.To("Planners").Send("Recipe ready", "R-17 staged");

        var written = Assert.Single(writer.Events);
        var collected = Assert.Single(collector.Snapshot());
        Assert.Same(written, collected);
        Assert.Equal(AuditChannel.Notification, Row(collected).Channel);
        Assert.Equal(AuditKind.NotifySend, Row(collected).Kind);
        Assert.Equal("Planners", collected.Target);
        Assert.Equal(Guid.Parse(notificationId), Row(collected).CorrelationId);
    }

    [Fact]
    public async Task OutboundRows_WhenNoCollector_WriterAndTelemetryStillReceiveEveryRow()
    {
        using var saf = await TempStoreAndForward.CreateAsync();
        var gateway = SqliteGateway(out var keepAlive);
        using var _ = keepAlive;
        var writer = new CapturingAuditWriter();
        var forwarder = new CapturingForwarder();
        var client = CachedClient(new ExternalCallResult(true, null, null, WasBuffered: true));
        client
            .Setup(c => c.CallAsync("ERP", "GetOrder", It.IsAny<IReadOnlyDictionary<string, object?>?>(), It.IsAny<CancellationToken>()))
            .ReturnsAsync(new ExternalCallResult(true, "{}", null));
        var ctx = CreateContext(CreateTestProbe().Ref, writer, traced: false, collector: null,
            externalSystemClient: client.Object, databaseGateway: gateway.Object,
            cachedForwarder: forwarder, storeAndForward: saf.Service);

        await ctx.ExternalSystem.Call("ERP", "GetOrder");
        await ctx.ExternalSystem.CachedCall("ERP", "GetOrder");
        await using (var conn = await ctx.Database.Connection("machineData"))
        await using (var cmd = conn.CreateCommand())
        {
            cmd.CommandText = "INSERT INTO t (id, name) VALUES (1, 'alpha')";
            await cmd.ExecuteNonQueryAsync();
        }
        await ctx.Notify.To("Planners").Send("s", "b");

        Assert.False(ctx.IsTraced);
        Assert.Equal(
            new[] { AuditKind.ApiCall, AuditKind.DbWrite, AuditKind.NotifySend },
            writer.Events.Select(e => Row(e).Kind));
        Assert.Equal(AuditKind.CachedSubmit, Row(Assert.Single(forwarder.Telemetry).Audit).Kind);
    }

    [Fact]
    public async Task ThrowingTraceSink_NeverFailsTheCall_AndTheWriterStillGetsTheRow()
    {
        var writer = new CapturingAuditWriter();
        var helper = new ScriptRuntimeContext.ExternalSystemHelper(
            SyncClient(new ExternalCallResult(true, "{}", null)).Object, InstanceName, NullLogger.Instance, Guid.NewGuid(),
            auditWriter: writer, siteId: "site-a", sourceScript: "S",
            traceSink: _ => throw new InvalidOperationException("observer broke"));

        var result = await helper.Call("ERP", "GetOrder");

        Assert.True(result.Success);
        var written = Assert.Single(writer.Events);
        Assert.Equal(AuditKind.ApiCall, Row(written).Kind);
    }
}