From f64a8049d8ed6d0ee2a126c0ba93e8016d242dd4 Mon Sep 17 00:00:00 2001
From: Joseph Doherty <dohejw01@gmail.com>
Date: Mon, 20 Apr 2026 21:23:31 -0400
Subject: [PATCH] =?UTF-8?q?Phase=207=20follow-up=20#243=20=E2=80=94=20Cach?=
 =?UTF-8?q?edTagUpstreamSource=20+=20Phase7EngineComposer?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Ships the composition kernel that maps Config DB rows (Script / VirtualTag /
ScriptedAlarm) to the runtime definitions VirtualTagEngine + ScriptedAlarmEngine
consume, builds the engine instances, and wires OnEvent → historian-sink routing.

## src/ZB.MOM.WW.OtOpcUa.Server/Phase7/

- CachedTagUpstreamSource — implements both Core.VirtualTags.ITagUpstreamSource and
  Core.ScriptedAlarms.ITagUpstreamSource (identical shape, distinct namespaces) on one
  concrete type so the composer can hand one instance to both engines. Thread-safe
  ConcurrentDictionary value cache with synchronous ReadTag + fire-on-write
  Push(path, snapshot) that fans out to every observer registered via SubscribeTag.
  Unknown-path reads return a BadNodeIdUnknown-quality snapshot (status 0x80340000)
  so scripts branch on quality naturally.
- Phase7EngineComposer.Compose(scripts, virtualTags, scriptedAlarms, upstream,
  alarmStateStore, historianSink, rootScriptLogger, loggerFactory) — single static
  entry point that:
  * Indexes scripts by ScriptId, resolves VirtualTag.ScriptId + ScriptedAlarm.PredicateScriptId
    to full SourceCode
  * Projects DB rows to VirtualTagDefinition + ScriptedAlarmDefinition (mapping
    DataType string → DriverDataType enum, AlarmType string → AlarmKind enum,
    Severity 1..1000 → AlarmSeverity bucket matching the OPC UA Part 9 bands
    that AbCipAlarmProjection + OpcUaClient MapSeverity already use)
  * Constructs VirtualTagEngine + loads definitions (throws InvalidOperationException
    with the list of scripts that failed to compile — aggregated like Streams B+C)
  * Constructs ScriptedAlarmEngine + loads definitions + wires OnEvent →
    IAlarmHistorianSink.EnqueueAsync using ScriptedAlarmEvent.Emission as the event
    kind + Condition.LastAckUser/LastAckComment for audit fields
  * Returns Phase7ComposedSources with Disposables list the caller owns

Empty Phase 7 config returns Phase7ComposedSources.Empty so deployments without
scripts / alarms behave exactly as pre-Phase-7. Non-null sources flow into
OpcUaApplicationHost's virtualReadable / scriptedAlarmReadable plumbing landed by
task #239 — DriverNodeManager then dispatches reads by NodeSourceKind per PR #186.

## Tests — 12/12

CachedTagUpstreamSourceTests (6):
- Unknown-path read returns BadNodeIdUnknown-quality snapshot
- Push-then-Read returns cached value
- Push fans out to subscribers in registration order
- Push to one path doesn't fire another path's observer
- Dispose of subscription handle stops fan-out
- Satisfies both Core.VirtualTags + Core.ScriptedAlarms ITagUpstreamSource interfaces

Phase7EngineComposerTests (6):
- Empty rows → Phase7ComposedSources.Empty (both sources null)
- VirtualTag rows → VirtualReadable non-null + Disposables populated
- Missing script reference throws InvalidOperationException with the missing ScriptId
  in the message
- Disabled VirtualTag row skipped by projection
- TimerIntervalMs → TimeSpan.FromMilliseconds round-trip
- Severity 1..1000 maps to Low/Medium/High/Critical at 250/500/750 boundaries
  (matches AbCipAlarmProjection + OpcUaClient.MapSeverity banding)

## Scope — what this PR does NOT do

The composition kernel is the tricky part; the remaining wiring is three narrower
follow-ups that each build on this PR:

- task #244 — driver-bridge feed that populates CachedTagUpstreamSource from live
  driver subscriptions. Without this, ctx.GetTag returns BadNodeIdUnknown even when
  the driver has a fresh value.
- task #245 — ScriptedAlarmReadable adapter exposing each alarm's current Active
  state as IReadable. Phase7EngineComposer.Compose currently returns
  ScriptedAlarmReadable=null so reads on Source=ScriptedAlarm variables return
  BadNotFound per the ADR-002 "misconfiguration not silent fallback" signal.
- task #246 — Program.cs call to Phase7EngineComposer.Compose with config rows
  loaded from the sealed-cache DB read, plus SqliteStoreAndForwardSink lifecycle
  wiring at %ProgramData%/OtOpcUa/alarm-historian-queue.db with the Galaxy.Host
  IPC writer from Stream D.

Task #240 (live OPC UA E2E smoke) depends on all three follow-ups landing.
---
 .../Phase7/CachedTagUpstreamSource.cs         |  84 +++++++
 .../Phase7/Phase7EngineComposer.cs            | 208 ++++++++++++++++++
 .../ZB.MOM.WW.OtOpcUa.Server.csproj           |   4 +
 .../Phase7/CachedTagUpstreamSourceTests.cs    |  83 +++++++
 .../Phase7/Phase7EngineComposerTests.cs       | 143 ++++++++++++
 5 files changed, 522 insertions(+)
 create mode 100644 src/ZB.MOM.WW.OtOpcUa.Server/Phase7/CachedTagUpstreamSource.cs
 create mode 100644 src/ZB.MOM.WW.OtOpcUa.Server/Phase7/Phase7EngineComposer.cs
 create mode 100644 tests/ZB.MOM.WW.OtOpcUa.Server.Tests/Phase7/CachedTagUpstreamSourceTests.cs
 create mode 100644 tests/ZB.MOM.WW.OtOpcUa.Server.Tests/Phase7/Phase7EngineComposerTests.cs

diff --git a/src/ZB.MOM.WW.OtOpcUa.Server/Phase7/CachedTagUpstreamSource.cs b/src/ZB.MOM.WW.OtOpcUa.Server/Phase7/CachedTagUpstreamSource.cs
new file mode 100644
index 0000000..f0b21c2
--- /dev/null
+++ b/src/ZB.MOM.WW.OtOpcUa.Server/Phase7/CachedTagUpstreamSource.cs
@@ -0,0 +1,84 @@
+using System.Collections.Concurrent;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Server.Phase7;
+
+/// <summary>
+///     Production <c>ITagUpstreamSource</c> for the Phase 7 engines (implements both the
+///     Core.VirtualTags and Core.ScriptedAlarms variants — identical shape, distinct
+///     namespaces). Per the interface docstring, reads are synchronous — user scripts
+///     call <c>ctx.GetTag</c> inline — so we serve from a last-known-value cache that
+///     the driver-bridge populates asynchronously via <see cref="Push"/>.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <see cref="Push"/> is called by the driver-bridge (wiring added by task #244)
+///         every time a driver's <c>ISubscribable.OnDataChange</c> fires. Subscribers
+///         registered via <see cref="SubscribeTag"/> are notified synchronously on the
+///         calling thread — the VirtualTagEngine + ScriptedAlarmEngine handle their own
+///         async hand-off via <c>SemaphoreSlim</c>.
+///     </para>
+///     <para>
+///         Reads of a path that has never been <see cref="Push"/>-ed return
+///         <see cref="UpstreamNotConfigured"/>-quality — which scripts see as
+///         <c>ctx.GetTag("...").StatusCode == BadNodeIdUnknown</c> and can branch on.
+///     </para>
+/// </remarks>
+public sealed class CachedTagUpstreamSource
+    : Core.VirtualTags.ITagUpstreamSource,
+      Core.ScriptedAlarms.ITagUpstreamSource
+{
+    private readonly ConcurrentDictionary<string, DataValueSnapshot> _values = new(StringComparer.Ordinal);
+    private readonly ConcurrentDictionary<string, List<Action<string, DataValueSnapshot>>> _observers
+        = new(StringComparer.Ordinal);
+
+    public DataValueSnapshot ReadTag(string path)
+    {
+        if (string.IsNullOrEmpty(path)) throw new ArgumentException("path required", nameof(path));
+        return _values.TryGetValue(path, out var snap)
+            ? snap
+            : new DataValueSnapshot(null, UpstreamNotConfigured, null, DateTime.UtcNow);
+    }
+
+    public IDisposable SubscribeTag(string path, Action<string, DataValueSnapshot> observer)
+    {
+        if (string.IsNullOrEmpty(path)) throw new ArgumentException("path required", nameof(path));
+        ArgumentNullException.ThrowIfNull(observer);
+
+        var list = _observers.GetOrAdd(path, _ => []);
+        lock (list) list.Add(observer);
+        return new Unsub(this, path, observer);
+    }
+
+    /// <summary>
+    ///     Driver-bridge write path — called when a driver delivers a value change for
+    ///     <paramref name="path"/>. Updates the cache + fans out to every observer.
+    ///     Safe for concurrent callers; observers fire on the caller's thread.
+    /// </summary>
+    public void Push(string path, DataValueSnapshot snapshot)
+    {
+        if (string.IsNullOrEmpty(path)) throw new ArgumentException("path required", nameof(path));
+        ArgumentNullException.ThrowIfNull(snapshot);
+
+        _values[path] = snapshot;
+        if (!_observers.TryGetValue(path, out var list)) return;
+        Action<string, DataValueSnapshot>[] snapshotList;
+        lock (list) snapshotList = list.ToArray();
+        foreach (var observer in snapshotList) observer(path, snapshot);
+    }
+
+    /// <summary>Mirror of OPC UA <c>StatusCodes.BadNodeIdUnknown</c> without pulling the OPC stack dependency.</summary>
+    public const uint UpstreamNotConfigured = 0x80340000;
+
+    private sealed class Unsub(CachedTagUpstreamSource owner, string path, Action<string, DataValueSnapshot> observer) : IDisposable
+    {
+        private bool _disposed;
+        public void Dispose()
+        {
+            if (_disposed) return;
+            _disposed = true;
+            if (owner._observers.TryGetValue(path, out var list))
+                lock (list) list.Remove(observer);
+        }
+    }
+}
diff --git a/src/ZB.MOM.WW.OtOpcUa.Server/Phase7/Phase7EngineComposer.cs b/src/ZB.MOM.WW.OtOpcUa.Server/Phase7/Phase7EngineComposer.cs
new file mode 100644
index 0000000..289d841
--- /dev/null
+++ b/src/ZB.MOM.WW.OtOpcUa.Server/Phase7/Phase7EngineComposer.cs
@@ -0,0 +1,208 @@
+using Microsoft.Extensions.Logging;
+using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
+using ZB.MOM.WW.OtOpcUa.Core.Scripting;
+using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
+using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
+
+namespace ZB.MOM.WW.OtOpcUa.Server.Phase7;
+
+/// <summary>
+///     Phase 7 follow-up (task #243) — maps the generation's <see cref="Script"/> /
+///     <see cref="VirtualTag"/> / <see cref="ScriptedAlarm"/> rows into the runtime
+///     definitions <see cref="VirtualTagEngine"/> + <see cref="ScriptedAlarmEngine"/>
+///     expect, builds the engine instances, and returns the <see cref="IReadable"/>
+///     sources plus an <see cref="IAlarmSource"/> for the <c>DriverNodeManager</c>
+///     wiring added by task #239.
+/// </summary>
+/// <remarks>
+///     <para>
+///         Empty Phase 7 config (no virtual tags + no scripted alarms) is a valid state:
+///         <see cref="Compose"/> returns a <see cref="Phase7ComposedSources"/> with null
+///         sources so Program.cs can pass them through to <c>OpcUaApplicationHost</c>
+///         unchanged — deployments without scripts behave exactly as they did before
+///         Phase 7.
+///     </para>
+///     <para>
+///         The caller owns the returned <see cref="Phase7ComposedSources.Disposables"/>
+///         and must dispose them on shutdown. Engine cascades + timer ticks run off
+///         background threads until then.
+///     </para>
+/// </remarks>
+public static class Phase7EngineComposer
+{
+    public static Phase7ComposedSources Compose(
+        IReadOnlyList<Script> scripts,
+        IReadOnlyList<VirtualTag> virtualTags,
+        IReadOnlyList<ScriptedAlarm> scriptedAlarms,
+        CachedTagUpstreamSource upstream,
+        IAlarmStateStore alarmStateStore,
+        IAlarmHistorianSink historianSink,
+        Serilog.ILogger rootScriptLogger,
+        ILoggerFactory loggerFactory)
+    {
+        ArgumentNullException.ThrowIfNull(scripts);
+        ArgumentNullException.ThrowIfNull(virtualTags);
+        ArgumentNullException.ThrowIfNull(scriptedAlarms);
+        ArgumentNullException.ThrowIfNull(upstream);
+        ArgumentNullException.ThrowIfNull(alarmStateStore);
+        ArgumentNullException.ThrowIfNull(historianSink);
+        ArgumentNullException.ThrowIfNull(rootScriptLogger);
+        ArgumentNullException.ThrowIfNull(loggerFactory);
+
+        if (virtualTags.Count == 0 && scriptedAlarms.Count == 0)
+            return Phase7ComposedSources.Empty;
+
+        var scriptById = scripts
+            .Where(s => s.Enabled())
+            .ToDictionary(s => s.ScriptId, StringComparer.Ordinal);
+
+        var scriptLoggerFactory = new ScriptLoggerFactory(rootScriptLogger);
+        var disposables = new List<IDisposable>();
+
+        // Engines take Serilog.ILogger — each engine gets its own so rolling-file emissions
+        // stay keyed to the right source in the scripts-*.log.
+        VirtualTagSource? vtSource = null;
+        if (virtualTags.Count > 0)
+        {
+            var vtDefs = ProjectVirtualTags(virtualTags, scriptById).ToList();
+            var vtEngine = new VirtualTagEngine(upstream, scriptLoggerFactory, rootScriptLogger);
+            vtEngine.Load(vtDefs);
+            vtSource = new VirtualTagSource(vtEngine);
+            disposables.Add(vtEngine);
+        }
+
+        ScriptedAlarmSource? alarmSource = null;
+        if (scriptedAlarms.Count > 0)
+        {
+            var alarmDefs = ProjectScriptedAlarms(scriptedAlarms, scriptById).ToList();
+            var alarmEngine = new ScriptedAlarmEngine(upstream, alarmStateStore, scriptLoggerFactory, rootScriptLogger);
+            // Wire alarm emissions to the historian sink (Stream D). Fire-and-forget because
+            // the sink's EnqueueAsync is already non-blocking from the producer's view.
+            var engineLogger = loggerFactory.CreateLogger("Phase7HistorianRouter");
+            alarmEngine.OnEvent += (_, e) => _ = RouteToHistorianAsync(e, historianSink, engineLogger);
+            alarmEngine.LoadAsync(alarmDefs, CancellationToken.None).GetAwaiter().GetResult();
+            alarmSource = new ScriptedAlarmSource(alarmEngine);
+            disposables.Add(alarmEngine);
+            disposables.Add(alarmSource);
+        }
+
+        // ScriptedAlarmSource is an IAlarmSource, not an IReadable — scripted-alarm
+        // variable-read dispatch (task #245) needs a dedicated engine-state adapter. Until
+        // that ships, reads against Source=ScriptedAlarm nodes return BadNotFound per the
+        // DriverNodeManager null-check path (the ADR-002 "misconfiguration not silent
+        // fallback" signal). The alarm event stream still fires via IAlarmSource.
+        return new Phase7ComposedSources(vtSource, ScriptedAlarmReadable: null, disposables);
+    }
+
+    internal static IEnumerable<VirtualTagDefinition> ProjectVirtualTags(
+        IReadOnlyList<VirtualTag> rows, IReadOnlyDictionary<string, Script> scriptById)
+    {
+        foreach (var row in rows)
+        {
+            if (!row.Enabled) continue;
+            if (!scriptById.TryGetValue(row.ScriptId, out var script))
+                throw new InvalidOperationException(
+                    $"VirtualTag '{row.VirtualTagId}' references unknown / disabled Script '{row.ScriptId}' in this generation");
+
+            yield return new VirtualTagDefinition(
+                Path: row.VirtualTagId,
+                DataType: ParseDataType(row.DataType),
+                ScriptSource: script.SourceCode,
+                ChangeTriggered: row.ChangeTriggered,
+                TimerInterval: row.TimerIntervalMs.HasValue
+                    ? TimeSpan.FromMilliseconds(row.TimerIntervalMs.Value)
+                    : null,
+                Historize: row.Historize);
+        }
+    }
+
+    internal static IEnumerable<ScriptedAlarmDefinition> ProjectScriptedAlarms(
+        IReadOnlyList<ScriptedAlarm> rows, IReadOnlyDictionary<string, Script> scriptById)
+    {
+        foreach (var row in rows)
+        {
+            if (!row.Enabled) continue;
+            if (!scriptById.TryGetValue(row.PredicateScriptId, out var script))
+                throw new InvalidOperationException(
+                    $"ScriptedAlarm '{row.ScriptedAlarmId}' references unknown / disabled predicate Script '{row.PredicateScriptId}'");
+
+            yield return new ScriptedAlarmDefinition(
+                AlarmId: row.ScriptedAlarmId,
+                EquipmentPath: row.EquipmentId,
+                AlarmName: row.Name,
+                Kind: ParseAlarmKind(row.AlarmType),
+                Severity: MapSeverity(row.Severity),
+                MessageTemplate: row.MessageTemplate,
+                PredicateScriptSource: script.SourceCode,
+                HistorizeToAveva: row.HistorizeToAveva,
+                Retain: row.Retain);
+        }
+    }
+
+    private static DriverDataType ParseDataType(string raw) =>
+        Enum.TryParse<DriverDataType>(raw, ignoreCase: true, out var parsed) ? parsed : DriverDataType.String;
+
+    private static AlarmKind ParseAlarmKind(string raw) => raw switch
+    {
+        "AlarmCondition" => AlarmKind.AlarmCondition,
+        "LimitAlarm" => AlarmKind.LimitAlarm,
+        "DiscreteAlarm" => AlarmKind.DiscreteAlarm,
+        "OffNormalAlarm" => AlarmKind.OffNormalAlarm,
+        _ => throw new InvalidOperationException($"Unknown AlarmType '{raw}' — DB check constraint should have caught this"),
+    };
+
+    // OPC UA Part 9 severity bands (1..1000) → AlarmSeverity enum. Matches the same
+    // banding the AB CIP ALMA projection + OpcUaClient MapSeverity use.
+    private static AlarmSeverity MapSeverity(int s) => s switch
+    {
+        <= 250 => AlarmSeverity.Low,
+        <= 500 => AlarmSeverity.Medium,
+        <= 750 => AlarmSeverity.High,
+        _ => AlarmSeverity.Critical,
+    };
+
+    private static async Task RouteToHistorianAsync(
+        ScriptedAlarmEvent e, IAlarmHistorianSink sink, Microsoft.Extensions.Logging.ILogger log)
+    {
+        try
+        {
+            var historianEvent = new AlarmHistorianEvent(
+                AlarmId: e.AlarmId,
+                EquipmentPath: e.EquipmentPath,
+                AlarmName: e.AlarmName,
+                AlarmTypeName: e.Kind.ToString(),
+                Severity: e.Severity,
+                EventKind: e.Emission.ToString(),
+                Message: e.Message,
+                User: e.Condition.LastAckUser ?? "system",
+                Comment: e.Condition.LastAckComment,
+                TimestampUtc: e.TimestampUtc);
+            await sink.EnqueueAsync(historianEvent, CancellationToken.None).ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            log.LogWarning(ex, "Historian enqueue failed for alarm {AlarmId}/{Emission}", e.AlarmId, e.Emission);
+        }
+    }
+}
+
+/// <summary>What <see cref="Phase7EngineComposer.Compose"/> returns.</summary>
+/// <param name="VirtualReadable">Non-null when virtual tags were composed; pass to <c>OpcUaApplicationHost.virtualReadable</c>.</param>
+/// <param name="ScriptedAlarmReadable">Non-null when scripted alarms were composed; pass to <c>OpcUaApplicationHost.scriptedAlarmReadable</c>.</param>
+/// <param name="Disposables">Engine + source instances the caller owns. Dispose on shutdown.</param>
+public sealed record Phase7ComposedSources(
+    IReadable? VirtualReadable,
+    IReadable? ScriptedAlarmReadable,
+    IReadOnlyList<IDisposable> Disposables)
+{
+    public static readonly Phase7ComposedSources Empty =
+        new(null, null, Array.Empty<IDisposable>());
+}
+
+internal static class ScriptEnabledExtensions
+{
+    // Script has no explicit Enabled column; every row in the generation is a live script.
+    public static bool Enabled(this Script _) => true;
+}
diff --git a/src/ZB.MOM.WW.OtOpcUa.Server/ZB.MOM.WW.OtOpcUa.Server.csproj b/src/ZB.MOM.WW.OtOpcUa.Server/ZB.MOM.WW.OtOpcUa.Server.csproj
index 6014c93..9d870a1 100644
--- a/src/ZB.MOM.WW.OtOpcUa.Server/ZB.MOM.WW.OtOpcUa.Server.csproj
+++ b/src/ZB.MOM.WW.OtOpcUa.Server/ZB.MOM.WW.OtOpcUa.Server.csproj
@@ -30,6 +30,10 @@
 
     <ItemGroup>
         <ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Core\ZB.MOM.WW.OtOpcUa.Core.csproj"/>
+        <ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Core.Scripting\ZB.MOM.WW.OtOpcUa.Core.Scripting.csproj"/>
+        <ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Core.VirtualTags\ZB.MOM.WW.OtOpcUa.Core.VirtualTags.csproj"/>
+        <ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms\ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.csproj"/>
+        <ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian\ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.csproj"/>
         <ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Analyzers\ZB.MOM.WW.OtOpcUa.Analyzers.csproj"
                           OutputItemType="Analyzer" ReferenceOutputAssembly="false"/>
     </ItemGroup>
diff --git a/tests/ZB.MOM.WW.OtOpcUa.Server.Tests/Phase7/CachedTagUpstreamSourceTests.cs b/tests/ZB.MOM.WW.OtOpcUa.Server.Tests/Phase7/CachedTagUpstreamSourceTests.cs
new file mode 100644
index 0000000..13f5c0c
--- /dev/null
+++ b/tests/ZB.MOM.WW.OtOpcUa.Server.Tests/Phase7/CachedTagUpstreamSourceTests.cs
@@ -0,0 +1,83 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Server.Phase7;
+
+namespace ZB.MOM.WW.OtOpcUa.Server.Tests.Phase7;
+
+/// <summary>
+///     Covers the Phase 7 driver-to-engine bridge cache (task #243). Verifies the
+///     cache serves last-known values synchronously, fans out Push updates to
+///     subscribers, and cleans up on Dispose.
+/// </summary>
+[Trait("Category", "Unit")]
+public sealed class CachedTagUpstreamSourceTests
+{
+    private static DataValueSnapshot Snap(object? v) =>
+        new(v, 0u, DateTime.UtcNow, DateTime.UtcNow);
+
+    [Fact]
+    public void ReadTag_unknown_path_returns_BadNodeIdUnknown_snapshot()
+    {
+        var c = new CachedTagUpstreamSource();
+        var snap = c.ReadTag("/nowhere");
+        snap.Value.ShouldBeNull();
+        snap.StatusCode.ShouldBe(CachedTagUpstreamSource.UpstreamNotConfigured);
+    }
+
+    [Fact]
+    public void Push_then_Read_returns_cached_value()
+    {
+        var c = new CachedTagUpstreamSource();
+        c.Push("/Line1/Temp", Snap(42));
+        c.ReadTag("/Line1/Temp").Value.ShouldBe(42);
+    }
+
+    [Fact]
+    public void Push_fans_out_to_subscribers_in_registration_order()
+    {
+        var c = new CachedTagUpstreamSource();
+        var events = new List<string>();
+        c.SubscribeTag("/X", (p, s) => events.Add($"A:{p}:{s.Value}"));
+        c.SubscribeTag("/X", (p, s) => events.Add($"B:{p}:{s.Value}"));
+
+        c.Push("/X", Snap(7));
+
+        events.ShouldBe(["A:/X:7", "B:/X:7"]);
+    }
+
+    [Fact]
+    public void Push_to_different_path_does_not_fire_foreign_observer()
+    {
+        var c = new CachedTagUpstreamSource();
+        var fired = 0;
+        c.SubscribeTag("/X", (_, _) => fired++);
+
+        c.Push("/Y", Snap(1));
+        fired.ShouldBe(0);
+    }
+
+    [Fact]
+    public void Dispose_of_subscription_stops_fan_out()
+    {
+        var c = new CachedTagUpstreamSource();
+        var fired = 0;
+        var sub = c.SubscribeTag("/X", (_, _) => fired++);
+
+        c.Push("/X", Snap(1));
+        sub.Dispose();
+        c.Push("/X", Snap(2));
+
+        fired.ShouldBe(1);
+    }
+
+    [Fact]
+    public void Satisfies_both_VirtualTag_and_ScriptedAlarm_upstream_interfaces()
+    {
+        var c = new CachedTagUpstreamSource();
+        // Single instance is assignable to both — the composer passes it through for
+        // both engine constructors per the task #243 wiring.
+        ((Core.VirtualTags.ITagUpstreamSource)c).ShouldNotBeNull();
+        ((Core.ScriptedAlarms.ITagUpstreamSource)c).ShouldNotBeNull();
+    }
+}
diff --git a/tests/ZB.MOM.WW.OtOpcUa.Server.Tests/Phase7/Phase7EngineComposerTests.cs b/tests/ZB.MOM.WW.OtOpcUa.Server.Tests/Phase7/Phase7EngineComposerTests.cs
new file mode 100644
index 0000000..03f9f3e
--- /dev/null
+++ b/tests/ZB.MOM.WW.OtOpcUa.Server.Tests/Phase7/Phase7EngineComposerTests.cs
@@ -0,0 +1,143 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using Serilog;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
+using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
+using ZB.MOM.WW.OtOpcUa.Server.Phase7;
+
+namespace ZB.MOM.WW.OtOpcUa.Server.Tests.Phase7;
+
+/// <summary>
+///     Phase 7 follow-up (task #243) — verifies the composer that maps Config DB
+///     rows to runtime engine definitions + wires up VirtualTagEngine +
+///     ScriptedAlarmEngine + historian routing.
+/// </summary>
+[Trait("Category", "Unit")]
+public sealed class Phase7EngineComposerTests
+{
+    private static Script ScriptRow(string id, string source) => new()
+    {
+        ScriptRowId = Guid.NewGuid(), GenerationId = 1,
+        ScriptId = id, Name = id, SourceCode = source, SourceHash = "h",
+    };
+
+    private static VirtualTag VtRow(string id, string scriptId) => new()
+    {
+        VirtualTagRowId = Guid.NewGuid(), GenerationId = 1,
+        VirtualTagId = id, EquipmentId = "eq-1", Name = id,
+        DataType = "Float32", ScriptId = scriptId,
+    };
+
+    private static ScriptedAlarm AlarmRow(string id, string scriptId) => new()
+    {
+        ScriptedAlarmRowId = Guid.NewGuid(), GenerationId = 1,
+        ScriptedAlarmId = id, EquipmentId = "eq-1", Name = id,
+        AlarmType = "LimitAlarm", Severity = 500,
+        MessageTemplate = "x", PredicateScriptId = scriptId,
+    };
+
+    [Fact]
+    public void Compose_empty_rows_returns_Empty_sentinel()
+    {
+        var result = Phase7EngineComposer.Compose(
+            scripts: [],
+            virtualTags: [],
+            scriptedAlarms: [],
+            upstream: new CachedTagUpstreamSource(),
+            alarmStateStore: new InMemoryAlarmStateStore(),
+            historianSink: NullAlarmHistorianSink.Instance,
+            rootScriptLogger: new LoggerConfiguration().CreateLogger(),
+            loggerFactory: NullLoggerFactory.Instance);
+
+        result.ShouldBeSameAs(Phase7ComposedSources.Empty);
+        result.VirtualReadable.ShouldBeNull();
+        result.ScriptedAlarmReadable.ShouldBeNull();
+    }
+
+    [Fact]
+    public void Compose_VirtualTag_rows_returns_non_null_VirtualReadable()
+    {
+        var scripts = new[] { ScriptRow("scr-1", "return 1;") };
+        var vtags = new[] { VtRow("vt-1", "scr-1") };
+
+        var result = Phase7EngineComposer.Compose(
+            scripts, vtags, [],
+            upstream: new CachedTagUpstreamSource(),
+            alarmStateStore: new InMemoryAlarmStateStore(),
+            historianSink: NullAlarmHistorianSink.Instance,
+            rootScriptLogger: new LoggerConfiguration().CreateLogger(),
+            loggerFactory: NullLoggerFactory.Instance);
+
+        result.VirtualReadable.ShouldNotBeNull();
+        result.Disposables.Count.ShouldBeGreaterThan(0);
+    }
+
+    [Fact]
+    public void Compose_missing_script_reference_throws_with_actionable_message()
+    {
+        var vtags = new[] { VtRow("vt-1", "scr-missing") };
+
+        Should.Throw<InvalidOperationException>(() =>
+            Phase7EngineComposer.Compose(
+                scripts: [],
+                vtags, [],
+                upstream: new CachedTagUpstreamSource(),
+                alarmStateStore: new InMemoryAlarmStateStore(),
+                historianSink: NullAlarmHistorianSink.Instance,
+                rootScriptLogger: new LoggerConfiguration().CreateLogger(),
+                loggerFactory: NullLoggerFactory.Instance))
+            .Message.ShouldContain("scr-missing");
+    }
+
+    [Fact]
+    public void Compose_disabled_VirtualTag_is_skipped()
+    {
+        var scripts = new[] { ScriptRow("scr-1", "return 1;") };
+        var disabled = VtRow("vt-1", "scr-1");
+        disabled.Enabled = false;
+
+        var defs = Phase7EngineComposer.ProjectVirtualTags(
+            new[] { disabled },
+            new Dictionary<string, Script> { ["scr-1"] = scripts[0] }).ToList();
+
+        defs.ShouldBeEmpty();
+    }
+
+    [Fact]
+    public void ProjectVirtualTags_maps_timer_interval_milliseconds_to_TimeSpan()
+    {
+        var scripts = new[] { ScriptRow("scr-1", "return 1;") };
+        var vt = VtRow("vt-1", "scr-1");
+        vt.TimerIntervalMs = 2500;
+
+        var def = Phase7EngineComposer.ProjectVirtualTags(
+            new[] { vt },
+            new Dictionary<string, Script> { ["scr-1"] = scripts[0] }).Single();
+
+        def.TimerInterval.ShouldBe(TimeSpan.FromMilliseconds(2500));
+    }
+
+    [Fact]
+    public void ProjectScriptedAlarms_maps_Severity_numeric_to_AlarmSeverity_bucket()
+    {
+        var scripts = new[] { ScriptRow("scr-1", "return true;") };
+
+        var buckets = new[] { (1, AlarmSeverity.Low), (250, AlarmSeverity.Low),
+                               (251, AlarmSeverity.Medium), (500, AlarmSeverity.Medium),
+                               (501, AlarmSeverity.High), (750, AlarmSeverity.High),
+                               (751, AlarmSeverity.Critical), (1000, AlarmSeverity.Critical) };
+        foreach (var (input, expected) in buckets)
+        {
+            var row = AlarmRow("a1", "scr-1");
+            row.Severity = input;
+            var def = Phase7EngineComposer.ProjectScriptedAlarms(
+                new[] { row },
+                new Dictionary<string, Script> { ["scr-1"] = scripts[0] }).Single();
+            def.Severity.ShouldBe(expected, $"severity {input} should map to {expected}");
+        }
+    }
+}
+