feat(site-runtime): outbound API/DB/notification audit rows ride the flow-trace collector

ScriptRuntimeContext caches a trace sink (null on an untraced run, so the production path is byte-identical) and threads it into the ExternalSystem / Database / Notify helpers and AuditingDbConnection→AuditingDbCommand (onAudit). All eight outbound emit sites hand the SAME AuditEvent instance to the sink first, then to the writer/forwarder. FlowTraceCollector gains Completed + Complete(): ScriptRunLauncher closes the collector where it builds the reply snapshot, so un-awaited calls finishing after the run returned never mutate a list the reply was built from. Cap (500) unchanged. Plan 2026-09-06 D5. Claude-Session: https://claude.ai/code/session_01NcMG6xXdsJbmD2wYAgkK29
2026-09-06 01:51:19 -04:00
parent 865c4be651
commit f394f58405
6 changed files with 464 additions and 19 deletions
@@ -1,13 +1,21 @@
 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;

@@ -16,6 +24,10 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Scripts;
 /// 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
 {
@@ -31,18 +43,117 @@ public class ScriptRuntimeContextTraceEmissionTests : TestKit
        }
    }

+    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)
+        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);

@@ -203,4 +314,197 @@ public class ScriptRuntimeContextTraceEmissionTests : TestKit
        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);
+    }
 }