diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs
new file mode 100644
index 00000000..9694b975
--- /dev/null
+++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs
@@ -0,0 +1,298 @@
+using System.Collections.Concurrent;
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using Google.Protobuf;
+using Grpc.Core;
+using Microsoft.Extensions.Logging.Abstractions;
+using NSubstitute;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+using ZB.MOM.WW.ScadaBridge.Communication.Actors;
+using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
+using ZB.MOM.WW.ScadaBridge.SiteRuntime;
+using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming;
+
+namespace ZB.MOM.WW.ScadaBridge.IntegrationTests.Grpc;
+
+///
+/// End-to-end trace for the aggregated live alarm stream (plan #10 deferred item, Task 7).
+///
+/// Every per-task unit test (T1–T6) exercises exactly one layer with its neighbour mocked:
+/// T1 the site broadcast filter (subscriber = TestKit probe, domain events), T2/the existing
+/// the server+relay+channel with a MOCKED
+/// stopping at the proto event, T3 the proto→domain client
+/// mapping in isolation, and T4 the aggregator fed hand-built domain events. Nothing wires the
+/// whole pipe together.
+///
+/// This test assembles the REAL chain end to end — mocking only the gRPC HTTP/2 transport:
+///
+/// domain AlarmStateChanged
+/// → SiteStreamManager.SubscribeSiteAlarms (real site-wide broadcast + alarm-only filter)
+/// → SiteStreamGrpcServer.SubscribeSite (real server handler)
+/// → StreamRelayActor (real domain→proto mapping + placeholder drop)
+/// → SiteStreamEvent proto → wire round-trip (ToByteArray/ParseFrom, simulates HTTP/2)
+/// → SiteStreamGrpcClient.ConvertToAlarmEvent (real proto→domain mapping)
+/// → SiteAlarmAggregatorActor cache (real dedup/seed/live)
+///
+/// and asserts the invariants the seams can't prove alone: AlarmKey identity
+/// (instance, name, sourceRef) + native enrichment survive every boundary; attributes and
+/// placeholder rows never reach the live cache; the single site-wide stream carries alarms
+/// for MULTIPLE instances (no per-instance filter); and a snapshot-seed row and a live delta
+/// for the same native alarm — both mapped through the real pipe — collapse onto ONE cache row.
+///
+/// Full gRPC-over-HTTP/2 remains a manual docker-cluster smoke (plan §4 Task 7).
+///
+public class SiteAlarmStreamEndToEndTests : TestKit
+{
+ private const string InstanceA = "SiteA.Pump01";
+ private const string InstanceB = "SiteA.Motor07";
+
+ // ── The full site→proto→client trace ────────────────────────────────────────
+
+ [Fact]
+ public async Task SiteWideAlarm_TraversesManagerToRelayToClient_PreservingIdentityAndEnrichment_DroppingAttributesAndPlaceholders()
+ {
+ // Site side: a REAL broadcast hub (no mock subscriber) wired straight into the
+ // REAL SubscribeSite server handler.
+ var manager = new SiteStreamManager(
+ new SiteRuntimeOptions { StreamBufferSize = 256 },
+ NullLogger.Instance);
+ manager.Initialize(Sys);
+
+ var server = new SiteStreamGrpcServer(manager, NullLogger.Instance);
+ server.SetReady(Sys);
+
+ var written = new ConcurrentQueue();
+ var writer = Substitute.For>();
+ writer.WriteAsync(Arg.Any(), Arg.Any())
+ .Returns(Task.CompletedTask)
+ .AndDoes(ci => written.Enqueue(ci.Arg()));
+
+ using var cts = new CancellationTokenSource();
+ var context = CreateMockContext(cts.Token);
+
+ var streamTask = Task.Run(() => server.SubscribeSite(
+ new SiteStreamRequest { CorrelationId = "e2e-site" }, writer, context));
+
+ // Wait until the server has actually subscribed to the site-wide hub, otherwise a
+ // publish can race ahead of the materialized subscription and be missed.
+ await WaitForConditionAsync(() => manager.SubscriptionCount == 1);
+
+ var raise = new DateTimeOffset(2026, 4, 2, 8, 30, 0, TimeSpan.Zero);
+ var ts = raise.AddSeconds(5);
+
+ // A fully-enriched NATIVE alarm on instance A (the payload whose fidelity we trace).
+ var native = new AlarmStateChanged(InstanceA, "Tank01.LevelAlarm", AlarmState.Active, 725, ts)
+ {
+ Level = AlarmLevel.HighHigh,
+ Message = "Tank 01 level critically high",
+ Kind = AlarmKind.NativeOpcUa,
+ Condition = new AlarmConditionState(
+ Active: true, Acknowledged: true, Confirmed: false,
+ Shelve: AlarmShelveState.OneShotShelved, Suppressed: false, Severity: 725),
+ SourceReference = "Tank01.Level.HiHi",
+ AlarmTypeName = "AnalogLimitAlarm.HiHi",
+ Category = "Process",
+ OperatorUser = "op.jane",
+ OperatorComment = "ack — investigating",
+ OriginalRaiseTime = raise,
+ CurrentValue = "98.4",
+ LimitValue = "95.0",
+ NativeSourceCanonicalName = "Tank01.LevelAlarm",
+ IsConfiguredPlaceholder = false
+ };
+
+ // A COMPUTED alarm on a DIFFERENT instance — must arrive over the SAME site-wide
+ // stream (proves the per-instance filter is gone).
+ var computedOtherInstance = new AlarmStateChanged(InstanceB, "OverSpeed", AlarmState.Active, 400, ts);
+
+ // Noise that MUST be dropped somewhere in the pipe:
+ var attribute = new AttributeValueChanged(InstanceA, "Modules.Flow", "GPM", 12.3, "Good", ts);
+ var placeholder = new AlarmStateChanged(InstanceA, "Motor1.MotorAlarms", AlarmState.Normal, 0, ts)
+ {
+ Kind = AlarmKind.NativeOpcUa,
+ IsConfiguredPlaceholder = true
+ };
+
+ // Interleave so a leak of the dropped rows would change the observed count/order.
+ manager.PublishAttributeValueChanged(attribute); // filtered at the manager (alarm-only)
+ manager.PublishAlarmStateChanged(native); // → 1 proto
+ manager.PublishAlarmStateChanged(placeholder); // dropped by StreamRelayActor
+ manager.PublishAlarmStateChanged(computedOtherInstance); // → 1 proto
+
+ await WaitForConditionAsync(() => written.Count >= 2);
+ // Give any leaked (attribute/placeholder) event a chance to show up before asserting.
+ await Task.Delay(150);
+
+ cts.Cancel();
+ await streamTask;
+
+ var protos = written.ToArray();
+ Assert.Equal(2, protos.Length);
+ Assert.All(protos, p =>
+ Assert.Equal(SiteStreamEvent.EventOneofCase.AlarmChanged, p.EventCase)); // no attribute leaked
+ Assert.DoesNotContain(protos, p => p.AlarmChanged.IsConfiguredPlaceholder); // no placeholder leaked
+
+ // Wire round-trip (serialize→deserialize) then the REAL client mapping back to domain.
+ var mapped = protos
+ .Select(p => SiteStreamGrpcClient.ConvertToAlarmEvent(RoundTripOverWire(p)))
+ .ToList();
+ Assert.All(mapped, m => Assert.NotNull(m));
+
+ // Both instances present — one site-wide stream, no per-instance filter.
+ var backA = mapped.Single(m => m!.InstanceUniqueName == InstanceA)!;
+ var backB = mapped.Single(m => m!.InstanceUniqueName == InstanceB)!;
+ Assert.Equal("OverSpeed", backB.AlarmName);
+
+ // AlarmKey identity survived end to end.
+ Assert.Equal("Tank01.LevelAlarm", backA.AlarmName);
+ Assert.Equal("Tank01.Level.HiHi", backA.SourceReference);
+
+ // Native enrichment survived every boundary (manager→relay→proto→wire→client).
+ Assert.Equal(AlarmKind.NativeOpcUa, backA.Kind);
+ Assert.Equal(AlarmState.Active, backA.State);
+ Assert.Equal(725, backA.Priority);
+ Assert.Equal(AlarmLevel.HighHigh, backA.Level);
+ Assert.Equal("Tank 01 level critically high", backA.Message);
+ Assert.True(backA.Condition.Active);
+ Assert.True(backA.Condition.Acknowledged);
+ Assert.Equal(AlarmShelveState.OneShotShelved, backA.Condition.Shelve);
+ Assert.Equal(725, backA.Condition.Severity);
+ Assert.Equal("AnalogLimitAlarm.HiHi", backA.AlarmTypeName);
+ Assert.Equal("Process", backA.Category);
+ Assert.Equal("op.jane", backA.OperatorUser);
+ Assert.Equal("ack — investigating", backA.OperatorComment);
+ Assert.Equal(raise, backA.OriginalRaiseTime);
+ Assert.Equal("98.4", backA.CurrentValue);
+ Assert.Equal("95.0", backA.LimitValue);
+ Assert.Equal("Tank01.LevelAlarm", backA.NativeSourceCanonicalName);
+ Assert.Equal(ts, backA.Timestamp);
+ Assert.False(backA.IsConfiguredPlaceholder);
+ }
+
+ // ── Snapshot-seed vs live-delta identity parity, through the real pipe ────────
+
+ [Fact]
+ public async Task RealPipeMapped_SeedRow_And_LiveDelta_ForSameNativeAlarm_CollapseOntoOneCacheRow()
+ {
+ // Produce TWO proto events for the SAME native alarm identity through the real
+ // manager→relay→proto pipe: an older one (used as the snapshot-seed row) and a
+ // newer one (fed as the live delta). Because both are mapped by the SAME real
+ // ConvertToAlarmEvent, their AlarmKeys are identical by construction — so the real
+ // aggregator must collapse them onto ONE row (no seed/live ghost duplicate).
+ var manager = new SiteStreamManager(
+ new SiteRuntimeOptions { StreamBufferSize = 256 },
+ NullLogger.Instance);
+ manager.Initialize(Sys);
+ var server = new SiteStreamGrpcServer(manager, NullLogger.Instance);
+ server.SetReady(Sys);
+
+ var written = new ConcurrentQueue();
+ var writer = Substitute.For>();
+ writer.WriteAsync(Arg.Any(), Arg.Any())
+ .Returns(Task.CompletedTask)
+ .AndDoes(ci => written.Enqueue(ci.Arg()));
+ using var cts = new CancellationTokenSource();
+ var streamTask = Task.Run(() => server.SubscribeSite(
+ new SiteStreamRequest { CorrelationId = "e2e-parity" }, writer, CreateMockContext(cts.Token)));
+ await WaitForConditionAsync(() => manager.SubscriptionCount == 1);
+
+ AlarmStateChanged Native(int priority, DateTimeOffset t) =>
+ new(InstanceA, "Tank01.LevelAlarm", AlarmState.Active, priority, t)
+ {
+ Kind = AlarmKind.NativeOpcUa,
+ SourceReference = "Tank01.Level.HiHi",
+ NativeSourceCanonicalName = "Tank01.LevelAlarm"
+ };
+
+ var t0 = new DateTimeOffset(2026, 4, 2, 9, 0, 0, TimeSpan.Zero);
+ manager.PublishAlarmStateChanged(Native(300, t0)); // → seed row (older)
+ manager.PublishAlarmStateChanged(Native(800, t0.AddSeconds(5))); // → live delta (newer)
+ await WaitForConditionAsync(() => written.Count >= 2);
+ cts.Cancel();
+ await streamTask;
+
+ var protos = written.ToArray();
+ var seedRow = SiteStreamGrpcClient.ConvertToAlarmEvent(RoundTripOverWire(protos[0]))!;
+ var liveRow = SiteStreamGrpcClient.ConvertToAlarmEvent(RoundTripOverWire(protos[1]))!;
+
+ // Drive a REAL aggregator: seed returns the older mapped row; then Tell it the newer
+ // mapped delta on the live path.
+ var factory = new NoopSiteStreamClientFactory();
+ var sink = new PublishSink();
+ var aggregator = Sys.ActorOf(Props.Create(() => new SiteAlarmAggregatorActor(
+ "site-alpha", "e2e-parity", _ => Task.FromResult>(new[] { seedRow }),
+ sink.Publish, factory, "http://a:5100", "http://b:5100", TimeSpan.FromMinutes(10))));
+
+ await WaitForConditionAsync(() => sink.Latest is { Count: 1 }); // seed applied
+ aggregator.Tell(liveRow);
+ await WaitForConditionAsync(() =>
+ sink.Latest is { Count: 1 } l && l[0].Priority == 800); // live delta collapsed in place
+
+ var final = sink.Latest!;
+ Assert.Single(final); // exactly ONE row for the (instance, name, sourceRef) identity
+ Assert.Equal("Tank01.Level.HiHi", final[0].SourceReference);
+ Assert.Equal(AlarmKind.NativeOpcUa, final[0].Kind);
+ }
+
+ // ── helpers ──────────────────────────────────────────────────────────────────
+
+ ///
+ /// Serialize→deserialize a proto event to mimic the gRPC HTTP/2 hop, proving the
+ /// enriched AlarmStateUpdate is fully wire-representable (no field lost in codec).
+ ///
+ private static SiteStreamEvent RoundTripOverWire(SiteStreamEvent evt) =>
+ SiteStreamEvent.Parser.ParseFrom(evt.ToByteArray());
+
+ private static ServerCallContext CreateMockContext(CancellationToken cancellationToken)
+ {
+ var context = Substitute.For();
+ context.CancellationToken.Returns(cancellationToken);
+ return context;
+ }
+
+ private static async Task WaitForConditionAsync(Func condition, int timeoutMs = 5000)
+ {
+ var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs);
+ while (!condition() && DateTime.UtcNow < deadline)
+ await Task.Delay(25);
+ Assert.True(condition(), $"Condition not met within {timeoutMs}ms");
+ }
+
+ private sealed class PublishSink
+ {
+ private readonly object _lock = new();
+ private IReadOnlyList? _latest;
+ public void Publish(IReadOnlyList snapshot)
+ {
+ lock (_lock) { _latest = snapshot; }
+ }
+ public IReadOnlyList? Latest
+ {
+ get { lock (_lock) { return _latest; } }
+ }
+ }
+
+ /// Aggregator transport double: its live stream simply stays open until cancelled.
+ private sealed class NoopSiteStreamClient : SiteStreamGrpcClient
+ {
+ public override Task SubscribeSiteAsync(
+ string correlationId, Action onAlarmEvent, Action onError, CancellationToken ct)
+ {
+ var tcs = new TaskCompletionSource();
+ ct.Register(() => tcs.TrySetResult());
+ return tcs.Task;
+ }
+
+ public override void Unsubscribe(string correlationId) { }
+ }
+
+ private sealed class NoopSiteStreamClientFactory : SiteStreamGrpcClientFactory
+ {
+ private readonly NoopSiteStreamClient _client = new();
+ public NoopSiteStreamClientFactory() : base(NullLoggerFactory.Instance) { }
+ public override SiteStreamGrpcClient GetOrCreate(string siteIdentifier, string grpcEndpoint) => _client;
+ public override SiteStreamGrpcClient? TryGet(string siteIdentifier, string grpcEndpoint) => _client;
+ }
+}