fix: observe endpointstream shutdown in SSE closed()

net/net-endpoint_buf.c++m

@@ -20,13 +20,22 @@ public:
     }
 
     // Wake blocking I/O on this buffer's socket (see socket::shutdown).
+    // Also latches shut_down() so observers (e.g. SSE/MCP session::closed)
+    // notice stop() without waiting for a write to set iostream failbit.
     void shutdown() noexcept
     {
         m_socket.shutdown();
+        m_shut_down.store(true, std::memory_order_release);
+    }
+
+    [[nodiscard]] bool shut_down() const noexcept
+    {
+        return m_shut_down.load(std::memory_order_acquire);
     }
 
 protected:
     socket m_socket;
+    std::atomic<bool> m_shut_down{false};
 };
 
 constexpr std::size_t tcp_buffer_size = 4096;

net/net-endpoint_stream.c++m

@@ -149,6 +149,15 @@ public:
         if(m_buf)
             m_buf->shutdown();
     }
+
+    // True after shutdown() (or close()). Used by SSE/MCP closed() checks so
+    // stop() is visible without a write failing first — otherwise handlers that
+    // only poll closed() never exit and wait_for_handlers() deadlocks.
+    [[nodiscard]] bool shut_down() const noexcept
+    {
+        return not m_buf or m_buf->shut_down();
+    }
+
     bool wait_for(const std::chrono::milliseconds& timeout)
     {
         if(not m_buf) return false;

net/net-sse.c++m

@@ -129,7 +129,15 @@ public:
 
     [[nodiscard]] bool closed() const noexcept
     {
-        return m_closed or not m_stream.good();
+        if(m_closed or not m_stream.good())
+            return true;
+        // http::server::stop() calls endpointstream::shutdown(), which does not
+        // set iostream failbit until the next write. Without this latch, MCP
+        // handlers that only poll closed() never observe stop() and
+        // wait_for_handlers() deadlocks (inbound-queue tests; custom transports).
+        if(auto* ep = dynamic_cast<const net::endpointstream*>(&m_stream))
+            return ep->shut_down();
+        return false;
     }
 
     [[nodiscard]] std::size_t bytes_out() const noexcept

net/net-sse.test.c++

@@ -3,6 +3,10 @@
 // See the LICENSE file in the project root for full license text.
 
 module net;
+import :endpointbuf;
+import :endpointstream;
+import :posix;
+import :socket;
 import :sse;
 import tester;
 import std;
@@ -81,6 +85,21 @@ auto register_sse_tests()
             check_false(s.send_comment("more"));
             check_false(s.flush());
         };
+
+        // Regression: stop() → endpointstream::shutdown() left failbit clear, so
+        // MCP handlers that only poll session.closed() never exited and
+        // wait_for_handlers() deadlocked after drain-on-listen-exit.
+        section("closed after endpointstream shutdown latch") = []
+        {
+            auto stream = net::endpointstream{
+                new net::tcp_endpointbuf{net::socket{net::posix::af_inet, net::posix::sock_stream}}};
+            auto s = session{stream};
+            check_false(stream.shut_down());
+            check_false(s.closed());
+            stream.shutdown();
+            check_true(stream.shut_down());
+            check_true(s.closed());
+        };
     };
 
     return true;