From 01be39019e48b12db2cfdaaca7ac98e2accb5a7b Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 5 Sep 2026 23:43:33 -0400 Subject: [PATCH] feat(site-runtime): InstanceActor trace arm stamps the device echo of a traced write; Trace rides on published changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Flow Debugger (design 2026-09-05 §5.3): a traced SetStaticAttributeCommand on a data-sourced attribute records (attribute → writer, expiry) for TraceArmWindowSeconds (new option, default 30, validated > 0); the next DCL echo for that attribute is published with the writer's SourceExecutionId + Trace whatever its value, and the arm is consumed once. Static writes publish Trace directly. Untraced writes never arm (pinned). Claude-Session: https://claude.ai/code/session_01GSLkkHKz9JrAtmJjmQK3cP --- .../Actors/InstanceActor.cs | 42 ++++- .../SiteRuntimeOptions.cs | 11 ++ .../SiteRuntimeOptionsValidator.cs | 5 + .../Actors/InstanceActorTraceArmTests.cs | 178 ++++++++++++++++++ .../SiteRuntimeOptionsValidatorTests.cs | 11 ++ 5 files changed, 243 insertions(+), 4 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTraceArmTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs index 65bbf00a..3989ca20 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs @@ -172,6 +172,15 @@ public class InstanceActor : ReceiveActor // DCL manager actor reference for subscribing to tag values private readonly IActorRef? _dclManager; + + /// + /// Flow Debugger trace arm (design 2026-09-05 §5.3): attribute → (writer's ExecutionId, + /// expiry) recorded when a TRACED script write targets a data-sourced attribute. The next + /// DCL value change for that attribute is published stamped with the writer's id and + /// Trace, and the arm is consumed. Bounded by the attribute count (one entry per + /// attribute, newest wins); no sweep — an unconsumed arm is overwritten or expires in place. + /// + private readonly Dictionary _traceArms = new(); // Maps each tag path to every attribute canonical name that references it. // A tag path can back more than one attribute (e.g. two composed modules // whose members reference the same PLC node), so a tag value update must @@ -592,7 +601,10 @@ public class InstanceActor : ReceiveActor // execution's id onto the change so a child Alarm Actor this write // trips can chain its on-trigger script run under the writer. Null for // writes with no audited origin (central Test Run, deployment tooling). - SourceExecutionId: command.SourceExecutionId); + SourceExecutionId: command.SourceExecutionId, + // Flow Debugger (design §5.2): a traced write's change carries the flag so the + // value-change run it spawns is traced and parented to the writer. + Trace: command.Trace); PublishAndNotifyChildren(changed); @@ -742,6 +754,14 @@ public class InstanceActor : ReceiveActor writeValue, DateTimeOffset.UtcNow); + // Flow Debugger trace arm (design §5.3): only a TRACED write with a known writer arms; + // the device echo that follows is stamped in HandleTagValueUpdate. Untraced writes + // never touch the map, so production behaviour is unchanged. + if (command.Trace && command.SourceExecutionId is { } traceSource) + { + _traceArms[attributeName] = (traceSource, DateTimeOffset.UtcNow.AddSeconds(_options.TraceArmWindowSeconds)); + } + // Ask the DCL and pipe the result back to the original caller. The DCL // returns the failure synchronously so the script can handle it. _dclManager.Ask(writeRequest, TimeSpan.FromSeconds(30)) @@ -1035,6 +1055,17 @@ public class InstanceActor : ReceiveActor // O(1) lookup off the hot ingest path (was a linear FirstOrDefault). _resolvedAttributeByName.TryGetValue(attrName, out var resolved); + // Flow Debugger trace arm (design §5.3): consume the arm for this attribute — the + // echo is stamped whatever its value (a device that clamps or rejects the write + // still shows what it did) — unless the arm has already expired. + Guid? traceSource = null; + var traced = false; + if (_traceArms.Remove(attrName, out var arm) && arm.ExpiresAt > DateTimeOffset.UtcNow) + { + traceSource = arm.SourceExecutionId; + traced = true; + } + // A List-typed attribute coerces the incoming OPC UA array // (a CLR array/IEnumerable from the SDK) into a typed List. On an // element-type mismatch we set the attribute's quality to Bad, log a @@ -1045,7 +1076,8 @@ public class InstanceActor : ReceiveActor { HandleAttributeValueChanged(new AttributeValueChanged( _instanceUniqueName, update.TagPath, attrName, - typedList, QualityToString(update.Quality), update.Timestamp)); + typedList, QualityToString(update.Quality), update.Timestamp, + SourceExecutionId: traceSource, Trace: traced)); } else { @@ -1062,7 +1094,8 @@ public class InstanceActor : ReceiveActor // non-change. PublishAndNotifyChildren(new AttributeValueChanged( _instanceUniqueName, update.TagPath, attrName, - currentValue, "Bad", update.Timestamp), evaluateWaiters: false); + currentValue, "Bad", update.Timestamp, + SourceExecutionId: traceSource, Trace: traced), evaluateWaiters: false); } continue; } @@ -1076,7 +1109,8 @@ public class InstanceActor : ReceiveActor HandleAttributeValueChanged(new AttributeValueChanged( _instanceUniqueName, update.TagPath, attrName, - value, QualityToString(update.Quality), update.Timestamp)); + value, QualityToString(update.Quality), update.Timestamp, + SourceExecutionId: traceSource, Trace: traced)); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs index 3326577b..c5ec6952 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs @@ -232,4 +232,15 @@ public class SiteRuntimeOptions /// (the row-67 EnsureInstanceStartedCommand path is unaffected). Default: 60. /// public int InstanceSweepIntervalSeconds { get; set; } = 60; + + /// + /// Flow Debugger trace arm window (design 2026-09-05 §5.3). When a TRACED script write + /// targets a data-sourced attribute, the Instance Actor remembers + /// (attribute → writer's ExecutionId) for this many seconds; the next DCL value + /// change for that attribute — whatever the value — is published stamped with that + /// SourceExecutionId and Trace, and the arm is consumed. An arm that expires + /// unconsumed emits nothing. Only traced writes arm, so production behaviour is unchanged. + /// Default: 30. + /// + public int TraceArmWindowSeconds { get; set; } = 30; } diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptionsValidator.cs index fac982c8..d77a3fd3 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptionsValidator.cs @@ -108,5 +108,10 @@ public sealed class SiteRuntimeOptionsValidator : OptionsValidatorBase 0, + $"ScadaBridge:SiteRuntime:TraceArmWindowSeconds must be greater than 0 " + + $"(was {options.TraceArmWindowSeconds}); it is how long the Instance Actor waits for the " + + "device echo of a Flow Debugger traced write before the arm expires (design 2026-09-05 §5.3)."); } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTraceArmTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTraceArmTests.cs new file mode 100644 index 00000000..3e0d1f81 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTraceArmTests.cs @@ -0,0 +1,178 @@ +using Akka.Actor; +using Akka.TestKit; +using Akka.TestKit.Xunit2; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Instance; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming; +using ZB.MOM.WW.ScadaBridge.TestSupport; +using System.Text.Json; + +namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors; + +/// +/// Flow Debugger design 2026-09-05 §5.3 — the Instance Actor trace arm: a TRACED write to a +/// data-sourced attribute stamps the next DCL echo for that attribute with the writer's +/// SourceExecutionId and Trace; the arm is consumed once and expires unconsumed. +/// Untraced writes never arm (the production-unchanged pin). A traced static write publishes +/// the flag directly. +/// +public class InstanceActorTraceArmTests : TestKit, IDisposable +{ + private const string Tag = "ns=3;s=Recipe.Processed"; + private readonly SiteStorageService _storage; + private readonly ScriptCompilationService _compilationService; + private readonly SharedScriptLibrary _sharedScriptLibrary; + private readonly TestLocalDb _localDb; + + public InstanceActorTraceArmTests() + { + _localDb = TestLocalDb.CreateTemp("instance-tracearm-test"); + _storage = new SiteStorageService(_localDb.Db, NullLogger.Instance); + _storage.InitializeAsync().GetAwaiter().GetResult(); + _compilationService = new ScriptCompilationService(NullLogger.Instance); + _sharedScriptLibrary = new SharedScriptLibrary(_compilationService, NullLogger.Instance); + } + + void IDisposable.Dispose() + { + Shutdown(); + var path = _localDb.Path; + _localDb.Dispose(); + TestLocalDb.DeleteFiles(path); + } + + private static FlattenedConfiguration Config(string instance) => new() + { + InstanceUniqueName = instance, + Attributes = + [ + new ResolvedAttribute + { + CanonicalName = "Processed", Value = "false", DataType = "Boolean", + DataSourceReference = Tag, BoundDataConnectionName = "PLC" + }, + new ResolvedAttribute { CanonicalName = "Status", Value = "Idle", DataType = "String" }, + ] + }; + + private (IActorRef Actor, TestProbe Dcl, TestProbe Subscriber) Create(string instance, int armWindowSeconds = 30) + { + var options = new SiteRuntimeOptions { StreamBufferSize = 100, TraceArmWindowSeconds = armWindowSeconds }; + var streamManager = new SiteStreamManager(options, NullLogger.Instance); + streamManager.Initialize(Sys); + var dcl = CreateTestProbe(); + var actor = ActorOf(Props.Create(() => new InstanceActor( + instance, JsonSerializer.Serialize(Config(instance)), _storage, _compilationService, + _sharedScriptLibrary, streamManager, options, NullLogger.Instance, dcl.Ref))); + var subscriber = CreateTestProbe(); + streamManager.Subscribe(instance, subscriber.Ref); + return (actor, dcl, subscriber); + } + + private static WriteTagRequest ExpectWrite(TestProbe dcl) + => dcl.FishForMessage(_ => true, TimeSpan.FromSeconds(5)); + + private void Write(IActorRef actor, TestProbe dcl, string instance, Guid? source, bool trace) + { + actor.Tell(new SetStaticAttributeCommand("c", instance, "Processed", "true", DateTimeOffset.UtcNow, + SourceExecutionId: source, Trace: trace)); + var write = ExpectWrite(dcl); + Assert.Equal(Tag, write.TagPath); + dcl.Reply(new WriteTagResponse(write.CorrelationId, true, null, DateTimeOffset.UtcNow)); + Assert.True(ExpectMsg(TimeSpan.FromSeconds(5)).Success); + } + + private static AttributeValueChanged Echo(IActorRef actor, TestProbe subscriber, object? value) + { + actor.Tell(new TagValueUpdate("PLC", Tag, value, QualityCode.Good, DateTimeOffset.UtcNow)); + return subscriber.FishForMessage( + m => m.AttributeName == "Processed" && Equals(m.Value, value), TimeSpan.FromSeconds(10)); + } + + [Fact] + public void TracedDataWrite_ThenEcho_PublishesChangeStampedWithSourceAndTrace() + { + var (actor, dcl, sub) = Create("P1"); + var writer = Guid.NewGuid(); + + Write(actor, dcl, "P1", writer, trace: true); + // The device clamps/echoes whatever it wants — the stamp does not depend on the value. + var published = Echo(actor, sub, "clamped"); + + Assert.Equal(writer, published.SourceExecutionId); + Assert.True(published.Trace); + } + + [Fact] + public void UntracedDataWrite_ThenEcho_PublishesUnstampedChange() + { + var (actor, dcl, sub) = Create("P2"); + + // Stamped writer but NOT traced — the ordinary script-write path. + Write(actor, dcl, "P2", Guid.NewGuid(), trace: false); + var published = Echo(actor, sub, true); + + Assert.Null(published.SourceExecutionId); + Assert.False(published.Trace); + } + + [Fact] + public void Arm_IsConsumedOnce() + { + var (actor, dcl, sub) = Create("P3"); + var writer = Guid.NewGuid(); + + Write(actor, dcl, "P3", writer, trace: true); + var first = Echo(actor, sub, "one"); + var second = Echo(actor, sub, "two"); + + Assert.Equal(writer, first.SourceExecutionId); + Assert.True(first.Trace); + Assert.Null(second.SourceExecutionId); + Assert.False(second.Trace); + } + + [Fact] + [Trait("Category", "Slow")] + public void Arm_Expired_IsIgnored() + { + var (actor, dcl, sub) = Create("P4", armWindowSeconds: 1); + + Write(actor, dcl, "P4", Guid.NewGuid(), trace: true); + Thread.Sleep(1100); + var published = Echo(actor, sub, true); + + Assert.Null(published.SourceExecutionId); + Assert.False(published.Trace); + } + + [Fact] + public void StaticWrite_Traced_PublishesTraceFlag() + { + var (actor, _, sub) = Create("P5"); + var writer = Guid.NewGuid(); + + actor.Tell(new SetStaticAttributeCommand("c", "P5", "Status", "Running", DateTimeOffset.UtcNow, + SourceExecutionId: writer, Trace: true)); + Assert.True(ExpectMsg(TimeSpan.FromSeconds(5)).Success); + + var published = sub.FishForMessage( + m => m.AttributeName == "Status" && Equals(m.Value, "Running"), TimeSpan.FromSeconds(10)); + Assert.Equal(writer, published.SourceExecutionId); + Assert.True(published.Trace); + + // And an untraced static write publishes Trace=false (pin). + actor.Tell(new SetStaticAttributeCommand("c2", "P5", "Status", "Idle2", DateTimeOffset.UtcNow, SourceExecutionId: writer)); + ExpectMsg(TimeSpan.FromSeconds(5)); + var plain = sub.FishForMessage( + m => m.AttributeName == "Status" && Equals(m.Value, "Idle2"), TimeSpan.FromSeconds(10)); + Assert.False(plain.Trace); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/SiteRuntimeOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/SiteRuntimeOptionsValidatorTests.cs index e9f130eb..d8a416ae 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/SiteRuntimeOptionsValidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/SiteRuntimeOptionsValidatorTests.cs @@ -136,6 +136,17 @@ public class SiteRuntimeOptionsValidatorTests } /// A zero cap would shed every trigger, silently disabling all scripts. + [Fact] + public void ZeroTraceArmWindowSeconds_IsRejected() + { + // Flow Debugger design 2026-09-05 §5.3: the arm window must be positive. + var result = Validate(new SiteRuntimeOptions { TraceArmWindowSeconds = 0 }); + + Assert.True(result.Failed); + Assert.Contains("TraceArmWindowSeconds", result.FailureMessage); + Assert.Equal(30, new SiteRuntimeOptions().TraceArmWindowSeconds); + } + [Fact] public void ZeroMaxConcurrentRunsPerScript_IsRejected() {