fix(transport): detect a physically dropped serial/TCP connection so ConnectionStatus.Lost is reachable

docs/DEVICE_INTERFACES.md

@@ -315,14 +315,10 @@ The policy is asked once per connection attempt, not again after connecting.
 **Ownership and liveness.** The registry disconnects and disposes every device it removes,
 including duplicates it rejects — once a device is passed to `ConnectAsync` or `Register`, don't
 dispose it yourself. Registrations whose device stops reporting `IsConnected` are pruned before
-every registration attempt (and by `PruneDisconnected()` on demand). The registry does not
-reconnect on its own.
-
-> **Know this limit:** pruning is only as good as the transport's drop detection.
-> `SerialStreamTransport.IsConnected` reports the OS handle's state, which stays open after a USB
-> device is physically unplugged, so that device keeps reporting `IsConnected` and is *not* pruned —
-> a later `ConnectAsync` under the same key hands back the dead handle rather than reconnecting.
-> Devices you `Disconnect()` yourself, and drops a transport does report, prune as described.
+every registration attempt (and by `PruneDisconnected()` on demand). Pruning covers a physically
+unplugged USB device: the transport detects the drop itself (see
+[Detecting a dropped connection](#detecting-a-dropped-connection)) and stops reporting
+`IsConnected`, within the bounds documented there. The registry does not reconnect on its own.
 
 All members are safe to call from any thread; reads return snapshots, and the duplicate policy is
 always invoked without the internal lock held, so a policy that blocks on a user prompt never
@@ -398,6 +394,55 @@ device.StatusChanged += (sender, args) =>
 };
 ```
 
+### Detecting a dropped connection
+
+`ConnectionStatus.Lost` means the connection ended without anyone asking it to — the USB cable was
+pulled, the device lost power, the network dropped. `Disconnect()` reports
+`ConnectionStatus.Disconnected` instead, never `Lost`, so the two are always distinguishable.
+
+Neither OS handle is a liveness signal on its own: `SerialPort.IsOpen` stays `true` after a USB
+device is physically unplugged, and `TcpClient.Connected` describes the last completed operation
+rather than the current link. The transports therefore watch for a drop directly, and the detection
+time is bounded:
+
+| Transport | Drop | Reported within |
+|---|---|---|
+| Serial | Cable unplugged / device re-enumerated | **~3 s** — port-presence poll (1 s cadence, 2 consecutive misses) |
+| Serial | Reads or writes failing while traffic flows | **< 1 s** — 5 consecutive I/O failures with no successful transfer between them |
+| TCP | Peer closed, connection reset | **< 1 s** — same I/O fault escalation |
+| TCP | Link silently severed (power loss, WiFi drop) | **~20 s** — TCP keep-alive (10 s idle, 3 s probes, 3 retries) |
+
+Both serial detectors are cross-platform and behave identically on Windows, macOS, and Linux: port
+presence is `SerialPort.GetPortNames()` (falling back to the device node on Unix), with no WMI or
+other OS-specific device-change watcher. The presence poll is armed only if the port is visible to
+that probe at connect time, so an exotic port-name spelling disables the check rather than
+producing a false drop.
+
+A single failed read never disconnects anything. Escalation requires a *run* of failures with no
+successful transfer in between, so a stream that glitches and recovers stays connected. A read or
+write *timeout* is not a failure at all — it is what an idle or momentarily busy device looks
+like. To disable the serial presence poll and rely on I/O escalation alone, construct the transport
+with `livenessCheckInterval: TimeSpan.Zero`.
+
+When a drop is detected the transport closes its handle, so `IsConnected` already reads `false` by
+the time `StatusChanged` fires. The device's message consumer and producer are *not* stopped
+automatically — call `Disconnect()` (or `Dispose()`) once you have handled `Lost` to release them.
+
+`Lost` is raised on an internal thread — the reader loop or the liveness timer, whichever detected
+the drop — so treat the handler like any background callback: do the minimum, and push teardown or
+reconnection onto your own thread rather than blocking inside it.
+
+```csharp
+device.StatusChanged += (_, e) =>
+{
+    if (e.Status != ConnectionStatus.Lost) return;
+    _ = Task.Run(() => device.Disconnect());   // don't tear down inside the callback
+};
+```
+
+Custom transports can opt into the same escalation by implementing `ITransportHealthSink`; the
+reader and writer loops report every read/write outcome to a transport that does.
+
 ### Working with Device Metadata
 
 After initialization, device metadata is populated:

src/Daqifi.Core.Tests/Communication/Consumers/StreamMessageConsumerHealthReportingTests.cs

@@ -0,0 +1,243 @@
+using Daqifi.Core.Communication.Consumers;
+using Daqifi.Core.Communication.Transport;
+using System.Text;
+
+namespace Daqifi.Core.Tests.Communication.Consumers;
+
+/// <summary>
+/// Issue #382: the reader loop is the first component to see a device that has gone away, so it
+/// must tell the transport — both when reads keep failing (a disconnect) and when a read succeeds
+/// again (a blip that must not disconnect anything).
+/// </summary>
+public class StreamMessageConsumerHealthReportingTests
+{
+    [Fact]
+    public void ReaderLoop_WhenStreamStartsFailing_ReportsFaultsToTheTransport()
+    {
+        using var stream = new ScriptedStream();
+        var sink = new RecordingHealthSink();
+        using var consumer = new StreamMessageConsumer<string>(
+            stream, new LineBasedMessageParser(), healthSink: sink);
+
+        stream.FailReads = true;
+        consumer.Start();
+
+        Assert.True(WaitUntil(() => sink.FaultCount >= 3, TimeSpan.FromSeconds(5)),
+            $"expected repeated fault reports, saw {sink.FaultCount}");
+        Assert.Equal(0, sink.SuccessCount);
+        Assert.All(sink.Faults, ex => Assert.IsType<IOException>(ex));
+
+        consumer.StopSafely(timeoutMs: 2000);
+    }
+
+    [Fact]
+    public void ReaderLoop_WhenAFailingStreamRecovers_ReportsSuccessSoTheBlipIsNotADisconnect()
+    {
+        using var stream = new ScriptedStream();
+        var sink = new RecordingHealthSink();
+        using var consumer = new StreamMessageConsumer<string>(
+            stream, new LineBasedMessageParser(), healthSink: sink);
+
+        stream.FailReads = true;
+        consumer.Start();
+
+        Assert.True(WaitUntil(() => sink.FaultCount >= 1, TimeSpan.FromSeconds(5)));
+
+        // The blip ends: reads work again.
+        stream.Enqueue("$DAQiFi\r\n");
+        stream.FailReads = false;
+
+        Assert.True(WaitUntil(() => sink.SuccessCount >= 1, TimeSpan.FromSeconds(5)),
+            "a successful read must be reported so the transport clears the failure run");
+
+        consumer.StopSafely(timeoutMs: 2000);
+    }
+
+    [Fact]
+    public void ReaderLoop_OnATimeout_ReportsNeitherFaultNorSuccess()
+    {
+        // An idle read timeout is the normal state of a quiet device, not evidence of anything.
+        using var stream = new ScriptedStream();
+        var sink = new RecordingHealthSink();
+        using var consumer = new StreamMessageConsumer<string>(
+            stream, new LineBasedMessageParser(), healthSink: sink);
+
+        stream.TimeoutReads = true;
+        consumer.Start();
+
+        Assert.True(WaitUntil(() => stream.ReadCount >= 5, TimeSpan.FromSeconds(5)));
+        Assert.Equal(0, sink.FaultCount);
+        Assert.Equal(0, sink.SuccessCount);
+
+        consumer.StopSafely(timeoutMs: 2000);
+    }
+
+    [Fact]
+    public void ReaderLoop_OnAZeroByteReadFromANonSocketStream_ReportsNoFault()
+    {
+        // Plenty of stream types return 0 for "nothing right now"; only a socket's 0 means closed.
+        using var stream = new ScriptedStream();
+        var sink = new RecordingHealthSink();
+        using var consumer = new StreamMessageConsumer<string>(
+            stream, new LineBasedMessageParser(), healthSink: sink);
+
+        consumer.Start();
+
+        Assert.True(WaitUntil(() => stream.ReadCount >= 5, TimeSpan.FromSeconds(5)));
+        Assert.Equal(0, sink.FaultCount);
+
+        consumer.StopSafely(timeoutMs: 2000);
+    }
+
+    [Fact]
+    public void ReaderLoop_WithNoHealthSink_BehavesExactlyAsBefore()
+    {
+        // The feedback path is opt-in; a consumer constructed without one still just raises
+        // ErrorOccurred and keeps going.
+        using var stream = new ScriptedStream();
+        var errors = 0;
+        using var consumer = new StreamMessageConsumer<string>(stream, new LineBasedMessageParser());
+        consumer.ErrorOccurred += (_, _) => Interlocked.Increment(ref errors);
+
+        stream.FailReads = true;
+        consumer.Start();
+
+        Assert.True(WaitUntil(() => Volatile.Read(ref errors) >= 2, TimeSpan.FromSeconds(5)));
+        Assert.True(consumer.IsRunning);
+
+        consumer.StopSafely(timeoutMs: 2000);
+    }
+
+    private static bool WaitUntil(Func<bool> condition, TimeSpan timeout)
+    {
+        var deadline = DateTime.UtcNow + timeout;
+        while (DateTime.UtcNow < deadline)
+        {
+            if (condition())
+            {
+                return true;
+            }
+
+            Thread.Sleep(10);
+        }
+
+        return condition();
+    }
+
+    /// <summary>
+    /// Records what the reader loop reports, standing in for a real transport.
+    /// </summary>
+    private sealed class RecordingHealthSink : ITransportHealthSink
+    {
+        private readonly object _gate = new();
+        private readonly List<Exception> _faults = [];
+        private int _successCount;
+
+        public int FaultCount
+        {
+            get
+            {
+                lock (_gate)
+                {
+                    return _faults.Count;
+                }
+            }
+        }
+
+        public IReadOnlyList<Exception> Faults
+        {
+            get
+            {
+                lock (_gate)
+                {
+                    return _faults.ToArray();
+                }
+            }
+        }
+
+        public int SuccessCount => Volatile.Read(ref _successCount);
+
+        public void ReportIoFault(Exception error)
+        {
+            lock (_gate)
+            {
+                _faults.Add(error);
+            }
+        }
+
+        public void ReportIoSuccess() => Interlocked.Increment(ref _successCount);
+    }
+
+    /// <summary>
+    /// A stream whose reads can be made to fail, time out, or deliver scripted data, so a
+    /// "connection that starts failing" can be reproduced without hardware.
+    /// </summary>
+    private sealed class ScriptedStream : Stream
+    {
+        private readonly Queue<byte[]> _pending = new();
+        private readonly object _gate = new();
+        private int _readCount;
+
+        public volatile bool FailReads;
+        public volatile bool TimeoutReads;
+
+        public int ReadCount => Volatile.Read(ref _readCount);
+
+        public void Enqueue(string text)
+        {
+            lock (_gate)
+            {
+                _pending.Enqueue(Encoding.UTF8.GetBytes(text));
+            }
+        }
+
+        public override int Read(byte[] buffer, int offset, int count)
+        {
+            Interlocked.Increment(ref _readCount);
+
+            if (TimeoutReads)
+            {
+                Thread.Sleep(5);
+                throw new TimeoutException("no data within the read timeout");
+            }
+
+            if (FailReads)
+            {
+                Thread.Sleep(5);
+                throw new IOException("the device is gone");
+            }
+
+            lock (_gate)
+            {
+                if (_pending.Count == 0)
+                {
+                    return 0;
+                }
+
+                var chunk = _pending.Dequeue();
+                var length = Math.Min(chunk.Length, count);
+                Array.Copy(chunk, 0, buffer, offset, length);
+                return length;
+            }
+        }
+
+        public override bool CanRead => true;
+        public override bool CanSeek => false;
+        public override bool CanWrite => false;
+        public override long Length => throw new NotSupportedException();
+
+        public override long Position
+        {
+            get => throw new NotSupportedException();
+            set => throw new NotSupportedException();
+        }
+
+        public override void Flush()
+        {
+        }
+
+        public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
+        public override void SetLength(long value) => throw new NotSupportedException();
+        public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+    }
+}