Merge branch 'worktree-agent-ae22af64445b321d4' into arch-review-remediation

docs/requirements/Component-DataConnectionLayer.md

@@ -38,6 +38,23 @@ IDataConnection : IAsyncDisposable
 
 The `Disconnected` event is raised by an adapter when it detects an unexpected connection loss (server offline, network failure, keep-alive timeout). The `DataConnectionActor` subscribes to this event to trigger the reconnection state machine. Additional protocols can be added by implementing this interface.
 
+### Batch Capability Seam
+
+A protocol whose wire form can subscribe MANY tags in one round trip additionally implements the optional capability interface `IBatchSubscribableConnection` (same pattern as `IBrowsableDataConnection` / `IAlarmSubscribableConnection`):
+
+```
+IBatchSubscribableConnection
+├── SubscribeBatchAsync(tagPaths, callback) → IReadOnlyList<TagSubscribeResult>
+└── UnsubscribeBatchAsync(subscriptionIds) → void
+```
+
+- **One shared callback, plain tag list.** Every tag in a batch shares one callback; `SubscriptionCallback` already carries the tag path, so per-tag delegates would carry nothing.
+- **Partial-failure contract.** A per-tag fault (bad node id, unresolvable path) is a result row with `Success:false` and never aborts the batch — mirroring the gateway's `SubscribeResult` and OPC UA's per-monitored-item create status. A **thrown** exception means the whole batch failed; the actor classifies it exactly as the per-tag path does and drives the Reconnecting state machine on a connection-level fault.
+- Implementing the interface also asserts that `ReadBatchAsync` / `WriteBatchAsync` are **true bulk** service calls rather than loops over the single-tag methods. Both shipped adapters (`OpcUaDataConnection`, `MxGatewayDataConnection`) implement it; an adapter that does not keeps the historical per-tag behaviour, so the capability is purely additive.
+- Single-tag `SubscribeAsync` / `ReadAsync` / `WriteAsync` are **kept** (heartbeat monitor, interactive read/write paths) and delegate to the batch form as a batch of one.
+
+Sizing: at the 37,500-tag site target (`docs/deployment/topology-guide.md`), the pre-batch path issued one adapter round trip — and, on OPC UA, one `ApplyChanges` — per tag on every subscribe and every reconnect.
+
 ### Common Value Type
 
 All protocols produce the same value tuple consumed by Instance Actors. Before the first value update arrives from the DCL, data-sourced attributes are held at **uncertain** quality by the Instance Actor (see Site Runtime — Initialization):
@@ -147,6 +164,7 @@ All settings are parsed from the data connection's configuration JSON dictionari
 | `MaxNotificationsPerPublish` | int | `100` | Max notifications batched per publish cycle |
 | `SamplingIntervalMs` | int | `1000` | Per-item server sampling rate in milliseconds |
 | `QueueSize` | int | `10` | Per-item notification buffer size |
+| `MaxMonitoredItemsPerSubscription` | int | `5000` | Monitored-item budget per OPC UA subscription. Above it the adapter **shards** onto an additional subscription on the same session (37,500 tags → 8 shards). Per endpoint, because item-count ceilings are a property of the target server. |
 | `SecurityMode` | string | `None` | Preferred endpoint security: `None`, `Sign`, or `SignAndEncrypt` |
 | `AutoAcceptUntrustedCerts` | bool | `true` | Accept untrusted server certificates |
 
@@ -172,9 +190,17 @@ These are configured via `DataConnectionOptions` in `appsettings.json`, not per-
 | Setting | Default | Description |
 |---------|---------|-------------|
 | `ReconnectInterval` | 5s | Fixed interval between reconnection attempts |
-| `TagResolutionRetryInterval` | 10s | Retry interval for unresolved tag paths |
+| `TagResolutionRetryInterval` | 10s | **Floor** interval for the unresolved-tag retry; the retry backs off from here |
+| `TagResolutionRetryMaxInterval` | 5m | Ceiling for the exponential tag-resolution backoff |
 | `WriteTimeout` | 30s | Timeout for write operations |
-| `SeedReadTimeout` | 30s | Per-tag timeout for seed reads on the initial-subscribe (and reconnect re-seed) path. A hung device read is treated as a failed seed: retried up to `SeedReadMaxAttempts`, then the tag stays Uncertain until a change notification arrives. |
+| `SeedReadTimeout` | 30s | **Per-chunk** timeout for seed reads on the initial-subscribe (and reconnect re-seed) path. A hung device read is treated as a failed seed: retried up to `SeedReadMaxAttempts`, then the tag stays Uncertain until a change notification arrives. |
+| `SeedOverallTimeout` | 120s | Wall-clock deadline for the WHOLE seed (all chunks, all retry rounds). Replaces the old 30s-per-tag serial worst case; remaining tags are logged and stay Uncertain. |
+| `SeedReadBatchSize` | 250 | Tags per seed-read chunk |
+| `SeedReadMaxParallelism` | 4 | Seed-read chunks in flight concurrently |
+| `SubscribeBatchSize` | 500 | Tags per adapter subscribe round trip (initial subscribe, reconnect re-subscribe, resolution probes) |
+| `SubscribeBatchDelay` | 50ms | Delay BETWEEN reconnect re-subscribe chunks. Not applied on the instance-driven subscribe path — the Deployment Manager already staggers instance startup there. |
+| `QualityFlushInterval` | 1s | Coalescing window for pushing tag-quality counters to the health collector |
+| `MxSupervisoryAdviseParallelism` | 16 | Max in-flight supervisory advise commands on the MxGateway bulk-subscribe path when the endpoint has no write-user context |
 
 ## Subscription Management
 
@@ -184,6 +210,22 @@ These are configured via `DataConnectionOptions` in `appsettings.json`, not per-
 - When an Instance Actor is stopped (due to disable, delete, or redeployment), the DCL cleans up the associated subscriptions.
 - When a new Instance Actor is created for a redeployment, subscriptions are established fresh based on the new configuration.
 
+### Batching, chunking and pacing
+
+Three subscribe paths, deliberately paced differently:
+
+| Path | Shape |
+|---|---|
+| **Instance-driven subscribe** (`SubscribeTagsRequest`, i.e. site failover / deploy) | One `SubscribeBatchAsync` per `SubscribeBatchSize` chunk with **no** added delay. The Deployment Manager already staggers instance startup (`SiteRuntimeOptions.StartupBatchSize`/`StartupBatchDelayMs`), so requests arrive pre-spaced at ~75 tags each — double-staggering would only slow failover. |
+| **Reconnect re-subscribe** (`ReSubscribeAll`) | **Sequential** chunks of `SubscribeBatchSize` with `SubscribeBatchDelay` between them, all inside ONE background task; each chunk reports its per-tag results back to the actor as its own message. Replaces a tight loop that fired one fire-and-forget task per tag. |
+| **Tag-resolution probe** | The whole not-in-flight unresolved set, chunked at `SubscribeBatchSize`, as one probe round per timer tick. |
+
+Seeding (the initial-value read that keeps a STATIC tag from staying Uncertain) issues chunked bulk reads — `SeedReadBatchSize` per chunk, `SeedReadMaxParallelism` chunks in flight, `SeedReadTimeout` per chunk — under a single `SeedOverallTimeout` deadline covering every chunk and retry round.
+
+**OPC UA subscription sharding.** `RealOpcUaClient` places monitored items on the first subscription shard with free capacity (`MaxMonitoredItemsPerSubscription`), creating a shard when all are full and deleting one once it empties. A batch is N × `AddItem` plus **one** `ApplyChanges` per touched shard. Alarms & Conditions event items are pinned to a **dedicated event shard**, so `ConditionRefresh` has a single subscription id to target and `QueueSize:1000` event items never consume data-shard capacity. Every structural mutation (AddItem / RemoveItem / ApplyChanges / Create / Delete) and every shard-list mutation serializes behind ONE per-client gate — the SDK `Subscription` is not safe under concurrent structural mutation, and the subscribe task, resolution probe and alarm subscribe are otherwise unordered.
+
+**MxGateway.** A batch is one `SubscribeBulk` gateway command (AddItem + Advise per tag in a single worker pass), replacing the historical 2 RPCs per tag. When the endpoint has no write-user context (`WriteUserId == 0`) the worker offers no BULK supervisory advise, so the client issues one `AddItemBulk` and then pipelines the per-item supervisory advises with a bounded window (`MxSupervisoryAdviseParallelism`). A cross-repo follow-up (an additive `supervisory` flag on `SubscribeBulkCommand` in mxaccessgw) would collapse that to one RPC as well.
+
 ## Write-Back Support
 
 - When a script calls `Instance.SetAttribute` for an attribute with a data source reference, the Instance Actor sends a write request to the DCL.
@@ -263,6 +305,8 @@ The registered `SourceReference` is a **NodeId** for OPC UA (the picker, CSV and
 - **Capability check**: if `_adapter is not IAlarmSubscribableConnection`, the actor replies `SubscribeAlarmsResponse(Success = false, ...)`.
 - **Reconnect handling**: on entering **Reconnecting**, the actor pushes a `NativeAlarmSourceUnavailable` to every alarm subscriber (consumers mark mirrored alarms uncertain rather than clearing them). On successful reconnection it re-subscribes the feed; the adapter re-emits a snapshot, reconciling state.
 - **Shared condition filter (last-subscriber-wins)**: because the feed is opened once per source, it carries a single condition filter. A second subscriber that registers a *different* filter for the same source overwrites it (last writer wins) and the actor logs a warning — co-subscribers to one source are expected to agree on the filter.
+- **Routing index**: subscribed sources are bucketed by the **first path segment** of their reference (everything before the first `.`), and a transition is matched only against the bucket for its own first segment, plus a residue list of sources that carry no separator at all (a prefix shorter than one segment, which can match other buckets). This is sound because a source reference containing a separator can only prefix a reference sharing its entire first segment. The per-source `StartsWith` test and the condition-type gate inside the bucket are unchanged, so routing decisions are identical to the previous linear scan over every subscribed source; the `SnapshotComplete` sentinel still bypasses the index entirely and is broadcast to every subscriber.
+- **Gateway union filter (MxGateway)**: `StreamAlarms` carries a single `alarm_filter_prefix`, so the adapter opens the stream on the **longest common prefix** of the currently subscribed source references and restarts it only when a NEW source falls outside that prefix (the source then replays a fresh snapshot, which `NativeAlarmActor` already handles). An unsubscribe never restarts the stream — the prefix it leaves behind is at worst too broad, which costs bandwidth, not correctness. The prefix is a bandwidth optimisation only; the actor's per-source + condition-type gate remains authoritative, mirroring the OPC UA server-side WhereClause stance.
 
 ### Protocol-Neutral Types & Messages
 
@@ -359,8 +403,9 @@ When the DCL subscribes to a tag path from the flattened configuration but the p
 
 1. The failure is **logged to Site Event Logging**.
 2. The attribute is marked with quality `bad`.
-3. The DCL **periodically retries resolution** at a configurable interval, accommodating devices that come online in stages or load modules after startup.
+3. The DCL **retries resolution with exponential backoff**, accommodating devices that come online in stages or load modules after startup: the interval starts at `TagResolutionRetryInterval` (10s), doubles after a round in which nothing resolved, and is capped at `TagResolutionRetryMaxInterval` (5m). It resets to the floor as soon as **any** tag resolves or the connection reconnects, so a device that is merely slow to boot is still picked up quickly while a dead one is not probed at full width forever. Each round probes the whole unresolved set in `SubscribeBatchSize` chunks, not one call per tag.
-4. On successful resolution, the subscription activates normally and quality reflects the live value from the device.
+4. The retry timer is **single-shot, rescheduled on probe completion** — a periodic timer cannot back off, and rescheduling only on completion is what keeps a fan-out of failures from resetting the clock (the anti-starvation property previously defended by an `IsTimerActive` gate, which is retained).
+5. On successful resolution, the subscription activates normally and quality reflects the live value from the device.
 
 Note: Pre-deployment validation at central does **not** verify that tag paths resolve to real tags on physical devices — that is a runtime concern handled here.
 
@@ -370,6 +415,7 @@ The DCL reports the following metrics to the Health Monitoring component via the
 
 - **Connection status**: `connected`, `disconnected`, or `reconnecting` per data connection.
 - **Tag resolution counts**: Per connection, the number of total subscribed tags vs. successfully resolved tags. This gives operators visibility into misconfigured templates without needing to open the debug view for individual instances.
+- **Tag quality counters** are pushed on a genuine quality **transition** only, coalesced onto a `QualityFlushInterval` (1s) single-shot timer. Counter arithmetic still runs per message; only the collector push is deferred, and a value whose quality is unchanged moves no counter at all. Health reports poll at 30s, so the coalescing loses nothing. Three paths flush **synchronously** because their correctness depends on it: the bad-quality push on disconnect, unsubscribe, and the reconnect counter reset.
 
 ## Dependencies
 

src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Protocol/IBatchSubscribableConnection.cs

@@ -0,0 +1,60 @@
+namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
+
+/// <summary>
+/// Per-tag outcome of a batch subscribe. A failed row is a genuine per-tag fault
+/// (bad node id, unresolvable path) and never aborts the batch — mirroring the
+/// MxAccess Gateway's <c>SubscribeResult</c> and OPC UA's per-monitored-item create
+/// status. A batch that fails at connection level throws instead, so the
+/// DataConnectionActor can drive the reconnect state machine.
+/// </summary>
+/// <param name="TagPath">The requested tag path.</param>
+/// <param name="Success">Whether the tag was subscribed.</param>
+/// <param name="SubscriptionId">Adapter subscription handle when <paramref name="Success"/>; otherwise <c>null</c>.</param>
+/// <param name="ErrorMessage">Per-tag failure reason when not successful.</param>
+public record TagSubscribeResult(string TagPath, bool Success, string? SubscriptionId, string? ErrorMessage);
+
+/// <summary>
+/// Optional capability for an <see cref="IDataConnection"/> implementation whose
+/// protocol can subscribe/unsubscribe MANY tags in one round trip, and whose
+/// <see cref="IDataConnection.ReadBatchAsync"/> / <see cref="IDataConnection.WriteBatchAsync"/>
+/// are TRUE bulk service calls rather than a loop over the single-tag methods.
+/// Mirrors the <see cref="IBrowsableDataConnection"/> / <see cref="IAlarmSubscribableConnection"/>
+/// capability-interface pattern; consumed by the DataConnectionActor only.
+///
+/// <para>
+/// Implementing this interface is the adapter's assertion that batching is genuinely
+/// cheaper than N single calls: the actor then subscribes, unsubscribes and seed-reads
+/// in bounded chunks instead of per tag. An adapter that does NOT implement it keeps
+/// the historical per-tag behaviour, so the capability is purely additive.
+/// </para>
+/// </summary>
+public interface IBatchSubscribableConnection
+{
+    /// <summary>
+    /// Subscribes every tag in <paramref name="tagPaths"/> in as few protocol round
+    /// trips as the adapter allows, returning one <see cref="TagSubscribeResult"/> per
+    /// requested tag (in any order — callers key by <see cref="TagSubscribeResult.TagPath"/>).
+    /// All tags share ONE <paramref name="callback"/>; the callback already carries the
+    /// tag path, so per-tag delegates would carry nothing extra.
+    /// </summary>
+    /// <param name="tagPaths">The tag paths to subscribe.</param>
+    /// <param name="callback">Callback invoked for every value change on any of the tags.</param>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    /// <returns>One result row per requested tag path.</returns>
+    /// <exception cref="Exception">
+    /// A connection-level fault (adapter not connected / transport down) is thrown so the
+    /// caller classifies it as a connection failure; per-tag faults are result rows.
+    /// </exception>
+    Task<IReadOnlyList<TagSubscribeResult>> SubscribeBatchAsync(
+        IReadOnlyList<string> tagPaths, SubscriptionCallback callback, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Releases every supplied subscription id in as few protocol round trips as the
+    /// adapter allows. Unknown/stale ids are ignored, mirroring
+    /// <see cref="IDataConnection.UnsubscribeAsync"/>.
+    /// </summary>
+    /// <param name="subscriptionIds">Subscription ids previously returned by a subscribe call.</param>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    Task UnsubscribeBatchAsync(IReadOnlyList<string> subscriptionIds, CancellationToken cancellationToken = default);
+}

src/ZB.MOM.WW.ScadaBridge.Commons/Serialization/OpcUaEndpointConfigSerializer.cs

@@ -195,6 +195,7 @@ public static class OpcUaEndpointConfigSerializer
             ["DiscardOldest"] = config.DiscardOldest.ToString(),
             ["SubscriptionPriority"] = config.SubscriptionPriority.ToString(),
             ["SubscriptionDisplayName"] = config.SubscriptionDisplayName,
+            ["MaxMonitoredItemsPerSubscription"] = config.MaxMonitoredItemsPerSubscription.ToString(),
             ["TimestampsToReturn"] = config.TimestampsToReturn.ToString(),
         };
         if (config.Heartbeat is { } hb)
@@ -248,6 +249,7 @@ public static class OpcUaEndpointConfigSerializer
         TryAssignInt(dict, "KeepAliveCount", v => c.KeepAliveCount = v);
         TryAssignInt(dict, "LifetimeCount", v => c.LifetimeCount = v);
         TryAssignInt(dict, "MaxNotificationsPerPublish", v => c.MaxNotificationsPerPublish = v);
+        TryAssignInt(dict, "MaxMonitoredItemsPerSubscription", v => c.MaxMonitoredItemsPerSubscription = v);
 
         if (dict.TryGetValue("DiscardOldest", out var doStr) && bool.TryParse(doStr, out var doVal))
             c.DiscardOldest = doVal;

src/ZB.MOM.WW.ScadaBridge.Commons/Types/DataConnections/OpcUaEndpointConfig.cs

@@ -66,6 +66,14 @@ public sealed class OpcUaEndpointConfig
     /// Display name for the subscription.
     /// </summary>
     public string SubscriptionDisplayName { get; set; } = "ScadaBridge";
+    /// <summary>
+    /// Maximum monitored items placed on a single OPC UA subscription before the
+    /// adapter shards onto an additional subscription. Item-count ceilings are a
+    /// property of the target server, so this is per endpoint rather than global.
+    /// A site sized at 37,500 tags shards into 8 subscriptions at the default.
+    /// Values &lt;= 0 are treated as the default.
+    /// </summary>
+    public int MaxMonitoredItemsPerSubscription { get; set; } = 5000;
 
     // Read / filter
     /// <summary>

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/AlarmFilterPrefix.cs

@@ -0,0 +1,63 @@
+namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
+
+/// <summary>
+/// Computes the single source-reference prefix an MxAccess Gateway alarm stream can be
+/// opened with while still covering every subscribed source. <c>StreamAlarms</c> carries
+/// ONE <c>alarm_filter_prefix</c>, so the only pushable union of N source references is
+/// their longest common prefix.
+///
+/// <para>
+/// The prefix is a BANDWIDTH optimisation, never correctness: the DataConnectionActor's
+/// per-source match plus condition-type gate stays authoritative, mirroring the OPC UA
+/// server-side WhereClause stance. A too-broad prefix (including the empty,
+/// gateway-wide one) is therefore always safe.
+/// </para>
+/// </summary>
+internal static class AlarmFilterPrefix
+{
+    /// <summary>
+    /// Longest common prefix of the supplied source references, or the empty string when
+    /// they share none (or when any source is empty — an empty source means "everything").
+    /// </summary>
+    /// <param name="sourceReferences">Currently subscribed source references.</param>
+    /// <returns>The prefix to open the gateway alarm stream with; empty = gateway-wide.</returns>
+    public static string LongestCommonPrefix(IEnumerable<string> sourceReferences)
+    {
+        string? prefix = null;
+        foreach (var source in sourceReferences)
+        {
+            if (string.IsNullOrEmpty(source))
+                return string.Empty;
+
+            if (prefix is null)
+            {
+                prefix = source;
+                continue;
+            }
+
+            var max = Math.Min(prefix.Length, source.Length);
+            var common = 0;
+            while (common < max && prefix[common] == source[common])
+                common++;
+
+            if (common == 0)
+                return string.Empty;
+            prefix = prefix[..common];
+        }
+
+        return prefix ?? string.Empty;
+    }
+
+    /// <summary>
+    /// Whether an existing stream opened with <paramref name="currentPrefix"/> already
+    /// covers <paramref name="sourceReference"/>. Only a source falling OUTSIDE the
+    /// current prefix justifies restarting the stream; an unsubscribe never does,
+    /// because the prefix it leaves behind is at worst too broad — which costs
+    /// bandwidth, not correctness.
+    /// </summary>
+    /// <param name="currentPrefix">Prefix the live stream was opened with.</param>
+    /// <param name="sourceReference">Newly subscribed source reference.</param>
+    /// <returns><c>true</c> when the live stream already carries this source.</returns>
+    public static bool Covers(string currentPrefix, string sourceReference) =>
+        currentPrefix.Length == 0 || sourceReference.StartsWith(currentPrefix, StringComparison.Ordinal);
+}

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/AsyncSerialGate.cs

@@ -0,0 +1,55 @@
+namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
+
+/// <summary>
+/// Serializes async critical sections — one caller inside at a time, FIFO. Used by
+/// <see cref="RealOpcUaClient"/> so every structural mutation of an OPC UA Subscription
+/// (AddItem / RemoveItem / ApplyChanges / Create / Delete) and every shard-list mutation
+/// runs alone: the OPC Foundation Subscription object is not safe under concurrent
+/// structural mutation, and nothing else orders the subscribe background task, the
+/// tag-resolution probe and an alarm subscribe against each other.
+///
+/// <para>
+/// A named type rather than a bare <see cref="SemaphoreSlim"/> field so the "exactly one
+/// gate, released on every path" discipline is visible at each call site and testable on
+/// its own.
+/// </para>
+/// </summary>
+internal sealed class AsyncSerialGate
+{
+    private readonly SemaphoreSlim _gate = new(1, 1);
+
+    /// <summary>Runs <paramref name="action"/> with no other gated section in flight.</summary>
+    /// <param name="action">The critical section.</param>
+    /// <param name="cancellationToken">Cancellation token observed while waiting to enter.</param>
+    /// <returns>A task that completes when the section has run and the gate is released.</returns>
+    public async Task RunAsync(Func<Task> action, CancellationToken cancellationToken = default)
+    {
+        await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
+        try
+        {
+            await action().ConfigureAwait(false);
+        }
+        finally
+        {
+            _gate.Release();
+        }
+    }
+
+    /// <summary>Value-returning counterpart of <see cref="RunAsync(Func{Task}, CancellationToken)"/>.</summary>
+    /// <typeparam name="T">Result type of the critical section.</typeparam>
+    /// <param name="action">The critical section.</param>
+    /// <param name="cancellationToken">Cancellation token observed while waiting to enter.</param>
+    /// <returns>The section's result.</returns>
+    public async Task<T> RunAsync<T>(Func<Task<T>> action, CancellationToken cancellationToken = default)
+    {
+        await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
+        try
+        {
+            return await action().ConfigureAwait(false);
+        }
+        finally
+        {
+            _gate.Release();
+        }
+    }
+}

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/BulkPipeline.cs

@@ -0,0 +1,79 @@
+namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
+
+/// <summary>
+/// Bounded-parallel execution of a per-item operation. Used where a protocol offers no
+/// bulk form of a command and the only alternative is N serial round trips — today the
+/// MxAccess Gateway's supervisory advise, which exists per item only (the worker's
+/// <c>SubscribeBulk</c> issues plain advises). Pipelining 37,500 tags at a window of 16
+/// replaces 37,500 serial round trips with ~2,344 windows.
+///
+/// <para>
+/// Factored out of the gateway client so the concurrency bound is unit-testable without
+/// a live MXAccess session.
+/// </para>
+/// </summary>
+internal static class BulkPipeline
+{
+    /// <summary>
+    /// Runs <paramref name="operation"/> for every item with at most
+    /// <paramref name="maxParallelism"/> in flight, preserving result order. A per-item
+    /// fault is captured into that item's result slot via <paramref name="onError"/>
+    /// rather than aborting the pipeline; cancellation propagates.
+    /// </summary>
+    /// <typeparam name="TItem">Input item type.</typeparam>
+    /// <typeparam name="TResult">Per-item result type.</typeparam>
+    /// <param name="items">Items to process.</param>
+    /// <param name="maxParallelism">Maximum operations in flight; values &lt; 1 are treated as 1.</param>
+    /// <param name="operation">The per-item operation.</param>
+    /// <param name="onError">Maps an item plus its exception onto a result row.</param>
+    /// <param name="ct">Cancellation token.</param>
+    /// <returns>One result per item, in input order.</returns>
+    public static async Task<TResult[]> RunAsync<TItem, TResult>(
+        IReadOnlyList<TItem> items,
+        int maxParallelism,
+        Func<TItem, CancellationToken, Task<TResult>> operation,
+        Func<TItem, Exception, TResult> onError,
+        CancellationToken ct = default)
+    {
+        var results = new TResult[items.Count];
+        if (items.Count == 0)
+            return results;
+
+        // Deliberately NOT disposed: on cancellation Task.WhenAll rethrows while sibling
+        // operations may still be unwinding, and a disposed SemaphoreSlim would turn that
+        // into an ObjectDisposedException. SemaphoreSlim without AvailableWaitHandle holds
+        // no unmanaged resource, so letting it be collected is safe.
+        var gate = new SemaphoreSlim(Math.Max(1, maxParallelism));
+        var tasks = new Task[items.Count];
+        for (var i = 0; i < items.Count; i++)
+        {
+            var index = i;
+            var item = items[i];
+            tasks[i] = RunOneAsync(item, index);
+        }
+
+        await Task.WhenAll(tasks).ConfigureAwait(false);
+        return results;
+
+        async Task RunOneAsync(TItem item, int index)
+        {
+            await gate.WaitAsync(ct).ConfigureAwait(false);
+            try
+            {
+                results[index] = await operation(item, ct).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException)
+            {
+                throw;
+            }
+            catch (Exception ex)
+            {
+                results[index] = onError(item, ex);
+            }
+            finally
+            {
+                gate.Release();
+            }
+        }
+    }
+}

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/IMxGatewayClient.cs

@@ -6,7 +6,11 @@ namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
 /// <summary>Connection parameters resolved from the flat config dict.</summary>
 public record MxGatewayConnectionOptions(
     string Endpoint, string ApiKey, string ClientName, int WriteUserId,
-    bool UseTls, string? CaFile, string? ServerName, int ReadTimeoutMs);
+    bool UseTls, string? CaFile, string? ServerName, int ReadTimeoutMs,
+    // Maximum in-flight supervisory advise commands on the bulk-subscribe path when
+    // WriteUserId == 0 (the gateway worker has no BULK supervisory advise). Sourced from
+    // DataConnectionOptions.MxSupervisoryAdviseParallelism.
+    int SupervisoryAdviseParallelism = 16);
 
 /// <summary>One advised-tag value change pushed from the gateway event stream.</summary>
 public record MxValueUpdate(string TagPath, object? Value, QualityCode Quality, DateTimeOffset Timestamp);
@@ -17,6 +21,13 @@ public record MxReadOutcome(string TagPath, bool Success, object? Value, Quality
 /// <summary>Per-tag write outcome.</summary>
 public record MxWriteOutcome(string TagPath, bool Success, string? Error);
 
+/// <summary>Per-tag outcome of a bulk subscribe (AddItem + Advise in one gateway command).</summary>
+/// <param name="TagPath">The requested tag address.</param>
+/// <param name="Success">Whether the item was added and advised.</param>
+/// <param name="SubscriptionId">Gateway item handle (as a string) when successful.</param>
+/// <param name="Error">Per-tag failure reason when not successful.</param>
+public record MxSubscribeOutcome(string TagPath, bool Success, string? SubscriptionId, string? Error);
+
 /// <summary>One node in a Galaxy browse level.</summary>
 public record MxBrowseChild(string NodeId, string DisplayName, BrowseNodeClass NodeClass, bool HasChildren, string? DataType = null);
 
@@ -51,6 +62,25 @@ public interface IMxGatewayClient : IAsyncDisposable
     /// <returns>A task that represents the asynchronous operation.</returns>
     Task UnsubscribeAsync(string subscriptionId, CancellationToken ct = default);
 
+    /// <summary>
+    /// Adds and advises MANY tags in as few gateway commands as the worker allows —
+    /// ONE <c>SubscribeBulk</c> round trip in plain-advise mode, or one
+    /// <c>AddItemBulk</c> plus bounded-parallel supervisory advises when the connection
+    /// has no write-user context (the worker has no bulk supervisory advise).
+    /// Replaces the historical 2-RPC-per-tag AddItem + Advise pair.
+    /// </summary>
+    /// <param name="tagPaths">Tag addresses to subscribe.</param>
+    /// <param name="ct">Cancellation token.</param>
+    /// <returns>One outcome per requested tag path, in request order.</returns>
+    Task<IReadOnlyList<MxSubscribeOutcome>> SubscribeBulkAsync(
+        IReadOnlyList<string> tagPaths, CancellationToken ct = default);
+
+    /// <summary>UnAdvise + RemoveItem for many subscription ids in one gateway command.</summary>
+    /// <param name="subscriptionIds">Subscription ids previously returned by a subscribe call.</param>
+    /// <param name="ct">Cancellation token.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    Task UnsubscribeBulkAsync(IReadOnlyList<string> subscriptionIds, CancellationToken ct = default);
+
     /// <summary>Snapshot read of one or more tags (ReadBulk).</summary>
     /// <param name="tagPaths">Tag addresses to read.</param>
     /// <param name="ct">Cancellation token.</param>

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/IOpcUaClient.cs

@@ -26,7 +26,11 @@ public record OpcUaConnectionOptions(
     string SubscriptionDisplayName = "ScadaBridge",
     string TimestampsToReturn = "Source",
     OpcUaDeadbandOptions? Deadband = null,
-    OpcUaUserIdentityOptions? UserIdentity = null);
+    OpcUaUserIdentityOptions? UserIdentity = null,
+    // Monitored items above this ceiling are sharded onto an additional
+    // subscription on the same session (see RealOpcUaClient). Per endpoint, because
+    // item-count limits are a property of the target server.
+    int MaxMonitoredItemsPerSubscription = 5000);
 
 public record OpcUaDeadbandOptions(string Type, double Value);
 
@@ -37,6 +41,32 @@ public record OpcUaUserIdentityOptions(
     string CertificatePath,
     string CertificatePassword);
 
+/// <summary>
+/// Per-node outcome of a batch monitored-item create. A node that fails to resolve or
+/// whose monitored item comes back with a bad create status is reported as a failed row
+/// — never by aborting the whole batch (a connection-level fault still throws).
+/// </summary>
+/// <param name="NodeId">The requested node id, verbatim.</param>
+/// <param name="Success">Whether the monitored item was created.</param>
+/// <param name="SubscriptionHandle">Handle for <see cref="IOpcUaClient.RemoveSubscriptionAsync"/> when successful.</param>
+/// <param name="Error">Failure reason when not successful.</param>
+public record OpcUaSubscribeOutcome(string NodeId, bool Success, string? SubscriptionHandle, string? Error);
+
+/// <summary>Per-node outcome of a batch read.</summary>
+/// <param name="NodeId">The requested node id, verbatim.</param>
+/// <param name="Value">The value read, when the read produced one.</param>
+/// <param name="SourceTimestamp">Source timestamp reported by the server.</param>
+/// <param name="StatusCode">OPC UA status code for this node's read.</param>
+/// <param name="Error">Set when the node could not be read at all (e.g. an unresolvable node id).</param>
+public record OpcUaReadOutcome(
+    string NodeId, object? Value, DateTime SourceTimestamp, uint StatusCode, string? Error);
+
+/// <summary>Per-node outcome of a batch write.</summary>
+/// <param name="NodeId">The requested node id, verbatim.</param>
+/// <param name="StatusCode">OPC UA status code for this node's write (0 = Good).</param>
+/// <param name="Error">Set when the node could not be written at all (e.g. an unresolvable node id).</param>
+public record OpcUaWriteOutcome(string NodeId, uint StatusCode, string? Error);
+
 /// <summary>
 /// Abstraction over OPC UA client library for testability.
 /// The real implementation would wrap an OPC UA SDK (e.g., OPC Foundation .NET Standard Library).
@@ -74,6 +104,22 @@ public interface IOpcUaClient : IAsyncDisposable
         Action<string, object?, DateTime, uint> onValueChanged,
         CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Creates monitored items for MANY nodes with ONE ApplyChanges per touched
+    /// subscription — the batch counterpart of <see cref="CreateSubscriptionAsync"/> and
+    /// the reason the DCL no longer issues one ApplyChanges per tag. All nodes share the
+    /// single <paramref name="onValueChanged"/> callback (its first argument is the node
+    /// id, so per-node delegates would carry nothing extra).
+    /// </summary>
+    /// <param name="nodeIds">The node ids to monitor.</param>
+    /// <param name="onValueChanged">Callback invoked with (nodeId, value, sourceTimestamp, statusCode) on each change.</param>
+    /// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
+    /// <returns>One outcome per requested node id, in request order.</returns>
+    Task<IReadOnlyList<OpcUaSubscribeOutcome>> CreateSubscriptionsAsync(
+        IReadOnlyList<string> nodeIds,
+        Action<string, object?, DateTime, uint> onValueChanged,
+        CancellationToken cancellationToken = default);
+
     /// <summary>
     /// Removes a monitored item subscription by handle.
     /// </summary>
@@ -82,6 +128,16 @@ public interface IOpcUaClient : IAsyncDisposable
     /// <returns>A task representing the asynchronous operation.</returns>
     Task RemoveSubscriptionAsync(string subscriptionHandle, CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Removes many monitored items, grouping ONE ApplyChanges per touched subscription.
+    /// Unknown handles are ignored.
+    /// </summary>
+    /// <param name="subscriptionHandles">Handles previously returned by a create call.</param>
+    /// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
+    /// <returns>A task representing the asynchronous operation.</returns>
+    Task RemoveSubscriptionsAsync(
+        IReadOnlyList<string> subscriptionHandles, CancellationToken cancellationToken = default);
+
     /// <summary>
     /// Subscribes to OPC UA Alarms &amp; Conditions events under
     /// <paramref name="sourceNodeId"/> (or the Server object when null). On
@@ -124,6 +180,26 @@ public interface IOpcUaClient : IAsyncDisposable
     /// <returns>A task that completes with the OPC UA status code of the write operation.</returns>
     Task<uint> WriteValueAsync(string nodeId, object? value, CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Reads many nodes in ONE OPC UA Read service call (chunked internally against the
+    /// server's operation limits).
+    /// </summary>
+    /// <param name="nodeIds">The node ids to read.</param>
+    /// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
+    /// <returns>One outcome per requested node id, in request order.</returns>
+    Task<IReadOnlyList<OpcUaReadOutcome>> ReadValuesAsync(
+        IReadOnlyList<string> nodeIds, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Writes many nodes in ONE OPC UA Write service call (chunked internally against the
+    /// server's operation limits).
+    /// </summary>
+    /// <param name="values">The node id / value pairs to write.</param>
+    /// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
+    /// <returns>One outcome per requested node id, in request order.</returns>
+    Task<IReadOnlyList<OpcUaWriteOutcome>> WriteValuesAsync(
+        IReadOnlyList<(string NodeId, object? Value)> values, CancellationToken cancellationToken = default);
+
     /// <summary>
     /// Raised when the OPC UA session detects a keep-alive failure or the server
     /// becomes unreachable. The adapter layer uses this to trigger reconnection.
@@ -320,12 +396,29 @@ internal class StubOpcUaClient : IOpcUaClient
         return Task.FromResult(Guid.NewGuid().ToString());
     }
 
+    /// <inheritdoc />
+    public Task<IReadOnlyList<OpcUaSubscribeOutcome>> CreateSubscriptionsAsync(
+        IReadOnlyList<string> nodeIds,
+        Action<string, object?, DateTime, uint> onValueChanged,
+        CancellationToken cancellationToken = default)
+    {
+        IReadOnlyList<OpcUaSubscribeOutcome> outcomes = nodeIds
+            .Select(n => new OpcUaSubscribeOutcome(n, true, Guid.NewGuid().ToString(), null))
+            .ToList();
+        return Task.FromResult(outcomes);
+    }
+
     /// <inheritdoc />
     public Task RemoveSubscriptionAsync(string subscriptionHandle, CancellationToken cancellationToken = default)
     {
         return Task.CompletedTask;
     }
 
+    /// <inheritdoc />
+    public Task RemoveSubscriptionsAsync(
+        IReadOnlyList<string> subscriptionHandles, CancellationToken cancellationToken = default)
+        => Task.CompletedTask;
+
     /// <inheritdoc />
     public Task<string> CreateAlarmSubscriptionAsync(
         string? sourceNodeId, string? conditionFilter,
@@ -352,6 +445,26 @@ internal class StubOpcUaClient : IOpcUaClient
         return Task.FromResult<uint>(0); // Good status
     }
 
+    /// <inheritdoc />
+    public Task<IReadOnlyList<OpcUaReadOutcome>> ReadValuesAsync(
+        IReadOnlyList<string> nodeIds, CancellationToken cancellationToken = default)
+    {
+        IReadOnlyList<OpcUaReadOutcome> outcomes = nodeIds
+            .Select(n => new OpcUaReadOutcome(n, null, DateTime.UtcNow, 0u, null))
+            .ToList();
+        return Task.FromResult(outcomes);
+    }
+
+    /// <inheritdoc />
+    public Task<IReadOnlyList<OpcUaWriteOutcome>> WriteValuesAsync(
+        IReadOnlyList<(string NodeId, object? Value)> values, CancellationToken cancellationToken = default)
+    {
+        IReadOnlyList<OpcUaWriteOutcome> outcomes = values
+            .Select(v => new OpcUaWriteOutcome(v.NodeId, 0u, null))
+            .ToList();
+        return Task.FromResult(outcomes);
+    }
+
     /// <inheritdoc />
     public Task<BrowseChildrenResult> BrowseChildrenAsync(
         string? parentNodeId, string? continuationToken = null, CancellationToken cancellationToken = default)

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MonitoredItemShardPlanner.cs

@@ -0,0 +1,63 @@
+namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
+
+/// <summary>
+/// Pure placement policy for monitored items across OPC UA subscription shards: first
+/// shard with free capacity, otherwise a new shard. Factored out of
+/// <see cref="RealOpcUaClient"/> so the budget arithmetic is testable without a live
+/// server (the SDK's Session/Subscription types cannot be faked).
+/// </summary>
+internal static class MonitoredItemShardPlanner
+{
+    /// <summary>
+    /// Plans where <paramref name="itemCount"/> new items go given the current per-shard
+    /// item counts. Returns one shard index per item, in order; an index equal to or above
+    /// <paramref name="currentCounts"/>.Count means "a shard that must be created first".
+    /// </summary>
+    /// <param name="currentCounts">Item count of each existing shard, in shard order.</param>
+    /// <param name="budget">Maximum items per shard; values &lt;= 0 fall back to 1.</param>
+    /// <param name="itemCount">Number of items to place.</param>
+    /// <returns>Shard index per item, in request order.</returns>
+    public static IReadOnlyList<int> Plan(IReadOnlyList<int> currentCounts, int budget, int itemCount)
+    {
+        var effectiveBudget = budget > 0 ? budget : 1;
+        var counts = currentCounts.ToList();
+        var placement = new List<int>(itemCount);
+
+        for (var i = 0; i < itemCount; i++)
+        {
+            var target = -1;
+            for (var shard = 0; shard < counts.Count; shard++)
+            {
+                if (counts[shard] < effectiveBudget)
+                {
+                    target = shard;
+                    break;
+                }
+            }
+
+            if (target < 0)
+            {
+                counts.Add(0);
+                target = counts.Count - 1;
+            }
+
+            counts[target]++;
+            placement.Add(target);
+        }
+
+        return placement;
+    }
+
+    /// <summary>
+    /// Number of shards needed to hold <paramref name="itemCount"/> items at
+    /// <paramref name="budget"/> items per shard, starting from nothing.
+    /// </summary>
+    /// <param name="itemCount">Total monitored items.</param>
+    /// <param name="budget">Maximum items per shard; values &lt;= 0 fall back to 1.</param>
+    /// <returns>The shard count.</returns>
+    public static int ShardCountFor(int itemCount, int budget)
+    {
+        var effectiveBudget = budget > 0 ? budget : 1;
+        return (itemCount + effectiveBudget - 1) / effectiveBudget;
+    }
+}

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs

@@ -21,11 +21,13 @@ namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
 /// <see cref="Disconnected"/>, the actor disposes this adapter, creates a fresh one,
 /// reconnects and re-subscribes all tags.
 /// </summary>
-public class MxGatewayDataConnection : IDataConnection, IBrowsableDataConnection, IAlarmSubscribableConnection
+public class MxGatewayDataConnection
+    : IDataConnection, IBrowsableDataConnection, IAlarmSubscribableConnection, IBatchSubscribableConnection
 {
     private readonly IMxGatewayClientFactory _clientFactory;
     private readonly ILogger<MxGatewayDataConnection> _logger;
     private readonly ISecretResolver? _secretResolver;
+    private readonly int _supervisoryAdviseParallelism;
     private IMxGatewayClient? _client;
     private ConnectionHealth _status = ConnectionHealth.Disconnected;
     private CancellationTokenSource? _eventLoopCts;
@@ -39,6 +41,13 @@ public class MxGatewayDataConnection : IDataConnection, IBrowsableDataConnection
     private int _alarmSubCount;
     private readonly object _alarmLock = new();
 
+    // Source references currently subscribed (alarm subscription id → source reference),
+    // and the prefix the live stream was opened with. StreamAlarms carries ONE prefix, so
+    // the pushable union is the longest common prefix of the active sources: the stream is
+    // restarted only when a NEW source falls outside it. Guarded by _alarmLock.
+    private readonly Dictionary<string, string> _alarmSubSources = new(StringComparer.Ordinal);
+    private string _alarmStreamPrefix = string.Empty;
+
     // subscriptionId → (tagPath, callback) so the event loop can route updates by tag,
     // plus tagPath → subscriptionId for reverse lookup. Concurrent because the event
     // loop reads from a background thread while Subscribe/Unsubscribe mutate.
@@ -58,14 +67,20 @@ public class MxGatewayDataConnection : IDataConnection, IBrowsableDataConnection
     /// connect time. When null, a literal ApiKey still works, but a <c>secret:</c> reference
     /// fails closed (a connection is never established with an unresolved key).
     /// </param>
+    /// <param name="supervisoryAdviseParallelism">
+    /// Maximum in-flight supervisory advise commands on the bulk-subscribe path when the
+    /// endpoint has no write-user context (<c>DataConnectionOptions.MxSupervisoryAdviseParallelism</c>).
+    /// </param>
     public MxGatewayDataConnection(
         IMxGatewayClientFactory clientFactory,
         ILogger<MxGatewayDataConnection> logger,
-        ISecretResolver? secretResolver = null)
+        ISecretResolver? secretResolver = null,
+        int supervisoryAdviseParallelism = 16)
     {
         _clientFactory = clientFactory;
         _logger = logger;
         _secretResolver = secretResolver;
+        _supervisoryAdviseParallelism = Math.Max(1, supervisoryAdviseParallelism);
     }
 
     /// <inheritdoc />
@@ -129,7 +144,8 @@ public class MxGatewayDataConnection : IDataConnection, IBrowsableDataConnection
             cfg.UseTls,
             string.IsNullOrWhiteSpace(cfg.CaFile) ? null : cfg.CaFile,
             string.IsNullOrWhiteSpace(cfg.ServerName) ? null : cfg.ServerName,
-            cfg.ReadTimeoutMs), cancellationToken);
+            cfg.ReadTimeoutMs,
+            _supervisoryAdviseParallelism), cancellationToken);
 
         _status = ConnectionHealth.Connected;
 
@@ -192,6 +208,8 @@ public class MxGatewayDataConnection : IDataConnection, IBrowsableDataConnection
             _alarmCts?.Dispose();
             _alarmCts = null;
             _alarmSubCount = 0;
+            _alarmSubSources.Clear();
+            _alarmStreamPrefix = string.Empty;
         }
         if (_client is not null)
             await _client.DisconnectAsync(cancellationToken);
@@ -215,28 +233,79 @@ public class MxGatewayDataConnection : IDataConnection, IBrowsableDataConnection
         await _client!.UnsubscribeAsync(subscriptionId, cancellationToken);
     }
 
+    /// <inheritdoc />
+    public async Task<IReadOnlyList<TagSubscribeResult>> SubscribeBatchAsync(
+        IReadOnlyList<string> tagPaths, SubscriptionCallback callback, CancellationToken cancellationToken = default)
+    {
+        // ONE gateway round trip per batch (SubscribeBulk = AddItem + Advise per tag in a
+        // single worker pass) instead of the historical 2 RPCs per tag.
+        var outcomes = await _client!.SubscribeBulkAsync(tagPaths, cancellationToken);
+        var results = new List<TagSubscribeResult>(outcomes.Count);
+        foreach (var outcome in outcomes)
+        {
+            if (outcome is { Success: true, SubscriptionId: not null })
+            {
+                _subs[outcome.SubscriptionId] = (outcome.TagPath, callback);
+                _tagToSub[outcome.TagPath] = outcome.SubscriptionId;
+            }
+
+            results.Add(new TagSubscribeResult(
+                outcome.TagPath, outcome.Success, outcome.SubscriptionId, outcome.Error));
+        }
+
+        return results;
+    }
+
+    /// <inheritdoc />
+    public async Task UnsubscribeBatchAsync(
+        IReadOnlyList<string> subscriptionIds, CancellationToken cancellationToken = default)
+    {
+        foreach (var id in subscriptionIds)
+        {
+            if (_subs.TryRemove(id, out var s))
+                _tagToSub.TryRemove(s.TagPath, out _);
+        }
+
+        await _client!.UnsubscribeBulkAsync(subscriptionIds, cancellationToken);
+    }
+
     /// <inheritdoc />
     public Task<string> SubscribeAlarmsAsync(
         string sourceReference, string? conditionFilter,
         AlarmTransitionCallback callback, CancellationToken cancellationToken = default)
     {
+        var subscriptionId = Guid.NewGuid().ToString();
         lock (_alarmLock)
         {
             _alarmSubCount++;
-            if (_alarmCts == null)
+            _alarmSubSources[subscriptionId] = sourceReference;
+
+            // Push the UNION of the subscribed source references to the gateway as the
+            // stream's single filter prefix. It is a bandwidth optimisation only — the
+            // MxGateway has no server-side CONDITION filter either, so conditionFilter is
+            // still not forwarded: the DataConnectionActor's per-source + per-condition-type
+            // gate stays authoritative, uniform with the OPC UA WhereClause stance.
+            if (_alarmCts != null && AlarmFilterPrefix.Covers(_alarmStreamPrefix, sourceReference))
+                return Task.FromResult(subscriptionId);
+
+            // A new source outside the live prefix (or the first subscriber): (re)open the
+            // stream on the recomputed union. The source replays a fresh snapshot, which
+            // NativeAlarmActor already handles.
+            if (_alarmCts != null)
             {
-                _alarmCts = new CancellationTokenSource();
+                _alarmCts.Cancel();
-                var token = _alarmCts.Token;
+                _alarmCts.Dispose();
-                var client = _client!;
-                // Gateway-wide feed (null prefix). The MxGateway has no server-side
-                // condition filter, so conditionFilter is intentionally NOT forwarded
-                // here: the DataConnectionActor applies it as the authoritative
-                // client-side gate per source reference AND per condition type
-                // (AlarmConditionFilter), uniform with the OPC UA path.
-                _ = Task.Run(() => client.RunAlarmStreamAsync(null, t => callback(t), token), token);
             }
+
+            _alarmStreamPrefix = AlarmFilterPrefix.LongestCommonPrefix(_alarmSubSources.Values);
+            _alarmCts = new CancellationTokenSource();
+            var token = _alarmCts.Token;
+            var client = _client!;
+            var prefix = _alarmStreamPrefix.Length == 0 ? null : _alarmStreamPrefix;
+            _ = Task.Run(() => client.RunAlarmStreamAsync(prefix, t => callback(t), token), token);
         }
-        return Task.FromResult(Guid.NewGuid().ToString());
+
+        return Task.FromResult(subscriptionId);
     }
 
     /// <inheritdoc />
@@ -244,13 +313,17 @@ public class MxGatewayDataConnection : IDataConnection, IBrowsableDataConnection
     {
         lock (_alarmLock)
         {
+            _alarmSubSources.Remove(subscriptionId);
             if (_alarmSubCount > 0)
                 _alarmSubCount--;
+            // Deliberately no stream restart here: dropping a source can only leave the
+            // prefix too BROAD, which costs bandwidth, never correctness.
             if (_alarmSubCount == 0)
             {
                 _alarmCts?.Cancel();
                 _alarmCts?.Dispose();
                 _alarmCts = null;
+                _alarmStreamPrefix = string.Empty;
             }
         }
         return Task.CompletedTask;
@@ -354,6 +427,8 @@ public class MxGatewayDataConnection : IDataConnection, IBrowsableDataConnection
             _alarmCts?.Dispose();
             _alarmCts = null;
             _alarmSubCount = 0;
+            _alarmSubSources.Clear();
+            _alarmStreamPrefix = string.Empty;
         }
         if (_client is not null)
             await _client.DisposeAsync();

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/OpcUaDataConnection.cs

@@ -18,7 +18,9 @@ namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
 /// - Read/Write → Read/Write service calls
 /// - Quality → OPC UA StatusCode mapping
 /// </summary>
-public class OpcUaDataConnection : IDataConnection, IBrowsableDataConnection, IAlarmSubscribableConnection, IAddressSpaceSearchable
+public class OpcUaDataConnection
+    : IDataConnection, IBrowsableDataConnection, IAlarmSubscribableConnection, IAddressSpaceSearchable,
+      IBatchSubscribableConnection
 {
     private readonly IOpcUaClientFactory _clientFactory;
     private readonly ILogger<OpcUaDataConnection> _logger;
@@ -93,7 +95,8 @@ public class OpcUaDataConnection : IDataConnection, IBrowsableDataConnection, IA
                 ? new OpcUaUserIdentityOptions(
                     ui.TokenType.ToString(), ui.Username, ui.Password,
                     ui.CertificatePath, ui.CertificatePassword)
-                : null);
+                : null,
+            MaxMonitoredItemsPerSubscription: config.MaxMonitoredItemsPerSubscription);
 
         _status = ConnectionHealth.Connecting;
 
@@ -193,6 +196,36 @@ public class OpcUaDataConnection : IDataConnection, IBrowsableDataConnection, IA
             cancellationToken);
     }
 
+    /// <inheritdoc />
+    public async Task<IReadOnlyList<TagSubscribeResult>> SubscribeBatchAsync(
+        IReadOnlyList<string> tagPaths, SubscriptionCallback callback, CancellationToken cancellationToken = default)
+    {
+        EnsureConnected();
+
+        // ONE shared callback for the whole batch — the client hands back the node id
+        // (== tag path for OPC UA), so no per-tag closure is needed.
+        var outcomes = await _client!.CreateSubscriptionsAsync(
+            tagPaths,
+            (nodeId, value, timestamp, statusCode) =>
+            {
+                var quality = MapStatusCode(statusCode);
+                callback(nodeId, new TagValue(value, quality, new DateTimeOffset(timestamp, TimeSpan.Zero)));
+            },
+            cancellationToken);
+
+        return outcomes
+            .Select(o => new TagSubscribeResult(o.NodeId, o.Success, o.SubscriptionHandle, o.Error))
+            .ToList();
+    }
+
+    /// <inheritdoc />
+    public async Task UnsubscribeBatchAsync(
+        IReadOnlyList<string> subscriptionIds, CancellationToken cancellationToken = default)
+    {
+        if (_client != null)
+            await _client.RemoveSubscriptionsAsync(subscriptionIds, cancellationToken);
+    }
+
     /// <inheritdoc />
     public async Task<string> SubscribeAlarmsAsync(
         string sourceReference, string? conditionFilter,
@@ -249,27 +282,59 @@ public class OpcUaDataConnection : IDataConnection, IBrowsableDataConnection, IA
     /// <inheritdoc />
     public async Task<IReadOnlyDictionary<string, ReadResult>> ReadBatchAsync(IEnumerable<string> tagPaths, CancellationToken cancellationToken = default)
     {
-        // A single failing tag must not abort the whole batch.
+        EnsureConnected();
-        // ReadAsync re-throws non-cancellation exceptions; catch them per tag and record
+
-        // a failed ReadResult so the caller receives a complete result map for every
+        // TRUE bulk: one OPC UA Read service call (chunked inside the client against the
-        // requested tag (the ReadResult shape already carries per-tag Success/error).
+        // server's operation limits), not a loop over ReadAsync. A single failing tag
-        var results = new Dictionary<string, ReadResult>();
+        // still comes back as a failed ReadResult row rather than aborting the batch;
-        foreach (var tagPath in tagPaths)
+        // OperationCanceledException still aborts the whole batch.
+        var requested = tagPaths as IReadOnlyList<string> ?? tagPaths.ToList();
+        var results = new Dictionary<string, ReadResult>(requested.Count);
+        if (requested.Count == 0)
+            return results;
+
+        try
         {
-            try
+            var outcomes = await _client!.ReadValuesAsync(requested, cancellationToken);
+            foreach (var outcome in outcomes)
             {
-                results[tagPath] = await ReadAsync(tagPath, cancellationToken);
+                if (outcome.Error != null)
-            }
+                {
-            catch (OperationCanceledException)
+                    results[outcome.NodeId] = new ReadResult(false, null, outcome.Error);
-            {
+                    continue;
-                // Cancellation aborts the whole batch — propagate it.
+                }
-                throw;
+
-            }
+                var quality = MapStatusCode(outcome.StatusCode);
-            catch (Exception ex)
+                results[outcome.NodeId] = quality == QualityCode.Bad
-            {
+                    ? new ReadResult(false, null, $"OPC UA read returned bad status: 0x{outcome.StatusCode:X8}")
-                results[tagPath] = new ReadResult(false, null, ex.Message);
+                    : new ReadResult(true,
+                        new TagValue(outcome.Value, quality, new DateTimeOffset(outcome.SourceTimestamp, TimeSpan.Zero)),
+                        null);
             }
         }
+        catch (OperationCanceledException)
+        {
+            // Cancellation aborts the whole batch — propagate it.
+            throw;
+        }
+        catch (Exception ex)
+        {
+            // A batch-level fault (session dropped mid-call) mirrors ReadAsync: signal the
+            // disconnect, then report every requested tag as failed so the caller still
+            // receives a complete map instead of an exception.
+            _logger.LogWarning(ex, "OPC UA batch read failed — connection may be lost");
+            RaiseDisconnected();
+            foreach (var tagPath in requested)
+                results[tagPath] = new ReadResult(false, null, ex.Message);
+        }
+
+        // Defensive: a non-conformant client/server that omits a requested tag.
+        foreach (var tagPath in requested)
+        {
+            if (!results.ContainsKey(tagPath))
+                results[tagPath] = new ReadResult(false, null, "Tag missing from OPC UA read result.");
+        }
+
         return results;
     }
 
@@ -288,29 +353,46 @@ public class OpcUaDataConnection : IDataConnection, IBrowsableDataConnection, IA
     /// <inheritdoc />
     public async Task<IReadOnlyDictionary<string, WriteResult>> WriteBatchAsync(IDictionary<string, object?> values, CancellationToken cancellationToken = default)
     {
-        // A mid-batch fault must not abort the whole batch.
+        EnsureConnected();
-        // WriteAsync calls EnsureConnected(), which throws InvalidOperationException when
+
-        // the connection drops partway through; catch per-tag exceptions and record a
+        // TRUE bulk: one OPC UA Write service call (chunked inside the client). A mid-batch
-        // failed WriteResult so the caller (including WriteBatchAndWaitAsync) receives a
+        // fault must not abort the batch — every requested tag gets a WriteResult row —
-        // complete result map. OperationCanceledException is still propagated so a
+        // while OperationCanceledException still aborts as a whole.
-        // cancelled batch aborts as a whole — mirrors the ReadBatchAsync fix.
+        var requested = values.ToList();
-        var results = new Dictionary<string, WriteResult>();
+        var results = new Dictionary<string, WriteResult>(requested.Count);
-        foreach (var (tagPath, value) in values)
+        if (requested.Count == 0)
+            return results;
+
+        try
         {
-            try
+            var outcomes = await _client!.WriteValuesAsync(
+                requested.Select(kv => (kv.Key, kv.Value)).ToList(), cancellationToken);
+            foreach (var outcome in outcomes)
             {
-                results[tagPath] = await WriteAsync(tagPath, value, cancellationToken);
+                results[outcome.NodeId] = outcome.Error != null
-            }
+                    ? new WriteResult(false, outcome.Error)
-            catch (OperationCanceledException)
+                    : outcome.StatusCode != 0
-            {
+                        ? new WriteResult(false, $"OPC UA write failed with status: 0x{outcome.StatusCode:X8}")
-                // Cancellation aborts the whole batch — propagate it.
+                        : new WriteResult(true, null);
-                throw;
-            }
-            catch (Exception ex)
-            {
-                results[tagPath] = new WriteResult(false, ex.Message);
             }
         }
+        catch (OperationCanceledException)
+        {
+            // Cancellation aborts the whole batch — propagate it.
+            throw;
+        }
+        catch (Exception ex)
+        {
+            foreach (var (tagPath, _) in requested)
+                results[tagPath] = new WriteResult(false, ex.Message);
+        }
+
+        foreach (var (tagPath, _) in requested)
+        {
+            if (!results.ContainsKey(tagPath))
+                results[tagPath] = new WriteResult(false, "Tag missing from OPC UA write result.");
+        }
+
         return results;
     }
 

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealMxGatewayClient.cs

@@ -26,6 +26,7 @@ public sealed class RealMxGatewayClient : IMxGatewayClient
     private int _serverHandle;
     private int _writeUserId;
     private int _readTimeoutMs;
+    private int _supervisoryAdviseParallelism = 16;
     private ulong _lastSeq;
 
     // tag ↔ MXAccess item handle, maintained across subscribe/write.
@@ -65,6 +66,7 @@ public sealed class RealMxGatewayClient : IMxGatewayClient
     {
         _writeUserId = options.WriteUserId;
         _readTimeoutMs = options.ReadTimeoutMs;
+        _supervisoryAdviseParallelism = Math.Max(1, options.SupervisoryAdviseParallelism);
 
         var clientOptions = new MxGatewayClientOptions
         {
@@ -102,6 +104,122 @@ public sealed class RealMxGatewayClient : IMxGatewayClient
         return handle.ToString(CultureInfo.InvariantCulture);
     }
 
+    /// <inheritdoc />
+    public async Task<IReadOnlyList<MxSubscribeOutcome>> SubscribeBulkAsync(
+        IReadOnlyList<string> tagPaths, CancellationToken ct = default)
+    {
+        if (tagPaths.Count == 0)
+            return [];
+
+        // Plain-advise mode (a configured write user): the worker's SubscribeBulk does
+        // AddItem + Advise for every tag in ONE gateway command, collapsing the historical
+        // 2 RPCs per tag to 1 RPC per chunk.
+        if (!UseSupervisoryAdvise)
+        {
+            var replies = await _session!.SubscribeBulkAsync(_serverHandle, tagPaths, ct).ConfigureAwait(false);
+            return MapSubscribeResults(tagPaths, replies);
+        }
+
+        // Supervisory mode (WriteUserId == 0): the worker has no BULK supervisory advise
+        // (SubscribeBulk issues plain Advise), so bulk-add the items in one command and
+        // pipeline the per-item AdviseSupervisory commands with a bounded window instead
+        // of issuing them serially. Cross-repo follow-up: an additive `supervisory` flag on
+        // SubscribeBulkCommand in mxaccessgw would collapse this to one RPC as well.
+        var added = await _session!.AddItemBulkAsync(_serverHandle, tagPaths, ct).ConfigureAwait(false);
+        var outcomes = MapSubscribeResults(tagPaths, added);
+
+        var advisable = outcomes
+            .Select((o, i) => (Outcome: o, Index: i))
+            .Where(x => x.Outcome.Success)
+            .ToList();
+
+        var advised = await BulkPipeline.RunAsync(
+            advisable,
+            _supervisoryAdviseParallelism,
+            async (x, token) =>
+            {
+                var handle = int.Parse(x.Outcome.SubscriptionId!, NumberStyles.Integer, CultureInfo.InvariantCulture);
+                await EnsureSupervisoryAdvisedAsync(handle, token).ConfigureAwait(false);
+                return x.Outcome;
+            },
+            (x, ex) => x.Outcome with { Success = false, SubscriptionId = null, Error = ex.Message },
+            ct).ConfigureAwait(false);
+
+        var result = outcomes.ToArray();
+        for (var i = 0; i < advisable.Count; i++)
+            result[advisable[i].Index] = advised[i];
+
+        // A failed advise leaves an added-but-unadvised item; drop our handle mapping so a
+        // later retry re-adds cleanly rather than reusing a half-subscribed handle.
+        foreach (var failed in result.Where(r => !r.Success))
+        {
+            if (_tagToHandle.TryRemove(failed.TagPath, out var staleHandle))
+                _handleToTag.TryRemove(staleHandle, out _);
+        }
+
+        return result;
+    }
+
+    /// <summary>
+    /// Maps the gateway's per-item bulk results back onto the requested tag paths. Results
+    /// are returned in request order; the tag address on the reply is preferred when the
+    /// gateway echoes it. Successful rows populate the tag ↔ item-handle maps that the
+    /// event loop and the write path key off.
+    /// </summary>
+    private MxSubscribeOutcome[] MapSubscribeResults(
+        IReadOnlyList<string> tagPaths, IReadOnlyList<SubscribeResult> replies)
+    {
+        var outcomes = new MxSubscribeOutcome[tagPaths.Count];
+        for (var i = 0; i < tagPaths.Count; i++)
+        {
+            var tag = tagPaths[i];
+            if (i >= replies.Count)
+            {
+                outcomes[i] = new MxSubscribeOutcome(tag, false, null, "Gateway returned no result for this tag.");
+                continue;
+            }
+
+            var reply = replies[i];
+            if (!reply.WasSuccessful)
+            {
+                outcomes[i] = new MxSubscribeOutcome(tag, false, null,
+                    string.IsNullOrEmpty(reply.ErrorMessage) ? "Bulk subscribe failed." : reply.ErrorMessage);
+                continue;
+            }
+
+            _tagToHandle[tag] = reply.ItemHandle;
+            _handleToTag[reply.ItemHandle] = tag;
+            outcomes[i] = new MxSubscribeOutcome(
+                tag, true, reply.ItemHandle.ToString(CultureInfo.InvariantCulture), null);
+        }
+
+        return outcomes;
+    }
+
+    /// <inheritdoc />
+    public async Task UnsubscribeBulkAsync(IReadOnlyList<string> subscriptionIds, CancellationToken ct = default)
+    {
+        var handles = new List<int>(subscriptionIds.Count);
+        foreach (var id in subscriptionIds)
+        {
+            if (int.TryParse(id, NumberStyles.Integer, CultureInfo.InvariantCulture, out var handle))
+                handles.Add(handle);
+        }
+
+        if (handles.Count == 0)
+            return;
+
+        await _session!.UnsubscribeBulkAsync(_serverHandle, handles, ct).ConfigureAwait(false);
+
+        foreach (var handle in handles)
+        {
+            // UnsubscribeBulk unadvises AND removes the item, releasing every advice kind.
+            _supervisoryAdvised.TryRemove(handle, out _);
+            if (_handleToTag.TryRemove(handle, out var tag))
+                _tagToHandle.TryRemove(tag, out _);
+        }
+    }
+
     /// <inheritdoc />
     public async Task UnsubscribeAsync(string subscriptionId, CancellationToken ct = default)
     {

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealOpcUaClient.cs

@@ -19,7 +19,42 @@ namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
 public class RealOpcUaClient : IOpcUaClient
 {
     private ISession? _session;
-    private Subscription? _subscription;
+
+    /// <summary>
+    /// One data shard = one OPC UA Subscription holding at most
+    /// <see cref="OpcUaConnectionOptions.MaxMonitoredItemsPerSubscription"/> monitored
+    /// items. The count is tracked here rather than read back off the SDK object so
+    /// placement decisions are made from state this class owns (and mutates only under
+    /// <see cref="_applyGate"/>).
+    /// </summary>
+    private sealed class DataShard(Subscription subscription)
+    {
+        public Subscription Subscription { get; } = subscription;
+        public int ItemCount { get; set; }
+    }
+
+    // Data-item shards, in creation order. A monitored item is placed on the first shard
+    // with free capacity; when all are full a new shard is created, and a shard is
+    // deleted once it empties. Mutated ONLY under _applyGate.
+    private readonly List<DataShard> _dataShards = new();
+
+    // Alarms & Conditions event items live on their own shard: TriggerConditionRefresh
+    // needs a single subscription id to target, and QueueSize:1000 event items would
+    // otherwise consume data-shard capacity. Created lazily, torn down on disconnect.
+    private Subscription? _eventShard;
+
+    // subscription handle → owning shard's Subscription, so a removal can group ONE
+    // ApplyChanges per touched shard.
+    private readonly ConcurrentDictionary<string, Subscription> _itemShard = new();
+
+    // The OPC Foundation Subscription object is not safe under concurrent structural
+    // mutation, and nothing else orders HandleSubscribe's background task, the
+    // tag-resolution probe and an alarm subscribe against each other. Every
+    // AddItem/RemoveItem/ApplyChanges/Create/Delete — and every shard-list mutation —
+    // serializes behind this ONE per-client gate. Batching makes contention rare; a
+    // single gate (rather than per-shard) also serializes shard creation and is the
+    // simplest thing that is correct.
+    private readonly AsyncSerialGate _applyGate = new();
 
     // These maps are read from the OPC Foundation SDK's
     // internal publish threads (the MonitoredItem.Notification handler reads
@@ -146,20 +181,68 @@ public class RealOpcUaClient : IOpcUaClient
         // Store options for monitored item creation
         _options = opts;
 
-        // Create a default subscription for all monitored items
+        // Create the first data shard up front so a session is immediately usable;
-        _subscription = new Subscription(_session.DefaultSubscription)
+        // further shards are added on demand as the item budget fills.
+        await _applyGate.RunAsync(async () =>
         {
-            DisplayName = opts.SubscriptionDisplayName,
+            _dataShards.Clear();
-            Priority = opts.SubscriptionPriority,
+            _itemShard.Clear();
+            _eventShard = null;
+            await CreateDataShardAsync(cancellationToken);
+        }, cancellationToken);
+    }
+
+    /// <summary>Item budget per subscription, guarding against a non-positive configured value.</summary>
+    private int ItemsPerShard => _options.MaxMonitoredItemsPerSubscription > 0
+        ? _options.MaxMonitoredItemsPerSubscription
+        : 5000;
+
+    /// <summary>
+    /// Builds a Subscription from the current connection options. Callers must hold
+    /// <see cref="_applyGate"/>.
+    /// </summary>
+    private Subscription NewSubscription(string displayName) =>
+        new(_session!.DefaultSubscription)
+        {
+            DisplayName = displayName,
+            Priority = _options.SubscriptionPriority,
             PublishingEnabled = true,
-            PublishingInterval = opts.PublishingIntervalMs,
+            PublishingInterval = _options.PublishingIntervalMs,
-            KeepAliveCount = (uint)opts.KeepAliveCount,
+            KeepAliveCount = (uint)_options.KeepAliveCount,
-            LifetimeCount = (uint)opts.LifetimeCount,
+            LifetimeCount = (uint)_options.LifetimeCount,
-            MaxNotificationsPerPublish = (uint)opts.MaxNotificationsPerPublish
+            MaxNotificationsPerPublish = (uint)_options.MaxNotificationsPerPublish
         };
 
-        _session.AddSubscription(_subscription);
+    /// <summary>
-        await _subscription.CreateAsync(cancellationToken);
+    /// Adds and creates a new data shard. Callers must hold <see cref="_applyGate"/>.
+    /// </summary>
+    private async Task<DataShard> CreateDataShardAsync(CancellationToken cancellationToken)
+    {
+        var suffix = _dataShards.Count == 0 ? string.Empty : $"-{_dataShards.Count + 1}";
+        var subscription = NewSubscription(_options.SubscriptionDisplayName + suffix);
+        _session!.AddSubscription(subscription);
+        await subscription.CreateAsync(cancellationToken);
+        var shard = new DataShard(subscription);
+        _dataShards.Add(shard);
+        _logger.LogDebug("OPC UA data shard {Shard} created (budget {Budget} items)",
+            subscription.DisplayName, ItemsPerShard);
+        return shard;
+    }
+
+    /// <summary>
+    /// Returns the dedicated Alarms &amp; Conditions event shard, creating it on first
+    /// use. Callers must hold <see cref="_applyGate"/>.
+    /// </summary>
+    private async Task<Subscription> GetOrCreateEventShardAsync(CancellationToken cancellationToken)
+    {
+        if (_eventShard != null)
+            return _eventShard;
+
+        var subscription = NewSubscription(_options.SubscriptionDisplayName + "-events");
+        _session!.AddSubscription(subscription);
+        await subscription.CreateAsync(cancellationToken);
+        _eventShard = subscription;
+        return subscription;
     }
 
     /// <summary>
@@ -439,11 +522,19 @@ public class RealOpcUaClient : IOpcUaClient
     /// <inheritdoc />
     public async Task DisconnectAsync(CancellationToken cancellationToken = default)
     {
-        if (_subscription != null)
+        await _applyGate.RunAsync(async () =>
         {
-            await _subscription.DeleteAsync(true);
+            foreach (var shard in _dataShards)
-            _subscription = null;
+                await shard.Subscription.DeleteAsync(true);
-        }
+            _dataShards.Clear();
+            _itemShard.Clear();
+            if (_eventShard != null)
+            {
+                await _eventShard.DeleteAsync(true);
+                _eventShard = null;
+            }
+        }, cancellationToken);
+
         if (_session != null)
         {
             _session.KeepAlive -= OnSessionKeepAlive;
@@ -459,55 +550,230 @@ public class RealOpcUaClient : IOpcUaClient
         string nodeId, Action<string, object?, DateTime, uint> onValueChanged,
         CancellationToken cancellationToken = default)
     {
-        if (_subscription == null || _session == null)
+        // Batch-of-one delegation: the single-node entry point is kept (heartbeat monitor,
-            throw new InvalidOperationException("Not connected.");
+        // interactive paths) but shares the batch implementation so there is exactly one
+        // placement / ApplyChanges code path. A per-item failure throws here — preserving
+        // the caller's historical "create throws on failure" contract AND the exception
+        // TYPE, so DataConnectionActor.IsConnectionLevelFailure still classifies a
+        // resolution failure as a tag-resolution failure rather than a connection fault.
+        var (outcomes, errors) = await CreateSubscriptionsCoreAsync(
+            [nodeId], onValueChanged, cancellationToken);
 
-        var handle = Guid.NewGuid().ToString();
+        var outcome = outcomes[0];
-        var monitoredItem = new MonitoredItem(_subscription.DefaultItem)
+        if (outcome.Success)
-        {
+            return outcome.SubscriptionHandle!;
-            DisplayName = nodeId,
-            StartNodeId = OpcUaNodeReference.Resolve(nodeId, _session.NamespaceUris),
-            AttributeId = Attributes.Value,
-            SamplingInterval = _options.SamplingIntervalMs,
-            QueueSize = (uint)_options.QueueSize,
-            DiscardOldest = _options.DiscardOldest,
-            Filter = BuildDataChangeFilter(_options.Deadband)
-        };
 
-        _callbacks[handle] = onValueChanged;
+        if (errors[0] is { } captured)
-
+            System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(captured).Throw();
-        monitoredItem.Notification += (item, e) =>
+        throw new ServiceResultException(StatusCodes.BadNodeIdUnknown,
-        {
+            outcome.Error ?? $"Monitored item for '{nodeId}' could not be created.");
-            if (e.NotificationValue is MonitoredItemNotification notification)
-            {
-                var value = notification.Value?.Value;
-                var timestamp = notification.Value?.SourceTimestamp ?? DateTime.UtcNow;
-                var statusCode = notification.Value?.StatusCode.Code ?? 0;
-
-                if (_callbacks.TryGetValue(handle, out var cb))
-                {
-                    cb(nodeId, value, timestamp, statusCode);
-                }
-            }
-        };
-
-        _subscription.AddItem(monitoredItem);
-        await _subscription.ApplyChangesAsync(cancellationToken);
-
-        _monitoredItems[handle] = monitoredItem;
-        return handle;
     }
 
     /// <inheritdoc />
-    public async Task RemoveSubscriptionAsync(string subscriptionHandle, CancellationToken cancellationToken = default)
+    public async Task<IReadOnlyList<OpcUaSubscribeOutcome>> CreateSubscriptionsAsync(
+        IReadOnlyList<string> nodeIds,
+        Action<string, object?, DateTime, uint> onValueChanged,
+        CancellationToken cancellationToken = default)
     {
-        if (_subscription != null && _monitoredItems.TryGetValue(subscriptionHandle, out var item))
+        var (outcomes, _) = await CreateSubscriptionsCoreAsync(nodeIds, onValueChanged, cancellationToken);
+        return outcomes;
+    }
+
+    /// <summary>
+    /// Creates monitored items for every requested node with ONE ApplyChanges per touched
+    /// shard — the core of the batch subscribe seam. Node-id resolution runs outside the
+    /// apply lock (it is pure over the session's namespace table); placement, AddItem and
+    /// ApplyChanges run under it. Returns per-node outcomes plus the captured exception
+    /// (when any) per node, so the single-node entry point can rethrow the original.
+    /// </summary>
+    private async Task<(OpcUaSubscribeOutcome[] Outcomes, Exception?[] Errors)> CreateSubscriptionsCoreAsync(
+        IReadOnlyList<string> nodeIds,
+        Action<string, object?, DateTime, uint> onValueChanged,
+        CancellationToken cancellationToken)
+    {
+        var session = _session;
+        if (session == null)
+            throw new InvalidOperationException("Not connected.");
+
+        var outcomes = new OpcUaSubscribeOutcome[nodeIds.Count];
+        var errors = new Exception?[nodeIds.Count];
+
+        // Resolve first: a bad node id is a per-tag fault and must not abort the batch.
+        var resolved = new List<(int Index, string NodeId, NodeId Resolved)>(nodeIds.Count);
+        for (var i = 0; i < nodeIds.Count; i++)
         {
-            _subscription.RemoveItem(item);
+            try
-            await _subscription.ApplyChangesAsync(cancellationToken);
+            {
-            _monitoredItems.TryRemove(subscriptionHandle, out _);
+                resolved.Add((i, nodeIds[i], OpcUaNodeReference.Resolve(nodeIds[i], session.NamespaceUris)));
-            _callbacks.TryRemove(subscriptionHandle, out _);
+            }
+            catch (Exception ex)
+            {
+                outcomes[i] = new OpcUaSubscribeOutcome(nodeIds[i], false, null, ex.Message);
+                errors[i] = ex;
+            }
         }
+
+        if (resolved.Count == 0)
+            return (outcomes, errors);
+
+        return await _applyGate.RunAsync(async () =>
+        {
+            var created = new List<(int Index, string NodeId, string Handle, MonitoredItem Item)>(resolved.Count);
+            var touched = new List<DataShard>();
+
+            foreach (var (index, nodeId, startNode) in resolved)
+            {
+                var shard = await PlaceOnDataShardAsync(cancellationToken);
+                var handle = Guid.NewGuid().ToString();
+                var monitoredItem = new MonitoredItem(shard.Subscription.DefaultItem)
+                {
+                    DisplayName = nodeId,
+                    StartNodeId = startNode,
+                    AttributeId = Attributes.Value,
+                    SamplingInterval = _options.SamplingIntervalMs,
+                    QueueSize = (uint)_options.QueueSize,
+                    DiscardOldest = _options.DiscardOldest,
+                    Filter = BuildDataChangeFilter(_options.Deadband)
+                };
+
+                _callbacks[handle] = onValueChanged;
+                monitoredItem.Notification += (_, e) =>
+                {
+                    if (e.NotificationValue is MonitoredItemNotification notification)
+                    {
+                        var value = notification.Value?.Value;
+                        var timestamp = notification.Value?.SourceTimestamp ?? DateTime.UtcNow;
+                        var statusCode = notification.Value?.StatusCode.Code ?? 0;
+
+                        if (_callbacks.TryGetValue(handle, out var cb))
+                        {
+                            cb(nodeId, value, timestamp, statusCode);
+                        }
+                    }
+                };
+
+                shard.Subscription.AddItem(monitoredItem);
+                shard.ItemCount++;
+                _itemShard[handle] = shard.Subscription;
+                created.Add((index, nodeId, handle, monitoredItem));
+                if (!touched.Contains(shard))
+                    touched.Add(shard);
+            }
+
+            // ONE ApplyChanges per touched shard — the whole point of the batch seam.
+            foreach (var shard in touched)
+                await shard.Subscription.ApplyChangesAsync(cancellationToken);
+
+            foreach (var (index, nodeId, handle, item) in created)
+            {
+                var error = item.Status?.Error;
+                if (error != null && ServiceResult.IsBad(error))
+                {
+                    // Per-item create failure (unknown node, access denied): release the
+                    // half-created item so the shard budget stays honest, and report a
+                    // failed row. The actor treats it as an unresolved tag and retries it
+                    // on the backoff timer.
+                    ReleaseFailedItem(handle, item);
+                    outcomes[index] = new OpcUaSubscribeOutcome(nodeId, false, null, error.ToString());
+                }
+                else
+                {
+                    _monitoredItems[handle] = item;
+                    outcomes[index] = new OpcUaSubscribeOutcome(nodeId, true, handle, null);
+                }
+            }
+
+            return (outcomes, errors);
+        }, cancellationToken);
+    }
+
+    /// <summary>
+    /// Returns the shard this item belongs on per
+    /// <see cref="MonitoredItemShardPlanner"/> — first shard with free capacity, else a
+    /// freshly created one. Callers must hold <see cref="_applyGate"/>.
+    /// </summary>
+    private async Task<DataShard> PlaceOnDataShardAsync(CancellationToken cancellationToken)
+    {
+        var index = MonitoredItemShardPlanner.Plan(
+            _dataShards.Select(s => s.ItemCount).ToList(), ItemsPerShard, 1)[0];
+        return index < _dataShards.Count
+            ? _dataShards[index]
+            : await CreateDataShardAsync(cancellationToken);
+    }
+
+    /// <summary>
+    /// Drops a monitored item that failed to create from the shard it was placed on.
+    /// Callers must hold <see cref="_applyGate"/>. No ApplyChanges is issued: the server
+    /// never created the item, so removing it locally keeps the budget honest.
+    /// </summary>
+    private void ReleaseFailedItem(string handle, MonitoredItem item)
+    {
+        _callbacks.TryRemove(handle, out _);
+        if (!_itemShard.TryRemove(handle, out var subscription))
+            return;
+
+        subscription.RemoveItem(item);
+        var shard = _dataShards.FirstOrDefault(s => ReferenceEquals(s.Subscription, subscription));
+        if (shard is { ItemCount: > 0 })
+            shard.ItemCount--;
+    }
+
+    /// <inheritdoc />
+    public Task RemoveSubscriptionAsync(string subscriptionHandle, CancellationToken cancellationToken = default)
+        => RemoveSubscriptionsAsync([subscriptionHandle], cancellationToken);
+
+    /// <inheritdoc />
+    public async Task RemoveSubscriptionsAsync(
+        IReadOnlyList<string> subscriptionHandles, CancellationToken cancellationToken = default)
+    {
+        if (subscriptionHandles.Count == 0)
+            return;
+
+        await _applyGate.RunAsync(async () =>
+        {
+            var touched = new List<DataShard>();
+            foreach (var handle in subscriptionHandles)
+            {
+                if (!_monitoredItems.TryGetValue(handle, out var item))
+                    continue;
+                if (!_itemShard.TryGetValue(handle, out var subscription))
+                    continue;
+
+                subscription.RemoveItem(item);
+                _monitoredItems.TryRemove(handle, out _);
+                _callbacks.TryRemove(handle, out _);
+                _itemShard.TryRemove(handle, out _);
+
+                var shard = _dataShards.FirstOrDefault(s => ReferenceEquals(s.Subscription, subscription));
+                if (shard == null)
+                    continue;
+                if (shard.ItemCount > 0)
+                    shard.ItemCount--;
+                if (!touched.Contains(shard))
+                    touched.Add(shard);
+            }
+
+            // ONE ApplyChanges per touched shard, then drop the shards that emptied.
+            foreach (var shard in touched)
+            {
+                await shard.Subscription.ApplyChangesAsync(cancellationToken);
+                if (shard.ItemCount != 0)
+                    continue;
+
+                _dataShards.Remove(shard);
+                try
+                {
+                    await shard.Subscription.DeleteAsync(true);
+                }
+                catch (Exception ex)
+                {
+                    // Best-effort: an empty shard the server already dropped must not fail
+                    // the unsubscribe path.
+                    _logger.LogDebug(ex, "Deleting emptied OPC UA data shard {Shard} failed (ignored).",
+                        shard.Subscription.DisplayName);
+                }
+            }
+        }, cancellationToken);
     }
 
     // ── Native alarm (Alarms & Conditions) subscription ──
@@ -523,7 +789,7 @@ public class RealOpcUaClient : IOpcUaClient
         string? sourceNodeId, string? conditionFilter,
         Action<NativeAlarmTransition> onTransition, CancellationToken cancellationToken = default)
     {
-        if (_subscription == null || _session == null)
+        if (_session == null)
             throw new InvalidOperationException("Not connected.");
 
         var handle = Guid.NewGuid().ToString();
@@ -533,7 +799,14 @@ public class RealOpcUaClient : IOpcUaClient
         var startNode = string.IsNullOrEmpty(sourceNodeId)
             ? ObjectIds.Server
             : OpcUaNodeReference.Resolve(sourceNodeId, _session.NamespaceUris);
-        var item = new MonitoredItem(_subscription.DefaultItem)
+
+        // A&C event items live on their own shard (see _eventShard): ConditionRefresh
+        // needs one subscription id to target, and QueueSize:1000 event items must not
+        // occupy data-shard capacity.
+        var eventShard = await _applyGate.RunAsync(
+            () => GetOrCreateEventShardAsync(cancellationToken), cancellationToken);
+
+        var item = new MonitoredItem(eventShard.DefaultItem)
         {
             DisplayName = $"alarm:{sourceNodeId ?? "Server"}",
             StartNodeId = startNode,
@@ -554,22 +827,31 @@ public class RealOpcUaClient : IOpcUaClient
                 HandleAlarmEvent(handle, sourceNodeId, efl, onTransition);
         };
 
-        _subscription.AddItem(item);
+        await _applyGate.RunAsync(async () =>
-        await _subscription.ApplyChangesAsync(cancellationToken);
+        {
+            eventShard.AddItem(item);
+            await eventShard.ApplyChangesAsync(cancellationToken);
+        }, cancellationToken);
         _alarmItems[handle] = item;
 
         // Replay currently-active conditions as a Snapshot…SnapshotComplete sequence.
-        await TriggerConditionRefreshAsync(handle, cancellationToken);
+        await TriggerConditionRefreshAsync(handle, eventShard, cancellationToken);
         return handle;
     }
 
     /// <inheritdoc />
     public async Task RemoveAlarmSubscriptionAsync(string subscriptionHandle, CancellationToken cancellationToken = default)
     {
-        if (_subscription != null && _alarmItems.TryRemove(subscriptionHandle, out var item))
+        if (_alarmItems.TryRemove(subscriptionHandle, out var item))
         {
-            _subscription.RemoveItem(item);
+            await _applyGate.RunAsync(async () =>
-            await _subscription.ApplyChangesAsync(cancellationToken);
+            {
+                if (_eventShard != null)
+                {
+                    _eventShard.RemoveItem(item);
+                    await _eventShard.ApplyChangesAsync(cancellationToken);
+                }
+            }, cancellationToken);
         }
         _alarmInRefresh.TryRemove(subscriptionHandle, out _);
         _alarmLastState.TryRemove(subscriptionHandle, out _);
@@ -775,15 +1057,17 @@ public class RealOpcUaClient : IOpcUaClient
         };
     }
 
-    private async Task TriggerConditionRefreshAsync(string handle, CancellationToken cancellationToken)
+    private async Task TriggerConditionRefreshAsync(
+        string handle, Subscription eventShard, CancellationToken cancellationToken)
     {
         try
         {
             // ConditionRefresh replays active conditions; RefreshStart/End events
-            // bracket the replay so HandleAlarmEvent can mark them Snapshot.
+            // bracket the replay so HandleAlarmEvent can mark them Snapshot. It targets
+            // the EVENT shard's subscription id — the shard every A&C item is pinned to.
             await _session!.CallAsync(
                 ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh,
-                cancellationToken, _subscription!.Id);
+                cancellationToken, eventShard.Id);
         }
         catch (Exception ex)
         {
@@ -906,42 +1190,143 @@ public class RealOpcUaClient : IOpcUaClient
         new Commons.Types.Alarms.AlarmConditionState(false, true, null, AlarmShelveState.Unshelved, false, 0),
         "", "", "", "", "", null, DateTimeOffset.UtcNow, "", "");
 
+    /// <summary>
+    /// Nodes per OPC UA Read/Write service call. Stays under the
+    /// <c>MaxNodesPerRead</c> / <c>MaxNodesPerWrite</c> operation limits typical servers
+    /// advertise, so a 37,500-tag re-seed never builds one oversized request.
+    /// </summary>
+    private const int MaxNodesPerServiceCall = 1000;
+
     /// <inheritdoc />
     public async Task<(object? Value, DateTime SourceTimestamp, uint StatusCode)> ReadValueAsync(
         string nodeId, CancellationToken cancellationToken = default)
     {
-        if (_session == null) throw new InvalidOperationException("Not connected.");
+        // Batch-of-one delegation; a resolution failure still surfaces as the thrown
+        // exception the single-node callers expect.
+        var outcomes = await ReadValuesAsync([nodeId], cancellationToken);
+        var outcome = outcomes[0];
+        if (outcome.Error != null)
+            throw new ServiceResultException(StatusCodes.BadNodeIdUnknown, outcome.Error);
+        return (outcome.Value, outcome.SourceTimestamp, outcome.StatusCode);
+    }
 
-        var readValue = new ReadValueId
+    /// <inheritdoc />
+    public async Task<IReadOnlyList<OpcUaReadOutcome>> ReadValuesAsync(
+        IReadOnlyList<string> nodeIds, CancellationToken cancellationToken = default)
+    {
+        var session = _session;
+        if (session == null) throw new InvalidOperationException("Not connected.");
+
+        var outcomes = new OpcUaReadOutcome[nodeIds.Count];
+
+        // Resolve first — an unresolvable node id is a per-node fault, not a batch abort.
+        var resolvable = new List<(int Index, string NodeId, ReadValueId Read)>(nodeIds.Count);
+        for (var i = 0; i < nodeIds.Count; i++)
         {
-            NodeId = OpcUaNodeReference.Resolve(nodeId, _session.NamespaceUris),
+            try
-            AttributeId = Attributes.Value
+            {
-        };
+                resolvable.Add((i, nodeIds[i], new ReadValueId
+                {
+                    NodeId = OpcUaNodeReference.Resolve(nodeIds[i], session.NamespaceUris),
+                    AttributeId = Attributes.Value
+                }));
+            }
+            catch (Exception ex)
+            {
+                outcomes[i] = new OpcUaReadOutcome(nodeIds[i], null, DateTime.UtcNow, StatusCodes.BadNodeIdUnknown, ex.Message);
+            }
+        }
 
-        var response = await _session.ReadAsync(
+        for (var offset = 0; offset < resolvable.Count; offset += MaxNodesPerServiceCall)
-            null, 0, MapTimestampsToReturn(_options.TimestampsToReturn),
+        {
-            new ReadValueIdCollection { readValue }, cancellationToken);
+            var chunk = resolvable.GetRange(offset, Math.Min(MaxNodesPerServiceCall, resolvable.Count - offset));
+            var collection = new ReadValueIdCollection(chunk.Select(c => c.Read));
+            var response = await session.ReadAsync(
+                null, 0, MapTimestampsToReturn(_options.TimestampsToReturn), collection, cancellationToken);
 
-        var result = response.Results[0];
+            for (var i = 0; i < chunk.Count; i++)
-        return (result.Value, result.SourceTimestamp, result.StatusCode.Code);
+            {
+                var (index, nodeId, _) = chunk[i];
+                if (response.Results is { Count: > 0 } results && i < results.Count)
+                {
+                    var result = results[i];
+                    outcomes[index] = new OpcUaReadOutcome(
+                        nodeId, result.Value, result.SourceTimestamp, result.StatusCode.Code, null);
+                }
+                else
+                {
+                    // Non-conformant server: fewer results than requested nodes.
+                    outcomes[index] = new OpcUaReadOutcome(
+                        nodeId, null, DateTime.UtcNow, StatusCodes.BadUnexpectedError,
+                        "OPC UA read returned no result for this node.");
+                }
+            }
+        }
+
+        return outcomes;
     }
 
     /// <inheritdoc />
     public async Task<uint> WriteValueAsync(string nodeId, object? value, CancellationToken cancellationToken = default)
     {
-        if (_session == null) throw new InvalidOperationException("Not connected.");
+        // Batch-of-one delegation, mirroring ReadValueAsync.
+        var outcomes = await WriteValuesAsync([(nodeId, value)], cancellationToken);
+        var outcome = outcomes[0];
+        if (outcome.Error != null)
+            throw new ServiceResultException(StatusCodes.BadNodeIdUnknown, outcome.Error);
+        return outcome.StatusCode;
+    }
 
-        var writeValue = new WriteValue
+    /// <inheritdoc />
+    public async Task<IReadOnlyList<OpcUaWriteOutcome>> WriteValuesAsync(
+        IReadOnlyList<(string NodeId, object? Value)> values, CancellationToken cancellationToken = default)
+    {
+        var session = _session;
+        if (session == null) throw new InvalidOperationException("Not connected.");
+
+        var outcomes = new OpcUaWriteOutcome[values.Count];
+
+        var resolvable = new List<(int Index, string NodeId, WriteValue Write)>(values.Count);
+        for (var i = 0; i < values.Count; i++)
         {
-            NodeId = OpcUaNodeReference.Resolve(nodeId, _session.NamespaceUris),
+            try
-            AttributeId = Attributes.Value,
+            {
-            Value = new DataValue(new Variant(value))
+                resolvable.Add((i, values[i].NodeId, new WriteValue
-        };
+                {
+                    NodeId = OpcUaNodeReference.Resolve(values[i].NodeId, session.NamespaceUris),
+                    AttributeId = Attributes.Value,
+                    Value = new DataValue(new Variant(values[i].Value))
+                }));
+            }
+            catch (Exception ex)
+            {
+                outcomes[i] = new OpcUaWriteOutcome(values[i].NodeId, StatusCodes.BadNodeIdUnknown, ex.Message);
+            }
+        }
 
-        var response = await _session.WriteAsync(
+        for (var offset = 0; offset < resolvable.Count; offset += MaxNodesPerServiceCall)
-            null, new WriteValueCollection { writeValue }, cancellationToken);
+        {
+            var chunk = resolvable.GetRange(offset, Math.Min(MaxNodesPerServiceCall, resolvable.Count - offset));
+            var collection = new WriteValueCollection(chunk.Select(c => c.Write));
+            var response = await session.WriteAsync(null, collection, cancellationToken);
 
-        return response.Results[0].Code;
+            for (var i = 0; i < chunk.Count; i++)
+            {
+                var (index, nodeId, _) = chunk[i];
+                if (response.Results is { Count: > 0 } results && i < results.Count)
+                {
+                    outcomes[index] = new OpcUaWriteOutcome(nodeId, results[i].Code, null);
+                }
+                else
+                {
+                    outcomes[index] = new OpcUaWriteOutcome(
+                        nodeId, StatusCodes.BadUnexpectedError,
+                        "OPC UA write returned no result for this node.");
+                }
+            }
+        }
+
+        return outcomes;
     }
 
     /// <summary>

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/DataConnectionFactory.cs

@@ -37,11 +37,16 @@ public class DataConnectionFactory : IDataConnectionFactory
     public DataConnectionFactory(
         ILoggerFactory loggerFactory,
         IOptions<OpcUaGlobalOptions> opcUaGlobalOptions,
-        ISecretResolver? secretResolver = null)
+        ISecretResolver? secretResolver = null,
+        IOptions<DataConnectionOptions>? dataConnectionOptions = null)
     {
         _loggerFactory = loggerFactory;
         _secretResolver = secretResolver;
         var globalOptions = opcUaGlobalOptions.Value;
+        // Only the MxGateway supervisory-advise window is adapter-level today; the rest of
+        // DataConnectionOptions is consumed by the actor, which is constructed elsewhere.
+        var supervisoryAdviseParallelism =
+            (dataConnectionOptions?.Value ?? new DataConnectionOptions()).MxSupervisoryAdviseParallelism;
 
         // Register built-in protocols.
         // Pass the ILoggerFactory into RealOpcUaClientFactory so
@@ -57,7 +62,8 @@ public class DataConnectionFactory : IDataConnectionFactory
         RegisterAdapter("MxGateway", details => new MxGatewayDataConnection(
             new RealMxGatewayClientFactory(_loggerFactory),
             _loggerFactory.CreateLogger<MxGatewayDataConnection>(),
-            _secretResolver));
+            _secretResolver,
+            supervisoryAdviseParallelism));
     }
 
     /// <summary>

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/DataConnectionOptions.cs

@@ -8,9 +8,70 @@ public class DataConnectionOptions
     /// <summary>Fixed interval between reconnect attempts after disconnect.</summary>
     public TimeSpan ReconnectInterval { get; set; } = TimeSpan.FromSeconds(5);
 
-    /// <summary>Interval for retrying failed tag path resolution.</summary>
+    /// <summary>
+    /// Floor interval for retrying failed tag path resolution. The retry backs off
+    /// exponentially (doubling per fully-failed round) up to
+    /// <see cref="TagResolutionRetryMaxInterval"/>, and resets to this floor as soon as
+    /// any tag resolves or the connection reconnects.
+    /// </summary>
     public TimeSpan TagResolutionRetryInterval { get; set; } = TimeSpan.FromSeconds(10);
 
+    /// <summary>
+    /// Ceiling for the exponential tag-resolution retry backoff. A dead device with
+    /// thousands of unresolved tags would otherwise probe forever at full width every
+    /// <see cref="TagResolutionRetryInterval"/>.
+    /// </summary>
+    public TimeSpan TagResolutionRetryMaxInterval { get; set; } = TimeSpan.FromMinutes(5);
+
+    /// <summary>
+    /// Number of tags per adapter subscribe round trip on the batch subscribe path
+    /// (initial subscribe, reconnect re-subscribe, tag-resolution probes).
+    /// </summary>
+    public int SubscribeBatchSize { get; set; } = 500;
+
+    /// <summary>
+    /// Delay inserted BETWEEN reconnect re-subscribe chunks so a site-wide reconnect
+    /// paces the device instead of issuing one task per tag in a tight loop. Not applied
+    /// on the instance-driven subscribe path — the Deployment Manager already staggers
+    /// instance startup there, and double-staggering would only slow failover.
+    /// </summary>
+    public TimeSpan SubscribeBatchDelay { get; set; } = TimeSpan.FromMilliseconds(50);
+
+    /// <summary>
+    /// Number of tags per seed-read chunk. Moderate on purpose: chunking plus the
+    /// per-chunk <see cref="SeedReadTimeout"/> answers the "some gateways time out on a
+    /// large batch" caveat that originally motivated per-tag seed reads.
+    /// </summary>
+    public int SeedReadBatchSize { get; set; } = 250;
+
+    /// <summary>Maximum seed-read chunks in flight concurrently.</summary>
+    public int SeedReadMaxParallelism { get; set; } = 4;
+
+    /// <summary>
+    /// Wall-clock deadline for the WHOLE seed (all chunks and all retry rounds). On
+    /// expiry the remaining tags are logged and stay Uncertain until a change
+    /// notification arrives — the same outcome as exhausting
+    /// <see cref="SeedReadMaxAttempts"/>.
+    /// </summary>
+    public TimeSpan SeedOverallTimeout { get; set; } = TimeSpan.FromSeconds(120);
+
+    /// <summary>
+    /// Coalescing window for pushing tag-quality counters to the health collector.
+    /// Counter math stays per message; only the collector push is deferred, and only
+    /// when a genuine quality transition occurred. Health reports poll at 30s, so a
+    /// short window loses nothing. Disconnect / unsubscribe / reconnect-reset flushes
+    /// stay synchronous.
+    /// </summary>
+    public TimeSpan QualityFlushInterval { get; set; } = TimeSpan.FromSeconds(1);
+
+    /// <summary>
+    /// Maximum in-flight supervisory advise commands when the MxGateway adapter has no
+    /// write-user context. The gateway worker has no BULK supervisory advise, so the
+    /// client pipelines per-item advises after one bulk AddItem instead of issuing them
+    /// serially.
+    /// </summary>
+    public int MxSupervisoryAdviseParallelism { get; set; } = 16;
+
     /// <summary>Timeout for synchronous write operations to devices.</summary>
     public TimeSpan WriteTimeout { get; set; } = TimeSpan.FromSeconds(30);
 

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/DataConnectionOptionsValidator.cs

@@ -35,5 +35,30 @@ public sealed class DataConnectionOptionsValidator : OptionsValidatorBase<DataCo
 
         builder.RequireThat(options.SeedReadMaxAttempts > 0,
             $"ScadaBridge:DataConnection:SeedReadMaxAttempts must be positive (was {options.SeedReadMaxAttempts}).");
+
+        builder.RequireThat(options.TagResolutionRetryMaxInterval >= options.TagResolutionRetryInterval,
+            "ScadaBridge:DataConnection:TagResolutionRetryMaxInterval must be at least " +
+            $"TagResolutionRetryInterval (was {options.TagResolutionRetryMaxInterval} vs {options.TagResolutionRetryInterval}).");
+
+        builder.RequireThat(options.SubscribeBatchSize > 0,
+            $"ScadaBridge:DataConnection:SubscribeBatchSize must be positive (was {options.SubscribeBatchSize}).");
+
+        builder.RequireThat(options.SubscribeBatchDelay >= TimeSpan.Zero,
+            $"ScadaBridge:DataConnection:SubscribeBatchDelay must not be negative (was {options.SubscribeBatchDelay}).");
+
+        builder.RequireThat(options.SeedReadBatchSize > 0,
+            $"ScadaBridge:DataConnection:SeedReadBatchSize must be positive (was {options.SeedReadBatchSize}).");
+
+        builder.RequireThat(options.SeedReadMaxParallelism > 0,
+            $"ScadaBridge:DataConnection:SeedReadMaxParallelism must be positive (was {options.SeedReadMaxParallelism}).");
+
+        builder.RequireThat(options.SeedOverallTimeout > TimeSpan.Zero,
+            $"ScadaBridge:DataConnection:SeedOverallTimeout must be a positive duration (was {options.SeedOverallTimeout}).");
+
+        builder.RequireThat(options.QualityFlushInterval > TimeSpan.Zero,
+            $"ScadaBridge:DataConnection:QualityFlushInterval must be a positive duration (was {options.QualityFlushInterval}).");
+
+        builder.RequireThat(options.MxSupervisoryAdviseParallelism > 0,
+            $"ScadaBridge:DataConnection:MxSupervisoryAdviseParallelism must be positive (was {options.MxSupervisoryAdviseParallelism}).");
     }
 }

tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionActorBatchTests.cs

@@ -0,0 +1,291 @@
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using NSubstitute;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
+using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
+
+namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;
+
+/// <summary>
+/// WP2.1b — the DataConnectionActor against a batch-capable adapter: chunked subscribe,
+/// chunked/bounded re-subscribe, the seed deadline, batched tag-resolution probes, the
+/// partial-failure surface and the coalesced quality push. The existing suite covers the
+/// same actor against a NON batch-capable adapter, which keeps the per-tag fallback pinned.
+/// </summary>
+public class DataConnectionActorBatchTests : TestKit
+{
+    private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
+    private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();
+
+    private static DataConnectionOptions Options() => new()
+    {
+        ReconnectInterval = TimeSpan.FromMilliseconds(100),
+        TagResolutionRetryInterval = TimeSpan.FromMilliseconds(150),
+        TagResolutionRetryMaxInterval = TimeSpan.FromMilliseconds(600),
+        WriteTimeout = TimeSpan.FromSeconds(5),
+        SeedReadMaxAttempts = 1,
+        SeedReadRetryDelay = TimeSpan.FromMilliseconds(10),
+        SubscribeBatchSize = 5,
+        SubscribeBatchDelay = TimeSpan.FromMilliseconds(50),
+        SeedReadBatchSize = 4,
+        SeedReadMaxParallelism = 2,
+        SeedOverallTimeout = TimeSpan.FromSeconds(30),
+        QualityFlushInterval = TimeSpan.FromMilliseconds(200)
+    };
+
+    private IActorRef CreateActor(FakeBatchDataConnection adapter, DataConnectionOptions options, string name) =>
+        Sys.ActorOf(Props.Create(() => new DataConnectionActor(
+            name, adapter, options, _health, _factory, "OpcUa")), name);
+
+    private static string[] Tags(int count) => Enumerable.Range(1, count).Select(i => $"tag{i}").ToArray();
+
+    [Fact]
+    public void Subscribe_IssuesOneAdapterRoundTripPerChunk_NotPerTag()
+    {
+        // 12 tags at SubscribeBatchSize 5 → 3 batch calls (5/5/2) and ZERO per-tag
+        // subscribes. This is finding #3: the old path was one adapter call (and one OPC UA
+        // ApplyChanges) per tag.
+        var adapter = new FakeBatchDataConnection();
+        var actor = CreateActor(adapter, Options(), "batch-chunking");
+
+        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-chunking", Tags(12), DateTimeOffset.UtcNow));
+        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
+
+        var batches = adapter.SubscribeBatches.ToList();
+        Assert.Equal(3, batches.Count);
+        Assert.Equal([5, 5, 2], batches.Select(b => b.Count));
+        Assert.Equal(0, Volatile.Read(ref adapter.SingleSubscribeCalls));
+    }
+
+    [Fact]
+    public void Seed_UsesChunkedBulkReads_NotPerTagReads()
+    {
+        // Seeding 6 tags at SeedReadBatchSize 4 → 2 bulk reads, no single-tag reads.
+        var adapter = new FakeBatchDataConnection();
+        var actor = CreateActor(adapter, Options(), "batch-seed");
+
+        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-seed", Tags(6), DateTimeOffset.UtcNow));
+        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
+
+        var reads = adapter.ReadBatches.ToList();
+        Assert.Equal(2, reads.Count);
+        Assert.Equal(6, reads.Sum(r => r.Count));
+        Assert.Equal(0, Volatile.Read(ref adapter.SingleReadCalls));
+        // The seeded value still reaches the instance actor after registration.
+        ExpectMsg<TagValueUpdate>(u => u.Quality == QualityCode.Good, TimeSpan.FromSeconds(5));
+    }
+
+    [Fact]
+    public void Seed_HangingReads_ReturnAtTheOverallDeadline_AndTheSubscribeIsStillAcked()
+    {
+        // A device whose reads never answer must not hold SubscribeCompleted (and therefore
+        // the instance's ack) open: the whole seed is bounded by SeedOverallTimeout.
+        var options = Options();
+        options.SeedOverallTimeout = TimeSpan.FromMilliseconds(600);
+        options.SeedReadTimeout = TimeSpan.FromSeconds(30); // deliberately longer than the deadline
+        options.SeedReadMaxAttempts = 3;
+
+        var adapter = new FakeBatchDataConnection { HangReads = true };
+        var actor = CreateActor(adapter, options, "batch-seed-deadline");
+
+        var started = DateTimeOffset.UtcNow;
+        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-seed-deadline", Tags(3), DateTimeOffset.UtcNow));
+        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(10));
+        var elapsed = DateTimeOffset.UtcNow - started;
+
+        // Bounded by the overall deadline, not by SeedReadTimeout (30s) or the retry budget.
+        Assert.True(elapsed < TimeSpan.FromSeconds(5), $"seed took {elapsed.TotalSeconds:F1}s");
+        // No value was seeded — the tags stay Uncertain until a change notification.
+        ExpectNoMsg(TimeSpan.FromMilliseconds(200));
+    }
+
+    [Fact]
+    public void PartialFailure_PerTagRow_MarksTagBad_ButTheRequestStillSucceeds()
+    {
+        // A per-tag failure row is a tag-resolution problem: Bad quality is pushed for that
+        // tag and the subscribe is still acked Success (the surviving tags are live).
+        var adapter = new FakeBatchDataConnection();
+        adapter.FailingTags.Add("tag2");
+        var actor = CreateActor(adapter, Options(), "batch-partial");
+
+        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-partial", Tags(3), DateTimeOffset.UtcNow));
+
+        ExpectMsg<TagValueUpdate>(u => u.TagPath == "tag2" && u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
+        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
+    }
+
+    [Fact]
+    public void ThrownBatch_AtConnectionLevel_FailsTheRequestAndDrivesReconnect()
+    {
+        // A THROWN batch means the whole chunk failed at connection level: the response must
+        // say so, and the actor must enter Reconnecting (which pushes bad quality for the
+        // connection).
+        var adapter = new FakeBatchDataConnection
+        {
+            BatchSubscribeThrows = () => new InvalidOperationException("client is not connected")
+        };
+        var actor = CreateActor(adapter, Options(), "batch-connection-fault");
+
+        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-connection-fault", Tags(3), DateTimeOffset.UtcNow));
+
+        ExpectMsg<SubscribeTagsResponse>(m => !m.Success && m.ErrorMessage != null, TimeSpan.FromSeconds(5));
+        ExpectMsg<ConnectionQualityChanged>(q => q.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
+    }
+
+    [Fact]
+    public void ResolutionProbes_AreBatched_NotOnePerTag()
+    {
+        // Every unresolved tag is probed in ONE batch call per chunk. Pre-fix the retry tick
+        // fired one SubscribeAsync task per unresolved tag, every tick, forever.
+        var options = Options();
+        options.SubscribeBatchSize = 10;
+        var adapter = new FakeBatchDataConnection();
+        foreach (var tag in Tags(6))
+            adapter.FailingTags.Add(tag);
+
+        var actor = CreateActor(adapter, options, "batch-probe");
+        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-probe", Tags(6), DateTimeOffset.UtcNow));
+
+        for (var i = 0; i < 6; i++)
+            ExpectMsg<TagValueUpdate>(u => u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
+        ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));
+
+        var initialCalls = adapter.SubscribeBatches.Count;
+        AwaitCondition(() => adapter.SubscribeBatches.Count > initialCalls, TimeSpan.FromSeconds(5));
+
+        // The probe round carried all six unresolved tags in a single call.
+        var probe = adapter.SubscribeBatches.Skip(initialCalls).First();
+        Assert.Equal(6, probe.Count);
+        Assert.Equal(0, Volatile.Read(ref adapter.SingleSubscribeCalls));
+    }
+
+    [Fact]
+    public void ResolutionProbes_BackOff_SoADeadDeviceIsNotProbedAtFullRate()
+    {
+        // Floor 150ms, ceiling 600ms: a fixed-interval retry would fire ~13 rounds in 2s;
+        // the backoff sequence (150, 300, 600, 600 …) fires far fewer.
+        var options = Options();
+        options.SubscribeBatchSize = 10;
+        var adapter = new FakeBatchDataConnection();
+        adapter.FailingTags.Add("tag1");
+
+        var actor = CreateActor(adapter, options, "batch-backoff");
+        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-backoff", ["tag1"], DateTimeOffset.UtcNow));
+        ExpectMsg<TagValueUpdate>(u => u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
+        ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));
+
+        var initialCalls = adapter.SubscribeBatches.Count;
+        Thread.Sleep(2000);
+        var rounds = adapter.SubscribeBatches.Count - initialCalls;
+
+        Assert.InRange(rounds, 2, 8);
+    }
+
+    [Fact]
+    public void QualityCounters_ArePushedOnTransitionOnly_AndCoalesced()
+    {
+        // Repeated values at UNCHANGED quality move no counter and must produce no collector
+        // push at all; a genuine transition produces exactly one push per coalescing window.
+        var adapter = new FakeBatchDataConnection();
+        var actor = CreateActor(adapter, Options(), "batch-quality");
+
+        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-quality", ["tag1"], DateTimeOffset.UtcNow));
+        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
+        AwaitCondition(() => adapter.ValueCallback != null, TimeSpan.FromSeconds(5));
+
+        // Let the seed's own transition flush, then start counting.
+        Thread.Sleep(400);
+        _health.ClearReceivedCalls();
+
+        for (var i = 0; i < 20; i++)
+            adapter.ValueCallback!("tag1", new TagValue(i, QualityCode.Good, DateTimeOffset.UtcNow));
+
+        Thread.Sleep(400);
+        Assert.DoesNotContain(_health.ReceivedCalls(), c => c.GetMethodInfo().Name == "UpdateTagQuality");
+
+        // One genuine transition → exactly one coalesced push.
+        adapter.ValueCallback!("tag1", new TagValue(1, QualityCode.Bad, DateTimeOffset.UtcNow));
+        AwaitCondition(
+            () => _health.ReceivedCalls().Count(c => c.GetMethodInfo().Name == "UpdateTagQuality") == 1,
+            TimeSpan.FromSeconds(3));
+        Thread.Sleep(300);
+        Assert.Single(_health.ReceivedCalls(), c => c.GetMethodInfo().Name == "UpdateTagQuality");
+    }
+
+    [Fact]
+    public void QualityCounters_FlushImmediatelyOnDisconnect()
+    {
+        // "Immediate bad quality on disconnect" must never wait for the coalescing timer.
+        var options = Options();
+        options.QualityFlushInterval = TimeSpan.FromSeconds(30); // would mask a deferred push
+        var adapter = new FakeBatchDataConnection();
+        var actor = CreateActor(adapter, options, "batch-quality-disconnect");
+
+        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-quality-disconnect", ["tag1"], DateTimeOffset.UtcNow));
+        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
+        ExpectMsg<TagValueUpdate>(TimeSpan.FromSeconds(5)); // the seed
+        _health.ClearReceivedCalls();
+
+        adapter.RaiseDisconnected();
+
+        ExpectMsg<ConnectionQualityChanged>(q => q.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
+        AwaitCondition(
+            () => _health.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "UpdateTagQuality"),
+            TimeSpan.FromSeconds(3));
+    }
+
+    [Fact]
+    public void Unsubscribe_ReleasesEveryHandleInOneRoundTrip()
+    {
+        var adapter = new FakeBatchDataConnection();
+        var actor = CreateActor(adapter, Options(), "batch-unsubscribe");
+
+        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-unsubscribe", Tags(6), DateTimeOffset.UtcNow));
+        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
+
+        actor.Tell(new UnsubscribeTagsRequest("c2", "inst1", "batch-unsubscribe", DateTimeOffset.UtcNow));
+
+        AwaitCondition(() => adapter.UnsubscribeBatches.Count == 1, TimeSpan.FromSeconds(5));
+        Assert.True(adapter.UnsubscribeBatches.TryDequeue(out var released));
+        Assert.Equal(6, released!.Count);
+    }
+
+    [Fact]
+    public void Reconnect_ReSubscribesInPacedChunks_AndRestoresTheResolvedCount()
+    {
+        // The reconnect re-subscribe is chunked and paced (SubscribeBatchDelay between
+        // chunks) rather than firing one task per tag, and the per-tag results repopulate
+        // _subscriptionIds so a later unsubscribe still releases every adapter handle.
+        var options = Options();
+        options.SubscribeBatchSize = 4;
+        options.SubscribeBatchDelay = TimeSpan.FromMilliseconds(120);
+
+        var adapter = new FakeBatchDataConnection();
+        var actor = CreateActor(adapter, options, "batch-resubscribe");
+
+        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-resubscribe", Tags(8), DateTimeOffset.UtcNow));
+        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
+
+        var beforeReconnect = adapter.SubscribeBatches.Count;
+        adapter.RaiseDisconnected();
+
+        // Two re-subscribe chunks of 4.
+        AwaitCondition(() => adapter.SubscribeBatches.Count >= beforeReconnect + 2, TimeSpan.FromSeconds(10));
+        var reconnectChunks = adapter.SubscribeBatches.Skip(beforeReconnect).Take(2).ToList();
+        Assert.Equal([4, 4], reconnectChunks.Select(c => c.Count));
+
+        // Paced: the second chunk is not issued back-to-back with the first.
+        var times = adapter.SubscribeBatchTimes.Skip(beforeReconnect).Take(2).ToList();
+        Assert.True(times[1] - times[0] >= TimeSpan.FromMilliseconds(100),
+            $"chunks were {(times[1] - times[0]).TotalMilliseconds:F0}ms apart");
+
+        // Every tag is registered again, so unsubscribe releases all 8 handles.
+        actor.Tell(new UnsubscribeTagsRequest("c2", "inst1", "batch-resubscribe", DateTimeOffset.UtcNow));
+        AwaitCondition(
+            () => adapter.UnsubscribeBatches.Any(b => b.Count == 8),
+            TimeSpan.FromSeconds(10));
+    }
+}

tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/FakeBatchDataConnection.cs

@@ -0,0 +1,161 @@
+using System.Collections.Concurrent;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+
+namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;
+
+/// <summary>
+/// In-memory batch-capable <see cref="IDataConnection"/> for the WP2.1b seam tests. It
+/// records every batch call (so "one round trip per chunk, not per tag" is assertable),
+/// can fail individual tags or a whole batch, and can hang its bulk read so the seed
+/// deadline is observable.
+/// </summary>
+public sealed class FakeBatchDataConnection
+    : IDataConnection, IBatchSubscribableConnection, IAlarmSubscribableConnection
+{
+    private int _nextId;
+
+    /// <summary>Tag lists handed to <see cref="SubscribeBatchAsync"/>, one entry per call.</summary>
+    public readonly ConcurrentQueue<IReadOnlyList<string>> SubscribeBatches = new();
+    /// <summary>Id lists handed to <see cref="UnsubscribeBatchAsync"/>, one entry per call.</summary>
+    public readonly ConcurrentQueue<IReadOnlyList<string>> UnsubscribeBatches = new();
+    /// <summary>Tag lists handed to <see cref="ReadBatchAsync"/>, one entry per call.</summary>
+    public readonly ConcurrentQueue<IReadOnlyList<string>> ReadBatches = new();
+    /// <summary>Count of SINGLE-tag subscribe calls; must stay 0 on a batch-capable adapter.</summary>
+    public int SingleSubscribeCalls;
+    /// <summary>Count of SINGLE-tag read calls; must stay 0 on a batch-capable adapter.</summary>
+    public int SingleReadCalls;
+    /// <summary>Wall-clock instant of each <see cref="SubscribeBatchAsync"/> call.</summary>
+    public readonly ConcurrentQueue<DateTimeOffset> SubscribeBatchTimes = new();
+
+    /// <summary>Tags reported as failed rows (per-tag resolution failure).</summary>
+    public readonly HashSet<string> FailingTags = new(StringComparer.Ordinal);
+    /// <summary>When set, every batch subscribe throws this — a batch-level fault.</summary>
+    public Func<Exception>? BatchSubscribeThrows;
+    /// <summary>When true, bulk reads never return until the caller's token cancels.</summary>
+    public bool HangReads;
+    /// <summary>Value returned for every readable tag.</summary>
+    public object? SeedValue = 42;
+
+    /// <summary>Callback the last batch subscribe registered; drives value pushes in tests.</summary>
+    public SubscriptionCallback? ValueCallback;
+    /// <summary>Callback the last alarm subscribe registered.</summary>
+    public AlarmTransitionCallback? AlarmCallback;
+
+    /// <inheritdoc />
+    public ConnectionHealth Status { get; private set; } = ConnectionHealth.Disconnected;
+
+    /// <inheritdoc />
+    public event Action? Disconnected;
+
+    /// <summary>Raises <see cref="Disconnected"/> as a real adapter would on a transport fault.</summary>
+    public void RaiseDisconnected() => Disconnected?.Invoke();
+
+    /// <inheritdoc />
+    public Task ConnectAsync(IDictionary<string, string> connectionDetails, CancellationToken cancellationToken = default)
+    {
+        Status = ConnectionHealth.Connected;
+        return Task.CompletedTask;
+    }
+
+    /// <inheritdoc />
+    public Task DisconnectAsync(CancellationToken cancellationToken = default)
+    {
+        Status = ConnectionHealth.Disconnected;
+        return Task.CompletedTask;
+    }
+
+    /// <inheritdoc />
+    public Task<IReadOnlyList<TagSubscribeResult>> SubscribeBatchAsync(
+        IReadOnlyList<string> tagPaths, SubscriptionCallback callback, CancellationToken cancellationToken = default)
+    {
+        SubscribeBatches.Enqueue(tagPaths.ToList());
+        SubscribeBatchTimes.Enqueue(DateTimeOffset.UtcNow);
+        ValueCallback = callback;
+
+        if (BatchSubscribeThrows is { } factory)
+            throw factory();
+
+        IReadOnlyList<TagSubscribeResult> rows = tagPaths
+            .Select(t => FailingTags.Contains(t)
+                ? new TagSubscribeResult(t, false, null, "node not found")
+                : new TagSubscribeResult(t, true, $"sub-{Interlocked.Increment(ref _nextId)}", null))
+            .ToList();
+        return Task.FromResult(rows);
+    }
+
+    /// <inheritdoc />
+    public Task UnsubscribeBatchAsync(IReadOnlyList<string> subscriptionIds, CancellationToken cancellationToken = default)
+    {
+        UnsubscribeBatches.Enqueue(subscriptionIds.ToList());
+        return Task.CompletedTask;
+    }
+
+    /// <inheritdoc />
+    public Task<string> SubscribeAsync(string tagPath, SubscriptionCallback callback, CancellationToken cancellationToken = default)
+    {
+        Interlocked.Increment(ref SingleSubscribeCalls);
+        ValueCallback = callback;
+        return Task.FromResult($"sub-{Interlocked.Increment(ref _nextId)}");
+    }
+
+    /// <inheritdoc />
+    public Task UnsubscribeAsync(string subscriptionId, CancellationToken cancellationToken = default)
+        => Task.CompletedTask;
+
+    /// <inheritdoc />
+    public async Task<ReadResult> ReadAsync(string tagPath, CancellationToken cancellationToken = default)
+    {
+        Interlocked.Increment(ref SingleReadCalls);
+        if (HangReads)
+            await Task.Delay(Timeout.Infinite, cancellationToken);
+        return new ReadResult(true, new TagValue(SeedValue, QualityCode.Good, DateTimeOffset.UtcNow), null);
+    }
+
+    /// <inheritdoc />
+    public async Task<IReadOnlyDictionary<string, ReadResult>> ReadBatchAsync(
+        IEnumerable<string> tagPaths, CancellationToken cancellationToken = default)
+    {
+        var tags = tagPaths.ToList();
+        ReadBatches.Enqueue(tags);
+        if (HangReads)
+            await Task.Delay(Timeout.Infinite, cancellationToken);
+
+        return tags.ToDictionary(
+            t => t,
+            t => new ReadResult(true, new TagValue(SeedValue, QualityCode.Good, DateTimeOffset.UtcNow), null));
+    }
+
+    /// <inheritdoc />
+    public Task<WriteResult> WriteAsync(string tagPath, object? value, CancellationToken cancellationToken = default)
+        => Task.FromResult(new WriteResult(true, null));
+
+    /// <inheritdoc />
+    public Task<IReadOnlyDictionary<string, WriteResult>> WriteBatchAsync(
+        IDictionary<string, object?> values, CancellationToken cancellationToken = default)
+        => Task.FromResult<IReadOnlyDictionary<string, WriteResult>>(
+            values.ToDictionary(kv => kv.Key, _ => new WriteResult(true, null)));
+
+    /// <inheritdoc />
+    public Task<bool> WriteBatchAndWaitAsync(
+        IDictionary<string, object?> values, string flagPath, object? flagValue, string responsePath,
+        object? responseValue, TimeSpan timeout, CancellationToken cancellationToken = default)
+        => Task.FromResult(true);
+
+    /// <inheritdoc />
+    public Task<string> SubscribeAlarmsAsync(
+        string sourceReference, string? conditionFilter, AlarmTransitionCallback callback,
+        CancellationToken cancellationToken = default)
+    {
+        AlarmCallback = callback;
+        return Task.FromResult($"alarm-{Interlocked.Increment(ref _nextId)}");
+    }
+
+    /// <inheritdoc />
+    public Task UnsubscribeAlarmsAsync(string subscriptionId, CancellationToken cancellationToken = default)
+        => Task.CompletedTask;
+
+    /// <inheritdoc />
+    public ValueTask DisposeAsync() => ValueTask.CompletedTask;
+}

tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/BatchSeamPrimitiveTests.cs

@@ -0,0 +1,204 @@
+using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
+using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
+
+namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters;
+
+/// <summary>
+/// WP2.1b — the primitives the batch seam is built from, tested in isolation because the
+/// OPC Foundation Session/Subscription types they drive cannot be faked without a live
+/// server: monitored-item shard placement, the apply-serialization gate, the bounded
+/// pipeline behind the MxGateway supervisory advise, the alarm-stream union prefix, and
+/// the tag-resolution backoff step.
+/// </summary>
+public class BatchSeamPrimitiveTests
+{
+    // ── Shard placement ──
+
+    [Fact]
+    public void Sharding_SplitsItemsAtTheBudget()
+    {
+        // 12,001 items at the 5,000 default → 3 shards (memo §2 sizing example).
+        Assert.Equal(3, MonitoredItemShardPlanner.ShardCountFor(12_001, 5_000));
+        // 37,500 tags → 8 shards.
+        Assert.Equal(8, MonitoredItemShardPlanner.ShardCountFor(37_500, 5_000));
+
+        var placement = MonitoredItemShardPlanner.Plan([], 5_000, 12_001);
+        Assert.Equal(12_001, placement.Count);
+        Assert.Equal(0, placement[0]);
+        Assert.Equal(0, placement[4_999]);
+        Assert.Equal(1, placement[5_000]);
+        Assert.Equal(2, placement[10_000]);
+        Assert.Equal(2, placement[12_000]);
+    }
+
+    [Fact]
+    public void Sharding_FillsTheFirstShardWithFreeCapacityBeforeCreatingOne()
+    {
+        // Shard 0 full, shard 1 has one free slot: the next two items fill shard 1 then
+        // open shard 2 — the "first shard with free capacity, else a new shard" policy,
+        // which is also what makes an emptied-and-deleted shard's capacity reusable.
+        var placement = MonitoredItemShardPlanner.Plan([5_000, 4_999], 5_000, 2);
+        Assert.Equal([1, 2], placement);
+    }
+
+    [Fact]
+    public void Sharding_NonPositiveBudgetDegradesToOneItemPerShard()
+    {
+        Assert.Equal([0, 1, 2], MonitoredItemShardPlanner.Plan([], 0, 3));
+    }
+
+    // ── Apply serialization ──
+
+    [Fact]
+    public async Task ApplyGate_NeverRunsTwoSectionsConcurrently()
+    {
+        var gate = new AsyncSerialGate();
+        var inFlight = 0;
+        var maxObserved = 0;
+
+        var tasks = Enumerable.Range(0, 32).Select(_ => gate.RunAsync(async () =>
+        {
+            var now = Interlocked.Increment(ref inFlight);
+            InterlockedMax(ref maxObserved, now);
+            await Task.Delay(5);
+            Interlocked.Decrement(ref inFlight);
+        })).ToArray();
+
+        await Task.WhenAll(tasks);
+
+        Assert.Equal(1, maxObserved);
+    }
+
+    [Fact]
+    public async Task ApplyGate_ReleasesWhenASectionThrows()
+    {
+        var gate = new AsyncSerialGate();
+
+        await Assert.ThrowsAsync<InvalidOperationException>(() =>
+            gate.RunAsync(() => throw new InvalidOperationException("apply failed")));
+
+        // The gate must still admit the next caller — a faulted ApplyChanges must not
+        // wedge every later subscribe/unsubscribe on this client.
+        var ran = false;
+        await gate.RunAsync(() =>
+        {
+            ran = true;
+            return Task.CompletedTask;
+        });
+        Assert.True(ran);
+    }
+
+    // ── Bounded pipeline (MxGateway supervisory advise) ──
+
+    [Fact]
+    public async Task BulkPipeline_KeepsInFlightCountWithinTheWindow()
+    {
+        const int parallelism = 16;
+        var inFlight = 0;
+        var maxObserved = 0;
+        var items = Enumerable.Range(0, 200).ToList();
+
+        var results = await BulkPipeline.RunAsync(
+            items,
+            parallelism,
+            async (item, _) =>
+            {
+                var now = Interlocked.Increment(ref inFlight);
+                InterlockedMax(ref maxObserved, now);
+                await Task.Delay(2);
+                Interlocked.Decrement(ref inFlight);
+                return item * 2;
+            },
+            (item, _) => -item);
+
+        Assert.Equal(items.Count, results.Length);
+        Assert.Equal(items.Select(i => i * 2), results);
+        Assert.InRange(maxObserved, 1, parallelism);
+    }
+
+    [Fact]
+    public async Task BulkPipeline_CapturesPerItemFaultsWithoutAbortingTheBatch()
+    {
+        var results = await BulkPipeline.RunAsync(
+            [1, 2, 3],
+            4,
+            (item, _) => item == 2
+                ? throw new InvalidOperationException("advise failed")
+                : Task.FromResult(item),
+            (item, _) => -item);
+
+        Assert.Equal([1, -2, 3], results);
+    }
+
+    // ── Alarm-stream union prefix ──
+
+    [Theory]
+    [InlineData(new[] { "Area1.Tank1", "Area1.Tank2" }, "Area1.Tank")]
+    [InlineData(new[] { "Area1.Tank1" }, "Area1.Tank1")]
+    [InlineData(new[] { "Area1.Tank1", "Area2.Tank1" }, "Area")]
+    [InlineData(new[] { "Plant.A", "Zone.B" }, "")]
+    [InlineData(new[] { "Area1.Tank1", "" }, "")]
+    [InlineData(new string[0], "")]
+    public void AlarmPrefix_IsTheLongestCommonPrefix(string[] sources, string expected)
+    {
+        Assert.Equal(expected, AlarmFilterPrefix.LongestCommonPrefix(sources));
+    }
+
+    [Fact]
+    public void AlarmPrefix_CoversOnlySourcesUnderTheLivePrefix()
+    {
+        Assert.True(AlarmFilterPrefix.Covers("Area1.", "Area1.Tank1"));
+        Assert.False(AlarmFilterPrefix.Covers("Area1.", "Area2.Tank1"));
+        // A gateway-wide stream covers everything.
+        Assert.True(AlarmFilterPrefix.Covers("", "Anything.At.All"));
+    }
+
+    // ── Tag-resolution backoff ──
+
+    [Fact]
+    public void Backoff_DoublesPerFailedRoundAndCapsAtTheMaximum()
+    {
+        var floor = TimeSpan.FromSeconds(10);
+        var max = TimeSpan.FromMinutes(5);
+
+        var sequence = new List<TimeSpan>();
+        var current = floor;
+        for (var round = 0; round < 8; round++)
+        {
+            current = DataConnectionActor.NextTagResolutionInterval(current, floor, max);
+            sequence.Add(current);
+        }
+
+        Assert.Equal(
+        [
+            TimeSpan.FromSeconds(20),
+            TimeSpan.FromSeconds(40),
+            TimeSpan.FromSeconds(80),
+            TimeSpan.FromSeconds(160),
+            TimeSpan.FromSeconds(300),
+            TimeSpan.FromSeconds(300),
+            TimeSpan.FromSeconds(300),
+            TimeSpan.FromSeconds(300),
+        ], sequence);
+    }
+
+    [Fact]
+    public void Backoff_MisconfiguredCeilingBelowFloorDegradesToAFixedInterval()
+    {
+        var floor = TimeSpan.FromSeconds(10);
+        var next = DataConnectionActor.NextTagResolutionInterval(floor, floor, TimeSpan.FromSeconds(1));
+        Assert.Equal(floor, next);
+    }
+
+    private static void InterlockedMax(ref int target, int value)
+    {
+        int seen;
+        do
+        {
+            seen = Volatile.Read(ref target);
+            if (value <= seen)
+                return;
+        }
+        while (Interlocked.CompareExchange(ref target, value, seen) != seen);
+    }
+}

tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/FakeMxGatewayClient.cs

@@ -11,6 +11,14 @@ public sealed class FakeMxGatewayClient : IMxGatewayClient, IMxGatewayClientFact
     public MxGatewayConnectionOptions? ConnectedWith;
     public readonly List<string> Subscribed = new();
     public readonly List<string> Unsubscribed = new();
+    /// <summary>One entry per SubscribeBulkAsync call, carrying that call's tag list.</summary>
+    public readonly List<IReadOnlyList<string>> BulkSubscribeCalls = new();
+    /// <summary>One entry per UnsubscribeBulkAsync call, carrying that call's id list.</summary>
+    public readonly List<IReadOnlyList<string>> BulkUnsubscribeCalls = new();
+    /// <summary>Tags the fake reports as failed rows from a bulk subscribe.</summary>
+    public readonly HashSet<string> BulkSubscribeFailures = new(StringComparer.Ordinal);
+    /// <summary>Every alarm-stream prefix the adapter opened a stream with (null = gateway-wide).</summary>
+    public readonly List<string?> AlarmStreamPrefixes = new();
     public readonly TaskCompletionSource EventLoopGate = new(TaskCreationOptions.RunContinuationsAsynchronously);
     public Action<MxValueUpdate>? OnUpdate;
     public Func<IReadOnlyList<string>, IReadOnlyList<MxReadOutcome>>? ReadHandler;
@@ -41,6 +49,34 @@ public sealed class FakeMxGatewayClient : IMxGatewayClient, IMxGatewayClientFact
         return Task.CompletedTask;
     }
 
+    public Task<IReadOnlyList<MxSubscribeOutcome>> SubscribeBulkAsync(
+        IReadOnlyList<string> tagPaths, CancellationToken ct = default)
+    {
+        BulkSubscribeCalls.Add(tagPaths.ToList());
+        var outcomes = new List<MxSubscribeOutcome>(tagPaths.Count);
+        foreach (var tag in tagPaths)
+        {
+            if (BulkSubscribeFailures.Contains(tag))
+            {
+                outcomes.Add(new MxSubscribeOutcome(tag, false, null, "not found"));
+                continue;
+            }
+
+            Subscribed.Add(tag);
+            outcomes.Add(new MxSubscribeOutcome(
+                tag, true, (++_nextHandle).ToString(), null));
+        }
+
+        return Task.FromResult<IReadOnlyList<MxSubscribeOutcome>>(outcomes);
+    }
+
+    public Task UnsubscribeBulkAsync(IReadOnlyList<string> subscriptionIds, CancellationToken ct = default)
+    {
+        BulkUnsubscribeCalls.Add(subscriptionIds.ToList());
+        Unsubscribed.AddRange(subscriptionIds);
+        return Task.CompletedTask;
+    }
+
     public Task<IReadOnlyList<MxReadOutcome>> ReadAsync(IReadOnlyList<string> tags, CancellationToken ct = default)
         => Task.FromResult(ReadHandler!(tags));
 
@@ -62,7 +98,13 @@ public sealed class FakeMxGatewayClient : IMxGatewayClient, IMxGatewayClientFact
         string? alarmFilterPrefix,
         Action<ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms.NativeAlarmTransition> onTransition,
         CancellationToken ct = default)
-        => Task.CompletedTask; // no alarm feed in the fake
+    {
+        // No alarm feed in the fake — but the prefix each stream is opened with is
+        // recorded so the union-filter (longest-common-prefix) behaviour is testable.
+        lock (AlarmStreamPrefixes)
+            AlarmStreamPrefixes.Add(alarmFilterPrefix);
+        return Task.CompletedTask;
+    }
 
     public ValueTask DisposeAsync() => ValueTask.CompletedTask;
 

tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayBatchSeamTests.cs

@@ -0,0 +1,146 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
+using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
+
+namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters;
+
+/// <summary>
+/// WP2.1b — the MxGateway adapter's batch subscribe seam and the alarm-stream union
+/// filter. The plain-vs-supervisory advise choice itself lives inside
+/// <c>RealMxGatewayClient</c> (it needs a live MXAccess session); its bounded-pipeline
+/// mechanism is pinned separately by <see cref="BatchSeamPrimitiveTests"/>.
+/// </summary>
+[Collection("DataConnectionManagerActor")]
+public class MxGatewayBatchSeamTests
+{
+    private static MxGatewayDataConnection NewAdapter(FakeMxGatewayClient fake) =>
+        new(fake, NullLogger<MxGatewayDataConnection>.Instance);
+
+    private static Dictionary<string, string> Details() => new()
+    {
+        ["Endpoint"] = "http://gw:5000",
+        ["ApiKey"] = "key",
+        ["ClientName"] = "client-a",
+        ["WriteUserId"] = "0",
+        ["ReadTimeoutMs"] = "2000",
+    };
+
+    [Fact]
+    public async Task SubscribeBatch_IssuesExactlyOneBulkCallPerBatch()
+    {
+        var fake = new FakeMxGatewayClient();
+        var adapter = NewAdapter(fake);
+        await adapter.ConnectAsync(Details());
+
+        var results = await adapter.SubscribeBatchAsync(
+            ["A.x", "A.y", "A.z"], (_, _) => { });
+
+        // ONE bulk RPC for the whole batch — replacing the historical AddItem + Advise
+        // pair PER TAG (6 RPCs for these three tags).
+        Assert.Single(fake.BulkSubscribeCalls);
+        Assert.Equal(3, fake.BulkSubscribeCalls[0].Count);
+        Assert.Equal(3, results.Count);
+        Assert.All(results, r => Assert.True(r.Success));
+        Assert.All(results, r => Assert.NotNull(r.SubscriptionId));
+    }
+
+    [Fact]
+    public async Task SubscribeBatch_ReportsPerTagFailuresWithoutFailingTheBatch()
+    {
+        var fake = new FakeMxGatewayClient();
+        fake.BulkSubscribeFailures.Add("A.y");
+        var adapter = NewAdapter(fake);
+        await adapter.ConnectAsync(Details());
+
+        var results = await adapter.SubscribeBatchAsync(["A.x", "A.y"], (_, _) => { });
+
+        Assert.True(results.Single(r => r.TagPath == "A.x").Success);
+        var failed = results.Single(r => r.TagPath == "A.y");
+        Assert.False(failed.Success);
+        Assert.Null(failed.SubscriptionId);
+        Assert.NotNull(failed.ErrorMessage);
+    }
+
+    [Fact]
+    public async Task UnsubscribeBatch_ReleasesEveryIdInOneBulkCall()
+    {
+        var fake = new FakeMxGatewayClient();
+        var adapter = NewAdapter(fake);
+        await adapter.ConnectAsync(Details());
+
+        var results = await adapter.SubscribeBatchAsync(["A.x", "A.y"], (_, _) => { });
+        await adapter.UnsubscribeBatchAsync(results.Select(r => r.SubscriptionId!).ToList());
+
+        Assert.Single(fake.BulkUnsubscribeCalls);
+        Assert.Equal(2, fake.BulkUnsubscribeCalls[0].Count);
+    }
+
+    [Fact]
+    public async Task AlarmStream_OpensOnTheLongestCommonPrefix_AndOnlyRestartsForAnEscapingSource()
+    {
+        var fake = new FakeMxGatewayClient();
+        var adapter = NewAdapter(fake);
+        await adapter.ConnectAsync(Details());
+
+        void NoTransition(NativeAlarmTransition _) { }
+
+        // First source → the stream opens scoped to it.
+        await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, NoTransition);
+        await WaitForPrefixCountAsync(fake, 1);
+        Assert.Equal("Area1.Tank1", fake.AlarmStreamPrefixes[0]);
+
+        // A sibling under the same prefix widens the union → one restart on "Area1.Tank".
+        await adapter.SubscribeAlarmsAsync("Area1.Tank2", null, NoTransition);
+        await WaitForPrefixCountAsync(fake, 2);
+        Assert.Equal("Area1.Tank", fake.AlarmStreamPrefixes[1]);
+
+        // A source ALREADY covered by the live prefix must NOT restart the stream.
+        await adapter.SubscribeAlarmsAsync("Area1.Tank3.Sub", null, NoTransition);
+        await Task.Delay(100);
+        Assert.Equal(2, fake.AlarmStreamPrefixes.Count);
+
+        // A source outside the prefix widens it again — here down to gateway-wide.
+        await adapter.SubscribeAlarmsAsync("Zone9.Pump", null, NoTransition);
+        await WaitForPrefixCountAsync(fake, 3);
+        Assert.Null(fake.AlarmStreamPrefixes[2]); // "" → gateway-wide
+    }
+
+    [Fact]
+    public async Task AlarmStream_UnsubscribeNeverRestartsTheStream()
+    {
+        var fake = new FakeMxGatewayClient();
+        var adapter = NewAdapter(fake);
+        await adapter.ConnectAsync(Details());
+
+        var first = await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, _ => { });
+        // Wait for the first stream to actually open before widening: the open runs on a
+        // Task.Run whose token the restart cancels, so a restart racing an unstarted task
+        // would leave the first prefix unrecorded.
+        await WaitForPrefixCountAsync(fake, 1);
+        await adapter.SubscribeAlarmsAsync("Area1.Tank2", null, _ => { });
+        await WaitForPrefixCountAsync(fake, 2);
+
+        // Dropping a source leaves the prefix too BROAD at worst — bandwidth, not
+        // correctness — so no restart is issued.
+        await adapter.UnsubscribeAlarmsAsync(first);
+        await Task.Delay(100);
+        Assert.Equal(2, fake.AlarmStreamPrefixes.Count);
+    }
+
+    private static async Task WaitForPrefixCountAsync(FakeMxGatewayClient fake, int expected)
+    {
+        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
+        while (DateTime.UtcNow < deadline)
+        {
+            lock (fake.AlarmStreamPrefixes)
+            {
+                if (fake.AlarmStreamPrefixes.Count >= expected)
+                    return;
+            }
+            await Task.Delay(20);
+        }
+
+        Assert.Fail($"alarm stream was opened {fake.AlarmStreamPrefixes.Count} time(s), expected {expected}");
+    }
+}

tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/DataConnectionActorAlarmIndexTests.cs

@@ -0,0 +1,166 @@
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using NSubstitute;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
+using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
+
+namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests;
+
+/// <summary>
+/// WP2.1b — the alarm subscriber first-segment bucket index must make EXACTLY the routing
+/// decisions the previous linear scan made. The expectation in each case is computed with
+/// the original rule (plain <c>StartsWith</c> against every subscribed source), so the
+/// test pins equivalence rather than restating the new implementation.
+/// </summary>
+public class DataConnectionActorAlarmIndexTests : TestKit
+{
+    private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
+    private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();
+    private readonly DataConnectionOptions _options = new()
+    {
+        ReconnectInterval = TimeSpan.FromMilliseconds(100),
+        TagResolutionRetryInterval = TimeSpan.FromMilliseconds(200),
+        WriteTimeout = TimeSpan.FromSeconds(5)
+    };
+
+    // Sources chosen to exercise every index path:
+    //  - "Area1"            → NO separator: the residue list (can match other buckets)
+    //  - "Area1.Tank"       → bucket "Area1", shared prefix with the next one
+    //  - "Area1.Tank1.Sub"  → bucket "Area1", deeper
+    //  - "Area2.Tank"       → a different bucket
+    private static readonly string[] Sources = ["Area1", "Area1.Tank", "Area1.Tank1.Sub", "Area2.Tank"];
+
+    private static readonly (string SourceRef, string SourceObjectRef)[] Transitions =
+    [
+        ("Area1.Tank1.Hi", "Area1.Tank1"),   // Area1 (residue) + Area1.Tank
+        ("Area1.Tank1.Sub.Hi", "Area1.Tank1.Sub"), // + Area1.Tank1.Sub
+        ("Area1X.Pump.Hi", "Area1X.Pump"),   // sub-segment prefix: Area1 ONLY
+        ("Area2.Tank9.Hi", "Area2.Tank9"),   // Area2.Tank only
+        ("Zone.A.Hi", "Zone.A"),             // nobody
+    ];
+
+    private static NativeAlarmTransition Transition(string sourceRef, string sourceObj) =>
+        new(sourceRef, sourceObj, "AnalogLimit.Hi", AlarmTransitionKind.Raise,
+            new AlarmConditionState(true, false, null, AlarmShelveState.Unshelved, false, 500),
+            "Process", "hi", "hi", "", "", null, DateTimeOffset.UtcNow, "92", "90");
+
+    /// <summary>The ORIGINAL linear-scan rule, kept here as the oracle.</summary>
+    private static bool LinearScanMatches(string sourceRef, string transitionSourceRef, string transitionSourceObjectRef) =>
+        transitionSourceObjectRef.StartsWith(sourceRef, StringComparison.Ordinal)
+        || transitionSourceRef.StartsWith(sourceRef, StringComparison.Ordinal);
+
+    [Fact]
+    public void BucketIndex_RoutesExactlyLikeTheLinearScan()
+    {
+        AlarmTransitionCallback? cb = null;
+        var adapter = Substitute.For<IDataConnection, IAlarmSubscribableConnection>();
+        adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
+            .Returns(Task.CompletedTask);
+        ((IAlarmSubscribableConnection)adapter)
+            .SubscribeAlarmsAsync(Arg.Any<string>(), Arg.Any<string?>(),
+                Arg.Do<AlarmTransitionCallback>(c => cb = c), Arg.Any<CancellationToken>())
+            .Returns(ci => Task.FromResult("alarm-" + ci.ArgAt<string>(0)));
+
+        var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
+            "conn", adapter, _options, _health, _factory, "OpcUa")), "alarm-index");
+
+        // One probe per source so the routing decision per source is observable.
+        var probes = Sources.ToDictionary(s => s, _ => CreateTestProbe());
+        foreach (var source in Sources)
+        {
+            actor.Tell(new SubscribeAlarmsRequest("c", "inst-" + source, "conn", source, null, DateTimeOffset.UtcNow),
+                probes[source].Ref);
+            probes[source].ExpectMsg<SubscribeAlarmsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
+        }
+
+        Assert.NotNull(cb);
+
+        foreach (var (sourceRef, sourceObjectRef) in Transitions)
+        {
+            cb!(Transition(sourceRef, sourceObjectRef));
+
+            foreach (var source in Sources)
+            {
+                if (LinearScanMatches(source, sourceRef, sourceObjectRef))
+                {
+                    probes[source].ExpectMsg<NativeAlarmTransitionUpdate>(
+                        u => u.Transition.SourceReference == sourceRef, TimeSpan.FromSeconds(5));
+                }
+            }
+        }
+
+        // Nothing extra was delivered to anybody.
+        foreach (var source in Sources)
+            probes[source].ExpectNoMsg(TimeSpan.FromMilliseconds(200));
+    }
+
+    [Fact]
+    public void SnapshotComplete_StillBroadcastsToEverySubscriber_BypassingTheIndex()
+    {
+        // The framing sentinel carries an EMPTY source reference, so it matches no bucket.
+        // It must still reach every alarm subscriber, or statically-active conditions
+        // delivered only in the snapshot would buffer forever.
+        AlarmTransitionCallback? cb = null;
+        var adapter = Substitute.For<IDataConnection, IAlarmSubscribableConnection>();
+        adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
+            .Returns(Task.CompletedTask);
+        ((IAlarmSubscribableConnection)adapter)
+            .SubscribeAlarmsAsync(Arg.Any<string>(), Arg.Any<string?>(),
+                Arg.Do<AlarmTransitionCallback>(c => cb = c), Arg.Any<CancellationToken>())
+            .Returns(ci => Task.FromResult("alarm-" + ci.ArgAt<string>(0)));
+
+        var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
+            "conn", adapter, _options, _health, _factory, "OpcUa")), "alarm-index-snapshot");
+
+        var probes = Sources.ToDictionary(s => s, _ => CreateTestProbe());
+        foreach (var source in Sources)
+        {
+            actor.Tell(new SubscribeAlarmsRequest("c", "inst-" + source, "conn", source, null, DateTimeOffset.UtcNow),
+                probes[source].Ref);
+            probes[source].ExpectMsg<SubscribeAlarmsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
+        }
+
+        Assert.NotNull(cb);
+        cb!(new NativeAlarmTransition(
+            "", "", "", AlarmTransitionKind.SnapshotComplete,
+            new AlarmConditionState(false, true, null, AlarmShelveState.Unshelved, false, 0),
+            "", "", "", "", "", null, DateTimeOffset.UtcNow, "", ""));
+
+        foreach (var source in Sources)
+        {
+            probes[source].ExpectMsg<NativeAlarmTransitionUpdate>(
+                u => u.Transition.Kind == AlarmTransitionKind.SnapshotComplete, TimeSpan.FromSeconds(5));
+        }
+    }
+
+    [Fact]
+    public void UnsubscribedSource_IsDroppedFromTheIndex()
+    {
+        AlarmTransitionCallback? cb = null;
+        var adapter = Substitute.For<IDataConnection, IAlarmSubscribableConnection>();
+        adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
+            .Returns(Task.CompletedTask);
+        ((IAlarmSubscribableConnection)adapter)
+            .SubscribeAlarmsAsync(Arg.Any<string>(), Arg.Any<string?>(),
+                Arg.Do<AlarmTransitionCallback>(c => cb = c), Arg.Any<CancellationToken>())
+            .Returns(ci => Task.FromResult("alarm-" + ci.ArgAt<string>(0)));
+
+        var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
+            "conn", adapter, _options, _health, _factory, "OpcUa")), "alarm-index-unsub");
+
+        var probe = CreateTestProbe();
+        actor.Tell(new SubscribeAlarmsRequest("c", "inst", "conn", "Area1.Tank", null, DateTimeOffset.UtcNow), probe.Ref);
+        probe.ExpectMsg<SubscribeAlarmsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
+
+        actor.Tell(new UnsubscribeAlarmsRequest("c2", "inst", "conn", "Area1.Tank", DateTimeOffset.UtcNow), probe.Ref);
+        Thread.Sleep(200);
+
+        Assert.NotNull(cb);
+        cb!(Transition("Area1.Tank1.Hi", "Area1.Tank1"));
+        probe.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+    }
+}

tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/OpcUaDataConnectionTests.cs

@@ -323,33 +323,41 @@ public class OpcUaDataConnectionTests
     [Fact]
     public async Task ReadBatch_ReadsAllTags()
     {
+        // WP2.1b: ReadBatchAsync is TRUE bulk — ONE client ReadValuesAsync call carrying
+        // every requested node, not a per-tag loop.
         _mockClient.IsConnected.Returns(true);
-        _mockClient.ReadValueAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
+        _mockClient.ReadValuesAsync(Arg.Any<IReadOnlyList<string>>(), Arg.Any<CancellationToken>())
-            .Returns((1.0, DateTime.UtcNow, 0u));
+            .Returns(ci => Task.FromResult<IReadOnlyList<OpcUaReadOutcome>>(
+                ci.Arg<IReadOnlyList<string>>()
+                    .Select(n => new OpcUaReadOutcome(n, 1.0, DateTime.UtcNow, 0u, null))
+                    .ToList()));
 
         await _adapter.ConnectAsync(new Dictionary<string, string>());
         var results = await _adapter.ReadBatchAsync(["tag1", "tag2", "tag3"]);
 
         Assert.Equal(3, results.Count);
         Assert.All(results.Values, r => Assert.True(r.Success));
+        Assert.Single(_mockClient.ReceivedCalls(), c => c.GetMethodInfo().Name == "ReadValuesAsync");
     }
 
     [Fact]
     public async Task DCL007_ReadBatch_ReturnsPerTagResults_WhenOneTagFails()
     {
-        // Regression test for DataConnectionLayer-007. ReadBatchAsync looped calling
+        // Regression test for DataConnectionLayer-007. ReadBatchAsync originally looped
-        // ReadAsync per tag; ReadAsync re-throws any non-cancellation exception, so a
+        // calling ReadAsync per tag; ReadAsync re-throws any non-cancellation exception, so
-        // single failing tag aborted the whole batch and the caller got NO results for
+        // a single failing tag aborted the whole batch and the caller got NO results for
-        // the tags that did read successfully — even though ReadResult already carries
+        // the tags that did read successfully — even though ReadResult already carries a
-        // a per-tag Success/ErrorMessage shape. After the fix the batch catches per-tag
+        // per-tag Success/ErrorMessage shape. The batch is now one bulk service call, and
-        // exceptions and returns a complete map.
+        // the same invariant holds: a per-node failure row never aborts the batch, and
+        // every requested tag comes back in the map.
         _mockClient.IsConnected.Returns(true);
-        _mockClient.ReadValueAsync("good1", Arg.Any<CancellationToken>())
+        _mockClient.ReadValuesAsync(Arg.Any<IReadOnlyList<string>>(), Arg.Any<CancellationToken>())
-            .Returns((1.0, DateTime.UtcNow, 0u));
+            .Returns(ci => Task.FromResult<IReadOnlyList<OpcUaReadOutcome>>(
-        _mockClient.ReadValueAsync("bad", Arg.Any<CancellationToken>())
+                ci.Arg<IReadOnlyList<string>>()
-            .Returns<(object?, DateTime, uint)>(_ => throw new InvalidOperationException("node not found"));
+                    .Select(n => n == "bad"
-        _mockClient.ReadValueAsync("good2", Arg.Any<CancellationToken>())
+                        ? new OpcUaReadOutcome(n, null, DateTime.UtcNow, 0x80340000u, "node not found")
-            .Returns((2.0, DateTime.UtcNow, 0u));
+                        : new OpcUaReadOutcome(n, 1.0, DateTime.UtcNow, 0u, null))
+                    .ToList()));
 
         await _adapter.ConnectAsync(new Dictionary<string, string>());
 
@@ -365,28 +373,25 @@ public class OpcUaDataConnectionTests
     }
 
     [Fact]
-    public async Task DCL017_WriteBatch_ReturnsPerTagResults_WhenConnectionDropsMidBatch()
+    public async Task DCL017_WriteBatch_ReturnsPerTagResults_WhenSomeTagsFail()
     {
-        // Regression test for DataConnectionLayer-017. WriteBatchAsync looped calling
+        // Regression test for DataConnectionLayer-017. WriteBatchAsync originally looped
-        // WriteAsync per tag; WriteAsync first calls EnsureConnected(), which throws
+        // calling WriteAsync per tag; a mid-batch fault made the whole call throw and the
-        // InvalidOperationException when the client is disconnected. WriteBatchAsync did
+        // caller lost the per-tag outcomes for the tags that already wrote. The batch is
-        // not catch that, so a connection dropping partway through a batch made the whole
+        // now ONE bulk service call (WP2.1b), and the invariant is unchanged: per-node
-        // WriteBatchAsync throw — the caller lost the per-tag outcomes for the tags that
+        // failures are reported as failed WriteResult rows and every requested tag is
-        // already wrote. After the fix (mirroring DCL-007's ReadBatchAsync) each per-tag
+        // present in the returned map.
-        // failure is recorded as a failed WriteResult and the batch returns a complete map.
-        var writeCount = 0;
-        // First write succeeds; then the client "disconnects" so EnsureConnected throws.
-        _mockClient.IsConnected.Returns(_ => Interlocked.Increment(ref writeCount) <= 1);
-        _mockClient.WriteValueAsync(Arg.Any<string>(), Arg.Any<object?>(), Arg.Any<CancellationToken>())
-            .Returns((uint)0);
-
-        // Connect leaves IsConnected true for the first WriteAsync's EnsureConnected check.
         _mockClient.IsConnected.Returns(true);
         await _adapter.ConnectAsync(new Dictionary<string, string>());
 
-        // Re-arm: IsConnected true for tag1's check, false for tag2 and tag3.
+        _mockClient.WriteValuesAsync(
-        var checks = 0;
+                Arg.Any<IReadOnlyList<(string NodeId, object? Value)>>(), Arg.Any<CancellationToken>())
-        _mockClient.IsConnected.Returns(_ => Interlocked.Increment(ref checks) <= 1);
+            .Returns(ci => Task.FromResult<IReadOnlyList<OpcUaWriteOutcome>>(
+                ci.Arg<IReadOnlyList<(string NodeId, object? Value)>>()
+                    .Select(v => v.NodeId == "tag1"
+                        ? new OpcUaWriteOutcome(v.NodeId, 0u, null)
+                        : new OpcUaWriteOutcome(v.NodeId, 0x80AE0000u, null))
+                    .ToList()));
 
         var results = await _adapter.WriteBatchAsync(new Dictionary<string, object?>
         {
@@ -398,11 +403,12 @@ public class OpcUaDataConnectionTests
         // Every requested tag is present in the result map — the batch was not aborted.
         Assert.Equal(3, results.Count);
         Assert.True(results["tag1"].Success);
-        // tag2 and tag3 fail at the connection check but are reported per-tag.
         Assert.False(results["tag2"].Success);
         Assert.NotNull(results["tag2"].ErrorMessage);
         Assert.False(results["tag3"].Success);
         Assert.NotNull(results["tag3"].ErrorMessage);
+        // ONE bulk write, not three single writes.
+        Assert.Single(_mockClient.ReceivedCalls(), c => c.GetMethodInfo().Name == "WriteValuesAsync");
     }
 
     [Fact]
@@ -415,8 +421,9 @@ public class OpcUaDataConnectionTests
 
         using var cts = new CancellationTokenSource();
         cts.Cancel();
-        _mockClient.WriteValueAsync(Arg.Any<string>(), Arg.Any<object?>(), Arg.Any<CancellationToken>())
+        _mockClient.WriteValuesAsync(
-            .Returns<uint>(_ => throw new OperationCanceledException());
+                Arg.Any<IReadOnlyList<(string NodeId, object? Value)>>(), Arg.Any<CancellationToken>())
+            .Returns<IReadOnlyList<OpcUaWriteOutcome>>(_ => throw new OperationCanceledException());
 
         await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
             _adapter.WriteBatchAsync(new Dictionary<string, object?> { ["tag1"] = 1 }, cts.Token));