Merge #600.1/#603: provision changed-to-historized tags; retire dormant VirtualTagEngine; cache ScriptSandbox references; one LocalDb replication predicate for registration and drain gate

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-08 07:01:41 -04:00
parent 93e381561b 4f6231fa40
commit e9e35baa70
31 changed files with 220 additions and 2211 deletions
@@ -203,9 +203,9 @@ prefix. The concrete implementation `ScriptTagCatalog` reads `Tag` +
 literal against `DriverInstanceActor.AttributeValuePublished.FullReference`,
 which is the `FullName` field from `Tag.TagConfig`
 (see `ScriptTagCatalog.ExtractFullNameFromTagConfig` + `EquipmentNodeWalker.ExtractFullName`).
-The UNS-path engine (`Core.VirtualTags.VirtualTagEngine`, keyed by the
+The former UNS-path cascade engine (`Core.VirtualTags.VirtualTagEngine`, keyed by the
-slash-joined `Enterprise/Site/Area/Line/Equipment/TagName` browse path) is
+slash-joined `Enterprise/Site/Area/Line/Equipment/TagName` browse path) was
-dormant — it is not wired into the host — so UNS browse paths never resolve at
+retired (#603) — it was never wired into the host — so UNS browse paths never resolve at
 runtime and are intentionally NOT offered as completions.
 The catalog is scoped per Blazor circuit; each call creates and disposes its own
@@ -233,8 +233,8 @@ deploy time. An unrelated `anything.GetTag("…")` no longer triggers it.
 **`SetVirtualTag` is a no-op in production.** The live single-tag evaluator
 (`RoslynVirtualTagEvaluator`) drops cross-tag `ctx.SetVirtualTag(...)` writes
-(logged); the cascade `VirtualTagEngine` that honours them is dormant (not wired
+(logged); the cascade engine that would have honoured them (`VirtualTagEngine`) was retired
-into the host). So hover on a `ctx.SetVirtualTag("…")` path literal appends a
+unwired (#603). So hover on a `ctx.SetVirtualTag("…")` path literal appends a
 note that the write is dropped in single-tag mode and will not take effect at
 runtime — the path completions still help author the literal, but the author is
 not misled into relying on a no-op. Making `SetVirtualTag` functional would
@@ -60,9 +60,7 @@ Parses the script source with `CSharpSyntaxTree.ParseText` (script kind), walks
 ## Virtual tag engine (`Core.VirtualTags`)
-### `VirtualTagDefinition`
+> **Retired (#603):** the standalone cascade engine (`VirtualTagEngine`, `VirtualTagSource`, `VirtualTagDefinition`, `TimerTriggerScheduler`) had no production consumer and was deleted; virtual tags run through `VirtualTagHostActor` / `VirtualTagActor` + `RoslynVirtualTagEvaluator`. `Core.VirtualTags` now holds `DependencyGraph` (deploy-gate cycle check), `ITagUpstreamSource` and `IHistoryWriter`. Descriptions below of the engine's `Load`/cascade/timer behaviour are historical.
 One row per operator-authored tag. Fields: `Path` (UNS browse path; also the engine's internal id), `DataType` (`DriverDataType` enum; the evaluator coerces the script's return value to this and mismatch surfaces as `BadTypeMismatch`), `ScriptSource` (Roslyn C# script text), `ChangeTriggered` (re-evaluate on any input delta), `TimerInterval` (optional periodic cadence; null disables), `Historize` (route every evaluation result to `IHistoryWriter`). Change-trigger and timer are independent — a tag can be either, both, or neither.
 ### `VirtualTagContext`
@@ -77,56 +75,30 @@ Directed graph of tag paths. Edges run from a virtual tag to each path it reads.
 Cycle detection uses an **iterative** Tarjan's SCC implementation (no recursion, deep graphs cannot stack-overflow). Cycles of length > 1 and self-loops both reject; leaf references cannot form cycles with internal nodes.
 ### `VirtualTagEngine` lifecycle
 - **`Load(definitions)`** — clears prior state, compiles every script through `DependencyExtractor.Extract` + `ScriptEvaluator.Compile` (wrapped in `TimedScriptEvaluator`), registers each in `_tags` + `_graph`, runs `TopologicalSort` (cycle check), then for every upstream (non-virtual) path subscribes via `ITagUpstreamSource.SubscribeTag` and seeds `_valueCache` with `ReadTag`. Throws `InvalidOperationException` aggregating every compile failure at once so operators see the whole set; throws `DependencyCycleException` on cycles. Re-entrant — supports config-publish reloads by disposing the prior upstream subscriptions first.
 - **`EvaluateAllAsync(ct)`** — evaluates every tag once in topological order. Called at startup so virtual tags have a defined initial value before subscriptions start.
 - **`EvaluateOneAsync(path, ct)`** — single-tag evaluation. Entry point for `TimerTriggerScheduler` + tests.
 - **`Read(path)`** — synchronous last-known-value lookup from `_valueCache`. Returns `BadNodeIdUnknown`-quality for unregistered paths.
 - **`Subscribe(path, observer)`** — register a change observer; returns `IDisposable`. Does **not** emit a seed value.
 - **`OnUpstreamChange(path, value)`** (internal, wired from the upstream subscription) — updates cache, notifies observers, launches `CascadeAsync` fire-and-forget so the driver's dispatcher isn't blocked.
 Evaluations are **serial across all tags** — `_evalGate` is a `SemaphoreSlim(1, 1)` held around every `EvaluateInternalAsync`. Parallelism is deferred (Phase 7 plan decision #19). Rationale: serial execution preserves the "earlier topological nodes computed before later dependents" invariant when two cascades race. Per-tag error isolation: a script exception or timeout sets that tag's quality to `BadInternalError` and logs a structured error; other tags keep evaluating. `OperationCanceledException` is re-thrown (shutdown path).
 Result coercion: `CoerceResult` maps the script's return value to the declared `DriverDataType` via `Convert.ToXxx`. Coercion failure returns null which the outer pipeline maps to `BadInternalError`; `BadTypeMismatch` is documented in the definition shape (`VirtualTagDefinition.DataType` doc) rather than emitted distinctly today.
 `IHistoryWriter.Record` fires per evaluation when `Historize = true`. The default `NullHistoryWriter` drops silently.
 ### `TimerTriggerScheduler`
 Groups `VirtualTagDefinition`s by `TimerInterval`, one `System.Threading.Timer` per unique interval. Each tick evaluates the group's paths serially via `VirtualTagEngine.EvaluateOneAsync`. Errors per-tag log and continue. `Dispose()` cancels an internal `CancellationTokenSource` and disposes every timer. Independent of the change-trigger path — a tag with both triggers fires from both scheduling sources.
 ### `ITagUpstreamSource`
-What the engine pulls driver-tag values from. Reads are **synchronous** because user scripts call `ctx.GetTag(path)` inline — a blocking wire call per evaluation would kill throughput. Implementations are expected to serve from a last-known-value cache populated by subscription callbacks. The server's production implementation is `CachedTagUpstreamSource` (see Composition below).
+What a value consumer (scripted alarms) pulls driver-tag values from. Reads are **synchronous** because user scripts call `ctx.GetTag(path)` inline — a blocking wire call per evaluation would kill throughput. Implementations are expected to serve from a last-known-value cache populated by subscription callbacks. The server's production implementation is `CachedTagUpstreamSource` (see Composition below).
 ### `IHistoryWriter`
-Fire-and-forget sink for evaluation results when `VirtualTagDefinition.Historize = true`. Implementations must queue internally and drain on their own cadence — a slow historian must not block script evaluation. `NullHistoryWriter.Instance` is the no-op default. Scripted-alarm emissions flow through `Core.AlarmHistorian` via `HistorianAdapterActor` → the registered `IAlarmHistorianSink` (a separate concern; see [AlarmTracking.md](AlarmTracking.md)). *(The `Phase7EngineComposer.RouteToHistorianAsync` this used to name is gone — not renamed; the routing moved into the actor.)*
+Fire-and-forget sink for evaluation results when a virtual tag is `Historize`d. Implementations must queue internally and drain on their own cadence — a slow historian must not block script evaluation. `NullHistoryWriter.Instance` is the no-op default. Scripted-alarm emissions flow through `Core.AlarmHistorian` via `HistorianAdapterActor` → the registered `IAlarmHistorianSink` (a separate concern; see [AlarmTracking.md](AlarmTracking.md)). *(The `Phase7EngineComposer.RouteToHistorianAsync` this used to name is gone — not renamed; the routing moved into the actor.)*
 **Equipment-namespace path (H5).** The `Historize` flag is threaded end-to-end on the equipment path: `VirtualTag.Historize` → composer + artifact-decode (byte-parity) → `EquipmentVirtualTagPlan.Historize` → `VirtualTagHostActor`, which calls `IHistoryWriter.Record(nodeId, snapshot)` for every historized result (in addition to publishing the live value). The writer is injectable via DI — `DriverHostActor` resolves `IHistoryWriter` (`TryAddSingleton`, `NullHistoryWriter` default) and threads it into `VirtualTagHostActor`. **This `IHistoryWriter` seam still ships no durable binding** (`NullHistoryWriter` default). Durable continuous historization of driver/virtual values is now handled by the separate `ContinuousHistorizationRecorder` (it taps the dependency-mux value fan-out → a crash-safe FasterLog outbox → the HistorianGateway's `WriteLiveValues` path; see [Historian.md](Historian.md)), not through this seam. A deployment can still bind a custom `IHistoryWriter` via DI.
 ## Dispatch integration
-Per [ADR-002](v2/implementation/adr-002-driver-vs-virtual-dispatch.md) Option B, there is a single `GenericDriverNodeManager`. `VirtualTagSource` implements `IReadable` + `ISubscribable` over a `VirtualTagEngine`:
+Virtual-tag nodes are materialised by `AddressSpaceApplier` and fed by `VirtualTagActor` publishes; they are not client-writable (`AccessLevel = CurrentRead`), scripts being the only write path via `ctx.SetVirtualTag`. See [ReadWriteOperations.md](ReadWriteOperations.md).
 - `ReadAsync` fans each path through `engine.Read(...)`.
 - `SubscribeAsync` calls `engine.Subscribe` per path and forwards each engine observer callback as an `OnDataChange` event; emits an initial-data callback per OPC UA convention.
 - `UnsubscribeAsync` disposes every per-path engine subscription it holds.
 - **`IWritable` is deliberately not implemented.** Virtual-tag nodes are not client-writable because `OtOpcUaNodeManager.EnsureVariable` materialises every SDK variable with `AccessLevel = AccessLevels.CurrentRead`; the SDK base `CustomNodeManager2.Write` returns `BadNotWritable` for read-only nodes and v2 has no client-write dispatch path. Scripts are the only write path via `ctx.SetVirtualTag`.
 `NodeSourceKind` on each `DriverAttributeInfo` (set by `EquipmentNodeWalker` at address-space build time) drives which backend handles a read. See [ReadWriteOperations.md](ReadWriteOperations.md) and [v1/Subscriptions.md](v1/Subscriptions.md) for the broader dispatch framing.
 ## Upstream reads + history
-`ITagUpstreamSource` and `IHistoryWriter` are the two ports the engine requires from its host. Both live in `Core.VirtualTags`. In the v2 actor system:
+`ITagUpstreamSource` and `IHistoryWriter` are the two ports value consumers require from their host. Both live in `Core.VirtualTags`. In the v2 actor system:
 - **Upstream-tag feed.** `DependencyMuxActor` (`src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs`) routes `DriverInstanceActor.AttributeValuePublished` events to the `VirtualTagActor` instances that declared interest in those tag refs. Each `VirtualTagActor` holds the in-memory per-tag dependency map; the `IVirtualTagEvaluator` (`RoslynVirtualTagEvaluator`) receives the dependency snapshot synchronously on the actor message thread. Reads of never-pushed dependency refs return `null` values in the dependency snapshot.
- **`IHistoryWriter`** — the equipment-namespace path threads `Historize` end-to-end and `VirtualTagHostActor` invokes the injected writer on historized results (H5); the writer is resolved through `DriverHostActor` DI with a `NullHistoryWriter` default. The standalone `VirtualTagEngine` likewise receives `NullHistoryWriter` by default. No *durable* writer ships on this seam — durable continuous historization now flows through the separate `ContinuousHistorizationRecorder` → HistorianGateway `WriteLiveValues` path (see the `IHistoryWriter` section above and [Historian.md](Historian.md)).
+- **`IHistoryWriter`** — the equipment-namespace path threads `Historize` end-to-end and `VirtualTagHostActor` invokes the injected writer on historized results (H5); the writer is resolved through `DriverHostActor` DI with a `NullHistoryWriter` default. No *durable* writer ships on this seam — durable continuous historization now flows through the separate `ContinuousHistorizationRecorder` → HistorianGateway `WriteLiveValues` path (see the `IHistoryWriter` section above and [Historian.md](Historian.md)).
 ## Composition
-`AddressSpaceComposer` (`src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceComposer.cs`) is a pure static function that flattens config-DB entities into an `AddressSpaceComposition` value (UNS topology + raw subtree + driver-instance plans + scripted-alarm plans). `AddressSpaceApplier` applies that result into the OPC UA SDK node manager. Neither class has knowledge of `VirtualTagEngine` or `ScriptedAlarmEngine`.
+`AddressSpaceComposer` (`src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceComposer.cs`) is a pure static function that flattens config-DB entities into an `AddressSpaceComposition` value (UNS topology + raw subtree + driver-instance plans + scripted-alarm plans). `AddressSpaceApplier` applies that result into the OPC UA SDK node manager. Neither class has knowledge of `ScriptedAlarmEngine`.
 In the v2 actor system, virtual-tag engine composition is owned by the driver-role host actor tree:
@@ -134,7 +106,7 @@ In the v2 actor system, virtual-tag engine composition is owned by the driver-ro
 - `RoslynVirtualTagEvaluator` (`src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs`) is injected into each `VirtualTagActor` as its `IVirtualTagEvaluator`. It holds a per-source `CompiledScriptCache` keyed by script source and compiles on first use.
 - `DependencyMuxActor` (`src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs`) receives every `DriverInstanceActor.AttributeValuePublished` event and routes it to the `VirtualTagActor` instances that registered interest in that tag ref.
-`VirtualTagEngine`, `VirtualTagSource`, `TimerTriggerScheduler`, and `ITagUpstreamSource` are available as standalone Core.VirtualTags primitives and remain the correct composition path for non-actor deployments (integration tests, future standalone runtimes).
+`ITagUpstreamSource` and `IHistoryWriter` remain available as `Core.VirtualTags` ports.
 ## Key source files
@@ -148,14 +120,10 @@ In the v2 actor system, virtual-tag engine composition is owned by the driver-ro
 - `src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/DependencyExtractor.cs` — static `ctx.GetTag` / `ctx.SetVirtualTag` inference
 - `src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLoggerFactory.cs` — per-script Serilog logger
 - `src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogCompanionSink.cs` — error mirror to main log
 - `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagDefinition.cs` — per-tag config record
 - `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagContext.cs` — evaluation-scoped `ctx`
 - `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/DependencyGraph.cs` — Kahn topo-sort + iterative Tarjan SCC
 - `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagEngine.cs` — load / evaluate / cascade pipeline
 - `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/TimerTriggerScheduler.cs` — periodic re-evaluation
 - `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/ITagUpstreamSource.cs` — driver-tag read + subscribe port
 - `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/IHistoryWriter.cs` — historize sink port + `NullHistoryWriter`
 - `src/Core/ZB.MOM.WW.OtOpcUa.Core.VirtualTags/VirtualTagSource.cs` — `IReadable` + `ISubscribable` adapter
 - `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs` — actor that receives `DependencyValueChanged` from the mux and invokes `IVirtualTagEvaluator` per expression
 - `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs` — routes `DriverInstanceActor.AttributeValuePublished` to interested `VirtualTagActor` subscribers
 - `src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs` — production `IVirtualTagEvaluator` binding; holds a per-source `CompiledScriptCache`
@@ -5,8 +5,7 @@ namespace ZB.MOM.WW.OtOpcUa.Commons.Engines;
 /// <summary>
 ///     Abstraction over the compiled virtual-tag expression engine. Runtime consumes this so
 ///     <see cref="VirtualTagActor"/> can stay free of Roslyn / scripting machinery and the
-///     production wiring binds an adapter over <c>VirtualTagEngine</c> from
+///     production wiring binds <c>RoslynVirtualTagEvaluator</c> (Host).
 ///     <c>Core.VirtualTags</c>.
 /// </summary>
 public interface IVirtualTagEvaluator
 {
@@ -98,7 +97,7 @@ public static class ScriptEvalFailureReasons
 }
 /// <summary>Returns <see cref="VirtualTagEvalResult.NoChange"/> from every call. Bound by default
-/// when the production <c>VirtualTagEngine</c> adapter hasn't been registered (Mac dev, tests).</summary>
+/// when the production <c>RoslynVirtualTagEvaluator</c> hasn't been registered (Mac dev, tests).</summary>
 public sealed class NullVirtualTagEvaluator : IVirtualTagEvaluator
 {
    public static readonly NullVirtualTagEvaluator Instance = new();
@@ -39,7 +39,7 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
 /// </param>
 /// <param name="VirtualTagId">
 ///     Set when <paramref name="Source"/> is <see cref="NodeSourceKind.Virtual"/> — stable
-///     logical id the VirtualTagEngine addresses by. Null otherwise.
+///     logical id the virtual-tag runtime addresses by. Null otherwise.
 /// </param>
 /// <param name="ScriptedAlarmId">
 ///     Set when <paramref name="Source"/> is <see cref="NodeSourceKind.ScriptedAlarm"/> —
@@ -78,7 +78,7 @@ public sealed record DriverAttributeInfo(
 ///     Discriminates which runtime subsystem owns this node's Read/Write/
 ///     Subscribe dispatch. <c>Driver</c> = a real IDriver capability surface;
 ///     <c>Virtual</c> = a Phase 7 <see cref="DriverAttributeInfo"/>.VirtualTagId'd tag
-///     computed by the VirtualTagEngine; <c>ScriptedAlarm</c> = a scripted Part 9 alarm
+///     computed by the virtual-tag runtime; <c>ScriptedAlarm</c> = a scripted Part 9 alarm
 ///     materialized by the ScriptedAlarmEngine.
 /// </summary>
 public enum NodeSourceKind
@@ -8,9 +8,7 @@ namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
 ///     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
 ///     <c>ITagUpstreamSource</c> subscription, virtual tags updated by prior
-///     evaluations). Writes route through <c>VirtualTagEngine</c>'s
+///     evaluations). Writes route through the host-supplied set-virtual-tag callback.
 ///     <c>OnScriptSetVirtualTag</c> callback so cross-tag write side effects still
 ///     participate in change-trigger cascades (via the engine's <c>CascadeAsync</c>).
 /// </summary>
 /// <remarks>
 ///     <para>
@@ -81,10 +81,12 @@ public static class ScriptSandbox
        // surface (operator-authored hosting helpers, third-party packages, app code)
        // is not on the list and stays unreachable.
        var references = new List<MetadataReference>();
        // #603: both halves are cached. CreateFromFile reads + maps the DLL, and the BCL half is ~170
        // assemblies, so doing it per compile was the dominant fixed cost of every ScriptEvaluator build.
        // MetadataReference is immutable and thread-safe, so sharing instances across compilations is safe.
        foreach (var asm in pinnedAssemblies)
-            references.Add(MetadataReference.CreateFromFile(asm.Location));
+            references.Add(PinnedReferences.GetOrAdd(asm, static a => MetadataReference.CreateFromFile(a.Location)));
-        foreach (var path in EnumerateBclAssemblyPaths())
+        references.AddRange(BclReferences.Value);
            references.Add(MetadataReference.CreateFromFile(path));
        var imports = new[]
        {
@@ -97,6 +99,13 @@ public static class ScriptSandbox
        return new SandboxConfig(references, imports);
    }
    private static readonly System.Collections.Concurrent.ConcurrentDictionary<System.Reflection.Assembly, MetadataReference> PinnedReferences = new();
    // The trusted-platform set is process-constant, so the BCL reference list is context-independent.
    private static readonly Lazy<IReadOnlyList<MetadataReference>> BclReferences = new(
        static () => EnumerateBclAssemblyPaths().Select(static p => (MetadataReference)MetadataReference.CreateFromFile(p)).ToArray(),
        LazyThreadSafetyMode.ExecutionAndPublication);
    private static IEnumerable<string> EnumerateBclAssemblyPaths()
    {
        // The .NET host advertises the resolved runtime-shared-framework + BCL DLL set
@@ -15,11 +15,8 @@ namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
 ///         from a last-known-value cache populated by the subscription callbacks.
 ///     </para>
 ///     <para>
-///         The subscription path feeds <see cref="VirtualTagEngine"/>'s
+///         The subscription path feeds the consumer's change callback (scripted alarms via
-///         <c>OnUpstreamChange</c> callback, which updates the engine's value cache and
+///         <c>DependencyMuxTagUpstreamSource</c>). One subscription per distinct upstream tag path.
 ///         schedules <c>CascadeAsync</c> to re-evaluate every change-driven dependent in
 ///         topological order. One subscription per distinct upstream tag path; the
 ///         engine tracks the mapping itself.
 ///     </para>
 /// </remarks>
 public interface ITagUpstreamSource
@@ -1,162 +0,0 @@
 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>
 /// <remarks>
 ///     <para>
 ///         Each timer group carries a per-group in-flight flag (see
 ///         <c>TickGroup.InFlight</c>). When the timer fires while a tick for the same
 ///         group is still running, the new callback skips the work and increments
 ///         <see cref="SkippedTickCount"/> rather than blocking a thread-pool thread on
 ///         the engine's evaluation gate. This bounds the work outstanding at one tick
 ///         per group, regardless of how long an individual evaluation takes.
 ///     </para>
 /// </remarks>
 public sealed class TimerTriggerScheduler : IDisposable
 {
    private readonly VirtualTagEngine _engine;
    private readonly ILogger _logger;
    private readonly List<Timer> _timers = [];
    private readonly List<TickGroup> _groups = [];
    private readonly CancellationTokenSource _cts = new();
    private long _skippedTickCount;
    private bool _disposed;
    private bool _started;
    /// <summary>Initializes a new instance of the <see cref="TimerTriggerScheduler"/> class.</summary>
    /// <param name="engine">The virtual tag engine to trigger evaluations on.</param>
    /// <param name="logger">A logger for diagnostic output.</param>
    public TimerTriggerScheduler(VirtualTagEngine engine, ILogger logger)
    {
        _engine = engine ?? throw new ArgumentNullException(nameof(engine));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
    }
    /// <summary>
    ///     Diagnostic counter: number of timer callbacks that skipped their work because
    ///     the prior tick for the same group was still running. Exposed for tests +
    ///     operational metrics. Monotonic; never resets.
    /// </summary>
    public long SkippedTickCount => Interlocked.Read(ref _skippedTickCount);
    /// <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>
    /// <param name="definitions">The virtual tag definitions to schedule timers for.</param>
    /// <exception cref="InvalidOperationException">Thrown when <c>Start</c> is called more than once on the same instance.</exception>
    public void Start(IReadOnlyList<VirtualTagDefinition> definitions)
    {
        if (_disposed) throw new ObjectDisposedException(nameof(TimerTriggerScheduler));
        if (_started) throw new InvalidOperationException(
            "TimerTriggerScheduler.Start has already been called. Create a new instance for each Load cycle.");
        _started = true;
        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 ctx = new TickGroup(paths);
            _groups.Add(ctx);
            var timer = new Timer(_ => OnTimer(ctx), null, interval, interval);
            _timers.Add(timer);
            _logger.Information("TimerTriggerScheduler: {TagCount} tag(s) on {Interval} cadence",
                paths.Length, interval);
        }
    }
    private void OnTimer(TickGroup ctx)
    {
        if (_cts.IsCancellationRequested) return;
        // Skip the tick when the prior one for this group is still running. Without
        // this guard a slow evaluation (or one waiting on the engine's _evalGate) would
        // cause subsequent timer callbacks to each pin a thread-pool thread on the
        // gate, compounding under high tick rates.
        if (Interlocked.CompareExchange(ref ctx.InFlight, 1, 0) != 0)
        {
            Interlocked.Increment(ref _skippedTickCount);
            return;
        }
        // Run async without blocking the timer's pool-thread callback. The task is
        // fire-and-forget — failures are logged inside RunTickAsync; the InFlight flag
        // is reset in the finally block so the next tick can proceed.
        _ = RunTickAsync(ctx);
    }
    private async Task RunTickAsync(TickGroup ctx)
    {
        try
        {
            foreach (var p in ctx.Paths)
            {
                if (_cts.IsCancellationRequested) return;
                try
                {
                    await _engine.EvaluateOneAsync(p, _cts.Token).ConfigureAwait(false);
                }
                catch (OperationCanceledException)
                {
                    return;
                }
                catch (Exception ex)
                {
                    _logger.Error(ex, "TimerTriggerScheduler evaluate failed for {Path}", p);
                }
            }
        }
        finally
        {
            Interlocked.Exchange(ref ctx.InFlight, 0);
        }
    }
    /// <summary>Releases all timers and disposes the scheduler's resources.</summary>
    public void Dispose()
    {
        if (_disposed) return;
        _disposed = true;
        _cts.Cancel();
        foreach (var t in _timers)
        {
            try { t.Dispose(); } catch { }
        }
        _timers.Clear();
        _groups.Clear();
        _cts.Dispose();
    }
    /// <summary>Groups tags that share the same timer interval.</summary>
    private sealed class TickGroup
    {
        // 0 = idle, 1 = a tick is currently running (or queued) for this group. Use
        // Interlocked.CompareExchange so a timer callback observes a consistent "is the
        // prior tick still running" answer without taking a lock.
        /// <summary>Flag indicating whether a tick is currently in flight for this group.</summary>
        public int InFlight;
        /// <summary>Gets the list of tag paths in this group.</summary>
        public IReadOnlyList<string> Paths { get; }
        /// <summary>Initializes a new instance of the <see cref="TickGroup"/> class.</summary>
        /// <param name="paths">The tag paths in this timer group.</param>
        public TickGroup(IReadOnlyList<string> paths)
        {
            Paths = paths;
        }
    }
 }
@@ -1,41 +0,0 @@
 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 end at
 ///     <see cref="VirtualTagEngine"/>'s <c>EvaluateInternalAsync</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);
@@ -1,568 +0,0 @@
 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>
 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);
    /// <summary>
    ///     Compile cache for every virtual-tag script. Routes <see cref="Load"/>'s
    ///     <see cref="ScriptEvaluator{TContext, TResult}.Compile"/> calls through the
    ///     cache so the collectible <see cref="System.Runtime.Loader.AssemblyLoadContext"/>
    ///     each compile produces is actually disposed on the publish-replace path: the
    ///     cache's <see cref="CompiledScriptCache{TContext, TResult}.Clear"/> disposes
    ///     every materialised evaluator before dropping its dictionary entry, so a
    ///     config-publish releases the prior generation's ALCs and the per-publish
    ///     accretion is actually freed in production. Pre-fix, the engine called
    ///     <c>ScriptEvaluator.Compile</c> directly, which left the ALCs rooted until
    ///     the process exited.
    /// </summary>
    private readonly CompiledScriptCache<VirtualTagContext, object?> _compileCache = new();
    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;
    /// <summary>
    /// Initializes a new instance of the <see cref="VirtualTagEngine"/> class.
    /// </summary>
    /// <param name="upstream">The upstream tag source for reading input tags.</param>
    /// <param name="loggerFactory">Factory for creating script loggers.</param>
    /// <param name="engineLogger">Logger for engine events and errors.</param>
    /// <param name="historyWriter">Optional history writer for recording tag evaluations; defaults to no-op.</param>
    /// <param name="clock">Optional function providing current UTC time; defaults to <see cref="DateTime.UtcNow"/>.</param>
    /// <param name="scriptTimeout">Optional timeout for script evaluation; defaults to <see cref="TimedScriptEvaluator{TContext, TResult}.DefaultTimeout"/>.</param>
    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>
    /// <param name="definitions">List of virtual tag definitions to load and compile.</param>
    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();
        _compileCache.Clear();
        var compileFailures = new List<string>();
        var seenPaths = new HashSet<string>(StringComparer.Ordinal);
        foreach (var def in definitions)
        {
            if (!seenPaths.Add(def.Path))
            {
                compileFailures.Add($"{def.Path}: duplicate path — only one definition per path is allowed");
                continue;
            }
            if (!IsSupportedDataType(def.DataType))
            {
                compileFailures.Add(
                    $"{def.Path}: unsupported DataType DriverDataType.{def.DataType} — virtual tags only support scalar primitive types");
                continue;
            }
            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 = _compileCache.GetOrCompile(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}");
            }
        }
        // Validate every ctx.SetVirtualTag write target resolves to a registered virtual
        // tag. A script writing to a non-existent virtual path would otherwise be silently
        // dropped at runtime by OnScriptSetVirtualTag's warning-and-drop branch; catching
        // it here surfaces operator typos as a publish failure.
        foreach (var (path, state) in _tags)
        {
            foreach (var writeTarget in state.Writes)
            {
                if (!_tags.ContainsKey(writeTarget))
                {
                    compileFailures.Add(
                        $"{path}: ctx.SetVirtualTag target '{writeTarget}' is not a registered virtual tag");
                }
            }
        }
        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. Iterate _tags.Values (the registered set) rather
        // than definitions to avoid indexing by a raw input list that may contain duplicates.
        var upstreamPaths = _tags.Values
            .SelectMany(s => s.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>
    /// <param name="ct">Cancellation token to stop evaluation.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    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>
    /// <param name="path">Path of the virtual tag to evaluate.</param>
    /// <param name="ct">Cancellation token to stop evaluation.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    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>
    /// <param name="path">Path of the tag to read.</param>
    /// <returns>The tag's most recently evaluated value, or a Bad-quality snapshot if the path is unrecognized.</returns>
    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>
    /// <param name="path">Path of the tag to subscribe to.</param>
    /// <param name="observer">Callback invoked with the tag path and new value on each evaluation.</param>
    /// <returns>An <see cref="IDisposable"/> that removes the observer when disposed.</returns>
    public IDisposable Subscribe(string path, Action<string, DataValueSnapshot> observer)
    {
        // Race-safe pattern paired with Unsub.Dispose: if Unsub.Dispose removed the
        // dictionary entry between our GetOrAdd and the lock-protected Add, the list
        // we hold a reference to is orphaned. Re-check the map under the lock and
        // re-insert the list (or grab the current one) if needed, retrying until the
        // dictionary observably contains the list we just added our observer to.
        while (true)
        {
            var list = _observers.GetOrAdd(path, _ => []);
            lock (list)
            {
                // Confirm the list is still the dictionary's value for this key. If
                // Dispose removed the entry, _observers[path] either doesn't exist or
                // points at a different (newer) list — retry.
                if (_observers.TryGetValue(path, out var current) && ReferenceEquals(current, list))
                {
                    list.Add(observer);
                    return new Unsub(this, path, observer);
                }
            }
            // Lost the race — Dispose pruned the list out from under us. Loop and
            // either re-create or pick up the newer list.
        }
    }
    /// <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>
    /// <param name="path">Path of the upstream tag that changed.</param>
    /// <param name="value">New value of the tag.</param>
    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;
        await _evalGate.WaitAsync(ct).ConfigureAwait(false);
        try
        {
            var ctxCache = BuildReadCache(state.Reads);
            // Cold-start guard — when any upstream input is still unset or Bad-quality,
            // publish a BadWaitingForInitialData snapshot so OPC UA clients see a defined
            // quality rather than observing "not yet computed" as a stale Good value.
            // Evaluating with nulls would throw inside the script (scripts cast
            // ctx.GetTag(path).Value directly) and produce a persistent BadInternalError.
            if (!AreInputsReady(ctxCache))
            {
                var notReady = new DataValueSnapshot(null, 0x80320000u /* BadWaitingForInitialData */, null, _clock());
                _valueCache[path] = notReady;
                NotifyObservers(path, notReady);
                return;
            }
            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);
                // null from CoerceResult means the conversion threw (raw was non-null but
                // not convertible to the declared type). Surface as BadInternalError so
                // the OPC UA client sees a defined Bad quality rather than a Good null.
                result = (raw is not null && coerced is null)
                    ? new DataValueSnapshot(null, 0x80020000u /* BadInternalError */, null, _clock())
                    : 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;
    }
    /// <summary>
    ///     Returns true when every entry in <paramref name="cache"/> has a
    ///     Good/Uncertain-quality <see cref="DataValueSnapshot.StatusCode"/> (bit 31 = 0).
    ///     Readiness is gated on <c>StatusCode</c> alone — a Good-quality snapshot with a
    ///     null <c>Value</c> (e.g. a script that returns <c>null</c>, or a
    ///     <c>ctx.SetVirtualTag</c> call with a null argument) is considered ready and the
    ///     downstream script is allowed to run. The downstream script will observe the null
    ///     value and typically throw an NRE, which the outer evaluation catch maps to
    ///     <c>BadInternalError</c> — the correct sentinel for "script ran but faulted"
    ///     rather than "inputs not yet available". Previously the null-value check
    ///     caused a Good-quality null upstream to permanently block all its dependents at
    ///     <c>BadWaitingForInitialData</c>.
    /// </summary>
    private static bool AreInputsReady(IReadOnlyDictionary<string, DataValueSnapshot> cache)
    {
        foreach (var kv in cache)
        {
            if (kv.Value is null) return false;
            if ((kv.Value.StatusCode & 0x80000000u) != 0) return false;
        }
        return true;
    }
    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);
        // A cross-tag write must participate in the change-trigger cascade, exactly
        // like an upstream delta — any change-triggered tag that reads this path
        // would otherwise go stale until an unrelated trigger fires (see
        // docs/VirtualTags.md, VirtualTagContext section). Fire-and-forget: this
        // callback runs inside EvaluateInternalAsync with the non-reentrant
        // _evalGate held, so the cascade must be scheduled, not invoked inline.
        _ = CascadeAsync(path, CancellationToken.None);
    }
    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.Int16 => Convert.ToInt16(raw),
                DriverDataType.Int32 => Convert.ToInt32(raw),
                DriverDataType.Int64 => Convert.ToInt64(raw),
                DriverDataType.UInt16 => Convert.ToUInt16(raw),
                DriverDataType.UInt32 => Convert.ToUInt32(raw),
                DriverDataType.UInt64 => Convert.ToUInt64(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),
                // Any DriverDataType not in the explicit list (currently Reference, or any
                // future enum member added without coercion support) must NOT silently
                // return the uncoerced raw value — that would surface as a wire-level
                // type mismatch on the OPC UA Variant. Throwing here is caught by the
                // outer catch and mapped to BadInternalError. Load-time validation in
                // IsSupportedDataType ensures operators never publish such a tag.
                _ => throw new InvalidOperationException(
                    $"Virtual-tag CoerceResult does not support DriverDataType.{target}"),
            };
        }
        catch
        {
            // Caller maps null to BadInternalError (0x80020000) — we let null propagate
            // so the outer evaluation path sets the Bad quality on the snapshot.
            return null;
        }
    }
    /// <summary>
    ///     The set of <see cref="DriverDataType"/> values <see cref="CoerceResult"/> can
    ///     honour. Definitions declaring any other type are rejected at <see cref="Load"/>
    ///     so an operator typo (or a future enum member added without coercion support) is
    ///     caught at publish time rather than silently producing a type-mismatched value.
    /// </summary>
    private static bool IsSupportedDataType(DriverDataType t) => t switch
    {
        DriverDataType.Boolean => true,
        DriverDataType.Int16 => true,
        DriverDataType.Int32 => true,
        DriverDataType.Int64 => true,
        DriverDataType.UInt16 => true,
        DriverDataType.UInt32 => true,
        DriverDataType.UInt64 => true,
        DriverDataType.Float32 => true,
        DriverDataType.Float64 => true,
        DriverDataType.String => true,
        DriverDataType.DateTime => true,
        _ => false,
    };
    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.");
    }
    /// <summary>
    /// Releases all resources held by the engine, including upstream subscriptions and compiled script caches.
    /// </summary>
    public void Dispose()
    {
        if (_disposed) return;
        _disposed = true;
        UnsubscribeFromUpstream();
        _tags.Clear();
        _graph.Clear();
        _compileCache.Dispose();
    }
    /// <summary>
    /// Gets the dependency graph for testing purposes.
    /// </summary>
    internal DependencyGraph GraphForTesting => _graph;
    private sealed class Unsub : IDisposable
    {
        private readonly VirtualTagEngine _engine;
        private readonly string _path;
        private readonly Action<string, DataValueSnapshot> _observer;
        /// <summary>
        /// Initializes a new instance of the <see cref="Unsub"/> class.
        /// </summary>
        /// <param name="e">The parent virtual tag engine.</param>
        /// <param name="path">Path of the subscribed tag.</param>
        /// <param name="observer">The observer callback to unsubscribe.</param>
        public Unsub(VirtualTagEngine e, string path, Action<string, DataValueSnapshot> observer)
        {
            _engine = e; _path = path; _observer = observer;
        }
        /// <summary>
        /// Removes the observer from the subscription list.
        /// </summary>
        public void Dispose()
        {
            if (_engine._observers.TryGetValue(_path, out var list))
            {
                lock (list)
                {
                    list.Remove(_observer);
                    // If we removed the last observer, prune the dictionary entry so a
                    // long-running server doesn't accumulate empty Lists for paths that
                    // saw transient subscriptions. The emptiness check is inside the same
                    // lock so a concurrent Subscribe can't slip an observer in after we
                    // observe the list as empty.
                    if (list.Count == 0)
                    {
                        // ICollection<KeyValuePair<,>> removal is value-typed — only removes
                        // if both key + value still match (i.e. the dictionary still points
                        // at this list). This keeps a racing Subscribe's brand-new list safe.
                        ((ICollection<KeyValuePair<string, List<Action<string, DataValueSnapshot>>>>)_engine._observers)
                            .Remove(new KeyValuePair<string, List<Action<string, DataValueSnapshot>>>(_path, list));
                    }
                }
            }
        }
    }
    internal sealed record VirtualTagState(
        VirtualTagDefinition Definition,
        TimedScriptEvaluator<VirtualTagContext, object?> Evaluator,
        IReadOnlySet<string> Reads,
        IReadOnlySet<string> Writes,
        ILogger Logger);
 }
@@ -1,106 +0,0 @@
 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>. 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. 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);
    /// <summary>
    ///     Initializes a new instance of the <see cref="VirtualTagSource"/> class.
    /// </summary>
    /// <param name="engine">The virtual tag engine.</param>
    public VirtualTagSource(VirtualTagEngine engine)
    {
        _engine = engine ?? throw new ArgumentNullException(nameof(engine));
    }
    /// <summary>
    ///     Occurs when data values change.
    /// </summary>
    public event EventHandler<DataChangeEventArgs>? OnDataChange;
    /// <inheritdoc />
    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);
    }
    /// <inheritdoc />
    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);
        // OPC UA convention: for each path, emit the initial-data callback BEFORE
        // registering the change observer. This prevents a race where an upstream change
        // fires the observer between the Subscribe call and the Read call, which would
        // deliver a newer change event before the initial-data event, leaving the client
        // with a stale last-known value.
        foreach (var path in fullReferences)
        {
            var snap = _engine.Read(path);
            OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, path, snap));
            observers.Add(_engine.Subscribe(path, (p, s) =>
                OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, p, s))));
        }
        _subs[handle.DiagnosticId] = new Subscription(handle, observers);
        return Task.FromResult<ISubscriptionHandle>(handle);
    }
    /// <inheritdoc />
    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
    {
        /// <summary>
        ///     Initializes a new instance of the <see cref="SubscriptionHandle"/> class.
        /// </summary>
        /// <param name="id">The subscription identifier.</param>
        public SubscriptionHandle(string id) { DiagnosticId = id; }
        /// <inheritdoc />
        public string DiagnosticId { get; }
    }
    private sealed record Subscription(SubscriptionHandle Handle, IReadOnlyList<IDisposable> Observers);
 }
@@ -3,12 +3,10 @@ using Microsoft.Extensions.Logging;
 namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation;
 /// <summary>
-///     Periodic re-evaluation scheduler for calc tags carrying a <c>timerIntervalMs</c>. A deliberate
+///     Periodic re-evaluation scheduler for calc tags carrying a <c>timerIntervalMs</c>. One
 ///     mirror of the VirtualTag <c>TimerTriggerScheduler</c> (<c>Core.VirtualTags</c>): one
 ///     <see cref="System.Threading.Timer"/> per distinct interval group (so the wire count stays low
 ///     regardless of tag count) with a per-group in-flight flag that skips a tick when the prior tick for
-///     the same group is still running. Revived <b>inside the driver</b> because the VT scheduler is welded
+///     the same group is still running. Each tick invokes the driver's own evaluate
 ///     to <c>VirtualTagEngine.EvaluateOneAsync</c>; here each tick invokes the driver's own evaluate
 ///     callback per tag in the group.
 /// </summary>
 internal sealed class CalculationTimerScheduler : IDisposable
@@ -22,7 +22,7 @@ public sealed class LastValueCache
 {
    // OPC UA BadWaitingForInitialData (0x80320000) — the repo-wide convention for "no value has
    // been observed yet" (mirrored by CalculationDriver before its first evaluation,
-    // VirtualTagEngine for a freshly-materialised node, and OtOpcUaNodeManager /
+    // and OtOpcUaNodeManager /
    // AddressSpaceApplier for a just-deployed variable). MQTT is subscribe-first, so an unseen
    // RawPath is in exactly that state until its first publish arrives. Deliberately NOT
    // GoodNoData — that code is reserved (see NullHistorianDataSource, OtOpcUaNodeManager
@@ -347,7 +347,7 @@ public sealed class ScriptAnalysisService
                    : $"**Tag path** `{Code(info.Path)}`\n\n{info.Kind} · Type **{info.DataType}**"
                        + (info.DriverInstanceId is null ? "" : $" · Driver `{Code(info.DriverInstanceId)}`");
                // Truthfulness: ctx.SetVirtualTag(...) is a no-op in the live single-tag evaluator
-                // (RoslynVirtualTagEvaluator drops cross-tag writes; the cascade VirtualTagEngine is dormant).
+                // (RoslynVirtualTagEvaluator drops cross-tag writes).
                if (isSetVirtualTag)
                    tagMd += "\n\n⚠ Cross-tag `SetVirtualTag` writes are currently dropped in single-tag mode " +
                             "(the cascade engine is not wired into the host); this write will not take effect at runtime.";
@@ -40,7 +40,7 @@ public static class LocalDbRegistration
    public const string LocalDbSectionPath = "LocalDb";
    /// <summary>Configuration section holding <c>ReplicationOptions</c>.</summary>
-    public const string ReplicationSectionPath = "LocalDb:Replication";
+    public const string ReplicationSectionPath = Runtime.LocalDbReplicationPolicy.ReplicationSectionPath;
    /// <summary>
    ///     Configuration key carrying the dedicated h2c sync listener's port. Zero or absent means
@@ -170,12 +170,9 @@ public static class LocalDbSetup
    /// <returns><c>true</c> when this node is configured to replicate.</returns>
    public static bool ReplicationIsConfigured(IConfiguration configuration)
    {
-        ArgumentNullException.ThrowIfNull(configuration);
+        // Shared with the alarm-historian drain gate (#603): one predicate, so the nodes that capture
-
+        // changes and the nodes that stand down for a peer can never disagree.
-        var section = configuration.GetSection(LocalDbRegistration.ReplicationSectionPath);
+        return Runtime.LocalDbReplicationPolicy.IsConfigured(configuration);
        return !string.IsNullOrWhiteSpace(section["PeerAddress"])
            || !string.IsNullOrWhiteSpace(section["ApiKey"]);
    }
    /// <summary>Opts every replicated table into change capture, baselining what is already there.</summary>
@@ -16,7 +16,7 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Engines;
 ///
 /// Single-tag mode: cross-tag <c>ctx.SetVirtualTag</c> writes are dropped (logged) because
 /// fan-out between actors is owned by <c>DependencyMuxActor</c>, not by the eval engine.
-/// Cycle detection + cascade ordering live in <see cref="VirtualTagEngine"/>; this adapter
+/// Cycle detection lives in the deploy gate (<c>DependencyGraph</c>); this adapter
 /// stays single-tag scoped to keep <see cref="VirtualTagActor"/>'s message loop simple.
 ///
 /// <para>
@@ -579,7 +579,14 @@ public sealed class AddressSpaceApplier
    }
    /// <summary>
-    ///     Auto-provision the historian for the added historized equipment tags. Runs on the OPC UA
+    ///     Resolve the historian name EXACTLY as MaterialiseRawSubtree does: a null/blank override falls
    ///     back to the RawPath (== NodeId).
    /// </summary>
    private static string ResolveHistorianName(RawTagPlan tag) =>
        string.IsNullOrWhiteSpace(tag.HistorianTagname) ? tag.NodeId : tag.HistorianTagname;
    /// <summary>
    ///     Auto-provision the historian for the added (and newly-historized / renamed-historian) raw tags. Runs on the OPC UA
    ///     publish actor's pinned thread, so the synchronous portion is kept to building the request
    ///     list only and the gateway round-trip is dispatched fire-and-forget. The whole hook is wrapped
    ///     in try/catch — a synchronously-throwing provisioner (or any request-building fault) is
@@ -597,22 +604,39 @@ public sealed class AddressSpaceApplier
            foreach (var tag in plan.AddedRawTags)
            {
                if (!tag.IsHistorized || tag.Alarm is not null) continue;
                AddRequest(tag);
            }
-                // Parse the driver-agnostic data type from the tag's DataType string. An unparseable
+            // #600.1: a historization flip (false->true) or a historian-name / data-type change on an
-                // type is skipped (logged at Debug) rather than faulting the hook.
+            // EXISTING raw tag never appears in AddedRawTags, so EnsureTags never fired for it. EnsureTags is
            // an upsert, so re-ensuring is safe; an unchanged pair is a no-op. Only raw deltas are used - a
            // UNS mirror delta must not double-provision the same historian tag.
            foreach (var d in plan.ChangedRawTags)
            {
                var cur = d.Current;
                if (!cur.IsHistorized || cur.Alarm is not null) continue;
                var prev = d.Previous;
                var wasProvisioned = prev.IsHistorized && prev.Alarm is null;
                if (wasProvisioned
                    && ResolveHistorianName(prev) == ResolveHistorianName(cur)
                    && string.Equals(prev.DataType, cur.DataType, StringComparison.OrdinalIgnoreCase))
                    continue;
                AddRequest(cur);
            }
            void AddRequest(RawTagPlan tag)
            {
                // An unparseable data type is skipped (logged at Debug) rather than faulting the hook.
                if (!Enum.TryParse<DriverDataType>(tag.DataType, ignoreCase: true, out var dataType))
                {
                    _logger.LogDebug(
-                        "AddressSpaceApplier: skipping historian provisioning for an added historized tag whose data type '{DataType}' is not a DriverDataType",
+                        "AddressSpaceApplier: skipping historian provisioning for a historized tag whose data type '{DataType}' is not a DriverDataType",
                        tag.DataType);
-                    continue;
+                    return;
                }
                // Resolve the historian name EXACTLY as MaterialiseRawSubtree does: a null/blank override
                // falls back to the RawPath (== NodeId).
                var historianName = string.IsNullOrWhiteSpace(tag.HistorianTagname) ? tag.NodeId : tag.HistorianTagname;
                (requests ??= new List<HistorianTagProvisionRequest>()).Add(
-                    new HistorianTagProvisionRequest(historianName, dataType, EngineeringUnit: null, Description: tag.Name));
+                    new HistorianTagProvisionRequest(ResolveHistorianName(tag), dataType, EngineeringUnit: null, Description: tag.Name));
            }
            if (requests is null) return;
@@ -0,0 +1,43 @@
 using Microsoft.Extensions.Configuration;
 namespace ZB.MOM.WW.OtOpcUa.Runtime;
 /// <summary>
 ///     The ONE definition of "this node's LocalDb is replicated to a peer" (#603). Two consumers must agree
 ///     on it, and used to carry separate predicates that diverged: the Host's table-registration step
 ///     (<c>LocalDbSetup.ReplicationIsConfigured</c> - which nodes capture changes) and the alarm-historian
 ///     drain gate (<c>AddAlarmHistorian</c> - which nodes stand down in favour of a peer). Lives in Runtime
 ///     because Host references Runtime and not the reverse.
 /// </summary>
 /// <remarks>
 ///     <para>
 ///         <b>Predicate: <c>LocalDb:Replication:PeerAddress</c> OR <c>LocalDb:Replication:ApiKey</c> is
 ///         non-blank.</b> The OR is required: only the dialling half of a pair sets <c>PeerAddress</c>, while
 ///         the passive half carries the shared <c>ApiKey</c> (plus a listen port) and nothing else.
 ///     </para>
 ///     <para>
 ///         <b>Why the drain gate takes the same predicate (it previously used PeerAddress OR
 ///         <c>SyncListenPort</c>).</b> The gate's question is "does a peer hold the same rows?". The passive
 ///         sync endpoint is fail-closed on the key, so a node with a listen port but no <c>ApiKey</c> rejects
 ///         every inbound stream and shares its queue with nobody - gating it would park its alarm events
 ///         behind a Primary that cannot deliver them. Conversely an <c>ApiKey</c>-only passive node IS
 ///         registered and does receive its peer's rows, so it must be gated or both halves drain. The port
 ///         is not evidence of a peer; the key is.
 ///     </para>
 /// </remarks>
 public static class LocalDbReplicationPolicy
 {
    /// <summary>Configuration section holding the LocalDb replication options.</summary>
    public const string ReplicationSectionPath = "LocalDb:Replication";
    /// <summary>Whether this node's LocalDb replicates to a pair peer.</summary>
    /// <param name="configuration">The application configuration.</param>
    /// <returns><c>true</c> when <c>PeerAddress</c> or <c>ApiKey</c> is set under the replication section.</returns>
    public static bool IsConfigured(IConfiguration configuration)
    {
        ArgumentNullException.ThrowIfNull(configuration);
        var section = configuration.GetSection(ReplicationSectionPath);
        return !string.IsNullOrWhiteSpace(section["PeerAddress"])
            || !string.IsNullOrWhiteSpace(section["ApiKey"]);
    }
 }
@@ -177,12 +177,11 @@ public static class ServiceCollectionExtensions
        // possibly deliver its events. Scoping the gate to "is my queue actually shared?" is what
        // keeps the two scopes from disagreeing.
        // BOTH halves of a pair share the queue, but only one of them dials: the initiator sets
-        // Replication:PeerAddress, its partner only sets SyncListenPort and waits. Testing the dial
+        // Replication:PeerAddress, its partner carries the shared ApiKey and waits. The predicate is
-        // side alone would leave the listening half permanently ungated — one drainer per pair by
+        // LocalDbReplicationPolicy - the same one LocalDbSetup uses to decide which nodes capture
-        // accident rather than by role, and the wrong one whenever the roles swap.
+        // changes at all. It deliberately does NOT count SyncListenPort: the passive endpoint is
-        var replicated =
+        // fail-closed on the key, so a port with no ApiKey shares nothing and must keep draining.
-            !string.IsNullOrWhiteSpace(configuration["LocalDb:Replication:PeerAddress"])
+        var replicated = LocalDbReplicationPolicy.IsConfigured(configuration);
            || !string.IsNullOrWhiteSpace(configuration["LocalDb:SyncListenPort"]);
        if (!replicated)
        {
@@ -366,7 +366,7 @@ public sealed class VirtualTagActor : ReceiveActor
            _virtualTagId, _hasLastValue ? _lastValue : null, timestampUtc, CorrelationId.NewId(), OpcUaStatus.BadInternalError));
    }
-    /// <summary>Inputs-ready gate (mirrors the dormant <c>VirtualTagEngine.AreInputsReady</c>): true once
+    /// <summary>Inputs-ready gate (the retired cascade engine's rule): true once
    /// every declared dependency ref has been seen at least once. Called once per failure — the ref list
    /// is small, so no caching. A vacuously-true empty ref list means a no-dependency script degrades on
    /// its first failure.</summary>
@@ -57,4 +57,18 @@ public sealed class ScriptSandboxBuildTests
            """return (double)ctx.GetTag("X").Value;""");
        evaluator.ShouldNotBeNull();
    }
    /// <summary>#603: the BCL reference set is built once per process, so two Build calls share the same
    /// <c>MetadataReference</c> instances rather than re-reading ~170 DLLs per compile.</summary>
    [Fact]
    public void Build_reuses_metadata_reference_instances_across_calls()
    {
        var first = ScriptSandbox.Build(typeof(ScriptContext));
        var second = ScriptSandbox.Build(typeof(ScriptContext));
        first.References.Count.ShouldBeGreaterThan(50);
        first.References.Count.ShouldBe(second.References.Count);
        for (var i = 0; i < first.References.Count; i++)
            ReferenceEquals(first.References[i], second.References[i]).ShouldBeTrue($"reference {i} was rebuilt");
    }
 }
@@ -1,91 +0,0 @@
 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);
    /// <summary>Gets the number of active subscriptions.</summary>
    public int ActiveSubscriptionCount { get; private set; }
    /// <summary>Sets a tag value without notifying subscribers.</summary>
    /// <param name="path">The path to the tag.</param>
    /// <param name="value">The value to set.</param>
    /// <param name="statusCode">The OPC UA status code for the value.</param>
    public void Set(string path, object value, uint statusCode = 0u)
    {
        var now = DateTime.UtcNow;
        _values[path] = new DataValueSnapshot(value, statusCode, now, now);
    }
    /// <summary>Sets a tag value and notifies all subscribers.</summary>
    /// <param name="path">The path to the tag.</param>
    /// <param name="value">The value to set.</param>
    /// <param name="statusCode">The OPC UA status code for the value.</param>
    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]);
        }
    }
    /// <summary>Reads the current value of a tag.</summary>
    /// <param name="path">The path to the tag.</param>
    public DataValueSnapshot ReadTag(string path)
        => _values.TryGetValue(path, out var v)
            ? v
            : new DataValueSnapshot(null, 0x80340000u, null, DateTime.UtcNow);
    /// <summary>Subscribes to tag value changes.</summary>
    /// <param name="path">The path to the tag.</param>
    /// <param name="observer">The callback to invoke when the tag value changes.</param>
    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;
        /// <summary>Initializes a new subscription unsubscription handler.</summary>
        /// <param name="up">The parent FakeUpstream instance.</param>
        /// <param name="path">The tag path.</param>
        /// <param name="observer">The observer callback to unsubscribe.</param>
        public Unsub(FakeUpstream up, string path, Action<string, DataValueSnapshot> observer)
        {
            _up = up; _path = path; _observer = observer;
        }
        /// <summary>Unsubscribes from tag value changes.</summary>
        public void Dispose()
        {
            if (_up._subs.TryGetValue(_path, out var list))
            {
                lock (list)
                {
                    if (list.Remove(_observer))
                        _up.ActiveSubscriptionCount--;
                }
            }
        }
    }
 }
@@ -1,203 +0,0 @@
 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
 {
    /// <summary>Verifies that timer interval causes periodic reevaluation of virtual tags.</summary>
    [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");
    }
    /// <summary>Verifies that tags without TimerInterval are not scheduled.</summary>
    [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");
    }
    /// <summary>Verifies that Start groups tags by interval into shared timers.</summary>
    [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)),
        }));
    }
    // ----- Core.VirtualTags-007: timer ticks must not block pool threads and must skip when prior tick is still running -----
    /// <summary>Verifies that tick is skipped when the prior tick for the same group is still running.</summary>
    [Fact]
    public async Task Tick_skips_when_prior_tick_for_the_same_group_is_still_running()
    {
        // Regression for Core.VirtualTags-007: if a single tick takes longer than the
        // interval, subsequent timer callbacks must NOT each pin a thread-pool thread
        // waiting on the same evaluation gate. The scheduler tracks an in-flight flag
        // per group and skips a new tick when the prior one is still running.
        var up = new FakeUpstream();
        up.Set("In", 1);
        var logger = new LoggerConfiguration().CreateLogger();
        // Slow script — each evaluation takes longer than several timer intervals.
        const int slowMs = 250;
        const int intervalMs = 50;
        using var engine = new VirtualTagEngine(up,
            new ScriptLoggerFactory(logger), logger);
        engine.Load([new VirtualTagDefinition(
            "Slow", DriverDataType.Int32,
            $$"""
            var end = DateTime.UtcNow.AddMilliseconds({{slowMs}});
            while (DateTime.UtcNow < end) { }
            return (int)ctx.GetTag("In").Value;
            """,
            ChangeTriggered: false,
            TimerInterval: TimeSpan.FromMilliseconds(intervalMs))]);
        using var sched = new TimerTriggerScheduler(engine, logger);
        sched.Start([new VirtualTagDefinition(
            "Slow", DriverDataType.Int32,
            "",
            ChangeTriggered: false,
            TimerInterval: TimeSpan.FromMilliseconds(intervalMs))]);
        // Wait long enough for many timer ticks at 50ms while one evaluation
        // (~250ms each) holds the engine. Window is 600ms ~ 12 ticks.
        await Task.Delay(600);
        // With the fix in place, ticks that fire while the previous one for the same
        // group is still running are skipped. The skipped count must be measurable; if
        // SkippedTickCount is still 0 after 600ms with ~12 ticks fired and a 250ms eval,
        // the fix is not working — at minimum 3-4 ticks must have been skipped.
        sched.SkippedTickCount.ShouldBeGreaterThan(2,
            "ticks that fire while the prior tick for the same group is still running must be skipped");
    }
    /// <summary>Verifies that a disposed scheduler stops firing timers.</summary>
    [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))]));
    }
    // ----- Core.VirtualTags-015: Start must not be callable twice on the same instance -----
    /// <summary>
    ///     Verifies that a second call to Start on the same (non-disposed) scheduler
    ///     throws InvalidOperationException. Without this guard, calling Start twice
    ///     appends a duplicate set of timers and TickGroups, causing every timer-group
    ///     to fire and evaluate twice per interval with no diagnostic.
    /// </summary>
    [Fact]
    public void Start_called_twice_throws_InvalidOperationException()
    {
        // Regression for Core.VirtualTags-015: without the _started guard, two Start
        // calls silently double the timer count and the evaluation frequency.
        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.FromSeconds(10))]);
        using var sched = new TimerTriggerScheduler(engine, logger);
        sched.Start([new VirtualTagDefinition(
            "T", DriverDataType.Int32, """return 1;""",
            TimerInterval: TimeSpan.FromSeconds(10))]);
        // Second call on the same live instance must throw — not silently create duplicates.
        Should.Throw<InvalidOperationException>(() =>
            sched.Start([new VirtualTagDefinition(
                "T", DriverDataType.Int32, """return 1;""",
                TimerInterval: TimeSpan.FromSeconds(10))]));
    }
 }
@@ -1,800 +0,0 @@
 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);
    }
    /// <summary>Verifies that a simple virtual tag script can read an upstream tag and return a coerced value.</summary>
    [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);
    }
    /// <summary>Verifies that an upstream tag change triggers a cascade re-evaluation through two levels of dependent virtual tags.</summary>
    [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).
        // Both B and C are updated in the same CascadeAsync call (topological order:
        // B then C), but we must wait for each independently: the WaitForConditionAsync
        // on B returns as soon as _valueCache["B"] is set (before the semaphore is
        // released for C's evaluation), so asserting C immediately after the B-wait
        // races against C's still-in-progress evaluation. Wait for C explicitly.
        up.Push("A", 5.0);
        await WaitForConditionAsync(() => Equals(engine.Read("B").Value, 15.0));
        await WaitForConditionAsync(() => Equals(engine.Read("C").Value, 30.0));
        engine.Read("B").Value.ShouldBe(15.0);
        engine.Read("C").Value.ShouldBe(30.0);
    }
    /// <summary>Verifies that a circular dependency among virtual tags is rejected at load time.</summary>
    [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;
    }
    /// <summary>Verifies that script compilation errors surface at load time with all failures aggregated.</summary>
    [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;
    }
    /// <summary>Verifies that a runtime exception in one virtual tag's script is isolated and does not affect other tags.</summary>
    [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();
    }
    /// <summary>Verifies that script timeout is mapped to BadInternalError status without killing the engine.</summary>
    [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 = DateTime.UtcNow.AddSeconds(5);
                while (DateTime.UtcNow < 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);
    }
    /// <summary>Verifies that subscribers receive notifications when the engine emits value changes.</summary>
    [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);
    }
    /// <summary>Verifies that the historize flag routes virtual tag values to the history writer.</summary>
    [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");
    }
    /// <summary>Verifies that upstream pushes are ignored when change-driven is false.</summary>
    [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");
    }
    /// <summary>Verifies that reloading the engine replaces existing tags and resubscribes to upstream sources cleanly.</summary>
    [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;
    }
    /// <summary>Verifies that disposing the engine releases all upstream subscriptions.</summary>
    [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;
    }
    /// <summary>Verifies that SetVirtualTag called within a script updates the target and triggers observers.</summary>
    [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);
    }
    /// <summary>Verifies that SetVirtualTag within a script cascades to change-triggered dependents of the written tag.</summary>
    [Fact]
    public async Task SetVirtualTag_within_script_cascades_to_dependents_of_the_written_tag()
    {
        // Regression for Core.VirtualTags-001: a ctx.SetVirtualTag write must schedule
        // the change-trigger cascade for tags depending on the written path, exactly
        // like an upstream delta. "Dependent" reads "Target" and is ChangeTriggered, so
        // a write to Target via ctx.SetVirtualTag must re-evaluate Dependent.
        var up = new FakeUpstream();
        up.Set("In", 5);
        using var engine = Build(up);
        engine.Load([
            new VirtualTagDefinition("Target", DriverDataType.Int32,
                """return 0;""", ChangeTriggered: false),  // operator-written via SetVirtualTag
            new VirtualTagDefinition("Dependent", DriverDataType.Int32,
                """return (int)ctx.GetTag("Target").Value + 1;""", ChangeTriggered: true),
            new VirtualTagDefinition("Writer", DriverDataType.Int32,
                """
                var v = (int)ctx.GetTag("In").Value;
                ctx.SetVirtualTag("Target", v * 100);
                return v;
                """),
        ]);
        await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
        // Writer set Target = 500; the cascade must have re-evaluated Dependent to 501.
        engine.Read("Target").Value.ShouldBe(500);
        await WaitForConditionAsync(() => Equals(engine.Read("Dependent").Value, 501));
        engine.Read("Dependent").Value.ShouldBe(501,
            "a ctx.SetVirtualTag write must cascade to change-triggered dependents");
    }
    /// <summary>Verifies that type coercion from script double to configured int32 works correctly.</summary>
    [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");
    }
    // ----- Core.VirtualTags-012: previously-missing coverage -----
    /// <summary>Verifies that the AreInputsReady guard publishes BadWaitingForInitialData when upstream tags have bad status.</summary>
    [Fact]
    public async Task AreInputsReady_guard_publishes_BadWaitingForInitialData_when_upstream_is_bad()
    {
        // Arrange: upstream tag is Bad-quality (not yet available).
        var up = new FakeUpstream();
        up.Set("BadIn", null!, 0x80000000u); // bad status, null value
        using var engine = Build(up);
        engine.Load([new VirtualTagDefinition(
            "Derived", DriverDataType.Int32,
            """return (int)ctx.GetTag("BadIn").Value * 2;""")]);
        // Act: evaluate — inputs are not ready.
        await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
        // Assert: tag publishes BadWaitingForInitialData, not a stale null/Good.
        var result = engine.Read("Derived");
        result.StatusCode.ShouldBe(0x80320000u, "BadWaitingForInitialData expected when inputs are bad");
        result.Value.ShouldBeNull();
    }
    /// <summary>Verifies that the AreInputsReady guard recovers when an upstream tag transitions from bad to good status.</summary>
    [Fact]
    public async Task AreInputsReady_guard_publishes_BadWaitingForInitialData_then_recovers_when_upstream_becomes_good()
    {
        // Arrange: upstream tag starts absent (null/Bad).
        var up = new FakeUpstream();
        using var engine = Build(up);
        engine.Load([new VirtualTagDefinition(
            "Derived", DriverDataType.Int32,
            """return (int)ctx.GetTag("In").Value + 1;""")]);
        // First evaluation: upstream not ready → BadWaitingForInitialData.
        await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
        engine.Read("Derived").StatusCode.ShouldBe(0x80320000u);
        // Upstream becomes available.
        up.Push("In", 10);
        await WaitForConditionAsync(() => engine.Read("Derived").StatusCode == 0u);
        // Tag should now have a Good value.
        engine.Read("Derived").StatusCode.ShouldBe(0u);
        engine.Read("Derived").Value.ShouldBe(11);
    }
    /// <summary>Verifies that SetVirtualTag cascades to change-triggered dependents.</summary>
    [Fact]
    public async Task SetVirtualTag_cascades_to_change_triggered_dependent()
    {
        // Arrange: "Writer" writes to "Target"; "Consumer" reads "Target" and is change-triggered.
        var up = new FakeUpstream();
        up.Set("In", 3);
        using var engine = Build(up);
        engine.Load([
            new VirtualTagDefinition("Target", DriverDataType.Int32,
                """return 0;""", ChangeTriggered: false),
            new VirtualTagDefinition("Writer", DriverDataType.Int32,
                """
                var v = (int)ctx.GetTag("In").Value;
                ctx.SetVirtualTag("Target", v * 10);
                return v;
                """),
            new VirtualTagDefinition("Consumer", DriverDataType.Int32,
                """return (int)ctx.GetTag("Target").Value + 1;""",
                ChangeTriggered: true),
        ]);
        await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
        // Writer sets Target = 30; Consumer should cascade and compute 31.
        await WaitForConditionAsync(() => engine.Read("Consumer").Value is int v && v == 31);
        engine.Read("Target").Value.ShouldBe(30);
        engine.Read("Consumer").Value.ShouldBe(31);
    }
    /// <summary>Verifies that SetVirtualTag calls targeting unregistered paths are caught at load time.</summary>
    [Fact]
    public async Task SetVirtualTag_on_non_registered_path_is_caught_at_Load()
    {
        // Originally validated the runtime warning-and-drop branch in OnScriptSetVirtualTag.
        // After Core.VirtualTags-011 the static DependencyExtractor.Writes set is validated
        // at Load time, so a literal-string write to a non-existent path is now rejected
        // at publish — the dynamic warning path is reserved as a defensive guard for cases
        // the static extractor cannot see (currently none, since dynamic paths are also
        // rejected at extraction).
        var up = new FakeUpstream();
        up.Set("In", 1);
        using var engine = Build(up);
        Should.Throw<InvalidOperationException>(() => engine.Load([
            new VirtualTagDefinition("Writer", DriverDataType.Int32,
                """
                ctx.SetVirtualTag("NonExistentPath", 99);
                return (int)ctx.GetTag("In").Value;
                """)
        ])).Message.ShouldContain("NonExistentPath");
        await Task.CompletedTask;
    }
    /// <summary>Verifies that EvaluateOneAsync throws ArgumentException when called for an unregistered path.</summary>
    [Fact]
    public async Task EvaluateOneAsync_throws_ArgumentException_for_unregistered_path()
    {
        var up = new FakeUpstream();
        using var engine = Build(up);
        engine.Load([new VirtualTagDefinition("A", DriverDataType.Int32, """return 1;""")]);
        await Should.ThrowAsync<ArgumentException>(async () =>
            await engine.EvaluateOneAsync("NoSuchTag", TestContext.Current.CancellationToken));
    }
    /// <summary>Verifies that a type coercion failure maps to BadInternalError status.</summary>
    [Fact]
    public async Task CoerceResult_failure_maps_to_BadInternalError()
    {
        // Arrange: script returns an object that cannot be coerced to the declared type.
        var up = new FakeUpstream();
        using var engine = Build(up);
        engine.Load([new VirtualTagDefinition(
            "Bad", DriverDataType.Int32,
            // Return a non-numeric string — Convert.ToInt32("not-a-number") throws.
            """return "not-a-number";""")]);
        await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
        // CoerceResult returns null on failure; the null propagates as BadInternalError.
        engine.Read("Bad").StatusCode.ShouldBe(0x80020000u, "type-coercion failure must map to BadInternalError");
        engine.Read("Bad").Value.ShouldBeNull();
    }
    // ----- Core.VirtualTags-011: Writes target validation at Load time -----
    /// <summary>Verifies that Load rejects scripts that write to unregistered virtual tag paths.</summary>
    [Fact]
    public async Task Load_rejects_script_writing_to_unregistered_virtual_tag_path()
    {
        // Regression for Core.VirtualTags-011: a script that calls
        // ctx.SetVirtualTag("Typo", ...) must be caught at publish/load time rather than
        // silently dropped at runtime, so operator typos surface as a publish failure.
        var up = new FakeUpstream();
        using var engine = Build(up);
        var ex = Should.Throw<InvalidOperationException>(() => engine.Load([
            new VirtualTagDefinition("Writer", DriverDataType.Int32,
                """
                ctx.SetVirtualTag("NonRegisteredTarget", 1);
                return 0;
                """),
            new VirtualTagDefinition("RegisteredTarget", DriverDataType.Int32,
                """return 1;"""),
        ]));
        ex.Message.ShouldContain("Writer");
        ex.Message.ShouldContain("NonRegisteredTarget");
        await Task.CompletedTask;
    }
    /// <summary>Verifies that Load accepts scripts that write to registered virtual tag paths.</summary>
    [Fact]
    public async Task Load_accepts_script_writing_to_registered_virtual_tag_path()
    {
        // Companion to the rejection test: a write to a registered tag must continue to
        // load successfully.
        var up = new FakeUpstream();
        up.Set("In", 1);
        using var engine = Build(up);
        // No throw — Writer writes to Target which is registered.
        engine.Load([
            new VirtualTagDefinition("Target", DriverDataType.Int32,
                """return 0;""", ChangeTriggered: false),
            new VirtualTagDefinition("Writer", DriverDataType.Int32,
                """
                ctx.SetVirtualTag("Target", (int)ctx.GetTag("In").Value);
                return 0;
                """),
        ]);
        await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
        engine.Read("Target").Value.ShouldBe(1);
    }
    // ----- Core.VirtualTags-006: empty observer list left in _observers map -----
    /// <summary>Verifies that disposing the last subscriber for a path removes the empty observer list entry.</summary>
    [Fact]
    public void Subscribe_then_Unsub_prunes_empty_observer_list_for_path()
    {
        // Regression for Core.VirtualTags-006: disposing the last subscriber for a path
        // must remove the dictionary entry so a long-running server with churning OPC UA
        // subscriptions does not accumulate an unbounded number of empty List entries.
        var up = new FakeUpstream();
        using var engine = Build(up);
        engine.Load([new VirtualTagDefinition(
            "T", DriverDataType.Int32, """return 1;""")]);
        // Subscribe, then immediately Dispose — both the only observer.
        var sub1 = engine.Subscribe("T", (_, _) => { });
        var sub2 = engine.Subscribe("T", (_, _) => { });
        sub1.Dispose();
        sub2.Dispose();
        // The internal map should no longer hold an entry for the path.
        // Use the same ConcurrentDictionary type the engine uses; we check via reflection
        // on the test-private field so this is robust to future renames inside engine.
        var observersField = typeof(VirtualTagEngine).GetField(
            "_observers",
            System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance);
        observersField.ShouldNotBeNull();
        var observers = observersField!.GetValue(engine);
        observers.ShouldNotBeNull();
        var containsKey = observers!.GetType().GetMethod("ContainsKey")!;
        var result = (bool)containsKey.Invoke(observers, new object[] { "T" })!;
        result.ShouldBeFalse("disposing the last subscriber must remove the dictionary entry");
    }
    // ----- Core.VirtualTags-004: CoerceResult default arm leaks uncoerced values -----
    /// <summary>Verifies that CoerceResult correctly handles Int16, UInt16, UInt32, and UInt64 types.</summary>
    [Fact]
    public async Task CoerceResult_handles_Int16_UInt16_UInt32_UInt64()
    {
        // Regression for Core.VirtualTags-004: before the fix, CoerceResult had a default
        // arm that returned the script's raw double/string for these types, producing a
        // type-mismatched DataValueSnapshot. Verify every integer DriverDataType the engine
        // is allowed to declare coerces correctly.
        var up = new FakeUpstream();
        using var engine = Build(up);
        engine.Load([
            new VirtualTagDefinition("AsInt16",  DriverDataType.Int16,  """return 7.0;"""),
            new VirtualTagDefinition("AsUInt16", DriverDataType.UInt16, """return 8.0;"""),
            new VirtualTagDefinition("AsUInt32", DriverDataType.UInt32, """return 9.0;"""),
            new VirtualTagDefinition("AsUInt64", DriverDataType.UInt64, """return 10.0;"""),
        ]);
        await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
        engine.Read("AsInt16").Value.ShouldBeOfType<short>();
        engine.Read("AsInt16").Value.ShouldBe((short)7);
        engine.Read("AsUInt16").Value.ShouldBeOfType<ushort>();
        engine.Read("AsUInt16").Value.ShouldBe((ushort)8);
        engine.Read("AsUInt32").Value.ShouldBeOfType<uint>();
        engine.Read("AsUInt32").Value.ShouldBe((uint)9);
        engine.Read("AsUInt64").Value.ShouldBeOfType<ulong>();
        engine.Read("AsUInt64").Value.ShouldBe((ulong)10);
    }
    /// <summary>Verifies that Load rejects virtual tag definitions with unsupported DriverDataType values.</summary>
    [Fact]
    public async Task Load_rejects_definition_with_unsupported_DriverDataType()
    {
        // Regression for Core.VirtualTags-004: any DriverDataType that CoerceResult cannot
        // honour must be rejected at Load time so an operator typo (or a future enum
        // member added without coercion support) does not silently emit a type-mismatched
        // value to OPC UA clients. Reference is unsupported for virtual tags (the engine
        // does not synthesize Galaxy attribute references).
        var up = new FakeUpstream();
        using var engine = Build(up);
        var ex = Should.Throw<InvalidOperationException>(() => engine.Load([
            new VirtualTagDefinition("Ref", DriverDataType.Reference, """return "Some.Attribute";"""),
        ]));
        ex.Message.ShouldContain("Reference");
        ex.Message.ShouldContain("Ref");
        await Task.CompletedTask;
    }
    /// <summary>Verifies that Load rejects duplicate virtual tag paths with an aggregated error message.</summary>
    [Fact]
    public async Task Load_rejects_duplicate_path_with_aggregated_error()
    {
        var up = new FakeUpstream();
        using var engine = Build(up);
        var ex = Should.Throw<InvalidOperationException>(() => engine.Load([
            new VirtualTagDefinition("Dup", DriverDataType.Int32, """return 1;"""),
            new VirtualTagDefinition("Dup", DriverDataType.Int32, """return 2;"""),
            new VirtualTagDefinition("Good", DriverDataType.Int32, """return 3;"""),
        ]));
        ex.Message.ShouldContain("Dup");
        ex.Message.ShouldContain("duplicate");
        await Task.CompletedTask;
    }
    /// <summary>Verifies that Read returns BadNodeIdUnknown for all paths before Load is called.</summary>
    [Fact]
    public void Read_before_Load_returns_BadNodeIdUnknown()
    {
        var up = new FakeUpstream();
        using var engine = Build(up);
        // Read is allowed before Load — it just returns BadNodeIdUnknown for everything.
        var result = engine.Read("AnyPath");
        result.StatusCode.ShouldBe(0x80340000u, "BadNodeIdUnknown before Load");
    }
    /// <summary>Verifies that EvaluateOneAsync throws InvalidOperationException when called before Load.</summary>
    [Fact]
    public void EvaluateOneAsync_before_Load_throws_InvalidOperationException()
    {
        var up = new FakeUpstream();
        using var engine = Build(up);
        Should.Throw<InvalidOperationException>(() =>
            engine.EvaluateOneAsync("A").GetAwaiter().GetResult());
    }
    /// <summary>Verifies that Subscribe does not throw when called before Load.</summary>
    [Fact]
    public void Subscribe_before_Load_does_not_throw()
    {
        // Subscribe uses GetOrAdd and does not call EnsureLoaded — it should work
        // (returns an Unsub handle) without a Load. The observer won't fire because
        // no tag is registered, but it must not throw.
        var up = new FakeUpstream();
        using var engine = Build(up);
        var fired = false;
        var sub = engine.Subscribe("AnyPath", (_, _) => fired = true);
        sub.ShouldNotBeNull();
        sub.Dispose();
        fired.ShouldBeFalse("no evaluation has happened, observer must not fire");
    }
    // ----- Core.VirtualTags-014: AreInputsReady must not block on Good-quality null value -----
    /// <summary>
    ///     Verifies that a Good-quality upstream snapshot with a null Value (e.g. a
    ///     script returning null) does not permanently block downstream change-triggered
    ///     dependents at BadWaitingForInitialData. A Good-quality null means "the
    ///     upstream is online and returned null" — readiness is determined by StatusCode,
    ///     not by the value's nullability.
    /// </summary>
    [Fact]
    public async Task AreInputsReady_Good_quality_null_upstream_does_not_block_downstream()
    {
        // Regression for Core.VirtualTags-014: before the fix, AreInputsReady checked
        // kv.Value.Value is null, so a Good-quality DataValueSnapshot(null, 0u, ...) —
        // produced by an upstream script returning null — kept every downstream stuck at
        // BadWaitingForInitialData forever. The fix gates only on StatusCode (bit 31).
        var up = new FakeUpstream();
        up.Set("In", 1);
        using var engine = Build(up);
        engine.Load([
            // Producer script returns null — stores DataValueSnapshot(null, Good, ...).
            new VirtualTagDefinition("NullProducer", DriverDataType.String,
                """return null;"""),
            // Consumer depends on NullProducer and is change-triggered.
            // With the bug: Consumer is stuck at BadWaitingForInitialData.
            // After the fix: Consumer evaluates and surfaces the NRE as BadInternalError
            // (since it casts null unconditionally), which is the correct sentinel for
            // "the script ran but produced an error" rather than "inputs not ready".
            new VirtualTagDefinition("Consumer", DriverDataType.String,
                """return (string)ctx.GetTag("NullProducer").Value + "_suffix";""",
                ChangeTriggered: true),
        ]);
        await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
        // NullProducer should be Good with null value (script returned null).
        var producerResult = engine.Read("NullProducer");
        producerResult.StatusCode.ShouldBe(0u, "Good quality: script returned null cleanly");
        producerResult.Value.ShouldBeNull();
        // Consumer must NOT be stuck at BadWaitingForInitialData — it should have
        // evaluated (and produced BadInternalError from the cast of null + "_suffix").
        var consumerResult = engine.Read("Consumer");
        consumerResult.StatusCode.ShouldNotBe(0x80320000u,
            "Consumer must not be stuck at BadWaitingForInitialData when its upstream is Good-quality");
    }
    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");
    }
    /// <summary>Test implementation of IHistoryWriter that records values to a buffer.</summary>
    private sealed class TestHistory : IHistoryWriter
    {
        private readonly List<(string, DataValueSnapshot)> _buf;
        /// <summary>Initializes a new instance of the TestHistory class with the specified buffer.</summary>
        /// <param name="buf">The list to store recorded history entries.</param>
        public TestHistory(List<(string, DataValueSnapshot)> buf) => _buf = buf;
        /// <summary>Records a virtual tag path and value snapshot to the history buffer.</summary>
        /// <param name="path">The virtual tag path.</param>
        /// <param name="value">The data value snapshot to record.</param>
        public void Record(string path, DataValueSnapshot value)
        {
            lock (_buf) { _buf.Add((path, value)); }
        }
    }
    // --- Core.Scripting-016: engine routes compiles through CompiledScriptCache ---
    /// <summary>Verifies that disposing the engine unloads the compiled script assembly.</summary>
    [Fact]
    public void Dispose_unloads_compiled_script_assembly()
    {
        // Pre-fix the engine called ScriptEvaluator.Compile directly, so the emitted
        // script assembly's ALC stayed loaded for the process lifetime. After the fix
        // the engine routes through CompiledScriptCache; engine Dispose triggers cache
        // Dispose which unloads every cached evaluator's ALC. Assert via WeakReference
        // + GC that the assembly is actually reclaimed.
        var weak = CompileTagAndCaptureWeak();
        for (int i = 0; i < 10 && weak.IsAlive; i++)
        {
            GC.Collect();
            GC.WaitForPendingFinalizers();
            GC.Collect();
        }
        weak.IsAlive.ShouldBeFalse(
            "engine Dispose must release the compiled-tag assembly via " +
            "CompiledScriptCache (Core.Scripting-016). If this fails the engine is " +
            "back to calling ScriptEvaluator.Compile directly and -008's headline " +
            "fix doesn't run in production.");
    }
    [System.Runtime.CompilerServices.MethodImpl(
        System.Runtime.CompilerServices.MethodImplOptions.NoInlining)]
    private static WeakReference CompileTagAndCaptureWeak()
    {
        var up = new FakeUpstream();
        up.Set("X", 10);
        var engine = Build(up);
        engine.Load([new VirtualTagDefinition(
            Path: "Doubled",
            DataType: DriverDataType.Float64,
            ScriptSource: """return (double)ctx.GetTag("X").Value * 2.0;""")]);
        // Reach into the engine's compile cache via reflection — internals scoped to
        // Tests-via-InternalsVisibleTo and the field is private.
        var cacheField = typeof(VirtualTagEngine).GetField(
            "_compileCache", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!;
        var cache = cacheField.GetValue(engine)!;
        var cacheDictField = cache.GetType().GetField(
            "_cache", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!;
        var cacheDict = (System.Collections.IDictionary)cacheDictField.GetValue(cache)!;
        var lazy = cacheDict.Values.Cast<object>().Single();
        var lazyValueProp = lazy.GetType().GetProperty("Value")!;
        var evaluator = lazyValueProp.GetValue(lazy)!;
        var funcField = evaluator.GetType().GetField(
            "_func", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!;
        var del = (Delegate)funcField.GetValue(evaluator)!;
        var weak = new WeakReference(del.Method.Module.Assembly);
        engine.Dispose();
        return weak;
    }
 }
@@ -1,138 +0,0 @@
 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);
    }
    /// <summary>Verifies that ReadAsync returns cached engine values.</summary>
    [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();
    }
    /// <summary>Verifies that unknown paths return Bad status quality.</summary>
    [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();
    }
    /// <summary>Verifies that subscribe fires the initial data callback immediately.</summary>
    [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();
    }
    /// <summary>Verifies that subscription fires on upstream changes via engine cascade.</summary>
    [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();
    }
    /// <summary>Verifies that unsubscribe stops further event emissions.</summary>
    [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();
    }
    /// <summary>Verifies that null arguments are rejected.</summary>
    [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();
    }
 }
@@ -7,11 +7,6 @@
        <IsPackable>false</IsPackable>
        <IsTestProject>true</IsTestProject>
        <RootNamespace>ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests</RootNamespace>
        <!-- OTOPCUA0001 (arch-review 07/C-1): these tests invoke VirtualTagSource's IReadable/ISubscribable
             methods DIRECTLY to exercise its wire-level behavior, which is the analyzer's own documented
             intentional case ("a unit test invoking the driver directly ... move the suppression into a
             NoWarn for the test project"). Genuinely intentional — not a resilience-dispatch gap. -->
        <NoWarn>$(NoWarn);OTOPCUA0001</NoWarn>
    </PropertyGroup>
    <ItemGroup>
@@ -7,7 +7,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
 public sealed class LastValueCacheTests
 {
    // OPC UA BadWaitingForInitialData (0x80320000) — the repo-wide convention for "no value
-    // observed yet" (mirrored by CalculationDriver, VirtualTagEngine, FOCAS, and
+    // observed yet" (mirrored by CalculationDriver, FOCAS, and
    // OtOpcUaNodeManager for freshly-materialised / not-yet-evaluated nodes). MQTT is
    // subscribe-first, so an unseen RawPath is in exactly that state until its first publish.
    private const uint BadWaitingForInitialData = 0x80320000u;
@@ -146,6 +146,46 @@ public sealed class AddressSpaceApplierProvisioningTests
        prov.Seen[0].TagName.ShouldBe("Pump1.Good");
    }
    /// <summary>#600.1: a CHANGED raw tag that flips to historized, or whose historian name / data type
    /// changes, is provisioned; an unchanged-historization edit and an alarm-bearing tag are not. The
    /// added "Control" tag is the positive control so the absences are not a vacuous no-dispatch.</summary>
    [Fact]
    public async Task Apply_provisions_changed_tags_whose_historization_flips_or_name_changes()
    {
        var prov = new CapturingProvisioner();
        var applier = new AddressSpaceApplier(NullOpcUaAddressSpaceSink.Instance, NullLogger<AddressSpaceApplier>.Instance, prov);
        var flipOff = NonHistorizedTag("Flip", "Float32") with { TagId = "t-flip", NodeId = "Plant/Flip" };
        var flipOn = flipOff with { IsHistorized = true };
        var renOld = HistorizedTag("Ren", "Old.Name", "Float32") with { TagId = "t-ren" };
        var renNew = renOld with { HistorianTagname = "New.Name" };
        var sameOld = HistorizedTag("Same", "Same.Name", "Float32") with { TagId = "t-same", Writable = false };
        var sameNew = sameOld with { Writable = true }; // non-historian edit
        var alarmOld = NonHistorizedTag("Alm", "Float32") with { TagId = "t-alm" };
        var alarmNew = alarmOld with
        {
            IsHistorized = true,
            Alarm = new EquipmentTagAlarmInfo("AlarmCondition", 500),
        };
        var plan = PlanWithAddedTags(HistorizedTag("Control", "Control.Name", "Float32")) with
        {
            ChangedRawTags = new[]
            {
                new AddressSpacePlan.RawTagDelta(flipOff, flipOn),
                new AddressSpacePlan.RawTagDelta(renOld, renNew),
                new AddressSpacePlan.RawTagDelta(sameOld, sameNew),
                new AddressSpacePlan.RawTagDelta(alarmOld, alarmNew),
            },
        };
        applier.Apply(plan);
        await prov.Called.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
        prov.Seen.Select(r => r.TagName).ShouldBe(
            new[] { "Control.Name", "Plant/Flip", "New.Name" }, ignoreOrder: true);
    }
    /// <summary>Capturing <see cref="ILogger{T}"/> that records every log line and signals a
    /// <see cref="TaskCompletionSource{TResult}"/> the first time a "historian provisioning completed" tally is
    /// emitted, so the fire-and-forget continuation's log can be awaited deterministically.</summary>
@@ -122,14 +122,18 @@ public sealed class AlarmHistorianRegistrationTests
    /// </remarks>
    [Theory]
    // Neither key: an unpaired node. Its rows exist nowhere else.
-    [InlineData(null, null, true)]
+    [InlineData(null, null, null, true)]
    // The dialing half of a pair — gated, because its partner holds the same rows.
-    [InlineData("http://peer:9001", null, false)]
+    [InlineData("http://peer:9001", null, null, false)]
    // The LISTENING half. Same shared queue, but it never dials, so a PeerAddress-only test would
-    // wrongly class it as unpaired and leave it permanently ungated.
+    // wrongly class it as unpaired and leave it permanently ungated. It is identified by the shared
-    [InlineData(null, "9001", false)]
+    // ApiKey (the same predicate LocalDbSetup registers tables on), not by its listen port.
    [InlineData(null, "9001", "k", false)]
    // A listen port with NO key: the fail-closed passive endpoint rejects every inbound stream, so the
    // queue is shared with nobody and the node must keep draining (#603: this used to be gated).
    [InlineData(null, "9001", null, true)]
    public async Task Only_an_unreplicated_node_ignores_the_role_view(
-        string? peerAddress, string? syncListenPort, bool expectDrain)
+        string? peerAddress, string? syncListenPort, string? apiKey, bool expectDrain)
    {
        var tempDir = Path.Combine(Path.GetTempPath(), "otopcua-alarmhist-test-" + Guid.NewGuid().ToString("N"));
        Directory.CreateDirectory(tempDir);
@@ -143,6 +147,7 @@ public sealed class AlarmHistorianRegistrationTests
                ["LocalDb:Path"] = dbPath,
                ["LocalDb:Replication:PeerAddress"] = peerAddress,
                ["LocalDb:SyncListenPort"] = syncListenPort,
                ["LocalDb:Replication:ApiKey"] = apiKey,
            });
            var services = BaseServices();
@@ -0,0 +1,32 @@
 using Microsoft.Extensions.Configuration;
 using Shouldly;
 using Xunit;
 namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian;
 /// <summary>#603: the single "is this node's LocalDb replicated" predicate shared by table registration and the alarm drain gate.</summary>
 public sealed class LocalDbReplicationPolicyTests
 {
    private static IConfiguration Config(params (string Key, string? Value)[] pairs) =>
        new ConfigurationBuilder()
            .AddInMemoryCollection(pairs.Select(p => new KeyValuePair<string, string?>(p.Key, p.Value)))
            .Build();
    /// <summary>PeerAddress OR ApiKey counts; the listen port alone and blank values do not.</summary>
    [Theory]
    [InlineData("LocalDb:Replication:PeerAddress", "http://peer:9001", true)]
    [InlineData("LocalDb:Replication:ApiKey", "k", true)]
    [InlineData("LocalDb:SyncListenPort", "9001", false)]
    [InlineData("LocalDb:Replication:ApiKey", "  ", false)]
    public void IsConfigured_follows_the_PeerAddress_or_ApiKey_rule(string key, string value, bool expected)
    {
        LocalDbReplicationPolicy.IsConfigured(Config((key, value))).ShouldBe(expected);
    }
    /// <summary>No replication keys at all is an unpaired node.</summary>
    [Fact]
    public void IsConfigured_is_false_for_an_empty_configuration()
    {
        LocalDbReplicationPolicy.IsConfigured(Config()).ShouldBeFalse();
    }
 }