From ff2d7853100a25dad073341120e167c327700b8a Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Sat, 1 Aug 2026 23:11:57 +0000
Subject: [PATCH] fix: observe endpointstream shutdown in SSE closed()

stop() shuts down live client sockets but does not set iostream failbit
until the next write. MCP handlers that only poll session.closed() (including
the inbound-queue tests) never exited, so wait_for_handlers() deadlocked after
drain-on-listen-exit. Latch shut_down on shutdown() and check it from
sse::session::closed().

Co-authored-by: Kaius  Ruokonen <ruoka@users.noreply.github.com>
---
 net/net-endpoint_buf.c++m    |  9 +++++++++
 net/net-endpoint_stream.c++m |  9 +++++++++
 net/net-sse.c++m             | 10 +++++++++-
 net/net-sse.test.c++         | 19 +++++++++++++++++++
 4 files changed, 46 insertions(+), 1 deletion(-)

diff --git a/net/net-endpoint_buf.c++m b/net/net-endpoint_buf.c++m
index ad2bac6..4eea7b1 100644
--- a/net/net-endpoint_buf.c++m
+++ b/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;
diff --git a/net/net-endpoint_stream.c++m b/net/net-endpoint_stream.c++m
index 05b1d72..6948257 100644
--- a/net/net-endpoint_stream.c++m
+++ b/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;
diff --git a/net/net-sse.c++m b/net/net-sse.c++m
index 2341ca4..95b0739 100644
--- a/net/net-sse.c++m
+++ b/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
diff --git a/net/net-sse.test.c++ b/net/net-sse.test.c++
index 7f31b90..b85eacb 100644
--- a/net/net-sse.test.c++
+++ b/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;