From eca1434160e63dd992f580cb9f17c0b95f47cdaa Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 5 Sep 2026 23:43:33 -0400 Subject: [PATCH] feat(site-runtime): traced runs emit ScriptRun rows, return ExecutionId + trace events; traced value-change runs inherit the parent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Flow Debugger (design 2026-09-05 §5.1/§5.4): ScriptRunLauncher emits one ScriptRun row per traced run on every exit path (OccurredAtUtc = run start), replies with the run's ExecutionId (always) and the collector snapshot (traced only). ScriptActor forwards ScriptCallRequest.Trace and, for a value-change trigger fired by a traced stamped change, parents the run to the writer with trigger detail in Extra; every other trigger stays a root. Claude-Session: https://claude.ai/code/session_01GSLkkHKz9JrAtmJjmQK3cP --- .../Actors/ScriptActor.cs | 41 +++- .../Scripts/ScriptRunLauncher.cs | 112 +++++++++-- .../Actors/ScriptActorTraceTests.cs | 183 ++++++++++++++++++ 3 files changed, 319 insertions(+), 17 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptActorTraceTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/ScriptActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/ScriptActor.cs index 195d4176..d4f25ba3 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/ScriptActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/ScriptActor.cs @@ -4,6 +4,8 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using ZB.MOM.WW.ScadaBridge.Commons.Messages.ScriptExecution; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; +using ZB.MOM.WW.ScadaBridge.Commons.Types; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging; using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening; using ZB.MOM.WW.ScadaBridge.HealthMonitoring; using ZB.MOM.WW.ScadaBridge.SiteEventLogging; @@ -272,7 +274,9 @@ public class ScriptActor : ReceiveActor, IWithTimers // (tag-change / timer) runs and nested Script.Call invocations. SpawnExecution( request.Parameters, request.CurrentCallDepth, Sender, request.CorrelationId, - request.ParentExecutionId); + request.ParentExecutionId, + // Flow Debugger (design §5.2): the trace flag rides beside the parent id. + request.Trace); } /// @@ -289,7 +293,23 @@ public class ScriptActor : ReceiveActor, IWithTimers { if (valueTrigger.AttributeName == changed.AttributeName) { - TrySpawnExecution(null); + // Flow Debugger (design §5.2/§5.3): a change stamped by a traced write parents + // the spawned run to the writing execution and carries the trace flag; every + // other value-change run stays a root by design (CLAUDE.md, alarm-leg note). + if (changed.Trace && changed.SourceExecutionId is { } sourceExecutionId) + { + TrySpawnExecution(null, sourceExecutionId, trace: true, new ScriptTraceExtra + { + TriggerKind = "valuechange", + TriggerAttribute = changed.AttributeName, + TriggerValue = AttributeValueCodec.Encode(changed.Value), + TriggerSource = "value-change", + }); + } + else + { + TrySpawnExecution(null); + } } } else if (_triggerConfig is ConditionalTriggerConfig conditional) @@ -496,7 +516,11 @@ public class ScriptActor : ReceiveActor, IWithTimers /// /// Attempts to spawn a script execution, respecting MinTimeBetweenRuns. /// - private void TrySpawnExecution(IReadOnlyDictionary? parameters) + private void TrySpawnExecution( + IReadOnlyDictionary? parameters, + Guid? parentExecutionId = null, + bool trace = false, + ScriptTraceExtra? triggerExtra = null) { if (_compiledScript == null) return; @@ -513,7 +537,7 @@ public class ScriptActor : ReceiveActor, IWithTimers } _lastExecutionTime = DateTimeOffset.UtcNow; - SpawnExecution(parameters, 0, ActorRefs.NoSender!, Guid.NewGuid().ToString()); + SpawnExecution(parameters, 0, ActorRefs.NoSender!, Guid.NewGuid().ToString(), parentExecutionId, trace, triggerExtra); } /// @@ -544,7 +568,9 @@ public class ScriptActor : ReceiveActor, IWithTimers int callDepth, IActorRef replyTo, string correlationId, - Guid? parentExecutionId = null) + Guid? parentExecutionId = null, + bool trace = false, + ScriptTraceExtra? triggerExtra = null) { if (callDepth == 0 && _runsInFlight >= _options.MaxConcurrentRunsPerScript) { @@ -590,7 +616,10 @@ public class ScriptActor : ReceiveActor, IWithTimers _executionTimeoutSeconds, // Scheduler seam (#18): thread this actor's scheduler override down so // launched script bodies share the same pool (null = process-wide shared). - _scheduler); + _scheduler, + // Flow Debugger (design §5.2): trace flag + value-change trigger detail. + trace, + triggerExtra); } catch (Exception ex) { diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRunLauncher.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRunLauncher.cs index 95395553..44cc50c5 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRunLauncher.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRunLauncher.cs @@ -7,7 +7,10 @@ using Microsoft.Extensions.Options; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; using ZB.MOM.WW.ScadaBridge.Commons.Messages.ScriptExecution; +using System.Text.Json; 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 ZB.MOM.WW.ScadaBridge.Commons.Types.Scripts; using ZB.MOM.WW.ScadaBridge.HealthMonitoring; @@ -74,6 +77,8 @@ internal static class ScriptRunLauncher /// ExecutionId of the spawning execution for audit correlation; null for root runs. /// Per-script execution timeout in seconds. Null or non-positive falls back to the global value. /// Script-execution scheduler seam (#18); null selects the process-wide shared instance. + /// Flow Debugger trace flag (design 2026-09-05 §5.2): when true the run emits ScriptTrace rows and returns them inline on the . + /// Flow Debugger: trigger detail for a traced value-change run (stamped into the ScriptRun row's Extra); null for calls. public static void LaunchScript( string scriptName, string instanceName, @@ -94,7 +99,9 @@ internal static class ScriptRunLauncher IServiceProvider? serviceProvider, Guid? parentExecutionId, int? executionTimeoutSeconds, - ScriptExecutionScheduler? scheduler) + ScriptExecutionScheduler? scheduler, + bool trace = false, + ScriptTraceExtra? triggerExtra = null) { var timeout = ResolveTimeout(executionTimeoutSeconds, options); var executionScheduler = scheduler ?? ScriptExecutionScheduler.Shared(options); @@ -111,7 +118,7 @@ internal static class ScriptRunLauncher instanceActor, sharedScriptLibrary, options, replyTo, correlationId, completionTarget, siteCommunicationActor, logger, scope, runId, healthCollector, serviceProvider, parentExecutionId, timeout, - executionScheduler, cts), + executionScheduler, cts, trace, triggerExtra), CancellationToken.None, TaskCreationOptions.DenyChildAttach, executionScheduler) .Unwrap(); } @@ -231,9 +238,19 @@ internal static class ScriptRunLauncher Guid? parentExecutionId, TimeSpan timeout, ScriptExecutionScheduler executionScheduler, - CancellationTokenSource cts) + CancellationTokenSource cts, + bool trace = false, + ScriptTraceExtra? triggerExtra = null) { IServiceScope? serviceScope = null; + // Flow Debugger (design §5.4): the inline sink exists only for a traced run; the + // ScriptRun row itself is emitted below on every exit path of a traced run. + var collector = trace ? new FlowTraceCollector() : null; + var startedUtc = DateTime.UtcNow; + ScriptRuntimeContext? context = null; + IAuditWriter? auditWriter = null; + var siteId = string.Empty; + string? sourceNode = null; // ISiteEventLogger is a singleton; resolve from the root provider so // it is available to the catch blocks regardless of scope state. var siteEventLogger = serviceProvider?.GetService(); @@ -281,11 +298,11 @@ internal static class ScriptRunLauncher // Notification Outbox: the S&F engine is a singleton; the site identity // provider supplies the site id stamped on enqueued notifications. StoreAndForwardService? storeAndForward = null; - var siteId = string.Empty; // The writer is a singleton (FallbackAuditWriter // composes the SQLite hot-path + drop-oldest ring); null in tests / hosts // that haven't called AddAuditLog, which the helper handles as a no-op. - IAuditWriter? auditWriter = null; + // (siteId / auditWriter / sourceNode are declared above the try so the + // traced ScriptRun emission in the catch blocks can reach them.) // Site-local tracking store // backing Tracking.Status(id). Singleton; null in tests / hosts // that haven't wired the store, which the helper handles by @@ -304,7 +321,6 @@ internal static class ScriptRunLauncher // provider, in which case NotificationSubmit.SourceNode and // SiteCallOperational.SourceNode stay null and central // persists the rows with SourceNode NULL. - string? sourceNode = null; if (serviceProvider != null) { @@ -320,7 +336,7 @@ internal static class ScriptRunLauncher sourceNode = serviceScope.ServiceProvider.GetService()?.NodeName; } - var context = new ScriptRuntimeContext( + context = new ScriptRuntimeContext( instanceActor, sharedScriptLibrary, callDepth, @@ -371,7 +387,10 @@ internal static class ScriptRunLauncher // execution-timeout token so Attributes.WaitAsync's Ask is // bounded by the script's own ExecutionTimeoutSeconds — a // shorter script deadline wins over the wait's own timeout. - scriptTimeoutToken: cts.Token); + scriptTimeoutToken: cts.Token, + // Flow Debugger (design §5.2/§5.4): the trace flag and the inline sink. + isTraced: trace, + traceCollector: collector); var globals = new ScriptGlobals { @@ -397,10 +416,18 @@ 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. + EmitScriptRun(trace, collector, auditWriter, logger, scriptName, instanceName, siteId, sourceNode, + context.ExecutionId, parentExecutionId, startedUtc, runStopwatch.ElapsedMilliseconds, + parameters, state.ReturnValue, error: null, triggerExtra); + // Send result to requester if this was an Ask-based call if (!replyTo.IsNobody()) { - replyTo.Tell(new ScriptCallResult(correlationId, true, state.ReturnValue, null)); + replyTo.Tell(new ScriptCallResult( + correlationId, true, state.ReturnValue, null, + context.ExecutionId, collector?.Snapshot())); } // Operational `script` event — execution completed successfully. Same gating as @@ -428,9 +455,14 @@ internal static class ScriptRunLauncher _ = siteEventLogger?.LogEventAsync( "script", "Error", instanceName, $"ScriptActor:{scriptName}", errorMsg); + EmitScriptRun(trace, collector, auditWriter, logger, scriptName, instanceName, siteId, sourceNode, + context?.ExecutionId, parentExecutionId, startedUtc, runStopwatch.ElapsedMilliseconds, + parameters, null, errorMsg, triggerExtra); + if (!replyTo.IsNobody()) { - replyTo.Tell(new ScriptCallResult(correlationId, false, null, errorMsg)); + replyTo.Tell(new ScriptCallResult( + correlationId, false, null, errorMsg, context?.ExecutionId, collector?.Snapshot())); } completionTarget.Tell(new ScriptActor.ScriptExecutionCompleted(scriptName, false, errorMsg)); @@ -448,9 +480,14 @@ internal static class ScriptRunLauncher _ = siteEventLogger?.LogEventAsync( "script", "Error", instanceName, $"ScriptActor:{scriptName}", errorMsg, ex.ToString()); + EmitScriptRun(trace, collector, auditWriter, logger, scriptName, instanceName, siteId, sourceNode, + context?.ExecutionId, parentExecutionId, startedUtc, runStopwatch.ElapsedMilliseconds, + parameters, null, errorMsg, triggerExtra); + if (!replyTo.IsNobody()) { - replyTo.Tell(new ScriptCallResult(correlationId, false, null, errorMsg)); + replyTo.Tell(new ScriptCallResult( + correlationId, false, null, errorMsg, context?.ExecutionId, collector?.Snapshot())); } completionTarget.Tell(new ScriptActor.ScriptExecutionCompleted(scriptName, false, errorMsg)); @@ -466,6 +503,59 @@ internal static class ScriptRunLauncher } } + /// + /// Flow Debugger (design 2026-09-05 §5.1): emits the ScriptTrace.ScriptRun row for a + /// traced run — collector first (so the inline reply carries it), then the best-effort + /// audit writer. OccurredAtUtc is the run's START so the row sorts before the + /// writes/waits the run made. No-op when untraced; never throws. + /// + private static void EmitScriptRun( + bool trace, FlowTraceCollector? collector, IAuditWriter? auditWriter, ILogger logger, + string scriptName, string instanceName, string siteId, string? sourceNode, + Guid? executionId, Guid? parentExecutionId, DateTime startedUtc, long elapsedMs, + IReadOnlyDictionary? parameters, object? returnValue, string? error, + ScriptTraceExtra? triggerExtra) + { + if (!trace) return; + try + { + var sourceScript = $"ScriptActor:{scriptName}"; + var evt = ScadaBridgeAuditEventFactory.Create( + AuditChannel.ScriptTrace, AuditKind.ScriptRun, + error is null ? AuditStatus.Delivered : AuditStatus.Failed, + occurredAtUtc: DateTime.SpecifyKind(startedUtc, DateTimeKind.Utc), + actor: sourceScript, target: scriptName, sourceNode: sourceNode, + executionId: executionId, parentExecutionId: parentExecutionId, + sourceSiteId: string.IsNullOrEmpty(siteId) ? null : siteId, + sourceInstanceId: instanceName, sourceScript: sourceScript, + durationMs: (int)Math.Min(elapsedMs, int.MaxValue), + errorMessage: error, + requestSummary: TraceJson(parameters), + responseSummary: TraceJson(returnValue), + extra: (triggerExtra ?? new ScriptTraceExtra { TriggerKind = "call" }).ToJson()); + collector?.Add(evt); + if (auditWriter is null) return; + var t = auditWriter.WriteAsync(evt, CancellationToken.None); + if (!t.IsCompletedSuccessfully) + { + _ = t.ContinueWith( + x => logger.LogWarning(x.Exception, "Flow trace ScriptRun write failed for {Script}", scriptName), + TaskContinuationOptions.OnlyOnFaulted); + } + } + catch (Exception ex) + { + logger.LogWarning(ex, "Flow trace ScriptRun emission failed for {Script}", scriptName); + } + } + + private static string? TraceJson(object? value) + { + if (value is null) return null; + try { return JsonSerializer.Serialize(value, ScriptTraceExtra.JsonOptions); } + catch (Exception) { return null; } + } + private static async Task RunAlarmScriptAsync( string alarmName, string instanceName, diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptActorTraceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptActorTraceTests.cs new file mode 100644 index 00000000..c40f9214 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptActorTraceTests.cs @@ -0,0 +1,183 @@ +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Microsoft.CodeAnalysis.CSharp.Scripting; +using Microsoft.CodeAnalysis.Scripting; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.ScriptExecution; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening; +using IAuditWriter = ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services.IAuditWriter; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts; + +namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors; + +/// +/// Flow Debugger design 2026-09-05 §5.1/§5.4: a traced makes +/// the launched run emit a ScriptRun row and return its ExecutionId plus the +/// inline trace events on the ; a traced, stamped +/// parents the value-change run to the writer. Untraced +/// paths are pinned unchanged. +/// +public class ScriptActorTraceTests : TestKit, IDisposable +{ + private sealed class CapturingAuditWriter : IAuditWriter + { + public List Events { get; } = new(); + public Task WriteAsync(AuditEvent evt, CancellationToken ct = default) + { + lock (Events) Events.Add(evt); + return Task.CompletedTask; + } + public AuditEvent? WaitForKind(string action, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + lock (Events) + { + var hit = Events.FirstOrDefault(e => e.Action == action); + if (hit is not null) return hit; + } + Thread.Sleep(25); + } + return null; + } + } + + private readonly SharedScriptLibrary _sharedLibrary; + private readonly SiteRuntimeOptions _options; + private readonly ScriptExecutionScheduler _scheduler; + private readonly CapturingAuditWriter _writer = new(); + private readonly IServiceProvider _services; + + public ScriptActorTraceTests() + { + _sharedLibrary = new SharedScriptLibrary( + new ScriptCompilationService(NullLogger.Instance), + NullLogger.Instance); + _options = new SiteRuntimeOptions { MaxScriptCallDepth = 10, ScriptExecutionTimeoutSeconds = 30 }; + _scheduler = new ScriptExecutionScheduler(_options.ScriptExecutionThreadCount); + var sc = new ServiceCollection(); + sc.AddSingleton(_writer); + _services = sc.BuildServiceProvider(); + } + + void IDisposable.Dispose() + { + Shutdown(); + _scheduler.Dispose(); + } + + private static Script Compile(string code) + { + var script = CSharpScript.Create(code, + ScriptOptions.Default.WithReferences(typeof(object).Assembly, typeof(Enumerable).Assembly) + .WithImports("System", "System.Collections.Generic", "System.Linq", "System.Threading.Tasks"), + typeof(ScriptGlobals)); + script.Compile(); + return script; + } + + private IActorRef CreateActor(string name, string code, string? triggerJson = null) + { + var config = new ResolvedScript { CanonicalName = name, Code = code, TriggerType = triggerJson is null ? null : "valuechange", TriggerConfiguration = triggerJson }; + return ActorOf(Props.Create(() => new ScriptActor( + name, "Pump1", CreateTestProbe().Ref, Compile(code), config, _sharedLibrary, _options, + NullLogger.Instance, null, null, null, _services, _scheduler))); + } + + [Fact] + public void Call_WithTrace_ReplyCarriesExecutionId_AndAScriptRunEvent() + { + var actor = CreateActor("Add", "(int)Parameters[\"x\"] + 1"); + + actor.Tell(new ScriptCallRequest("Add", new Dictionary { ["x"] = 41 }, 0, "cid", Trace: true)); + + var result = ExpectMsg(TimeSpan.FromSeconds(10)); + Assert.True(result.Success, result.ErrorMessage); + Assert.Equal(42, result.ReturnValue); + Assert.NotNull(result.ExecutionId); + var evt = Assert.Single(result.TraceEvents!); + var row = evt.AsRow(); + Assert.Equal("ScriptTrace.ScriptRun", evt.Action); + Assert.Equal("Delivered", row.Status.ToString()); + Assert.Equal(result.ExecutionId, row.ExecutionId); + Assert.Equal("Add", evt.Target); + Assert.Equal("Pump1", row.SourceInstanceId); + Assert.Contains("\"x\":41", row.RequestSummary); + Assert.Equal("42", row.ResponseSummary); + Assert.Equal("call", ScriptTraceExtra.TryParse(row.Extra)!.TriggerKind); + + // The same row went through the audit writer. + Assert.NotNull(_writer.WaitForKind("ScriptTrace.ScriptRun", TimeSpan.FromSeconds(5))); + } + + [Fact] + public void Call_WithTrace_FailedRun_ReturnsFailedScriptRunEvent() + { + var actor = CreateActor("Boom", "throw new InvalidOperationException(\"kaboom\");"); + + actor.Tell(new ScriptCallRequest("Boom", null, 0, "cid", Trace: true)); + + var result = ExpectMsg(TimeSpan.FromSeconds(10)); + Assert.False(result.Success); + Assert.NotNull(result.ExecutionId); + var row = Assert.Single(result.TraceEvents!).AsRow(); + Assert.Equal("Failed", row.Status.ToString()); + Assert.Contains("kaboom", row.ErrorMessage); + } + + [Fact] + public void Call_WithoutTrace_ReplyHasNoTraceEvents_ButStillCarriesExecutionId() + { + var actor = CreateActor("Answer", "42"); + + actor.Tell(new ScriptCallRequest("Answer", null, 0, "cid")); + + var result = ExpectMsg(TimeSpan.FromSeconds(10)); + Assert.True(result.Success); + Assert.NotNull(result.ExecutionId); + Assert.Null(result.TraceEvents); + Thread.Sleep(200); + Assert.Empty(_writer.Events); + } + + [Fact] + public void ValueChangeTrigger_TracedChange_RunParentsToSourceExecutionId() + { + var actor = CreateActor("OnStatus", "1", "{\"attribute\":\"Status\"}"); + var writer = Guid.NewGuid(); + var now = DateTimeOffset.UtcNow; + + actor.Tell(new AttributeValueChanged("Pump1", "Status", "Status", "Running", "Good", now, SourceExecutionId: writer, Trace: true)); + + var evt = _writer.WaitForKind("ScriptTrace.ScriptRun", TimeSpan.FromSeconds(10)); + Assert.NotNull(evt); + var row = evt!.AsRow(); + Assert.Equal(writer, row.ParentExecutionId); + Assert.NotEqual(writer, row.ExecutionId); + var extra = ScriptTraceExtra.TryParse(row.Extra)!; + Assert.Equal("valuechange", extra.TriggerKind); + Assert.Equal("Status", extra.TriggerAttribute); + Assert.Equal("Running", extra.TriggerValue); + Assert.Equal("value-change", extra.TriggerSource); + } + + [Fact] + public void ValueChangeTrigger_UntracedChange_RunStaysRoot_AndEmitsNothing() + { + var actor = CreateActor("OnStatus", "1", "{\"attribute\":\"Status\"}"); + + // Stamped but NOT traced (an ordinary script write) — and a plain DCL change. + actor.Tell(new AttributeValueChanged("Pump1", "Status", "Status", "Running", "Good", DateTimeOffset.UtcNow, SourceExecutionId: Guid.NewGuid())); + actor.Tell(new AttributeValueChanged("Pump1", "Status", "Status", "Stopped", "Good", DateTimeOffset.UtcNow)); + + Assert.Null(_writer.WaitForKind("ScriptTrace.ScriptRun", TimeSpan.FromSeconds(1))); + Assert.Empty(_writer.Events); + } +}