diff --git a/src/zeroconf/_engine.py b/src/zeroconf/_engine.py
index 0e1c01a1..291a0d3c 100644
--- a/src/zeroconf/_engine.py
+++ b/src/zeroconf/_engine.py
@@ -126,6 +126,16 @@ async def _async_create_endpoints(self) -> None:
             if s in self._respond_sockets:
                 self._respond_sockets.remove(s)
 
+    def _async_remove_listener(self, listener: AsyncListener) -> None:
+        """Drop a listener and its wrapped transports from the engine lists."""
+        wrapped = listener.transport
+        transport = wrapped.transport if wrapped is not None else None
+        if listener in self.protocols:
+            self.protocols.remove(listener)
+        if transport is not None:
+            self.readers = [w for w in self.readers if w.transport is not transport]
+            self.senders = [w for w in self.senders if w.transport is not transport]
+
     def _async_cache_cleanup(self) -> None:
         """Periodic cache cleanup."""
         now = current_time_millis()
diff --git a/src/zeroconf/_listener.py b/src/zeroconf/_listener.py
index 7be2a828..ac7347da 100644
--- a/src/zeroconf/_listener.py
+++ b/src/zeroconf/_listener.py
@@ -356,4 +356,5 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None:
         self.sock_description = f"{wrapped_transport.fileno} ({wrapped_transport.sock_name})"
 
     def connection_lost(self, exc: Exception | None) -> None:
-        """Handle connection lost."""
+        """Prune this transport from the engine so a dead socket is not reused."""
+        self.zc.engine._async_remove_listener(self)
diff --git a/tests/test_engine.py b/tests/test_engine.py
index 66cfa356..43fcf510 100644
--- a/tests/test_engine.py
+++ b/tests/test_engine.py
@@ -10,7 +10,7 @@
 import pytest
 
 import zeroconf as r
-from zeroconf import _engine, const
+from zeroconf import _engine, _listener, const
 from zeroconf.asyncio import AsyncZeroconf
 
 log = logging.getLogger("zeroconf")
@@ -84,6 +84,48 @@ async def test_setup_releases_socket_ownership(aiozc_loopback: AsyncZeroconf) ->
     assert engine.senders
 
 
+@pytest.mark.asyncio
+async def test_connection_lost_prunes_transport(aiozc_loopback: AsyncZeroconf) -> None:
+    """A lost transport is removed from the engine reader/sender/protocol lists."""
+    await aiozc_loopback.zeroconf.async_wait_for_start()
+    engine = aiozc_loopback.zeroconf.engine
+    assert engine.senders
+    reader_count = len(engine.readers)
+    sender_count = len(engine.senders)
+    protocol_count = len(engine.protocols)
+
+    dead_transport = engine.senders[0].transport
+    protocol = next(
+        p for p in engine.protocols if p.transport is not None and p.transport.transport is dead_transport
+    )
+    protocol.connection_lost(None)
+
+    assert len(engine.senders) == sender_count - 1
+    assert len(engine.readers) == reader_count - 1
+    assert len(engine.protocols) == protocol_count - 1
+    assert all(w.transport is not dead_transport for w in engine.senders)
+    assert all(w.transport is not dead_transport for w in engine.readers)
+    assert protocol not in engine.protocols
+
+
+@pytest.mark.asyncio
+async def test_connection_lost_without_transport_is_noop(aiozc_loopback: AsyncZeroconf) -> None:
+    """connection_lost on a listener that never bound a transport leaves the lists intact."""
+    await aiozc_loopback.zeroconf.async_wait_for_start()
+    engine = aiozc_loopback.zeroconf.engine
+    reader_count = len(engine.readers)
+    sender_count = len(engine.senders)
+    protocol_count = len(engine.protocols)
+
+    orphan = _listener.AsyncListener(aiozc_loopback.zeroconf)
+    assert orphan.transport is None
+    orphan.connection_lost(None)
+
+    assert len(engine.readers) == reader_count
+    assert len(engine.senders) == sender_count
+    assert len(engine.protocols) == protocol_count
+
+
 @pytest.mark.asyncio
 async def test_async_close_propagates_outer_cancellation(aiozc_loopback: AsyncZeroconf) -> None:
     """Outer-task cancellation while awaiting setup propagates to the caller."""