diff --git a/archreview/plans/R2-07-surgical-pure-adds-plan.md.tasks.json b/archreview/plans/R2-07-surgical-pure-adds-plan.md.tasks.json
index d743e7e6..356b451c 100644
--- a/archreview/plans/R2-07-surgical-pure-adds-plan.md.tasks.json
+++ b/archreview/plans/R2-07-surgical-pure-adds-plan.md.tasks.json
@@ -45,10 +45,11 @@
{
"id": "T5",
"subject": "Phase 1: over-the-wire SubscriptionSurvivalTests \u2014 subscribe, pure-add, item survives + new node browsable",
- "status": "pending",
+ "status": "deferred-live",
"blockedBy": [
"T4b"
- ]
+ ],
+ "note": "authored + compiles (dotnet build green); NOT run \u2014 *.IntegrationTests suite leaks ~16GB/run per the memory constraint. MUST run in the serial heavy integration pass (single-test, boots one in-process server + client)."
},
{
"id": "T6",
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/SubscriptionSurvivalTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/SubscriptionSurvivalTests.cs
new file mode 100644
index 00000000..62ee88d7
--- /dev/null
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/SubscriptionSurvivalTests.cs
@@ -0,0 +1,204 @@
+using System.Net;
+using System.Net.Sockets;
+using Microsoft.Extensions.Logging.Abstractions;
+using Opc.Ua;
+using Opc.Ua.Client;
+using Shouldly;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
+
+///
+/// R2-07 03/P1 — the HEADLINE proof: an over-the-wire subscription on an existing tag SURVIVES a
+/// pure-add deploy (one new tag), instead of being killed by the full-rebuild teardown that pre-R2-07
+/// every topology change forced. Boots the real , materialises tag A,
+/// opens a real client subscription + monitored item on A, then drives a PURE-ADD apply adding tag B
+/// (classifier ⇒ PureAdd ⇒ no rebuild ⇒ the idempotent Materialise passes create B and leave A's
+/// NodeState untouched), and asserts:
+///
+/// - A's monitored item still delivers a subsequent value write (the item is alive — no
+/// BadNodeIdUnknown, no dead subscription);
+/// - B is browsable + readable (the add landed).
+///
+/// Written against the master (pre-T2) behaviour this FAILS (the rebuild recreates A as a new NodeState
+/// and the monitored item goes dead); it passes once the classifier routing (T2) + the publish-actor
+/// announce (T4b) land.
+/// Run note: this is a heavy in-process server+client integration test — run it in the
+/// serial integration pass (it is NOT part of the lightweight unit sweep). The
+/// namespace URI is ; the client resolves the
+/// folder-scoped variable NodeIds against it.
+///
+public sealed class SubscriptionSurvivalTests
+{
+ private const string ServerUri = "urn:OtOpcUa.SubscriptionSurvival";
+
+ /// A subscription on tag A survives a pure-add deploy adding tag B; B becomes readable.
+ [Fact]
+ public async Task Subscription_on_existing_tag_survives_pure_add_of_another_tag()
+ {
+ var pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-subsurvive-{Guid.NewGuid():N}");
+ var port = AllocateFreePort();
+ var ct = TestContext.Current.CancellationToken;
+
+ try
+ {
+ var options = new OpcUaApplicationHostOptions
+ {
+ ApplicationName = ServerUri,
+ ApplicationUri = ServerUri,
+ OpcUaPort = port,
+ PublicHostname = "127.0.0.1",
+ PkiStoreRoot = pkiRoot,
+ EnabledSecurityProfiles = new List { OpcUaSecurityProfile.None },
+ AutoAcceptUntrustedClientCertificates = true,
+ };
+ var server = new OtOpcUaSdkServer();
+ await using var host = new OpcUaApplicationHost(options, NullLogger.Instance);
+ await host.StartAsync(server, ct);
+
+ var nm = server.NodeManager!;
+ var sink = new SdkAddressSpaceSink(nm);
+ var applier = new AddressSpaceApplier(sink, NullLogger.Instance);
+
+ // --- Materialise tag A (equipment eq-1, tag "A") and seed it a Good value. ---
+ var withA = CompositionWith(
+ new EquipmentTagPlan("tag-a", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Int32", FullName: "1", Writable: false, Alarm: null));
+ applier.MaterialiseHierarchy(withA);
+ applier.MaterialiseEquipmentTags(withA);
+ var nodeIdAString = Commons.OpcUa.EquipmentNodeIds.Variable("eq-1", "", "A");
+ sink.WriteValue(nodeIdAString, 41, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow);
+
+ // --- Open a client session + subscription + monitored item on A. ---
+ using var session = await OpenSessionAsync($"opc.tcp://127.0.0.1:{port}/OtOpcUa", ct);
+ var ns = (ushort)session.NamespaceUris.GetIndex(OtOpcUaNodeManager.DefaultNamespaceUri);
+ ns.ShouldBeGreaterThan((ushort)0);
+ var nodeIdA = new NodeId(nodeIdAString, ns);
+
+ var received = new List();
+ var gate = new object();
+ var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 };
+ session.AddSubscription(subscription);
+ await subscription.CreateAsync(ct);
+
+ var monitored = new MonitoredItem(subscription.DefaultItem)
+ {
+ StartNodeId = nodeIdA,
+ AttributeId = Attributes.Value,
+ SamplingInterval = 50,
+ QueueSize = 10,
+ };
+ monitored.Notification += (item, _) =>
+ {
+ foreach (var v in item.DequeueValues())
+ lock (gate) received.Add(v);
+ };
+ subscription.AddItem(monitored);
+ await subscription.ApplyChangesAsync(ct);
+
+ // Baseline: the initial value (41) is delivered.
+ await WaitUntilAsync(() => { lock (gate) return received.Count >= 1; }, TimeSpan.FromSeconds(5));
+
+ // --- Drive a PURE-ADD apply adding tag B (previous={A}, next={A,B}). ---
+ var withAB = CompositionWith(
+ new EquipmentTagPlan("tag-a", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Int32", FullName: "1", Writable: false, Alarm: null),
+ new EquipmentTagPlan("tag-b", "eq-1", "drv", FolderPath: "", Name: "B", DataType: "Int32", FullName: "2", Writable: false, Alarm: null));
+ var plan = AddressSpacePlanner.Compute(withA, withAB);
+ plan.AddedEquipmentTags.Count.ShouldBe(1);
+
+ var outcome = applier.Apply(plan);
+ outcome.RebuildCalled.ShouldBeFalse(); // PureAdd — no teardown
+
+ // Materialise passes (as the publish actor runs them) then announce — over the FULL composition.
+ applier.MaterialiseHierarchy(withAB);
+ applier.MaterialiseEquipmentTags(withAB);
+ applier.AnnounceAddedNodes(plan);
+
+ // --- A's monitored item must still be alive: a new write to A is delivered. ---
+ lock (gate) received.Clear();
+ sink.WriteValue(nodeIdAString, 42, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow);
+
+ await WaitUntilAsync(() => { lock (gate) return received.Any(v => Equals(v.Value, 42)); }, TimeSpan.FromSeconds(5));
+ lock (gate)
+ {
+ received.ShouldContain(v => Equals(v.Value, 42));
+ received.ShouldNotContain(v => v.StatusCode == StatusCodes.BadNodeIdUnknown);
+ }
+
+ // --- B is browsable + readable (the add landed). ---
+ var nodeIdB = new NodeId(Commons.OpcUa.EquipmentNodeIds.Variable("eq-1", "", "B"), ns);
+ sink.WriteValue(nodeIdB.Identifier.ToString()!, 7, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow);
+ var bValue = await session.ReadValueAsync(nodeIdB, ct);
+ bValue.StatusCode.ShouldNotBe(StatusCodes.BadNodeIdUnknown);
+ bValue.Value.ShouldBe(7);
+ }
+ finally
+ {
+ if (Directory.Exists(pkiRoot)) Directory.Delete(pkiRoot, recursive: true);
+ }
+ }
+
+ private static AddressSpaceComposition CompositionWith(params EquipmentTagPlan[] tags) =>
+ new(
+ UnsAreas: Array.Empty(),
+ UnsLines: Array.Empty(),
+ EquipmentNodes: new[] { new EquipmentNode("eq-1", "Equipment 1", "") },
+ DriverInstancePlans: Array.Empty(),
+ ScriptedAlarmPlans: Array.Empty())
+ {
+ EquipmentTags = tags,
+ };
+
+ private static async Task OpenSessionAsync(string endpointUrl, CancellationToken ct)
+ {
+ var appConfig = new ApplicationConfiguration
+ {
+ ApplicationName = "OtOpcUa.SubscriptionSurvivalClient",
+ ApplicationUri = $"urn:OtOpcUa.SubscriptionSurvivalClient.{Guid.NewGuid():N}",
+ ApplicationType = ApplicationType.Client,
+ SecurityConfiguration = new SecurityConfiguration
+ {
+ ApplicationCertificate = new CertificateIdentifier(),
+ AutoAcceptUntrustedCertificates = true,
+ },
+ ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 60_000 },
+ };
+ await appConfig.ValidateAsync(ApplicationType.Client, ct);
+ appConfig.CertificateValidator.CertificateValidation += (_, e) => e.Accept = true;
+
+ var endpoint = await CoreClientUtils.SelectEndpointAsync(
+ appConfig, endpointUrl, false, DefaultTelemetry.Create(_ => { }), ct);
+ var endpointConfiguration = EndpointConfiguration.Create(appConfig);
+ var configuredEndpoint = new ConfiguredEndpoint(null, endpoint, endpointConfiguration);
+
+ return await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync(
+ appConfig,
+ configuredEndpoint,
+ updateBeforeConnect: false,
+ checkDomain: false,
+ sessionName: "SubscriptionSurvivalTests",
+ sessionTimeout: 60_000,
+ identity: new UserIdentity(new AnonymousIdentityToken()),
+ preferredLocales: null,
+ ct: ct);
+ }
+
+ private static async Task WaitUntilAsync(Func condition, TimeSpan timeout)
+ {
+ var deadline = DateTime.UtcNow + timeout;
+ while (DateTime.UtcNow < deadline)
+ {
+ if (condition()) return;
+ await Task.Delay(50);
+ }
+ condition().ShouldBeTrue("condition not met within timeout");
+ }
+
+ private static int AllocateFreePort()
+ {
+ var listener = new TcpListener(IPAddress.Loopback, 0);
+ listener.Start();
+ var port = ((IPEndPoint)listener.LocalEndpoint).Port;
+ listener.Stop();
+ return port;
+ }
+}