Compare commits

...

4 Commits

Author SHA1 Message Date
Joseph Doherty
df39809526 Phase 7 Stream C — Core.ScriptedAlarms project (Part 9 state machine + predicate engine + IAlarmSource adapter)
Ships the Part 9 alarm fidelity layer Phase 7 committed to in plan decision #5. Every scripted alarm gets a full OPC UA AlarmConditionType state machine — EnabledState, ActiveState, AckedState, ConfirmedState, ShelvingState — with persistent operator-supplied state across server restarts per Phase 7 plan decision #14. Runtime shape matches the Galaxy-native + AB CIP ALMD alarm sources: scripted alarms fan out through the existing IAlarmSource surface so Phase 6.1 AlarmTracker composition consumes them without per-source branching.

Part9StateMachine is a pure-functions module — no instance state, no I/O, no mutation. Every transition (ApplyPredicate, ApplyAcknowledge, ApplyConfirm, ApplyOneShotShelve, ApplyTimedShelve, ApplyUnshelve, ApplyEnable, ApplyDisable, ApplyAddComment, ApplyShelvingCheck) takes the current AlarmConditionState record plus the event and returns a fresh state + EmissionKind hint. Two structural invariants enforced: disabled alarms never transition ActiveState / AckedState / ConfirmedState; shelved alarms still advance state (so startup recovery reflects reality) but emit a Suppressed hint so subscribers do not see the transition. OneShot shelving expires on clear; Timed shelving expires via ApplyShelvingCheck against the UnshelveAtUtc timestamp. Comments are append-only — every acknowledge, confirm, shelve, unshelve, enable, disable, explicit add-comment, and auto-unshelve appends an AlarmComment record with user identity + timestamp + kind + text for the GxP / 21 CFR Part 11 audit surface.

AlarmConditionState is the persistent record the store saves. Fields: AlarmId, Enabled, Active, Acked, Confirmed, Shelving (kind + UnshelveAtUtc), LastTransitionUtc, LastActiveUtc, LastClearedUtc, LastAckUtc + LastAckUser + LastAckComment, LastConfirmUtc + LastConfirmUser + LastConfirmComment, Comments. Fresh factory initializes everything to the no-event position.

IAlarmStateStore is the persistence abstraction — LoadAsync, LoadAllAsync, SaveAsync, RemoveAsync. Stream E wires this to a SQL-backed store with IAuditLogger hooks; tests use InMemoryAlarmStateStore. Startup recovery per Phase 7 plan decision #14: LoadAsync runs every configured alarm predicate against current tag values to rederive ActiveState, but EnabledState / AckedState / ConfirmedState / ShelvingState + audit history are loaded verbatim from the store so operators do not re-ack after an outage and shelved alarms stay shelved through maintenance windows.

MessageTemplate implements Phase 7 plan decision #13 — static-with-substitution. {TagPath} tokens resolved at event emission time from the engine value cache. Missing paths, non-Good quality, or null values all resolve to {?} so the event still fires but the operator sees where the reference broke. ExtractTokenPaths enumerates tokens at publish time so the engine knows to subscribe to every template-referenced tag in addition to predicate-referenced tags.

AlarmPredicateContext is the ScriptContext subclass alarm scripts see. GetTag reads from the engine shared cache; SetVirtualTag is explicitly rejected at runtime with a pointed error message — alarm predicates must be pure so their output does not couple to virtual-tag state in ways that become impossible to reason about. If cross-tag side effects are needed, the operator authors a virtual tag and the alarm predicate reads it.

ScriptedAlarmEngine orchestrates. LoadAsync compiles every predicate through Stream A ScriptSandbox + ForbiddenTypeAnalyzer, runs DependencyExtractor to find the read set, adds template token paths to the input set, reports every compile failure as one aggregated InvalidOperationException (not one-at-a-time), subscribes to each unique referenced upstream path, seeds the value cache, loads persisted state for each alarm (falling back to Fresh for first-load), re-evaluates the predicate, and saves the recovered state. ChangeTrigger — when an upstream tag changes, look up every alarm referencing that path in a per-path inverse index, enqueue all of them for re-evaluation via a SemaphoreSlim-gated path. Unlike the virtual-tag engine, scripted alarms are leaves in the evaluation DAG (no alarm drives another alarm), so no topological sort is needed. Operator actions (AcknowledgeAsync, ConfirmAsync, OneShotShelveAsync, TimedShelveAsync, UnshelveAsync, EnableAsync, DisableAsync, AddCommentAsync) route through the state machine, persist, and emit if there is an emission. A 5-second shelving-check timer auto-expires Timed shelving and emits Unshelved events at the right moment. Predicate evaluation errors (script throws, timeout, compile-time reads bad tag) leave the state unchanged — the engine does NOT invent a clear transition on predicate failure. Logged as scripts-*.log Error; companion WARN in main log.

ScriptedAlarmSource implements IAlarmSource. SubscribeAlarmsAsync filter is a set of equipment-path prefixes; empty means all. AcknowledgeAsync from the base interface routes to the engine with user identity "opcua-client" — Stream G will replace this with the authenticated principal from the OPC UA dispatch layer. The adapter implements only the base IAlarmSource methods; richer Part 9 methods (Confirm, Shelve, Unshelve, AddComment) remain on the engine and will bind to OPC UA method nodes in Stream G.

47 unit tests across 5 files. Part9StateMachineTests (16) — every transition + noop edge cases: predicate true/false, same-state noop, disabled ignores predicate, acknowledge records user/comment/adds audit, idempotent acknowledge, reject no-user ack, full activate-ack-clear-confirm walk, one-shot shelve suppresses next activation, one-shot expires on clear, timed shelve requires future unshelve time, timed shelve expires via shelving-check, explicit unshelve emits, add-comment appends to audit, comments append-only through multiple operations, full lifecycle walk emits every expected EmissionKind. MessageTemplateTests (11) — no-token passthrough, single+multiple token substitution, bad quality becomes {?}, unknown path becomes {?}, null value becomes {?}, tokens with slashes+dots, empty + null template, ExtractTokenPaths returns every distinct path, whitespace inside tokens trimmed. ScriptedAlarmEngineTests (13) — load compiles+subscribes, compile failures aggregated, upstream change emits Activated, clearing emits Cleared, message template resolves at emission, ack persists to store, startup recovery preserves ack but rederives active, shelved activation state-advances but suppresses emission, runtime exception isolates to owning alarm, disable prevents activation until re-enable, AddComment appends audit without state change, SetVirtualTag from predicate rejected (state unchanged), Dispose releases upstream subscriptions. ScriptedAlarmSourceTests (5) — empty filter matches all, equipment-prefix filter, Unsubscribe stops events, AcknowledgeAsync routes with default user, null arguments rejected. FakeUpstream fixture gives tests an in-memory driver mock with subscription count tracking.

Full Phase 7 test count after Stream C: 146 green (63 Scripting + 36 VirtualTags + 47 ScriptedAlarms). Stream D (historian alarm sink with SQLite store-and-forward + Galaxy.Host IPC) consumes ScriptedAlarmEvent + similar Galaxy / AB CIP emissions to produce the unified alarm timeline. Stream G wires the OPC UA method calls and AlarmSource into DriverNodeManager dispatch.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-20 18:49:48 -04:00
2a8bcc8f60 Merge pull request 'Phase 7 Stream B — Core.VirtualTags engine + dep graph + timer + source' (#180) from phase-7-stream-b-virtual-tag-engine into v2 2026-04-20 17:05:13 -04:00
Joseph Doherty
479af166ab Phase 7 Stream B — Core.VirtualTags project (engine + dep graph + timer + source)
Ships the evaluation engine that consumes compiled scripts from Stream A, subscribes to upstream driver tags, runs on change + on timer, cascades evaluations through dependent virtual tags in topological order, and emits changes through a driver-capability-shaped adapter the DriverNodeManager can dispatch to per ADR-002.

DependencyGraph owns the directed dep-graph where nodes are tag paths (driver tags implicit leaves, virtual tags registered internal nodes) and edges run from a virtual tag to each tag it reads. Kahn algorithm produces the topological sort. Tarjan iterative SCC detects every cycle in one pass so publish-time rejection surfaces all offending cycles together. Both iterative so 10k-deep chains do not StackOverflow. Re-adding a node overwrites prior dependency set cleanly (supports config-publish reloads).

VirtualTagDefinition is the operator-authored config row (Path, DataType, ScriptSource, ChangeTriggered, TimerInterval, Historize). Stream E config DB materializes these on publish.

ITagUpstreamSource is the abstraction the engine pulls driver tag values from. Stream G bridges this to IReadable + ISubscribable on live drivers; tests use FakeUpstream that tracks subscription count for leak-test assertions.

IHistoryWriter is the per-tag Historize sink. NullHistoryWriter default when caller does not pass one.

VirtualTagContext is the per-evaluation ScriptContext. Reads from engine last-known-value cache, writes route through SetVirtualTag callback so cross-tag side effects participate in change cascades. Injectable Now clock for deterministic tests.

VirtualTagEngine orchestrates. Load compiles every script via ScriptSandbox, builds the dep graph via DependencyExtractor, checks for cycles, reports every compile failure in one error, subscribes to each referenced upstream path, seeds the value cache. EvaluateAllAsync runs topological order. EvaluateOneAsync is timer path. Read returns cached value. Subscribe registers observer. OnUpstreamChange updates cache, fans out, schedules transitive dependents (change-driven=false tags skipped). EvaluateInternalAsync holds a SemaphoreSlim so cascades do not interleave. Script exceptions and timeouts map per-tag to BadInternalError. Coercion from script double to config Int32 uses Convert.ToInt32.

TimerTriggerScheduler groups tags by interval into shared Timers. Tags without TimerInterval not scheduled.

VirtualTagSource implements IReadable + ISubscribable per ADR-002. ReadAsync returns cache. SubscribeAsync fires initial-data callback per OPC UA convention. IWritable deliberately not implemented — OPC UA writes to virtual tags rejected in DriverNodeManager per Phase 7 decision 6.

36 unit tests across 4 files: DependencyGraphTests 12, VirtualTagEngineTests 13, VirtualTagSourceTests 6, TimerTriggerSchedulerTests 4. Coverage includes cycle detection (self-loop, 2-node, 3-node, multiple disjoint), 2-level change cascade, per-tag error isolation (one tag throws, others keep working), timeout isolation, Historize toggle, ChangeTriggered=false ignore, reload cleans subscriptions, Dispose releases resources, SetVirtualTag fires observers, type coercion, 10k deep graph no stack overflow, initial-data callback, Unsubscribe stops events.

Fixed two bugs during implementation. Monitor.Enter/Exit cannot be held across await (Monitor ownership is thread-local and lost across suspension) — switched to SemaphoreSlim. Kahn edge-direction was inverted — for dependency ordering (X depends on Y means Y comes before X) in-degree should be count of a node own deps, not count of nodes pointing to it; was incrementing inDegree[dep] instead of inDegree[nodeId], causing false cycle detection on valid DAGs.

Full Phase 7 test count after Stream B: 99 green (63 Scripting + 36 VirtualTags). Streams C and G will plug engine + source into live OPC UA dispatch path.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-20 17:02:50 -04:00
00724e9784 Merge pull request 'Phase 7 Stream A.3 — ScriptLoggerFactory + ScriptLogCompanionSink (closes Stream A)' (#179) from phase-7-stream-a3-script-logger into v2 2026-04-20 16:45:09 -04:00
32 changed files with 3937 additions and 0 deletions

View File

@@ -4,6 +4,8 @@
<Project Path="src/ZB.MOM.WW.OtOpcUa.Configuration/ZB.MOM.WW.OtOpcUa.Configuration.csproj"/> <Project Path="src/ZB.MOM.WW.OtOpcUa.Configuration/ZB.MOM.WW.OtOpcUa.Configuration.csproj"/>
<Project Path="src/ZB.MOM.WW.OtOpcUa.Core/ZB.MOM.WW.OtOpcUa.Core.csproj"/> <Project Path="src/ZB.MOM.WW.OtOpcUa.Core/ZB.MOM.WW.OtOpcUa.Core.csproj"/>
<Project Path="src/ZB.MOM.WW.OtOpcUa.Core.Scripting/ZB.MOM.WW.OtOpcUa.Core.Scripting.csproj"/> <Project Path="src/ZB.MOM.WW.OtOpcUa.Core.Scripting/ZB.MOM.WW.OtOpcUa.Core.Scripting.csproj"/>
<Project Path="src/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.csproj"/>
<Project Path="src/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.csproj"/>
<Project Path="src/ZB.MOM.WW.OtOpcUa.Server/ZB.MOM.WW.OtOpcUa.Server.csproj"/> <Project Path="src/ZB.MOM.WW.OtOpcUa.Server/ZB.MOM.WW.OtOpcUa.Server.csproj"/>
<Project Path="src/ZB.MOM.WW.OtOpcUa.Admin/ZB.MOM.WW.OtOpcUa.Admin.csproj"/> <Project Path="src/ZB.MOM.WW.OtOpcUa.Admin/ZB.MOM.WW.OtOpcUa.Admin.csproj"/>
<Project Path="src/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.csproj"/> <Project Path="src/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.csproj"/>
@@ -28,6 +30,8 @@
<Project Path="tests/ZB.MOM.WW.OtOpcUa.Configuration.Tests/ZB.MOM.WW.OtOpcUa.Configuration.Tests.csproj"/> <Project Path="tests/ZB.MOM.WW.OtOpcUa.Configuration.Tests/ZB.MOM.WW.OtOpcUa.Configuration.Tests.csproj"/>
<Project Path="tests/ZB.MOM.WW.OtOpcUa.Core.Tests/ZB.MOM.WW.OtOpcUa.Core.Tests.csproj"/> <Project Path="tests/ZB.MOM.WW.OtOpcUa.Core.Tests/ZB.MOM.WW.OtOpcUa.Core.Tests.csproj"/>
<Project Path="tests/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests.csproj"/> <Project Path="tests/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests.csproj"/>
<Project Path="tests/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests/ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests.csproj"/>
<Project Path="tests/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests.csproj"/>
<Project Path="tests/ZB.MOM.WW.OtOpcUa.Server.Tests/ZB.MOM.WW.OtOpcUa.Server.Tests.csproj"/> <Project Path="tests/ZB.MOM.WW.OtOpcUa.Server.Tests/ZB.MOM.WW.OtOpcUa.Server.Tests.csproj"/>
<Project Path="tests/ZB.MOM.WW.OtOpcUa.Admin.Tests/ZB.MOM.WW.OtOpcUa.Admin.Tests.csproj"/> <Project Path="tests/ZB.MOM.WW.OtOpcUa.Admin.Tests/ZB.MOM.WW.OtOpcUa.Admin.Tests.csproj"/>
<Project Path="tests/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Tests/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Tests.csproj"/> <Project Path="tests/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Tests/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Tests.csproj"/>

View File

@@ -0,0 +1,84 @@
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
/// <summary>
/// Persistent per-alarm state tracked by the Part 9 state machine. Every field
/// carried here either participates in the state machine or contributes to the
/// audit trail required by Phase 7 plan decision #14 (GxP / 21 CFR Part 11).
/// </summary>
/// <remarks>
/// <para>
/// <see cref="Active"/> is re-derived from the predicate at startup per Phase 7
/// decision #14 — the engine runs every alarm's predicate against current tag
/// values at <c>Load</c>, overriding whatever Active state is in the store.
/// Every other state field persists verbatim across server restarts so
/// operators don't re-ack active alarms after an outage + shelved alarms stay
/// shelved + audit history survives.
/// </para>
/// <para>
/// <see cref="Comments"/> is append-only; comments + ack/confirm user identities
/// are the audit surface regulators consume. The engine never rewrites past
/// entries.
/// </para>
/// </remarks>
public sealed record AlarmConditionState(
string AlarmId,
AlarmEnabledState Enabled,
AlarmActiveState Active,
AlarmAckedState Acked,
AlarmConfirmedState Confirmed,
ShelvingState Shelving,
DateTime LastTransitionUtc,
DateTime? LastActiveUtc,
DateTime? LastClearedUtc,
DateTime? LastAckUtc,
string? LastAckUser,
string? LastAckComment,
DateTime? LastConfirmUtc,
string? LastConfirmUser,
string? LastConfirmComment,
IReadOnlyList<AlarmComment> Comments)
{
/// <summary>Initial-load state for a newly registered alarm — everything in the "no-event" position.</summary>
public static AlarmConditionState Fresh(string alarmId, DateTime nowUtc) => new(
AlarmId: alarmId,
Enabled: AlarmEnabledState.Enabled,
Active: AlarmActiveState.Inactive,
Acked: AlarmAckedState.Acknowledged,
Confirmed: AlarmConfirmedState.Confirmed,
Shelving: ShelvingState.Unshelved,
LastTransitionUtc: nowUtc,
LastActiveUtc: null,
LastClearedUtc: null,
LastAckUtc: null,
LastAckUser: null,
LastAckComment: null,
LastConfirmUtc: null,
LastConfirmUser: null,
LastConfirmComment: null,
Comments: []);
}
/// <summary>
/// Shelving state — kind plus, for <see cref="ShelvingKind.Timed"/>, the UTC
/// timestamp at which the shelving auto-expires. The engine polls the timer on its
/// evaluation cadence; callers should not rely on millisecond-precision expiry.
/// </summary>
public sealed record ShelvingState(ShelvingKind Kind, DateTime? UnshelveAtUtc)
{
public static readonly ShelvingState Unshelved = new(ShelvingKind.Unshelved, null);
}
/// <summary>
/// A single append-only audit record — acknowledgement / confirmation / explicit
/// comment / shelving action. Every entry carries a monotonic UTC timestamp plus the
/// user identity Phase 6.2 authenticated.
/// </summary>
/// <param name="TimestampUtc">When the action happened.</param>
/// <param name="User">OS / LDAP identity of the actor. For engine-internal events (shelving expiry, startup recovery) this is <c>"system"</c>.</param>
/// <param name="Kind">Human-readable classification — "Acknowledge", "Confirm", "ShelveOneShot", "ShelveTimed", "Unshelve", "AddComment", "Enable", "Disable", "AutoUnshelve".</param>
/// <param name="Text">Operator-supplied comment or engine-generated message.</param>
public sealed record AlarmComment(
DateTime TimestampUtc,
string User,
string Kind,
string Text);

View File

@@ -0,0 +1,55 @@
using Serilog;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
/// <summary>
/// <see cref="ScriptContext"/> subclass for alarm predicate evaluation. Reads from
/// the engine's shared tag cache (driver + virtual tags), writes are rejected —
/// predicates must be side-effect free so their output doesn't depend on evaluation
/// order or drive cascade behavior.
/// </summary>
/// <remarks>
/// Per Phase 7 plan Shape A decision, alarm scripts are one-script-per-alarm
/// returning <c>bool</c>. They read any tag they want but should not write
/// anything (the owning alarm's state is tracked by the engine, not the script).
/// </remarks>
public sealed class AlarmPredicateContext : ScriptContext
{
private readonly IReadOnlyDictionary<string, DataValueSnapshot> _readCache;
private readonly Func<DateTime> _clock;
public AlarmPredicateContext(
IReadOnlyDictionary<string, DataValueSnapshot> readCache,
ILogger logger,
Func<DateTime>? clock = null)
{
_readCache = readCache ?? throw new ArgumentNullException(nameof(readCache));
Logger = logger ?? throw new ArgumentNullException(nameof(logger));
_clock = clock ?? (() => DateTime.UtcNow);
}
public override DataValueSnapshot GetTag(string path)
{
if (string.IsNullOrWhiteSpace(path))
return new DataValueSnapshot(null, 0x80340000u, null, _clock());
return _readCache.TryGetValue(path, out var v)
? v
: new DataValueSnapshot(null, 0x80340000u, null, _clock());
}
public override void SetVirtualTag(string path, object? value)
{
// Predicates must be pure — writing from an alarm script couples alarm state to
// virtual-tag state in a way that's near-impossible to reason about. Rejected
// at runtime with a clear message; operators see it in the scripts-*.log.
throw new InvalidOperationException(
"Alarm predicate scripts cannot write to virtual tags. Move the write logic " +
"into a virtual tag whose value the alarm predicate then reads.");
}
public override DateTime Now => _clock();
public override ILogger Logger { get; }
}

View File

@@ -0,0 +1,40 @@
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
/// <summary>
/// The concrete OPC UA Part 9 alarm subtype a scripted alarm materializes as. The
/// engine's internal state machine is identical regardless of kind — the
/// <c>AlarmKind</c> only affects how the alarm node appears to OPC UA clients
/// (which ObjectType it maps to) and what diagnostic fields are populated.
/// </summary>
public enum AlarmKind
{
/// <summary>Base AlarmConditionType — no numeric or discrete interpretation.</summary>
AlarmCondition,
/// <summary>LimitAlarmType — the condition reflects a numeric setpoint / threshold breach.</summary>
LimitAlarm,
/// <summary>DiscreteAlarmType — the condition reflects a specific discrete value match.</summary>
DiscreteAlarm,
/// <summary>OffNormalAlarmType — the condition reflects deviation from a configured "normal" state.</summary>
OffNormalAlarm,
}
/// <summary>OPC UA Part 9 EnabledState — operator-controlled alarm enable/disable.</summary>
public enum AlarmEnabledState { Enabled, Disabled }
/// <summary>OPC UA Part 9 ActiveState — reflects the current predicate truth.</summary>
public enum AlarmActiveState { Inactive, Active }
/// <summary>OPC UA Part 9 AckedState — operator has acknowledged the active transition.</summary>
public enum AlarmAckedState { Unacknowledged, Acknowledged }
/// <summary>OPC UA Part 9 ConfirmedState — operator has confirmed the clear transition.</summary>
public enum AlarmConfirmedState { Unconfirmed, Confirmed }
/// <summary>
/// OPC UA Part 9 shelving mode.
/// <see cref="OneShot"/> suppresses the next active transition; once cleared
/// the shelving expires and the alarm returns to normal behavior.
/// <see cref="Timed"/> suppresses until a configured expiry timestamp passes.
/// <see cref="Unshelved"/> is the default state — no suppression.
/// </summary>
public enum ShelvingKind { Unshelved, OneShot, Timed }

View File

@@ -0,0 +1,47 @@
using System.Collections.Concurrent;
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
/// <summary>
/// Persistence for <see cref="AlarmConditionState"/> across server restarts. Phase 7
/// plan decision #14: operator-supplied state (EnabledState / AckedState /
/// ConfirmedState / ShelvingState + audit trail) persists; ActiveState is
/// recomputed from the live predicate on startup so operators never re-ack.
/// </summary>
/// <remarks>
/// Stream E wires this to a SQL-backed store against the <c>ScriptedAlarmState</c>
/// table with audit logging through <see cref="Core.Abstractions"/> IAuditLogger.
/// Tests + local dev use <see cref="InMemoryAlarmStateStore"/>.
/// </remarks>
public interface IAlarmStateStore
{
Task<AlarmConditionState?> LoadAsync(string alarmId, CancellationToken ct);
Task<IReadOnlyList<AlarmConditionState>> LoadAllAsync(CancellationToken ct);
Task SaveAsync(AlarmConditionState state, CancellationToken ct);
Task RemoveAsync(string alarmId, CancellationToken ct);
}
/// <summary>In-memory default — used by tests + by dev deployments without a SQL backend.</summary>
public sealed class InMemoryAlarmStateStore : IAlarmStateStore
{
private readonly ConcurrentDictionary<string, AlarmConditionState> _map
= new(StringComparer.Ordinal);
public Task<AlarmConditionState?> LoadAsync(string alarmId, CancellationToken ct)
=> Task.FromResult(_map.TryGetValue(alarmId, out var v) ? v : null);
public Task<IReadOnlyList<AlarmConditionState>> LoadAllAsync(CancellationToken ct)
=> Task.FromResult<IReadOnlyList<AlarmConditionState>>(_map.Values.ToArray());
public Task SaveAsync(AlarmConditionState state, CancellationToken ct)
{
_map[state.AlarmId] = state;
return Task.CompletedTask;
}
public Task RemoveAsync(string alarmId, CancellationToken ct)
{
_map.TryRemove(alarmId, out _);
return Task.CompletedTask;
}
}

View File

@@ -0,0 +1,64 @@
using System.Text.RegularExpressions;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
/// <summary>
/// Per Phase 7 plan decision #13, alarm messages are static-with-substitution
/// templates. The engine resolves <c>{TagPath}</c> tokens at event emission time
/// against current tag values; unresolvable tokens become <c>{?}</c> so the event
/// still fires but the operator sees where the reference broke.
/// </summary>
/// <remarks>
/// <para>
/// Token syntax: <c>{path/with/slashes}</c>. Brace-stripped the contents must
/// match a path the caller's resolver function can look up. No escaping
/// currently — if you need literal braces in the message, reach for a feature
/// request.
/// </para>
/// <para>
/// Pure function. Same inputs always produce the same string. Tests verify the
/// edge cases (no tokens / one token / many / nested / unresolvable / bad
/// quality / null value).
/// </para>
/// </remarks>
public static class MessageTemplate
{
private static readonly Regex TokenRegex = new(@"\{([^{}]+)\}",
RegexOptions.Compiled | RegexOptions.CultureInvariant);
/// <summary>
/// Resolve every <c>{path}</c> token in <paramref name="template"/> using
/// <paramref name="resolveTag"/>. Tokens whose returned <see cref="DataValueSnapshot"/>
/// has a non-Good <see cref="DataValueSnapshot.StatusCode"/> or a null
/// <see cref="DataValueSnapshot.Value"/> resolve to <c>{?}</c>.
/// </summary>
public static string Resolve(string template, Func<string, DataValueSnapshot?> resolveTag)
{
if (string.IsNullOrEmpty(template)) return template ?? string.Empty;
if (resolveTag is null) throw new ArgumentNullException(nameof(resolveTag));
return TokenRegex.Replace(template, match =>
{
var path = match.Groups[1].Value.Trim();
if (path.Length == 0) return "{?}";
var snap = resolveTag(path);
if (snap is null) return "{?}";
if (snap.StatusCode != 0u) return "{?}";
return snap.Value?.ToString() ?? "{?}";
});
}
/// <summary>Enumerate the token paths the template references. Used at publish time to validate references exist.</summary>
public static IReadOnlyList<string> ExtractTokenPaths(string? template)
{
if (string.IsNullOrEmpty(template)) return Array.Empty<string>();
var tokens = new List<string>();
foreach (Match m in TokenRegex.Matches(template))
{
var path = m.Groups[1].Value.Trim();
if (path.Length > 0) tokens.Add(path);
}
return tokens;
}
}

View File

@@ -0,0 +1,294 @@
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
/// <summary>
/// Pure functions for OPC UA Part 9 alarm-condition state transitions. Input = the
/// current <see cref="AlarmConditionState"/> + the event; output = the new state +
/// optional emission hint. The engine calls these; persistence happens around them.
/// </summary>
/// <remarks>
/// <para>
/// No instance state, no I/O, no mutation of the input record. Every transition
/// returns a fresh record. Makes the state machine trivially unit-testable —
/// tests assert on (input, event) -> (output) without standing anything else up.
/// </para>
/// <para>
/// Two invariants the machine enforces:
/// (1) Disabled alarms never transition ActiveState / AckedState / ConfirmedState
/// — all predicate evaluations while disabled produce a no-op result and a
/// diagnostic log line. Re-enable restores normal flow with ActiveState
/// re-derived from the next predicate evaluation.
/// (2) Shelved alarms (OneShot / Timed) don't fire active transitions to
/// subscribers, but the state record still advances so that when shelving
/// expires the ActiveState reflects current reality. OneShot expires on the
/// next clear; Timed expires at <see cref="ShelvingState.UnshelveAtUtc"/>.
/// </para>
/// </remarks>
public static class Part9StateMachine
{
/// <summary>
/// Apply a predicate re-evaluation result. Handles activation, clearing,
/// branch-stack increment when a new active arrives while prior active is
/// still un-acked, and shelving suppression.
/// </summary>
public static TransitionResult ApplyPredicate(
AlarmConditionState current,
bool predicateTrue,
DateTime nowUtc)
{
if (current.Enabled == AlarmEnabledState.Disabled)
return TransitionResult.NoOp(current, "disabled — predicate result ignored");
// Expire timed shelving if the configured clock has passed.
var shelving = MaybeExpireShelving(current.Shelving, nowUtc);
var stateWithShelving = current with { Shelving = shelving };
// Shelved alarms still update state but skip event emission.
var shelved = shelving.Kind != ShelvingKind.Unshelved;
if (predicateTrue && current.Active == AlarmActiveState.Inactive)
{
// Inactive -> Active transition.
// OneShotShelving is consumed on the NEXT clear, not activation — so we
// still suppress this transition's emission.
var next = stateWithShelving with
{
Active = AlarmActiveState.Active,
Acked = AlarmAckedState.Unacknowledged,
Confirmed = AlarmConfirmedState.Unconfirmed,
LastActiveUtc = nowUtc,
LastTransitionUtc = nowUtc,
};
return new TransitionResult(next, shelved ? EmissionKind.Suppressed : EmissionKind.Activated);
}
if (!predicateTrue && current.Active == AlarmActiveState.Active)
{
// Active -> Inactive transition.
var next = stateWithShelving with
{
Active = AlarmActiveState.Inactive,
LastClearedUtc = nowUtc,
LastTransitionUtc = nowUtc,
// OneShotShelving expires on clear — resetting here so the next
// activation fires normally.
Shelving = shelving.Kind == ShelvingKind.OneShot
? ShelvingState.Unshelved
: shelving,
};
return new TransitionResult(next, shelved ? EmissionKind.Suppressed : EmissionKind.Cleared);
}
// Predicate matches current Active — no state change beyond possible shelving
// expiry.
return new TransitionResult(stateWithShelving, EmissionKind.None);
}
/// <summary>Operator acknowledges the currently-active transition.</summary>
public static TransitionResult ApplyAcknowledge(
AlarmConditionState current,
string user,
string? comment,
DateTime nowUtc)
{
if (string.IsNullOrWhiteSpace(user))
throw new ArgumentException("User identity required for audit.", nameof(user));
if (current.Acked == AlarmAckedState.Acknowledged)
return TransitionResult.NoOp(current, "already acknowledged");
var audit = AppendComment(current.Comments, nowUtc, user, "Acknowledge", comment);
var next = current with
{
Acked = AlarmAckedState.Acknowledged,
LastAckUtc = nowUtc,
LastAckUser = user,
LastAckComment = comment,
LastTransitionUtc = nowUtc,
Comments = audit,
};
return new TransitionResult(next, EmissionKind.Acknowledged);
}
/// <summary>Operator confirms the cleared transition. Part 9 requires confirm after clear for retain-flag alarms.</summary>
public static TransitionResult ApplyConfirm(
AlarmConditionState current,
string user,
string? comment,
DateTime nowUtc)
{
if (string.IsNullOrWhiteSpace(user))
throw new ArgumentException("User identity required for audit.", nameof(user));
if (current.Confirmed == AlarmConfirmedState.Confirmed)
return TransitionResult.NoOp(current, "already confirmed");
var audit = AppendComment(current.Comments, nowUtc, user, "Confirm", comment);
var next = current with
{
Confirmed = AlarmConfirmedState.Confirmed,
LastConfirmUtc = nowUtc,
LastConfirmUser = user,
LastConfirmComment = comment,
LastTransitionUtc = nowUtc,
Comments = audit,
};
return new TransitionResult(next, EmissionKind.Confirmed);
}
public static TransitionResult ApplyOneShotShelve(
AlarmConditionState current, string user, DateTime nowUtc)
{
if (string.IsNullOrWhiteSpace(user)) throw new ArgumentException("User required.", nameof(user));
if (current.Shelving.Kind == ShelvingKind.OneShot)
return TransitionResult.NoOp(current, "already one-shot shelved");
var audit = AppendComment(current.Comments, nowUtc, user, "ShelveOneShot", null);
var next = current with
{
Shelving = new ShelvingState(ShelvingKind.OneShot, null),
LastTransitionUtc = nowUtc,
Comments = audit,
};
return new TransitionResult(next, EmissionKind.Shelved);
}
public static TransitionResult ApplyTimedShelve(
AlarmConditionState current, string user, DateTime unshelveAtUtc, DateTime nowUtc)
{
if (string.IsNullOrWhiteSpace(user)) throw new ArgumentException("User required.", nameof(user));
if (unshelveAtUtc <= nowUtc)
throw new ArgumentOutOfRangeException(nameof(unshelveAtUtc), "Unshelve time must be in the future.");
var audit = AppendComment(current.Comments, nowUtc, user, "ShelveTimed",
$"UnshelveAtUtc={unshelveAtUtc:O}");
var next = current with
{
Shelving = new ShelvingState(ShelvingKind.Timed, unshelveAtUtc),
LastTransitionUtc = nowUtc,
Comments = audit,
};
return new TransitionResult(next, EmissionKind.Shelved);
}
public static TransitionResult ApplyUnshelve(AlarmConditionState current, string user, DateTime nowUtc)
{
if (string.IsNullOrWhiteSpace(user)) throw new ArgumentException("User required.", nameof(user));
if (current.Shelving.Kind == ShelvingKind.Unshelved)
return TransitionResult.NoOp(current, "not shelved");
var audit = AppendComment(current.Comments, nowUtc, user, "Unshelve", null);
var next = current with
{
Shelving = ShelvingState.Unshelved,
LastTransitionUtc = nowUtc,
Comments = audit,
};
return new TransitionResult(next, EmissionKind.Unshelved);
}
public static TransitionResult ApplyEnable(AlarmConditionState current, string user, DateTime nowUtc)
{
if (string.IsNullOrWhiteSpace(user)) throw new ArgumentException("User required.", nameof(user));
if (current.Enabled == AlarmEnabledState.Enabled)
return TransitionResult.NoOp(current, "already enabled");
var audit = AppendComment(current.Comments, nowUtc, user, "Enable", null);
var next = current with
{
Enabled = AlarmEnabledState.Enabled,
LastTransitionUtc = nowUtc,
Comments = audit,
};
return new TransitionResult(next, EmissionKind.Enabled);
}
public static TransitionResult ApplyDisable(AlarmConditionState current, string user, DateTime nowUtc)
{
if (string.IsNullOrWhiteSpace(user)) throw new ArgumentException("User required.", nameof(user));
if (current.Enabled == AlarmEnabledState.Disabled)
return TransitionResult.NoOp(current, "already disabled");
var audit = AppendComment(current.Comments, nowUtc, user, "Disable", null);
var next = current with
{
Enabled = AlarmEnabledState.Disabled,
LastTransitionUtc = nowUtc,
Comments = audit,
};
return new TransitionResult(next, EmissionKind.Disabled);
}
public static TransitionResult ApplyAddComment(
AlarmConditionState current, string user, string text, DateTime nowUtc)
{
if (string.IsNullOrWhiteSpace(user)) throw new ArgumentException("User required.", nameof(user));
if (string.IsNullOrWhiteSpace(text)) throw new ArgumentException("Comment text required.", nameof(text));
var audit = AppendComment(current.Comments, nowUtc, user, "AddComment", text);
var next = current with { Comments = audit };
return new TransitionResult(next, EmissionKind.CommentAdded);
}
/// <summary>
/// Re-evaluate whether a currently timed-shelved alarm has expired. Returns
/// the (possibly unshelved) state + emission hint so the engine knows to
/// publish an Unshelved event at the right moment.
/// </summary>
public static TransitionResult ApplyShelvingCheck(AlarmConditionState current, DateTime nowUtc)
{
if (current.Shelving.Kind != ShelvingKind.Timed) return TransitionResult.None(current);
if (current.Shelving.UnshelveAtUtc is DateTime t && nowUtc >= t)
{
var audit = AppendComment(current.Comments, nowUtc, "system", "AutoUnshelve",
$"Timed shelving expired at {nowUtc:O}");
var next = current with
{
Shelving = ShelvingState.Unshelved,
LastTransitionUtc = nowUtc,
Comments = audit,
};
return new TransitionResult(next, EmissionKind.Unshelved);
}
return TransitionResult.None(current);
}
private static ShelvingState MaybeExpireShelving(ShelvingState s, DateTime nowUtc)
{
if (s.Kind != ShelvingKind.Timed) return s;
return s.UnshelveAtUtc is DateTime t && nowUtc >= t ? ShelvingState.Unshelved : s;
}
private static IReadOnlyList<AlarmComment> AppendComment(
IReadOnlyList<AlarmComment> existing, DateTime ts, string user, string kind, string? text)
{
var list = new List<AlarmComment>(existing.Count + 1);
list.AddRange(existing);
list.Add(new AlarmComment(ts, user, kind, text ?? string.Empty));
return list;
}
}
/// <summary>Result of a state-machine operation — new state + what to emit (if anything).</summary>
public sealed record TransitionResult(AlarmConditionState State, EmissionKind Emission)
{
public static TransitionResult None(AlarmConditionState state) => new(state, EmissionKind.None);
public static TransitionResult NoOp(AlarmConditionState state, string reason) => new(state, EmissionKind.None);
}
/// <summary>What kind of event, if any, the engine should emit after a transition.</summary>
public enum EmissionKind
{
/// <summary>State did not change meaningfully — no event to emit.</summary>
None,
/// <summary>Predicate transitioned to true while shelving was suppressing events.</summary>
Suppressed,
Activated,
Cleared,
Acknowledged,
Confirmed,
Shelved,
Unshelved,
Enabled,
Disabled,
CommentAdded,
}

View File

@@ -0,0 +1,50 @@
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
/// <summary>
/// Operator-authored scripted-alarm configuration. Phase 7 Stream E (config DB schema)
/// materializes these from the <c>ScriptedAlarm</c> + <c>Script</c> tables on publish.
/// </summary>
/// <param name="AlarmId">
/// Stable identity for the alarm — used as the OPC UA ConditionId + the key in the
/// state store. Should be globally unique within the cluster; convention is
/// <c>{EquipmentPath}::{AlarmName}</c>.
/// </param>
/// <param name="EquipmentPath">
/// UNS path of the Equipment node the alarm hangs under. Alarm browse lives here;
/// ACL binding inherits this equipment's scope per Phase 6.2.
/// </param>
/// <param name="AlarmName">Human-readable alarm name — used in the browse tree + Admin UI.</param>
/// <param name="Kind">Concrete OPC UA Part 9 subtype the alarm materializes as.</param>
/// <param name="Severity">Static severity per Phase 7 plan decision #13; not currently computed by the predicate.</param>
/// <param name="MessageTemplate">
/// Message text with <c>{TagPath}</c> tokens resolved at event-emission time per
/// Phase 7 plan decision #13. Unresolvable tokens emit <c>{?}</c> + a structured
/// error so operators can spot stale references.
/// </param>
/// <param name="PredicateScriptSource">
/// Roslyn C# script returning <c>bool</c>. <c>true</c> = alarm condition currently holds (active);
/// <c>false</c> = condition has cleared. Same sandbox rules as virtual tags per Phase 7 decision #6.
/// </param>
/// <param name="HistorizeToAveva">
/// When true, every transition emission of this alarm flows to the Historian alarm
/// sink (Stream D). Defaults to true — plant alarm history is usually the
/// operator's primary diagnostic. Galaxy-native alarms default false since Galaxy
/// historises them directly.
/// </param>
/// <param name="Retain">
/// Part 9 retain flag — when true, the condition node remains visible after the
/// predicate clears as long as it has un-acknowledged or un-confirmed transitions.
/// Default true.
/// </param>
public sealed record ScriptedAlarmDefinition(
string AlarmId,
string EquipmentPath,
string AlarmName,
AlarmKind Kind,
AlarmSeverity Severity,
string MessageTemplate,
string PredicateScriptSource,
bool HistorizeToAveva = true,
bool Retain = true);

View File

@@ -0,0 +1,429 @@
using System.Collections.Concurrent;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
/// <summary>
/// Phase 7 scripted-alarm orchestrator. Compiles every configured alarm's predicate
/// against the Stream A sandbox, subscribes to the referenced upstream tags,
/// re-evaluates the predicate on every input change + on a shelving-check timer,
/// applies the resulting transition through <see cref="Part9StateMachine"/>,
/// persists state via <see cref="IAlarmStateStore"/>, and emits the resulting events
/// through <see cref="ScriptedAlarmSource"/> (which wires into the existing
/// <c>IAlarmSource</c> fan-out).
/// </summary>
/// <remarks>
/// <para>
/// Scripted alarms are leaves in the evaluation DAG — no alarm's state drives
/// another alarm's predicate. The engine maintains only an inverse index from
/// upstream tag path → alarms referencing it; no topological sort needed
/// (unlike the virtual-tag engine).
/// </para>
/// <para>
/// Evaluation errors (script throws, timeout, coercion fail) surface as
/// structured errors in the dedicated scripts-*.log sink plus a WARN companion
/// in the main log. The alarm's ActiveState stays at its prior value — the
/// engine does NOT invent a clear transition just because the predicate broke.
/// Operators investigating a broken predicate shouldn't see a phantom
/// clear-event preceding the failure.
/// </para>
/// </remarks>
public sealed class ScriptedAlarmEngine : IDisposable
{
private readonly ITagUpstreamSource _upstream;
private readonly IAlarmStateStore _store;
private readonly ScriptLoggerFactory _loggerFactory;
private readonly ILogger _engineLogger;
private readonly Func<DateTime> _clock;
private readonly TimeSpan _scriptTimeout;
private readonly Dictionary<string, AlarmState> _alarms = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, DataValueSnapshot> _valueCache
= new(StringComparer.Ordinal);
private readonly Dictionary<string, HashSet<string>> _alarmsReferencing
= new(StringComparer.Ordinal); // tag path -> alarm ids
private readonly List<IDisposable> _upstreamSubscriptions = [];
private readonly SemaphoreSlim _evalGate = new(1, 1);
private Timer? _shelvingTimer;
private bool _loaded;
private bool _disposed;
public ScriptedAlarmEngine(
ITagUpstreamSource upstream,
IAlarmStateStore store,
ScriptLoggerFactory loggerFactory,
ILogger engineLogger,
Func<DateTime>? clock = null,
TimeSpan? scriptTimeout = null)
{
_upstream = upstream ?? throw new ArgumentNullException(nameof(upstream));
_store = store ?? throw new ArgumentNullException(nameof(store));
_loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory));
_engineLogger = engineLogger ?? throw new ArgumentNullException(nameof(engineLogger));
_clock = clock ?? (() => DateTime.UtcNow);
_scriptTimeout = scriptTimeout ?? TimedScriptEvaluator<AlarmPredicateContext, bool>.DefaultTimeout;
}
/// <summary>Raised for every emission the Part9StateMachine produces that the engine should publish.</summary>
public event EventHandler<ScriptedAlarmEvent>? OnEvent;
public IReadOnlyCollection<string> LoadedAlarmIds => _alarms.Keys;
/// <summary>
/// Load a batch of alarm definitions. Compiles every predicate, aggregates any
/// compile failures into one <see cref="InvalidOperationException"/>, subscribes
/// to upstream input tags, seeds the value cache, loads persisted state from
/// the store (falling back to Fresh for first-load alarms), and recomputes
/// ActiveState per Phase 7 plan decision #14 (startup recovery).
/// </summary>
public async Task LoadAsync(IReadOnlyList<ScriptedAlarmDefinition> definitions, CancellationToken ct)
{
if (_disposed) throw new ObjectDisposedException(nameof(ScriptedAlarmEngine));
if (definitions is null) throw new ArgumentNullException(nameof(definitions));
await _evalGate.WaitAsync(ct).ConfigureAwait(false);
try
{
UnsubscribeFromUpstream();
_alarms.Clear();
_alarmsReferencing.Clear();
var compileFailures = new List<string>();
foreach (var def in definitions)
{
try
{
var extraction = DependencyExtractor.Extract(def.PredicateScriptSource);
if (!extraction.IsValid)
{
var joined = string.Join("; ", extraction.Rejections.Select(r => r.Message));
compileFailures.Add($"{def.AlarmId}: dependency extraction rejected — {joined}");
continue;
}
var evaluator = ScriptEvaluator<AlarmPredicateContext, bool>.Compile(def.PredicateScriptSource);
var timed = new TimedScriptEvaluator<AlarmPredicateContext, bool>(evaluator, _scriptTimeout);
var logger = _loggerFactory.Create(def.AlarmId);
var templateTokens = MessageTemplate.ExtractTokenPaths(def.MessageTemplate);
var allInputs = new HashSet<string>(extraction.Reads, StringComparer.Ordinal);
foreach (var t in templateTokens) allInputs.Add(t);
_alarms[def.AlarmId] = new AlarmState(def, timed, extraction.Reads, templateTokens, logger,
AlarmConditionState.Fresh(def.AlarmId, _clock()));
foreach (var path in allInputs)
{
if (!_alarmsReferencing.TryGetValue(path, out var set))
_alarmsReferencing[path] = set = new HashSet<string>(StringComparer.Ordinal);
set.Add(def.AlarmId);
}
}
catch (Exception ex)
{
compileFailures.Add($"{def.AlarmId}: {ex.Message}");
}
}
if (compileFailures.Count > 0)
{
throw new InvalidOperationException(
$"ScriptedAlarmEngine load failed. {compileFailures.Count} alarm(s) did not compile:\n "
+ string.Join("\n ", compileFailures));
}
// Seed the value cache with current upstream values + subscribe for changes.
foreach (var path in _alarmsReferencing.Keys)
{
_valueCache[path] = _upstream.ReadTag(path);
_upstreamSubscriptions.Add(_upstream.SubscribeTag(path, OnUpstreamChange));
}
// Restore persisted state, falling back to Fresh where nothing was saved,
// then re-derive ActiveState from the current predicate per decision #14.
foreach (var (alarmId, state) in _alarms)
{
var persisted = await _store.LoadAsync(alarmId, ct).ConfigureAwait(false);
var seed = persisted ?? state.Condition;
var afterPredicate = await EvaluatePredicateToStateAsync(state, seed, nowUtc: _clock(), ct)
.ConfigureAwait(false);
_alarms[alarmId] = state with { Condition = afterPredicate };
await _store.SaveAsync(afterPredicate, ct).ConfigureAwait(false);
}
_loaded = true;
_engineLogger.Information("ScriptedAlarmEngine loaded {Count} alarm(s)", _alarms.Count);
// Start the shelving-check timer — ticks every 5s, expires any timed shelves
// that have passed their UnshelveAtUtc.
_shelvingTimer = new Timer(_ => RunShelvingCheck(),
null, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(5));
}
finally
{
_evalGate.Release();
}
}
/// <summary>
/// Current persisted state for <paramref name="alarmId"/>. Returns null for
/// unknown alarm. Mainly used for diagnostics + the Admin UI status page.
/// </summary>
public AlarmConditionState? GetState(string alarmId)
=> _alarms.TryGetValue(alarmId, out var s) ? s.Condition : null;
public IReadOnlyCollection<AlarmConditionState> GetAllStates()
=> _alarms.Values.Select(a => a.Condition).ToArray();
public Task AcknowledgeAsync(string alarmId, string user, string? comment, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyAcknowledge(cur, user, comment, _clock()));
public Task ConfirmAsync(string alarmId, string user, string? comment, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyConfirm(cur, user, comment, _clock()));
public Task OneShotShelveAsync(string alarmId, string user, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyOneShotShelve(cur, user, _clock()));
public Task TimedShelveAsync(string alarmId, string user, DateTime unshelveAtUtc, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyTimedShelve(cur, user, unshelveAtUtc, _clock()));
public Task UnshelveAsync(string alarmId, string user, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyUnshelve(cur, user, _clock()));
public Task EnableAsync(string alarmId, string user, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyEnable(cur, user, _clock()));
public Task DisableAsync(string alarmId, string user, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyDisable(cur, user, _clock()));
public Task AddCommentAsync(string alarmId, string user, string text, CancellationToken ct)
=> ApplyAsync(alarmId, ct, cur => Part9StateMachine.ApplyAddComment(cur, user, text, _clock()));
private async Task ApplyAsync(string alarmId, CancellationToken ct, Func<AlarmConditionState, TransitionResult> op)
{
EnsureLoaded();
if (!_alarms.TryGetValue(alarmId, out var state))
throw new ArgumentException($"Unknown alarm {alarmId}", nameof(alarmId));
await _evalGate.WaitAsync(ct).ConfigureAwait(false);
try
{
var result = op(state.Condition);
_alarms[alarmId] = state with { Condition = result.State };
await _store.SaveAsync(result.State, ct).ConfigureAwait(false);
if (result.Emission != EmissionKind.None) EmitEvent(state, result.State, result.Emission);
}
finally { _evalGate.Release(); }
}
/// <summary>
/// Upstream-change callback. Updates the value cache + enqueues predicate
/// re-evaluation for every alarm referencing the changed path. Fire-and-forget
/// so driver-side dispatch isn't blocked.
/// </summary>
internal void OnUpstreamChange(string path, DataValueSnapshot value)
{
_valueCache[path] = value;
if (_alarmsReferencing.TryGetValue(path, out var alarmIds))
{
_ = ReevaluateAsync(alarmIds.ToArray(), CancellationToken.None);
}
}
private async Task ReevaluateAsync(IReadOnlyList<string> alarmIds, CancellationToken ct)
{
try
{
await _evalGate.WaitAsync(ct).ConfigureAwait(false);
try
{
foreach (var id in alarmIds)
{
if (!_alarms.TryGetValue(id, out var state)) continue;
var newState = await EvaluatePredicateToStateAsync(
state, state.Condition, _clock(), ct).ConfigureAwait(false);
if (!ReferenceEquals(newState, state.Condition))
{
_alarms[id] = state with { Condition = newState };
await _store.SaveAsync(newState, ct).ConfigureAwait(false);
}
}
}
finally { _evalGate.Release(); }
}
catch (Exception ex)
{
_engineLogger.Error(ex, "ScriptedAlarmEngine reevaluate failed");
}
}
/// <summary>
/// Evaluate the predicate + apply the resulting state-machine transition.
/// Returns the new condition state. Emits the appropriate event if the
/// transition produces one.
/// </summary>
private async Task<AlarmConditionState> EvaluatePredicateToStateAsync(
AlarmState state, AlarmConditionState seed, DateTime nowUtc, CancellationToken ct)
{
var inputs = BuildReadCache(state.Inputs);
var context = new AlarmPredicateContext(inputs, state.Logger, _clock);
bool predicateTrue;
try
{
predicateTrue = await state.Evaluator.RunAsync(context, ct).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
throw;
}
catch (ScriptTimeoutException tex)
{
state.Logger.Warning("Alarm predicate timed out after {Timeout} — state unchanged", tex.Timeout);
return seed;
}
catch (Exception ex)
{
state.Logger.Error(ex, "Alarm predicate threw — state unchanged");
return seed;
}
var result = Part9StateMachine.ApplyPredicate(seed, predicateTrue, nowUtc);
if (result.Emission != EmissionKind.None)
EmitEvent(state, result.State, result.Emission);
return result.State;
}
private IReadOnlyDictionary<string, DataValueSnapshot> BuildReadCache(IReadOnlySet<string> inputs)
{
var d = new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal);
foreach (var p in inputs)
d[p] = _valueCache.TryGetValue(p, out var v) ? v : _upstream.ReadTag(p);
return d;
}
private void EmitEvent(AlarmState state, AlarmConditionState condition, EmissionKind kind)
{
// Suppressed kind means shelving ate the emission — we don't fire for subscribers
// but the state record still advanced so startup recovery reflects reality.
if (kind == EmissionKind.Suppressed || kind == EmissionKind.None) return;
var message = MessageTemplate.Resolve(state.Definition.MessageTemplate, TryLookup);
var evt = new ScriptedAlarmEvent(
AlarmId: state.Definition.AlarmId,
EquipmentPath: state.Definition.EquipmentPath,
AlarmName: state.Definition.AlarmName,
Kind: state.Definition.Kind,
Severity: state.Definition.Severity,
Message: message,
Condition: condition,
Emission: kind,
TimestampUtc: _clock());
try { OnEvent?.Invoke(this, evt); }
catch (Exception ex)
{
_engineLogger.Warning(ex, "ScriptedAlarmEngine OnEvent subscriber threw for {AlarmId}", state.Definition.AlarmId);
}
}
private DataValueSnapshot? TryLookup(string path)
=> _valueCache.TryGetValue(path, out var v) ? v : null;
private void RunShelvingCheck()
{
if (_disposed) return;
var ids = _alarms.Keys.ToArray();
_ = ShelvingCheckAsync(ids, CancellationToken.None);
}
private async Task ShelvingCheckAsync(IReadOnlyList<string> alarmIds, CancellationToken ct)
{
try
{
await _evalGate.WaitAsync(ct).ConfigureAwait(false);
try
{
var now = _clock();
foreach (var id in alarmIds)
{
if (!_alarms.TryGetValue(id, out var state)) continue;
var result = Part9StateMachine.ApplyShelvingCheck(state.Condition, now);
if (!ReferenceEquals(result.State, state.Condition))
{
_alarms[id] = state with { Condition = result.State };
await _store.SaveAsync(result.State, ct).ConfigureAwait(false);
if (result.Emission != EmissionKind.None)
EmitEvent(state, result.State, result.Emission);
}
}
}
finally { _evalGate.Release(); }
}
catch (Exception ex)
{
_engineLogger.Warning(ex, "ScriptedAlarmEngine shelving-check failed");
}
}
private void UnsubscribeFromUpstream()
{
foreach (var s in _upstreamSubscriptions)
{
try { s.Dispose(); } catch { }
}
_upstreamSubscriptions.Clear();
}
private void EnsureLoaded()
{
if (!_loaded) throw new InvalidOperationException(
"ScriptedAlarmEngine not loaded. Call LoadAsync first.");
}
public void Dispose()
{
if (_disposed) return;
_disposed = true;
_shelvingTimer?.Dispose();
UnsubscribeFromUpstream();
_alarms.Clear();
_alarmsReferencing.Clear();
}
private sealed record AlarmState(
ScriptedAlarmDefinition Definition,
TimedScriptEvaluator<AlarmPredicateContext, bool> Evaluator,
IReadOnlySet<string> Inputs,
IReadOnlyList<string> TemplateTokens,
ILogger Logger,
AlarmConditionState Condition);
}
/// <summary>
/// One alarm emission the engine pushed to subscribers. Carries everything
/// downstream consumers (OPC UA alarm-source adapter + historian sink) need to
/// publish the event without re-querying the engine.
/// </summary>
public sealed record ScriptedAlarmEvent(
string AlarmId,
string EquipmentPath,
string AlarmName,
AlarmKind Kind,
AlarmSeverity Severity,
string Message,
AlarmConditionState Condition,
EmissionKind Emission,
DateTime TimestampUtc);
/// <summary>
/// Upstream source abstraction — intentionally identical shape to the virtual-tag
/// engine's so Stream G can compose them behind one driver bridge.
/// </summary>
public interface ITagUpstreamSource
{
DataValueSnapshot ReadTag(string path);
IDisposable SubscribeTag(string path, Action<string, DataValueSnapshot> observer);
}

View File

@@ -0,0 +1,122 @@
using System.Collections.Concurrent;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
/// <summary>
/// Adapter that exposes <see cref="ScriptedAlarmEngine"/> through the driver-agnostic
/// <see cref="IAlarmSource"/> surface. The existing Phase 6.1 <c>AlarmTracker</c>
/// composition fan-out consumes this alongside Galaxy / AB CIP / FOCAS alarm
/// sources — no per-source branching in the fan-out.
/// </summary>
/// <remarks>
/// <para>
/// Per Phase 7 plan Stream C.6, ack / confirm / shelve / unshelve are OPC UA
/// method calls per-condition. This adapter implements <see cref="AcknowledgeAsync"/>
/// from the base interface; the richer Part 9 methods (Confirm / Shelve /
/// Unshelve / AddComment) live directly on the engine, invoked from OPC UA
/// method handlers wired up in Stream G.
/// </para>
/// <para>
/// SubscribeAlarmsAsync takes a list of source-node-id filters (typically an
/// Equipment path prefix). When the list is empty every alarm matches. The
/// adapter doesn't maintain per-subscription state beyond the filter set — it
/// checks each emission against every live subscription.
/// </para>
/// </remarks>
public sealed class ScriptedAlarmSource : IAlarmSource, IDisposable
{
private readonly ScriptedAlarmEngine _engine;
private readonly ConcurrentDictionary<string, Subscription> _subscriptions
= new(StringComparer.Ordinal);
private bool _disposed;
public ScriptedAlarmSource(ScriptedAlarmEngine engine)
{
_engine = engine ?? throw new ArgumentNullException(nameof(engine));
_engine.OnEvent += OnEngineEvent;
}
public event EventHandler<AlarmEventArgs>? OnAlarmEvent;
public Task<IAlarmSubscriptionHandle> SubscribeAlarmsAsync(
IReadOnlyList<string> sourceNodeIds, CancellationToken cancellationToken)
{
if (sourceNodeIds is null) throw new ArgumentNullException(nameof(sourceNodeIds));
var handle = new SubscriptionHandle(Guid.NewGuid().ToString("N"));
_subscriptions[handle.DiagnosticId] = new Subscription(handle,
new HashSet<string>(sourceNodeIds, StringComparer.Ordinal));
return Task.FromResult<IAlarmSubscriptionHandle>(handle);
}
public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken)
{
if (handle is null) throw new ArgumentNullException(nameof(handle));
_subscriptions.TryRemove(handle.DiagnosticId, out _);
return Task.CompletedTask;
}
public async Task AcknowledgeAsync(
IReadOnlyList<AlarmAcknowledgeRequest> acknowledgements, CancellationToken cancellationToken)
{
if (acknowledgements is null) throw new ArgumentNullException(nameof(acknowledgements));
foreach (var a in acknowledgements)
{
// The base interface doesn't carry a user identity — Stream G provides the
// authenticated principal at the OPC UA dispatch layer + proxies through
// the engine's richer AcknowledgeAsync. Here we default to "opcua-client"
// so callers using the raw IAlarmSource still produce an audit entry.
await _engine.AcknowledgeAsync(a.ConditionId, "opcua-client", a.Comment, cancellationToken)
.ConfigureAwait(false);
}
}
private void OnEngineEvent(object? sender, ScriptedAlarmEvent evt)
{
if (_disposed) return;
foreach (var sub in _subscriptions.Values)
{
if (!Matches(sub, evt)) continue;
var payload = new AlarmEventArgs(
SubscriptionHandle: sub.Handle,
SourceNodeId: evt.EquipmentPath,
ConditionId: evt.AlarmId,
AlarmType: evt.Kind.ToString(),
Message: evt.Message,
Severity: evt.Severity,
SourceTimestampUtc: evt.TimestampUtc);
try { OnAlarmEvent?.Invoke(this, payload); }
catch { /* subscriber exceptions don't crash the adapter */ }
}
}
private static bool Matches(Subscription sub, ScriptedAlarmEvent evt)
{
if (sub.Filter.Count == 0) return true;
// A subscription matches if any filter is a prefix of the alarm's equipment
// path — typical use is "Enterprise/Site/Area/Line" filtering a whole line.
foreach (var f in sub.Filter)
{
if (evt.EquipmentPath.Equals(f, StringComparison.Ordinal)) return true;
if (evt.EquipmentPath.StartsWith(f + "/", StringComparison.Ordinal)) return true;
}
return false;
}
public void Dispose()
{
if (_disposed) return;
_disposed = true;
_engine.OnEvent -= OnEngineEvent;
_subscriptions.Clear();
}
private sealed class SubscriptionHandle : IAlarmSubscriptionHandle
{
public SubscriptionHandle(string id) { DiagnosticId = id; }
public string DiagnosticId { get; }
}
private sealed record Subscription(SubscriptionHandle Handle, IReadOnlySet<string> Filter);
}

View File

@@ -0,0 +1,32 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<LangVersion>latest</LangVersion>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
<NoWarn>$(NoWarn);CS1591</NoWarn>
<RootNamespace>ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms</RootNamespace>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Serilog" Version="4.2.0"/>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Core.Abstractions\ZB.MOM.WW.OtOpcUa.Core.Abstractions.csproj"/>
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Core.Scripting\ZB.MOM.WW.OtOpcUa.Core.Scripting.csproj"/>
</ItemGroup>
<ItemGroup>
<InternalsVisibleTo Include="ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests"/>
</ItemGroup>
<ItemGroup>
<NuGetAuditSuppress Include="https://github.com/advisories/GHSA-37gx-xxp4-5rgx"/>
<NuGetAuditSuppress Include="https://github.com/advisories/GHSA-w3x6-4m5h-cxqf"/>
</ItemGroup>
</Project>

View File

@@ -0,0 +1,271 @@
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
/// <summary>
/// Directed dependency graph over tag paths. Nodes are tag paths (either driver
/// tags — leaves — or virtual tags — internal nodes). Edges run from a virtual tag
/// to each tag it reads via <c>ctx.GetTag(...)</c>. Supports cycle detection at
/// publish time and topological sort for evaluation ordering.
/// </summary>
/// <remarks>
/// <para>
/// Cycle detection uses Tarjan's strongly-connected-components algorithm,
/// iterative implementation (no recursion) so deeply-nested graphs can't blow
/// the stack. A cycle of length > 1 (or a self-loop) is a publish-time error;
/// the engine refuses to load such a config.
/// </para>
/// <para>
/// Topological sort uses Kahn's algorithm. The output order guarantees that when
/// tag X depends on tag Y, Y appears before X — so a change cascade starting at
/// Y can evaluate the full downstream closure in one serial pass without needing
/// a second iteration.
/// </para>
/// <para>
/// Missing leaf dependencies (a virtual tag reads a driver tag that doesn't
/// exist in the live config) are NOT rejected here — the graph treats any
/// unregistered path as an implicit leaf. Leaf validity is a separate concern
/// handled at engine-load time against the authoritative tag catalog.
/// </para>
/// </remarks>
public sealed class DependencyGraph
{
private readonly Dictionary<string, HashSet<string>> _dependsOn = new(StringComparer.Ordinal);
private readonly Dictionary<string, HashSet<string>> _dependents = new(StringComparer.Ordinal);
/// <summary>
/// Register a node and the set of tags it depends on. Idempotent — re-adding
/// the same node overwrites the prior dependency set, so re-publishing an edited
/// script works without a separate "remove" call.
/// </summary>
public void Add(string nodeId, IReadOnlySet<string> dependsOn)
{
if (string.IsNullOrWhiteSpace(nodeId)) throw new ArgumentException("Node id required.", nameof(nodeId));
if (dependsOn is null) throw new ArgumentNullException(nameof(dependsOn));
// Remove any prior dependents pointing at the previous version of this node.
if (_dependsOn.TryGetValue(nodeId, out var previous))
{
foreach (var dep in previous)
{
if (_dependents.TryGetValue(dep, out var set))
set.Remove(nodeId);
}
}
_dependsOn[nodeId] = new HashSet<string>(dependsOn, StringComparer.Ordinal);
foreach (var dep in dependsOn)
{
if (!_dependents.TryGetValue(dep, out var set))
_dependents[dep] = set = new HashSet<string>(StringComparer.Ordinal);
set.Add(nodeId);
}
}
/// <summary>Tag paths <paramref name="nodeId"/> directly reads.</summary>
public IReadOnlySet<string> DirectDependencies(string nodeId) =>
_dependsOn.TryGetValue(nodeId, out var set) ? set : (IReadOnlySet<string>)new HashSet<string>();
/// <summary>
/// Tags whose evaluation depends on <paramref name="nodeId"/> — i.e. when
/// <paramref name="nodeId"/> changes, these need to re-evaluate. Direct only;
/// transitive propagation falls out of the topological sort.
/// </summary>
public IReadOnlySet<string> DirectDependents(string nodeId) =>
_dependents.TryGetValue(nodeId, out var set) ? set : (IReadOnlySet<string>)new HashSet<string>();
/// <summary>
/// Full transitive dependent closure of <paramref name="nodeId"/> in topological
/// order (direct dependents first, then their dependents, and so on). Used by the
/// change-trigger dispatcher to schedule the right sequence of re-evaluations
/// when a single upstream value changes.
/// </summary>
public IReadOnlyList<string> TransitiveDependentsInOrder(string nodeId)
{
if (string.IsNullOrWhiteSpace(nodeId)) return [];
var result = new List<string>();
var visited = new HashSet<string>(StringComparer.Ordinal);
var order = TopologicalSort();
var rank = new Dictionary<string, int>(StringComparer.Ordinal);
for (var i = 0; i < order.Count; i++) rank[order[i]] = i;
// DFS from the changed node collecting every reachable dependent.
var stack = new Stack<string>();
stack.Push(nodeId);
while (stack.Count > 0)
{
var cur = stack.Pop();
foreach (var dep in DirectDependents(cur))
{
if (visited.Add(dep))
{
result.Add(dep);
stack.Push(dep);
}
}
}
// Sort by topological rank so when re-evaluation runs serial, earlier entries
// are computed before later entries that might depend on them.
result.Sort((a, b) =>
{
var ra = rank.TryGetValue(a, out var va) ? va : int.MaxValue;
var rb = rank.TryGetValue(b, out var vb) ? vb : int.MaxValue;
return ra.CompareTo(rb);
});
return result;
}
/// <summary>Iterable of every registered node id (inputs-only tags excluded).</summary>
public IReadOnlyCollection<string> RegisteredNodes => _dependsOn.Keys;
/// <summary>
/// Produce an evaluation order where every node appears after all its
/// dependencies. Throws <see cref="DependencyCycleException"/> if any cycle
/// exists. Implemented via Kahn's algorithm.
/// </summary>
public IReadOnlyList<string> TopologicalSort()
{
// Kahn's framing: edge u -> v means "u must come before v". For dependencies,
// if X depends on Y, Y must come before X, so the edge runs Y -> X and X has
// an incoming edge from Y. inDegree[X] = count of X's registered (virtual) deps
// — leaf driver-tag deps don't contribute to ordering since they're never emitted.
var inDegree = new Dictionary<string, int>(StringComparer.Ordinal);
foreach (var node in _dependsOn.Keys) inDegree[node] = 0;
foreach (var kv in _dependsOn)
{
var nodeId = kv.Key;
foreach (var dep in kv.Value)
{
if (_dependsOn.ContainsKey(dep))
inDegree[nodeId]++;
}
}
var ready = new Queue<string>(inDegree.Where(kv => kv.Value == 0).Select(kv => kv.Key));
var result = new List<string>();
while (ready.Count > 0)
{
var n = ready.Dequeue();
result.Add(n);
// In our edge direction (node -> deps), removing n means decrementing in-degree
// of every node that DEPENDS on n.
foreach (var dependent in DirectDependents(n))
{
if (inDegree.TryGetValue(dependent, out var d))
{
inDegree[dependent] = d - 1;
if (inDegree[dependent] == 0) ready.Enqueue(dependent);
}
}
}
if (result.Count != inDegree.Count)
{
var cycles = DetectCycles();
throw new DependencyCycleException(cycles);
}
return result;
}
/// <summary>
/// Returns every strongly-connected component of size &gt; 1 + every self-loop.
/// Empty list means the graph is a DAG. Useful for surfacing every cycle in one
/// rejection pass so operators see all of them, not just one at a time.
/// </summary>
public IReadOnlyList<IReadOnlyList<string>> DetectCycles()
{
// Iterative Tarjan's SCC. Avoids recursion so deep graphs don't StackOverflow.
var index = 0;
var indexOf = new Dictionary<string, int>(StringComparer.Ordinal);
var lowlinkOf = new Dictionary<string, int>(StringComparer.Ordinal);
var onStack = new HashSet<string>(StringComparer.Ordinal);
var sccStack = new Stack<string>();
var cycles = new List<IReadOnlyList<string>>();
foreach (var root in _dependsOn.Keys)
{
if (indexOf.ContainsKey(root)) continue;
var work = new Stack<(string node, IEnumerator<string> iter)>();
indexOf[root] = index;
lowlinkOf[root] = index;
index++;
onStack.Add(root);
sccStack.Push(root);
work.Push((root, _dependsOn[root].GetEnumerator()));
while (work.Count > 0)
{
var (v, iter) = work.Peek();
if (iter.MoveNext())
{
var w = iter.Current;
if (!_dependsOn.ContainsKey(w))
continue; // leaf — not part of any cycle with us
if (!indexOf.ContainsKey(w))
{
indexOf[w] = index;
lowlinkOf[w] = index;
index++;
onStack.Add(w);
sccStack.Push(w);
work.Push((w, _dependsOn[w].GetEnumerator()));
}
else if (onStack.Contains(w))
{
lowlinkOf[v] = Math.Min(lowlinkOf[v], indexOf[w]);
}
}
else
{
// v fully explored — unwind
work.Pop();
if (lowlinkOf[v] == indexOf[v])
{
var component = new List<string>();
string w;
do
{
w = sccStack.Pop();
onStack.Remove(w);
component.Add(w);
} while (w != v);
if (component.Count > 1 || _dependsOn[v].Contains(v))
cycles.Add(component);
}
else if (work.Count > 0)
{
var parent = work.Peek().node;
lowlinkOf[parent] = Math.Min(lowlinkOf[parent], lowlinkOf[v]);
}
}
}
}
return cycles;
}
public void Clear()
{
_dependsOn.Clear();
_dependents.Clear();
}
}
/// <summary>Thrown when <see cref="DependencyGraph.TopologicalSort"/> finds one or more cycles.</summary>
public sealed class DependencyCycleException : Exception
{
public IReadOnlyList<IReadOnlyList<string>> Cycles { get; }
public DependencyCycleException(IReadOnlyList<IReadOnlyList<string>> cycles)
: base(BuildMessage(cycles))
{
Cycles = cycles;
}
private static string BuildMessage(IReadOnlyList<IReadOnlyList<string>> cycles)
{
var lines = cycles.Select(c => " - " + string.Join(" -> ", c) + " -> " + c[0]);
return "Virtual-tag dependency graph contains cycle(s):\n" + string.Join("\n", lines);
}
}

View File

@@ -0,0 +1,25 @@
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
/// <summary>
/// Sink for virtual-tag evaluation results that the operator marked
/// <c>Historize = true</c>. Stream G wires this to the existing history-write path
/// drivers use; tests inject a fake recorder.
/// </summary>
/// <remarks>
/// Emission is fire-and-forget from the evaluation pipeline — a slow historian must
/// not block script evaluations. Implementations queue internally and drain on their
/// own cadence.
/// </remarks>
public interface IHistoryWriter
{
void Record(string path, DataValueSnapshot value);
}
/// <summary>No-op default used when no historian is configured.</summary>
public sealed class NullHistoryWriter : IHistoryWriter
{
public static readonly NullHistoryWriter Instance = new();
public void Record(string path, DataValueSnapshot value) { }
}

View File

@@ -0,0 +1,40 @@
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
/// <summary>
/// What the virtual-tag engine pulls driver-tag values from. Implementations
/// shipped in Stream G bridge this to <see cref="IReadable"/> + <see cref="ISubscribable"/>
/// on the live driver instances; tests use an in-memory fake.
/// </summary>
/// <remarks>
/// <para>
/// The read path is synchronous because user scripts call
/// <c>ctx.GetTag(path)</c> inline — blocking on a driver wire call per-script
/// evaluation would kill throughput. Implementations are expected to serve
/// from a last-known-value cache populated by the subscription callbacks.
/// </para>
/// <para>
/// The subscription path feeds the engine's <c>ChangeTriggerDispatcher</c> so
/// change-driven virtual tags re-evaluate on any upstream delta (value, status,
/// or timestamp). One subscription per distinct upstream tag path; the engine
/// tracks the mapping itself.
/// </para>
/// </remarks>
public interface ITagUpstreamSource
{
/// <summary>
/// Synchronous read returning the last-known value + quality for
/// <paramref name="path"/>. Returns a <c>BadNodeIdUnknown</c>-quality snapshot
/// when the path isn't configured.
/// </summary>
DataValueSnapshot ReadTag(string path);
/// <summary>
/// Register an observer that fires every time the upstream value at
/// <paramref name="path"/> changes. Returns an <see cref="IDisposable"/> the
/// engine disposes when the virtual-tag config is reloaded or the engine shuts
/// down, so source-side subscriptions don't leak.
/// </summary>
IDisposable SubscribeTag(string path, Action<string, DataValueSnapshot> observer);
}

View File

@@ -0,0 +1,83 @@
using Serilog;
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
/// <summary>
/// Periodic re-evaluation scheduler for tags with a non-null
/// <see cref="VirtualTagDefinition.TimerInterval"/>. Independent of the
/// change-trigger path — a tag can be timer-only, change-only, or both. One
/// <see cref="System.Threading.Timer"/> per interval-group keeps the wire count
/// low regardless of tag count.
/// </summary>
public sealed class TimerTriggerScheduler : IDisposable
{
private readonly VirtualTagEngine _engine;
private readonly ILogger _logger;
private readonly List<Timer> _timers = [];
private readonly CancellationTokenSource _cts = new();
private bool _disposed;
public TimerTriggerScheduler(VirtualTagEngine engine, ILogger logger)
{
_engine = engine ?? throw new ArgumentNullException(nameof(engine));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
}
/// <summary>
/// Stand up one <see cref="Timer"/> per unique interval. All tags with
/// matching interval share a timer; each tick triggers re-evaluation of the
/// group in topological order so cascades are consistent with change-triggered
/// behavior.
/// </summary>
public void Start(IReadOnlyList<VirtualTagDefinition> definitions)
{
if (_disposed) throw new ObjectDisposedException(nameof(TimerTriggerScheduler));
var byInterval = definitions
.Where(d => d.TimerInterval.HasValue && d.TimerInterval.Value > TimeSpan.Zero)
.GroupBy(d => d.TimerInterval!.Value);
foreach (var group in byInterval)
{
var paths = group.Select(d => d.Path).ToArray();
var interval = group.Key;
var timer = new Timer(_ => Tick(paths), null, interval, interval);
_timers.Add(timer);
_logger.Information("TimerTriggerScheduler: {TagCount} tag(s) on {Interval} cadence",
paths.Length, interval);
}
}
private void Tick(IReadOnlyList<string> paths)
{
if (_cts.IsCancellationRequested) return;
foreach (var p in paths)
{
try
{
_engine.EvaluateOneAsync(p, _cts.Token).GetAwaiter().GetResult();
}
catch (OperationCanceledException)
{
return;
}
catch (Exception ex)
{
_logger.Error(ex, "TimerTriggerScheduler evaluate failed for {Path}", p);
}
}
}
public void Dispose()
{
if (_disposed) return;
_disposed = true;
_cts.Cancel();
foreach (var t in _timers)
{
try { t.Dispose(); } catch { }
}
_timers.Clear();
_cts.Dispose();
}
}

View File

@@ -0,0 +1,64 @@
using Serilog;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
/// <summary>
/// Per-evaluation <see cref="ScriptContext"/> for a virtual-tag script. Reads come
/// out of the engine's last-known-value cache (driver tags updated via the
/// <see cref="ITagUpstreamSource"/> subscription, virtual tags updated by prior
/// evaluations). Writes route through the engine's <c>SetVirtualTag</c> callback so
/// cross-tag write side effects still participate in change-trigger cascades.
/// </summary>
/// <remarks>
/// <para>
/// Context instances are evaluation-scoped, not tag-scoped. The engine
/// constructs a fresh context for every run — cheap because the constructor
/// just captures references — so scripts can't cache mutable state across runs
/// via <c>ctx</c>. Mutable state across runs is a future decision (e.g. a
/// dedicated <c>ctx.Memory</c> dictionary); not in scope for Phase 7.
/// </para>
/// <para>
/// The <see cref="Now"/> clock is injectable so tests can pin time
/// deterministically. Production wires to <see cref="DateTime.UtcNow"/>.
/// </para>
/// </remarks>
public sealed class VirtualTagContext : ScriptContext
{
private readonly IReadOnlyDictionary<string, DataValueSnapshot> _readCache;
private readonly Action<string, object?> _setVirtualTag;
private readonly Func<DateTime> _clock;
public VirtualTagContext(
IReadOnlyDictionary<string, DataValueSnapshot> readCache,
Action<string, object?> setVirtualTag,
ILogger logger,
Func<DateTime>? clock = null)
{
_readCache = readCache ?? throw new ArgumentNullException(nameof(readCache));
_setVirtualTag = setVirtualTag ?? throw new ArgumentNullException(nameof(setVirtualTag));
Logger = logger ?? throw new ArgumentNullException(nameof(logger));
_clock = clock ?? (() => DateTime.UtcNow);
}
public override DataValueSnapshot GetTag(string path)
{
if (string.IsNullOrWhiteSpace(path))
return new DataValueSnapshot(null, 0x80340000u /* BadNodeIdUnknown */, null, _clock());
return _readCache.TryGetValue(path, out var v)
? v
: new DataValueSnapshot(null, 0x80340000u /* BadNodeIdUnknown */, null, _clock());
}
public override void SetVirtualTag(string path, object? value)
{
if (string.IsNullOrWhiteSpace(path))
throw new ArgumentException("Virtual tag path required.", nameof(path));
_setVirtualTag(path, value);
}
public override DateTime Now => _clock();
public override ILogger Logger { get; }
}

View File

@@ -0,0 +1,41 @@
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
/// <summary>
/// Operator-authored virtual-tag configuration row. Phase 7 Stream E (config DB
/// schema) materializes these from the <c>VirtualTag</c> + <c>Script</c> tables on
/// publish; the engine ingests a list of them at load time.
/// </summary>
/// <param name="Path">
/// UNS tag path — <c>Enterprise/Site/Area/Line/Equipment/TagName</c>. Used both as
/// the engine's internal id and the OPC UA browse path.
/// </param>
/// <param name="DataType">
/// Expected return type. The evaluator coerces the script's return value to this
/// type before publishing; mismatch surfaces as <c>BadTypeMismatch</c> quality on
/// the tag.
/// </param>
/// <param name="ScriptSource">Roslyn C# script source. Must compile under <c>ScriptSandbox</c>.</param>
/// <param name="ChangeTriggered">
/// True if any input tag's change (value / status / timestamp delta) should trigger
/// re-evaluation. Operator picks per tag — usually true for inputs that change at
/// protocol rates.
/// </param>
/// <param name="TimerInterval">
/// Optional periodic re-evaluation cadence. Null = timer-driven disabled. Both can
/// be enabled simultaneously; independent scheduling paths both feed
/// <c>EvaluationPipeline</c>.
/// </param>
/// <param name="Historize">
/// When true, every evaluation result is forwarded to the configured
/// <see cref="IHistoryWriter"/>. Operator-set per tag; the Admin UI exposes as a
/// checkbox.
/// </param>
public sealed record VirtualTagDefinition(
string Path,
DriverDataType DataType,
string ScriptSource,
bool ChangeTriggered = true,
TimeSpan? TimerInterval = null,
bool Historize = false);

View File

@@ -0,0 +1,385 @@
using System.Collections.Concurrent;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
/// <summary>
/// The Phase 7 virtual-tag evaluation engine. Ingests a set of
/// <see cref="VirtualTagDefinition"/>s at load time, compiles each script against
/// <see cref="ScriptSandbox"/>, builds the dependency graph, subscribes to every
/// referenced upstream tag, and schedules re-evaluations on change + on timer.
/// </summary>
/// <remarks>
/// <para>
/// Evaluation order is topological per ADR-001 / Phase 7 plan decision #19 —
/// serial for the v1 rollout, parallel promoted to a follow-up. When upstream
/// tag X changes, the engine computes the transitive dependent closure of X in
/// topological rank and evaluates each in turn, so a cascade through multiple
/// levels of virtual tags settles within one change-trigger pass.
/// </para>
/// <para>
/// Per-tag error isolation per Phase 7 plan decision #11 — a script exception
/// (or timeout) fails that tag's latest value with <c>BadInternalError</c> or
/// <c>BadTypeMismatch</c> quality and logs a structured error; every other tag
/// keeps evaluating. The engine itself never faults from a user script.
/// </para>
/// </remarks>
public sealed class VirtualTagEngine : IDisposable
{
private readonly ITagUpstreamSource _upstream;
private readonly IHistoryWriter _history;
private readonly ScriptLoggerFactory _loggerFactory;
private readonly ILogger _engineLogger;
private readonly Func<DateTime> _clock;
private readonly TimeSpan _scriptTimeout;
private readonly DependencyGraph _graph = new();
private readonly Dictionary<string, VirtualTagState> _tags = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, DataValueSnapshot> _valueCache = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, List<Action<string, DataValueSnapshot>>> _observers
= new(StringComparer.Ordinal);
private readonly List<IDisposable> _upstreamSubscriptions = [];
private readonly SemaphoreSlim _evalGate = new(1, 1);
private bool _loaded;
private bool _disposed;
public VirtualTagEngine(
ITagUpstreamSource upstream,
ScriptLoggerFactory loggerFactory,
ILogger engineLogger,
IHistoryWriter? historyWriter = null,
Func<DateTime>? clock = null,
TimeSpan? scriptTimeout = null)
{
_upstream = upstream ?? throw new ArgumentNullException(nameof(upstream));
_loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory));
_engineLogger = engineLogger ?? throw new ArgumentNullException(nameof(engineLogger));
_history = historyWriter ?? NullHistoryWriter.Instance;
_clock = clock ?? (() => DateTime.UtcNow);
_scriptTimeout = scriptTimeout ?? TimedScriptEvaluator<VirtualTagContext, object?>.DefaultTimeout;
}
/// <summary>Registered tag paths, in topological order. Empty before <see cref="Load"/>.</summary>
public IReadOnlyCollection<string> LoadedTagPaths => _tags.Keys;
/// <summary>Compile + register every tag in <paramref name="definitions"/>. Throws on cycle or any compile failure.</summary>
public void Load(IReadOnlyList<VirtualTagDefinition> definitions)
{
if (_disposed) throw new ObjectDisposedException(nameof(VirtualTagEngine));
if (definitions is null) throw new ArgumentNullException(nameof(definitions));
// Start from a clean slate — supports config-publish reloads.
UnsubscribeFromUpstream();
_tags.Clear();
_graph.Clear();
var compileFailures = new List<string>();
foreach (var def in definitions)
{
try
{
var extraction = DependencyExtractor.Extract(def.ScriptSource);
if (!extraction.IsValid)
{
var msgs = string.Join("; ", extraction.Rejections.Select(r => r.Message));
compileFailures.Add($"{def.Path}: dependency extraction rejected — {msgs}");
continue;
}
var evaluator = ScriptEvaluator<VirtualTagContext, object?>.Compile(def.ScriptSource);
var timed = new TimedScriptEvaluator<VirtualTagContext, object?>(evaluator, _scriptTimeout);
var scriptLogger = _loggerFactory.Create(def.Path);
_tags[def.Path] = new VirtualTagState(def, timed, extraction.Reads, extraction.Writes, scriptLogger);
_graph.Add(def.Path, extraction.Reads);
}
catch (Exception ex)
{
compileFailures.Add($"{def.Path}: {ex.Message}");
}
}
if (compileFailures.Count > 0)
{
var joined = string.Join("\n ", compileFailures);
throw new InvalidOperationException(
$"Virtual-tag engine load failed. {compileFailures.Count} script(s) did not compile:\n {joined}");
}
// Cycle check — throws DependencyCycleException on offense.
_ = _graph.TopologicalSort();
// Subscribe to every referenced upstream path (driver tags only — virtual tags
// cascade internally). Seed the cache with current upstream values so first
// evaluations see something real.
var upstreamPaths = definitions
.SelectMany(d => _tags[d.Path].Reads)
.Where(p => !_tags.ContainsKey(p))
.Distinct(StringComparer.Ordinal);
foreach (var path in upstreamPaths)
{
_valueCache[path] = _upstream.ReadTag(path);
_upstreamSubscriptions.Add(_upstream.SubscribeTag(path, OnUpstreamChange));
}
_loaded = true;
_engineLogger.Information(
"VirtualTagEngine loaded {TagCount} tag(s), {UpstreamCount} upstream subscription(s)",
_tags.Count, _upstreamSubscriptions.Count);
}
/// <summary>
/// Evaluate every registered tag once in topological order — used at startup so
/// virtual tags have a defined initial value rather than inheriting the cache
/// default. Also called after a config reload.
/// </summary>
public async Task EvaluateAllAsync(CancellationToken ct = default)
{
EnsureLoaded();
var order = _graph.TopologicalSort();
foreach (var path in order)
{
if (_tags.ContainsKey(path))
await EvaluateOneAsync(path, ct).ConfigureAwait(false);
}
}
/// <summary>Evaluate a single tag — used by the timer trigger + test hooks.</summary>
public Task EvaluateOneAsync(string path, CancellationToken ct = default)
{
EnsureLoaded();
if (!_tags.ContainsKey(path))
throw new ArgumentException($"Not a registered virtual tag: {path}", nameof(path));
return EvaluateInternalAsync(path, ct);
}
/// <summary>
/// Read the most recently evaluated value for <paramref name="path"/>. Driver
/// tags return the last-known upstream value; virtual tags return their last
/// evaluation result.
/// </summary>
public DataValueSnapshot Read(string path)
{
if (string.IsNullOrWhiteSpace(path))
return new DataValueSnapshot(null, 0x80340000u, null, _clock());
return _valueCache.TryGetValue(path, out var v)
? v
: new DataValueSnapshot(null, 0x80340000u /* BadNodeIdUnknown */, null, _clock());
}
/// <summary>
/// Register an observer that fires on every evaluation of the given tag.
/// Returns an <see cref="IDisposable"/> to unsubscribe. Does NOT fire a seed
/// value — subscribers call <see cref="Read"/> for the current value if needed.
/// </summary>
public IDisposable Subscribe(string path, Action<string, DataValueSnapshot> observer)
{
var list = _observers.GetOrAdd(path, _ => []);
lock (list) { list.Add(observer); }
return new Unsub(this, path, observer);
}
/// <summary>
/// Change-trigger entry point — called by the upstream subscription callback.
/// Updates the cache, fans out to observers (so OPC UA clients see the upstream
/// change too if they subscribed via the engine), and schedules every
/// change-triggered dependent for re-evaluation in topological order.
/// </summary>
internal void OnUpstreamChange(string path, DataValueSnapshot value)
{
_valueCache[path] = value;
NotifyObservers(path, value);
// Fire-and-forget — the upstream subscription callback must not block the
// driver's dispatcher. Exceptions during cascade are handled per-tag inside
// EvaluateInternalAsync.
_ = CascadeAsync(path, CancellationToken.None);
}
private async Task CascadeAsync(string upstreamPath, CancellationToken ct)
{
try
{
var dependents = _graph.TransitiveDependentsInOrder(upstreamPath);
foreach (var dep in dependents)
{
if (_tags.TryGetValue(dep, out var state) && state.Definition.ChangeTriggered)
await EvaluateInternalAsync(dep, ct).ConfigureAwait(false);
}
}
catch (Exception ex)
{
_engineLogger.Error(ex, "VirtualTagEngine cascade failed for upstream {Path}", upstreamPath);
}
}
private async Task EvaluateInternalAsync(string path, CancellationToken ct)
{
if (!_tags.TryGetValue(path, out var state)) return;
// Serial evaluation across all tags. Phase 7 plan decision #19 — parallel is a
// follow-up. The semaphore bounds the evaluation graph so two cascades don't
// interleave, which would break the "earlier nodes computed first" invariant.
// SemaphoreSlim.WaitAsync is async-safe where Monitor.Enter is not (Monitor
// ownership is thread-local and lost across await).
await _evalGate.WaitAsync(ct).ConfigureAwait(false);
try
{
var ctxCache = BuildReadCache(state.Reads);
var context = new VirtualTagContext(
ctxCache,
(p, v) => OnScriptSetVirtualTag(p, v),
state.Logger,
_clock);
DataValueSnapshot result;
try
{
var raw = await state.Evaluator.RunAsync(context, ct).ConfigureAwait(false);
var coerced = CoerceResult(raw, state.Definition.DataType);
result = new DataValueSnapshot(coerced, 0u, _clock(), _clock());
}
catch (ScriptTimeoutException tex)
{
state.Logger.Warning("Script timed out after {Timeout}", tex.Timeout);
result = new DataValueSnapshot(null, 0x80020000u /* BadInternalError */, null, _clock());
}
catch (OperationCanceledException)
{
throw; // shutdown path — don't misclassify
}
catch (Exception ex)
{
state.Logger.Error(ex, "Virtual-tag script threw");
result = new DataValueSnapshot(null, 0x80020000u /* BadInternalError */, null, _clock());
}
_valueCache[path] = result;
NotifyObservers(path, result);
if (state.Definition.Historize) _history.Record(path, result);
}
finally
{
_evalGate.Release();
}
}
private IReadOnlyDictionary<string, DataValueSnapshot> BuildReadCache(IReadOnlySet<string> reads)
{
var map = new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal);
foreach (var r in reads)
{
map[r] = _valueCache.TryGetValue(r, out var v)
? v
: _upstream.ReadTag(r);
}
return map;
}
private void OnScriptSetVirtualTag(string path, object? value)
{
if (!_tags.ContainsKey(path))
{
_engineLogger.Warning(
"Script attempted ctx.SetVirtualTag on non-virtual or non-registered path {Path}", path);
return;
}
var snap = new DataValueSnapshot(value, 0u, _clock(), _clock());
_valueCache[path] = snap;
NotifyObservers(path, snap);
if (_tags[path].Definition.Historize) _history.Record(path, snap);
}
private void NotifyObservers(string path, DataValueSnapshot value)
{
if (!_observers.TryGetValue(path, out var list)) return;
Action<string, DataValueSnapshot>[] snapshot;
lock (list) { snapshot = list.ToArray(); }
foreach (var obs in snapshot)
{
try { obs(path, value); }
catch (Exception ex)
{
_engineLogger.Warning(ex, "Virtual-tag observer for {Path} threw", path);
}
}
}
private static object? CoerceResult(object? raw, DriverDataType target)
{
if (raw is null) return null;
try
{
return target switch
{
DriverDataType.Boolean => Convert.ToBoolean(raw),
DriverDataType.Int32 => Convert.ToInt32(raw),
DriverDataType.Int64 => Convert.ToInt64(raw),
DriverDataType.Float32 => Convert.ToSingle(raw),
DriverDataType.Float64 => Convert.ToDouble(raw),
DriverDataType.String => Convert.ToString(raw) ?? string.Empty,
DriverDataType.DateTime => raw is DateTime dt ? dt : Convert.ToDateTime(raw),
_ => raw,
};
}
catch
{
// Caller logs + maps to BadTypeMismatch — we let null propagate so the
// outer evaluation path sets the Bad quality.
return null;
}
}
private void UnsubscribeFromUpstream()
{
foreach (var s in _upstreamSubscriptions)
{
try { s.Dispose(); } catch { /* best effort */ }
}
_upstreamSubscriptions.Clear();
}
private void EnsureLoaded()
{
if (!_loaded) throw new InvalidOperationException(
"VirtualTagEngine not loaded. Call Load(definitions) first.");
}
public void Dispose()
{
if (_disposed) return;
_disposed = true;
UnsubscribeFromUpstream();
_tags.Clear();
_graph.Clear();
}
internal DependencyGraph GraphForTesting => _graph;
private sealed class Unsub : IDisposable
{
private readonly VirtualTagEngine _engine;
private readonly string _path;
private readonly Action<string, DataValueSnapshot> _observer;
public Unsub(VirtualTagEngine e, string path, Action<string, DataValueSnapshot> observer)
{
_engine = e; _path = path; _observer = observer;
}
public void Dispose()
{
if (_engine._observers.TryGetValue(_path, out var list))
{
lock (list) { list.Remove(_observer); }
}
}
}
internal sealed record VirtualTagState(
VirtualTagDefinition Definition,
TimedScriptEvaluator<VirtualTagContext, object?> Evaluator,
IReadOnlySet<string> Reads,
IReadOnlySet<string> Writes,
ILogger Logger);
}

View File

@@ -0,0 +1,89 @@
using System.Collections.Concurrent;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
/// <summary>
/// Implements the driver-agnostic capability surface the
/// <c>DriverNodeManager</c> dispatches to when a node resolves to
/// <c>NodeSource.Virtual</c> per ADR-002. Reads return the engine's last-known
/// evaluation result; subscriptions forward engine-emitted change events as
/// <see cref="ISubscribable.OnDataChange"/> events.
/// </summary>
/// <remarks>
/// <para>
/// <see cref="IWritable"/> is deliberately not implemented — OPC UA client
/// writes to virtual tags are rejected in <c>DriverNodeManager</c> before they
/// reach here per Phase 7 decision #6. Scripts are the only write path, routed
/// through <c>ctx.SetVirtualTag</c>.
/// </para>
/// </remarks>
public sealed class VirtualTagSource : IReadable, ISubscribable
{
private readonly VirtualTagEngine _engine;
private readonly ConcurrentDictionary<string, Subscription> _subs = new(StringComparer.Ordinal);
public VirtualTagSource(VirtualTagEngine engine)
{
_engine = engine ?? throw new ArgumentNullException(nameof(engine));
}
public event EventHandler<DataChangeEventArgs>? OnDataChange;
public Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
{
if (fullReferences is null) throw new ArgumentNullException(nameof(fullReferences));
var results = new DataValueSnapshot[fullReferences.Count];
for (var i = 0; i < fullReferences.Count; i++)
results[i] = _engine.Read(fullReferences[i]);
return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(results);
}
public Task<ISubscriptionHandle> SubscribeAsync(
IReadOnlyList<string> fullReferences,
TimeSpan publishingInterval,
CancellationToken cancellationToken)
{
if (fullReferences is null) throw new ArgumentNullException(nameof(fullReferences));
var handle = new SubscriptionHandle(Guid.NewGuid().ToString("N"));
var observers = new List<IDisposable>(fullReferences.Count);
foreach (var path in fullReferences)
{
observers.Add(_engine.Subscribe(path, (p, snap) =>
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, p, snap))));
}
_subs[handle.DiagnosticId] = new Subscription(handle, observers);
// OPC UA convention: emit initial-data callback for each path with the current value.
foreach (var path in fullReferences)
{
var snap = _engine.Read(path);
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, path, snap));
}
return Task.FromResult<ISubscriptionHandle>(handle);
}
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
{
if (handle is null) throw new ArgumentNullException(nameof(handle));
if (_subs.TryRemove(handle.DiagnosticId, out var sub))
{
foreach (var d in sub.Observers)
{
try { d.Dispose(); } catch { }
}
}
return Task.CompletedTask;
}
private sealed class SubscriptionHandle : ISubscriptionHandle
{
public SubscriptionHandle(string id) { DiagnosticId = id; }
public string DiagnosticId { get; }
}
private sealed record Subscription(SubscriptionHandle Handle, IReadOnlyList<IDisposable> Observers);
}

View File

@@ -0,0 +1,32 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<LangVersion>latest</LangVersion>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
<NoWarn>$(NoWarn);CS1591</NoWarn>
<RootNamespace>ZB.MOM.WW.OtOpcUa.Core.VirtualTags</RootNamespace>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Serilog" Version="4.2.0"/>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Core.Abstractions\ZB.MOM.WW.OtOpcUa.Core.Abstractions.csproj"/>
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Core.Scripting\ZB.MOM.WW.OtOpcUa.Core.Scripting.csproj"/>
</ItemGroup>
<ItemGroup>
<InternalsVisibleTo Include="ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests"/>
</ItemGroup>
<ItemGroup>
<NuGetAuditSuppress Include="https://github.com/advisories/GHSA-37gx-xxp4-5rgx"/>
<NuGetAuditSuppress Include="https://github.com/advisories/GHSA-w3x6-4m5h-cxqf"/>
</ItemGroup>
</Project>

View File

@@ -0,0 +1,61 @@
using System.Collections.Concurrent;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests;
public sealed class FakeUpstream : ITagUpstreamSource
{
private readonly ConcurrentDictionary<string, DataValueSnapshot> _values = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, List<Action<string, DataValueSnapshot>>> _subs
= new(StringComparer.Ordinal);
public int ActiveSubscriptionCount { get; private set; }
public void Set(string path, object? value, uint statusCode = 0u)
{
var now = DateTime.UtcNow;
_values[path] = new DataValueSnapshot(value, statusCode, now, now);
}
public void Push(string path, object? value, uint statusCode = 0u)
{
Set(path, value, statusCode);
if (_subs.TryGetValue(path, out var list))
{
Action<string, DataValueSnapshot>[] snap;
lock (list) { snap = list.ToArray(); }
foreach (var obs in snap) obs(path, _values[path]);
}
}
public DataValueSnapshot ReadTag(string path)
=> _values.TryGetValue(path, out var v) ? v
: new DataValueSnapshot(null, 0x80340000u, null, DateTime.UtcNow);
public IDisposable SubscribeTag(string path, Action<string, DataValueSnapshot> observer)
{
var list = _subs.GetOrAdd(path, _ => []);
lock (list) { list.Add(observer); }
ActiveSubscriptionCount++;
return new Unsub(this, path, observer);
}
private sealed class Unsub : IDisposable
{
private readonly FakeUpstream _up;
private readonly string _path;
private readonly Action<string, DataValueSnapshot> _observer;
public Unsub(FakeUpstream up, string path, Action<string, DataValueSnapshot> observer)
{ _up = up; _path = path; _observer = observer; }
public void Dispose()
{
if (_up._subs.TryGetValue(_path, out var list))
{
lock (list)
{
if (list.Remove(_observer)) _up.ActiveSubscriptionCount--;
}
}
}
}
}

View File

@@ -0,0 +1,107 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests;
[Trait("Category", "Unit")]
public sealed class MessageTemplateTests
{
private static DataValueSnapshot Good(object? v) =>
new(v, 0u, DateTime.UtcNow, DateTime.UtcNow);
private static DataValueSnapshot Bad() =>
new(null, 0x80050000u, null, DateTime.UtcNow);
private static DataValueSnapshot? Resolver(Dictionary<string, DataValueSnapshot> map, string path)
=> map.TryGetValue(path, out var v) ? v : null;
[Fact]
public void No_tokens_returns_template_unchanged()
{
MessageTemplate.Resolve("No tokens here", _ => null).ShouldBe("No tokens here");
}
[Fact]
public void Single_token_substituted()
{
var map = new Dictionary<string, DataValueSnapshot> { ["Tank/Temp"] = Good(75.5) };
MessageTemplate.Resolve("Temp={Tank/Temp}C", p => Resolver(map, p)).ShouldBe("Temp=75.5C");
}
[Fact]
public void Multiple_tokens_substituted()
{
var map = new Dictionary<string, DataValueSnapshot>
{
["A"] = Good(10),
["B"] = Good("on"),
};
MessageTemplate.Resolve("{A}/{B}", p => Resolver(map, p)).ShouldBe("10/on");
}
[Fact]
public void Bad_quality_token_becomes_question_mark()
{
var map = new Dictionary<string, DataValueSnapshot> { ["Bad"] = Bad() };
MessageTemplate.Resolve("value={Bad}", p => Resolver(map, p)).ShouldBe("value={?}");
}
[Fact]
public void Unknown_path_becomes_question_mark()
{
MessageTemplate.Resolve("value={DoesNotExist}", _ => null).ShouldBe("value={?}");
}
[Fact]
public void Null_value_with_good_quality_becomes_question_mark()
{
var map = new Dictionary<string, DataValueSnapshot> { ["X"] = Good(null) };
MessageTemplate.Resolve("{X}", p => Resolver(map, p)).ShouldBe("{?}");
}
[Fact]
public void Tokens_with_slashes_and_dots_resolved()
{
var map = new Dictionary<string, DataValueSnapshot>
{
["Line1/Pump.Speed"] = Good(1200),
};
MessageTemplate.Resolve("rpm={Line1/Pump.Speed}", p => Resolver(map, p))
.ShouldBe("rpm=1200");
}
[Fact]
public void Empty_template_returns_empty()
{
MessageTemplate.Resolve("", _ => null).ShouldBe("");
}
[Fact]
public void Null_template_returns_empty_without_throwing()
{
MessageTemplate.Resolve(null!, _ => null).ShouldBe("");
}
[Fact]
public void ExtractTokenPaths_returns_every_distinct_token()
{
var tokens = MessageTemplate.ExtractTokenPaths("{A}/{B}/{A}/{C}");
tokens.ShouldBe(new[] { "A", "B", "A", "C" });
}
[Fact]
public void ExtractTokenPaths_empty_for_tokenless_template()
{
MessageTemplate.ExtractTokenPaths("No tokens").ShouldBeEmpty();
MessageTemplate.ExtractTokenPaths("").ShouldBeEmpty();
MessageTemplate.ExtractTokenPaths(null).ShouldBeEmpty();
}
[Fact]
public void Whitespace_inside_token_is_trimmed()
{
var map = new Dictionary<string, DataValueSnapshot> { ["A"] = Good(42) };
MessageTemplate.Resolve("{ A }", p => Resolver(map, p)).ShouldBe("42");
}
}

View File

@@ -0,0 +1,205 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests;
/// <summary>
/// Pure state-machine tests — no engine, no I/O, no async. Every transition rule
/// from Phase 7 plan Stream C.2 / C.3 has at least one locking test so regressions
/// surface as clear failures rather than subtle alarm-behavior drift.
/// </summary>
[Trait("Category", "Unit")]
public sealed class Part9StateMachineTests
{
private static readonly DateTime T0 = new(2026, 1, 1, 12, 0, 0, DateTimeKind.Utc);
private static AlarmConditionState Fresh() => AlarmConditionState.Fresh("alarm-1", T0);
[Fact]
public void Predicate_true_on_inactive_becomes_active_and_emits_Activated()
{
var r = Part9StateMachine.ApplyPredicate(Fresh(), predicateTrue: true, T0.AddSeconds(1));
r.State.Active.ShouldBe(AlarmActiveState.Active);
r.State.Acked.ShouldBe(AlarmAckedState.Unacknowledged);
r.State.Confirmed.ShouldBe(AlarmConfirmedState.Unconfirmed);
r.Emission.ShouldBe(EmissionKind.Activated);
r.State.LastActiveUtc.ShouldNotBeNull();
}
[Fact]
public void Predicate_false_on_active_becomes_inactive_and_emits_Cleared()
{
var active = Part9StateMachine.ApplyPredicate(Fresh(), true, T0.AddSeconds(1)).State;
var r = Part9StateMachine.ApplyPredicate(active, false, T0.AddSeconds(2));
r.State.Active.ShouldBe(AlarmActiveState.Inactive);
r.Emission.ShouldBe(EmissionKind.Cleared);
r.State.LastClearedUtc.ShouldNotBeNull();
}
[Fact]
public void Predicate_unchanged_state_emits_None()
{
var r = Part9StateMachine.ApplyPredicate(Fresh(), false, T0);
r.Emission.ShouldBe(EmissionKind.None);
}
[Fact]
public void Disabled_alarm_ignores_predicate()
{
var disabled = Part9StateMachine.ApplyDisable(Fresh(), "op1", T0.AddSeconds(1)).State;
var r = Part9StateMachine.ApplyPredicate(disabled, true, T0.AddSeconds(2));
r.State.Active.ShouldBe(AlarmActiveState.Inactive);
r.Emission.ShouldBe(EmissionKind.None);
}
[Fact]
public void Acknowledge_from_unacked_records_user_and_emits()
{
var active = Part9StateMachine.ApplyPredicate(Fresh(), true, T0.AddSeconds(1)).State;
var r = Part9StateMachine.ApplyAcknowledge(active, "alice", "looking into it", T0.AddSeconds(2));
r.State.Acked.ShouldBe(AlarmAckedState.Acknowledged);
r.State.LastAckUser.ShouldBe("alice");
r.State.LastAckComment.ShouldBe("looking into it");
r.State.Comments.Count.ShouldBe(1);
r.Emission.ShouldBe(EmissionKind.Acknowledged);
}
[Fact]
public void Acknowledge_when_already_acked_is_noop()
{
var active = Part9StateMachine.ApplyPredicate(Fresh(), true, T0.AddSeconds(1)).State;
var acked = Part9StateMachine.ApplyAcknowledge(active, "alice", null, T0.AddSeconds(2)).State;
var r = Part9StateMachine.ApplyAcknowledge(acked, "alice", null, T0.AddSeconds(3));
r.Emission.ShouldBe(EmissionKind.None);
}
[Fact]
public void Acknowledge_without_user_throws()
{
Should.Throw<ArgumentException>(() =>
Part9StateMachine.ApplyAcknowledge(Fresh(), "", null, T0));
}
[Fact]
public void Confirm_after_clear_records_user_and_emits()
{
// Walk: activate -> ack -> clear -> confirm
var s = Fresh();
s = Part9StateMachine.ApplyPredicate(s, true, T0.AddSeconds(1)).State;
s = Part9StateMachine.ApplyAcknowledge(s, "alice", null, T0.AddSeconds(2)).State;
s = Part9StateMachine.ApplyPredicate(s, false, T0.AddSeconds(3)).State;
var r = Part9StateMachine.ApplyConfirm(s, "bob", "resolved", T0.AddSeconds(4));
r.State.Confirmed.ShouldBe(AlarmConfirmedState.Confirmed);
r.State.LastConfirmUser.ShouldBe("bob");
r.Emission.ShouldBe(EmissionKind.Confirmed);
}
[Fact]
public void OneShotShelve_suppresses_next_activation_emission()
{
var s = Part9StateMachine.ApplyOneShotShelve(Fresh(), "alice", T0.AddSeconds(1)).State;
var r = Part9StateMachine.ApplyPredicate(s, true, T0.AddSeconds(2));
r.State.Active.ShouldBe(AlarmActiveState.Active, "state still advances");
r.Emission.ShouldBe(EmissionKind.Suppressed, "but subscribers don't see it");
}
[Fact]
public void OneShotShelve_expires_on_clear()
{
var s = Fresh();
s = Part9StateMachine.ApplyOneShotShelve(s, "alice", T0.AddSeconds(1)).State;
s = Part9StateMachine.ApplyPredicate(s, true, T0.AddSeconds(2)).State;
var r = Part9StateMachine.ApplyPredicate(s, false, T0.AddSeconds(3));
r.State.Shelving.Kind.ShouldBe(ShelvingKind.Unshelved, "OneShot expires on clear");
}
[Fact]
public void TimedShelve_requires_future_unshelve_time()
{
Should.Throw<ArgumentOutOfRangeException>(() =>
Part9StateMachine.ApplyTimedShelve(Fresh(), "alice", T0, T0.AddSeconds(5)));
}
[Fact]
public void TimedShelve_expires_via_shelving_check()
{
var until = T0.AddMinutes(5);
var shelved = Part9StateMachine.ApplyTimedShelve(Fresh(), "alice", until, T0).State;
shelved.Shelving.Kind.ShouldBe(ShelvingKind.Timed);
// Before expiry — still shelved.
var earlier = Part9StateMachine.ApplyShelvingCheck(shelved, T0.AddMinutes(3));
earlier.State.Shelving.Kind.ShouldBe(ShelvingKind.Timed);
earlier.Emission.ShouldBe(EmissionKind.None);
// After expiry — auto-unshelved + emission.
var after = Part9StateMachine.ApplyShelvingCheck(shelved, T0.AddMinutes(6));
after.State.Shelving.Kind.ShouldBe(ShelvingKind.Unshelved);
after.Emission.ShouldBe(EmissionKind.Unshelved);
after.State.Comments.Any(c => c.Kind == "AutoUnshelve").ShouldBeTrue();
}
[Fact]
public void Unshelve_from_unshelved_is_noop()
{
var r = Part9StateMachine.ApplyUnshelve(Fresh(), "alice", T0);
r.Emission.ShouldBe(EmissionKind.None);
}
[Fact]
public void Explicit_Unshelve_emits_event()
{
var s = Part9StateMachine.ApplyOneShotShelve(Fresh(), "alice", T0).State;
var r = Part9StateMachine.ApplyUnshelve(s, "bob", T0.AddSeconds(30));
r.State.Shelving.Kind.ShouldBe(ShelvingKind.Unshelved);
r.Emission.ShouldBe(EmissionKind.Unshelved);
}
[Fact]
public void AddComment_appends_to_audit_trail_with_event()
{
var r = Part9StateMachine.ApplyAddComment(Fresh(), "alice", "investigating", T0.AddSeconds(5));
r.State.Comments.Count.ShouldBe(1);
r.State.Comments[0].Kind.ShouldBe("AddComment");
r.State.Comments[0].User.ShouldBe("alice");
r.State.Comments[0].Text.ShouldBe("investigating");
r.Emission.ShouldBe(EmissionKind.CommentAdded);
}
[Fact]
public void Comments_are_append_only_never_rewritten()
{
var s = Part9StateMachine.ApplyAddComment(Fresh(), "alice", "first", T0.AddSeconds(1)).State;
s = Part9StateMachine.ApplyAddComment(s, "bob", "second", T0.AddSeconds(2)).State;
s = Part9StateMachine.ApplyAddComment(s, "carol", "third", T0.AddSeconds(3)).State;
s.Comments.Count.ShouldBe(3);
s.Comments[0].User.ShouldBe("alice");
s.Comments[1].User.ShouldBe("bob");
s.Comments[2].User.ShouldBe("carol");
}
[Fact]
public void Full_lifecycle_walk_produces_every_expected_emission()
{
// Walk a condition through its whole lifecycle and make sure emissions line up.
var emissions = new List<EmissionKind>();
var s = Fresh();
s = Capture(Part9StateMachine.ApplyPredicate(s, true, T0.AddSeconds(1)));
s = Capture(Part9StateMachine.ApplyAcknowledge(s, "alice", null, T0.AddSeconds(2)));
s = Capture(Part9StateMachine.ApplyAddComment(s, "alice", "need to investigate", T0.AddSeconds(3)));
s = Capture(Part9StateMachine.ApplyPredicate(s, false, T0.AddSeconds(4)));
s = Capture(Part9StateMachine.ApplyConfirm(s, "bob", null, T0.AddSeconds(5)));
emissions.ShouldBe(new[] {
EmissionKind.Activated,
EmissionKind.Acknowledged,
EmissionKind.CommentAdded,
EmissionKind.Cleared,
EmissionKind.Confirmed,
});
AlarmConditionState Capture(TransitionResult r) { emissions.Add(r.Emission); return r.State; }
}
}

View File

@@ -0,0 +1,316 @@
using Serilog;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests;
/// <summary>
/// End-to-end engine tests: load, predicate evaluation, change-triggered
/// re-evaluation, state persistence, startup recovery, error isolation.
/// </summary>
[Trait("Category", "Unit")]
public sealed class ScriptedAlarmEngineTests
{
private static ScriptedAlarmEngine Build(FakeUpstream up, out IAlarmStateStore store)
{
store = new InMemoryAlarmStateStore();
var logger = new LoggerConfiguration().CreateLogger();
return new ScriptedAlarmEngine(up, store, new ScriptLoggerFactory(logger), logger);
}
private static ScriptedAlarmDefinition Alarm(string id, string predicate,
string msg = "condition", AlarmSeverity sev = AlarmSeverity.High) =>
new(AlarmId: id,
EquipmentPath: "Plant/Line1/Reactor",
AlarmName: id,
Kind: AlarmKind.AlarmCondition,
Severity: sev,
MessageTemplate: msg,
PredicateScriptSource: predicate);
[Fact]
public async Task Load_compiles_and_subscribes_to_referenced_upstreams()
{
var up = new FakeUpstream();
up.Set("Temp", 50);
using var eng = Build(up, out _);
await eng.LoadAsync([Alarm("a1", """return (int)ctx.GetTag("Temp").Value > 100;""")],
TestContext.Current.CancellationToken);
eng.LoadedAlarmIds.ShouldContain("a1");
up.ActiveSubscriptionCount.ShouldBe(1);
}
[Fact]
public async Task Compile_failures_aggregated_into_one_error()
{
var up = new FakeUpstream();
using var eng = Build(up, out _);
var ex = await Should.ThrowAsync<InvalidOperationException>(async () =>
await eng.LoadAsync([
Alarm("bad1", "return unknownIdentifier;"),
Alarm("good", "return true;"),
Alarm("bad2", "var x = alsoUnknown; return x;"),
], TestContext.Current.CancellationToken));
ex.Message.ShouldContain("2 alarm(s) did not compile");
}
[Fact]
public async Task Upstream_change_re_evaluates_predicate_and_emits_Activated()
{
var up = new FakeUpstream();
up.Set("Temp", 50);
using var eng = Build(up, out _);
await eng.LoadAsync([Alarm("HighTemp", """return (int)ctx.GetTag("Temp").Value > 100;""")],
TestContext.Current.CancellationToken);
var events = new List<ScriptedAlarmEvent>();
eng.OnEvent += (_, e) => events.Add(e);
up.Push("Temp", 150);
await WaitForAsync(() => events.Count > 0);
events[0].AlarmId.ShouldBe("HighTemp");
events[0].Emission.ShouldBe(EmissionKind.Activated);
eng.GetState("HighTemp")!.Active.ShouldBe(AlarmActiveState.Active);
}
[Fact]
public async Task Clearing_upstream_emits_Cleared_event()
{
var up = new FakeUpstream();
up.Set("Temp", 150);
using var eng = Build(up, out _);
await eng.LoadAsync([Alarm("HighTemp", """return (int)ctx.GetTag("Temp").Value > 100;""")],
TestContext.Current.CancellationToken);
// Startup sees 150 → active.
eng.GetState("HighTemp")!.Active.ShouldBe(AlarmActiveState.Active);
var events = new List<ScriptedAlarmEvent>();
eng.OnEvent += (_, e) => events.Add(e);
up.Push("Temp", 50);
await WaitForAsync(() => events.Any(e => e.Emission == EmissionKind.Cleared));
eng.GetState("HighTemp")!.Active.ShouldBe(AlarmActiveState.Inactive);
}
[Fact]
public async Task Message_template_resolves_tag_values_at_emission()
{
var up = new FakeUpstream();
up.Set("Temp", 50);
up.Set("Limit", 100);
using var eng = Build(up, out _);
await eng.LoadAsync([
new ScriptedAlarmDefinition(
"HighTemp", "Plant/Line1", "HighTemp",
AlarmKind.LimitAlarm, AlarmSeverity.High,
"Temp {Temp}C exceeded limit {Limit}C",
"""return (int)ctx.GetTag("Temp").Value > (int)ctx.GetTag("Limit").Value;"""),
], TestContext.Current.CancellationToken);
var events = new List<ScriptedAlarmEvent>();
eng.OnEvent += (_, e) => events.Add(e);
up.Push("Temp", 150);
await WaitForAsync(() => events.Any());
events[0].Message.ShouldBe("Temp 150C exceeded limit 100C");
}
[Fact]
public async Task Ack_records_user_and_persists_to_store()
{
var up = new FakeUpstream();
up.Set("Temp", 150);
using var eng = Build(up, out var store);
await eng.LoadAsync([Alarm("HighTemp", """return (int)ctx.GetTag("Temp").Value > 100;""")],
TestContext.Current.CancellationToken);
await eng.AcknowledgeAsync("HighTemp", "alice", "checking", TestContext.Current.CancellationToken);
var persisted = await store.LoadAsync("HighTemp", TestContext.Current.CancellationToken);
persisted.ShouldNotBeNull();
persisted!.Acked.ShouldBe(AlarmAckedState.Acknowledged);
persisted.LastAckUser.ShouldBe("alice");
persisted.LastAckComment.ShouldBe("checking");
persisted.Comments.Any(c => c.Kind == "Acknowledge" && c.User == "alice").ShouldBeTrue();
}
[Fact]
public async Task Startup_recovery_preserves_ack_but_rederives_active_from_predicate()
{
var up = new FakeUpstream();
up.Set("Temp", 50); // predicate will go false on second load
// First run — alarm goes active + operator acks.
using (var eng1 = Build(up, out var sharedStore))
{
up.Set("Temp", 150);
await eng1.LoadAsync([Alarm("HighTemp", """return (int)ctx.GetTag("Temp").Value > 100;""")],
TestContext.Current.CancellationToken);
eng1.GetState("HighTemp")!.Active.ShouldBe(AlarmActiveState.Active);
await eng1.AcknowledgeAsync("HighTemp", "alice", null, TestContext.Current.CancellationToken);
eng1.GetState("HighTemp")!.Acked.ShouldBe(AlarmAckedState.Acknowledged);
}
// Simulate restart — temp is back to 50 (below threshold).
up.Set("Temp", 50);
var logger = new LoggerConfiguration().CreateLogger();
var store2 = new InMemoryAlarmStateStore();
// seed store2 with the acked state from before restart
await store2.SaveAsync(new AlarmConditionState(
"HighTemp",
AlarmEnabledState.Enabled,
AlarmActiveState.Active, // was active pre-restart
AlarmAckedState.Acknowledged, // ack persisted
AlarmConfirmedState.Unconfirmed,
ShelvingState.Unshelved,
DateTime.UtcNow,
DateTime.UtcNow, null,
DateTime.UtcNow, "alice", null,
null, null, null,
[new AlarmComment(DateTime.UtcNow, "alice", "Acknowledge", "")]),
TestContext.Current.CancellationToken);
using var eng2 = new ScriptedAlarmEngine(up, store2, new ScriptLoggerFactory(logger), logger);
await eng2.LoadAsync([Alarm("HighTemp", """return (int)ctx.GetTag("Temp").Value > 100;""")],
TestContext.Current.CancellationToken);
var s = eng2.GetState("HighTemp")!;
s.Active.ShouldBe(AlarmActiveState.Inactive, "Active recomputed from current tag value");
s.Acked.ShouldBe(AlarmAckedState.Acknowledged, "Ack persisted across restart");
s.LastAckUser.ShouldBe("alice");
}
[Fact]
public async Task Shelved_active_transitions_state_but_suppresses_emission()
{
var up = new FakeUpstream();
up.Set("Temp", 50);
using var eng = Build(up, out _);
await eng.LoadAsync([Alarm("HighTemp", """return (int)ctx.GetTag("Temp").Value > 100;""")],
TestContext.Current.CancellationToken);
await eng.OneShotShelveAsync("HighTemp", "alice", TestContext.Current.CancellationToken);
var events = new List<ScriptedAlarmEvent>();
eng.OnEvent += (_, e) => events.Add(e);
up.Push("Temp", 150);
await Task.Delay(200);
events.Any(e => e.Emission == EmissionKind.Activated).ShouldBeFalse(
"OneShot shelve suppresses activation emission");
eng.GetState("HighTemp")!.Active.ShouldBe(AlarmActiveState.Active,
"state still advances so startup recovery is consistent");
}
[Fact]
public async Task Predicate_runtime_exception_does_not_transition_state()
{
var up = new FakeUpstream();
up.Set("Temp", 150);
using var eng = Build(up, out _);
await eng.LoadAsync([
Alarm("BadScript", """throw new InvalidOperationException("boom");"""),
Alarm("GoodScript", """return (int)ctx.GetTag("Temp").Value > 100;"""),
], TestContext.Current.CancellationToken);
// Bad script doesn't activate + doesn't disable other alarms.
eng.GetState("BadScript")!.Active.ShouldBe(AlarmActiveState.Inactive);
eng.GetState("GoodScript")!.Active.ShouldBe(AlarmActiveState.Active);
}
[Fact]
public async Task Disable_prevents_activation_until_re_enabled()
{
var up = new FakeUpstream();
up.Set("Temp", 50);
using var eng = Build(up, out _);
await eng.LoadAsync([Alarm("HighTemp", """return (int)ctx.GetTag("Temp").Value > 100;""")],
TestContext.Current.CancellationToken);
await eng.DisableAsync("HighTemp", "alice", TestContext.Current.CancellationToken);
up.Push("Temp", 150);
await Task.Delay(100);
eng.GetState("HighTemp")!.Active.ShouldBe(AlarmActiveState.Inactive,
"disabled alarm ignores predicate");
await eng.EnableAsync("HighTemp", "alice", TestContext.Current.CancellationToken);
up.Push("Temp", 160);
await WaitForAsync(() => eng.GetState("HighTemp")!.Active == AlarmActiveState.Active);
}
[Fact]
public async Task AddComment_appends_to_audit_without_state_change()
{
var up = new FakeUpstream();
up.Set("Temp", 50);
using var eng = Build(up, out var store);
await eng.LoadAsync([Alarm("A", """return false;""")], TestContext.Current.CancellationToken);
await eng.AddCommentAsync("A", "alice", "peeking at this", TestContext.Current.CancellationToken);
var s = await store.LoadAsync("A", TestContext.Current.CancellationToken);
s.ShouldNotBeNull();
s!.Comments.Count.ShouldBe(1);
s.Comments[0].User.ShouldBe("alice");
s.Comments[0].Kind.ShouldBe("AddComment");
}
[Fact]
public async Task Predicate_scripts_cannot_SetVirtualTag()
{
var up = new FakeUpstream();
up.Set("Temp", 100);
using var eng = Build(up, out _);
// The script compiles fine but throws at runtime when SetVirtualTag is called.
// The engine swallows the exception + leaves state unchanged.
await eng.LoadAsync([
new ScriptedAlarmDefinition(
"Bad", "Plant/Line1", "Bad",
AlarmKind.AlarmCondition, AlarmSeverity.High, "bad",
"""
ctx.SetVirtualTag("NotAllowed", 1);
return true;
"""),
], TestContext.Current.CancellationToken);
// Bad alarm's predicate threw — state unchanged.
eng.GetState("Bad")!.Active.ShouldBe(AlarmActiveState.Inactive);
}
[Fact]
public async Task Dispose_releases_upstream_subscriptions()
{
var up = new FakeUpstream();
up.Set("Temp", 50);
var eng = Build(up, out _);
await eng.LoadAsync([Alarm("A", """return (int)ctx.GetTag("Temp").Value > 100;""")],
TestContext.Current.CancellationToken);
up.ActiveSubscriptionCount.ShouldBe(1);
eng.Dispose();
up.ActiveSubscriptionCount.ShouldBe(0);
}
private static async Task WaitForAsync(Func<bool> cond, int timeoutMs = 2000)
{
var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs);
while (DateTime.UtcNow < deadline)
{
if (cond()) return;
await Task.Delay(25);
}
throw new TimeoutException("Condition did not become true in time");
}
}

View File

@@ -0,0 +1,142 @@
using Serilog;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests;
[Trait("Category", "Unit")]
public sealed class ScriptedAlarmSourceTests
{
private static async Task<(ScriptedAlarmEngine e, ScriptedAlarmSource s, FakeUpstream u)> BuildAsync()
{
var up = new FakeUpstream();
up.Set("Temp", 50);
var logger = new LoggerConfiguration().CreateLogger();
var engine = new ScriptedAlarmEngine(up, new InMemoryAlarmStateStore(),
new ScriptLoggerFactory(logger), logger);
await engine.LoadAsync([
new ScriptedAlarmDefinition(
"Plant/Line1::HighTemp",
"Plant/Line1",
"HighTemp",
AlarmKind.LimitAlarm,
AlarmSeverity.High,
"Temp {Temp}C",
"""return (int)ctx.GetTag("Temp").Value > 100;"""),
new ScriptedAlarmDefinition(
"Plant/Line2::OtherAlarm",
"Plant/Line2",
"OtherAlarm",
AlarmKind.AlarmCondition,
AlarmSeverity.Low,
"other",
"""return false;"""),
], CancellationToken.None);
var source = new ScriptedAlarmSource(engine);
return (engine, source, up);
}
[Fact]
public async Task Subscribe_with_empty_filter_receives_every_alarm_emission()
{
var (engine, source, up) = await BuildAsync();
using var _e = engine;
using var _s = source;
var events = new List<AlarmEventArgs>();
source.OnAlarmEvent += (_, e) => events.Add(e);
var handle = await source.SubscribeAlarmsAsync([], TestContext.Current.CancellationToken);
up.Push("Temp", 150);
await Task.Delay(200);
events.Count.ShouldBe(1);
events[0].ConditionId.ShouldBe("Plant/Line1::HighTemp");
events[0].SourceNodeId.ShouldBe("Plant/Line1");
events[0].Severity.ShouldBe(AlarmSeverity.High);
events[0].AlarmType.ShouldBe("LimitAlarm");
events[0].Message.ShouldBe("Temp 150C");
await source.UnsubscribeAlarmsAsync(handle, TestContext.Current.CancellationToken);
}
[Fact]
public async Task Subscribe_with_equipment_prefix_filters_by_that_prefix()
{
var (engine, source, up) = await BuildAsync();
using var _e = engine;
using var _s = source;
var events = new List<AlarmEventArgs>();
source.OnAlarmEvent += (_, e) => events.Add(e);
// Subscribe only to Line1 alarms.
var handle = await source.SubscribeAlarmsAsync(["Plant/Line1"], TestContext.Current.CancellationToken);
up.Push("Temp", 150);
await Task.Delay(200);
events.Count.ShouldBe(1);
events[0].SourceNodeId.ShouldBe("Plant/Line1");
await source.UnsubscribeAlarmsAsync(handle, TestContext.Current.CancellationToken);
}
[Fact]
public async Task Unsubscribe_stops_further_events()
{
var (engine, source, up) = await BuildAsync();
using var _e = engine;
using var _s = source;
var events = new List<AlarmEventArgs>();
source.OnAlarmEvent += (_, e) => events.Add(e);
var handle = await source.SubscribeAlarmsAsync([], TestContext.Current.CancellationToken);
await source.UnsubscribeAlarmsAsync(handle, TestContext.Current.CancellationToken);
up.Push("Temp", 150);
await Task.Delay(200);
events.Count.ShouldBe(0);
}
[Fact]
public async Task AcknowledgeAsync_routes_to_engine_with_default_user()
{
var (engine, source, up) = await BuildAsync();
using var _e = engine;
using var _s = source;
up.Push("Temp", 150);
await Task.Delay(200);
engine.GetState("Plant/Line1::HighTemp")!.Acked.ShouldBe(AlarmAckedState.Unacknowledged);
await source.AcknowledgeAsync([new AlarmAcknowledgeRequest(
"Plant/Line1", "Plant/Line1::HighTemp", "ack via opcua")],
TestContext.Current.CancellationToken);
var state = engine.GetState("Plant/Line1::HighTemp")!;
state.Acked.ShouldBe(AlarmAckedState.Acknowledged);
state.LastAckUser.ShouldBe("opcua-client");
state.LastAckComment.ShouldBe("ack via opcua");
}
[Fact]
public async Task Null_arguments_rejected()
{
var (engine, source, _) = await BuildAsync();
using var _e = engine;
using var _s = source;
await Should.ThrowAsync<ArgumentNullException>(async () =>
await source.SubscribeAlarmsAsync(null!, TestContext.Current.CancellationToken));
await Should.ThrowAsync<ArgumentNullException>(async () =>
await source.UnsubscribeAlarmsAsync(null!, TestContext.Current.CancellationToken));
await Should.ThrowAsync<ArgumentNullException>(async () =>
await source.AcknowledgeAsync(null!, TestContext.Current.CancellationToken));
}
}

View File

@@ -0,0 +1,31 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<IsPackable>false</IsPackable>
<IsTestProject>true</IsTestProject>
<RootNamespace>ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests</RootNamespace>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="xunit.v3" Version="1.1.0"/>
<PackageReference Include="Shouldly" Version="4.3.0"/>
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.12.0"/>
<PackageReference Include="xunit.runner.visualstudio" Version="3.0.2">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\src\ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms\ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.csproj"/>
</ItemGroup>
<ItemGroup>
<NuGetAuditSuppress Include="https://github.com/advisories/GHSA-37gx-xxp4-5rgx"/>
<NuGetAuditSuppress Include="https://github.com/advisories/GHSA-w3x6-4m5h-cxqf"/>
</ItemGroup>
</Project>

View File

@@ -0,0 +1,166 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests;
/// <summary>
/// Verifies cycle detection + topological sort on the virtual-tag dependency
/// graph. Publish-time correctness depends on these being right — a missed cycle
/// would deadlock cascade evaluation; a wrong topological order would miscompute
/// chained virtual tags.
/// </summary>
[Trait("Category", "Unit")]
public sealed class DependencyGraphTests
{
private static IReadOnlySet<string> Set(params string[] items) =>
new HashSet<string>(items, StringComparer.Ordinal);
[Fact]
public void Empty_graph_produces_empty_sort_and_no_cycles()
{
var g = new DependencyGraph();
g.TopologicalSort().ShouldBeEmpty();
g.DetectCycles().ShouldBeEmpty();
}
[Fact]
public void Single_node_with_no_deps()
{
var g = new DependencyGraph();
g.Add("A", Set());
g.TopologicalSort().ShouldBe(new[] { "A" });
g.DetectCycles().ShouldBeEmpty();
}
[Fact]
public void Topological_order_places_dependencies_before_dependents()
{
var g = new DependencyGraph();
g.Add("B", Set("A")); // B depends on A
g.Add("C", Set("B", "A")); // C depends on B + A
g.Add("A", Set()); // A is a leaf
var order = g.TopologicalSort();
var idx = order.Select((x, i) => (x, i)).ToDictionary(p => p.x, p => p.i);
idx["A"].ShouldBeLessThan(idx["B"]);
idx["B"].ShouldBeLessThan(idx["C"]);
}
[Fact]
public void Self_loop_detected_as_cycle()
{
var g = new DependencyGraph();
g.Add("A", Set("A"));
var cycles = g.DetectCycles();
cycles.Count.ShouldBe(1);
cycles[0].ShouldContain("A");
}
[Fact]
public void Two_node_cycle_detected()
{
var g = new DependencyGraph();
g.Add("A", Set("B"));
g.Add("B", Set("A"));
var cycles = g.DetectCycles();
cycles.Count.ShouldBe(1);
cycles[0].Count.ShouldBe(2);
}
[Fact]
public void Three_node_cycle_detected()
{
var g = new DependencyGraph();
g.Add("A", Set("B"));
g.Add("B", Set("C"));
g.Add("C", Set("A"));
var cycles = g.DetectCycles();
cycles.Count.ShouldBe(1);
cycles[0].Count.ShouldBe(3);
}
[Fact]
public void Multiple_disjoint_cycles_all_reported()
{
var g = new DependencyGraph();
// Cycle 1: A -> B -> A
g.Add("A", Set("B"));
g.Add("B", Set("A"));
// Cycle 2: X -> Y -> Z -> X
g.Add("X", Set("Y"));
g.Add("Y", Set("Z"));
g.Add("Z", Set("X"));
// Clean leaf: M
g.Add("M", Set());
var cycles = g.DetectCycles();
cycles.Count.ShouldBe(2);
}
[Fact]
public void Topological_sort_throws_DependencyCycleException_on_cycle()
{
var g = new DependencyGraph();
g.Add("A", Set("B"));
g.Add("B", Set("A"));
Should.Throw<DependencyCycleException>(() => g.TopologicalSort())
.Cycles.ShouldNotBeEmpty();
}
[Fact]
public void DirectDependents_returns_direct_only()
{
var g = new DependencyGraph();
g.Add("B", Set("A"));
g.Add("C", Set("B"));
g.DirectDependents("A").ShouldBe(new[] { "B" });
g.DirectDependents("B").ShouldBe(new[] { "C" });
g.DirectDependents("C").ShouldBeEmpty();
}
[Fact]
public void TransitiveDependentsInOrder_returns_topological_closure()
{
var g = new DependencyGraph();
g.Add("B", Set("A"));
g.Add("C", Set("B"));
g.Add("D", Set("C"));
var closure = g.TransitiveDependentsInOrder("A");
closure.ShouldBe(new[] { "B", "C", "D" });
}
[Fact]
public void Readding_a_node_overwrites_prior_dependencies()
{
var g = new DependencyGraph();
g.Add("X", Set("A"));
g.DirectDependencies("X").ShouldBe(new[] { "A" });
// Re-add with different deps (simulates script edit + republish).
g.Add("X", Set("B", "C"));
g.DirectDependencies("X").OrderBy(s => s).ShouldBe(new[] { "B", "C" });
// A should no longer list X as a dependent.
g.DirectDependents("A").ShouldBeEmpty();
}
[Fact]
public void Leaf_dependencies_not_registered_as_nodes_are_treated_as_implicit()
{
// A is referenced but never Add'd as a node — it's an upstream driver tag.
var g = new DependencyGraph();
g.Add("B", Set("A"));
g.TopologicalSort().ShouldBe(new[] { "B" });
g.DirectDependents("A").ShouldBe(new[] { "B" });
}
[Fact]
public void Deep_graph_no_stack_overflow()
{
// Iterative Tarjan's + Kahn's — 10k deep chain must complete without blowing the stack.
var g = new DependencyGraph();
for (var i = 1; i < 10_000; i++)
g.Add($"N{i}", Set($"N{i - 1}"));
var order = g.TopologicalSort();
order.Count.ShouldBe(9_999);
}
}

View File

@@ -0,0 +1,70 @@
using System.Collections.Concurrent;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests;
/// <summary>
/// In-memory <see cref="ITagUpstreamSource"/> for tests. Seed tag values via
/// <see cref="Set"/>, push changes via <see cref="Push"/>. Tracks subscriptions so
/// tests can assert the engine disposes them on reload / shutdown.
/// </summary>
public sealed class FakeUpstream : ITagUpstreamSource
{
private readonly ConcurrentDictionary<string, DataValueSnapshot> _values = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, List<Action<string, DataValueSnapshot>>> _subs = new(StringComparer.Ordinal);
public int ActiveSubscriptionCount { get; private set; }
public void Set(string path, object value, uint statusCode = 0u)
{
var now = DateTime.UtcNow;
_values[path] = new DataValueSnapshot(value, statusCode, now, now);
}
public void Push(string path, object value, uint statusCode = 0u)
{
Set(path, value, statusCode);
if (_subs.TryGetValue(path, out var list))
{
Action<string, DataValueSnapshot>[] snap;
lock (list) { snap = list.ToArray(); }
foreach (var obs in snap) obs(path, _values[path]);
}
}
public DataValueSnapshot ReadTag(string path)
=> _values.TryGetValue(path, out var v)
? v
: new DataValueSnapshot(null, 0x80340000u, null, DateTime.UtcNow);
public IDisposable SubscribeTag(string path, Action<string, DataValueSnapshot> observer)
{
var list = _subs.GetOrAdd(path, _ => []);
lock (list) { list.Add(observer); }
ActiveSubscriptionCount++;
return new Unsub(this, path, observer);
}
private sealed class Unsub : IDisposable
{
private readonly FakeUpstream _up;
private readonly string _path;
private readonly Action<string, DataValueSnapshot> _observer;
public Unsub(FakeUpstream up, string path, Action<string, DataValueSnapshot> observer)
{
_up = up; _path = path; _observer = observer;
}
public void Dispose()
{
if (_up._subs.TryGetValue(_path, out var list))
{
lock (list)
{
if (list.Remove(_observer))
_up.ActiveSubscriptionCount--;
}
}
}
}
}

View File

@@ -0,0 +1,118 @@
using Serilog;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests;
[Trait("Category", "Unit")]
public sealed class TimerTriggerSchedulerTests
{
[Fact]
public async Task Timer_interval_causes_periodic_reevaluation()
{
var up = new FakeUpstream();
// Counter source — re-eval should pick up new value each tick.
var counter = 0;
var logger = new LoggerConfiguration().CreateLogger();
using var engine = new VirtualTagEngine(up,
new ScriptLoggerFactory(logger),
logger);
engine.Load([new VirtualTagDefinition(
"Counter", DriverDataType.Int32,
"""return ctx.Now.Millisecond;""", // changes on every evaluation
ChangeTriggered: false,
TimerInterval: TimeSpan.FromMilliseconds(100))]);
using var sched = new TimerTriggerScheduler(engine, logger);
sched.Start([new VirtualTagDefinition(
"Counter", DriverDataType.Int32,
"""return ctx.Now.Millisecond;""",
ChangeTriggered: false,
TimerInterval: TimeSpan.FromMilliseconds(100))]);
// Watch the value change across ticks.
var snapshots = new List<object?>();
using var sub = engine.Subscribe("Counter", (_, v) => snapshots.Add(v.Value));
await Task.Delay(500);
snapshots.Count.ShouldBeGreaterThanOrEqualTo(3, "At least 3 ticks in 500ms at 100ms cadence");
}
[Fact]
public async Task Tags_without_TimerInterval_not_scheduled()
{
var up = new FakeUpstream();
var logger = new LoggerConfiguration().CreateLogger();
using var engine = new VirtualTagEngine(up,
new ScriptLoggerFactory(logger), logger);
engine.Load([new VirtualTagDefinition(
"NoTimer", DriverDataType.Int32, """return 1;""")]);
using var sched = new TimerTriggerScheduler(engine, logger);
sched.Start([new VirtualTagDefinition(
"NoTimer", DriverDataType.Int32, """return 1;""")]);
var events = new List<int>();
using var sub = engine.Subscribe("NoTimer", (_, v) => events.Add((int)(v.Value ?? 0)));
await Task.Delay(300);
events.Count.ShouldBe(0, "No TimerInterval = no timer ticks");
}
[Fact]
public void Start_groups_tags_by_interval_into_shared_timers()
{
// Smoke test — Start on a definition list with two distinct intervals must not
// throw. Group count matches unique intervals.
var up = new FakeUpstream();
var logger = new LoggerConfiguration().CreateLogger();
using var engine = new VirtualTagEngine(up,
new ScriptLoggerFactory(logger), logger);
engine.Load([
new VirtualTagDefinition("Fast", DriverDataType.Int32, """return 1;""",
TimerInterval: TimeSpan.FromSeconds(1)),
new VirtualTagDefinition("Slow", DriverDataType.Int32, """return 2;""",
TimerInterval: TimeSpan.FromSeconds(5)),
new VirtualTagDefinition("AlsoFast", DriverDataType.Int32, """return 3;""",
TimerInterval: TimeSpan.FromSeconds(1)),
]);
using var sched = new TimerTriggerScheduler(engine, logger);
Should.NotThrow(() => sched.Start(new[]
{
new VirtualTagDefinition("Fast", DriverDataType.Int32, """return 1;""", TimerInterval: TimeSpan.FromSeconds(1)),
new VirtualTagDefinition("Slow", DriverDataType.Int32, """return 2;""", TimerInterval: TimeSpan.FromSeconds(5)),
new VirtualTagDefinition("AlsoFast", DriverDataType.Int32, """return 3;""", TimerInterval: TimeSpan.FromSeconds(1)),
}));
}
[Fact]
public void Disposed_scheduler_stops_firing()
{
var up = new FakeUpstream();
var logger = new LoggerConfiguration().CreateLogger();
using var engine = new VirtualTagEngine(up,
new ScriptLoggerFactory(logger), logger);
engine.Load([new VirtualTagDefinition(
"T", DriverDataType.Int32, """return 1;""",
TimerInterval: TimeSpan.FromMilliseconds(50))]);
var sched = new TimerTriggerScheduler(engine, logger);
sched.Start([new VirtualTagDefinition(
"T", DriverDataType.Int32, """return 1;""",
TimerInterval: TimeSpan.FromMilliseconds(50))]);
sched.Dispose();
// After dispose, second Start throws ObjectDisposedException.
Should.Throw<ObjectDisposedException>(() =>
sched.Start([new VirtualTagDefinition(
"T", DriverDataType.Int32, """return 1;""",
TimerInterval: TimeSpan.FromMilliseconds(50))]));
}
}

View File

@@ -0,0 +1,307 @@
using Serilog;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests;
/// <summary>
/// End-to-end VirtualTagEngine behavior: load config, subscribe to upstream,
/// evaluate on change, cascade through dependent virtual tags, timer-driven
/// re-evaluation, error isolation, historize flag, cycle rejection.
/// </summary>
[Trait("Category", "Unit")]
public sealed class VirtualTagEngineTests
{
private static VirtualTagEngine Build(
FakeUpstream upstream,
IHistoryWriter? history = null,
TimeSpan? scriptTimeout = null,
Func<DateTime>? clock = null)
{
var rootLogger = new LoggerConfiguration().CreateLogger();
return new VirtualTagEngine(
upstream,
new ScriptLoggerFactory(rootLogger),
rootLogger,
history,
clock,
scriptTimeout);
}
[Fact]
public async Task Simple_script_reads_upstream_and_returns_coerced_value()
{
var up = new FakeUpstream();
up.Set("InTag", 10.0);
using var engine = Build(up);
engine.Load([new VirtualTagDefinition(
Path: "LineRate",
DataType: DriverDataType.Float64,
ScriptSource: """return (double)ctx.GetTag("InTag").Value * 2.0;""")]);
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
var result = engine.Read("LineRate");
result.StatusCode.ShouldBe(0u);
result.Value.ShouldBe(20.0);
}
[Fact]
public async Task Upstream_change_triggers_cascade_through_two_levels()
{
var up = new FakeUpstream();
up.Set("A", 1.0);
using var engine = Build(up);
engine.Load([
new VirtualTagDefinition("B", DriverDataType.Float64,
"""return (double)ctx.GetTag("A").Value + 10.0;"""),
new VirtualTagDefinition("C", DriverDataType.Float64,
"""return (double)ctx.GetTag("B").Value * 2.0;"""),
]);
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
engine.Read("B").Value.ShouldBe(11.0);
engine.Read("C").Value.ShouldBe(22.0);
// Change upstream — cascade should recompute B (11→15.0) then C (30.0)
up.Push("A", 5.0);
await WaitForConditionAsync(() => Equals(engine.Read("B").Value, 15.0));
engine.Read("B").Value.ShouldBe(15.0);
engine.Read("C").Value.ShouldBe(30.0);
}
[Fact]
public async Task Cycle_in_virtual_tags_rejected_at_Load()
{
var up = new FakeUpstream();
using var engine = Build(up);
Should.Throw<DependencyCycleException>(() => engine.Load([
new VirtualTagDefinition("A", DriverDataType.Int32, """return (int)ctx.GetTag("B").Value + 1;"""),
new VirtualTagDefinition("B", DriverDataType.Int32, """return (int)ctx.GetTag("A").Value + 1;"""),
]));
await Task.CompletedTask;
}
[Fact]
public async Task Script_compile_error_surfaces_at_Load_with_all_failures()
{
var up = new FakeUpstream();
using var engine = Build(up);
var ex = Should.Throw<InvalidOperationException>(() => engine.Load([
new VirtualTagDefinition("A", DriverDataType.Int32, """return undefinedIdentifier;"""),
new VirtualTagDefinition("B", DriverDataType.Int32, """return 42;"""),
new VirtualTagDefinition("C", DriverDataType.Int32, """var x = anotherUndefined; return x;"""),
]));
ex.Message.ShouldContain("2 script(s) did not compile");
ex.Message.ShouldContain("A");
ex.Message.ShouldContain("C");
await Task.CompletedTask;
}
[Fact]
public async Task Script_runtime_exception_isolates_to_owning_tag()
{
var up = new FakeUpstream();
up.Set("OK", 10);
using var engine = Build(up);
engine.Load([
new VirtualTagDefinition("GoodTag", DriverDataType.Int32,
"""return (int)ctx.GetTag("OK").Value * 2;"""),
new VirtualTagDefinition("BadTag", DriverDataType.Int32,
"""throw new InvalidOperationException("boom");"""),
]);
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
engine.Read("GoodTag").StatusCode.ShouldBe(0u);
engine.Read("GoodTag").Value.ShouldBe(20);
engine.Read("BadTag").StatusCode.ShouldBe(0x80020000u, "BadInternalError for thrown script");
engine.Read("BadTag").Value.ShouldBeNull();
}
[Fact]
public async Task Timeout_maps_to_BadInternalError_without_killing_the_engine()
{
var up = new FakeUpstream();
using var engine = Build(up, scriptTimeout: TimeSpan.FromMilliseconds(30));
engine.Load([
new VirtualTagDefinition("Hang", DriverDataType.Int32, """
var end = Environment.TickCount64 + 5000;
while (Environment.TickCount64 < end) { }
return 1;
"""),
new VirtualTagDefinition("Ok", DriverDataType.Int32, """return 42;"""),
]);
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
engine.Read("Hang").StatusCode.ShouldBe(0x80020000u);
engine.Read("Ok").Value.ShouldBe(42);
}
[Fact]
public async Task Subscribers_receive_engine_emitted_changes()
{
var up = new FakeUpstream();
up.Set("In", 1);
using var engine = Build(up);
engine.Load([new VirtualTagDefinition(
"Out", DriverDataType.Int32, """return (int)ctx.GetTag("In").Value + 100;""")]);
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
var received = new List<DataValueSnapshot>();
using var sub = engine.Subscribe("Out", (p, v) => received.Add(v));
up.Push("In", 5);
await WaitForConditionAsync(() => received.Count >= 1);
received[^1].Value.ShouldBe(105);
}
[Fact]
public async Task Historize_flag_routes_to_history_writer()
{
var recorded = new List<(string, DataValueSnapshot)>();
var history = new TestHistory(recorded);
var up = new FakeUpstream();
up.Set("In", 1);
using var engine = Build(up, history);
engine.Load([
new VirtualTagDefinition("H", DriverDataType.Int32,
"""return (int)ctx.GetTag("In").Value + 1;""", Historize: true),
new VirtualTagDefinition("NoH", DriverDataType.Int32,
"""return (int)ctx.GetTag("In").Value - 1;""", Historize: false),
]);
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
recorded.Select(p => p.Item1).ShouldContain("H");
recorded.Select(p => p.Item1).ShouldNotContain("NoH");
}
[Fact]
public async Task Change_driven_false_ignores_upstream_push()
{
var up = new FakeUpstream();
up.Set("In", 1);
using var engine = Build(up);
engine.Load([new VirtualTagDefinition(
"Manual", DriverDataType.Int32,
"""return (int)ctx.GetTag("In").Value * 10;""",
ChangeTriggered: false)]);
// Initial eval seeds the value.
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
engine.Read("Manual").Value.ShouldBe(10);
// Upstream change fires but change-driven is off — no recompute.
up.Push("In", 99);
await Task.Delay(100);
engine.Read("Manual").Value.ShouldBe(10, "change-driven=false ignores upstream deltas");
}
[Fact]
public async Task Reload_replaces_existing_tags_and_resubscribes_cleanly()
{
var up = new FakeUpstream();
up.Set("A", 1);
up.Set("B", 2);
using var engine = Build(up);
engine.Load([new VirtualTagDefinition(
"T", DriverDataType.Int32, """return (int)ctx.GetTag("A").Value * 2;""")]);
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
engine.Read("T").Value.ShouldBe(2);
up.ActiveSubscriptionCount.ShouldBe(1);
// Reload — T now depends on B instead of A.
engine.Load([new VirtualTagDefinition(
"T", DriverDataType.Int32, """return (int)ctx.GetTag("B").Value * 3;""")]);
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
engine.Read("T").Value.ShouldBe(6);
up.ActiveSubscriptionCount.ShouldBe(1, "previous subscription on A must be disposed");
await Task.CompletedTask;
}
[Fact]
public async Task Dispose_releases_upstream_subscriptions()
{
var up = new FakeUpstream();
up.Set("A", 1);
var engine = Build(up);
engine.Load([new VirtualTagDefinition(
"T", DriverDataType.Int32, """return (int)ctx.GetTag("A").Value;""")]);
up.ActiveSubscriptionCount.ShouldBe(1);
engine.Dispose();
up.ActiveSubscriptionCount.ShouldBe(0);
await Task.CompletedTask;
}
[Fact]
public async Task SetVirtualTag_within_script_updates_target_and_triggers_observers()
{
var up = new FakeUpstream();
up.Set("In", 5);
using var engine = Build(up);
engine.Load([
new VirtualTagDefinition("Target", DriverDataType.Int32,
"""return 0;""", ChangeTriggered: false), // placeholder value, operator-written via SetVirtualTag
new VirtualTagDefinition("Driver", DriverDataType.Int32,
"""
var v = (int)ctx.GetTag("In").Value;
ctx.SetVirtualTag("Target", v * 100);
return v;
"""),
]);
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
engine.Read("Target").Value.ShouldBe(500);
engine.Read("Driver").Value.ShouldBe(5);
}
[Fact]
public async Task Type_coercion_from_script_double_to_config_int32()
{
var up = new FakeUpstream();
up.Set("In", 3.7);
using var engine = Build(up);
engine.Load([new VirtualTagDefinition(
"Rounded", DriverDataType.Int32,
"""return (double)ctx.GetTag("In").Value;""")]);
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
engine.Read("Rounded").Value.ShouldBe(4, "Convert.ToInt32 rounds 3.7 to 4");
}
private static async Task WaitForConditionAsync(Func<bool> cond, int timeoutMs = 2000)
{
var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs);
while (DateTime.UtcNow < deadline)
{
if (cond()) return;
await Task.Delay(25);
}
throw new TimeoutException("Condition did not become true in time");
}
private sealed class TestHistory : IHistoryWriter
{
private readonly List<(string, DataValueSnapshot)> _buf;
public TestHistory(List<(string, DataValueSnapshot)> buf) => _buf = buf;
public void Record(string path, DataValueSnapshot value)
{
lock (_buf) { _buf.Add((path, value)); }
}
}
}

View File

@@ -0,0 +1,132 @@
using Serilog;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests;
/// <summary>
/// Verifies the IReadable + ISubscribable adapter that DriverNodeManager dispatches
/// to for NodeSource.Virtual per ADR-002. Key contract: OPC UA clients see virtual
/// tags via the same capability interfaces as driver tags, so dispatch stays
/// source-agnostic.
/// </summary>
[Trait("Category", "Unit")]
public sealed class VirtualTagSourceTests
{
private static (VirtualTagEngine engine, VirtualTagSource source, FakeUpstream up) Build()
{
var up = new FakeUpstream();
up.Set("In", 10);
var logger = new LoggerConfiguration().CreateLogger();
var engine = new VirtualTagEngine(up, new ScriptLoggerFactory(logger), logger);
engine.Load([new VirtualTagDefinition(
"Out", DriverDataType.Int32, """return (int)ctx.GetTag("In").Value * 2;""")]);
return (engine, new VirtualTagSource(engine), up);
}
[Fact]
public async Task ReadAsync_returns_engine_cached_values()
{
var (engine, source, _) = Build();
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
var results = await source.ReadAsync(["Out"], TestContext.Current.CancellationToken);
results.Count.ShouldBe(1);
results[0].Value.ShouldBe(20);
results[0].StatusCode.ShouldBe(0u);
engine.Dispose();
}
[Fact]
public async Task ReadAsync_unknown_path_returns_Bad_quality()
{
var (engine, source, _) = Build();
var results = await source.ReadAsync(["NoSuchTag"], TestContext.Current.CancellationToken);
results[0].StatusCode.ShouldBe(0x80340000u);
engine.Dispose();
}
[Fact]
public async Task SubscribeAsync_fires_initial_data_callback()
{
var (engine, source, _) = Build();
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
var events = new List<DataChangeEventArgs>();
source.OnDataChange += (_, e) => events.Add(e);
var handle = await source.SubscribeAsync(["Out"], TimeSpan.FromMilliseconds(100),
TestContext.Current.CancellationToken);
handle.ShouldNotBeNull();
// Per OPC UA convention, initial-data callback fires on subscribe.
events.Count.ShouldBeGreaterThanOrEqualTo(1);
events[0].FullReference.ShouldBe("Out");
events[0].Snapshot.Value.ShouldBe(20);
await source.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
engine.Dispose();
}
[Fact]
public async Task SubscribeAsync_fires_on_upstream_change_via_engine_cascade()
{
var (engine, source, up) = Build();
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
var events = new List<DataChangeEventArgs>();
source.OnDataChange += (_, e) => events.Add(e);
var handle = await source.SubscribeAsync(["Out"], TimeSpan.Zero,
TestContext.Current.CancellationToken);
var initialCount = events.Count;
up.Push("In", 50);
// Wait for the cascade.
var deadline = DateTime.UtcNow.AddSeconds(2);
while (DateTime.UtcNow < deadline && events.Count <= initialCount) await Task.Delay(25);
events.Count.ShouldBeGreaterThan(initialCount);
events[^1].Snapshot.Value.ShouldBe(100);
await source.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
engine.Dispose();
}
[Fact]
public async Task UnsubscribeAsync_stops_further_events()
{
var (engine, source, up) = Build();
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
var events = new List<DataChangeEventArgs>();
source.OnDataChange += (_, e) => events.Add(e);
var handle = await source.SubscribeAsync(["Out"], TimeSpan.Zero,
TestContext.Current.CancellationToken);
await source.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
var countAfterUnsub = events.Count;
up.Push("In", 99);
await Task.Delay(200);
events.Count.ShouldBe(countAfterUnsub, "Unsubscribe must stop OnDataChange emissions");
engine.Dispose();
}
[Fact]
public async Task Null_arguments_rejected()
{
var (engine, source, _) = Build();
await Should.ThrowAsync<ArgumentNullException>(async () =>
await source.ReadAsync(null!, TestContext.Current.CancellationToken));
await Should.ThrowAsync<ArgumentNullException>(async () =>
await source.SubscribeAsync(null!, TimeSpan.Zero, TestContext.Current.CancellationToken));
await Should.ThrowAsync<ArgumentNullException>(async () =>
await source.UnsubscribeAsync(null!, TestContext.Current.CancellationToken));
engine.Dispose();
}
}

View File

@@ -0,0 +1,31 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<IsPackable>false</IsPackable>
<IsTestProject>true</IsTestProject>
<RootNamespace>ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests</RootNamespace>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="xunit.v3" Version="1.1.0"/>
<PackageReference Include="Shouldly" Version="4.3.0"/>
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.12.0"/>
<PackageReference Include="xunit.runner.visualstudio" Version="3.0.2">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\src\ZB.MOM.WW.OtOpcUa.Core.VirtualTags\ZB.MOM.WW.OtOpcUa.Core.VirtualTags.csproj"/>
</ItemGroup>
<ItemGroup>
<NuGetAuditSuppress Include="https://github.com/advisories/GHSA-37gx-xxp4-5rgx"/>
<NuGetAuditSuppress Include="https://github.com/advisories/GHSA-w3x6-4m5h-cxqf"/>
</ItemGroup>
</Project>