From 93489cd7a2dca4e289d829a52178e6d6282523f1 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 5 Sep 2026 23:43:32 -0400 Subject: [PATCH] feat(site-runtime): ScriptRuntimeContext emits ScriptTrace AttributeWrite/AttributeWait rows and inherits the Trace flag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Flow Debugger (design 2026-09-05 §5.1/§5.2): the context is the single funnel for SetAttribute / WriteBatchAndWait / WaitInternal, so it emits the write/wait rows (collector first, then the best-effort IAuditWriter whose redactor applies the payload cap), stamps Trace on SetStaticAttributeCommand and ScriptCallRequest, and hands the flag + sink to the inline CallShared child context. Untraced contexts emit nothing (pinned). Claude-Session: https://claude.ai/code/session_01GSLkkHKz9JrAtmJjmQK3cP --- .../Scripts/ScriptRuntimeContext.cs | 196 ++++++++++++++++- .../ScriptRuntimeContextTraceEmissionTests.cs | 206 ++++++++++++++++++ 2 files changed, 393 insertions(+), 9 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Scripts/ScriptRuntimeContextTraceEmissionTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRuntimeContext.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRuntimeContext.cs index fa1b6a9c..8f64a7bb 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRuntimeContext.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRuntimeContext.cs @@ -12,6 +12,7 @@ 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.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 AuditEvent = ZB.MOM.WW.Audit.AuditEvent; @@ -164,6 +165,20 @@ public class ScriptRuntimeContext /// private readonly Guid? _parentExecutionId; + /// + /// Flow Debugger (design 2026-09-05 §5.2): true when this run is traced — the + /// Trace flag rode in beside ParentExecutionId on the spawning message + /// (or ApiMethod.TraceAllCalls set it at the root). Gates every + /// ScriptTrace row this context emits and is inherited by nested runs. + /// + private readonly bool _isTraced; + + /// + /// Flow Debugger (design §5.4): the per-run inline sink the launcher attaches to + /// ScriptCallResult.TraceEvents. Null when untraced. + /// + private readonly FlowTraceCollector? _traceCollector; + /// /// Initializes a new script runtime context with all dependencies for script /// execution, external system calls, database access, and notification delivery. @@ -213,6 +228,8 @@ public class ScriptRuntimeContext /// for callers / tests that do not /// thread one — those waits are bounded only by their own timeout. /// + /// Flow Debugger trace flag — see design 2026-09-05 §5.2. Defaults to untraced. + /// Flow Debugger per-run inline trace sink (design §5.4); null when untraced. public ScriptRuntimeContext( IActorRef instanceActor, SharedScriptLibrary sharedScriptLibrary, @@ -234,7 +251,9 @@ public class ScriptRuntimeContext Guid? parentExecutionId = null, string? sourceNode = null, ISiteEventLogger? siteEventLogger = null, - CancellationToken scriptTimeoutToken = default) + CancellationToken scriptTimeoutToken = default, + bool isTraced = false, + FlowTraceCollector? traceCollector = null) { _instanceActor = instanceActor; _sharedScriptLibrary = sharedScriptLibrary; @@ -266,8 +285,13 @@ public class ScriptRuntimeContext // WaitForAttribute (spec §4.3): default(CancellationToken) == None when // not threaded in — the WaitAsync Ask is then bounded only by its own timeout. _scriptTimeoutToken = scriptTimeoutToken; + _isTraced = isTraced; + _traceCollector = traceCollector; } + /// Flow Debugger (design 2026-09-05 §5.2): whether this run emits ScriptTrace rows. + public bool IsTraced => _isTraced; + /// /// This run's own per-execution id. Exposed so a /// nested Scripts.CallShared can record it as the spawned shared @@ -322,7 +346,69 @@ public class ScriptRuntimeContext // WaitForAttribute (spec §4.3): an inline shared-script call shares the // parent run's execution-timeout token so a WaitAsync inside the shared // script is bounded by the SAME script deadline. - scriptTimeoutToken: _scriptTimeoutToken); + scriptTimeoutToken: _scriptTimeoutToken, + // Flow Debugger (design §5.2): a traced root traces its whole tree — the inline + // shared script shares the same flag AND the same inline sink. + isTraced: _isTraced, + traceCollector: _traceCollector); + } + + /// + /// Flow Debugger (design 2026-09-05 §5.1): emits one ScriptTrace audit row for a + /// traced run — into the inline collector first (so a routed call can return it) and + /// then through the best-effort . No-op when untraced, never + /// throws into the script, and hands raw payloads to the writer whose redactor applies + /// the configured size cap (§9) exactly as the other channels do. + /// + private void EmitTrace( + AuditKind kind, AuditStatus status, string target, DateTime startedUtc, int durationMs, + string? input, string? output, string? error, ScriptTraceExtra? extra) + { + if (!_isTraced) return; + AuditEvent evt; + try + { + evt = ScadaBridgeAuditEventFactory.Create( + AuditChannel.ScriptTrace, kind, status, + occurredAtUtc: DateTime.SpecifyKind(startedUtc, DateTimeKind.Utc), + actor: _sourceScript, target: target, sourceNode: _sourceNode, + executionId: _executionId, parentExecutionId: _parentExecutionId, + sourceSiteId: string.IsNullOrEmpty(_siteId) ? null : _siteId, + sourceInstanceId: _instanceName, sourceScript: _sourceScript, + durationMs: durationMs, errorMessage: error, + requestSummary: input, responseSummary: output, + extra: extra?.ToJson()); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Flow trace event build failed for {Kind}", kind); + return; + } + + _traceCollector?.Add(evt); + if (_auditWriter is null) return; + try + { + var t = _auditWriter.WriteAsync(evt, CancellationToken.None); + if (!t.IsCompletedSuccessfully) + { + _ = t.ContinueWith( + x => _logger.LogWarning(x.Exception, "Flow trace write failed for {Kind}", kind), + TaskContinuationOptions.OnlyOnFaulted); + } + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Flow trace write failed for {Kind}", kind); + } + } + + /// Serialises a small trace payload; null on failure (best-effort, like the audit helpers). + private static string? TraceJson(object? value) + { + if (value is null) return null; + try { return JsonSerializer.Serialize(value, ScriptTraceExtra.JsonOptions); } + catch (Exception) { return null; } } /// @@ -477,9 +563,12 @@ public class ScriptRuntimeContext cid, _instanceName, name, targetValueEncoded, predicate, timeout, DateTimeOffset.UtcNow, requireGoodQuality); + var startedUtc = DateTime.UtcNow; + var sw = _isTraced ? Stopwatch.StartNew() : null; + WaitForAttributeResponse response; try { - return await _instanceActor.Ask( + response = await _instanceActor.Ask( req, timeout + _askTimeout, _scriptTimeoutToken); } catch (AskTimeoutException) @@ -491,9 +580,47 @@ public class ScriptRuntimeContext // OperationCanceledException / TaskCanceledException from the // script-deadline token are deliberately NOT caught here: they must // propagate to abort the script (§4.3). - return new WaitForAttributeResponse( + response = new WaitForAttributeResponse( cid, Matched: false, null, null, TimedOut: true); } + catch (OperationCanceledException) + { + // Script deadline hit mid-wait: record the cancelled wait for a traced run, then + // let the cancellation propagate exactly as before (§4.3). + if (sw is not null) + { + EmitTrace(AuditKind.AttributeWait, AuditStatus.Failed, name, startedUtc, + (int)sw.ElapsedMilliseconds, null, null, "Wait cancelled by the script deadline", + new ScriptTraceExtra + { + WaitAttribute = name, WaitTarget = targetValueEncoded ?? (predicate is null ? null : ""), + WaitTimeoutMs = (int)timeout.TotalMilliseconds, WaitOutcome = "cancelled", + }); + } + throw; + } + + if (sw is not null) + { + var outcome = response.Matched ? "satisfied" : response.TimedOut ? "timeout" : "error"; + var observed = AttributeValueCodec.Encode(response.Value); + EmitTrace(AuditKind.AttributeWait, + response.Matched ? AuditStatus.Delivered : AuditStatus.Failed, + name, startedUtc, (int)sw.ElapsedMilliseconds, + input: TraceJson(new { attribute = name, target = targetValueEncoded, timeoutMs = (int)timeout.TotalMilliseconds, requireGoodQuality }), + output: TraceJson(new { matched = response.Matched, timedOut = response.TimedOut, value = observed, quality = response.Quality }), + error: response.Matched ? null : response.ErrorMessage ?? (response.TimedOut ? $"Wait for '{name}' timed out after {timeout.TotalSeconds}s" : null), + extra: new ScriptTraceExtra + { + WaitAttribute = name, + WaitTarget = targetValueEncoded ?? (predicate is null ? null : ""), + WaitTimeoutMs = (int)timeout.TotalMilliseconds, + WaitOutcome = outcome, + ObservedValue = observed, + }); + } + + return response; } /// @@ -519,11 +646,41 @@ public class ScriptRuntimeContext // ExecutionId as the write's originating execution. The Instance Actor // carries it onto the published AttributeValueChanged, so an alarm this // write trips records this run as the on-trigger script's parent. - SourceExecutionId: _executionId); + SourceExecutionId: _executionId, + // Flow Debugger (design §5.2/§5.3): a traced run's write arms the instance + // actor's echo stamp and rides onto the published AttributeValueChanged. + Trace: _isTraced); - // Ask — mutation serialized through the Instance Actor mailbox; the reply - // carries the device-write outcome for data-connected attributes. - var response = await _instanceActor.Ask(command, _askTimeout); + var startedUtc = DateTime.UtcNow; + var sw = _isTraced ? Stopwatch.StartNew() : null; + SetStaticAttributeResponse response; + try + { + // Ask — mutation serialized through the Instance Actor mailbox; the reply + // carries the device-write outcome for data-connected attributes. + response = await _instanceActor.Ask(command, _askTimeout); + } + catch (Exception ex) when (sw is not null) + { + EmitTrace(AuditKind.AttributeWrite, AuditStatus.Failed, attributeName, startedUtc, + (int)sw.ElapsedMilliseconds, TraceJson(new { attribute = attributeName, value }), null, ex.Message, + new ScriptTraceExtra { Writes = [new AttributeWriteDetail(attributeName, value, false, ex.Message)] }); + throw; + } + + if (sw is not null) + { + EmitTrace(AuditKind.AttributeWrite, + response.Success ? AuditStatus.Delivered : AuditStatus.Failed, + attributeName, startedUtc, (int)sw.ElapsedMilliseconds, + input: TraceJson(new { attribute = attributeName, value }), + output: null, + error: response.ErrorMessage, + extra: new ScriptTraceExtra + { + Writes = [new AttributeWriteDetail(attributeName, value, response.Success, response.ErrorMessage)], + }); + } if (!response.Success) { @@ -560,9 +717,28 @@ public class ScriptRuntimeContext // 35s: the InstanceActor's DCL Ask is internally bounded at 30s, so allow a small // margin so the DCL's own typed failure/timeout reply is the one we observe rather // than an AskTimeoutException here. Honors the script execution-timeout token. + var startedUtc = DateTime.UtcNow; + var sw = _isTraced ? Stopwatch.StartNew() : null; var batchResp = await _instanceActor.Ask( batchReq, TimeSpan.FromSeconds(35), _scriptTimeoutToken); + if (sw is not null) + { + // Flow Debugger (design §5.1): one AttributeWrite row per batch — every value plus the + // trigger — with the batch's single outcome on each write. + var writes = encodedValues + .Select(kv => new AttributeWriteDetail(kv.Key, kv.Value, batchResp.Success, batchResp.ErrorMessage)) + .Append(new AttributeWriteDetail(triggerAttr, triggerEncoded, batchResp.Success, batchResp.ErrorMessage)) + .ToList(); + EmitTrace(AuditKind.AttributeWrite, + batchResp.Success ? AuditStatus.Delivered : AuditStatus.Failed, + triggerAttr, startedUtc, (int)sw.ElapsedMilliseconds, + input: TraceJson(new { values = encodedValues, trigger = triggerAttr, triggerValue = triggerEncoded }), + output: null, + error: batchResp.ErrorMessage, + extra: new ScriptTraceExtra { TriggerAttribute = triggerAttr, TriggerValue = triggerEncoded, Writes = writes }); + } + if (!batchResp.Success) throw new InvalidOperationException( $"WriteBatchAndWait write failed: {batchResp.ErrorMessage}"); @@ -605,7 +781,9 @@ public class ScriptRuntimeContext // So A → CallScript(B) yields B.ParentExecutionId == A.ExecutionId, // building a true multi-level execution tree rather than flattening // every nested call under the original inbound spawner. - ParentExecutionId: _executionId); + ParentExecutionId: _executionId, + // Flow Debugger (design §5.2): nested runs inherit the trace flag. + Trace: _isTraced); // Ask the Instance Actor, which routes to the appropriate Script Actor var result = await _instanceActor.Ask(request, _askTimeout); diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Scripts/ScriptRuntimeContextTraceEmissionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Scripts/ScriptRuntimeContextTraceEmissionTests.cs new file mode 100644 index 00000000..5ceab46f --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Scripts/ScriptRuntimeContextTraceEmissionTests.cs @@ -0,0 +1,206 @@ +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Instance; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.ScriptExecution; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging; +using IAuditWriter = ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services.IAuditWriter; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts; + +namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Scripts; + +/// +/// Flow Debugger design 2026-09-05 §5.1/§5.2: a traced +/// 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. +/// +public class ScriptRuntimeContextTraceEmissionTests : TestKit +{ + private const string InstanceName = "Plant.Pump42"; + + 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; + } + } + + 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) + => new( + instanceActor, NewLibrary(), + currentCallDepth: 0, maxCallDepth: 10, askTimeout: TimeSpan.FromSeconds(5), + instanceName: InstanceName, logger: NullLogger.Instance, + siteId: "site-a", sourceScript: "ScriptActor:A", auditWriter: writer, + 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(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(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); + } + + [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(TimeSpan.FromSeconds(5)); + probe.Reply(new SetStaticAttributeResponse(cmd.CorrelationId, InstanceName, "Setpoint", false, "device rejected", DateTimeOffset.UtcNow)); + + await Assert.ThrowsAsync(() => 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(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(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(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(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(TimeSpan.FromSeconds(5)); + Assert.False(req.Trace); + probe.Reply(new ScriptCallResult(req.CorrelationId, true, null, null)); + await call; + } +}