docs(galaxy): clarify native alarm acknowledgement outcomes

docs/AlarmHistorian.md

@@ -153,6 +153,19 @@ Ordering follows from that: the drain reads in `enqueued_at_utc` order (with `id
 as a tiebreak so the ordering is total), because a hashed key carries no
 insertion sequence.
 
+Gateway delivery carries a separate nonzero `DeliveryId` GUID. Direct event instances
+create it once; retries and record copies preserve it. It is excluded from queue JSON
+and business equality, so the existing payload fingerprint and pair deduplication stay
+unchanged. The drain derives its delivery GUID solely from the persisted row `id`,
+including migrated legacy rows. It never hashes a reconstructed payload or mints an ID
+per send. The versioned encoding in `AlarmDeliveryIds` therefore survives retry,
+restart, payload edits and replicated failover without a schema migration.
+
+Different business payloads at the same timestamp remain distinct; completely identical
+payloads deliberately converge. An explicitly empty direct delivery ID is rejected.
+The Gateway mapper sends this GUID as the wire event ID; the live alarm readback test
+requires that exact ID. SQL Server and QuestDB retain their existing event-ID schemes.
+
 `EnqueueAsync` does a single `INSERT … ON CONFLICT DO NOTHING` on the hot path.
 To avoid a `SELECT COUNT(*)` on every enqueue, the sink keeps an in-memory
 non-dead-lettered row counter (seeded at startup, kept current by every mutation,

docs/AlarmTracking.md

@@ -456,9 +456,15 @@ from the scripted `AlarmCommandRouter`:
    carrying the authored `FullName` as the `ConditionId` and the
    authenticated `OperatorUser`. The driver forwards this to the Galaxy
    gateway → AVEVA via `GatewayGalaxyAlarmAcknowledger`.
-   **Fire-and-forget** — a failed upstream ack is not surfaced back to the
-   OPC UA client (mirrors the Galaxy write-outcome limitation; the local
-   `AlarmConditionState` SDK update already committed at step 1).
+   The Galaxy adapter awaits the correlated `AcknowledgeAlarmAsync` RPC:
+   gateway protocol failures throw, while a non-zero native return code is
+   warning-logged. **The actor route back to OPC UA is fire-and-forget** —
+   `DriverInstanceActor` logs and drops failures without replying to the
+   OPC UA client, whose local `AlarmConditionState` SDK update already
+   committed at step 1. This is a native-ack routing limitation; Galaxy
+   value writes have a separate correlated write-outcome path. An awaited
+   ack RPC does not prove the ack reached AVEVA event history: the H6 live
+   verification remains [operator-gated](plans/2026-06-15-stillpending-phase-3-opcua-standards-design.md).
 
 Only the **Acknowledge** is routed to the driver. **A native condition
 supports Acknowledge and AddComment only** (Gitea #587 review): `Confirm`,

docs/Historian.md

@@ -573,7 +573,8 @@ node without `isHistorized` has no HistoryRead access bit and answers
 | Historized node, historian configured, time range empty | `GoodNoData` |
 | Historized node, historian NOT configured (`Enabled=false` / Null source) | `GoodNoData` (empty) for Raw / AtTime; Processed ⇒ `BadAggregateNotSupported` (#682) |
 | Non-historized node | `BadHistoryOperationUnsupported` |
-| Backend timeout or exception | `BadHistoryOperationUnsupported` per node; other nodes in the same batch are unaffected |
+| Backend transport outage (#697) | `BadNoCommunication` per node across Raw, Modified, Processed, AtTime and Events; cancelled/deadline-limited server reads retain `BadTimeout` |
+| Other backend exception / unsupported operation | `BadHistoryOperationUnsupported` per node; other nodes in the same batch are unaffected |
 | `Backend=SqlServer`, `AutoCreateSchema=false`, schema missing/behind | every read/write fails with a clear error naming the missing schema version (health `LastError`); reads answer `BadHistoryOperationUnsupported` per node; no host crash |
 
 A historized node with no historian configured never returns an error status — it returns
@@ -764,8 +765,11 @@ data.
   refusal of the remembered one. A **raised** ceiling (or a switch to streaming) is noticed by the only reads it matters to: when a
   read bounded by the remembered ceiling comes back full without proving completeness, it is re-probed **once** at its full budget
   — served ⇒ the ceiling is forgotten; refused naming a larger ceiling ⇒ that one is learned and the read re-issued at it; refused
-  naming the same ceiling ⇒ the bounded result stands. No timer: a TTL would only add refusals for reads that already fit; the cost
-  of the rule is one refused re-probe per genuinely over-ceiling read (which fails either way). So an uncapped read (`NumValuesPerNode = 0`, which used to be refused outright on a strict gateway, even for a 10-row
+  naming the same ceiling ⇒ the bounded result stands. Raw and AtTime re-probe immediately. Per-sample aggregates (#696) reuse
+  a refusal for **30 seconds** before re-probing a full stream; within that period an oversized window still fails honestly with
+  `BadTooManyOperations`. A lowered ceiling is still re-learned immediately; a raise becomes visible on the next incomplete
+  aggregate read after the cooldown. So an uncapped read (`NumValuesPerNode = 0`, which used to be refused outright on a strict
+  gateway, even for a 10-row
   window) and a page size within 16 of the ceiling are now served when the window fits. A ceiling-bounded read is answered only
   when it is **provably complete** — the stream came back short of the ceiling, or reached `EndTime` (or, capped, filled the page);
   a stream that delivered the whole ceiling short of `EndTime` may be truncated and fails **`BadTooManyOperations`** (not a backend
@@ -823,6 +827,21 @@ data.
   it could only ever be a Part 11 *interpolated* bound. The start bound on the Gateway is the real prior sample, found by the
   #686 walk-back.
 
+**Warm crowding ceiling (#695).** A raw crowding re-read spends the remaining room below a learned gateway ceiling before
+refusing; doubling the look-back allowance does not skip that final usable budget. Reverse paging retries a crowded segment
+with a smaller in-window budget derived from the learned ceiling and the observed pre-window rows.
+
+**Missing tags (#653).** Gateway Raw, Processed and AtTime first check the tag catalog. An empty exact-name catalog result or an explicit
+`NotFound` returns no data without a history RPC: Raw is empty; AtTime and Processed return null/`BadNoData` at the
+requested times or bucket timestamps. Unsupported aggregate validation still runs first. A bounded per-data-source cache
+holds at most 4,096 tag names, with **5 seconds** for absence and **60 seconds** for presence, so newly provisioned tags become
+visible after the short negative expiry. Catalog faults and cancellation never become missing-tag cache entries. The adapter drains the filtered catalog stream and checks exact names, avoiding older gateways that misparse a missing `GetTagMetadata` response as `Internal`. Empty
+AtTime/Processed requests need no check. Events and modified-history capability checks are unaffected.
+
+**Stopped-tag reverse discovery (#696).** On a strict Gateway, a truncated remainder probe retains its oldest-row evidence
+and density. Gap discovery uses the learned ceiling's available budget; ordinary page reads keep their page-sized budget.
+Only complete segments contribute to a page, preserving newest-first order, bounds, ties and continuation hints.
+
 ### Gateway AtTime is evaluated from raw reads (#693)
 
 > ⚠️ **Client-visible (#693).** Gateway AtTime used to pass AVEVA's own interpolated answer (the gateway's `ReadAtTime`) through,
@@ -880,9 +899,14 @@ data.
   here — the SQL backends store DateTime as its own kind and return a `DateTime`. Under an integer hint a NaN, ±∞ or
   out-of-Int64-range stored value is no integer: it is treated as unusable (as `BadNoData`), so a neighbour like that is never
   interpolated or saturated into a Good value.
-- **What differs between backends (by design, not changed by #693).** Only the **Gateway** converts an AtTime result to the node's
-  integer type (`Int32` ⇒ `int`): the SQL backends store and return integer nodes as `Int64` (and Boolean as Boolean). Gateway
-  **Raw** reads still return a Boolean node's samples as the stored Double 0/1 (the hint is passed only on AtTime).
+- **Client value types (#698, all backends).** Before returning Raw, Modified or AtTime history, the node manager converts non-null
+  scalar values to the node's declared DataType (including Boolean, integer widths and Float). Arrays retain their stored values.
+  Conversion failures produce a null value with `BadTypeMismatch` for that sample, preserving timestamps and modification metadata.
+  DateTime requires an actual DateTime value: the gateway's numeric DateTime encoding is unknown and cannot be inferred.
+  Processed results retain each aggregate's result type, so Count and averages can remain Double on a Boolean or integer node.
+- **Future timestamps (#696).** A bounded raw probe checks for stored samples between now and the requested future time. An empty
+  future interval starts the prior-sample search near now; a full probe seeds the existing walk-back with a stored future sample.
+  The result still carries the requested timestamp and the evaluator's Uncertain status when no later sample exists.
 - ⚠️ **Shared-evaluator fix (all backends that use it):** `AtTimeEvaluator` returned every interpolated `Int64` boxed as a
   **Double** (a C# `cond ? (long)x : double` conditional widens the long); its "keeping the Int64 type" tests passed because
   Shouldly's `ShouldBe` compares numbers across types. SqlServer / QuestDB interpolated `Int64` AtTime values are now `Int64`,
@@ -1764,7 +1788,7 @@ this family and **are** capped (they used to be pushed down). Raise the cap, or
 ### Health
 
 The read client reports `GetHealthSnapshot()` (query/failure counters, consecutive failures, `LastError` — e.g. a missing schema under
-`AutoCreateSchema=false`). A SQL error on a read surfaces as `BadHistoryOperationUnsupported` for **that node only**; a cancelled read
+`AutoCreateSchema=false`). A SQL connection outage on a read surfaces as `BadNoCommunication`; other SQL errors retain `BadHistoryOperationUnsupported` for **that node only**; a cancelled read
 (deadline, client abort) is not counted as a backend fault.
 
 ### Retention (`RetentionDays`)

docs/ScriptedAlarms.md

@@ -325,8 +325,10 @@ routes through a separate `NativeAlarmAckRouter` seam (instead of the scripted
 FullName)` inverse map, Primary-gated) → the owning driver's
 `IAlarmSource.AcknowledgeAsync` → Galaxy gateway → AVEVA. The call carries the
 authenticated operator's display name via the `AlarmAcknowledgeRequest.OperatorUser`
-field. This is fire-and-forget — a failed upstream ack is not surfaced back to the
-OPC UA client (mirrors the Galaxy write-outcome limitation). See [AlarmTracking.md
+field. The actor route back to OPC UA is fire-and-forget — a failed upstream ack
+is logged rather than surfaced back to the OPC UA client. The Galaxy adapter
+awaits its correlated RPC; this does not establish delivery to AVEVA event history.
+Galaxy value writes use a separate correlated write-outcome path. See [AlarmTracking.md
 §Native alarm acknowledge → AVEVA](AlarmTracking.md#native-alarm-acknowledge--aveva)
 for the full routing diagram.
 

docs/plans/2026-06-15-stillpending-phase-3-opcua-standards-design.md

@@ -120,7 +120,9 @@ OPC UA client Acknowledge on a NATIVE condition
   under Lock, like `AlarmCommandRouter`).
 - **Driver faulted / reconnecting** → `AcknowledgeAsync` fails or no-ops per the driver; the local condition
   state still reflects the SDK-applied ack (the OPC UA client sees Good). Surfacing a *failed* upstream ack
-  back to the client is out of scope (mirrors the Galaxy write-outcome fire-and-forget limitation).
+  back to the client is out of scope for this native-ack actor route. **Current clarification
+  (2026-10-13):** the Galaxy adapter awaits a correlated ack RPC, but the actor route does not
+  return its outcome to OPC UA; Galaxy value writes have a separate correlated write-outcome path.
 - **H4 native Enable/Disable** → `BadNotSupported` (decision #2).
 - **H2-bit:** no runtime behavior change for existing grants except closing the HistoryRead⇒HistoryUpdate
   authorization hole; the service itself stays rejected.

docs/plans/2026-10-13-alarm-delivery-identity-results.md

@@ -0,0 +1,9 @@
+# Alarm delivery identity results — lmxopcua #705
+
+Alarm events now carry a nonzero delivery GUID created once for direct delivery and preserved across retries and record copies. It is excluded from queue JSON and business-field equality/hash. Queued delivery overrides it with a frozen versioned UUID encoding derived only from the persisted queue row key, preserving existing payload JSON, fingerprints, schema, deduplication and AlarmId. The Gateway mapper sends that GUID and rejects explicitly empty identities. SQL Server and QuestDB identity logic is unchanged.
+
+Coverage includes literal historical JSON/fingerprint/GUID vectors; ten-field equality and hash behavior; direct copies/retries; queue retry/reopen/payload edits/dead-letter revival; identical-input convergence; distinct same-time business events; real replicated queue convergence; and legacy migration. Reconstructed direct objects receive new identities; durable reconstruction uses the queue. Completely identical business payloads remain intentionally deduplicated.
+
+TDD: direct wire regression failed on an empty ID; queued retry/reopen regression failed on changing IDs. Final affected runs: Core51/51, Host migration/convergence16/16, Gateway453passed/6live skipped/0failed. Full offline solution:14,635passed/611skipped/0failed across62suites, exit0. Independent final review and diffcheck clear. Fixture configuration variables were removed before all runs.
+
+No live event writes, deployed sender rollout or historical row rewrite occurred. The live round-trip test now requires exact delivery GUID readback; actual persistence remains the open Gateway #28 gate, alongside its separate #39 timestamp-offset provider failure. Upstream History controls #29 and lmxopcua #694/#700 remain open.

docs/plans/2026-10-13-cache-concurrency-results.md

@@ -0,0 +1,9 @@
+# Gateway tag-presence concurrency coverage — 2026-10-13
+
+Approved tests-only fix for #701. Added 15 deterministic cases for overlapping positive/negative cold lookups, expiry refreshes, capacity crossing, successful-result isolation from faults/cancellation, and cancellation after asynchronous completion. Fake presence calls support an async hook and atomic counter. Production cache is unchanged; no single-flight promise.
+
+The synthetic 16-caller cold/expiry bursts produced 16 catalog RPCs. Assertions permit one through16; this records amplification without making it a requirement or a live-cost claim. Newclass15/15 and Gateway449/449 passed. Independent review found no actionable defects; diff check clean.
+
+First full solution run:14,614passed/609skipped/1failed across62suites. Existing restart/reconnect wire test failed, then isolated retry passed1/1. Full repeat with detailed console logging passed14,615/609skip/0fail across62suites, exit0, including OPC UA integration122/122 and host integration268passed/12skip. #702 tracks the unresolved intermittent failure; it is not closed merely because the repeat passed.
+
+Both full runs excluded configured live fixture variables from the environment; environment-gated tests skipped. Existing NU1510/xUnit duplicate-InlineData warnings remain pre-existing. No production behavior or deployment changed by this patch.

docs/plans/2026-10-13-historian-653-696-results.md

@@ -0,0 +1,36 @@
+# Missing-tag and Gateway performance fixes — 2026-10-13
+
+Follow-up to pushed `97668d26` / PR #699. Implements #653 and the remaining #696 items 2/3 under the approved bounded design.
+
+## Changes
+
+- **#653:** Raw, Processed and AtTime check tag presence before history RPCs. The bounded 4,096-name cache uses a 5-second negative and 60-second positive lifetime. Only explicit absence or NotFound is cached as missing; failures and cancellation propagate. Raw is empty; AtTime and Processed retain requested/bucket timestamps with null/BadNoData. Unsupported aggregate validation runs first.
+- **Live compatibility adjustment:** the deployed gateway misparses missing `GetTagMetadata` replies as `Internal` (`InvalidDataException: GetTagInfo response ended unexpectedly`). The adapter therefore checks the filtered tag catalog and compares exact names, fully draining the stream before declaring presence/absence. It never turns the parser error into a missing tag. Wildcard matches alone do not prove presence, and a trailing stream fault cannot enter the cache.
+- **#696 item 2:** a truncated remainder probe seeds strict-ceiling reverse discovery with its observed rows/density; gap searches use the available discovery budget. Complete segments alone contribute to the newest-first page. Regression: a 7,200-row stopped 1 Hz tag, ceiling 1,000, requested latest three one day later, previously 17 raw calls; now meets a maximum of 12 with exactly 7199/7198/7197 returned. Ordinary page sizing and event discovery behavior are unchanged.
+- **#696 item 3:** per-sample aggregates reuse a ceiling refusal for 30 seconds instead of immediately re-probing on every full stream. The next incomplete aggregate read after expiry notices a raised ceiling; a lowered ceiling is re-learned immediately. Raw and AtTime re-probes remain immediate. During the cooldown an oversized aggregate remains BadTooManyOperations, never a truncated result.
+
+## Verification evidence
+
+New regressions failed before fixes: all initial 12 cases failed (missing preflight/caching, 17 discovery reads, duplicate aggregate refusal). The four catalog wire cases failed with the metadata adapter and pass with the catalog adapter. Final Gateway suite: 434 passed, six environment-gated skips.
+
+Independent read-only review found no important defects. Concurrent cold requests can duplicate a metadata/catalog check; cache access and capacity mutations are locked. No single-flight guarantee was introduced. Catalog RPC contract streams all matches; the upstream SDK pages the native tag-query results, with no configured browse result ceiling.
+
+## Live #653 confirmation
+
+Measured the updated GatewayHistorianDataSource through the real adapter and local gateway against the existing remote AVEVA connection. Temporary historian:read key had a two-hour expiry and was revoked after the run. No historian writes/provisioning and no gateway deployment changes.
+
+Nonexistent tag: `OtOpcUa.Issue653.Nonexistent.20261013`.
+
+| Operation, five consecutive runs | Elapsed milliseconds | Result |
+|---|---|---|
+| Raw | 228, 0, 0, 0, 0 | empty |
+| Count processed | 2, 0, 0, 0, 0 | six null/BadNoData buckets |
+| AtTime | 1, 0, 0, 0, 0 | null/BadNoData at requested time |
+
+These used one data source/cache: the first Raw call is cold, subsequent calls are cached. The prior direct-client measurement took ~30.6 seconds for missing-tag Raw/Aggregate and ~61.3 seconds for AtTime. Direct client RPCs are not modified; the fast-fail protection is in OtOpcUa.
+
+**Control limitation:** SysTimeSec catalog presence succeeded, but its Raw RPC twice timed out upstream with DeadlineExceeded at ~30 seconds. The data source propagated IOException/communication failure rather than returning no data. A fresh control using the original package client (no data-source presence checks) also timed out with DeadlineExceeded after 30,656 ms, confirming this timeout is independent of the new checks. The earlier direct-client control had succeeded. This run therefore verifies missing-tag fast-fail and failure propagation, not healthy-tag live history behavior. No end-to-end deployed OPC UA claim is made.
+
+#694 remains open for native aggregate rerun after historiangw#27. No sibling gateway code changed.
+
+Full final-tree solution: 14,600 passed, 609 skipped, 0 failed across 62 suites. `dotnet test ZB.MOM.WW.OtOpcUa.slnx --verbosity quiet`, exit 0. Includes 122 passing OPC UA integration tests and 268 passing host integration tests (12 host skips). Environment-gated suites are included in the skip count.

docs/plans/2026-10-13-historian-695-698-results.md

@@ -0,0 +1,28 @@
+# Historian follow-up results — 2026-10-13
+
+Scope: approved bounded fixes for #695–#698, plus live measurement of #653. Base: `66a2b0c3`.
+
+- #695: spend the remaining learned-ceiling room on crowding reads; reduce crowded reverse-read windows using the observed pre-window count. Regression cases cover ceilings 1,709 and 1,369, and reverse paging through a crowded millisecond.
+- #696 item 1: probe the future interval before starting the prior-sample search near now. Preserve stored future samples, requested timestamps and Uncertain status. Dense future probes still use the existing bounded walk. Items 2 and 3 remain open (stopped-tag strict-ceiling cost and aggregate ceiling re-probes).
+- #697: actual Gateway, SQL Server and QuestDB transport failures become `BadNoCommunication` on every HistoryRead arm. Cancellation remains `BadTimeout`; unsupported operations retain `BadHistoryOperationUnsupported`. Tests include real refused database connections, repeated reads and nested gRPC failures.
+- #698: normalize Raw, Modified and AtTime scalar values to the node DataType. Conversion failure gives per-sample null/`BadTypeMismatch`; arrays, audit metadata and timestamps survive. Processed results retain aggregate-specific types.
+- #694 item 3 remains blocked on historiangw#27. No upstream code was changed.
+
+## Live #653 measurement
+
+Five sequential calls per operation and tag, using client 0.3.0 against the running local gateway (`localhost:5221`) with its existing remote AVEVA connection. Commissioned a two-hour `historian:read` key, then revoked it after measurement. No historian writes or provisioning performed.
+
+| Operation | SysTimeSec, elapsed ms | Nonexistent tag, elapsed ms |
+|---|---|---|
+| Metadata | 299, 85, 81, 82, 48 (success) | 89, 53, 41, 77, 76 (gateway exception) |
+| Raw | 143, 99, 145, 133, 159 (success) | 30260, 30634, 30675, 30754, 30628 (gateway exception) |
+| Aggregate | 271, 275, 309, 249, 304 (success) | 30564, 30605, 30583, 30575, 30806 (gateway exception) |
+| AtTime | 140, 88, 89, 127, 92 (success) | 61074, 61330, 61601, 61249, 61271 (gateway exception) |
+
+Nonexistent tag: `OtOpcUa.Issue653.Nonexistent.20261013`. The delay is still present. Recommend a metadata pre-check with a short negative cache, caching only explicit missing-tag outcomes and never authentication, communication or arbitrary errors. This is follow-up implementation; #653 remains open. Direct gateway AtTime above differs from OtOpcUa's raw-based AtTime evaluator.
+
+## Consumer and validation notes
+
+ScadaBridge source and tests contain no HistoryRead consumer; no consumer change required. Independent review found a future-stored-sample edge case, which was fixed and covered before final verification. No deployment or live end-to-end OPC UA verification of these new fixes is claimed; the live run measures #653 through the gateway client.
+
+Full solution: 14,581 passed, 609 skipped, 0 failed across 62 test suites. Command: `dotnet test ZB.MOM.WW.OtOpcUa.slnx --verbosity quiet`, exit 0. Skips include environment-gated integration suites.

docs/plans/2026-10-13-restart-loss-reason-results.md

@@ -0,0 +1,9 @@
+# Restart loss reason test correction — #702
+
+The full integration reproduction failed at the restart wire test reason assertion: it required BadServerHalted but correctly received BadSessionIdInvalid. Production already supports that definitive loss reason.
+
+The test now accepts an explicit semantic set: four direct definitive session statuses and two sustained closed-channel statuses. All BadNoCommunication, retained-value, unhealthy-state, frozen-read-stamp, one-notification and explicit reinitialization/recovery assertions remain unchanged. Only test expectation/comments changed; production unchanged. Existing deterministic driver tests independently enforce direct-status ordering/count and the sustained-channel threshold.
+
+Verification: isolated restart1/1, deterministic status routes30/30, full OPC UA integration122/122. Full offline solution: 14,613 passed / 611 skipped / 0 failed, exit0, 61 reported suite summaries. Host integration266 passed /14 skipped; unavailable environment/fixture tests remain skips. Independent review found no important defects; git diff --check passed. Actual prior red run121pass/1fail, detailed console/TRX confirms the exact reason mismatch.
+
+Logs/TRX retained in the Codex thread workspace under work/issue702-*. No Gateway or deployment behavior changed by this test correction. Separate live gates remain open.

src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmDeliveryIds.cs

@@ -0,0 +1,25 @@
+using System.Security.Cryptography;
+using System.Text;
+
+namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
+
+/// <summary>Stable delivery GUIDs for persisted alarm queue rows.</summary>
+public static class AlarmDeliveryIds
+{
+    /// <summary>
+    /// Converts an existing durable row key into a nonzero version-8 GUID. Version 1 of this encoding hashes
+    /// UTF-8 "OtOpcUa.AlarmDelivery/v1\0" followed by the exact persisted key with SHA-256, takes the first
+    /// 16 bytes, sets the UUID version/variant bits, and interprets the bytes in network order.
+    /// This encoding must remain stable across retries, upgrades and replicated failover.
+    /// </summary>
+    /// <param name="queueId">The persisted row key, never a hash recomputed from the current payload.</param>
+    /// <returns>The row's stable delivery identity.</returns>
+    public static Guid FromQueueId(string queueId)
+    {
+        ArgumentException.ThrowIfNullOrWhiteSpace(queueId);
+        var digest = SHA256.HashData(Encoding.UTF8.GetBytes("OtOpcUa.AlarmDelivery/v1\0" + queueId));
+        digest[6] = (byte)((digest[6] & 0x0F) | 0x80);
+        digest[8] = (byte)((digest[8] & 0x3F) | 0x80);
+        return new Guid(digest.AsSpan(0, 16), bigEndian: true);
+    }
+}

src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmHistorianEvent.cs

@@ -1,3 +1,4 @@
+using System.Text.Json.Serialization;
 using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
 
 namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
@@ -34,4 +35,31 @@ public sealed record AlarmHistorianEvent(
     string Message,
     string User,
     string? Comment,
-    DateTime TimestampUtc);
+    DateTime TimestampUtc)
+{
+    /// <summary>
+    /// Delivery identity, minted once for direct sends and replaced by the durable queue's identity when drained.
+    /// Excluded from payload serialization and business equality so redundant enqueues still converge.
+    /// A record copy preserves this identity; an explicitly empty identity is invalid for delivery.
+    /// </summary>
+    [JsonIgnore]
+    public Guid DeliveryId { get; init; } = Guid.NewGuid();
+
+    /// <summary>Compares the ten business fields, excluding transport delivery identity.</summary>
+    /// <param name="other">The event to compare.</param>
+    /// <returns>Whether the business fields are equal.</returns>
+    public bool Equals(AlarmHistorianEvent? other) => ReferenceEquals(this, other) || other is not null
+        && AlarmId == other.AlarmId && EquipmentPath == other.EquipmentPath && AlarmName == other.AlarmName
+        && AlarmTypeName == other.AlarmTypeName && Severity == other.Severity && EventKind == other.EventKind
+        && Message == other.Message && User == other.User && Comment == other.Comment && TimestampUtc == other.TimestampUtc;
+
+    /// <summary>Hashes the same business fields used by equality, excluding delivery identity.</summary>
+    /// <returns>The business-field hash code.</returns>
+    public override int GetHashCode()
+    {
+        var hash = new HashCode();
+        hash.Add(AlarmId); hash.Add(EquipmentPath); hash.Add(AlarmName); hash.Add(AlarmTypeName); hash.Add(Severity);
+        hash.Add(EventKind); hash.Add(Message); hash.Add(User); hash.Add(Comment); hash.Add(TimestampUtc);
+        return hash.ToHashCode();
+    }
+}

src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/LocalDbStoreAndForwardSink.cs

@@ -603,6 +603,9 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl
             try
             {
                 evt = JsonSerializer.Deserialize<AlarmHistorianEvent>(payload);
+                // The durable key survives retry, payload edits, legacy migration and replicated failover.
+                // Never derive delivery identity from a reserialized payload or an individual RPC attempt.
+                if (evt is not null) evt = evt with { DeliveryId = AlarmDeliveryIds.FromQueueId(id) };
             }
             catch (JsonException)
             {

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayAtTimeReader.cs

@@ -227,7 +227,27 @@ internal static class GatewayAtTimeReader
                 {
                     if (!beforeKnown)
                     {
-                        before = await WalkBackAsync(ctx, qs, walkKnown, walkWidth).ConfigureAwait(false);
+                        // #696: prove the future stretch empty before jumping to now. Historian clocks or imports
+                        // can place real rows there; a full bounded probe seeds the original backward search.
+                        var searchEnd = qs;
+                        if (qs > now && walkKnown is null)
+                        {
+                            var future = await ReadTrustedAsync(ctx, MsFloor(now),
+                                Min(MsFloor(qs) + OneMs, horizon), ReadRows).ConfigureAwait(false);
+                            StoredSample? latest = null;
+                            foreach (var row in future.Rows)
+                                if (row.TimeUtc <= qs) latest = row;
+                            if (future.Complete)
+                            {
+                                before = latest;
+                                beforeKnown = latest is not null;
+                                searchEnd = now;
+                            }
+                            else if (latest is { } found)
+                                walkKnown = found.TimeUtc;
+                        }
+                        if (!beforeKnown)
+                            before = await WalkBackAsync(ctx, searchEnd, walkKnown, walkWidth).ConfigureAwait(false);
                         beforeKnown = true;
                     }
                     prev = before;

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs

@@ -68,7 +68,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
     internal int MaxPreWindowRows => _maxPreWindowRows;
 
     /// <summary>A strict gateway's row ceiling, learned from its first refusal (#691.3).</summary>
-    private readonly GatewayRowCeiling _ceiling = new();
+    private readonly GatewayRowCeiling _ceiling;
+    private readonly GatewayTagPresenceCache _tags;
 
     private readonly object _healthLock = new();
     private DateTime? _lastSuccessUtc;
@@ -107,6 +108,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
         _logger = logger;
         _maxRawRows = maxRawRows;
         _time = timeProvider ?? TimeProvider.System;
+        _ceiling = new GatewayRowCeiling(_time);
+        _tags = new GatewayTagPresenceCache(_client, _time);
     }
 
     /// <inheritdoc />
@@ -126,6 +129,11 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
 
         try
         {
+            if (!await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
+            {
+                RecordOutcome(success: true, error: null);
+                return new HistoryReadResult([], ContinuationPoint: null);
+            }
             var samples = await ReadRawWindowAsync(fullReference, startUtc, endUtc, maxValuesPerNode, cancellationToken)
                 .ConfigureAwait(false);
             var snapshots = SampleMapper.ToSnapshots(samples);
@@ -152,6 +160,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
         catch (Exception ex)
         {
             RecordReadFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
             throw;
         }
     }
@@ -239,7 +248,15 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
         for (var attempt = 0; attempt < MaxCrowdedReReads && preWindow <= _maxPreWindowRows; attempt++)
         {
             var budget = (int)Math.Min(int.MaxValue, (long)cap + room + RawAggregateReader.PreWindowAllowance);
-            if (_ceiling.Learned is var known and > 0 && budget > known) break; // over a strict ceiling: refused anyway
+            // #695: doubling can leap over the ceiling while the cluster still fits. Spend the remaining room once.
+            if (_ceiling.Learned is var known and > 0 && budget > known)
+            {
+                var available = known - cap - RawAggregateReader.PreWindowAllowance;
+                if (available <= 0) break;
+                room = Math.Min(room, available);
+                budget = cap + room + RawAggregateReader.PreWindowAllowance;
+            }
+            if (budget <= pass.Budget) break; // no more room, never retry the same full read
             pass = await PassAsync(budget).ConfigureAwait(false);
             if (pass.Result.Complete) return pass.Result.Samples;
             preWindow = Math.Max(preWindow, pass.Result.PreWindowRows);
@@ -342,6 +359,11 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
         if (plan.BucketCount == 0) return new HistoryReadResult([], ContinuationPoint: null);
         try
         {
+            if (!await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
+            {
+                RecordOutcome(success: true, error: null);
+                return new HistoryReadResult(plan.Assemble(new Dictionary<long, DataValueSnapshot>(), t => AggregateEvaluator.NoData(t, _time.GetUtcNow().UtcDateTime)), ContinuationPoint: null);
+            }
             var byBucket = new Dictionary<long, DataValueSnapshot>();
             foreach (var request in NativeAggregatePlan.Requests(plan))
             {
@@ -365,6 +387,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
         catch (Exception ex)
         {
             RecordReadFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
             throw;
         }
     }
@@ -380,6 +403,11 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
         try
         {
             var now = _time.GetUtcNow().UtcDateTime;
+            if (plan.BucketCount > 0 && !await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
+            {
+                RecordOutcome(success: true, error: null);
+                return new HistoryReadResult(plan.Assemble(new Dictionary<long, DataValueSnapshot>(), t => AggregateEvaluator.NoData(t, now)), ContinuationPoint: null);
+            }
             var values = await RawAggregateReader
                 .ReadAsync(_client, _ceiling, fullReference, aggregate, plan, _maxRawRows, now, cancellationToken)
                 .ConfigureAwait(false);
@@ -408,6 +436,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
         catch (Exception ex)
         {
             RecordReadFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
             throw;
         }
     }
@@ -449,6 +478,11 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
         try
         {
             var now = _time.GetUtcNow().UtcDateTime;
+            if (timestampsUtc.Count > 0 && !await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
+            {
+                RecordOutcome(success: true, error: null);
+                return MissingAtTime(timestampsUtc, now);
+            }
             var values = await GatewayAtTimeReader
                 .ReadAsync(_client, _ceiling, fullReference, timestampsUtc, now, cancellationToken, valueType)
                 .ConfigureAwait(false);
@@ -466,6 +500,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
         catch (Exception ex)
         {
             RecordReadFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
             throw;
         }
     }
@@ -476,6 +511,12 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
     {
         try
         {
+            if (timestampsUtc.Count > 0 && !await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
+            {
+                RecordOutcome(success: true, error: null);
+                return MissingAtTime(timestampsUtc, _time.GetUtcNow().UtcDateTime);
+            }
+
             var samples = await _client
                 .ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken)
                 .ConfigureAwait(false);
@@ -486,6 +527,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
         catch (Exception ex)
         {
             RecordReadFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
             throw;
         }
     }
@@ -581,6 +623,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
         catch (Exception ex)
         {
             RecordReadFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
             throw;
         }
     }
@@ -655,6 +698,19 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
         return result;
     }
 
+    /// <summary>No-data values for a missing tag, stamped with the client's requested times.</summary>
+    private static HistoryReadResult MissingAtTime(IReadOnlyList<DateTime> times, DateTime now) =>
+        new(times.Select(t => AtTimeEvaluator.Evaluate(t, null, null, now, now)).ToArray(), ContinuationPoint: null);
+
+    /// <summary>Classifies actual gRPC/package transport failures without treating refusals as outages (#697).</summary>
+    private static bool IsTransportFailure(Exception ex) => ex switch
+    {
+        ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayUnavailableException => true,
+        RpcException rpc => rpc.StatusCode is Grpc.Core.StatusCode.Unavailable or Grpc.Core.StatusCode.DeadlineExceeded,
+        TimeoutException => true,
+        _ => ex.InnerException is { } inner && IsTransportFailure(inner),
+    };
+
     /// <summary>
     ///     Records a failed read: bumps the health counters and logs a generic, redaction-safe
     ///     debug line (no tag, host, or value). The exception itself is rethrown by the caller.

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayReadLimits.cs

@@ -76,13 +76,15 @@ internal static class GatewayReadLimits
 ///         the read is re-issued at <c>N</c>; refused naming <c>C</c> again ⇒ the <c>C</c>-bounded result stands (no repeat).
 ///     </para>
 ///     <para>
-///         <b>Why no TTL.</b> A raise only matters to a read that needs more than <c>C</c>, and such a read re-probes anyway
-///         (rule 3), so a timer would only add refusals for reads that fit. The cost of the rule is one refused re-probe per
-///         genuinely over-ceiling read — a read that fails <c>BadTooManyOperations</c> whatever is done.
+///         <b>Re-probe timing (#696).</b> Raw and AtTime reads re-probe incomplete streams immediately. Per-sample
+///         aggregate reads opt into a 30-second cooldown after a refusal; a full stream during it remains a request
+///         limit, never a truncated answer. After the cooldown the next incomplete read detects a raised ceiling.
 ///     </para>
 /// </remarks>
-internal sealed class GatewayRowCeiling
+internal sealed class GatewayRowCeiling(TimeProvider? timeProvider = null)
 {
+    private readonly TimeProvider _time = timeProvider ?? TimeProvider.System;
+    private long _lastRefusalTicks;
     private int _learned; // 0 = unknown / not strict
 
     /// <summary>The learned ceiling (0 = none).</summary>
@@ -93,8 +95,9 @@ internal sealed class GatewayRowCeiling
     /// <param name="budget">The <c>max_values</c> the read wants.</param>
     /// <param name="pass">One drained read with the given budget.</param>
     /// <param name="needsMore">True when a pass with the given budget delivered it all without proving completeness.</param>
+    /// <param name="deferRecentReprobe">Aggregate-only opt-in: reuse a refusal for 30 seconds before re-probing an incomplete read.</param>
     /// <returns>The result and the budget it was read with.</returns>
-    public async Task<(T Result, int Budget)> ReadAsync<T>(int budget, Func<int, Task<T>> pass, Func<T, int, bool> needsMore)
+    public async Task<(T Result, int Budget)> ReadAsync<T>(int budget, Func<int, Task<T>> pass, Func<T, int, bool> needsMore, bool deferRecentReprobe = false)
     {
         var learned = Learned;
         if (learned <= 0 || learned >= budget) return await AttemptAsync(budget, pass).ConfigureAwait(false);
@@ -102,6 +105,10 @@ internal sealed class GatewayRowCeiling
         var bounded = await AttemptAsync(learned, pass).ConfigureAwait(false);
         if (bounded.Budget < learned || !needsMore(bounded.Result, bounded.Budget)) return bounded;
 
+        // #696: aggregate reads may reuse a recent refusal for 30 seconds. Raw paging keeps immediate re-probes.
+        if (deferRecentReprobe && _time.GetUtcNow().UtcTicks - Volatile.Read(ref _lastRefusalTicks) < TimeSpan.FromSeconds(30).Ticks)
+            return bounded;
+
         // Rule 3: full at a possibly stale ceiling — re-probe once at the full budget.
         int fresh;
         try
@@ -114,10 +121,16 @@ internal sealed class GatewayRowCeiling
         {
             // fall through (outside the catch)
         }
-        Volatile.Write(ref _learned, fresh);
+        Learn(fresh);
         return fresh > learned ? (await pass(fresh).ConfigureAwait(false), fresh) : bounded;
     }
 
+    private void Learn(int ceiling)
+    {
+        Volatile.Write(ref _lastRefusalTicks, _time.GetUtcNow().UtcTicks);
+        Volatile.Write(ref _learned, ceiling);
+    }
+
     private async Task<(T Result, int Budget)> AttemptAsync<T>(int budget, Func<int, Task<T>> pass)
     {
         int ceiling;
@@ -129,7 +142,7 @@ internal sealed class GatewayRowCeiling
         {
             // fall through to the retry (outside the catch, so a second refusal is not nested in this one)
         }
-        Volatile.Write(ref _learned, ceiling);
+        Learn(ceiling);
         return (await pass(ceiling).ConfigureAwait(false), ceiling);
     }
 }

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayTagPresenceCache.cs

@@ -0,0 +1,36 @@
+using Grpc.Core;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
+
+/// <summary>Bounded metadata cache (#653). Missing tags expire quickly; failures never become missing tags.</summary>
+internal sealed class GatewayTagPresenceCache(IHistorianGatewayClient client, TimeProvider time)
+{
+    internal const int Capacity = 4096;
+    private readonly object _sync = new();
+    private readonly Dictionary<string, (bool Exists, DateTimeOffset Expires)> _entries = new(StringComparer.Ordinal);
+
+    public async Task<bool> ExistsAsync(string tag, CancellationToken ct)
+    {
+        ct.ThrowIfCancellationRequested();
+        lock (_sync)
+            if (_entries.TryGetValue(tag, out var entry) && time.GetUtcNow() < entry.Expires) return entry.Exists;
+
+        bool exists;
+        try { exists = await client.TagExistsAsync(tag, ct).ConfigureAwait(false); }
+        catch (Exception ex) when (IsNotFound(ex)) { exists = false; }
+        ct.ThrowIfCancellationRequested();
+        lock (_sync)
+        {
+            if (!_entries.ContainsKey(tag) && _entries.Count >= Capacity) _entries.Remove(_entries.Keys.First());
+            _entries[tag] = (exists, time.GetUtcNow().AddSeconds(exists ? 60 : 5));
+        }
+        return exists;
+    }
+
+    private static bool IsNotFound(Exception ex)
+    {
+        for (Exception? e = ex; e is not null; e = e.InnerException)
+            if (e is RpcException { StatusCode: StatusCode.NotFound }) return true;
+        return false;
+    }
+}

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/HistorianGatewayClientAdapter.cs

@@ -81,6 +81,17 @@ public sealed class HistorianGatewayClientAdapter : IHistorianGatewayClient, IDi
         return new HistorianGatewayClientAdapter(HistorianGatewayClient.Create(clientOptions));
     }
 
+    /// <inheritdoc />
+    public async Task<bool> TagExistsAsync(string tag, CancellationToken ct)
+    {
+        // #653: older gateway SDKs misparse a missing GetTagInfo response as Internal. A catalog query proves
+        // absence without guessing from an error. Even a glob-like tag is checked by exact name, never by row count.
+        var found = false;
+        await foreach (var name in _inner.BrowseTagNamesAsync(tag, ct).ConfigureAwait(false))
+            found |= string.Equals(name, tag, StringComparison.OrdinalIgnoreCase);
+        return found; // drain first: a trailing stream fault must propagate and must not become a cache entry
+    }
+
     /// <inheritdoc />
     public IAsyncEnumerable<HistorianSample> ReadRawAsync(
         string tag, DateTime startUtc, DateTime endUtc, int maxValues, CancellationToken ct) =>

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/IHistorianGatewayClient.cs

@@ -11,6 +11,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
 /// </summary>
 public interface IHistorianGatewayClient : IAsyncDisposable
 {
+    /// <summary>Checks the tag catalog; false means explicitly absent, while failures propagate.</summary>
+    /// <param name="tag">Historian tag name.</param>
+    /// <param name="ct">Cancellation token.</param>
+    /// <returns>Whether the historian contains the tag.</returns>
+    Task<bool> TagExistsAsync(string tag, CancellationToken ct);
+
     /// <summary>Streams raw historian samples for a tag over a time window.</summary>
     /// <param name="tag">The historian tag name to read.</param>
     /// <param name="startUtc">Inclusive start of the requested time window (UTC).</param>

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/AlarmEventMapper.cs

@@ -35,16 +35,16 @@ internal static class AlarmEventMapper
     /// <returns>The gateway wire event ready for <c>SendEvent</c>.</returns>
     public static HistorianEvent ToHistorianEvent(AlarmHistorianEvent alarm)
     {
+        if (alarm.DeliveryId == Guid.Empty)
+            throw new ArgumentException("Alarm delivery identity must be nonzero.", nameof(alarm));
         // Timestamp.FromDateTime requires a Utc-kind DateTime; coerce defensively (TimestampUtc is
         // already Utc by contract, but a caller could pass Unspecified).
         var eventTime = Timestamp.FromDateTime(DateTime.SpecifyKind(alarm.TimestampUtc, DateTimeKind.Utc));
 
         var historianEvent = new HistorianEvent
         {
-            // Deliberately DO NOT set HistorianEvent.Id: the gateway's SendEvent path rejects a
-            // client-supplied event id (the server handler throws and the call fails permanently —
-            // confirmed live). The historian assigns event identity server-side; the alarm's own id
-            // is preserved below as a property for read-back correlation/traceability.
+            // Identity belongs to the event/durable queue, never to an individual RPC attempt.
+            Id = alarm.DeliveryId.ToString("D"),
             SourceName = alarm.EquipmentPath,
             Type = alarm.AlarmTypeName,
             EventTime = eventTime,

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs

@@ -188,7 +188,7 @@ internal static class RawAggregateReader
         var requested = MaxValuesFor(maxWindowRows);
         var (pass, maxValues) = await ceiling.ReadAsync(requested,
             budget => ReadWindowAsync(client, tag, aggregate, plan, requestStart, budget, maxWindowRows, ct),
-            (r, budget) => !r.OverCap && r.Received >= budget).ConfigureAwait(false);
+            (r, budget) => !r.OverCap && r.Received >= budget, deferRecentReprobe: true).ConfigureAwait(false);
         var (buckets, earliest, latest, received, overCap) = pass;
         if (overCap) throw LimitExceeded(tag, maxWindowRows);
         if (received >= maxValues) throw BudgetExhausted(tag, maxWindowRows, requested, maxValues);

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs

@@ -137,6 +137,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
         catch (Exception ex)
         {
             RecordFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
             throw;
         }
     }
@@ -232,6 +233,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
         catch (Exception ex)
         {
             RecordFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
             throw;
         }
     }
@@ -316,6 +318,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
         catch (Exception ex)
         {
             RecordFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
             throw;
         }
     }
@@ -443,6 +446,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
         catch (Exception ex)
         {
             RecordFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
             throw;
         }
     }
@@ -538,6 +542,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
         catch (Exception ex)
         {
             RecordFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
             throw;
         }
     }
@@ -701,6 +706,11 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
 
     private void RecordSuccess() => _health.RecordSuccess();
 
+    /// <summary>Classifies provider transport/timeouts, keeping SQL/query errors distinct (#697).</summary>
+    private static bool IsTransportFailure(Exception ex) => ex is NpgsqlException { IsTransient: true } and not PostgresException
+        || ex is PostgresException pg && (pg.SqlState.StartsWith("08", StringComparison.Ordinal)
+            || pg.SqlState is "57P01" or "57P02" or "57P03");
+
     private void RecordFailure(Exception ex)
     {
         // A cancelled read (deadline / client abort) is not a backend fault (the counters ignore it too).

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs

@@ -132,6 +132,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
         catch (Exception ex)
         {
             RecordFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
             throw;
         }
     }
@@ -229,6 +230,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
         catch (Exception ex) when (ex is not NotSupportedException)
         {
             RecordFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
             throw;
         }
     }
@@ -306,6 +308,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
         catch (Exception ex)
         {
             RecordFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
             throw;
         }
     }
@@ -401,6 +404,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
         catch (Exception ex)
         {
             RecordFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
             throw;
         }
     }
@@ -512,6 +516,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
         catch (Exception ex)
         {
             RecordFailure(ex);
+            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
             throw;
         }
     }
@@ -635,6 +640,9 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
 
     private void RecordSuccess() => _health.RecordSuccess();
 
+    /// <summary>Classifies provider transport/timeouts, keeping SQL/query errors distinct (#697).</summary>
+    private static bool IsTransportFailure(Exception ex) => ex is SqlException sql && sql.Number is -2 or 0 or 53 or 64 or 233 or 258 or 10053 or 10054 or 10060 or 10061 or 10065 or 11001 or 40197 or 40613;
+
     private void RecordFailure(Exception ex)
     {
         // A cancelled read (deadline / client abort) is not a backend fault (the counters ignore it too).

src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs

@@ -5129,6 +5129,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
 
             RunBounded(work, HistoryReadBatchParallelism);
         });
+        NormalizeHistoryValues(nodesToProcess, results);
     }
 
     /// <inheritdoc />
@@ -5225,6 +5226,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
 
             RunBounded(work, HistoryReadBatchParallelism);
         });
+        NormalizeHistoryValues(nodesToProcess, results);
     }
 
     /// <inheritdoc />
@@ -5516,8 +5518,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
             // One node's backend failure must not throw out of the batch — surface Bad for THIS node
             // only. Logged via the injected ILogger.
             _logger.LogError(ex, "OtOpcUaNodeManager: HistoryReadEvents failed for node {NodeId}", handle.NodeId);
-            errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
-            results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
+            errors[handle.Index] = HistoryReadFailureStatus(ex);
+            results[handle.Index] = new SdkHistoryReadResult { StatusCode = HistoryReadFailureStatus(ex) };
         }
     }
 
@@ -6233,7 +6235,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
             // Bad status for THIS node only. Logged through the injected ILogger
             // rather than swallowed silently.
             _logger.LogError(ex, "OtOpcUaNodeManager: HistoryRead failed for node {NodeId}", handle.NodeId);
-            errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
+            errors[handle.Index] = HistoryReadFailureStatus(ex);
+            results[handle.Index] = new SdkHistoryReadResult { StatusCode = HistoryReadFailureStatus(ex) };
         }
     }
 
@@ -6722,7 +6725,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
         {
             // One node's backend failure must not throw out of the batch — Bad for THIS node only.
             _logger.LogError(ex, "OtOpcUaNodeManager: HistoryReadRaw (paged) failed for node {NodeId}", handle.NodeId);
-            errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
+            errors[handle.Index] = HistoryReadFailureStatus(ex);
+            results[handle.Index] = new SdkHistoryReadResult { StatusCode = HistoryReadFailureStatus(ex) };
         }
     }
 
@@ -6829,8 +6833,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
         catch (Exception ex)
         {
             _logger.LogError(ex, "OtOpcUaNodeManager: HistoryRead (modified) failed for node {NodeId}", handle.NodeId);
-            errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
-            results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
+            errors[handle.Index] = HistoryReadFailureStatus(ex);
+            results[handle.Index] = new SdkHistoryReadResult { StatusCode = HistoryReadFailureStatus(ex) };
         }
     }
 
@@ -6879,6 +6883,49 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
         }
     }
 
+    /// <summary>Transport failures are retryable; unsupported operations retain their distinct status (#697).</summary>
+    private static uint HistoryReadFailureStatus(Exception error)
+    {
+        for (Exception? ex = error; ex is not null; ex = ex.InnerException)
+            if (ex is System.IO.IOException or System.Net.Sockets.SocketException or System.Net.Http.HttpRequestException or TimeoutException)
+                return StatusCodes.BadNoCommunication;
+        return StatusCodes.BadHistoryOperationUnsupported;
+    }
+
+    /// <summary>Normalize stored and AtTime history to each scalar node's declared DataType (#698).</summary>
+    private void NormalizeHistoryValues(List<NodeHandle> handles, IList<SdkHistoryReadResult> results)
+    {
+        foreach (var handle in handles)
+        {
+            if (AtTimeValueType(handle.NodeId) is not { } type || results[handle.Index]?.HistoryData is not { } body) continue;
+            var values = ExtensionObject.ToEncodeable(body) switch
+            {
+                HistoryData data => data.DataValues,
+                _ => null,
+            };
+            if (values is null) continue;
+            foreach (var value in values)
+            {
+                if (value.Value is null) continue;
+                try
+                {
+                    // A gateway number cannot reconstruct a DateTime or a String which was never stored.
+                    if ((type == TypeCode.DateTime && value.Value is not DateTime)
+                        || (type == TypeCode.String && value.Value is not string)) throw new InvalidCastException();
+                    var converted = Convert.ChangeType(value.Value, type, System.Globalization.CultureInfo.InvariantCulture);
+                    if (converted is float f && float.IsInfinity(f) && value.Value is double d && double.IsFinite(d))
+                        throw new OverflowException();
+                    value.WrappedValue = new Variant(converted);
+                }
+                catch (Exception ex) when (ex is InvalidCastException or OverflowException or FormatException)
+                {
+                    value.WrappedValue = Variant.Null;
+                    value.StatusCode = StatusCodes.BadTypeMismatch;
+                }
+            }
+        }
+    }
+
     /// <summary>Project a plain sample list into an SDK <see cref="HistoryData"/> (the paged Raw path
     /// works on a trimmed <see cref="IReadOnlyList{T}"/> rather than a whole <see cref="HistorianRead"/>).</summary>
     /// <param name="samples">The samples to project (already trimmed of boundary duplicates).</param>

src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs

@@ -100,16 +100,22 @@ internal static class ReverseRawReader
                     {
                         var asked = limit > 0 && cap > limit ? limit : cap; // cap 0 (the whole window) is never cut
                         IReadOnlyList<DataValueSnapshot> samples;
-                        try
+                        for (var attempt = 0; ; attempt++)
                         {
-                            samples = (await source.ReadRawAsync(tagname, start, end, (uint)asked, token).ConfigureAwait(false)).Samples;
-                        }
-                        catch (HistoryReadLimitExceededException ex) when (asked > 0 && ex.PreWindowRows > 0 && asked - ex.PreWindowRows >= 1)
-                        {
-                            // #694: rows just before the segment crowded the backend's budget — read it smaller; a full answer
-                            // then counts as truncated (the walk narrows), never as complete.
-                            asked -= ex.PreWindowRows;
-                            samples = (await source.ReadRawAsync(tagname, start, end, (uint)asked, token).ConfigureAwait(false)).Samples;
+                            try
+                            {
+                                samples = (await source.ReadRawAsync(tagname, start, end, (uint)asked, token).ConfigureAwait(false)).Samples;
+                                break;
+                            }
+                            catch (HistoryReadLimitExceededException ex) when (asked > 0 && ex.PreWindowRows > 0 && attempt < 2)
+                            {
+                                // #695: the discovery request may be smaller than the pre-window cluster. Size against
+                                // the ceiling, not that request; a second pass can discover more crowding rows.
+                                var ceilingRoom = source.MaxRawValuesPerRead is var learned and > 0 ? learned : asked;
+                                var smaller = Math.Min(asked - 1, ceilingRoom - ex.PreWindowRows - 1);
+                                if (smaller < 1) throw;
+                                asked = smaller;
+                            }
                         }
                         return new ForwardSegmentRead<DataValueSnapshot>(samples, asked > 0 && samples.Count >= asked);
                     },
@@ -120,7 +126,7 @@ internal static class ReverseRawReader
                         (budget < Math.Max(1, maxTieCluster)
                             ? "the backend's row ceiling (HistorianGateway Historian:ReadLimits:MaxValuesCeiling)."
                             : "ServerHistorian:MaxTieClusterOverfetch.")),
-                    ct)
+                    ct, useDiscoveryBudgetForGaps: limit > 0)
                 .ConfigureAwait(false);
             return new ReverseRawPage(rows, next.WidthTicks) { Hint = next };
         }

src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseWalk.cs

@@ -123,13 +123,14 @@ internal static class ReverseWalk
     /// <param name="hint">What the previous page of the same read carried (default = none).</param>
     /// <param name="tooManyAtOneTimestamp">Builds the exception for a timestamp holding more than M rows.</param>
     /// <param name="ct">Cancellation.</param>
+    /// <param name="useDiscoveryBudgetForGaps">Use ceiling-sized discovery and retain truncated remainder rows (#696, strict raw backends only).</param>
     /// <returns>At most <paramref name="want"/> rows, newest first, and the hint for the next page.</returns>
     public static async Task<(List<T> Rows, ReverseWalkHint Next)> WalkAsync<T>(
         Func<DateTime, DateTime, int, CancellationToken, Task<ForwardSegmentRead<T>>> read,
         Func<T, long> ticksOf,
         IComparer<T>? tieOrder,
         DateTime hi, DateTime lo, uint want, int tieBudget, long resolutionTicks, DateTime nowUtc,
-        ReverseWalkHint hint, Func<DateTime, Exception> tooManyAtOneTimestamp, CancellationToken ct)
+        ReverseWalkHint hint, Func<DateTime, Exception> tooManyAtOneTimestamp, CancellationToken ct, bool useDiscoveryBudgetForGaps = false)
     {
         var resolution = Math.Max(1L, resolutionTicks);
         var granularity = Math.Max(Alignment, resolution);
@@ -206,7 +207,8 @@ internal static class ReverseWalk
             var segLo = segHi - stopTicks <= w ? stopTicks : segHi - w;
             var start = Utc(AlignDown(segLo, granularity));
             var end = Utc(AlignUp(segHi + 1, granularity));
-            var segBudget = discovering && !probe ? discoveryBudget : pageBudget;
+            var segBudget = (discovering && !probe) || (useDiscoveryBudgetForGaps && (probe || known > 0))
+                ? discoveryBudget : pageBudget;
             var cap = segBudget + 2;
             IReadOnlyList<T> rows;
             var narrowed = false;
@@ -220,10 +222,14 @@ internal static class ReverseWalk
 
                 if (probe)
                 {
-                    // Deep history below: the probe saw only its oldest rows. Carry on walking from the same segHi.
+                    // #696: strict-ceiling discovery retains the rows/density instead of discarding the remainder probe.
                     probed = true;
-                    narrowed = true;
-                    break;
+                    if (!useDiscoveryBudgetForGaps)
+                    {
+                        narrowed = true;
+                        break;
+                    }
+                    probe = false;
                 }
 
                 var leading = 0;
@@ -284,7 +290,7 @@ internal static class ReverseWalk
                     // Rows exist at lastInner; what lies above it is unknown. Aim at about half the page budget at the
                     // density the truncated rows showed (their span, not the segment's — it may open on an empty stretch).
                     var aimed = inner > 1 && lastInner > firstInner
-                        ? (long)Math.Min((double)(lastInner - firstInner) * (pageBudget / 2.0) / (inner - 1), maxWidth)
+                        ? (long)Math.Min((double)(lastInner - firstInner) * ((useDiscoveryBudgetForGaps ? segBudget : pageBudget) / 2.0) / (inner - 1), maxWidth)
                         : resolution;
                     known = lastInner;
                     knownWidth = Math.Max(resolution, aimed);

tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/AlarmHistorianEventIdentityTests.cs

@@ -0,0 +1,78 @@
+using System.Text.Json;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests;
+
+public sealed class AlarmHistorianEventIdentityTests
+{
+    internal const string Payload = """{"AlarmId":"A","EquipmentPath":"S","AlarmName":"N","AlarmTypeName":"LimitAlarm","Severity":2,"EventKind":"Activated","Message":"m","User":"system","Comment":null,"TimestampUtc":"2026-01-01T00:00:00Z"}""";
+    internal const string QueueId = "52C8292271C629CCC2BFE4EC8C16A5E70CA7456C7D88BC282ABCEDF636F05901";
+    private static AlarmHistorianEvent Event() => new("A", "S", "N", "LimitAlarm", AlarmSeverity.High,
+        "Activated", "m", "system", null, new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc));
+
+    [Theory]
+    [InlineData(QueueId, "48bca092-b99d-8ea4-a2de-8b30d80a481a")]
+    [InlineData("persisted-row-v1", "39473040-002b-876c-a1e6-9a9ad4a60ded")]
+    public void Persisted_keys_have_frozen_versioned_delivery_GUID_vectors(string key, string expected)
+    {
+        Assert.Equal(Guid.Parse(expected), AlarmDeliveryIds.FromQueueId(key));
+    }
+
+    [Fact]
+    public void Delivery_identity_is_nonzero_and_a_copy_preserves_it()
+    {
+        var original = Event();
+        Assert.NotEqual(Guid.Empty, original.DeliveryId);
+        Assert.Equal(original.DeliveryId, (original with { }).DeliveryId);
+        Assert.Equal(original.DeliveryId, (original with { Comment = "copy" }).DeliveryId);
+    }
+
+    [Fact]
+    public void Business_equality_hash_and_historical_JSON_ignore_delivery_identity()
+    {
+        var first = Event();
+        var second = Event();
+        Assert.NotEqual(first.DeliveryId, second.DeliveryId);
+        Assert.Equal(first, second);
+        Assert.True(first == second);
+        Assert.Equal(first.GetHashCode(), second.GetHashCode());
+        Assert.Single(new HashSet<AlarmHistorianEvent> { first, second });
+        Assert.Equal(Payload, JsonSerializer.Serialize(first));
+        Assert.Equal(Payload, JsonSerializer.Serialize(second));
+        Assert.Equal(QueueId, AlarmSfSchema.DeriveId(Payload));
+        Assert.Equal(first, JsonSerializer.Deserialize<AlarmHistorianEvent>(Payload));
+    }
+
+    [Theory]
+    [InlineData("AlarmId")]
+    [InlineData("EquipmentPath")]
+    [InlineData("AlarmName")]
+    [InlineData("AlarmTypeName")]
+    [InlineData("Severity")]
+    [InlineData("EventKind")]
+    [InlineData("Message")]
+    [InlineData("User")]
+    [InlineData("Comment")]
+    [InlineData("TimestampUtc")]
+    public void Each_business_field_still_participates_in_equality(string field)
+    {
+        var first = Event();
+        var changed = field switch
+        {
+            "AlarmId" => first with { AlarmId = "B" },
+            "EquipmentPath" => first with { EquipmentPath = "T" },
+            "AlarmName" => first with { AlarmName = "Other" },
+            "AlarmTypeName" => first with { AlarmTypeName = "DiscreteAlarm" },
+            "Severity" => first with { Severity = AlarmSeverity.Low },
+            "EventKind" => first with { EventKind = "Cleared" },
+            "Message" => first with { Message = "other" },
+            "User" => first with { User = "operator" },
+            "Comment" => first with { Comment = "note" },
+            "TimestampUtc" => first with { TimestampUtc = first.TimestampUtc.AddTicks(1) },
+            _ => throw new ArgumentOutOfRangeException(nameof(field)),
+        };
+        Assert.NotEqual(first, changed);
+        Assert.False(first == changed);
+    }
+}

tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/LocalDbStoreAndForwardSinkTests.cs

@@ -123,6 +123,69 @@ public sealed class LocalDbStoreAndForwardSinkTests : IDisposable
         Comment: null,
         TimestampUtc: ts ?? DateTime.UtcNow);
 
+    [Fact]
+    public async Task Queued_delivery_identity_survives_retry_reopen_and_payload_changes()
+    {
+        // The persisted row key is authoritative: reserializing today's payload must never change a retry's identity.
+        var expected = Guid.Parse("48bca092-b99d-8ea4-a2de-8b30d80a481a");
+        var evt = System.Text.Json.JsonSerializer.Deserialize<AlarmHistorianEvent>(AlarmHistorianEventIdentityTests.Payload)!;
+        var writer = new FakeWriter { DefaultOutcome = HistorianWriteOutcome.RetryPlease };
+        using (var sink = new LocalDbStoreAndForwardSink(Db, writer, _log))
+        {
+            await sink.EnqueueAsync(evt, TestContext.Current.CancellationToken);
+            var row = (await Db.QueryAsync("SELECT id, payload_json FROM alarm_sf_events",
+                r => (Id: r.GetString(0), Payload: r.GetString(1)), null, TestContext.Current.CancellationToken)).Single();
+            row.Id.ShouldBe(AlarmHistorianEventIdentityTests.QueueId);
+            row.Payload.ShouldBe(AlarmHistorianEventIdentityTests.Payload);
+            await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
+            await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
+            writer.Batches.ShouldAllBe(batch => batch.Single().DeliveryId == expected);
+            await Db.ExecuteAsync("UPDATE alarm_sf_events SET payload_json = @Payload",
+                new { Payload = System.Text.Json.JsonSerializer.Serialize(evt with { Comment = "edited" }) },
+                TestContext.Current.CancellationToken);
+        }
+        _provider!.Dispose();
+        _provider = null;
+        _db = null;
+        var reopenedWriter = new FakeWriter();
+        using var reopened = new LocalDbStoreAndForwardSink(Db, reopenedWriter, _log);
+        await reopened.DrainOnceAsync(TestContext.Current.CancellationToken);
+        reopenedWriter.Batches.Single().Single().DeliveryId.ShouldBe(expected);
+        reopenedWriter.Batches.Single().Single().Comment.ShouldBe("edited");
+        reopened.GetStatus().QueueDepth.ShouldBe(0);
+    }
+
+    [Fact]
+    public async Task Independently_created_identical_events_keep_the_historical_queue_key()
+    {
+        var first = System.Text.Json.JsonSerializer.Deserialize<AlarmHistorianEvent>(AlarmHistorianEventIdentityTests.Payload)!;
+        var second = System.Text.Json.JsonSerializer.Deserialize<AlarmHistorianEvent>(AlarmHistorianEventIdentityTests.Payload)!;
+        first.DeliveryId.ShouldNotBe(second.DeliveryId);
+        using var sink = new LocalDbStoreAndForwardSink(Db, new FakeWriter(), _log);
+        await sink.EnqueueAsync(first, TestContext.Current.CancellationToken);
+        await sink.EnqueueAsync(second, TestContext.Current.CancellationToken);
+        sink.GetStatus().QueueDepth.ShouldBe(1);
+        (await Db.QueryAsync("SELECT id FROM alarm_sf_events", r => r.GetString(0), null, TestContext.Current.CancellationToken))
+            .ShouldBe([AlarmHistorianEventIdentityTests.QueueId]);
+    }
+
+    [Fact]
+    public async Task Distinct_business_events_at_the_same_time_have_distinct_queued_delivery_ids()
+    {
+        var first = Event("A", new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc));
+        var second = first with { EventKind = "Cleared", Comment = "different transition" };
+        var writer = new FakeWriter();
+        using var sink = new LocalDbStoreAndForwardSink(Db, writer, _log);
+        await sink.EnqueueAsync(first, TestContext.Current.CancellationToken);
+        await sink.EnqueueAsync(second, TestContext.Current.CancellationToken);
+        sink.GetStatus().QueueDepth.ShouldBe(2);
+        await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
+        var delivered = writer.Batches.Single();
+        delivered.Count.ShouldBe(2);
+        delivered.ShouldAllBe(evt => evt.DeliveryId != Guid.Empty);
+        delivered.Select(evt => evt.DeliveryId).Distinct().Count().ShouldBe(2);
+    }
+
     /// <summary>Verifies that acknowledged events are removed from the queue.</summary>
     [Fact]
     public async Task EnqueueThenDrain_Ack_removes_row()
@@ -297,6 +360,11 @@ public sealed class LocalDbStoreAndForwardSinkTests : IDisposable
         var status = sink.GetStatus();
         status.QueueDepth.ShouldBe(1);
         status.DeadLetterDepth.ShouldBe(0);
+        var originalDeliveryId = writer.Batches.Single().Single().DeliveryId;
+        await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
+        writer.Batches.Count.ShouldBe(2);
+        writer.Batches.Last().Single().DeliveryId.ShouldBe(originalDeliveryId);
+        sink.GetStatus().QueueDepth.ShouldBe(0);
     }
 
     /// <summary>Verifies that the exponential backoff ladder caps at 60 seconds.</summary>

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs

@@ -12,6 +12,20 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
 /// </summary>
 public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient
 {
+    public bool TagExistsResult = true;
+    public int TagExistsCallCount;
+    public Exception? TagExistsThrows;
+    /// <summary>Optional asynchronous catalog response for overlapping presence-cache tests.</summary>
+    public Func<string, CancellationToken, Task<bool>>? TagExistsSource;
+    public Task<bool> TagExistsAsync(string tag, CancellationToken ct)
+    {
+        ct.ThrowIfCancellationRequested();
+        Interlocked.Increment(ref TagExistsCallCount);
+        if (TagExistsSource is { } source) return source(tag, ct);
+        if (TagExistsThrows is { } error) throw error;
+        return Task.FromResult(TagExistsResult);
+    }
+
     // ---- ReadRaw -------------------------------------------------------------------------------
     public IReadOnlyList<HistorianSample> RawSamples = Array.Empty<HistorianSample>();
     public string? LastReadRawTag;

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAlarmHistorianWriterTests.cs

@@ -18,6 +18,33 @@ public sealed class GatewayAlarmHistorianWriterTests
     private static GatewayAlarmHistorianWriter Writer(FakeHistorianGatewayClient fake) =>
         new(fake, NullLogger<GatewayAlarmHistorianWriter>.Instance);
 
+    [Fact]
+    public async Task A_direct_retry_preserves_the_original_delivery_id()
+    {
+        var evt = Evt("A");
+        var fake = new FakeHistorianGatewayClient { SendEventThrows = new RpcException(new Status(StatusCode.Unavailable, "down")) };
+        using var writer = Writer(fake);
+        Assert.Equal(HistorianWriteOutcome.RetryPlease,
+            (await writer.WriteBatchAsync([evt], TestContext.Current.CancellationToken)).Single());
+        var first = fake.LastSendEvent!.Id;
+        fake.SendEventThrows = null;
+        fake.SendEventResult = new WriteAck { Success = true };
+        Assert.Equal(HistorianWriteOutcome.Ack,
+            (await writer.WriteBatchAsync([evt with { }], TestContext.Current.CancellationToken)).Single());
+        Assert.Equal(evt.DeliveryId.ToString("D"), first);
+        Assert.Equal(first, fake.LastSendEvent!.Id);
+    }
+
+    [Fact]
+    public async Task An_empty_delivery_id_is_permanent_failure_without_an_RPC()
+    {
+        var fake = new FakeHistorianGatewayClient();
+        using var writer = Writer(fake);
+        Assert.Equal(HistorianWriteOutcome.PermanentFail,
+            (await writer.WriteBatchAsync([Evt("A") with { DeliveryId = Guid.Empty }], TestContext.Current.CancellationToken)).Single());
+        Assert.Equal(0, fake.SendEventCallCount);
+    }
+
     [Fact]
     public async Task All_acked_when_SendEvent_succeeds()
     {

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAtTimeTests.cs

@@ -128,6 +128,35 @@ public sealed class GatewayAtTimeTests
         return r.Samples;
     }
 
+    [Fact]
+    public async Task A_future_time_on_a_live_tag_finds_the_prior_sample_without_walking_the_empty_future()
+    {
+        var store = new AvevaStore(Enumerable.Range(0, 7201).Select(i => (Now.AddSeconds(i - 7200), (double)i, false)));
+        var fake = store.Client();
+        var result = (await AtTime(fake, Now.AddHours(1))).Single();
+        result.Value.ShouldBe(7200d);
+        result.StatusCode.ShouldBe(UncertainInterpolated);
+        fake.ReadRawCallCount.ShouldBeLessThanOrEqualTo(4);
+    }
+
+    [Fact]
+    public async Task A_future_time_still_uses_a_stored_future_prior_sample()
+    {
+        var fake = new AvevaStore([(Now.AddSeconds(-1), 1d, false), (Now.AddMinutes(30), 2d, false)]).Client();
+        var result = (await AtTime(fake, Now.AddHours(1))).Single();
+        result.Value.ShouldBe(2d);
+        result.StatusCode.ShouldBe(UncertainInterpolated);
+    }
+
+    [Fact]
+    public async Task A_full_future_probe_walks_to_the_latest_stored_future_sample()
+    {
+        var fake = new AvevaStore(Enumerable.Range(0, 7201).Select(i => (Now.AddSeconds(i), (double)i, false))).Client();
+        var result = (await AtTime(fake, Now.AddHours(3))).Single();
+        result.Value.ShouldBe(7200d);
+        result.StatusCode.ShouldBe(UncertainInterpolated);
+    }
+
     // ---- the four live findings ------------------------------------------------------------------------------------
 
     [Fact] // live: 19:00:03 (a stored NULL) ⇒ 107 Good

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayCeilingPagingTests.cs

@@ -385,6 +385,21 @@ public sealed class GatewayCeilingPagingTests : IDisposable
         finally { await host.DisposeAsync(); }
     }
 
+    [Fact]
+    public async Task A_stopped_dense_tag_under_a_strict_ceiling_is_found_with_bounded_discovery_cost()
+    {
+        var stored = PerSecond(7200);
+        var fake = Gateway(stored, Ceiling);
+        var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
+        await ds.ReadRawAsync("T", At(0), At(1), 5000, Ct);
+        fake.RawCalls.Clear();
+        var page = await ReverseRawReader.ReadAsync(ds, "T", At(7200).AddDays(1), DateTime.UnixEpoch,
+            3, 65536, At(7200).AddDays(1), 0, Ct);
+        page.Samples.Select(s => s.Value).ShouldBe(new object?[] { 7199d, 7198d, 7197d });
+        fake.RawCalls.Count.ShouldBeLessThanOrEqualTo(12);
+        EveryRequestFits(fake);
+    }
+
     // ---- F3: rows crowding the look-back before a page start ---------------------------------------------------------------
 
     [Theory]
@@ -420,6 +435,54 @@ public sealed class GatewayCeilingPagingTests : IDisposable
         finally { await host.DisposeAsync(); }
     }
 
+    [Theory]
+    [InlineData(1709, 1366)] // #695 reverse-walk trace
+    [InlineData(1369, 1292)] // #695 bounds trace
+    public async Task A_warm_read_uses_the_last_room_below_the_ceiling_for_pre_window_ties(int ceiling, int cluster)
+    {
+        var start = At(1000);
+        var stored = Enumerable.Range(0, cluster).Select(i => (start.AddMilliseconds(-0.5), (double)i))
+            .Append((start, 9000d)).Append((start.AddSeconds(1), 9001d)).ToList();
+        var fake = Gateway(stored, ceiling);
+        using var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
+        await ds.ReadRawAsync("T", At(0), At(1), 5000, Ct); // learn without crowding
+        fake.RawCalls.Clear();
+        var result = await ds.ReadRawAsync("T", start, start.AddSeconds(2), 1, Ct);
+        result.Samples.Select(s => s.Value).ShouldBe([9000d]);
+        fake.RawCalls.ShouldAllBe(c => c.MaxValues <= ceiling);
+    }
+
+    [Fact]
+    public async Task A_warm_reverse_segment_can_shrink_again_when_the_pre_window_cluster_grows()
+    {
+        var start = At(1000);
+        var stored = Enumerable.Range(0, 1366).Select(i => (start.AddMilliseconds(-0.5), (double)i))
+            .Concat(Enumerable.Range(0, 500).Select(i => (start.AddMilliseconds(i), 9000d + i))).ToList();
+        var fake = Gateway(stored, 1709);
+        using var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
+        await ds.ReadRawAsync("T", At(0), At(1), 5000, Ct);
+        var page = await ReverseRawReader.ReadAsync(ds, "T", start.AddMilliseconds(499), start.AddTicks(-1),
+            249, 65536, At(2000), new ReverseWalkHint(499 * TimeSpan.TicksPerMillisecond), Ct);
+        page.Samples.Select(s => (double)s.Value!).ShouldBe(Enumerable.Range(9251, 249).Reverse().Select(i => (double)i));
+    }
+
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task Gateway_transport_faults_surface_as_BadNoCommunication(bool deadline)
+    {
+        var fake = new FakeHistorianGatewayClient { ThrowOnRead = deadline
+            ? new ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayException("deadline", new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.DeadlineExceeded, "deadline")))
+            : new ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayUnavailableException("unavailable") };
+        var (host, nm, nodeId) = await BootAsync(fake);
+        try
+        {
+            var r = ReadRaw(nm, nodeId, At(0), At(10), 10, null);
+            r.Error.StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
+        }
+        finally { await host.DisposeAsync(); }
+    }
+
     // ---- helpers ---------------------------------------------------------------------------------------------------------
 
     private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId NodeId)> BootAsync(

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayMissingTagTests.cs

@@ -0,0 +1,126 @@
+using Google.Protobuf.WellKnownTypes;
+using Grpc.Core;
+using Microsoft.Extensions.Logging.Abstractions;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
+
+public sealed class GatewayMissingTagTests
+{
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+    private static readonly DateTime Now = new(2026, 10, 13, 8, 0, 0, DateTimeKind.Utc);
+    private sealed class Clock : TimeProvider
+    {
+        public DateTime Current = Now;
+        public override DateTimeOffset GetUtcNow() => new(Current);
+    }
+    private static GatewayHistorianDataSource Source(FakeHistorianGatewayClient fake, Clock? clock = null, bool native = false) =>
+        new(fake, NullLogger<GatewayHistorianDataSource>.Instance, timeProvider: clock, atTimeFromRaw: !native);
+
+    [Theory]
+    [InlineData("Raw")]
+    [InlineData("AtTime")]
+    [InlineData("NativeAtTime")]
+    [InlineData("Count")]
+    [InlineData("TimeAverage")]
+    public async Task A_missing_tag_returns_no_data_without_a_history_RPC(string kind)
+    {
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = false };
+        var ds = Source(fake, native: kind == "NativeAtTime");
+        var result = kind switch
+        {
+            "Raw" => await ds.ReadRawAsync("Missing", Now.AddMinutes(-1), Now, 10, Ct),
+            "AtTime" or "NativeAtTime" => await ds.ReadAtTimeAsync("Missing", [Now], Ct),
+            _ => await ds.ReadProcessedAsync("Missing", Now.AddMinutes(-1), Now, TimeSpan.FromMinutes(1),
+                kind == "Count" ? HistoryAggregateType.Count : HistoryAggregateType.TimeAverage, Ct),
+        };
+        if (kind == "Raw") result.Samples.ShouldBeEmpty();
+        else
+        {
+            result.Samples.Count.ShouldBe(1);
+            result.Samples[0].Value.ShouldBeNull();
+            (result.Samples[0].StatusCode & 0xFFFF0000u).ShouldBe(0x809B0000u);
+            result.Samples[0].SourceTimestampUtc.ShouldBe(kind.Contains("AtTime") ? Now : Now.AddMinutes(-1));
+        }
+        fake.TagExistsCallCount.ShouldBe(1);
+        (fake.ReadRawCallCount + fake.ReadAggregateCallCount + fake.ReadAtTimeCallCount).ShouldBe(0);
+        ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
+    }
+
+    [Fact]
+    public async Task Negative_cache_expires_so_a_newly_provisioned_tag_can_be_read()
+    {
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = false };
+        var clock = new Clock(); var ds = Source(fake, clock);
+        await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct);
+        fake.TagExistsResult = true;
+        fake.RawSamples = [new HistorianSample { Timestamp = Timestamp.FromDateTime(Now.AddSeconds(-1)), NumericValue = 42, OpcQuality = 192 }];
+        (await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct)).Samples.ShouldBeEmpty();
+        fake.TagExistsCallCount.ShouldBe(1);
+        clock.Current = Now.AddSeconds(5);
+        (await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct)).Samples.Single().Value.ShouldBe(42d);
+        fake.TagExistsCallCount.ShouldBe(2);
+    }
+
+    [Fact]
+    public async Task Positive_cache_avoids_checks_per_page_and_expires_after_a_minute()
+    {
+        var fake = new FakeHistorianGatewayClient(); var clock = new Clock(); var ds = Source(fake, clock);
+        await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct);
+        await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct);
+        fake.TagExistsCallCount.ShouldBe(1);
+        clock.Current = Now.AddSeconds(60);
+        await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct);
+        fake.TagExistsCallCount.ShouldBe(2);
+    }
+
+    [Fact]
+    public async Task An_explicit_nested_NotFound_is_cached_as_missing()
+    {
+        var fake = new FakeHistorianGatewayClient { TagExistsThrows = new InvalidOperationException("wrapper",
+            new RpcException(new Status(StatusCode.NotFound, "absent"))) };
+        var ds = Source(fake);
+        (await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct)).Samples.ShouldBeEmpty();
+        (await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct)).Samples.ShouldBeEmpty();
+        fake.TagExistsCallCount.ShouldBe(1);
+        fake.ReadRawCallCount.ShouldBe(0);
+    }
+
+    [Fact]
+    public async Task The_presence_cache_is_bounded_and_a_displaced_entry_is_checked_again()
+    {
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = false }; var ds = Source(fake);
+        for (var i = 0; i <= GatewayTagPresenceCache.Capacity; i++)
+            await ds.ReadRawAsync($"T{i}", Now.AddMinutes(-1), Now, 10, Ct);
+        await ds.ReadRawAsync("T0", Now.AddMinutes(-1), Now, 10, Ct);
+        fake.TagExistsCallCount.ShouldBe(GatewayTagPresenceCache.Capacity + 2);
+    }
+
+    [Fact]
+    public async Task Caller_cancellation_is_not_cached_as_missing()
+    {
+        var fake = new FakeHistorianGatewayClient { TagExistsThrows = new OperationCanceledException() }; var ds = Source(fake);
+        await Should.ThrowAsync<OperationCanceledException>(() => ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct));
+        fake.TagExistsThrows = null;
+        await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct);
+        fake.TagExistsCallCount.ShouldBe(2);
+    }
+
+    [Theory]
+    [InlineData(StatusCode.Unavailable)]
+    [InlineData(StatusCode.PermissionDenied)]
+    [InlineData(StatusCode.Internal)]
+    public async Task A_metadata_fault_is_not_cached_as_a_missing_tag(StatusCode code)
+    {
+        var fake = new FakeHistorianGatewayClient { TagExistsThrows = new RpcException(new Status(code, "failure")) };
+        var ds = Source(fake);
+        await Should.ThrowAsync<Exception>(() => ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct));
+        fake.TagExistsThrows = null;
+        await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct);
+        fake.TagExistsCallCount.ShouldBe(2);
+        fake.ReadRawCallCount.ShouldBe(1);
+    }
+}

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayStrictCeilingTests.cs

@@ -57,6 +57,30 @@ public sealed class GatewayStrictCeilingTests
         public override DateTimeOffset GetUtcNow() => new(now);
     }
 
+    private sealed class MovingClock : TimeProvider
+    {
+        public DateTime Current = Now;
+        public override DateTimeOffset GetUtcNow() => new(Current);
+    }
+
+    [Fact]
+    public async Task Aggregate_ceiling_re_probes_wait_thirty_seconds_then_notice_a_raised_limit()
+    {
+        var fake = Store(100, ceiling: 50);
+        var clock = new MovingClock();
+        var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance, timeProvider: clock);
+        Task<HistoryReadResult> Read() => ds.ReadProcessedAsync("T", At(0), At(100), TimeSpan.FromSeconds(100), HistoryAggregateType.Count, Ct);
+        await Should.ThrowAsync<HistoryReadLimitExceededException>(Read);
+        await Should.ThrowAsync<HistoryReadLimitExceededException>(Read);
+        Refusals(fake).ShouldBe(1);
+        fake.MaxValuesCeiling = 200;
+        await Should.ThrowAsync<HistoryReadLimitExceededException>(Read);
+        Refusals(fake).ShouldBe(1);
+        clock.Current = Now.AddSeconds(30);
+        (await Read()).Samples.Single().Value.ShouldBe(100);
+        ds.MaxRawValuesPerRead.ShouldBe(200 - RawAggregateReader.PreWindowAllowance);
+    }
+
     // ---- raw, uncapped ---------------------------------------------------------------------------------------------------
 
     [Fact]

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagCatalogWireTests.cs

@@ -0,0 +1,90 @@
+using System.Net;
+using Grpc.Core;
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Server.Kestrel.Core;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
+using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
+
+public sealed class GatewayTagCatalogWireTests
+{
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+    [Theory]
+    [InlineData("Pump1", true)]
+    [InlineData("Missing", false)]
+    [InlineData("Pump*", false)]
+    public async Task Presence_is_proved_by_exact_names_without_using_the_broken_metadata_parser(string tag, bool expected)
+    {
+        var service = new TagsService();
+        await using var server = await StartAsync(service);
+        using var adapter = Create(server.Port);
+        (await adapter.TagExistsAsync(tag, Ct)).ShouldBe(expected);
+        service.Filter.ShouldBe(tag);
+        service.MetadataCalls.ShouldBe(0);
+        service.Completed.ShouldBe(1);
+    }
+
+    [Fact]
+    public async Task A_catalog_stream_fault_after_a_matching_name_is_not_cached_as_presence()
+    {
+        var service = new TagsService { FailAfterRow = true };
+        await using var server = await StartAsync(service);
+        using var adapter = Create(server.Port);
+        var source = new GatewayHistorianDataSource(adapter, NullLogger<GatewayHistorianDataSource>.Instance);
+        var now = DateTime.UtcNow;
+        await Should.ThrowAsync<Exception>(() => source.ReadRawAsync("Pump1", now.AddMinutes(-1), now, 10, Ct));
+        service.FailAfterRow = false; service.Names = [];
+        (await source.ReadRawAsync("Pump1", now.AddMinutes(-1), now, 10, Ct)).Samples.ShouldBeEmpty();
+        service.BrowseCalls.ShouldBe(2);
+    }
+
+    private static HistorianGatewayClientAdapter Create(int port) => HistorianGatewayClientAdapter.Create(
+        new ServerHistorianOptions { Endpoint = $"http://127.0.0.1:{port}", ApiKey = "histgw_test_key", UseTls = false },
+        NullLoggerFactory.Instance);
+
+    private sealed class TagsService : HistorianTags.HistorianTagsBase
+    {
+        public int MetadataCalls, BrowseCalls, Completed;
+        public string? Filter;
+        public bool FailAfterRow;
+        public string[] Names = ["Pump1", "Pump2"];
+        public override Task<GetTagMetadataReply> GetTagMetadata(GetTagMetadataRequest request, ServerCallContext context)
+        {
+            MetadataCalls++;
+            throw new RpcException(new Status(StatusCode.Internal, "GetTagInfo response ended unexpectedly."));
+        }
+        public override async Task BrowseTagNames(BrowseTagNamesRequest request, IServerStreamWriter<TagName> stream, ServerCallContext context)
+        {
+            BrowseCalls++; Filter = request.Filter;
+            foreach (var name in Names)
+            {
+                await stream.WriteAsync(new TagName { Name = name }, context.CancellationToken);
+                if (FailAfterRow) throw new RpcException(new Status(StatusCode.Internal, "catalog failure"));
+            }
+            Completed++;
+        }
+    }
+
+    private sealed class Server(WebApplication app, int port) : IAsyncDisposable
+    {
+        public int Port => port;
+        public async ValueTask DisposeAsync() { await app.StopAsync(); await app.DisposeAsync(); }
+    }
+    private static async Task<Server> StartAsync(TagsService service)
+    {
+        var builder = WebApplication.CreateSlimBuilder();
+        builder.Logging.ClearProviders();
+        builder.WebHost.ConfigureKestrel(k => k.Listen(IPAddress.Loopback, 0, o => o.Protocols = HttpProtocols.Http2));
+        builder.Services.AddGrpc(); builder.Services.AddSingleton(service);
+        var app = builder.Build(); app.MapGrpcService<TagsService>(); await app.StartAsync();
+        return new Server(app, new Uri(app.Urls.First()).Port);
+    }
+}

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagPresenceCacheConcurrencyTests.cs

@@ -0,0 +1,209 @@
+using Grpc.Core;
+using Shouldly;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
+
+public sealed class GatewayTagPresenceCacheConcurrencyTests(ITestOutputHelper output)
+{
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+    private const int BurstSize = 16;
+
+    private sealed class Clock : TimeProvider
+    {
+        public DateTimeOffset Current = new(2026, 10, 13, 8, 0, 0, TimeSpan.Zero);
+        public override DateTimeOffset GetUtcNow() => Current;
+    }
+
+    private static TaskCompletionSource<bool> Gate() => new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task Overlapping_cold_lookups_return_the_catalog_result_and_populate_the_cache(bool exists)
+    {
+        // Catches dropped/corrupted concurrent writes or a result cached with the wrong presence value.
+        var gate = Gate();
+        var fake = new FakeHistorianGatewayClient { TagExistsSource = (_, _) => gate.Task };
+        var cache = new GatewayTagPresenceCache(fake, new Clock());
+        var reads = Enumerable.Range(0, BurstSize).Select(_ => cache.ExistsAsync("T", Ct)).ToArray();
+        reads.ShouldAllBe(read => !read.IsCompleted);
+        var rpcCount = fake.TagExistsCallCount;
+        // Measure amplification without requiring either coalescing or one RPC per caller.
+        rpcCount.ShouldBeInRange(1, BurstSize);
+        output.WriteLine($"Cold {exists}: {BurstSize} overlapping callers, {rpcCount} catalog RPCs.");
+        gate.SetResult(exists);
+        (await Task.WhenAll(reads).WaitAsync(Ct)).ShouldAllBe(value => value == exists);
+
+        fake.TagExistsSource = null;
+        fake.TagExistsThrows = new InvalidOperationException("A warm lookup must not contact the catalog.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(exists);
+    }
+
+    [Theory]
+    [InlineData(false, 5)]
+    [InlineData(true, 60)]
+    public async Task Overlapping_expired_lookups_refresh_presence_at_the_TTL_boundary(bool initial, int ttlSeconds)
+    {
+        // Catches an inclusive expiry comparison or failure to replace an expired positive/negative entry.
+        var clock = new Clock();
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = initial };
+        var cache = new GatewayTagPresenceCache(fake, clock);
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(initial);
+        var expires = clock.Current.AddSeconds(ttlSeconds);
+        clock.Current = expires.AddTicks(-1);
+        fake.TagExistsThrows = new InvalidOperationException("The entry must still be warm before expiry.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(initial);
+
+        clock.Current = expires;
+        var gate = Gate();
+        fake.TagExistsThrows = null;
+        fake.TagExistsSource = (_, _) => gate.Task;
+        var reads = Enumerable.Range(0, BurstSize).Select(_ => cache.ExistsAsync("T", Ct)).ToArray();
+        reads.ShouldAllBe(read => !read.IsCompleted);
+        var rpcCount = fake.TagExistsCallCount - 1;
+        rpcCount.ShouldBeInRange(1, BurstSize);
+        output.WriteLine($"Expired {initial}: {BurstSize} overlapping callers, {rpcCount} catalog RPCs.");
+        gate.SetResult(!initial);
+        (await Task.WhenAll(reads).WaitAsync(Ct)).ShouldAllBe(value => value == !initial);
+
+        fake.TagExistsSource = null;
+        fake.TagExistsThrows = new InvalidOperationException("A refreshed lookup must stay cached.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(!initial);
+    }
+
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task Overlapping_distinct_tag_completions_keep_the_cache_bounded(bool exists)
+    {
+        // Catches racing capacity checks that allow more than Capacity resident entries.
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = exists };
+        var cache = new GatewayTagPresenceCache(fake, new Clock());
+        var tags = Enumerable.Range(0, GatewayTagPresenceCache.Capacity + 31).Select(i => $"T{i}").ToArray();
+        foreach (var tag in tags.Take(GatewayTagPresenceCache.Capacity - 1))
+            (await cache.ExistsAsync(tag, Ct)).ShouldBe(exists);
+
+        var gate = Gate();
+        fake.TagExistsSource = (_, _) => gate.Task;
+        var reads = tags.Skip(GatewayTagPresenceCache.Capacity - 1).Select(tag => cache.ExistsAsync(tag, Ct)).ToArray();
+        reads.ShouldAllBe(read => !read.IsCompleted);
+        gate.SetResult(exists);
+        (await Task.WhenAll(reads).WaitAsync(Ct)).ShouldAllBe(value => value == exists);
+
+        // Failed probes cannot insert or evict entries, so count resident hits through the public behavior.
+        fake.TagExistsSource = null;
+        var uncached = new InvalidOperationException("Uncached catalog probe.");
+        fake.TagExistsThrows = uncached;
+        var hits = 0;
+        foreach (var tag in tags)
+        {
+            try { (await cache.ExistsAsync(tag, Ct)).ShouldBe(exists); hits++; }
+            catch (InvalidOperationException error) when (ReferenceEquals(error, uncached)) { }
+        }
+        hits.ShouldBe(GatewayTagPresenceCache.Capacity);
+    }
+
+    [Theory]
+    [InlineData(StatusCode.Unavailable)]
+    [InlineData(StatusCode.PermissionDenied)]
+    [InlineData(StatusCode.Internal)]
+    [InlineData(StatusCode.Cancelled)]
+    public async Task A_faulting_lookup_cannot_overwrite_an_overlapping_positive_result(StatusCode code)
+    {
+        // Catches converting metadata faults to false, including a fault finishing after a successful lookup.
+        var success = Gate();
+        var failure = Gate();
+        var calls = 0;
+        var fake = new FakeHistorianGatewayClient
+        {
+            TagExistsSource = (_, _) => Interlocked.Increment(ref calls) == 1 ? success.Task : failure.Task,
+        };
+        var cache = new GatewayTagPresenceCache(fake, new Clock());
+        var goodRead = cache.ExistsAsync("T", Ct);
+        var badRead = cache.ExistsAsync("T", Ct);
+        goodRead.IsCompleted.ShouldBeFalse();
+        badRead.IsCompleted.ShouldBeFalse();
+        success.SetResult(true);
+        (await goodRead.WaitAsync(Ct)).ShouldBeTrue();
+        failure.SetException(new RpcException(new Status(code, "Catalog fault.")));
+        (await Should.ThrowAsync<RpcException>(() => badRead.WaitAsync(Ct))).StatusCode.ShouldBe(code);
+        fake.TagExistsSource = null;
+        fake.TagExistsThrows = new InvalidOperationException("The positive entry must survive the failed lookup.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBeTrue();
+    }
+
+    [Fact]
+    public async Task Cancellation_after_a_catalog_result_cannot_overwrite_an_overlapping_positive_result()
+    {
+        // Catches removal of the post-await token check when the client ignores cancellation and returns false.
+        var success = Gate();
+        var cancelled = Gate();
+        using var caller = CancellationTokenSource.CreateLinkedTokenSource(Ct);
+        var fake = new FakeHistorianGatewayClient
+        {
+            TagExistsSource = (_, token) => token == caller.Token ? cancelled.Task : success.Task,
+        };
+        var cache = new GatewayTagPresenceCache(fake, new Clock());
+        var goodRead = cache.ExistsAsync("T", Ct);
+        var cancelledRead = cache.ExistsAsync("T", caller.Token);
+        success.SetResult(true);
+        (await goodRead.WaitAsync(Ct)).ShouldBeTrue();
+        caller.Cancel();
+        cancelled.SetResult(false);
+        await Should.ThrowAsync<OperationCanceledException>(() => cancelledRead.WaitAsync(Ct));
+        fake.TagExistsSource = null;
+        fake.TagExistsThrows = new InvalidOperationException("The positive entry must survive cancellation.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBeTrue();
+        // Cached results still respect an already-cancelled caller.
+        await Should.ThrowAsync<OperationCanceledException>(() => cache.ExistsAsync("T", caller.Token));
+    }
+
+    [Theory]
+    [InlineData(false, 5)]
+    [InlineData(true, 60)]
+    public async Task An_expired_entry_is_rechecked_after_a_faulting_refresh(bool initial, int ttlSeconds)
+    {
+        // Catches faults renewing an expired negative or replacing an expired positive with a negative.
+        var clock = new Clock();
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = initial };
+        var cache = new GatewayTagPresenceCache(fake, clock);
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(initial);
+        clock.Current = clock.Current.AddSeconds(ttlSeconds);
+        var gate = Gate();
+        fake.TagExistsSource = (_, _) => gate.Task;
+        var read = cache.ExistsAsync("T", Ct);
+        read.IsCompleted.ShouldBeFalse();
+        gate.SetException(new RpcException(new Status(StatusCode.Unavailable, "Catalog unavailable.")));
+        await Should.ThrowAsync<RpcException>(() => read.WaitAsync(Ct));
+        fake.TagExistsSource = null;
+        fake.TagExistsResult = true;
+        (await cache.ExistsAsync("T", Ct)).ShouldBeTrue();
+        fake.TagExistsCallCount.ShouldBe(3);
+    }
+
+    [Theory]
+    [InlineData(false, 5)]
+    [InlineData(true, 60)]
+    public async Task An_expired_entry_is_rechecked_after_a_cancelled_refresh(bool initial, int ttlSeconds)
+    {
+        // Catches cancellation renewing expired entries when the catalog returns an absence result anyway.
+        var clock = new Clock();
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = initial };
+        var cache = new GatewayTagPresenceCache(fake, clock);
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(initial);
+        clock.Current = clock.Current.AddSeconds(ttlSeconds);
+        var gate = Gate();
+        fake.TagExistsSource = (_, _) => gate.Task;
+        using var caller = CancellationTokenSource.CreateLinkedTokenSource(Ct);
+        var read = cache.ExistsAsync("T", caller.Token);
+        read.IsCompleted.ShouldBeFalse();
+        caller.Cancel();
+        gate.SetResult(false);
+        await Should.ThrowAsync<OperationCanceledException>(() => read.WaitAsync(Ct));
+        fake.TagExistsSource = null;
+        fake.TagExistsResult = true;
+        (await cache.ExistsAsync("T", Ct)).ShouldBeTrue();
+        fake.TagExistsCallCount.ShouldBe(3);
+    }
+}

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Live/GatewayLiveIntegrationTests.cs

@@ -149,9 +149,8 @@ public sealed class GatewayLiveIntegrationTests(GatewayLiveFixture fixture) : IC
     ///     <see cref="GatewayAlarmHistorianWriter"/>, then <c>ReadEvents</c> the recent window for that
     ///     source through the real <see cref="GatewayHistorianDataSource"/> and assert the event is
     ///     present. Requires the gateway running <c>RuntimeDb:EventReadsEnabled=true</c> (the SQL
-    ///     alarm-history read path). Presence is asserted as "at least one event for the source surfaced
-    ///     in the post-send window" (the data source filters by source); the exact AlarmId / message
-    ///     match is surfaced as a diagnostic, since the SQL event store may re-key the row.
+    ///     alarm-history read path). The exact delivery GUID must return; an unrelated event for the
+    ///     same source cannot satisfy the assertion.
     /// </summary>
     [Fact]
     [Trait("Category", "LiveIntegration")]
@@ -196,7 +195,7 @@ public sealed class GatewayLiveIntegrationTests(GatewayLiveFixture fixture) : IC
             var read = await dataSource.ReadEventsAsync(
                 source, eventUtc - TimeSpan.FromMinutes(5), DateTime.UtcNow + TimeSpan.FromMinutes(1), maxEvents: 0, ct);
             events = read.Events;
-            if (events.Count > 0) break;
+            if (events.Any(e => Guid.TryParse(e.EventId, out var id) && id == alarm.DeliveryId)) break;
             await Task.Delay(TimeSpan.FromSeconds(2), ct);
         }
         while (DateTime.UtcNow < deadline);
@@ -211,18 +210,18 @@ public sealed class GatewayLiveIntegrationTests(GatewayLiveFixture fixture) : IC
             $"a source-filtered ReadEvents must return the just-sent alarm for source '{source}': the gateway now " +
             "populates Source_Object from the event SourceName (C4 fix), so OtOpcUa's own alarm sends round-trip.");
 
-        var exactMatch = events.Any(e => string.Equals(e.EventId, alarmId, StringComparison.Ordinal));
+        var exactMatch = events.Any(e => Guid.TryParse(e.EventId, out var id) && id == alarm.DeliveryId);
+        exactMatch.ShouldBeTrue("the just-sent event must retain its nonzero delivery GUID through persistence and readback.");
         TestContext.Current.SendDiagnosticMessage(
-            $"alarm round-trip: SendEvent source='{source}' id={alarmId} → ReadEvents returned {events.Count} event(s); exact-id match={exactMatch}.");
+            $"alarm round-trip: SendEvent source='{source}' delivery-id={alarm.DeliveryId:D} → exact ID returned among {events.Count} event(s).");
     }
 
     /// <summary>
     ///     Alarm send contract — the OtOpcUa responsibility in isolation: an <see cref="AlarmHistorianEvent"/>
     ///     maps to a wire event the gateway accepts and <c>SendEvent</c> returns <see cref="HistorianWriteOutcome.Ack"/>.
     ///     This is the half that must always hold (the readback half is historian-gated; see
-    ///     <see cref="Alarm_SendEvent_then_ReadEvents"/>). Regression guard for the event-id mapping —
-    ///     a client-supplied wire <c>Id</c> makes the gateway's SendEvent handler throw, so the mapper
-    ///     must leave it unset.
+    ///     <see cref="Alarm_SendEvent_then_ReadEvents"/>). The mapper sends the stable nonzero delivery GUID;
+    ///     the earlier assumption that the gateway rejected every client-supplied ID was incorrect (#705).
     /// </summary>
     [Fact]
     [Trait("Category", "LiveIntegration")]
@@ -251,7 +250,7 @@ public sealed class GatewayLiveIntegrationTests(GatewayLiveFixture fixture) : IC
 
         outcomes.ShouldHaveSingleItem().ShouldBe(
             HistorianWriteOutcome.Ack,
-            "the alarm SendEvent must be acked (the AlarmEventMapper must NOT set the wire event Id — the gateway rejects a client-supplied id).");
+            "the alarm SendEvent carrying its nonzero delivery GUID must be acked.");
     }
 
     /// <summary>

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Mapping/AlarmEventMapperTests.cs

@@ -7,6 +7,17 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests.Mapping;
 
 public sealed class AlarmEventMapperTests
 {
+    [Fact]
+    public void A_direct_event_has_a_nonzero_wire_id_that_survives_repeated_mapping()
+    {
+        var alarm = new AlarmHistorianEvent("A", "S", "N", "LimitAlarm", AlarmSeverity.High,
+            "Activated", "m", "system", null, new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc));
+        var first = AlarmEventMapper.ToHistorianEvent(alarm).Id;
+        Assert.True(Guid.TryParse(first, out var id), $"Expected a delivery GUID; received '{first}'.");
+        Assert.NotEqual(Guid.Empty, id);
+        Assert.Equal(first, AlarmEventMapper.ToHistorianEvent(alarm).Id);
+    }
+
     [Fact]
     public void Maps_source_time_type_and_rich_properties()
     {
@@ -17,9 +28,7 @@ public sealed class AlarmEventMapperTests
         Assert.Equal("Area/Line/Pump1", e.SourceName);
         Assert.Equal("LimitAlarm", e.Type);
         Assert.Equal(new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc), e.EventTime.ToDateTime());
-        // The wire Id is deliberately left unset — the gateway's SendEvent rejects a client-supplied
-        // event id. The alarm's own id is carried as a property instead.
-        Assert.Equal(string.Empty, e.Id);
+        Assert.Equal(a.DeliveryId.ToString("D"), e.Id);
         Assert.Equal("A1", e.Properties["AlarmId"]);
         Assert.Equal("HiHi", e.Properties["AlarmName"]);
         Assert.Equal("Activated", e.Properties["EventKind"]);
@@ -67,14 +76,21 @@ public sealed class AlarmEventMapperTests
     }
 
     [Fact]
-    public void Wire_id_is_never_set_and_blank_alarm_id_adds_no_property()
+    public void A_blank_condition_id_still_has_delivery_identity_and_adds_no_condition_property()
     {
-        // A blank AlarmId must NOT fall back to a generated wire Id (the gateway rejects any id) and
-        // must not add an empty AlarmId property.
+        // Condition identity and event delivery identity serve different purposes.
         var a = new AlarmHistorianEvent("", "S", "N", "DiscreteAlarm", AlarmSeverity.Low, "Cleared", "m", "system", null,
             new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc));
         var e = AlarmEventMapper.ToHistorianEvent(a);
-        Assert.Equal(string.Empty, e.Id);
+        Assert.Equal(a.DeliveryId.ToString("D"), e.Id);
         Assert.False(e.Properties.ContainsKey("AlarmId"));
     }
+
+    [Fact]
+    public void An_explicitly_empty_delivery_id_is_rejected_instead_of_reminted()
+    {
+        var alarm = new AlarmHistorianEvent("A", "S", "N", "LimitAlarm", AlarmSeverity.High,
+            "Activated", "m", "system", null, DateTime.UtcNow) { DeliveryId = Guid.Empty };
+        Assert.Throws<ArgumentException>(() => AlarmEventMapper.ToHistorianEvent(alarm));
+    }
 }

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs

@@ -11,6 +11,33 @@ public sealed class QuestDbHistorianProviderTests
         ConnectionString = "Host=localhost;Port=18812;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading",
     });
 
+    [Theory]
+    [InlineData("Raw")]
+    [InlineData("AtTime")]
+    [InlineData("Processed")]
+    [InlineData("Events")]
+    [InlineData("Modified")]
+    public async Task A_connection_failure_is_exposed_as_a_transport_failure(string kind)
+    {
+        using var provider = new QuestDbHistorianProvider(new QuestDbHistorianOptions {
+            ConnectionString = "Host=127.0.0.1;Port=1;Username=admin;Password=test;Database=qdb;Timeout=1;Pooling=False;ServerCompatibilityMode=NoTypeLoading", AutoCreateSchema = false
+        });
+        using var source = provider.CreateDataSource(new NullSp());
+        var t = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
+        Func<Task> read = kind switch
+        {
+            "Raw" => async () => await source.ReadRawAsync("T", t, t.AddHours(1), 10, TestContext.Current.CancellationToken),
+            "AtTime" => async () => await source.ReadAtTimeAsync("T", [t], TestContext.Current.CancellationToken),
+            "Processed" => async () => await source.ReadProcessedAsync("T", t, t.AddHours(1), TimeSpan.FromSeconds(10), HistoryAggregateType.Average, TestContext.Current.CancellationToken),
+            "Events" => async () => await source.ReadEventsAsync("T", t, t.AddHours(1), 10, TestContext.Current.CancellationToken),
+            _ => async () => await source.ReadModifiedAsync("T", t, t.AddHours(1), 10, null, TestContext.Current.CancellationToken)
+        };
+        var error = await Should.ThrowAsync<System.IO.IOException>(read);
+        error.InnerException.ShouldNotBeNull();
+        // A cached bootstrap outage must remain retryable on subsequent reads as well.
+        (await Should.ThrowAsync<System.IO.IOException>(read)).InnerException.ShouldNotBeNull();
+    }
+
     [Fact]
     public void Name_and_all_six_kinds()
     {

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianProviderTests.cs

@@ -11,6 +11,33 @@ public sealed class SqlServerHistorianProviderTests
         ConnectionString = "Server=localhost;User Id=sa;Password=x;TrustServerCertificate=True",
     });
 
+    [Theory]
+    [InlineData("Raw")]
+    [InlineData("AtTime")]
+    [InlineData("Processed")]
+    [InlineData("Events")]
+    [InlineData("Modified")]
+    public async Task A_connection_failure_is_exposed_as_a_transport_failure(string kind)
+    {
+        var provider = new SqlServerHistorianProvider(new SqlServerHistorianOptions {
+            ConnectionString = "Server=127.0.0.1,1;User Id=sa;Password=test;TrustServerCertificate=True;Connect Timeout=1;ConnectRetryCount=0;Pooling=False", AutoCreateSchema = false
+        });
+        using var source = provider.CreateDataSource(new NullSp());
+        var t = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
+        Func<Task> read = kind switch
+        {
+            "Raw" => async () => await source.ReadRawAsync("T", t, t.AddHours(1), 10, TestContext.Current.CancellationToken),
+            "AtTime" => async () => await source.ReadAtTimeAsync("T", [t], TestContext.Current.CancellationToken),
+            "Processed" => async () => await source.ReadProcessedAsync("T", t, t.AddHours(1), TimeSpan.FromSeconds(10), HistoryAggregateType.Average, TestContext.Current.CancellationToken),
+            "Events" => async () => await source.ReadEventsAsync("T", t, t.AddHours(1), 10, TestContext.Current.CancellationToken),
+            _ => async () => await source.ReadModifiedAsync("T", t, t.AddHours(1), 10, null, TestContext.Current.CancellationToken)
+        };
+        var error = await Should.ThrowAsync<System.IO.IOException>(read);
+        error.InnerException.ShouldNotBeNull();
+        // A cached bootstrap outage must remain retryable on subsequent reads as well.
+        (await Should.ThrowAsync<System.IO.IOException>(read)).InnerException.ShouldNotBeNull();
+    }
+
     [Fact]
     public void Name_and_all_six_kinds()
     {

tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmSfConvergenceTests.cs

@@ -41,10 +41,14 @@ public sealed class AlarmSfConvergenceTests
     /// <summary>A writer that never acknowledges — a historian outage, which is when the buffer matters.</summary>
     private sealed class OutageWriter : IAlarmHistorianWriter
     {
+        public List<AlarmHistorianEvent> Events { get; } = [];
         public Task<IReadOnlyList<HistorianWriteOutcome>> WriteBatchAsync(
-            IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken) =>
-            Task.FromResult<IReadOnlyList<HistorianWriteOutcome>>(
+            IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken)
+        {
+            Events.AddRange(batch);
+            return Task.FromResult<IReadOnlyList<HistorianWriteOutcome>>(
                 batch.Select(_ => HistorianWriteOutcome.RetryPlease).ToList());
+        }
     }
 
     private static AlarmHistorianEvent Event(string alarmId, int second) => new(
@@ -170,11 +174,15 @@ public sealed class AlarmSfConvergenceTests
         await pair.StartAsync();
 
         var shared = Event("eq/shared", 1);
-        using var sinkA = new LocalDbStoreAndForwardSink(pair.A, new OutageWriter(), Log);
-        using var sinkB = new LocalDbStoreAndForwardSink(pair.B, new OutageWriter(), Log);
+        var sharedOnB = Event("eq/shared", 1);
+        shared.DeliveryId.ShouldNotBe(sharedOnB.DeliveryId);
+        var writerA = new OutageWriter();
+        var writerB = new OutageWriter();
+        using var sinkA = new LocalDbStoreAndForwardSink(pair.A, writerA, Log);
+        using var sinkB = new LocalDbStoreAndForwardSink(pair.B, writerB, Log);
 
         await sinkA.EnqueueAsync(shared, TestContext.Current.CancellationToken);
-        await sinkB.EnqueueAsync(shared, TestContext.Current.CancellationToken);
+        await sinkB.EnqueueAsync(sharedOnB, TestContext.Current.CancellationToken);
 
         // A second, DISTINCT event on B. Without it the scenario would pass with replication
         // switched off entirely — each node would hold its own single copy and "count == 1 on
@@ -193,5 +201,13 @@ public sealed class AlarmSfConvergenceTests
                 async () => await CountAsync(pair.A) > 2 || await CountAsync(pair.B) > 2,
                 TimeSpan.FromSeconds(3)))
             .ShouldBeFalse("the duplicate must stay collapsed");
+
+        // Either node can retry after promotion, with the same GUID despite independent input identities.
+        await sinkA.DrainOnceAsync(TestContext.Current.CancellationToken);
+        await sinkB.DrainOnceAsync(TestContext.Current.CancellationToken);
+        var deliveryA = writerA.Events.Single(e => e.AlarmId == "eq/shared").DeliveryId;
+        var deliveryB = writerB.Events.Single(e => e.AlarmId == "eq/shared").DeliveryId;
+        deliveryA.ShouldNotBe(Guid.Empty);
+        deliveryB.ShouldBe(deliveryA);
     }
 }

tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmSfLegacyMigratorTests.cs

@@ -36,6 +36,38 @@ public sealed class AlarmSfLegacyMigratorTests : IDisposable
 
     private ServiceProvider? _provider;
 
+    private sealed class CapturingRetryWriter : IAlarmHistorianWriter
+    {
+        public List<AlarmHistorianEvent> Events { get; } = [];
+        public Task<IReadOnlyList<HistorianWriteOutcome>> WriteBatchAsync(
+            IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken)
+        {
+            Events.AddRange(batch);
+            return Task.FromResult<IReadOnlyList<HistorianWriteOutcome>>(batch.Select(_ => HistorianWriteOutcome.RetryPlease).ToArray());
+        }
+    }
+
+    [Fact]
+    public async Task A_legacy_payload_keeps_its_key_and_stable_delivery_identity_across_drains()
+    {
+        var payload = Payload("eq/a");
+        SeedLegacy((1, "eq/a", payload, 0, dead: 0));
+        var db = BuildDb();
+        AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());
+        var row = (await db.QueryAsync("SELECT id, payload_json FROM alarm_sf_events",
+            r => (Id: r.GetString(0), Payload: r.GetString(1)), null, TestContext.Current.CancellationToken)).Single();
+        row.Id.ShouldBe("B1D80B7207C3F3B6CB31105DBF1A6E0D6FE300DA1B3F877AB746B36535310A60");
+        row.Payload.ShouldBe(payload);
+        var writer = new CapturingRetryWriter();
+        using var log = new Serilog.LoggerConfiguration().CreateLogger();
+        using (var sink = new LocalDbStoreAndForwardSink(db, writer, log))
+            await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
+        using (var sink = new LocalDbStoreAndForwardSink(db, writer, log))
+            await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
+        writer.Events.Count.ShouldBe(2);
+        writer.Events.ShouldAllBe(evt => evt.DeliveryId == Guid.Parse("55372883-eaec-8dd7-a1a4-6f98f416a272"));
+    }
+
     // ---- the cases ---------------------------------------------------------------------------
 
     [Fact]

tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs

@@ -13,8 +13,8 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
 ///     comms-loss watchdog trips on). Recovery is the watchdog's sequence, replayed here as <c>DriverInstanceActor</c>
 ///     runs it: ShutdownAsync, InitializeAsync, re-subscribe — after which the value comes back Good. There is no
 ///     in-driver reconnect any more (#670: the SDK reconnect handler kept everything Good and Healthy forever).
-///     Since #676 a gracefully stopped server is a DEFINITIVE loss (<c>BadServerHalted</c>), so the driver also raises
-///     <c>ConnectionLost</c> and the host runs that sequence at once instead of after the watchdog's silence window.
+///     Since #676 definitive session loss also raises <c>ConnectionLost</c>, so the host runs that sequence at once
+///     instead of after the watchdog's silence window. Shutdown timing determines the SDK's loss status.
 /// </summary>
 [Trait("Category", "Integration")]
 public sealed class OpcUaClientDriverReconnectWireTests
@@ -72,12 +72,22 @@ public sealed class OpcUaClientDriverReconnectWireTests
             await Task.Delay(TimeSpan.FromSeconds(3), ct); // three keep-alive intervals
             drv.GetHealth().LastSuccessfulRead.ShouldBe(frozen);
 
-            // #676: a server that HALTED (graceful stop) is a definitive loss — measured here, the SDK reports
-            // BadServerHalted on every keep-alive and never re-opens the channel, so the session can never recover in
-            // place. The driver says so at once, so the host re-initialises it instead of waiting out the ~90 s watchdog.
+            // #676/#702: shutdown can report a halted server, an invalid/closed session, or a sustained closed
+            // channel. Each proves definitive loss; #702 reproduced BadSessionIdInvalid during this graceful stop.
+            // Deterministic ConnectionLossSignal tests cover the individual statuses and the sustained-channel gate.
             await WaitAsync(() => { lock (lost) return lost.Count > 0; },
                 "a halted upstream must raise ConnectionLost", TimeSpan.FromSeconds(10));
-            lock (lost) lost.Single().ShouldContain("BadServerHalted");
+            lock (lost)
+            {
+                var reason = lost.Single();
+                string[] definitiveStatuses =
+                [
+                    "BadSessionIdInvalid", "BadSessionClosed", "BadSessionNotActivated", "BadServerHalted",
+                    "BadSecureChannelClosed", "BadConnectionClosed",
+                ];
+                definitiveStatuses.ShouldContain(status => reason.EndsWith(status, StringComparison.Ordinal),
+                    $"the restart must report definitive loss, but reported: {reason}");
+            }
         });
 
         // The restarted server knows nothing of the old session, so the driver stays down on its own …

tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsTests.cs

@@ -295,12 +295,14 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
     /// HistoryRead call. The fake source MUST be invoked (proving we reached the bridge) and threw;
     /// the production catch now sets BOTH errors and results explicitly, so both are asserted here
     /// rather than relying on the SDK base pre-seeding results[i].</summary>
-    [Fact]
-    public async Task Events_backend_throw_yields_bad_status_and_does_not_escape()
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task Events_backend_throw_yields_bad_status_and_does_not_escape(bool communication)
     {
         var (host, server) = await BootAsync();
         var nm = server.NodeManager!;
-        var fake = new RecordingHistorianDataSource { ThrowOnRead = true };
+        var fake = new RecordingHistorianDataSource { ThrowOnRead = !communication, ReadException = communication ? new System.IO.IOException("lost connection") : null };
         nm.HistorianDataSource = fake;
 
         const string equipmentId = "eq-boom";
@@ -321,10 +323,10 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
 
         // Authoritative per-node signal: errors[0] must be Bad.
         ServiceResult.IsBad(errors[0]).ShouldBeTrue();
-        errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
+        errors[0].StatusCode.Code.ShouldBe(communication ? StatusCodes.BadNoCommunication : StatusCodes.BadHistoryOperationUnsupported);
         // The production catch now sets results[0] explicitly — assert it directly (not relying on base seeding).
         results[0].ShouldNotBeNull();
-        results[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
+        results[0].StatusCode.Code.ShouldBe(communication ? StatusCodes.BadNoCommunication : StatusCodes.BadHistoryOperationUnsupported);
         // The source WAS entered (proving the override reached the bridge and the throw was swallowed).
         fake.EventsEntered.ShouldBeTrue();
         // ThrowOnRead fires before EventsCalled/LastSourceName are set — that's the throw path.
@@ -380,6 +382,7 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
     private sealed class RecordingHistorianDataSource : IHistorianDataSource
     {
         public bool ThrowOnRead { get; init; }
+        public Exception? ReadException { get; init; }
         public HistoricalEventsResult EventsResult { get; set; } =
             new(Array.Empty<HistoricalEvent>(), null);
 
@@ -411,6 +414,7 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
             CancellationToken cancellationToken)
         {
             EventsEntered = true;
+            if (ReadException is not null) throw ReadException;
             if (ThrowOnRead) throw new InvalidOperationException("backend boom for events");
             EventsCalled = true;
             LastSourceName = sourceName;

tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadModifiedTests.cs

@@ -25,6 +25,34 @@ public sealed class NodeManagerHistoryReadModifiedTests : IDisposable
     private static readonly DateTime Epoch = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
     private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-modified-{Guid.NewGuid():N}");
 
+    [Fact]
+    public async Task Modified_values_use_the_scalar_node_type_and_preserve_the_audit_information()
+    {
+        var (host, nm, node, _) = await SetupAsync([Mod(1, 1, 2d, HistoryModificationType.Replace, "alice", Epoch)], "Float");
+        try
+        {
+            var r = Read(nm, node, Epoch, Epoch.AddHours(1), 10, null);
+            r.Data!.DataValues.Single().Value.ShouldBeOfType<float>().ShouldBe(2f);
+            r.Data.ModificationInfos.Single().UserName.ShouldBe("alice");
+            r.Data.ModificationInfos.Single().UpdateType.ShouldBe(HistoryUpdateType.Replace);
+        }
+        finally { await host.DisposeAsync(); }
+    }
+
+    [Fact]
+    public async Task A_modified_read_of_an_unreachable_backend_answers_BadNoCommunication()
+    {
+        var (host, nm, node, src) = await SetupAsync([]);
+        try
+        {
+            src.Throw = new System.IO.IOException("connection lost");
+            var r = Read(nm, node, Epoch, Epoch.AddHours(1), 10, null);
+            r.Error.StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
+            r.Result.StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
+        }
+        finally { await host.DisposeAsync(); }
+    }
+
     [Fact]
     public async Task A_modified_read_dispatches_to_the_trail_and_projects_modification_infos()
     {
@@ -289,7 +317,7 @@ public sealed class NodeManagerHistoryReadModifiedTests : IDisposable
     }
 
     private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId Node, SqlLikeTrail Src)> SetupAsync(
-        List<HistoryModifiedValue> trail)
+        List<HistoryModifiedValue> trail, string dataType = "Double")
     {
         var started = await OpcUaHostStart.StartAsync(
             port => new OpcUaApplicationHost(
@@ -307,7 +335,7 @@ public sealed class NodeManagerHistoryReadModifiedTests : IDisposable
         nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
         var src = new SqlLikeTrail(trail);
         nm.HistorianDataSource = src;
-        nm.EnsureVariable("eq-1/mod", null, "Mod", "Double", false, historianTagname: Tag, realm: AddressSpaceRealm.Uns);
+        nm.EnsureVariable("eq-1/mod", null, "Mod", dataType, false, historianTagname: Tag, realm: AddressSpaceRealm.Uns);
         return (started.Host, nm, nm.TryGetVariable("eq-1/mod")!.NodeId, src);
     }
 

tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs

@@ -786,6 +786,132 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
         return (results, errors);
     }
 
+    [Theory]
+    [InlineData("Raw")]
+    [InlineData("AtTime")]
+    [InlineData("Processed")]
+    public async Task An_unreachable_backend_answers_a_communication_status(string kind)
+    {
+        var (host, server) = await BootAsync();
+        try
+        {
+            var nm = server.NodeManager!;
+            nm.HistorianDataSource = new RecordingHistorianDataSource { ReadException = new System.IO.IOException("connection lost") };
+            nm.EnsureVariable("eq/comm", null, "Comm", "Double", false, historianTagname: "T", realm: AddressSpaceRealm.Uns);
+            var t = DateTime.UtcNow.AddMinutes(-1);
+            HistoryReadDetails details = kind switch
+            {
+                "Raw" => new ReadRawModifiedDetails { IsReadModified = false, ReturnBounds = false, StartTime = t, EndTime = t.AddSeconds(10) },
+                "AtTime" => new ReadAtTimeDetails { ReqTimes = new DateTimeCollection { t } },
+                _ => new ReadProcessedDetails { StartTime = t, EndTime = t.AddSeconds(10), ProcessingInterval = 1000,
+                    AggregateType = new NodeIdCollection { ObjectIds.AggregateFunction_Average } }
+            };
+            var (results, errors) = InvokeHistoryRead(server, nm, details, nm.TryGetVariable("eq/comm")!.NodeId);
+            errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
+            results[0].StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
+        }
+        finally { await host.DisposeAsync(); }
+    }
+
+    [Theory]
+    [InlineData("Int32", "Raw")]
+    [InlineData("Boolean", "Raw")]
+    [InlineData("Float", "Raw")]
+    [InlineData("Int32", "AtTime")]
+    [InlineData("Boolean", "AtTime")]
+    [InlineData("Float", "AtTime")]
+    public async Task History_values_use_the_nodes_scalar_type(string dataType, string kind)
+    {
+        var (host, server) = await BootAsync();
+        try
+        {
+            var nm = server.NodeManager!;
+            var t = DateTime.UtcNow.AddMinutes(-1);
+            var value = new HistorianRead([new DataValueSnapshot(1d, StatusCodes.Good, t, t)], null);
+            nm.HistorianDataSource = new RecordingHistorianDataSource { RawResult = value, AtTimeResult = value };
+            nm.EnsureVariable("eq/typed", null, "Typed", dataType, false, historianTagname: "T", realm: AddressSpaceRealm.Uns);
+            HistoryReadDetails details = kind == "Raw"
+                ? new ReadRawModifiedDetails { IsReadModified = false, ReturnBounds = false, StartTime = t, EndTime = t.AddSeconds(10) }
+                : new ReadAtTimeDetails { ReqTimes = new DateTimeCollection { t } };
+            var (results, errors) = InvokeHistoryRead(server, nm, details, nm.TryGetVariable("eq/typed")!.NodeId);
+            errors[0].StatusCode.Code.ShouldBe(StatusCodes.Good);
+            var actual = ((HistoryData)ExtensionObject.ToEncodeable(results[0].HistoryData)).DataValues.Single();
+            if (dataType == "Int32") actual.Value.ShouldBeOfType<int>().ShouldBe(1);
+            if (dataType == "Boolean") actual.Value.ShouldBeOfType<bool>().ShouldBeTrue();
+            if (dataType == "Float") actual.Value.ShouldBeOfType<float>().ShouldBe(1f);
+            actual.SourceTimestamp.ShouldBe(t);
+            actual.StatusCode.Code.ShouldBe(StatusCodes.Good);
+        }
+        finally { await host.DisposeAsync(); }
+    }
+
+    [Theory]
+    [InlineData("Int16", 40000d)]
+    [InlineData("DateTime", 1d)]
+    public async Task An_unrepresentable_historical_value_is_null_with_BadTypeMismatch(string type, double input)
+    {
+        var (host, server) = await BootAsync();
+        try
+        {
+            var nm = server.NodeManager!;
+            var t = DateTime.UtcNow.AddMinutes(-1);
+            nm.EnsureVariable("eq/mismatch", null, "Mismatch", type, false, historianTagname: "T", realm: AddressSpaceRealm.Uns);
+            nm.HistorianDataSource = new RecordingHistorianDataSource { RawResult = new HistorianRead([new(input, StatusCodes.Good, t, t)], null) };
+            var (results, _) = InvokeHistoryRead(server, nm, new ReadRawModifiedDetails { IsReadModified = false, ReturnBounds = false,
+                StartTime = t, EndTime = t.AddSeconds(10) }, nm.TryGetVariable("eq/mismatch")!.NodeId);
+            var value = ((HistoryData)ExtensionObject.ToEncodeable(results[0].HistoryData)).DataValues.Single();
+            value.Value.ShouldBeNull();
+            value.StatusCode.Code.ShouldBe(StatusCodes.BadTypeMismatch);
+            value.SourceTimestamp.ShouldBe(t);
+            value.ServerTimestamp.ShouldBe(t);
+        }
+        finally { await host.DisposeAsync(); }
+    }
+
+    [Fact]
+    public async Task Array_and_null_history_values_preserve_quality_and_timestamps()
+    {
+        var (host, server) = await BootAsync();
+        try
+        {
+            var nm = server.NodeManager!;
+            var t = DateTime.UtcNow.AddMinutes(-1);
+            var serverTime = t.AddSeconds(1);
+            nm.EnsureVariable("eq/array", null, "Array", "Int32", false, historianTagname: "T", realm: AddressSpaceRealm.Raw, isArray: true, arrayLength: 2);
+            nm.HistorianDataSource = new RecordingHistorianDataSource { RawResult = new HistorianRead([
+                new(new int[] { 1, 2 }, StatusCodes.Uncertain, t, serverTime),
+                new(null, StatusCodes.BadNoCommunication, t.AddSeconds(1), serverTime)], null) };
+            var (results, _) = InvokeHistoryRead(server, nm, new ReadRawModifiedDetails { IsReadModified = false, ReturnBounds = false,
+                StartTime = t, EndTime = t.AddSeconds(10) }, nm.TryGetVariable("eq/array", AddressSpaceRealm.Raw)!.NodeId);
+            var values = ((HistoryData)ExtensionObject.ToEncodeable(results[0].HistoryData)).DataValues;
+            values[0].Value.ShouldBeOfType<int[]>().ShouldBe([1, 2]);
+            values[0].StatusCode.Code.ShouldBe(StatusCodes.Uncertain);
+            values[0].SourceTimestamp.ShouldBe(t);
+            values[0].ServerTimestamp.ShouldBe(serverTime);
+            values[1].Value.ShouldBeNull();
+            values[1].StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
+        }
+        finally { await host.DisposeAsync(); }
+    }
+
+    [Fact]
+    public async Task Processed_Count_keeps_its_aggregate_type_on_a_Boolean_node()
+    {
+        var (host, server) = await BootAsync();
+        try
+        {
+            var nm = server.NodeManager!;
+            var t = DateTime.UtcNow.AddMinutes(-1);
+            nm.EnsureVariable("eq/count", null, "Count", "Boolean", false, historianTagname: "T", realm: AddressSpaceRealm.Uns);
+            nm.HistorianDataSource = new RecordingHistorianDataSource { ProcessedResult = new HistorianRead([new(3d, StatusCodes.Good, t, t)], null) };
+            var (results, _) = InvokeHistoryRead(server, nm, new ReadProcessedDetails { StartTime = t, EndTime = t.AddSeconds(10),
+                ProcessingInterval = 1000, AggregateType = new NodeIdCollection { ObjectIds.AggregateFunction_Count } }, nm.TryGetVariable("eq/count")!.NodeId);
+            var value = ((HistoryData)ExtensionObject.ToEncodeable(results[0].HistoryData)).DataValues.Single();
+            value.Value.ShouldBeOfType<double>().ShouldBe(3d);
+        }
+        finally { await host.DisposeAsync(); }
+    }
+
     private static HistorianRead Empty() => new(Array.Empty<DataValueSnapshot>(), null);
 
     /// <summary>A recording fake historian source — captures the last call's kind + arguments and returns
@@ -793,6 +919,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
     private sealed class RecordingHistorianDataSource : IHistorianDataSource
     {
         public bool ThrowOnRead { get; init; }
+        public Exception? ReadException { get; init; }
         public bool AggregateNotSupported { get; init; }
         public bool LimitExceeded { get; init; }
         public IReadOnlySet<HistoryAggregateType>? Supported { get; init; } = HistoryAggregateTypes.All;
@@ -814,6 +941,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
             string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
             CancellationToken cancellationToken)
         {
+            if (ReadException is not null) throw ReadException;
             if (ThrowOnRead) throw new InvalidOperationException("backend boom");
             LastCall = "Raw";
             LastTagname = fullReference;
@@ -827,6 +955,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
             string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
             HistoryAggregateType aggregate, CancellationToken cancellationToken)
         {
+            if (ReadException is not null) throw ReadException;
             if (ThrowOnRead) throw new InvalidOperationException("backend boom");
             if (AggregateNotSupported) throw new HistoryAggregateNotSupportedException(aggregate);
             if (LimitExceeded) throw new HistoryReadLimitExceededException(4, "too many raw rows");
@@ -842,6 +971,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
         public Task<HistorianRead> ReadAtTimeAsync(
             string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
         {
+            if (ReadException is not null) throw ReadException;
             if (ThrowOnRead) throw new InvalidOperationException("backend boom");
             LastCall = "AtTime";
             LastTagname = fullReference;