fix: observe endpointstream shutdown in SSE closed()

[cursor]

Bug and impact

After drain-on-listen-exit, http::server::stop() calls endpointstream::shutdown() on live clients and then wait_for_handlers(). shutdown() wakes blocking I/O but does not set iostream failbit until the next write.

http::sse::session::closed() only checked m_closed || !stream.good(). Custom SSE/MCP handlers that only poll closed() (including the MCP inbound-queue tests that hold a session open with while (!sess.closed())) never observe stop(), so listen() / join hang forever.

Production mcp::server::run_session already probes with heartbeats and is less affected; any attach handler that only polls closed() deadlocks shutdown.

Root cause

stop() → endpointstream::shutdown() does not surface as stream failure until a write. Session lifecycle state (m_state->closed) is only set in erase_session after on_session returns, so a poll-only handler has no signal.

Fix

• Latch shut_down on endpointbuf_base::shutdown()
• Expose it via endpointstream::shut_down()
• Have sse::session::closed() observe the latch (via dynamic_cast to endpointstream)

Validation

• Added unit section: closed after endpointstream shutdown latch
• Ran filtered tests matching shutdown latch|MCP SSE|SSE session (includes inbound-queue 503 scenarios that previously hung on stop()): all passed

  

---

Note

Medium Risk
Touches connection lifecycle and SSE shutdown visibility; limited scope but affects server stop and MCP/SSE handler exit paths.

Overview
Fixes a shutdown deadlock where http::server::stop() shuts down client endpointstreams but handlers that only poll http::sse::session::closed() never saw closure until a write failed.

endpointbuf_base::shutdown() now sets an atomic shut_down latch (in addition to waking the socket). endpointstream::shut_down() exposes that state (including after close()). sse::session::closed() treats an underlying endpointstream as closed when the latch is set, via dynamic_cast.

Adds a regression test: after stream.shutdown(), session.closed() is true while iostream good() may still be true.

^{Reviewed by Cursor Bugbot for commit ff2d785. Bugbot is set up for automated code reviews on this repo. Configure here.}