diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/AuditingDbCommand.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/AuditingDbCommand.cs index 15577e51..e27656c4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/AuditingDbCommand.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/AuditingDbCommand.cs @@ -48,6 +48,14 @@ internal sealed class AuditingDbCommand : DbCommand /// private readonly Guid? _parentExecutionId; + /// + /// Flow Debugger (plan 2026-09-06 D5): optional observer invoked with the SAME + /// instance hands the writer, + /// collector-first — the run's inline trace collector. Null when untraced, which + /// leaves the emit path byte-identical. + /// + private readonly Action? _onAudit; + private readonly ILogger _logger; private DbConnection? _wrappingConnection; @@ -63,6 +71,7 @@ internal sealed class AuditingDbCommand : DbCommand /// Logger for diagnostics and warnings. /// Unique identifier for this script execution. /// Optional identifier of the parent execution (for routed calls). + /// Optional observer of every emitted (the Flow Debugger trace collector). // Parameter ordering: executionId sits immediately after the ILogger, // consistent with the other three audit-threaded ctors (ExternalSystemHelper, // DatabaseHelper, AuditingDbConnection). parentExecutionId is a trailing @@ -76,7 +85,8 @@ internal sealed class AuditingDbCommand : DbCommand string? sourceScript, ILogger logger, Guid executionId, - Guid? parentExecutionId = null) + Guid? parentExecutionId = null, + Action? onAudit = null) { _inner = inner ?? throw new ArgumentNullException(nameof(inner)); _auditWriter = auditWriter ?? throw new ArgumentNullException(nameof(auditWriter)); @@ -87,6 +97,7 @@ internal sealed class AuditingDbCommand : DbCommand _logger = logger ?? throw new ArgumentNullException(nameof(logger)); _executionId = executionId; _parentExecutionId = parentExecutionId; + _onAudit = onAudit; } // -- Forwarded surface ------------------------------------------------ @@ -410,6 +421,8 @@ internal sealed class AuditingDbCommand : DbCommand try { + // Flow Debugger: collector first (same instance), then the writer. + _onAudit?.Invoke(evt); var writeTask = _auditWriter.WriteAsync(evt, CancellationToken.None); if (!writeTask.IsCompleted) { diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/AuditingDbConnection.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/AuditingDbConnection.cs index 5a9527cd..69f622c0 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/AuditingDbConnection.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/AuditingDbConnection.cs @@ -2,6 +2,7 @@ using System.Data; using System.Data.Common; using Microsoft.Extensions.Logging; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using AuditEvent = ZB.MOM.WW.Audit.AuditEvent; namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts; @@ -46,6 +47,13 @@ internal sealed class AuditingDbConnection : DbConnection /// private readonly Guid? _parentExecutionId; + /// + /// Flow Debugger (plan 2026-09-06 D5): optional observer handed the SAME + /// each gives the writer — + /// the run's inline trace collector. Null when untraced. + /// + private readonly Action? _onAudit; + private readonly ILogger _logger; /// @@ -60,6 +68,7 @@ internal sealed class AuditingDbConnection : DbConnection /// Logger for diagnostics and warnings. /// Unique identifier for this script execution. /// Optional identifier of the parent execution (for routed calls). + /// Optional observer of every emitted (the Flow Debugger trace collector). // Parameter ordering: executionId sits immediately after the ILogger, // consistent with the other three audit-threaded ctors (ExternalSystemHelper, // DatabaseHelper, AuditingDbCommand). parentExecutionId is a trailing @@ -73,7 +82,8 @@ internal sealed class AuditingDbConnection : DbConnection string? sourceScript, ILogger logger, Guid executionId, - Guid? parentExecutionId = null) + Guid? parentExecutionId = null, + Action? onAudit = null) { _inner = inner ?? throw new ArgumentNullException(nameof(inner)); _auditWriter = auditWriter ?? throw new ArgumentNullException(nameof(auditWriter)); @@ -84,6 +94,7 @@ internal sealed class AuditingDbConnection : DbConnection _logger = logger ?? throw new ArgumentNullException(nameof(logger)); _executionId = executionId; _parentExecutionId = parentExecutionId; + _onAudit = onAudit; } /// @@ -155,7 +166,9 @@ internal sealed class AuditingDbConnection : DbConnection _executionId, // ParentExecutionId: the spawning execution's id, // threaded alongside _executionId. Null for non-routed runs. - _parentExecutionId); + _parentExecutionId, + // Flow Debugger: the DbWrite rows ride the inline collector. + _onAudit); } /// diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/FlowTraceCollector.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/FlowTraceCollector.cs index 8dc855bc..46a92f1b 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/FlowTraceCollector.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/FlowTraceCollector.cs @@ -9,6 +9,11 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts; /// the collector keeps the first N, sets , and the audit rows written /// through IAuditWriter remain complete — Replay is never truncated, only the inline copy. /// An untraced run holds null rather than an inert collector; callers null-check. +/// Outbound-calls addendum (plan 2026-09-06 D5): the ApiOutbound / DbOutbound / +/// Notification rows a run emits ride the same sink, so closes it +/// when the launcher takes the reply snapshot — a lifecycle row the S&F forwarder writes +/// minutes later, or an un-awaited call finishing after the run returned, must not mutate a +/// list the reply has already been built from. is a no-op once completed. /// public sealed class FlowTraceCollector { @@ -22,6 +27,12 @@ public sealed class FlowTraceCollector /// True once an was dropped because the cap was reached. public bool Truncated { get; private set; } + /// + /// True once ran — the run's reply snapshot has been taken and + /// later calls are ignored (they never count as ). + /// + public bool Completed { get; private set; } + /// Creates a collector with the given inline cap (must be positive). public FlowTraceCollector(int maxEvents = DefaultMaxEvents) { @@ -29,17 +40,35 @@ public sealed class FlowTraceCollector _max = maxEvents; } - /// Appends an event, or marks the collector truncated once the cap is reached. + /// + /// Appends an event, or marks the collector truncated once the cap is reached. A no-op + /// after . + /// public void Add(AuditEvent evt) { ArgumentNullException.ThrowIfNull(evt); lock (_gate) { + if (Completed) return; if (_events.Count >= _max) { Truncated = true; return; } _events.Add(evt); } } + /// + /// Closes the collector and returns the final snapshot — the launcher calls this exactly + /// where it builds ScriptCallResult.TraceEvents. Idempotent: a second call returns + /// the same contents. + /// + public IReadOnlyList Complete() + { + lock (_gate) + { + Completed = true; + return _events.ToArray(); + } + } + /// Current event count. public int Count { get { lock (_gate) return _events.Count; } } diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRunLauncher.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRunLauncher.cs index 44cc50c5..f5a2b297 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRunLauncher.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRunLauncher.cs @@ -417,7 +417,9 @@ internal static class ScriptRunLauncher var state = await compiledScript.RunAsync(globals, cts.Token); // Flow Debugger: the ScriptRun row goes into the collector BEFORE the reply so the - // inline list the caller receives includes the run itself. + // inline list the caller receives includes the run itself. Complete() (not Snapshot) + // closes the sink at the same instant — outbound rows finishing after this point + // (plan 2026-09-06 D5) go to the audit writer only. EmitScriptRun(trace, collector, auditWriter, logger, scriptName, instanceName, siteId, sourceNode, context.ExecutionId, parentExecutionId, startedUtc, runStopwatch.ElapsedMilliseconds, parameters, state.ReturnValue, error: null, triggerExtra); @@ -427,7 +429,7 @@ internal static class ScriptRunLauncher { replyTo.Tell(new ScriptCallResult( correlationId, true, state.ReturnValue, null, - context.ExecutionId, collector?.Snapshot())); + context.ExecutionId, collector?.Complete())); } // Operational `script` event — execution completed successfully. Same gating as @@ -462,7 +464,7 @@ internal static class ScriptRunLauncher if (!replyTo.IsNobody()) { replyTo.Tell(new ScriptCallResult( - correlationId, false, null, errorMsg, context?.ExecutionId, collector?.Snapshot())); + correlationId, false, null, errorMsg, context?.ExecutionId, collector?.Complete())); } completionTarget.Tell(new ScriptActor.ScriptExecutionCompleted(scriptName, false, errorMsg)); @@ -487,7 +489,7 @@ internal static class ScriptRunLauncher if (!replyTo.IsNobody()) { replyTo.Tell(new ScriptCallResult( - correlationId, false, null, errorMsg, context?.ExecutionId, collector?.Snapshot())); + correlationId, false, null, errorMsg, context?.ExecutionId, collector?.Complete())); } completionTarget.Tell(new ScriptActor.ScriptExecutionCompleted(scriptName, false, errorMsg)); diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRuntimeContext.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRuntimeContext.cs index 8f64a7bb..04a773b8 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRuntimeContext.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRuntimeContext.cs @@ -179,6 +179,16 @@ public class ScriptRuntimeContext /// private readonly FlowTraceCollector? _traceCollector; + /// + /// Outbound-calls addendum (plan 2026-09-06 D5): as a delegate the + /// ExternalSystem / Database / Notify helpers hand every + /// ApiOutbound / DbOutbound / Notification row to — the same + /// instance the audit writer / telemetry forwarder receives, so + /// a Live run's inline reply carries them. Null when untraced so the helpers' emit paths + /// stay byte-identical to the pre-feature behaviour. + /// + private readonly Action? _traceSink; + /// /// Initializes a new script runtime context with all dependencies for script /// execution, external system calls, database access, and notification delivery. @@ -287,8 +297,17 @@ public class ScriptRuntimeContext _scriptTimeoutToken = scriptTimeoutToken; _isTraced = isTraced; _traceCollector = traceCollector; + _traceSink = traceCollector is null ? null : Collect; } + /// + /// Flow Debugger: the single feeder of the inline collector. Every row this run's + /// reply should carry — the ScriptTrace rows from and the + /// outbound rows the helpers emit — passes through here; a no-op when untraced or once + /// the launcher has d the run. + /// + private void Collect(AuditEvent evt) => _traceCollector?.Add(evt); + /// Flow Debugger (design 2026-09-05 §5.2): whether this run emits ScriptTrace rows. public bool IsTraced => _isTraced; @@ -385,7 +404,7 @@ public class ScriptRuntimeContext return; } - _traceCollector?.Add(evt); + Collect(evt); if (_auditWriter is null) return; try { @@ -818,7 +837,9 @@ public class ScriptRuntimeContext // SourceNode-stamping: the local node name (node-a/node-b), // threaded so the cached-call telemetry construction sites can stamp // it onto SiteCallOperational.SourceNode. - _sourceNode); + _sourceNode, + // Flow Debugger: the outbound rows ride the inline collector (null when untraced). + traceSink: _traceSink); /// /// Provides access to database operations. @@ -846,7 +867,9 @@ public class ScriptRuntimeContext // SourceNode-stamping: the local node name (node-a/node-b), // threaded so Database.CachedWrite's CachedSubmit telemetry can // stamp it onto SiteCallOperational.SourceNode. - _sourceNode); + _sourceNode, + // Flow Debugger: the outbound rows ride the inline collector (null when untraced). + traceSink: _traceSink); /// /// Provides access to the Notification Outbox API. @@ -869,7 +892,9 @@ public class ScriptRuntimeContext _parentExecutionId, // SourceNode-stamping: the local node name (node-a/node-b), // threaded so NotifyTarget.Send can stamp it onto NotificationSubmit. - _sourceNode); + _sourceNode, + // Flow Debugger: the NotifySend row rides the inline collector (null when untraced). + traceSink: _traceSink); /// /// Read-only access to the instance's CURRENT alarm conditions (MES alarm-status API @@ -1065,6 +1090,13 @@ public class ScriptRuntimeContext /// private readonly string? _sourceNode; + /// + /// Flow Debugger (plan 2026-09-06 D5): the run's inline collector. Each emit + /// site hands it the SAME it gives the writer / + /// forwarder, collector first. Null when untraced — the production default. + /// + private readonly Action? _traceSink; + /// /// Initializes a new external system helper for script API calls. /// @@ -1078,6 +1110,7 @@ public class ScriptRuntimeContext /// Optional forwarder for cached call telemetry. /// Optional identifier of the parent execution (for routed calls). /// Optional cluster node identifier (node-a/node-b) for audit stamping. + /// Flow Debugger inline collector — receives the same each emit hands to the writer/forwarder; null when untraced. // Internal constructor for tests living in ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests // (via InternalsVisibleTo). Production sites resolve the helper through // ScriptRuntimeContext.ExternalSystem. @@ -1100,7 +1133,8 @@ public class ScriptRuntimeContext string? sourceScript = null, ICachedCallTelemetryForwarder? cachedForwarder = null, Guid? parentExecutionId = null, - string? sourceNode = null) + string? sourceNode = null, + Action? traceSink = null) { _client = client; _instanceName = instanceName; @@ -1112,6 +1146,7 @@ public class ScriptRuntimeContext _cachedForwarder = cachedForwarder; _parentExecutionId = parentExecutionId; _sourceNode = sourceNode; + _traceSink = traceSink; } /// @@ -1329,6 +1364,8 @@ public class ScriptRuntimeContext try { + // Flow Debugger: collector first, then the forwarder. + _traceSink?.Invoke(telemetry.Audit); await _cachedForwarder.ForwardAsync(telemetry, cancellationToken) .ConfigureAwait(false); } @@ -1452,6 +1489,9 @@ public class ScriptRuntimeContext { try { + // Flow Debugger: the immediate-completion rows are emitted inline while + // the run is still open, so they legitimately ride the collector. + _traceSink?.Invoke(attempted.Audit); await _cachedForwarder.ForwardAsync(attempted, cancellationToken) .ConfigureAwait(false); } @@ -1519,6 +1559,7 @@ public class ScriptRuntimeContext try { + _traceSink?.Invoke(resolve.Audit); await _cachedForwarder.ForwardAsync(resolve, cancellationToken) .ConfigureAwait(false); } @@ -1569,6 +1610,8 @@ public class ScriptRuntimeContext try { + // Flow Debugger: collector first (same instance), then the writer. + _traceSink?.Invoke(evt); // Fire-and-forget so we never block the script on the audit // writer; the writer itself is responsible for fast, durable // enqueue (site SQLite hot-path). We DO observe failures via @@ -1758,6 +1801,14 @@ public class ScriptRuntimeContext /// private readonly string? _sourceNode; + /// + /// Flow Debugger (plan 2026-09-06 D5): the run's inline collector, handed the + /// same each emit gives the writer / forwarder and + /// threaded into for the sync DbWrite + /// rows. Null when untraced. + /// + private readonly Action? _traceSink; + /// /// Initializes a new database helper for script database access. /// @@ -1771,6 +1822,7 @@ public class ScriptRuntimeContext /// Optional forwarder for cached call telemetry. /// Optional identifier of the parent execution (for routed calls). /// Optional cluster node identifier (node-a/node-b) for audit stamping. + /// Flow Debugger inline collector — receives the same each emit hands to the writer/forwarder; null when untraced. // Parameter ordering: executionId sits immediately after the // ILogger — see the note on ExternalSystemHelper's ctor for why the // post-logger slot is the one consistent position across all four @@ -1785,7 +1837,8 @@ public class ScriptRuntimeContext string? sourceScript = null, ICachedCallTelemetryForwarder? cachedForwarder = null, Guid? parentExecutionId = null, - string? sourceNode = null) + string? sourceNode = null, + Action? traceSink = null) { _gateway = gateway; _instanceName = instanceName; @@ -1797,6 +1850,7 @@ public class ScriptRuntimeContext _cachedForwarder = cachedForwarder; _parentExecutionId = parentExecutionId; _sourceNode = sourceNode; + _traceSink = traceSink; } /// @@ -1837,7 +1891,9 @@ public class ScriptRuntimeContext executionId: _executionId, // (ParentExecutionId): the spawning execution's // id, threaded alongside _executionId. Null for non-routed runs. - parentExecutionId: _parentExecutionId); + parentExecutionId: _parentExecutionId, + // Flow Debugger: the sync DbWrite rows ride the inline collector. + onAudit: _traceSink); } /// @@ -1959,6 +2015,8 @@ public class ScriptRuntimeContext { try { + // Flow Debugger: emitted inline while the run is open — rides the collector. + _traceSink?.Invoke(attempted.Audit); await _cachedForwarder.ForwardAsync(attempted, cancellationToken) .ConfigureAwait(false); } @@ -1980,6 +2038,7 @@ public class ScriptRuntimeContext { try { + _traceSink?.Invoke(resolve.Audit); await _cachedForwarder.ForwardAsync(resolve, cancellationToken) .ConfigureAwait(false); } @@ -2107,6 +2166,8 @@ public class ScriptRuntimeContext try { + // Flow Debugger: collector first, then the forwarder. + _traceSink?.Invoke(telemetry.Audit); await _cachedForwarder.ForwardAsync(telemetry, cancellationToken) .ConfigureAwait(false); } @@ -2169,6 +2230,12 @@ public class ScriptRuntimeContext /// private readonly string? _sourceNode; + /// + /// Flow Debugger (plan 2026-09-06 D5): the run's inline collector, forwarded to + /// each . Null when untraced. + /// + private readonly Action? _traceSink; + /// /// Initializes a new notification helper for script notification delivery. /// @@ -2183,6 +2250,7 @@ public class ScriptRuntimeContext /// Optional writer for audit log entries. /// Optional identifier of the parent execution (for routed calls). /// Optional cluster node identifier (node-a/node-b) for audit stamping. + /// Flow Debugger inline collector — receives the same Send hands to the writer; null when untraced. // Parameter ordering: executionId sits immediately after the ILogger, // consistent with the other audit-threaded ctors. parentExecutionId is // a trailing optional param. @@ -2197,7 +2265,8 @@ public class ScriptRuntimeContext Guid executionId, IAuditWriter? auditWriter = null, Guid? parentExecutionId = null, - string? sourceNode = null) + string? sourceNode = null, + Action? traceSink = null) { _storeAndForward = storeAndForward; _siteCommunicationActor = siteCommunicationActor; @@ -2210,6 +2279,7 @@ public class ScriptRuntimeContext _auditWriter = auditWriter; _parentExecutionId = parentExecutionId; _sourceNode = sourceNode; + _traceSink = traceSink; } /// @@ -2232,7 +2302,9 @@ public class ScriptRuntimeContext _parentExecutionId, // SourceNode-stamping: the local node name, stamped // onto NotificationSubmit.SourceNode in Send(). - _sourceNode); + _sourceNode, + // Flow Debugger: the NotifySend row rides the inline collector. + traceSink: _traceSink); } /// @@ -2339,6 +2411,13 @@ public class ScriptRuntimeContext /// private readonly string? _sourceNode; + /// + /// Flow Debugger (plan 2026-09-06 D5): the run's inline collector — handed the + /// same NotifySend the writer receives. Null when + /// untraced. + /// + private readonly Action? _traceSink; + /// /// Initializes a new notification target for a specific notification list. /// @@ -2352,6 +2431,7 @@ public class ScriptRuntimeContext /// Optional writer for audit log entries. /// Optional identifier of the parent execution (for routed calls). /// Optional cluster node identifier (node-a/node-b) for audit stamping. + /// Flow Debugger inline collector; null when untraced. internal NotifyTarget( string listName, StoreAndForwardService? storeAndForward, @@ -2362,7 +2442,8 @@ public class ScriptRuntimeContext Guid executionId, IAuditWriter? auditWriter = null, Guid? parentExecutionId = null, - string? sourceNode = null) + string? sourceNode = null, + Action? traceSink = null) { _listName = listName; _storeAndForward = storeAndForward; @@ -2374,6 +2455,7 @@ public class ScriptRuntimeContext _auditWriter = auditWriter; _parentExecutionId = parentExecutionId; _sourceNode = sourceNode; + _traceSink = traceSink; } /// @@ -2558,6 +2640,8 @@ public class ScriptRuntimeContext try { + // Flow Debugger: collector first (same instance), then the writer. + _traceSink?.Invoke(evt); // Fire-and-forget (mirrors ExternalSystemHelper.EmitCallAudit) // so the script is never blocked on the audit writer; we observe // failures via ContinueWith so a thrown writer is logged rather diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Scripts/ScriptRuntimeContextTraceEmissionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Scripts/ScriptRuntimeContextTraceEmissionTests.cs index 5ceab46f..409b99c8 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Scripts/ScriptRuntimeContextTraceEmissionTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Scripts/ScriptRuntimeContextTraceEmissionTests.cs @@ -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 ScriptTrace rows for attribute writes and waits (into the inline collector /// AND the audit writer), stamps Trace 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 ApiOutbound / +/// DbOutbound / Notification rows a run emits ride the same collector — the +/// SAME instance the writer / telemetry forwarder receives — and +/// nothing lands after the collector is d. /// public class ScriptRuntimeContextTraceEmissionTests : TestKit { @@ -31,18 +43,117 @@ public class ScriptRuntimeContextTraceEmissionTests : TestKit } } + private sealed class CapturingForwarder : ICachedCallTelemetryForwarder + { + public List Telemetry { get; } = new(); + public Task ForwardAsync(CachedCallTelemetry telemetry, CancellationToken ct = default) + { + Telemetry.Add(telemetry); + return Task.CompletedTask; + } + } + + /// A real S&F service over a temp LocalDb file — Notify.Send needs the + /// concrete , there is no interface to stub. + 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 CreateAsync() + { + var localDb = TestLocalDb.CreateTemp("TraceEmissionNotify"); + var storage = new StoreAndForwardStorage(localDb.Db, NullLogger.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.Instance)); + } + + public void Dispose() + { + _localDb.Dispose(); + TestLocalDb.DeleteFiles(_localDb.Path); + } + } + + private static Mock SyncClient(ExternalCallResult result) + { + var client = new Mock(); + client + .Setup(c => c.CallAsync("ERP", "GetOrder", It.IsAny?>(), It.IsAny())) + .ReturnsAsync(result); + return client; + } + + private static Mock CachedClient(ExternalCallResult result) + { + var client = new Mock(); + client + .Setup(c => c.CachedCallAsync( + "ERP", "GetOrder", + It.IsAny?>(), + InstanceName, + It.IsAny(), + It.IsAny(), + It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(result); + return client; + } + + /// 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. + private static Mock 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(); + gateway + .Setup(g => g.GetConnectionAsync("machineData", It.IsAny())) + .ReturnsAsync(live); + return gateway; + } + + private static AuditRowProjection.AuditRowValues Row(AuditEvent evt) => evt.AsRow(); + private static SharedScriptLibrary NewLibrary() => new(new ScriptCompilationService(NullLogger.Instance), NullLogger.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(); + gateway + .Setup(g => g.CachedWriteAsync( + "machineData", "UPDATE t SET name = 'x'", + It.IsAny?>(), InstanceName, + It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny(), It.IsAny())) + .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?>(), It.IsAny())) + .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); + } }