merge: comms-loss watchdog — DisconnectObserved gets its first production sender (#527)

Closes the defect the #481 live gate surfaced: a driver whose device died mid-run
was never detected and never recovered. Live-gated twice on docker-dev.

Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo

CLAUDE.md

@@ -210,6 +210,38 @@ nothing ever writes a `DriverHostStatus` row (the table was re-created in the v3
 confirm intent before deleting). The separable `OnHostStatusChanged` event is the useful half. Tracked
 as Gitea **#521**.
 
+### Comms-loss detection: `GetHealth()` is load-bearing (Gitea #527, closed)
+
+Until #527, **`DriverInstanceActor.DisconnectObserved` had no production sender.** The only paths into
+`Reconnecting` were a failed *initial* connect and an operator's Force Reconnect, so a driver that connected
+and then lost its device stayed `Connected` indefinitely — publishing Bad for every tag, never
+re-initialising, and **not recovering when the device returned** (live-confirmed: a Modbus driver against a
+stopped fixture needed a node restart). `PublishHealthSnapshot` read `GetHealth()` for *reporting only*.
+
+The `Connected` health tick now runs a watchdog that self-sends `DisconnectObserved` when a driver reports a
+non-`Healthy` state **and** its `LastSuccessfulRead` has been frozen for 90 s. Things to know before touching it:
+
+- ⚠️ **The conjunction is the design — "non-Healthy for N polls" is wrong and was the issue's own proposal.**
+  `Degraded` means *connected but some data is bad*, and several drivers write health **per tag inside the
+  read loop** (`ModbusDriver.ReadAsync`), so one unreadable address latches `Degraded` forever against a
+  perfectly reachable device. State-only tripping would reconnect that driver every 90 s indefinitely.
+  `LastSuccessfulRead` is the honest evidence — preserved across `Degraded`/`Faulted` writes, advanced by any
+  success. Each conjunct has a test that goes red when the *other* is deleted.
+- **The mechanism rests on every driver reaching a non-Healthy state on sustained failure.** Surveyed across
+  all 12: nine poll drivers via `Degraded`, MQTT via live connection state, Galaxy via its reconnect
+  supervisor. `CalculationDriver` reports `Healthy` while running and so can never trip — correct, as it has
+  no transport. Re-check this if a new driver lands.
+- ⚠️ **The reconnect retry MUST tear down before re-initialising.** `ModbusDriver` (`_transport`) and
+  `MqttDriver` (`_connection`) overwrite an `IAsyncDisposable` field inside `InitializeAsync` without
+  disposing the old one, so re-initialising without `ShutdownAsync` leaks a socket per attempt — ~360 for an
+  hour-long outage at the 10 s retry interval. Latent while only a manual Force Reconnect reached this path.
+- The threshold is hardcoded (`DefaultCommsLossTimeout`), matching `HealthPollInterval` /
+  `DefaultReconnectInterval`. Unlike the #523 drift detector this generates **no** traffic on a healthy
+  system — it is a local `GetHealth()` read, and only reaches the network once a driver has already stopped
+  reading.
+
+See `docs/DriverLifecycle.md` §"The comms-loss watchdog".
+
 ## mxaccessgw
 
 The gateway lives in a sibling repo at `~/Desktop/MxAccessGateway/` (`mxaccessgw` — see [Related Projects](#related-projects-scadaproj-umbrella)). See `docs/v2/Galaxy.ParityRig.md` for the gw setup recipe (build, API key provisioning via `apikey create-key`, env-var overrides for HTTP/2 cleartext + worker path). The gw's MXAccess Toolkit reference (its `gateway.md`) is the canonical MxAccess API doc; the standalone `mxaccess_documentation.md` previously kept in this repo retired in PR 7.3.

docs/AlarmTracking.md

@@ -183,15 +183,22 @@ Two things worth knowing before changing this:
   and is the correct reading of "this input's source is unreachable", but it is a behaviour change beyond
   scripted alarms.
 
-⚠️ **The signal this hangs off does not currently fire for a mid-run device loss — see #527.** The #481 live
-gate on docker-dev found that `DriverInstanceActor.DisconnectObserved` has **no production sender**: the only
-paths into `Reconnecting` are a failed *initial* connect and an operator Force Reconnect. A Modbus driver
-whose fixture was stopped stayed `Connected` for >10 minutes publishing per-tag Bad, and did not recover when
-the fixture came back until the node was restarted. So the fan-out above — and the native `#477` Layer 2
-annotation it mirrors — are correctly wired to a signal that a dying device never raises. What partially
-covers the gap today is that Modbus publishes Bad *per tag* rather than going silent, so #478 Layer 3 does
-carry Bad into scripted input quality for that driver; that is a per-driver accident, not a guarantee, and it
-does nothing for native conditions or for recovery. Closing #527 is what makes this path reachable.
+**The signal this hangs off now fires for a mid-run device loss (#527, closed).** The #481 live gate on
+docker-dev found that `DriverInstanceActor.DisconnectObserved` had **no production sender**: the only paths
+into `Reconnecting` were a failed *initial* connect and an operator Force Reconnect. A Modbus driver whose
+fixture was stopped stayed `Connected` for >10 minutes publishing per-tag Bad, and did not recover when the
+fixture came back until the node was restarted. So the fan-out above — and the native #477 Layer 2
+annotation it mirrors — were correctly wired to a signal that a dying device never raised. #527 added the
+comms-loss watchdog on the Connected health tick, which self-sends `DisconnectObserved` once a driver has
+gone 90 s without a successful read *while* reporting itself non-Healthy. See
+[`DriverLifecycle.md`](DriverLifecycle.md) §"The comms-loss watchdog" for why the trip condition is that
+conjunction and not simply "non-Healthy for N polls" — the naive rule would reconnect a healthy device's
+driver every 90 s forever whenever one authored address is unreadable.
+
+Note the partial cover that existed before, because it explains why *some* of this appeared to work: Modbus
+publishes Bad *per tag* rather than going silent, so #478 Layer 3 did carry Bad into scripted input quality
+for that driver. That was a per-driver accident, not a guarantee, and it did nothing for native conditions
+or for recovery.
 
 **Cold start vs. a driver reporting Bad (#478 Finding 3, closed by #481).** `WorstInputStatus` used to skip
 every input whose *value* was null as "no data yet". A comms-lost driver produces exactly that shape — Bad

docs/DriverLifecycle.md

@@ -219,6 +219,57 @@ So a driver whose `InitializeAsync` disposes anything on its failure path must b
 `MTConnectDriver` does today, because it is the only one that tears down inside a `catch`. The
 shutdown paths deliberately keep the single trailing write.
 
+### The comms-loss watchdog (Gitea #527) — `GetHealth()` is now load-bearing
+
+`GetHealth()` used to be **reporting-only**: `DriverInstanceActor.PublishHealthSnapshot` read it every 30 s
+and forwarded it to the AdminUI, and nothing ever acted on a driver that reported itself unhealthy. That
+left `DisconnectObserved` with **no production sender at all** — the only paths into `Reconnecting` were a
+failed *initial* connect and an operator's Force Reconnect. A driver that connected and then lost its
+device stayed `Connected` forever: publishing Bad for every tag, never re-initialising, and **not
+recovering when the device came back** (live-confirmed against a stopped Modbus fixture, which needed a
+node restart). Both comms-loss quality paths — #477's native-condition annotation and #481's
+scripted-alarm input fan-out — hang off `ConnectivityChanged`, so both were wired to a signal that never
+fired.
+
+The `Connected`-state health tick now also evaluates a watchdog, and self-sends `DisconnectObserved` when
+**both** of these hold:
+
+1. the driver reports a state other than `Healthy`, **and**
+2. `DriverHealth.LastSuccessfulRead` has not advanced for `DefaultCommsLossTimeout` (90 s = three ticks),
+   measured from the Connected entry when the driver has never reported a successful read.
+
+**The conjunction is the whole design, and "non-Healthy for N polls" is the wrong rule.** `Degraded` means
+*connected but some data is bad* — the enum says so. Several drivers publish health **per tag inside their
+read loop** (`ModbusDriver.ReadAsync` writes `Healthy` or `Degraded` once per reference), so a single
+unreadable address leaves the driver latched `Degraded` whenever it happens to be read last, indefinitely,
+against a device that is perfectly reachable. A state-only watchdog would tear that driver down every 90 s
+forever — strictly worse than the bug it fixes. `LastSuccessfulRead` is the honest liveness evidence: every
+driver advances it on any successful read and *preserves* it across `Degraded`/`Faulted` writes, so one bad
+tag among good ones keeps it moving while a device that is genuinely gone freezes it. The state check is
+still needed as the second conjunct, because a `Healthy` driver with nothing subscribed legitimately never
+reads and must not be declared dead for being idle. Both halves are pinned by their own test in
+`DriverInstanceActorCommsLossWatchdogTests`, each of which goes red when the *other* conjunct is deleted.
+
+Two consequences worth knowing:
+
+- **`CalculationDriver` can never trip it, correctly.** It reports `Healthy` whenever it is running and has
+  no transport to be degraded about. Every other driver reaches a non-Healthy state on sustained failure:
+  the nine poll drivers via `Degraded`, MQTT via its live connection state, Galaxy via its reconnect
+  supervisor. This was surveyed across all 12 before relying on it — it is the assumption the whole
+  mechanism rests on.
+- **A device that accepts a connection but serves no data will cycle**, reconnecting once per threshold.
+  That is the correct reading of a broken device, it is rate-limited, and it is visible in the health
+  snapshot's error count — but do not mistake it for a bug when reading logs.
+
+**The reconnect retry now tears down before re-initialising.** `Reconnecting`'s `RetryConnect` calls
+`ShutdownAsync` before `InitializeAsync` — the sequence every driver's own `ReinitializeAsync` already
+uses — because several drivers **overwrite** an `IAsyncDisposable` transport field inside `InitializeAsync`
+without disposing the previous one (`ModbusDriver._transport`, `MqttDriver._connection`). Re-initialising
+without a teardown leaks one socket per attempt. That was latent while only a manual Force Reconnect
+re-entered this path; the watchdog makes it automatic and unbounded, so a device down for an hour would
+have leaked ~360 transports at the 10 s retry interval. The shutdown is best-effort: a driver that throws
+on it still gets its initialise attempt.
+
 ---
 
 ## IDriverConfigEditor — custom AdminUI config editor (plug-point)

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs

@@ -141,6 +141,23 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
     /// <summary>Interval between periodic health-poll heartbeats sent to the snapshot store.</summary>
     public static readonly TimeSpan HealthPollInterval = TimeSpan.FromSeconds(30);
 
+    /// <summary>
+    ///     How long a Connected driver may go without a single successful read — while also reporting
+    ///     itself non-Healthy — before the actor declares comms lost (Gitea #527). Evaluated on the
+    ///     existing <see cref="HealthPollInterval"/> tick, so the effective detection granularity is
+    ///     30 s and the worst-case detection latency is this value plus one tick.
+    /// </summary>
+    /// <remarks>
+    ///     90 s = three health-poll ticks. Long enough that a device answering slowly under load is not
+    ///     torn down, short enough that an operator watching a dead PLC sees it inside two minutes.
+    ///     Hardcoded rather than exposed in appsettings, matching <see cref="HealthPollInterval"/> and
+    ///     <see cref="DefaultReconnectInterval"/>. Unlike the periodic drift-detector removed by Gitea
+    ///     #523, this generates <b>no</b> outbound traffic on a healthy system: the watchdog is a local
+    ///     read of <see cref="IDriver.GetHealth"/>, and it only reaches the network once a driver has
+    ///     already stopped reading successfully.
+    /// </remarks>
+    public static readonly TimeSpan DefaultCommsLossTimeout = TimeSpan.FromSeconds(90);
+
     private readonly IDriver _driver;
 
     /// <summary>Phase 6.1 resilience seam: every guarded driver-capability call this actor makes is
@@ -158,6 +175,14 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
 
     private readonly TimeSpan _healthPollInterval;
 
+    /// <summary>Comms-loss watchdog threshold (#527); <see cref="TimeSpan.Zero"/> or negative disables it.</summary>
+    private readonly TimeSpan _commsLossTimeout;
+
+    /// <summary>When the actor last entered <c>Connected</c>. The comms-loss watchdog's baseline for a
+    /// driver that has never reported a successful read, so "connected but never read anything" is
+    /// measurable rather than indefinitely undefined.</summary>
+    private DateTime _connectedSinceUtc;
+
     private readonly ILoggingAdapter _log = Context.GetLogger();
     private string? _currentConfigJson;
 
@@ -229,6 +254,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
     /// <param name="healthPollInterval">Optional period of the health-poll heartbeat, which also drives the
     /// subscription reconcile (<see cref="ReconcileSubscription"/>); defaults to
     /// <see cref="HealthPollInterval"/>. Exists so a test can drive the reconcile without waiting 30 s.</param>
+    /// <param name="commsLossTimeout">Optional comms-loss watchdog threshold (#527); defaults to
+    /// <see cref="DefaultCommsLossTimeout"/>. <see cref="TimeSpan.Zero"/> disables the watchdog — used by
+    /// tests that need a Connected driver to stay Connected while reporting non-Healthy.</param>
     /// <returns>A <see cref="Akka.Actor.Props"/> instance configured to create the wrapped <see cref="DriverInstanceActor"/>.</returns>
     public static Props Props(
         IDriver driver,
@@ -237,7 +265,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
         IDriverHealthPublisher? healthPublisher = null,
         string? clusterId = null,
         IDriverCapabilityInvoker? invoker = null,
-        TimeSpan? healthPollInterval = null) =>
+        TimeSpan? healthPollInterval = null,
+        TimeSpan? commsLossTimeout = null) =>
         Akka.Actor.Props.Create(() => new DriverInstanceActor(
             driver,
             reconnectInterval ?? DefaultReconnectInterval,
@@ -245,7 +274,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
             healthPublisher ?? NullDriverHealthPublisher.Instance,
             clusterId ?? string.Empty,
             invoker,
-            healthPollInterval));
+            healthPollInterval,
+            commsLossTimeout));
 
     /// <summary>
     /// Returns true when the driver should boot in DEV-STUB mode based on host platform and
@@ -278,6 +308,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
     /// defaults to <see cref="NullDriverCapabilityInvoker"/> (pass-through) when null.</param>
     /// <param name="healthPollInterval">Period of the health-poll heartbeat, which also drives the
     /// subscription reconcile; defaults to <see cref="HealthPollInterval"/> when null.</param>
+    /// <param name="commsLossTimeout">Comms-loss watchdog threshold (#527); defaults to
+    /// <see cref="DefaultCommsLossTimeout"/> when null, and is disabled when zero or negative.</param>
     public DriverInstanceActor(
         IDriver driver,
         TimeSpan reconnectInterval,
@@ -285,7 +317,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
         IDriverHealthPublisher? healthPublisher = null,
         string? clusterId = null,
         IDriverCapabilityInvoker? invoker = null,
-        TimeSpan? healthPollInterval = null)
+        TimeSpan? healthPollInterval = null,
+        TimeSpan? commsLossTimeout = null)
     {
         _driver = driver;
         _invoker = invoker ?? NullDriverCapabilityInvoker.Instance;
@@ -294,6 +327,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
         _healthPublisher = healthPublisher ?? NullDriverHealthPublisher.Instance;
         _reconnectInterval = reconnectInterval;
         _healthPollInterval = healthPollInterval ?? HealthPollInterval;
+        _commsLossTimeout = commsLossTimeout ?? DefaultCommsLossTimeout;
         OtOpcUaTelemetry.DriverInstanceLifecycle.Add(1,
             new KeyValuePair<string, object?>("event", startStubbed ? "spawn_stub" : "spawn"),
             new KeyValuePair<string, object?>("driver_type", driver.DriverType));
@@ -403,6 +437,10 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
         // non-Good "waiting for initial data" quality at materialise, and this is what confirms them Good.
         Context.Parent.Tell(new ConnectivityChanged(_driverInstanceId, Connected: true));
 
+        // #527 — baseline for the comms-loss watchdog. Re-stamped on every Connected entry so a driver that
+        // reconnects but never manages a read is measured from THIS connect, not from a stale earlier one.
+        _connectedSinceUtc = DateTime.UtcNow;
+
         ReceiveAsync<ApplyDelta>(HandleApplyDeltaAsync);
         Receive<DisconnectObserved>(msg =>
         {
@@ -456,11 +494,76 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
         Receive<HostStatusRaised>(_ => PublishHealthSnapshot());
         Receive<HealthPollTick>(_ =>
         {
-            PublishHealthSnapshot();
+            // One GetHealth observation feeds both the snapshot publish and the watchdog — reading it
+            // twice would let the two disagree about the same tick.
+            var health = PublishHealthSnapshot();
+            EvaluateCommsLossWatchdog(health);
             ReconcileSubscription();
         });
     }
 
+    /// <summary>
+    ///     Gitea #527 — the comms-loss watchdog. Declares a Connected driver disconnected once it has gone
+    ///     <see cref="_commsLossTimeout"/> without a single successful read <b>while also</b> reporting
+    ///     itself non-Healthy, by self-sending <see cref="DisconnectObserved"/>.
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///     <b>Why this exists.</b> <see cref="DisconnectObserved"/> had no production sender at all. The only
+    ///     ways into <c>Reconnecting</c> were a failed <em>initial</em> connect and an operator's Force
+    ///     Reconnect, so a driver that connected successfully and then lost its device stayed Connected
+    ///     forever: it published Bad for every tag, never re-initialised, and did not recover when the device
+    ///     came back — live-confirmed against a stopped Modbus fixture, which needed a node restart. Both
+    ///     comms-loss quality paths (#477 native conditions, #481 scripted-alarm inputs) hang off
+    ///     <see cref="ConnectivityChanged"/>, so both were wired to a signal that never fired.
+    ///     <c>OpcUaClientDriver</c>'s own reconnect-exhausted branch already assumed this mechanism existed
+    ///     ("the Core never learns it must offer an operator reinitialize") — it did not.
+    ///     </para>
+    ///     <para>
+    ///     <b>Why the trip condition is a conjunction, and NOT simply "non-Healthy for N polls".</b>
+    ///     <see cref="DriverState.Degraded"/> means <em>connected but some data is bad</em> — the enum says
+    ///     so. Several drivers publish health per-tag inside their read loop (<c>ModbusDriver.ReadAsync</c>
+    ///     writes Healthy or Degraded once per reference), so a single unreadable address leaves the driver
+    ///     Degraded whenever it happens to be the last tag read, indefinitely, against a perfectly healthy
+    ///     device. Tripping on state alone would tear that driver down every watchdog period forever —
+    ///     strictly worse than the bug being fixed.
+    ///     </para>
+    ///     <para>
+    ///     <see cref="DriverHealth.LastSuccessfulRead"/> is the honest liveness evidence: every driver
+    ///     advances it on any successful read and <em>preserves</em> it across Degraded/Faulted writes. One
+    ///     bad tag among good ones keeps it moving; a device that is genuinely gone freezes it. The state
+    ///     check is still required as the second conjunct, because a Healthy driver with nothing subscribed
+    ///     legitimately never reads at all and must not be declared dead for being idle.
+    ///     </para>
+    ///     <para>
+    ///     <b>Drivers this cannot fire for, by construction.</b> <c>CalculationDriver</c> reports Healthy
+    ///     whenever it is running (it has no transport to be degraded about), so it never trips — correct,
+    ///     since there is nothing to reconnect to. Every other driver in the tree reaches a non-Healthy
+    ///     state on sustained failure: the nine poll drivers via Degraded, MQTT via its live connection
+    ///     state, Galaxy via its reconnect supervisor.
+    ///     </para>
+    /// </remarks>
+    /// <param name="health">The health observed on this tick, or null when <see cref="IDriver.GetHealth"/>
+    /// threw — a failed observation is not evidence of comms loss, so it never advances the watchdog.</param>
+    private void EvaluateCommsLossWatchdog(DriverHealth? health)
+    {
+        if (_commsLossTimeout <= TimeSpan.Zero) return;
+        if (health is not { } observed) return;
+        if (observed.State == DriverState.Healthy) return;
+
+        // A driver that has never reported a successful read is measured from the current Connected entry.
+        var lastEvidence = observed.LastSuccessfulRead ?? _connectedSinceUtc;
+        var silence = DateTime.UtcNow - lastEvidence;
+        if (silence < _commsLossTimeout) return;
+
+        _log.Warning(
+            "DriverInstance {Id}: no successful read for {Silence} while reporting {State} ({Error}) — " +
+            "declaring comms lost and reconnecting",
+            _driverInstanceId, silence, observed.State, observed.LastError ?? "no error reported");
+        Self.Tell(new DisconnectObserved(
+            $"no successful read for {silence.TotalSeconds:F0}s while reporting {observed.State}"));
+    }
+
     /// <summary>
     ///     Periodic desired-vs-actual subscription reconcile (#488), riding the existing <c>health-poll</c>
     ///     tick. While Connected, a driver that has a desired subscription set but no live handle is
@@ -514,7 +617,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
         // (a comms-lost driver emits no alarm events, so this is the ONLY signal that the source is unreachable).
         Context.Parent.Tell(new ConnectivityChanged(_driverInstanceId, Connected: false));
 
-        Receive<RetryConnect>(_ => InitializeAsync(_currentConfigJson ?? "{}"));
+        // #527 — tear the dead transport down before re-initialising, or every retry leaks one.
+        Receive<RetryConnect>(_ => InitializeAsync(_currentConfigJson ?? "{}", teardownFirst: true));
         // Fast-fail writes while reconnecting (same reason as Connecting — avoids the 8s host Ask
         // timeout on an inbound write to a transiently-down driver). Synchronous Receive.
         Receive<WriteAttribute>(_ =>
@@ -562,16 +666,45 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
         Timers.StartPeriodicTimer("retry-connect", RetryConnect.Instance, _reconnectInterval);
     }
 
-    private void InitializeAsync(string driverConfigJson)
+    /// <param name="driverConfigJson">The config blob to initialise the driver with.</param>
+    /// <param name="teardownFirst">
+    ///     When true, <see cref="IDriver.ShutdownAsync"/> is awaited before <see cref="IDriver.InitializeAsync"/>
+    ///     — the sequence every driver's own <c>ReinitializeAsync</c> uses. Set on the <c>Reconnecting</c>
+    ///     retry path (#527), because several drivers <em>overwrite</em> their transport field in
+    ///     <c>InitializeAsync</c> without disposing the previous one: <c>ModbusDriver</c> assigns
+    ///     <c>_transport = _transportFactory(...)</c> and <c>MqttDriver</c> assigns <c>_connection</c>, both
+    ///     over an <see cref="IAsyncDisposable"/>. Re-initialising without tearing down therefore leaks one
+    ///     socket (and, for Modbus, a poll engine) per attempt. That was latent while nothing but an
+    ///     operator's Force Reconnect ever re-entered this path; the watchdog makes it automatic and
+    ///     unbounded — a device down for an hour would leak ~360 transports at the 10 s retry interval.
+    ///     Best-effort and never fatal: a driver that throws on shutdown still gets its initialise attempt.
+    /// </param>
+    private void InitializeAsync(string driverConfigJson, bool teardownFirst = false)
     {
         _currentConfigJson = driverConfigJson;
         var generation = ++_initGeneration;
         var self = Self;
+        var log = _log;
+        var driverInstanceId = _driverInstanceId;
+        var driver = _driver;
         _ = Task.Run(async () =>
         {
             try
             {
-                await _driver.InitializeAsync(driverConfigJson, CancellationToken.None);
+                if (teardownFirst)
+                {
+                    try
+                    {
+                        await driver.ShutdownAsync(CancellationToken.None);
+                    }
+                    catch (Exception ex)
+                    {
+                        log.Warning(ex,
+                            "DriverInstance {Id}: ShutdownAsync threw before a reconnect attempt; " +
+                            "initialising anyway", driverInstanceId);
+                    }
+                }
+                await driver.InitializeAsync(driverConfigJson, CancellationToken.None);
                 self.Tell(new InitializeSucceeded(generation));
             }
             catch (Exception ex)
@@ -1006,11 +1139,16 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
     ///     every 30s when nothing has changed. Newly-joined SignalR clients still get the current
     ///     snapshot via <c>DriverStatusHub.JoinDriver</c> which reads the store directly.
     /// </summary>
-    private void PublishHealthSnapshot()
+    /// <returns>The health observed on this call, or null when <see cref="IDriver.GetHealth"/> threw.
+    /// Returned so the caller can reuse the same observation — the #527 comms-loss watchdog rides this
+    /// call rather than polling the driver a second time and risking a different answer.</returns>
+    private DriverHealth? PublishHealthSnapshot()
     {
+        DriverHealth? observed = null;
         try
         {
             var health = _driver.GetHealth();
+            observed = health;
             var errorCount = ErrorCount5Min();
             var hostStatuses = CurrentHostStatuses();
             // _rediscoveryNeededUtc is PART OF THE FINGERPRINT on purpose. A rediscovery raise on an
@@ -1027,7 +1165,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
             var fingerprint = (health.State, health.LastSuccessfulRead, health.LastError, errorCount,
                 _rediscoveryNeededUtc, HostStatusDigest(hostStatuses));
             if (_lastPublishedFingerprint is { } prev && prev.Equals(fingerprint))
-                return;
+                return observed;
             _lastPublishedFingerprint = fingerprint;
             _healthPublisher.Publish(
                 _clusterId, _driverInstanceId, health, errorCount, _rediscoveryNeededUtc, _rediscoveryReason,
@@ -1037,6 +1175,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
         {
             _log.Warning(ex, "DriverInstance {Id}: GetHealth threw during health publish; skipping", _driverInstanceId);
         }
+        return observed;
     }
 
     /// <summary>Order-insensitive value digest of a host-status list for the publish fingerprint. Null in,

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorCommsLossWatchdogTests.cs

@@ -0,0 +1,330 @@
+using Akka.Actor;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
+using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
+
+/// <summary>
+///     Gitea #527 — the comms-loss watchdog that gives <see cref="DriverInstanceActor.DisconnectObserved"/>
+///     its first production sender.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>The defect.</b> Nothing in <c>src/</c> ever sent <c>DisconnectObserved</c>. A driver that
+///         connected and then lost its device stayed <c>Connected</c> indefinitely — publishing Bad for
+///         every tag, never re-initialising, and not recovering when the device returned. Live-confirmed
+///         against a stopped Modbus fixture, which needed a node restart.
+///     </para>
+///     <para>
+///         <b>What these tests are really pinning is the trip CONDITION, not the plumbing.</b> The obvious
+///         design — "non-Healthy for N consecutive polls" — is wrong, and
+///         <see cref="A_degraded_driver_that_is_still_reading_successfully_is_left_alone"/> is the test that
+///         says so. <see cref="DriverState.Degraded"/> means <i>connected but some data is bad</i>, and
+///         several drivers write health per-tag inside their read loop, so one unreadable address can leave
+///         a perfectly healthy device's driver Degraded forever. Tripping on state alone would reconnect
+///         that driver every 90 s in production — worse than the bug being fixed. The conjunction with a
+///         frozen <see cref="DriverHealth.LastSuccessfulRead"/> is what makes the signal honest, and
+///         <see cref="An_idle_healthy_driver_with_a_stale_last_read_is_left_alone"/> pins the other half.
+///     </para>
+///     <para>
+///         Every absence assertion carries a positive control — a proven number of health-poll ticks
+///         actually processed (<see cref="HealthControlledStubDriver.HealthReadCount"/>) — so "the watchdog
+///         did not fire" can never pass merely because nothing ran.
+///     </para>
+/// </remarks>
+public sealed class DriverInstanceActorCommsLossWatchdogTests : RuntimeActorTestBase
+{
+    private static readonly TimeSpan FastPoll = TimeSpan.FromMilliseconds(50);
+    private static readonly TimeSpan FastTimeout = TimeSpan.FromMilliseconds(250);
+
+    /// <summary>Drives the actor to Connected and returns once the connect has been announced.</summary>
+    private (IActorRef Actor, Akka.TestKit.TestProbe Parent) ConnectDriver(
+        HealthControlledStubDriver driver,
+        TimeSpan? commsLossTimeout = null,
+        TimeSpan? reconnectInterval = null)
+    {
+        var parent = CreateTestProbe();
+        var actor = parent.ChildActorOf(DriverInstanceActor.Props(
+            driver,
+            reconnectInterval: reconnectInterval ?? TimeSpan.FromSeconds(30),
+            healthPollInterval: FastPoll,
+            commsLossTimeout: commsLossTimeout ?? FastTimeout));
+
+        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
+        // Connected entry announces Connected=true; consuming it here leaves the probe holding only
+        // whatever the watchdog does (or does not) send next.
+        parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(TimeSpan.FromSeconds(3))
+            .Connected.ShouldBeTrue();
+        return (actor, parent);
+    }
+
+    /// <summary>
+    ///     The core of #527: a Connected driver whose device has gone away — non-Healthy, and no successful
+    ///     read since well before the threshold — is declared disconnected and announces comms loss to the
+    ///     host, which is the signal #477 and #481 both hang off.
+    /// </summary>
+    [Fact]
+    public void A_connected_driver_that_stops_reading_successfully_is_declared_disconnected()
+    {
+        var driver = new HealthControlledStubDriver
+        {
+            // The live shape: reads fail, the driver reports Degraded, and LastSuccessfulRead is frozen
+            // at the instant the device died.
+            Health = new DriverHealth(DriverState.Degraded, DateTime.UtcNow.AddMinutes(-5), "socket timeout"),
+        };
+        var (_, parent) = ConnectDriver(driver);
+
+        parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(TimeSpan.FromSeconds(3))
+            .Connected.ShouldBeFalse();
+    }
+
+    /// <summary>
+    ///     The falsifying control for the trip condition, and the reason it is not "non-Healthy for N polls".
+    ///     This driver is <b>permanently Degraded</b> — modelling one unreadable address among many, which
+    ///     <c>ModbusDriver.ReadAsync</c> leaves latched whenever the bad tag is read last — but it is still
+    ///     reading successfully, so <c>LastSuccessfulRead</c> keeps advancing. It must be left alone.
+    ///     A state-only watchdog would tear this driver down on every threshold, forever.
+    /// </summary>
+    [Fact]
+    public void A_degraded_driver_that_is_still_reading_successfully_is_left_alone()
+    {
+        var driver = new HealthControlledStubDriver
+        {
+            // Recomputed per call: Degraded, but LastSuccessfulRead is always fresh.
+            HealthFactory = () => new DriverHealth(DriverState.Degraded, DateTime.UtcNow, "one bad tag"),
+        };
+        var (_, parent) = ConnectDriver(driver);
+
+        // Positive control: prove the watchdog genuinely ran, many times, over a span far longer than the
+        // trip threshold. Without this the absence assertion below would pass against a watchdog that
+        // never executed at all.
+        AwaitCondition(
+            () => driver.HealthReadCount >= 12,
+            TimeSpan.FromSeconds(5),
+            "health-poll ticks did not run — the absence assertion below would be vacuous");
+        driver.ObservedSpan.ShouldBeGreaterThan(FastTimeout, "the observation window never exceeded the trip threshold");
+
+        parent.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
+    }
+
+    /// <summary>
+    ///     The other conjunct. A Healthy driver with nothing subscribed legitimately never reads, so its
+    ///     <c>LastSuccessfulRead</c> is arbitrarily stale. Staleness alone must not declare it dead.
+    /// </summary>
+    [Fact]
+    public void An_idle_healthy_driver_with_a_stale_last_read_is_left_alone()
+    {
+        var driver = new HealthControlledStubDriver
+        {
+            Health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow.AddHours(-2), null),
+        };
+        var (_, parent) = ConnectDriver(driver);
+
+        AwaitCondition(
+            () => driver.HealthReadCount >= 12,
+            TimeSpan.FromSeconds(5),
+            "health-poll ticks did not run — the absence assertion below would be vacuous");
+
+        parent.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
+    }
+
+    /// <summary>
+    ///     A driver that connects but never manages a single read reports <c>LastSuccessfulRead = null</c>,
+    ///     so staleness has no natural baseline. The watchdog measures from the Connected entry instead —
+    ///     otherwise a never-read driver would be the one case that could never be declared dead.
+    /// </summary>
+    [Fact]
+    public void A_driver_that_never_reads_after_connecting_is_declared_disconnected()
+    {
+        var driver = new HealthControlledStubDriver
+        {
+            Health = new DriverHealth(DriverState.Degraded, LastSuccessfulRead: null, "never read"),
+        };
+        var (_, parent) = ConnectDriver(driver);
+
+        parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(TimeSpan.FromSeconds(3))
+            .Connected.ShouldBeFalse();
+    }
+
+    /// <summary>
+    ///     A <see cref="IDriver.GetHealth"/> that throws is a failed observation, not evidence that the
+    ///     device is gone. Treating it as a trip would let a buggy health accessor cycle a working driver.
+    /// </summary>
+    [Fact]
+    public void A_throwing_GetHealth_is_not_evidence_of_comms_loss()
+    {
+        var driver = new HealthControlledStubDriver { GetHealthShouldThrow = true };
+        var (_, parent) = ConnectDriver(driver);
+
+        AwaitCondition(
+            () => driver.HealthReadCount >= 12,
+            TimeSpan.FromSeconds(5),
+            "health-poll ticks did not run — the absence assertion below would be vacuous");
+
+        parent.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
+    }
+
+    /// <summary>
+    ///     #527's second acceptance criterion, end to end: the driver recovers by itself once the device
+    ///     returns, with no node restart. Trip → Reconnecting → retry re-initialises → Connected, and the
+    ///     host is told connectivity is back.
+    /// </summary>
+    [Fact]
+    public void The_driver_recovers_on_its_own_once_the_device_returns()
+    {
+        var driver = new HealthControlledStubDriver
+        {
+            Health = new DriverHealth(DriverState.Degraded, DateTime.UtcNow.AddMinutes(-5), "socket timeout"),
+        };
+        var (_, parent) = ConnectDriver(driver, reconnectInterval: TimeSpan.FromMilliseconds(100));
+
+        parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(TimeSpan.FromSeconds(3))
+            .Connected.ShouldBeFalse();
+
+        // The device comes back: the next initialise attempt succeeds and reads resume.
+        driver.Health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
+
+        parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(TimeSpan.FromSeconds(5))
+            .Connected.ShouldBeTrue();
+        driver.InitializeCount.ShouldBeGreaterThan(1, "the driver was never re-initialised");
+    }
+
+    /// <summary>
+    ///     The leak guard. Several drivers overwrite an <see cref="IAsyncDisposable"/> transport field
+    ///     inside <c>InitializeAsync</c> without disposing the previous one (<c>ModbusDriver._transport</c>,
+    ///     <c>MqttDriver._connection</c>), so re-initialising without a teardown leaks one socket per
+    ///     attempt. Latent while only an operator's Force Reconnect re-entered this path; the watchdog
+    ///     makes it automatic and unbounded, so the retry must shut down first.
+    /// </summary>
+    [Fact]
+    public void A_reconnect_attempt_tears_the_dead_transport_down_before_re_initialising()
+    {
+        var driver = new HealthControlledStubDriver
+        {
+            Health = new DriverHealth(DriverState.Degraded, DateTime.UtcNow.AddMinutes(-5), "socket timeout"),
+        };
+        var (_, parent) = ConnectDriver(driver, reconnectInterval: TimeSpan.FromMilliseconds(100));
+
+        parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(TimeSpan.FromSeconds(3))
+            .Connected.ShouldBeFalse();
+
+        AwaitCondition(
+            () => driver.InitializeCount >= 2,
+            TimeSpan.FromSeconds(5),
+            "the reconnect retry never re-initialised the driver");
+
+        // Ordering, not just counts: EVERY re-initialise must be immediately preceded by a shutdown, since
+        // it is the initialise that overwrites the transport field. Asserting ShutdownCount > 0 alone would
+        // also pass on a teardown that happened afterwards, which frees nothing before the overwrite.
+        //
+        // The sequence is deliberately not pinned to a fixed length: this driver reports stale health even
+        // after a successful re-init, so the watchdog trips again and the cycle repeats — the real
+        // behaviour for a device that accepts a connection but serves no data. Every additional cycle is
+        // another chance for the invariant to be violated, so a longer run is a stronger test, not a
+        // flakier one.
+        var lifecycle = driver.Lifecycle;
+        lifecycle[0].ShouldBe("init", "the first connect must not tear down a driver that was never up");
+        for (var i = 1; i < lifecycle.Count; i++)
+        {
+            if (lifecycle[i] != "init") continue;
+            lifecycle[i - 1].ShouldBe(
+                "shutdown",
+                $"re-initialise at index {i} was not preceded by a teardown — the previous transport leaks. " +
+                $"Full sequence: [{string.Join(", ", lifecycle)}]");
+        }
+    }
+
+    /// <summary>
+    ///     The escape hatch is real: a zero threshold disables the watchdog entirely, leaving the
+    ///     pre-#527 behaviour. Pinned because tests elsewhere rely on a Connected driver staying Connected
+    ///     while reporting non-Healthy.
+    /// </summary>
+    [Fact]
+    public void A_zero_threshold_disables_the_watchdog()
+    {
+        var driver = new HealthControlledStubDriver
+        {
+            Health = new DriverHealth(DriverState.Degraded, DateTime.UtcNow.AddMinutes(-5), "socket timeout"),
+        };
+        var (_, parent) = ConnectDriver(driver, commsLossTimeout: TimeSpan.Zero);
+
+        AwaitCondition(
+            () => driver.HealthReadCount >= 12,
+            TimeSpan.FromSeconds(5),
+            "health-poll ticks did not run — the absence assertion below would be vacuous");
+
+        parent.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
+    }
+}
+
+/// <summary>
+///     A <see cref="StubDriver"/> whose health is fully driven by the test, counting observations so an
+///     absence assertion can prove the watchdog actually ran, and recording lifecycle call ORDER so the
+///     teardown-before-reinitialise guard can assert sequence rather than mere counts.
+/// </summary>
+internal sealed class HealthControlledStubDriver : StubDriver
+{
+    private readonly Lock _gate = new();
+    private int _healthReadCount;
+    private DateTime _firstReadUtc;
+    private DateTime _lastReadUtc;
+
+    /// <summary>The health returned by every <see cref="GetHealth"/> call; ignored when
+    /// <see cref="HealthFactory"/> is set.</summary>
+    public DriverHealth Health { get; set; } = new(DriverState.Healthy, DateTime.UtcNow, null);
+
+    /// <summary>Per-call health, for modelling a driver whose <c>LastSuccessfulRead</c> keeps advancing.</summary>
+    public Func<DriverHealth>? HealthFactory { get; set; }
+
+    /// <summary>When true, <see cref="GetHealth"/> throws — a failed observation.</summary>
+    public bool GetHealthShouldThrow { get; set; }
+
+    /// <summary>Number of <see cref="GetHealth"/> calls; the positive control for absence assertions.</summary>
+    public int HealthReadCount => Volatile.Read(ref _healthReadCount);
+
+    /// <summary>Wall-clock span between the first and last observation — proves an absence window
+    /// genuinely exceeded the trip threshold rather than merely counting fast ticks.</summary>
+    public TimeSpan ObservedSpan
+    {
+        get { lock (_gate) return _lastReadUtc - _firstReadUtc; }
+    }
+
+    /// <summary>Ordered lifecycle log: <c>"init"</c> / <c>"shutdown"</c>, in call order.</summary>
+    public IReadOnlyList<string> Lifecycle
+    {
+        get { lock (_gate) return _lifecycle.ToArray(); }
+    }
+
+    private readonly List<string> _lifecycle = new();
+
+    /// <inheritdoc />
+    public override async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
+    {
+        lock (_gate) _lifecycle.Add("init");
+        await base.InitializeAsync(driverConfigJson, cancellationToken);
+    }
+
+    /// <inheritdoc />
+    public override Task ShutdownAsync(CancellationToken cancellationToken)
+    {
+        lock (_gate) _lifecycle.Add("shutdown");
+        return base.ShutdownAsync(cancellationToken);
+    }
+
+    /// <inheritdoc />
+    public override DriverHealth GetHealth()
+    {
+        Interlocked.Increment(ref _healthReadCount);
+        lock (_gate)
+        {
+            var now = DateTime.UtcNow;
+            if (_firstReadUtc == default) _firstReadUtc = now;
+            _lastReadUtc = now;
+        }
+        if (GetHealthShouldThrow) throw new InvalidOperationException("stub-health-fail");
+        return HealthFactory is { } factory ? factory() : Health;
+    }
+}

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs

@@ -33,7 +33,7 @@ internal class StubDriver : IDriver
     /// <summary>Initializes the driver with the specified configuration JSON.</summary>
     /// <param name="driverConfigJson">The driver configuration JSON.</param>
     /// <param name="cancellationToken">Cancellation token for the operation.</param>
-    public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
+    public virtual async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
     {
         Interlocked.Increment(ref InitializeCount);
         lock (_initConfigsLock) InitConfigs.Add(driverConfigJson);
@@ -56,13 +56,20 @@ internal class StubDriver : IDriver
 
     /// <summary>Shuts down the driver.</summary>
     /// <param name="cancellationToken">Cancellation token for the operation.</param>
-    public Task ShutdownAsync(CancellationToken cancellationToken)
+    public virtual Task ShutdownAsync(CancellationToken cancellationToken)
     {
         Interlocked.Increment(ref ShutdownCount);
         return Task.CompletedTask;
     }
     /// <summary>Gets the health status of the driver.</summary>
-    public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null);
+    /// <remarks>
+    ///     Virtual so the #527 comms-loss watchdog suite can drive health directly. Note the
+    ///     <see cref="DateTime.UtcNow"/>: this stub is permanently Healthy with an always-fresh
+    ///     <c>LastSuccessfulRead</c>, so the watchdog can never fire for it — which is why every existing
+    ///     test built on <see cref="StubDriver"/> is unaffected by the watchdog, and equally why a watchdog
+    ///     test built on it would be vacuous.
+    /// </remarks>
+    public virtual DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null);
 }
 
 /// <summary>