From 8aca0f14d9deceaae4ccde9d42a5c4762c4e608c Mon Sep 17 00:00:00 2001
From: kekxv <caesar@kekxv.com>
Date: Thu, 20 Aug 2026 09:41:26 +0000
Subject: [PATCH 1/6] build: upgrade starter example to khttpd 0.4.4

---
 example/MODULE.bazel | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/example/MODULE.bazel b/example/MODULE.bazel
index 95643af..e369abf 100644
--- a/example/MODULE.bazel
+++ b/example/MODULE.bazel
@@ -8,7 +8,7 @@ bazel_dep(name = "boost", version = "1.90.0.bcr.1")
 bazel_dep(name = "boost.asio", version = "1.90.0.bcr.1")
 bazel_dep(name = "boost.mysql", version = "1.90.0.bcr.1")
 bazel_dep(name = "spdlog", version = "1.17.0")
-bazel_dep(name = "khttpd", version = "0.4.2")
+bazel_dep(name = "khttpd", version = "0.4.4")
 local_path_override(
     module_name = "khttpd",
     path = "..",

From d61aeebaa2789380a5d586911148efd9f638e590 Mon Sep 17 00:00:00 2001
From: kekxv <caesar@kekxv.com>
Date: Fri, 21 Aug 2026 14:10:24 +0000
Subject: [PATCH 2/6] feat: add asynchronous server-sent events

---
 README.md                        |  49 ++++++++
 doc/index.md                     |   3 +
 doc/server-sent-events.md        |  75 +++++++++++
 framework/BUILD.bazel            |   2 +
 framework/client/sse_client.cpp  | 185 +++++++++++++++++++++++++++
 framework/client/sse_client.hpp  |  41 ++++++
 framework/router/http_router.cpp |  14 +++
 framework/router/http_router.hpp |   3 +
 framework/sse/sse_event.hpp      |  22 ++++
 framework/sse/sse_parser.cpp     | 122 ++++++++++++++++++
 framework/sse/sse_parser.hpp     |  30 +++++
 framework/sse/sse_session.cpp    | 126 +++++++++++++++++++
 framework/sse/sse_session.hpp    |  33 +++++
 framework/tests/BUILD.bazel      |  11 ++
 framework/tests/sse_test.cpp     | 208 +++++++++++++++++++++++++++++++
 15 files changed, 924 insertions(+)
 create mode 100644 doc/server-sent-events.md
 create mode 100644 framework/client/sse_client.cpp
 create mode 100644 framework/client/sse_client.hpp
 create mode 100644 framework/sse/sse_event.hpp
 create mode 100644 framework/sse/sse_parser.cpp
 create mode 100644 framework/sse/sse_parser.hpp
 create mode 100644 framework/sse/sse_session.cpp
 create mode 100644 framework/sse/sse_session.hpp
 create mode 100644 framework/tests/sse_test.cpp

diff --git a/README.md b/README.md
index 284a622..d82f861 100644
--- a/README.md
+++ b/README.md
@@ -23,6 +23,7 @@ and [Boost.Asio](https://www.boost.org/doc/libs/release/libs/asio/), managed wit
 - **Exception Handling** — Type-safe exception dispatcher with per-type handlers
 - **Chunked Streaming** — Server-sent chunked transfer encoding via `HttpContext::chunked()`
 - **Bidirectional HTTP Streaming** — Header-first request routing, fixed-buffer upload/download, proxy backpressure, and Range forwarding
+- **Async Server-Sent Events** — Non-blocking SSE routes and client with incremental parsing, cancellation, and backpressure
 - **Cookie Support** — Read / write cookies with configurable `CookieOptions` (path, domain, SameSite, etc.)
 - **Form & Multipart** — `application/x-www-form-urlencoded` and `multipart/form-data` parsing (file uploads)
 - **JSON** — Native `boost::json` integration with `get_json()`, `set_body_json()`, `set_body_from()`
@@ -168,6 +169,7 @@ framework/
 ├── client/
 │   ├── http_client.hpp/cpp     # Sync/async HTTP client with SSL
 │   ├── http_client_stream.hpp/cpp # Fixed-buffer HTTP streaming client
+│   ├── sse_client.hpp/cpp      # Async Server-Sent Events client
 │   ├── http_proxy_session.hpp/cpp # Bidirectional streaming proxy pump
 │   └── websocket_client.hpp/cpp # WebSocket client
 ├── interceptor/
@@ -182,6 +184,9 @@ framework/
 │   └── di_container.hpp        # Type-indexed DI container (singleton)
 ├── session/
 │   └── http_session.hpp/cpp    # Per-connection HTTP session
+├── sse/
+│   ├── sse_parser.hpp/cpp      # Incremental SSE parser and wire formatting
+│   └── sse_session.hpp/cpp     # Async server-side SSE write queue
 └── websocket/
     └── websocket_session.hpp/cpp # Per-connection WebSocket session
 ```
@@ -421,6 +426,50 @@ fixed-buffer behavior. Its default TLS context verifies the system trust store;
 an application can inject an `ssl::context` into `HttpClientStream` or
 `HttpProxySession` for private CAs and test certificates.
 
+### Async Server-Sent Events
+
+Register an SSE endpoint with `HttpRouter::sse`. The session serializes writes
+through an owned queue, so callers may publish from different threads without
+keeping event strings alive until the socket write completes.
+
+```cpp
+router.sse("/events",
+  [](HttpContext&, std::shared_ptr<sse::SseSession> session)
+  {
+    session->on_close([](boost::system::error_code ec) {
+      // Remove the subscriber from application state.
+    });
+    session->send({"config", R"({"name":"app.yaml"})", "42", 3000});
+    session->send_comment("heartbeat");
+
+    // Retain session in the application's subscriber collection when more
+    // events will be sent after this handler returns. Call close() for a
+    // graceful final chunk or cancel() to abort the connection.
+  });
+```
+
+`SseClient` uses the same fixed-buffer streaming transport and delivers each
+complete event as soon as it arrives. Arbitrarily split lines, CRLF/LF,
+multiline `data`, comments, `id`, and numeric `retry` fields are supported.
+
+```cpp
+auto events = std::make_shared<client::SseClient>(ioc);
+events->connect(
+  "https://config.internal/events",
+  {{"Authorization", "Bearer internal-token"}, {"Last-Event-ID", "41"}},
+  [](const sse::SseEvent& event) {
+    // event.event, event.data, event.id, event.retry
+  },
+  [](boost::system::error_code ec) {
+    // Schedule reconnect/backoff in application code when appropriate.
+  });
+```
+
+Keep the `SseClient` alive for the duration of the subscription. `cancel()`
+closes the transport and completes the close callback once with
+`operation_aborted`. Automatic reconnect is intentionally left to the caller,
+which can apply service-specific backoff and send `Last-Event-ID`.
+
 ### Tests
 
 Run the complete framework suite, including buffered-body boundaries, streaming
diff --git a/doc/index.md b/doc/index.md
index 2dd011f..84cbab8 100644
--- a/doc/index.md
+++ b/doc/index.md
@@ -9,6 +9,7 @@
 | [架构指南](architecture.md) | 框架设计、请求流程、线程模型、扩展点 |
 | [高级功能](advanced.md) | 拦截器、异常处理、动态 WebSocket、双向 HTTP 流、Cron、DI、Cookie |
 | [HTTP 与 WebSocket 客户端](http-client.md) | 缓冲/流式 HTTP 客户端、API_CALL 宏、WebSocket 帧客户端 |
+| [异步 Server-Sent Events](server-sent-events.md) | SSE 服务端路由、异步客户端、心跳、取消与重连策略 |
 
 ## 按主题查找
 
@@ -33,6 +34,7 @@
 - 设置响应 → [API 参考](api-reference.md#响应设置)
 - 分块流式响应 → [高级功能](advanced.md#分块流式响应)
 - 大文件双向流式代理 → [高级功能](advanced.md#双向-http-流与大文件代理)
+- 异步 SSE 服务端 → [Server-Sent Events](server-sent-events.md#服务端)
 - 普通请求体 413 上限 → [API 参考](api-reference.md#server)
 - Cookie 操作 → [高级功能](advanced.md#cookie-操作)
 
@@ -57,6 +59,7 @@
 - 多 Host 权重分发 → [HTTP 客户端](http-client.md#多-host-权重分发)
 - API_CALL 宏 → [HTTP 客户端](http-client.md#api_call-宏自动生成客户端方法)
 - WebSocket 客户端 → [HTTP 客户端](http-client.md#websocket-客户端)
+- 异步 SSE 客户端 → [Server-Sent Events](server-sent-events.md#客户端)
 
 ### 架构
 - 请求处理流程 → [架构指南](architecture.md#请求处理流程)
diff --git a/doc/server-sent-events.md b/doc/server-sent-events.md
new file mode 100644
index 0000000..de466b7
--- /dev/null
+++ b/doc/server-sent-events.md
@@ -0,0 +1,75 @@
+# 异步 Server-Sent Events
+
+khttpd 的 SSE 支持建立在异步流式 HTTP 传输之上，不占用阻塞线程。服务端通过
+`SseSession` 顺序写入事件，客户端通过 `SseClient` 增量读取事件；普通响应、
+`HttpContext::chunked()` 和 WebSocket 的处理路径不受影响。
+
+## 服务端
+
+使用 `HttpRouter::sse` 注册 GET 路由：
+
+```cpp
+#include "sse/sse_session.hpp"
+
+router.sse("/events",
+  [](HttpContext&, std::shared_ptr<sse::SseSession> session)
+  {
+    session->on_close([](boost::system::error_code ec) {
+      // 从业务侧的订阅者集合移除连接。
+    });
+
+    session->send({
+      "config",                     // event
+      R"({"name":"app.yaml"})",   // data，可包含多行
+      "42",                         // id
+      3000                           // retry，毫秒
+    });
+    session->send_comment("heartbeat");
+  });
+```
+
+响应会包含 `Content-Type: text/event-stream`、`Cache-Control: no-cache` 和
+`X-Accel-Buffering: no`，正文使用 HTTP 分块传输。`send()` 和
+`send_comment()` 可从不同线程调用，内部拥有待写入字符串并按 FIFO 顺序串行写入，
+自然继承底层 socket 的背压。
+
+如果路由处理函数返回后仍要继续推送，应用必须在订阅者集合中持有
+`std::shared_ptr<SseSession>`。`close()` 会等待已排队事件写完再发送结束块；
+`cancel()` 会立即中断连接。两者最终都只触发一次 `on_close`。
+
+心跳建议使用注释帧，例如 `send_comment("heartbeat")`；心跳周期和空闲连接管理由
+业务服务决定。
+
+## 客户端
+
+```cpp
+#include "client/sse_client.hpp"
+
+auto subscription = std::make_shared<client::SseClient>(ioc);
+subscription->connect(
+  "https://config.internal/events",
+  {
+    {"Authorization", "Bearer internal-token"},
+    {"Last-Event-ID", "41"},
+  },
+  [](const sse::SseEvent& event) {
+    // event.event、event.data、event.id、event.retry
+  },
+  [](boost::system::error_code ec) {
+    // 按业务策略决定是否重连。
+  });
+```
+
+客户端要求上游返回 HTTP 200，且 Content-Type 必须为 `text/event-stream`
+（允许带 charset 等参数）。解析器支持任意网络分片、CRLF/LF、多行 `data`、注释、
+`event`、`id` 和纯数字 `retry`；无效的 `retry` 会被忽略。
+
+调用方需要在订阅期间持有 `SseClient`。显式调用 `cancel()` 会关闭传输，并以
+`operation_aborted` 调用一次关闭回调。框架不自动重连：调用方可以结合事件的
+`retry`、指数退避和 `Last-Event-ID` 实现符合具体服务需求的恢复策略。
+
+## 选择 SSE 还是 WebSocket
+
+- 服务器向客户端持续单向推送配置、实例或通知时，优先使用 SSE。
+- 需要全双工消息、二进制帧或自定义控制帧时，使用 WebSocket。
+- 需要上传和下载同时承受背压时，使用双向 HTTP 流式接口。
diff --git a/framework/BUILD.bazel b/framework/BUILD.bazel
index a9b60f1..f6f8475 100644
--- a/framework/BUILD.bazel
+++ b/framework/BUILD.bazel
@@ -6,6 +6,7 @@ cc_library(
     srcs = glob([
         "*.cpp",
         "router/*.cpp",
+        "sse/*.cpp",
         "session/*.cpp",
         "websocket/*.cpp",
         "context/*.cpp",
@@ -21,6 +22,7 @@ cc_library(
         "interceptor/*.hpp",
         "di/*.hpp",
         "router/*.hpp",
+        "sse/*.hpp",
         "session/*.hpp",
         "websocket/*.hpp",
         "client/*.hpp",
diff --git a/framework/client/sse_client.cpp b/framework/client/sse_client.cpp
new file mode 100644
index 0000000..bf4c768
--- /dev/null
+++ b/framework/client/sse_client.cpp
@@ -0,0 +1,185 @@
+#include "client/sse_client.hpp"
+
+#include <boost/asio/buffer.hpp>
+#include <boost/asio/error.hpp>
+#include <boost/beast/http.hpp>
+#include <algorithm>
+#include <array>
+#include <cctype>
+#include <mutex>
+#include <utility>
+
+#include "client/http_client_stream.hpp"
+#include "sse/sse_parser.hpp"
+
+namespace khttpd::framework::client
+{
+  namespace http = boost::beast::http;
+  namespace net = boost::asio;
+
+  namespace
+  {
+    bool is_event_stream_content_type(const boost::beast::string_view value)
+    {
+      std::string media_type(value.data(), value.size());
+      const auto parameters = media_type.find(';');
+      if (parameters != std::string::npos) media_type.erase(parameters);
+      while (!media_type.empty() && std::isspace(static_cast<unsigned char>(media_type.back())))
+        media_type.pop_back();
+      auto first = media_type.begin();
+      while (first != media_type.end() && std::isspace(static_cast<unsigned char>(*first))) ++first;
+      media_type.erase(media_type.begin(), first);
+      std::transform(media_type.begin(), media_type.end(), media_type.begin(),
+                     [](const unsigned char c) { return static_cast<char>(std::tolower(c)); });
+      return media_type == "text/event-stream";
+    }
+  }
+
+  struct SseClient::Impl : std::enable_shared_from_this<Impl>
+  {
+    std::shared_ptr<HttpClientStream> stream;
+    sse::SseParser parser;
+    std::array<char, 8192> read_buffer{};
+    EventHandler event_handler;
+    CloseHandler close_handler;
+    std::mutex mutex;
+    bool connect_started = false;
+    bool terminal = false;
+
+    explicit Impl(net::io_context& ioc) : stream(std::make_shared<HttpClientStream>(ioc)) {}
+    Impl(net::io_context& ioc, net::ssl::context& context)
+      : stream(std::make_shared<HttpClientStream>(ioc, context)) {}
+
+    void connect(const std::string& url,
+                 const std::map<std::string, std::string>& headers,
+                 EventHandler on_event,
+                 CloseHandler on_close)
+    {
+      bool duplicate = false;
+      {
+        std::lock_guard lock(mutex);
+        if (connect_started)
+          duplicate = true;
+        else
+        {
+          connect_started = true;
+          event_handler = std::move(on_event);
+          close_handler = std::move(on_close);
+        }
+      }
+      if (duplicate)
+      {
+        if (on_close) on_close(make_error_code(boost::system::errc::operation_in_progress));
+        return;
+      }
+
+      HttpClientStream::RequestHead request{http::verb::get, "/", 11};
+      request.keep_alive(true);
+      for (const auto& [name, value] : headers) request.set(name, value);
+      request.set(http::field::accept, "text/event-stream");
+      request.set(http::field::cache_control, "no-cache");
+      stream->async_start(url, std::move(request),
+        [self = shared_from_this()](boost::system::error_code ec)
+        {
+          if (ec) return self->finish(ec);
+          self->stream->async_finish_request([self](boost::system::error_code finish_ec)
+          {
+            if (finish_ec) return self->finish(finish_ec);
+            self->read_head();
+          });
+        });
+    }
+
+    void read_head()
+    {
+      stream->async_read_response_head(
+        [self = shared_from_this()](boost::system::error_code ec, HttpClientStream::ResponseHead head)
+        {
+          if (ec) return self->finish(ec);
+          if (head.result() != http::status::ok ||
+              !is_event_stream_content_type(head[http::field::content_type]))
+            return self->finish(make_error_code(boost::system::errc::protocol_error));
+          self->read_next();
+        });
+    }
+
+    void read_next()
+    {
+      stream->async_read_some(net::buffer(read_buffer),
+        [self = shared_from_this()](boost::system::error_code ec, const std::size_t size, const bool done)
+        {
+          if (size != 0)
+          {
+            const auto events = self->parser.feed(std::string_view(self->read_buffer.data(), size));
+            for (const auto& event : events) self->emit(event);
+          }
+          if (self->is_terminal()) return;
+          if (done) return self->finish({});
+          if (ec == net::error::eof || ec == http::error::end_of_stream) return self->finish({});
+          if (ec) return self->finish(ec);
+          self->read_next();
+        });
+    }
+
+    void emit(const sse::SseEvent& event)
+    {
+      EventHandler handler;
+      {
+        std::lock_guard lock(mutex);
+        if (terminal) return;
+        handler = event_handler;
+      }
+      if (handler) handler(event);
+    }
+
+    bool is_terminal()
+    {
+      std::lock_guard lock(mutex);
+      return terminal;
+    }
+
+    void finish(const boost::system::error_code ec)
+    {
+      CloseHandler handler;
+      {
+        std::lock_guard lock(mutex);
+        if (terminal) return;
+        terminal = true;
+        handler = std::move(close_handler);
+        event_handler = {};
+      }
+      if (ec) stream->cancel();
+      if (handler) handler(ec);
+    }
+
+    void cancel()
+    {
+      stream->cancel();
+      finish(net::error::operation_aborted);
+    }
+
+    void abandon()
+    {
+      {
+        std::lock_guard lock(mutex);
+        terminal = true;
+        event_handler = {};
+        close_handler = {};
+      }
+      stream->cancel();
+    }
+  };
+
+  SseClient::SseClient(net::io_context& ioc) : impl_(std::make_shared<Impl>(ioc)) {}
+  SseClient::SseClient(net::io_context& ioc, net::ssl::context& context)
+    : impl_(std::make_shared<Impl>(ioc, context)) {}
+  SseClient::~SseClient() { if (impl_) impl_->abandon(); }
+  void SseClient::connect(const std::string& url,
+                          const std::map<std::string, std::string>& headers,
+                          EventHandler on_event,
+                          CloseHandler on_close)
+  {
+    impl_->connect(url, headers, std::move(on_event), std::move(on_close));
+  }
+  void SseClient::cancel() { impl_->cancel(); }
+}
diff --git a/framework/client/sse_client.hpp b/framework/client/sse_client.hpp
new file mode 100644
index 0000000..91ebf5d
--- /dev/null
+++ b/framework/client/sse_client.hpp
@@ -0,0 +1,41 @@
+#ifndef KHTTPD_FRAMEWORK_CLIENT_SSE_CLIENT_HPP
+#define KHTTPD_FRAMEWORK_CLIENT_SSE_CLIENT_HPP
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ssl/context.hpp>
+#include <boost/system/error_code.hpp>
+#include <functional>
+#include <map>
+#include <memory>
+#include <string>
+
+#include "sse/sse_event.hpp"
+
+namespace khttpd::framework::client
+{
+  class SseClient
+  {
+  public:
+    using EventHandler = std::function<void(const sse::SseEvent&)>;
+    using CloseHandler = std::function<void(boost::system::error_code)>;
+
+    explicit SseClient(boost::asio::io_context& ioc);
+    SseClient(boost::asio::io_context& ioc, boost::asio::ssl::context& ssl_context);
+    ~SseClient();
+
+    SseClient(const SseClient&) = delete;
+    SseClient& operator=(const SseClient&) = delete;
+
+    void connect(const std::string& url,
+                 const std::map<std::string, std::string>& headers,
+                 EventHandler on_event,
+                 CloseHandler on_close);
+    void cancel();
+
+  private:
+    struct Impl;
+    std::shared_ptr<Impl> impl_;
+  };
+}
+
+#endif
diff --git a/framework/router/http_router.cpp b/framework/router/http_router.cpp
index 15f8cdf..c6bc972 100644
--- a/framework/router/http_router.cpp
+++ b/framework/router/http_router.cpp
@@ -1,5 +1,6 @@
 // framework/router/http_router.cpp
 #include "http_router.hpp"
+#include "sse/sse_session.hpp"
 #include <fmt/core.h>
 #include <spdlog/spdlog.h>
 #include <algorithm>
@@ -333,6 +334,19 @@ namespace khttpd::framework
     return false;
   }
 
+  void HttpRouter::sse(const std::string& path, SseHandler handler)
+  {
+    stream(path, boost::beast::http::verb::get,
+      [handler = std::move(handler)](HttpContext& ctx, std::shared_ptr<HttpRequestStream>,
+                                     std::shared_ptr<HttpResponseStream> response, HttpStreamComplete)
+      {
+        auto session = std::make_shared<sse::SseSession>(std::move(response), ctx.get_request().version(),
+                                                        ctx.get_request().keep_alive());
+        session->start();
+        handler(ctx, std::move(session));
+      });
+  }
+
   bool HttpRouter::dispatch_stream(HttpContext& ctx, std::shared_ptr<HttpRequestStream> stream,
                                    std::shared_ptr<HttpResponseStream> response_stream,
                                    HttpStreamComplete complete) const
diff --git a/framework/router/http_router.hpp b/framework/router/http_router.hpp
index b25b090..ef59753 100644
--- a/framework/router/http_router.hpp
+++ b/framework/router/http_router.hpp
@@ -19,6 +19,7 @@
 
 namespace khttpd::framework
 {
+  namespace sse { class SseSession; }
   struct OpenApiInfo;
   class HttpRouter;
   void install_openapi_routes(HttpRouter& router, const OpenApiInfo& info,
@@ -30,6 +31,7 @@ namespace khttpd::framework
   using HttpStreamComplete = std::function<void()>;
   using HttpStreamHandler = std::function<void(HttpContext&, std::shared_ptr<HttpRequestStream>,
                                                std::shared_ptr<HttpResponseStream>, HttpStreamComplete)>;
+  using SseHandler = std::function<void(HttpContext&, std::shared_ptr<sse::SseSession>)>;
   using UnknownExceptionHandler = std::function<void(HttpContext&)>;
 
   struct RouteHeader
@@ -247,6 +249,7 @@ namespace khttpd::framework
     // Async handlers must invoke complete exactly once, from any thread.
     void async_route(const std::string& path, boost::beast::http::verb method, HttpAsyncHandler handler);
     void stream(const std::string& path, boost::beast::http::verb method, HttpStreamHandler handler);
+    void sse(const std::string& path, SseHandler handler);
 
     // Used by HttpSession after it has parsed only the request header.
     bool is_stream_route(const std::string& path, boost::beast::http::verb method) const;
diff --git a/framework/sse/sse_event.hpp b/framework/sse/sse_event.hpp
new file mode 100644
index 0000000..a871700
--- /dev/null
+++ b/framework/sse/sse_event.hpp
@@ -0,0 +1,22 @@
+#ifndef KHTTPD_FRAMEWORK_SSE_SSE_EVENT_HPP
+#define KHTTPD_FRAMEWORK_SSE_SSE_EVENT_HPP
+
+#include <cstdint>
+#include <optional>
+#include <string>
+
+namespace khttpd::framework::sse
+{
+  struct SseEvent
+  {
+    std::string event;
+    std::string data;
+    std::string id;
+    std::optional<std::uint64_t> retry;
+  };
+
+  std::string format_sse_event(const SseEvent& event);
+  std::string format_sse_comment(const std::string& comment);
+}
+
+#endif
diff --git a/framework/sse/sse_parser.cpp b/framework/sse/sse_parser.cpp
new file mode 100644
index 0000000..0607349
--- /dev/null
+++ b/framework/sse/sse_parser.cpp
@@ -0,0 +1,122 @@
+#include "sse/sse_parser.hpp"
+
+#include <algorithm>
+#include <cctype>
+#include <limits>
+
+namespace khttpd::framework::sse
+{
+  namespace
+  {
+    std::string single_line(std::string value)
+    {
+      value.erase(std::remove_if(value.begin(), value.end(), [](const char c) { return c == '\r' || c == '\n'; }), value.end());
+      return value;
+    }
+
+    void append_data_lines(std::string& output, const std::string& data)
+    {
+      std::size_t begin = 0;
+      while (true)
+      {
+        const auto end = data.find('\n', begin);
+        output += "data: " + data.substr(begin, end == std::string::npos ? end : end - begin) + "\n";
+        if (end == std::string::npos) break;
+        begin = end + 1;
+      }
+    }
+  }
+
+  std::string format_sse_event(const SseEvent& event)
+  {
+    std::string output;
+    if (!event.event.empty()) output += "event: " + single_line(event.event) + "\n";
+    if (!event.id.empty()) output += "id: " + single_line(event.id) + "\n";
+    if (event.retry) output += "retry: " + std::to_string(*event.retry) + "\n";
+    append_data_lines(output, event.data);
+    output += "\n";
+    return output;
+  }
+
+  std::string format_sse_comment(const std::string& comment)
+  {
+    return ": " + single_line(comment) + "\n\n";
+  }
+
+  std::vector<SseEvent> SseParser::feed(const std::string_view bytes)
+  {
+    buffer_.append(bytes.data(), bytes.size());
+    std::vector<SseEvent> events;
+
+    if (skip_leading_lf_ && !buffer_.empty())
+    {
+      if (buffer_.front() == '\n') buffer_.erase(0, 1);
+      skip_leading_lf_ = false;
+    }
+    if (at_stream_start_)
+    {
+      constexpr std::string_view utf8_bom{"\xef\xbb\xbf", 3};
+      const auto prefix_size = std::min(buffer_.size(), utf8_bom.size());
+      if (buffer_.compare(0, prefix_size, utf8_bom.data(), prefix_size) == 0 &&
+          buffer_.size() < utf8_bom.size())
+        return events;
+      if (buffer_.compare(0, utf8_bom.size(), utf8_bom.data(), utf8_bom.size()) == 0)
+        buffer_.erase(0, utf8_bom.size());
+      at_stream_start_ = false;
+    }
+
+    std::size_t consumed = 0;
+    while (consumed < buffer_.size())
+    {
+      const auto newline = buffer_.find_first_of("\r\n", consumed);
+      if (newline == std::string::npos) break;
+      const auto line = buffer_.substr(consumed, newline - consumed);
+      process_line(line, events);
+      if (buffer_[newline] == '\r' && newline + 1 < buffer_.size() && buffer_[newline + 1] == '\n')
+        consumed = newline + 2;
+      else
+      {
+        consumed = newline + 1;
+        if (buffer_[newline] == '\r' && consumed == buffer_.size()) skip_leading_lf_ = true;
+      }
+    }
+    if (consumed != 0) buffer_.erase(0, consumed);
+    return events;
+  }
+
+  void SseParser::process_line(const std::string& line, std::vector<SseEvent>& events)
+  {
+    if (line.empty())
+    {
+      if (!data_.empty())
+      {
+        data_.pop_back();
+        events.push_back({event_.empty() ? "message" : event_, data_, last_event_id_, retry_});
+      }
+      event_.clear();
+      data_.clear();
+      retry_.reset();
+      return;
+    }
+    if (line.front() == ':') return;
+    const auto colon = line.find(':');
+    const auto field = line.substr(0, colon);
+    auto value = colon == std::string::npos ? std::string{} : line.substr(colon + 1);
+    if (!value.empty() && value.front() == ' ') value.erase(0, 1);
+    if (field == "event") event_ = value;
+    else if (field == "data") { data_ += value; data_ += '\n'; }
+    else if (field == "id" && value.find('\0') == std::string::npos) last_event_id_ = value;
+    else if (field == "retry" && !value.empty() &&
+             std::all_of(value.begin(), value.end(),
+                         [](const unsigned char c) { return std::isdigit(c) != 0; }))
+    {
+      try { retry_ = std::stoull(value); } catch (...) { retry_.reset(); }
+    }
+  }
+
+  void SseParser::reset()
+  {
+    buffer_.clear(); event_.clear(); data_.clear(); last_event_id_.clear(); retry_.reset();
+    at_stream_start_ = true; skip_leading_lf_ = false;
+  }
+}
diff --git a/framework/sse/sse_parser.hpp b/framework/sse/sse_parser.hpp
new file mode 100644
index 0000000..24e304d
--- /dev/null
+++ b/framework/sse/sse_parser.hpp
@@ -0,0 +1,30 @@
+#ifndef KHTTPD_FRAMEWORK_SSE_SSE_PARSER_HPP
+#define KHTTPD_FRAMEWORK_SSE_SSE_PARSER_HPP
+
+#include "sse/sse_event.hpp"
+
+#include <string>
+#include <string_view>
+#include <vector>
+
+namespace khttpd::framework::sse
+{
+  class SseParser
+  {
+  public:
+    std::vector<SseEvent> feed(std::string_view bytes);
+    void reset();
+
+  private:
+    void process_line(const std::string& line, std::vector<SseEvent>& events);
+    std::string buffer_;
+    std::string event_;
+    std::string data_;
+    std::string last_event_id_;
+    std::optional<std::uint64_t> retry_;
+    bool at_stream_start_ = true;
+    bool skip_leading_lf_ = false;
+  };
+}
+
+#endif
diff --git a/framework/sse/sse_session.cpp b/framework/sse/sse_session.cpp
new file mode 100644
index 0000000..c149d7c
--- /dev/null
+++ b/framework/sse/sse_session.cpp
@@ -0,0 +1,126 @@
+#include "sse/sse_session.hpp"
+
+#include <boost/asio/buffer.hpp>
+#include <deque>
+#include <mutex>
+#include <utility>
+
+namespace khttpd::framework::sse
+{
+  struct SseSession::Impl : std::enable_shared_from_this<Impl>
+  {
+    std::shared_ptr<HttpResponseStream> response;
+    int version;
+    bool keep_alive;
+    mutable std::mutex mutex;
+    std::deque<std::shared_ptr<std::string>> queue;
+    bool started = false;
+    bool writing = false;
+    bool closing = false;
+    bool closed = false;
+    boost::system::error_code close_error;
+    CloseHandler close_handler;
+
+    Impl(std::shared_ptr<HttpResponseStream> value, const int v, const bool keep)
+      : response(std::move(value)), version(v), keep_alive(keep) {}
+
+    void start()
+    {
+      HttpResponseStream::ResponseHead head{boost::beast::http::status::ok, version};
+      head.keep_alive(keep_alive);
+      head.set(boost::beast::http::field::content_type, "text/event-stream");
+      head.set(boost::beast::http::field::cache_control, "no-cache");
+      head.set("X-Accel-Buffering", "no");
+      response->async_start(std::move(head), [self = shared_from_this()](boost::system::error_code ec)
+      {
+        if (ec) return self->finish(ec);
+        { std::lock_guard lock(self->mutex); self->started = true; }
+        self->write_next();
+      });
+    }
+
+    bool enqueue(std::string value)
+    {
+      {
+        std::lock_guard lock(mutex);
+        if (closing || closed) return false;
+        queue.push_back(std::make_shared<std::string>(std::move(value)));
+      }
+      write_next();
+      return true;
+    }
+
+    void write_next()
+    {
+      std::shared_ptr<std::string> value;
+      bool should_finish = false;
+      {
+        std::lock_guard lock(mutex);
+        if (!started || writing || closed) return;
+        if (queue.empty())
+        {
+          if (!closing) return;
+          writing = true;
+          should_finish = true;
+        }
+        else
+        {
+          writing = true;
+          value = queue.front();
+        }
+      }
+      if (should_finish)
+      {
+        response->async_finish([self = shared_from_this()](boost::system::error_code ec) { self->finish(ec); });
+        return;
+      }
+      response->async_write_some(boost::asio::buffer(*value),
+        [self = shared_from_this(), value](boost::system::error_code ec)
+        {
+          if (ec) return self->finish(ec);
+          {
+            std::lock_guard lock(self->mutex);
+            if (!self->queue.empty()) self->queue.pop_front();
+            self->writing = false;
+          }
+          self->write_next();
+        });
+    }
+
+    void finish(boost::system::error_code ec)
+    {
+      CloseHandler handler;
+      {
+        std::lock_guard lock(mutex);
+        if (closed) return;
+        closed = true;
+        close_error = ec;
+        writing = false;
+        queue.clear();
+        handler = std::move(close_handler);
+      }
+      if (handler) handler(ec);
+    }
+  };
+
+  SseSession::SseSession(std::shared_ptr<HttpResponseStream> response, const int version, const bool keep_alive)
+    : impl_(std::make_shared<Impl>(std::move(response), version, keep_alive)) {}
+  void SseSession::start() { impl_->start(); }
+  bool SseSession::send(SseEvent event) { return impl_->enqueue(format_sse_event(event)); }
+  bool SseSession::send_comment(std::string comment) { return impl_->enqueue(format_sse_comment(comment)); }
+  void SseSession::close() { { std::lock_guard lock(impl_->mutex); impl_->closing = true; } impl_->write_next(); }
+  void SseSession::cancel() { impl_->response->cancel(); impl_->finish(boost::asio::error::operation_aborted); }
+  void SseSession::on_close(CloseHandler handler)
+  {
+    boost::system::error_code error;
+    bool already_closed = false;
+    {
+      std::lock_guard lock(impl_->mutex);
+      already_closed = impl_->closed;
+      error = impl_->close_error;
+      if (!already_closed) impl_->close_handler = std::move(handler);
+    }
+    if (already_closed && handler) handler(error);
+  }
+  bool SseSession::is_open() const { std::lock_guard lock(impl_->mutex); return !impl_->closing && !impl_->closed; }
+}
diff --git a/framework/sse/sse_session.hpp b/framework/sse/sse_session.hpp
new file mode 100644
index 0000000..cf57cc8
--- /dev/null
+++ b/framework/sse/sse_session.hpp
@@ -0,0 +1,33 @@
+#ifndef KHTTPD_FRAMEWORK_SSE_SSE_SESSION_HPP
+#define KHTTPD_FRAMEWORK_SSE_SSE_SESSION_HPP
+
+#include "context/http_response_stream.hpp"
+#include "sse/sse_event.hpp"
+
+#include <functional>
+#include <memory>
+#include <string>
+
+namespace khttpd::framework::sse
+{
+  class SseSession : public std::enable_shared_from_this<SseSession>
+  {
+  public:
+    using CloseHandler = std::function<void(boost::system::error_code)>;
+
+    SseSession(std::shared_ptr<HttpResponseStream> response, int http_version, bool keep_alive);
+    void start();
+    bool send(SseEvent event);
+    bool send_comment(std::string comment);
+    void close();
+    void cancel();
+    void on_close(CloseHandler handler);
+    bool is_open() const;
+
+  private:
+    struct Impl;
+    std::shared_ptr<Impl> impl_;
+  };
+}
+
+#endif
diff --git a/framework/tests/BUILD.bazel b/framework/tests/BUILD.bazel
index 79076e0..001d1fd 100644
--- a/framework/tests/BUILD.bazel
+++ b/framework/tests/BUILD.bazel
@@ -1,5 +1,16 @@
 load("@rules_cc//cc:defs.bzl", "cc_test")
 
+cc_test(
+    name = "sse_test",
+    srcs = ["http_session_test_harness.hpp", "sse_test.cpp"],
+    copts = ["-std=c++17", "-Wall", "-pedantic"],
+    deps = [
+        "//framework",
+        "@googletest//:gtest",
+        "@googletest//:gtest_main",
+    ],
+)
+
 cc_test(
     name = "context_test",
     srcs = ["context_test.cpp"],
diff --git a/framework/tests/sse_test.cpp b/framework/tests/sse_test.cpp
new file mode 100644
index 0000000..1077a25
--- /dev/null
+++ b/framework/tests/sse_test.cpp
@@ -0,0 +1,208 @@
+#include <gtest/gtest.h>
+
+#include <boost/asio.hpp>
+#include <boost/beast/http.hpp>
+#include <atomic>
+#include <thread>
+
+#include "client/sse_client.hpp"
+#include "sse/sse_event.hpp"
+#include "sse/sse_parser.hpp"
+#include "sse/sse_session.hpp"
+#include "http_session_test_harness.hpp"
+
+namespace sse = khttpd::framework::sse;
+namespace fw = khttpd::framework;
+namespace test = khttpd::framework::tests;
+namespace http = boost::beast::http;
+namespace net = boost::asio;
+namespace client = khttpd::framework::client;
+using tcp = net::ip::tcp;
+
+TEST(SseParserTest, PreservesAFragmentedMultilineEvent)
+{
+  sse::SseParser parser;
+  EXPECT_TRUE(parser.feed("event: con").empty());
+  EXPECT_TRUE(parser.feed("fig\r\nid: 42\r\ndata: first\r\n").empty());
+  const auto events = parser.feed("data: second\r\nretry: 1500\r\n\r\n");
+
+  ASSERT_EQ(events.size(), 1U);
+  EXPECT_EQ(events[0].event, "config");
+  EXPECT_EQ(events[0].id, "42");
+  EXPECT_EQ(events[0].data, "first\nsecond");
+  ASSERT_TRUE(events[0].retry.has_value());
+  EXPECT_EQ(*events[0].retry, 1500U);
+}
+
+TEST(SseParserTest, IgnoresCommentsAndInvalidRetryWithoutLosingData)
+{
+  sse::SseParser parser;
+  const auto events = parser.feed(": heartbeat\nretry: soon\ndata: ready\n\n");
+  ASSERT_EQ(events.size(), 1U);
+  EXPECT_EQ(events[0].data, "ready");
+  EXPECT_FALSE(events[0].retry.has_value());
+}
+
+TEST(SseParserTest, AcceptsFragmentedUtf8BomAndStandaloneCarriageReturns)
+{
+  sse::SseParser parser;
+  EXPECT_TRUE(parser.feed("\xef").empty());
+  EXPECT_TRUE(parser.feed("\xbb\xbf" "data: one\rdata: two\r").empty());
+  const auto events = parser.feed("\r");
+
+  ASSERT_EQ(events.size(), 1U);
+  EXPECT_EQ(events[0].event, "message");
+  EXPECT_EQ(events[0].data, "one\ntwo");
+}
+
+TEST(SseEventTest, FormatsEveryDataLineAndOptionalFields)
+{
+  sse::SseEvent event{"instances", "one\ntwo", "9", 2500};
+  EXPECT_EQ(sse::format_sse_event(event),
+            "event: instances\nid: 9\nretry: 2500\ndata: one\ndata: two\n\n");
+}
+
+TEST(SseSessionTest, StreamsQueuedEventsInOrderWithSseHeaders)
+{
+  test::TempWebRoot web_root;
+  fw::HttpRouter router;
+  fw::WebsocketRouter websocket_router;
+  router.sse("/events", [](fw::HttpContext&, std::shared_ptr<sse::SseSession> session)
+  {
+    session->send({"config", "first", "1", std::nullopt});
+    session->send({"instances", "second", "2", std::nullopt});
+    session->close();
+  });
+  http::request<http::string_body> request{http::verb::get, "/events", 11};
+  request.keep_alive(false);
+  const auto response = test::round_trip(router, websocket_router, web_root.path, std::move(request));
+
+  EXPECT_EQ(response.result(), http::status::ok);
+  EXPECT_EQ(response[http::field::content_type], "text/event-stream");
+  EXPECT_EQ(response[http::field::cache_control], "no-cache");
+  EXPECT_EQ(response["X-Accel-Buffering"], "no");
+  EXPECT_EQ(response.body(), "event: config\nid: 1\ndata: first\n\nevent: instances\nid: 2\ndata: second\n\n");
+}
+
+TEST(SseClientTest, DeliversEventsFromAnAsyncEventStream)
+{
+  net::io_context server_ioc;
+  tcp::acceptor acceptor(server_ioc, {net::ip::address_v4::loopback(), 0});
+  const auto port = acceptor.local_endpoint().port();
+  std::atomic<bool> accepted_sse{false};
+  std::thread server([&]
+  {
+    tcp::socket socket(server_ioc);
+    acceptor.accept(socket);
+    boost::beast::flat_buffer buffer;
+    http::request<http::empty_body> request;
+    http::read(socket, buffer, request);
+    accepted_sse = request[http::field::accept] == "text/event-stream" &&
+                   request["Last-Event-ID"] == "41";
+
+    const std::string head =
+      "HTTP/1.1 200 OK\r\n"
+      "Content-Type: text/event-stream; charset=utf-8\r\n"
+      "Connection: close\r\n\r\n";
+    net::write(socket, net::buffer(head));
+    net::write(socket, net::buffer(std::string("event: con")));
+    net::write(socket, net::buffer(std::string("fig\ndata: changed\nid: 42\n\n")));
+    boost::system::error_code ignored;
+    socket.shutdown(tcp::socket::shutdown_send, ignored);
+  });
+
+  net::io_context ioc;
+  auto stream = std::make_shared<client::SseClient>(ioc);
+  std::vector<sse::SseEvent> events;
+  boost::system::error_code closed_with;
+  bool closed = false;
+  stream->connect(
+    "http://127.0.0.1:" + std::to_string(port) + "/events",
+    {{"Last-Event-ID", "41"}},
+    [&](const sse::SseEvent& event) { events.push_back(event); },
+    [&](boost::system::error_code ec) { closed_with = ec; closed = true; });
+  ioc.run();
+  server.join();
+
+  EXPECT_TRUE(accepted_sse);
+  EXPECT_TRUE(closed);
+  EXPECT_FALSE(closed_with) << closed_with.message();
+  ASSERT_EQ(events.size(), 1U);
+  EXPECT_EQ(events[0].event, "config");
+  EXPECT_EQ(events[0].data, "changed");
+  EXPECT_EQ(events[0].id, "42");
+}
+
+TEST(SseClientTest, CancelClosesTheStreamAndCompletesExactlyOnce)
+{
+  net::io_context server_ioc;
+  tcp::acceptor acceptor(server_ioc, {net::ip::address_v4::loopback(), 0});
+  const auto port = acceptor.local_endpoint().port();
+  std::atomic<bool> peer_closed{false};
+  std::thread server([&]
+  {
+    tcp::socket socket(server_ioc);
+    acceptor.accept(socket);
+    boost::beast::flat_buffer buffer;
+    http::request<http::empty_body> request;
+    http::read(socket, buffer, request);
+    const std::string event = "data: stop\n\n";
+    const std::string wire = std::string{
+      "HTTP/1.1 200 OK\r\n"
+      "Content-Type: text/event-stream\r\n"
+      "Transfer-Encoding: chunked\r\n\r\n"} +
+      "c\r\n" + event + "\r\n";
+    net::write(socket, net::buffer(wire));
+    std::array<char, 1> ignored{};
+    boost::system::error_code ec;
+    socket.read_some(net::buffer(ignored), ec);
+    peer_closed = static_cast<bool>(ec);
+  });
+
+  net::io_context ioc;
+  auto stream = std::make_shared<client::SseClient>(ioc);
+  int close_count = 0;
+  boost::system::error_code closed_with;
+  stream->connect(
+    "http://127.0.0.1:" + std::to_string(port) + "/events", {},
+    [stream](const sse::SseEvent&) { stream->cancel(); },
+    [&](boost::system::error_code ec) { ++close_count; closed_with = ec; });
+  ioc.run();
+  server.join();
+
+  EXPECT_TRUE(peer_closed);
+  EXPECT_EQ(close_count, 1);
+  EXPECT_EQ(closed_with, net::error::operation_aborted);
+}
+
+TEST(SseClientTest, RejectsANonEventStreamResponse)
+{
+  net::io_context server_ioc;
+  tcp::acceptor acceptor(server_ioc, {net::ip::address_v4::loopback(), 0});
+  const auto port = acceptor.local_endpoint().port();
+  std::thread server([&]
+  {
+    tcp::socket socket(server_ioc);
+    acceptor.accept(socket);
+    boost::beast::flat_buffer buffer;
+    http::request<http::empty_body> request;
+    http::read(socket, buffer, request);
+    http::response<http::empty_body> response{http::status::ok, 11};
+    response.set(http::field::content_type, "application/json");
+    response.content_length(0);
+    response.keep_alive(false);
+    http::write(socket, response);
+  });
+
+  net::io_context ioc;
+  auto stream = std::make_shared<client::SseClient>(ioc);
+  boost::system::error_code closed_with;
+  stream->connect(
+    "http://127.0.0.1:" + std::to_string(port) + "/events", {},
+    [](const sse::SseEvent&) { FAIL() << "non-SSE response emitted an event"; },
+    [&](boost::system::error_code ec) { closed_with = ec; });
+  ioc.run();
+  server.join();
+
+  EXPECT_EQ(closed_with, make_error_code(boost::system::errc::protocol_error));
+}

From 052f9f05ee7a7001a0c424d7f2bb891143658de2 Mon Sep 17 00:00:00 2001
From: kekxv <caesar@kekxv.com>
Date: Fri, 21 Aug 2026 14:45:08 +0000
Subject: [PATCH 3/6] fix: harden async SSE and release 0.4.5

---
 MODULE.bazel                     |   2 +-
 README.md                        |  11 +-
 doc/server-sent-events.md        |  14 ++-
 example/MODULE.bazel             |   2 +-
 framework/client/sse_client.cpp  |  27 +++--
 framework/client/sse_client.hpp  |   7 +-
 framework/router/http_router.cpp |  11 +-
 framework/router/http_router.hpp |   3 +-
 framework/sse/sse_parser.cpp     |  88 +++++++++------
 framework/sse/sse_parser.hpp     |   6 ++
 framework/sse/sse_session.cpp    |  28 ++++-
 framework/sse/sse_session.hpp    |   5 +-
 framework/tests/sse_test.cpp     | 178 +++++++++++++++++++++++++++++++
 13 files changed, 324 insertions(+), 58 deletions(-)

diff --git a/MODULE.bazel b/MODULE.bazel
index 498a022..63a3ed2 100644
--- a/MODULE.bazel
+++ b/MODULE.bazel
@@ -1,6 +1,6 @@
 module(
     name = "khttpd",
-    version = "0.4.4",
+    version = "0.4.5",
 )
 
 bazel_dep(name = "platforms", version = "1.1.0")
diff --git a/README.md b/README.md
index d82f861..5e3075a 100644
--- a/README.md
+++ b/README.md
@@ -448,12 +448,19 @@ router.sse("/events",
   });
 ```
 
+The optional third `router.sse` argument sets the maximum queued wire bytes per
+connection (1 MiB by default). `send()` returns `false` when that limit would be
+exceeded, allowing the application to drop or resynchronize a slow subscriber.
+SSE routes run the ordinary pre-request interceptor chain before their handler;
+an authentication or authorization interceptor can reject the request before
+any event-stream response headers are written.
+
 `SseClient` uses the same fixed-buffer streaming transport and delivers each
 complete event as soon as it arrives. Arbitrarily split lines, CRLF/LF,
 multiline `data`, comments, `id`, and numeric `retry` fields are supported.
 
 ```cpp
-auto events = std::make_shared<client::SseClient>(ioc);
+auto events = std::make_shared<client::SseClient>(ioc); // 1 MiB event limit
 events->connect(
   "https://config.internal/events",
   {{"Authorization", "Bearer internal-token"}, {"Last-Event-ID", "41"}},
@@ -469,6 +476,8 @@ Keep the `SseClient` alive for the duration of the subscription. `cancel()`
 closes the transport and completes the close callback once with
 `operation_aborted`. Automatic reconnect is intentionally left to the caller,
 which can apply service-specific backoff and send `Last-Event-ID`.
+The optional constructor limit bounds an unfinished line or event; exceeding it
+closes only that subscription and reports `message_size`.
 
 ### Tests
 
diff --git a/doc/server-sent-events.md b/doc/server-sent-events.md
index de466b7..62d77c7 100644
--- a/doc/server-sent-events.md
+++ b/doc/server-sent-events.md
@@ -28,6 +28,16 @@ router.sse("/events",
   });
 ```
 
+`router.sse` 的可选第三个参数用于设置单连接待发送队列的字节上限，默认 1 MiB：
+
+```cpp
+router.sse("/events", handler, 256 * 1024);
+```
+
+队列即将超过上限时 `send()` 返回 `false`，业务可丢弃慢订阅者或触发全量同步，
+不会继续占用内存。SSE 路由在 handler 前运行普通的全局同步/异步拦截器，鉴权或
+权限拦截器返回 `Stop` 时不会发送 SSE 响应头，也不会执行 handler。
+
 响应会包含 `Content-Type: text/event-stream`、`Cache-Control: no-cache` 和
 `X-Accel-Buffering: no`，正文使用 HTTP 分块传输。`send()` 和
 `send_comment()` 可从不同线程调用，内部拥有待写入字符串并按 FIFO 顺序串行写入，
@@ -45,7 +55,7 @@ router.sse("/events",
 ```cpp
 #include "client/sse_client.hpp"
 
-auto subscription = std::make_shared<client::SseClient>(ioc);
+auto subscription = std::make_shared<client::SseClient>(ioc); // 默认单事件上限 1 MiB
 subscription->connect(
   "https://config.internal/events",
   {
@@ -67,6 +77,8 @@ subscription->connect(
 调用方需要在订阅期间持有 `SseClient`。显式调用 `cancel()` 会关闭传输，并以
 `operation_aborted` 调用一次关闭回调。框架不自动重连：调用方可以结合事件的
 `retry`、指数退避和 `Last-Event-ID` 实现符合具体服务需求的恢复策略。
+构造函数的可选字节上限用于限制未完成行或单个事件；超限只会关闭当前订阅，并以
+`message_size` 完成关闭回调，不会终止进程。
 
 ## 选择 SSE 还是 WebSocket
 
diff --git a/example/MODULE.bazel b/example/MODULE.bazel
index e369abf..6b126f2 100644
--- a/example/MODULE.bazel
+++ b/example/MODULE.bazel
@@ -8,7 +8,7 @@ bazel_dep(name = "boost", version = "1.90.0.bcr.1")
 bazel_dep(name = "boost.asio", version = "1.90.0.bcr.1")
 bazel_dep(name = "boost.mysql", version = "1.90.0.bcr.1")
 bazel_dep(name = "spdlog", version = "1.17.0")
-bazel_dep(name = "khttpd", version = "0.4.4")
+bazel_dep(name = "khttpd", version = "0.4.5")
 local_path_override(
     module_name = "khttpd",
     path = "..",
diff --git a/framework/client/sse_client.cpp b/framework/client/sse_client.cpp
index bf4c768..8e6f705 100644
--- a/framework/client/sse_client.cpp
+++ b/framework/client/sse_client.cpp
@@ -7,6 +7,7 @@
 #include <array>
 #include <cctype>
 #include <mutex>
+#include <stdexcept>
 #include <utility>
 
 #include "client/http_client_stream.hpp"
@@ -46,9 +47,10 @@ namespace khttpd::framework::client
     bool connect_started = false;
     bool terminal = false;
 
-    explicit Impl(net::io_context& ioc) : stream(std::make_shared<HttpClientStream>(ioc)) {}
-    Impl(net::io_context& ioc, net::ssl::context& context)
-      : stream(std::make_shared<HttpClientStream>(ioc, context)) {}
+    Impl(net::io_context& ioc, const std::size_t max_event_bytes)
+      : stream(std::make_shared<HttpClientStream>(ioc)), parser(max_event_bytes) {}
+    Impl(net::io_context& ioc, net::ssl::context& context, const std::size_t max_event_bytes)
+      : stream(std::make_shared<HttpClientStream>(ioc, context)), parser(max_event_bytes) {}
 
     void connect(const std::string& url,
                  const std::map<std::string, std::string>& headers,
@@ -110,8 +112,15 @@ namespace khttpd::framework::client
         {
           if (size != 0)
           {
-            const auto events = self->parser.feed(std::string_view(self->read_buffer.data(), size));
-            for (const auto& event : events) self->emit(event);
+            try
+            {
+              const auto events = self->parser.feed(std::string_view(self->read_buffer.data(), size));
+              for (const auto& event : events) self->emit(event);
+            }
+            catch (const std::length_error&)
+            {
+              return self->finish(net::error::message_size);
+            }
           }
           if (self->is_terminal()) return;
           if (done) return self->finish({});
@@ -170,9 +179,11 @@ namespace khttpd::framework::client
     }
   };
 
-  SseClient::SseClient(net::io_context& ioc) : impl_(std::make_shared<Impl>(ioc)) {}
-  SseClient::SseClient(net::io_context& ioc, net::ssl::context& context)
-    : impl_(std::make_shared<Impl>(ioc, context)) {}
+  SseClient::SseClient(net::io_context& ioc, const std::size_t max_event_bytes)
+    : impl_(std::make_shared<Impl>(ioc, max_event_bytes)) {}
+  SseClient::SseClient(net::io_context& ioc, net::ssl::context& context,
+                       const std::size_t max_event_bytes)
+    : impl_(std::make_shared<Impl>(ioc, context, max_event_bytes)) {}
   SseClient::~SseClient() { if (impl_) impl_->abandon(); }
   void SseClient::connect(const std::string& url,
                           const std::map<std::string, std::string>& headers,
diff --git a/framework/client/sse_client.hpp b/framework/client/sse_client.hpp
index 91ebf5d..d309179 100644
--- a/framework/client/sse_client.hpp
+++ b/framework/client/sse_client.hpp
@@ -4,6 +4,7 @@
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/ssl/context.hpp>
 #include <boost/system/error_code.hpp>
+#include <cstddef>
 #include <functional>
 #include <map>
 #include <memory>
@@ -19,8 +20,10 @@ namespace khttpd::framework::client
     using EventHandler = std::function<void(const sse::SseEvent&)>;
     using CloseHandler = std::function<void(boost::system::error_code)>;
 
-    explicit SseClient(boost::asio::io_context& ioc);
-    SseClient(boost::asio::io_context& ioc, boost::asio::ssl::context& ssl_context);
+    explicit SseClient(boost::asio::io_context& ioc,
+                       std::size_t max_event_bytes = 1024 * 1024);
+    SseClient(boost::asio::io_context& ioc, boost::asio::ssl::context& ssl_context,
+              std::size_t max_event_bytes = 1024 * 1024);
     ~SseClient();
 
     SseClient(const SseClient&) = delete;
diff --git a/framework/router/http_router.cpp b/framework/router/http_router.cpp
index c6bc972..222f05d 100644
--- a/framework/router/http_router.cpp
+++ b/framework/router/http_router.cpp
@@ -334,16 +334,19 @@ namespace khttpd::framework
     return false;
   }
 
-  void HttpRouter::sse(const std::string& path, SseHandler handler)
+  void HttpRouter::sse(const std::string& path, SseHandler handler, const std::size_t max_pending_bytes)
   {
     stream(path, boost::beast::http::verb::get,
-      [handler = std::move(handler)](HttpContext& ctx, std::shared_ptr<HttpRequestStream>,
+      [handler = std::move(handler), max_pending_bytes](HttpContext& ctx, std::shared_ptr<HttpRequestStream>,
                                      std::shared_ptr<HttpResponseStream> response, HttpStreamComplete)
       {
+        const auto content_length = ctx.get_request()[boost::beast::http::field::content_length];
+        if (ctx.get_request().chunked() || (!content_length.empty() && content_length != "0"))
+          response->cancel_request_body();
         auto session = std::make_shared<sse::SseSession>(std::move(response), ctx.get_request().version(),
-                                                        ctx.get_request().keep_alive());
+                                                        ctx.get_request().keep_alive(), max_pending_bytes);
+        handler(ctx, session);
         session->start();
-        handler(ctx, std::move(session));
       });
   }
 
diff --git a/framework/router/http_router.hpp b/framework/router/http_router.hpp
index ef59753..af83fa2 100644
--- a/framework/router/http_router.hpp
+++ b/framework/router/http_router.hpp
@@ -249,7 +249,8 @@ namespace khttpd::framework
     // Async handlers must invoke complete exactly once, from any thread.
     void async_route(const std::string& path, boost::beast::http::verb method, HttpAsyncHandler handler);
     void stream(const std::string& path, boost::beast::http::verb method, HttpStreamHandler handler);
-    void sse(const std::string& path, SseHandler handler);
+    void sse(const std::string& path, SseHandler handler,
+             std::size_t max_pending_bytes = 1024 * 1024);
 
     // Used by HttpSession after it has parsed only the request header.
     bool is_stream_route(const std::string& path, boost::beast::http::verb method) const;
diff --git a/framework/sse/sse_parser.cpp b/framework/sse/sse_parser.cpp
index 0607349..aeab727 100644
--- a/framework/sse/sse_parser.cpp
+++ b/framework/sse/sse_parser.cpp
@@ -3,6 +3,7 @@
 #include <algorithm>
 #include <cctype>
 #include <limits>
+#include <stdexcept>
 
 namespace khttpd::framework::sse
 {
@@ -19,10 +20,10 @@ namespace khttpd::framework::sse
       std::size_t begin = 0;
       while (true)
       {
-        const auto end = data.find('\n', begin);
+        const auto end = data.find_first_of("\r\n", begin);
         output += "data: " + data.substr(begin, end == std::string::npos ? end : end - begin) + "\n";
         if (end == std::string::npos) break;
-        begin = end + 1;
+        begin = end + (data[end] == '\r' && end + 1 < data.size() && data[end + 1] == '\n' ? 2 : 1);
       }
     }
   }
@@ -45,42 +46,56 @@ namespace khttpd::framework::sse
 
   std::vector<SseEvent> SseParser::feed(const std::string_view bytes)
   {
-    buffer_.append(bytes.data(), bytes.size());
     std::vector<SseEvent> events;
-
-    if (skip_leading_lf_ && !buffer_.empty())
-    {
-      if (buffer_.front() == '\n') buffer_.erase(0, 1);
-      skip_leading_lf_ = false;
-    }
-    if (at_stream_start_)
+    std::size_t offset = 0;
+    const auto configured_chunk_limit = max_event_bytes_ == (std::numeric_limits<std::size_t>::max)()
+      ? max_event_bytes_ : max_event_bytes_ + 1;
+    const auto chunk_limit = std::max<std::size_t>(configured_chunk_limit, 3);
+    while (offset < bytes.size())
     {
-      constexpr std::string_view utf8_bom{"\xef\xbb\xbf", 3};
-      const auto prefix_size = std::min(buffer_.size(), utf8_bom.size());
-      if (buffer_.compare(0, prefix_size, utf8_bom.data(), prefix_size) == 0 &&
-          buffer_.size() < utf8_bom.size())
-        return events;
-      if (buffer_.compare(0, utf8_bom.size(), utf8_bom.data(), utf8_bom.size()) == 0)
-        buffer_.erase(0, utf8_bom.size());
-      at_stream_start_ = false;
-    }
+      const auto available = chunk_limit - buffer_.size();
+      const auto chunk_size = std::min(available, bytes.size() - offset);
+      buffer_.append(bytes.data() + offset, chunk_size);
+      offset += chunk_size;
 
-    std::size_t consumed = 0;
-    while (consumed < buffer_.size())
-    {
-      const auto newline = buffer_.find_first_of("\r\n", consumed);
-      if (newline == std::string::npos) break;
-      const auto line = buffer_.substr(consumed, newline - consumed);
-      process_line(line, events);
-      if (buffer_[newline] == '\r' && newline + 1 < buffer_.size() && buffer_[newline + 1] == '\n')
-        consumed = newline + 2;
-      else
+      if (skip_leading_lf_ && !buffer_.empty())
+      {
+        if (buffer_.front() == '\n') buffer_.erase(0, 1);
+        skip_leading_lf_ = false;
+      }
+      if (at_stream_start_)
+      {
+        constexpr std::string_view utf8_bom{"\xef\xbb\xbf", 3};
+        const auto prefix_size = std::min(buffer_.size(), utf8_bom.size());
+        if (buffer_.compare(0, prefix_size, utf8_bom.data(), prefix_size) == 0 &&
+            buffer_.size() < utf8_bom.size())
+          continue;
+        if (buffer_.compare(0, utf8_bom.size(), utf8_bom.data(), utf8_bom.size()) == 0)
+          buffer_.erase(0, utf8_bom.size());
+        at_stream_start_ = false;
+      }
+
+      std::size_t consumed = 0;
+      while (consumed < buffer_.size())
       {
-        consumed = newline + 1;
-        if (buffer_[newline] == '\r' && consumed == buffer_.size()) skip_leading_lf_ = true;
+        const auto newline = buffer_.find_first_of("\r\n", consumed);
+        if (newline == std::string::npos) break;
+        if (newline - consumed > max_event_bytes_)
+          throw std::length_error("SSE line exceeds configured event limit");
+        const auto line = buffer_.substr(consumed, newline - consumed);
+        process_line(line, events);
+        if (buffer_[newline] == '\r' && newline + 1 < buffer_.size() && buffer_[newline + 1] == '\n')
+          consumed = newline + 2;
+        else
+        {
+          consumed = newline + 1;
+          if (buffer_[newline] == '\r' && consumed == buffer_.size()) skip_leading_lf_ = true;
+        }
       }
+      if (consumed != 0) buffer_.erase(0, consumed);
+      if (!at_stream_start_ && buffer_.size() > max_event_bytes_)
+        throw std::length_error("SSE line exceeds configured event limit");
     }
-    if (consumed != 0) buffer_.erase(0, consumed);
     return events;
   }
 
@@ -103,8 +118,15 @@ namespace khttpd::framework::sse
     const auto field = line.substr(0, colon);
     auto value = colon == std::string::npos ? std::string{} : line.substr(colon + 1);
     if (!value.empty() && value.front() == ' ') value.erase(0, 1);
+    if (value.size() > max_event_bytes_)
+      throw std::length_error("SSE field exceeds configured event limit");
     if (field == "event") event_ = value;
-    else if (field == "data") { data_ += value; data_ += '\n'; }
+    else if (field == "data")
+    {
+      if (value.size() >= max_event_bytes_ || data_.size() > max_event_bytes_ - value.size() - 1)
+        throw std::length_error("SSE event data exceeds configured event limit");
+      data_ += value; data_ += '\n';
+    }
     else if (field == "id" && value.find('\0') == std::string::npos) last_event_id_ = value;
     else if (field == "retry" && !value.empty() &&
              std::all_of(value.begin(), value.end(),
diff --git a/framework/sse/sse_parser.hpp b/framework/sse/sse_parser.hpp
index 24e304d..c0cc3e0 100644
--- a/framework/sse/sse_parser.hpp
+++ b/framework/sse/sse_parser.hpp
@@ -3,6 +3,7 @@
 
 #include "sse/sse_event.hpp"
 
+#include <cstddef>
 #include <string>
 #include <string_view>
 #include <vector>
@@ -12,6 +13,10 @@ namespace khttpd::framework::sse
   class SseParser
   {
   public:
+    static constexpr std::size_t default_max_event_bytes = 1024 * 1024;
+
+    explicit SseParser(std::size_t max_event_bytes = default_max_event_bytes)
+      : max_event_bytes_(max_event_bytes) {}
     std::vector<SseEvent> feed(std::string_view bytes);
     void reset();
 
@@ -24,6 +29,7 @@ namespace khttpd::framework::sse
     std::optional<std::uint64_t> retry_;
     bool at_stream_start_ = true;
     bool skip_leading_lf_ = false;
+    std::size_t max_event_bytes_;
   };
 }
 
diff --git a/framework/sse/sse_session.cpp b/framework/sse/sse_session.cpp
index c149d7c..0240dd3 100644
--- a/framework/sse/sse_session.cpp
+++ b/framework/sse/sse_session.cpp
@@ -14,6 +14,9 @@ namespace khttpd::framework::sse
     bool keep_alive;
     mutable std::mutex mutex;
     std::deque<std::shared_ptr<std::string>> queue;
+    std::size_t pending_bytes = 0;
+    std::size_t max_pending_bytes;
+    bool start_requested = false;
     bool started = false;
     bool writing = false;
     bool closing = false;
@@ -21,11 +24,18 @@ namespace khttpd::framework::sse
     boost::system::error_code close_error;
     CloseHandler close_handler;
 
-    Impl(std::shared_ptr<HttpResponseStream> value, const int v, const bool keep)
-      : response(std::move(value)), version(v), keep_alive(keep) {}
+    Impl(std::shared_ptr<HttpResponseStream> value, const int v, const bool keep,
+         const std::size_t maximum_pending_bytes)
+      : response(std::move(value)), version(v), keep_alive(keep),
+        max_pending_bytes(maximum_pending_bytes) {}
 
     void start()
     {
+      {
+        std::lock_guard lock(mutex);
+        if (start_requested || closed) return;
+        start_requested = true;
+      }
       HttpResponseStream::ResponseHead head{boost::beast::http::status::ok, version};
       head.keep_alive(keep_alive);
       head.set(boost::beast::http::field::content_type, "text/event-stream");
@@ -44,6 +54,8 @@ namespace khttpd::framework::sse
       {
         std::lock_guard lock(mutex);
         if (closing || closed) return false;
+        if (value.size() > max_pending_bytes || pending_bytes > max_pending_bytes - value.size()) return false;
+        pending_bytes += value.size();
         queue.push_back(std::make_shared<std::string>(std::move(value)));
       }
       write_next();
@@ -80,7 +92,11 @@ namespace khttpd::framework::sse
           if (ec) return self->finish(ec);
           {
             std::lock_guard lock(self->mutex);
-            if (!self->queue.empty()) self->queue.pop_front();
+            if (!self->queue.empty())
+            {
+              self->pending_bytes -= self->queue.front()->size();
+              self->queue.pop_front();
+            }
             self->writing = false;
           }
           self->write_next();
@@ -96,6 +112,7 @@ namespace khttpd::framework::sse
         closed = true;
         close_error = ec;
         writing = false;
+        pending_bytes = 0;
         queue.clear();
         handler = std::move(close_handler);
       }
@@ -103,8 +120,9 @@ namespace khttpd::framework::sse
     }
   };
 
-  SseSession::SseSession(std::shared_ptr<HttpResponseStream> response, const int version, const bool keep_alive)
-    : impl_(std::make_shared<Impl>(std::move(response), version, keep_alive)) {}
+  SseSession::SseSession(std::shared_ptr<HttpResponseStream> response, const int version, const bool keep_alive,
+                         const std::size_t max_pending_bytes)
+    : impl_(std::make_shared<Impl>(std::move(response), version, keep_alive, max_pending_bytes)) {}
   void SseSession::start() { impl_->start(); }
   bool SseSession::send(SseEvent event) { return impl_->enqueue(format_sse_event(event)); }
   bool SseSession::send_comment(std::string comment) { return impl_->enqueue(format_sse_comment(comment)); }
diff --git a/framework/sse/sse_session.hpp b/framework/sse/sse_session.hpp
index cf57cc8..fb3a5d9 100644
--- a/framework/sse/sse_session.hpp
+++ b/framework/sse/sse_session.hpp
@@ -4,6 +4,7 @@
 #include "context/http_response_stream.hpp"
 #include "sse/sse_event.hpp"
 
+#include <cstddef>
 #include <functional>
 #include <memory>
 #include <string>
@@ -14,8 +15,10 @@ namespace khttpd::framework::sse
   {
   public:
     using CloseHandler = std::function<void(boost::system::error_code)>;
+    static constexpr std::size_t default_max_pending_bytes = 1024 * 1024;
 
-    SseSession(std::shared_ptr<HttpResponseStream> response, int http_version, bool keep_alive);
+    SseSession(std::shared_ptr<HttpResponseStream> response, int http_version, bool keep_alive,
+               std::size_t max_pending_bytes = default_max_pending_bytes);
     void start();
     bool send(SseEvent event);
     bool send_comment(std::string comment);
diff --git a/framework/tests/sse_test.cpp b/framework/tests/sse_test.cpp
index 1077a25..4458b4b 100644
--- a/framework/tests/sse_test.cpp
+++ b/framework/tests/sse_test.cpp
@@ -3,6 +3,7 @@
 #include <boost/asio.hpp>
 #include <boost/beast/http.hpp>
 #include <atomic>
+#include <stdexcept>
 #include <thread>
 
 #include "client/sse_client.hpp"
@@ -62,6 +63,39 @@ TEST(SseEventTest, FormatsEveryDataLineAndOptionalFields)
             "event: instances\nid: 9\nretry: 2500\ndata: one\ndata: two\n\n");
 }
 
+TEST(SseEventTest, BareCarriageReturnsCannotInjectEventFields)
+{
+  sse::SseEvent event{"config", "safe\rid: injected\r\nretry: 1", "trusted", std::nullopt};
+  EXPECT_EQ(sse::format_sse_event(event),
+            "event: config\nid: trusted\ndata: safe\ndata: id: injected\ndata: retry: 1\n\n");
+}
+
+TEST(SseParserTest, RejectsAnEventThatExceedsItsConfiguredMemoryLimit)
+{
+  sse::SseParser parser(32);
+  EXPECT_THROW(parser.feed("data: " + std::string(33, 'x') + "\n"), std::length_error);
+}
+
+TEST(SseParserTest, LimitAppliesPerEventRatherThanPerNetworkRead)
+{
+  sse::SseParser parser(32);
+  const auto events = parser.feed("data: one\n\ndata: two\n\ndata: three\n\n");
+
+  ASSERT_EQ(events.size(), 3U);
+  EXPECT_EQ(events[0].data, "one");
+  EXPECT_EQ(events[1].data, "two");
+  EXPECT_EQ(events[2].data, "three");
+}
+
+TEST(SseParserTest, MinimalLimitStillConsumesAFragmentedUtf8Bom)
+{
+  sse::SseParser parser(1);
+  EXPECT_TRUE(parser.feed("\xef").empty());
+  EXPECT_TRUE(parser.feed("\xbb").empty());
+  EXPECT_TRUE(parser.feed("\xbf").empty());
+  EXPECT_THROW(parser.feed("ab"), std::length_error);
+}
+
 TEST(SseSessionTest, StreamsQueuedEventsInOrderWithSseHeaders)
 {
   test::TempWebRoot web_root;
@@ -84,6 +118,114 @@ TEST(SseSessionTest, StreamsQueuedEventsInOrderWithSseHeaders)
   EXPECT_EQ(response.body(), "event: config\nid: 1\ndata: first\n\nevent: instances\nid: 2\ndata: second\n\n");
 }
 
+TEST(SseSessionTest, RejectsWritesBeyondTheConfiguredQueueLimit)
+{
+  test::TempWebRoot web_root;
+  fw::HttpRouter router;
+  fw::WebsocketRouter websocket_router;
+  bool accepted = true;
+  router.sse("/events", [&](fw::HttpContext&, std::shared_ptr<sse::SseSession> session)
+  {
+    accepted = session->send({"message", std::string(64, 'x'), "", std::nullopt});
+    session->close();
+  }, 32);
+  http::request<http::string_body> request{http::verb::get, "/events", 11};
+  request.keep_alive(false);
+  const auto response = test::round_trip(router, websocket_router, web_root.path, std::move(request));
+
+  EXPECT_FALSE(accepted);
+  EXPECT_TRUE(response.body().empty());
+}
+
+TEST(SseSessionTest, RequestBodiesForceConnectionCloseToPreventDesynchronization)
+{
+  test::TempWebRoot web_root;
+  fw::HttpRouter router;
+  fw::WebsocketRouter websocket_router;
+  router.sse("/events", [](fw::HttpContext&, std::shared_ptr<sse::SseSession> session)
+  {
+    session->send({"message", "ready", "", std::nullopt});
+    session->close();
+  });
+  http::request<http::string_body> request{http::verb::get, "/events", 11};
+  request.body() = "unconsumed-body";
+  request.prepare_payload();
+  request.keep_alive(true);
+  const auto response = test::round_trip(router, websocket_router, web_root.path, std::move(request));
+
+  EXPECT_FALSE(response.keep_alive());
+  EXPECT_EQ(response.body(), "event: message\ndata: ready\n\n");
+}
+
+TEST(SseSessionTest, PreRequestInterceptorCanDenyTheStreamBeforeItsHandlerRuns)
+{
+  class DenyInterceptor final : public fw::Interceptor
+  {
+  public:
+    fw::InterceptorResult handle_request(fw::HttpContext& ctx) override
+    {
+      ctx.set_status(http::status::forbidden);
+      ctx.set_body("denied");
+      return fw::InterceptorResult::Stop;
+    }
+  };
+
+  test::TempWebRoot web_root;
+  fw::HttpRouter router;
+  fw::WebsocketRouter websocket_router;
+  bool handler_called = false;
+  router.add_interceptor(std::make_shared<DenyInterceptor>());
+  router.sse("/events", [&](fw::HttpContext&, std::shared_ptr<sse::SseSession>)
+  {
+    handler_called = true;
+  });
+  http::request<http::string_body> request{http::verb::get, "/events", 11};
+  request.keep_alive(false);
+  const auto response = test::round_trip(router, websocket_router, web_root.path, std::move(request));
+
+  EXPECT_EQ(response.result(), http::status::forbidden);
+  EXPECT_EQ(response.body(), "denied");
+  EXPECT_FALSE(handler_called);
+}
+
+TEST(SseSessionTest, ThrowingHandlerReturnsOneSafeErrorResponse)
+{
+  test::TempWebRoot web_root;
+  fw::HttpRouter router;
+  fw::WebsocketRouter websocket_router;
+  router.sse("/events", [](fw::HttpContext&, std::shared_ptr<sse::SseSession>)
+  {
+    throw std::runtime_error("sensitive SSE failure");
+  });
+  http::request<http::string_body> request{http::verb::get, "/events", 11};
+  request.keep_alive(false);
+  const auto response = test::round_trip(router, websocket_router, web_root.path, std::move(request));
+
+  EXPECT_EQ(response.result(), http::status::internal_server_error);
+  EXPECT_EQ(response[http::field::content_type], "application/json");
+  EXPECT_EQ(response.body(), R"({"code":"INTERNAL_SERVER_ERROR","message":"Internal server error"})");
+  EXPECT_EQ(response.body().find("sensitive SSE failure"), std::string::npos);
+}
+
+TEST(SseSessionTest, ExplicitStartInsideRouteHandlerIsIdempotent)
+{
+  test::TempWebRoot web_root;
+  fw::HttpRouter router;
+  fw::WebsocketRouter websocket_router;
+  router.sse("/events", [](fw::HttpContext&, std::shared_ptr<sse::SseSession> session)
+  {
+    session->start();
+    session->send({"message", "ready", "", std::nullopt});
+    session->close();
+  });
+  http::request<http::string_body> request{http::verb::get, "/events", 11};
+  request.keep_alive(false);
+  const auto response = test::round_trip(router, websocket_router, web_root.path, std::move(request));
+
+  EXPECT_EQ(response.result(), http::status::ok);
+  EXPECT_EQ(response.body(), "event: message\ndata: ready\n\n");
+}
+
 TEST(SseClientTest, DeliversEventsFromAnAsyncEventStream)
 {
   net::io_context server_ioc;
@@ -206,3 +348,39 @@ TEST(SseClientTest, RejectsANonEventStreamResponse)
 
   EXPECT_EQ(closed_with, make_error_code(boost::system::errc::protocol_error));
 }
+
+TEST(SseClientTest, ClosesAnEventStreamThatExceedsTheConfiguredLimit)
+{
+  net::io_context server_ioc;
+  tcp::acceptor acceptor(server_ioc, {net::ip::address_v4::loopback(), 0});
+  const auto port = acceptor.local_endpoint().port();
+  std::thread server([&]
+  {
+    tcp::socket socket(server_ioc);
+    acceptor.accept(socket);
+    boost::beast::flat_buffer buffer;
+    http::request<http::empty_body> request;
+    http::read(socket, buffer, request);
+    const std::string wire =
+      "HTTP/1.1 200 OK\r\n"
+      "Content-Type: text/event-stream\r\n"
+      "Connection: close\r\n\r\n"
+      "data: " + std::string(64, 'x') + "\n\n";
+    boost::system::error_code ignored;
+    net::write(socket, net::buffer(wire), ignored);
+  });
+
+  net::io_context ioc;
+  auto stream = std::make_shared<client::SseClient>(ioc, 32);
+  int close_count = 0;
+  boost::system::error_code closed_with;
+  stream->connect(
+    "http://127.0.0.1:" + std::to_string(port) + "/events", {},
+    [](const sse::SseEvent&) { FAIL() << "oversized SSE response emitted an event"; },
+    [&](boost::system::error_code ec) { ++close_count; closed_with = ec; });
+  ioc.run();
+  server.join();
+
+  EXPECT_EQ(close_count, 1);
+  EXPECT_EQ(closed_with, net::error::message_size);
+}

From 08726815fde3fd21820df4660310d72d462418cf Mon Sep 17 00:00:00 2001
From: kekxv <caesar@kekxv.com>
Date: Sat, 22 Aug 2026 00:08:52 +0000
Subject: [PATCH 4/6] feat: add SSE example and harden callbacks

---
 README.md                           |  3 +
 doc/server-sent-events.md           |  9 +++
 example/BUILD.bazel                 |  8 +++
 example/SseDemo.hpp                 | 60 ++++++++++++++++++
 example/export_openapi_test.sh      |  2 +-
 example/homepage_docs_link_test.sh  |  2 +-
 example/main.cpp                    |  4 +-
 example/runtime_docs_toggle_test.sh |  2 +-
 example/sse_demo_test.sh            | 39 ++++++++++++
 example/typed_request_error_test.sh |  2 +-
 framework/client/sse_client.cpp     | 40 +++++++++++-
 framework/sse/sse_session.cpp       | 24 +++++++-
 framework/tests/sse_test.cpp        | 96 +++++++++++++++++++++++++++++
 13 files changed, 281 insertions(+), 10 deletions(-)
 create mode 100644 example/SseDemo.hpp
 create mode 100755 example/sse_demo_test.sh

diff --git a/README.md b/README.md
index 5e3075a..1b721de 100644
--- a/README.md
+++ b/README.md
@@ -459,6 +459,9 @@ any event-stream response headers are written.
 complete event as soon as it arrives. Arbitrarily split lines, CRLF/LF,
 multiline `data`, comments, `id`, and numeric `retry` fields are supported.
 
+The starter `example` exposes a running `/events` stream; after `bazel run //:app`,
+inspect it with `curl -N http://127.0.0.1:8080/events`.
+
 ```cpp
 auto events = std::make_shared<client::SseClient>(ioc); // 1 MiB event limit
 events->connect(
diff --git a/doc/server-sent-events.md b/doc/server-sent-events.md
index 62d77c7..f77c209 100644
--- a/doc/server-sent-events.md
+++ b/doc/server-sent-events.md
@@ -50,6 +50,11 @@ router.sse("/events", handler, 256 * 1024);
 心跳建议使用注释帧，例如 `send_comment("heartbeat")`；心跳周期和空闲连接管理由
 业务服务决定。
 
+仓库的 `example` 包含可直接运行的 `/events` 演示：连接后立即发送 `welcome` 和
+首个 `tick` 事件，之后每秒继续推送。运行 `bazel run //:app` 后可使用
+`curl -N http://127.0.0.1:8080/events` 查看输出。定时器在连接关闭时取消，避免为
+已断开的客户端保留后台任务。
+
 ## 客户端
 
 ```cpp
@@ -80,6 +85,10 @@ subscription->connect(
 构造函数的可选字节上限用于限制未完成行或单个事件；超限只会关闭当前订阅，并以
 `message_size` 完成关闭回调，不会终止进程。
 
+事件和关闭回调中的异常会被框架记录并收敛：事件回调异常会取消当前订阅并以
+`operation_aborted` 关闭，关闭回调异常不会穿透 Asio 的执行器。应用仍应自行处理
+业务错误，避免把可恢复错误作为控制流异常。
+
 ## 选择 SSE 还是 WebSocket
 
 - 服务器向客户端持续单向推送配置、实例或通知时，优先使用 SSE。
diff --git a/example/BUILD.bazel b/example/BUILD.bazel
index e4faa55..75fcac3 100644
--- a/example/BUILD.bazel
+++ b/example/BUILD.bazel
@@ -7,6 +7,7 @@ cc_binary(
         "HelloController.hpp",
         "HelloStreamController.hpp",
         "HelloWsController.hpp",
+        "SseDemo.hpp",
         "TypedHelloController.hpp",
         "main.cpp",
     ],
@@ -42,3 +43,10 @@ sh_test(
     data = [":app"],
     tags = ["exclusive"],
 )
+
+sh_test(
+    name = "sse_demo_test",
+    srcs = ["sse_demo_test.sh"],
+    data = [":app"],
+    tags = ["exclusive"],
+)
diff --git a/example/SseDemo.hpp b/example/SseDemo.hpp
new file mode 100644
index 0000000..1d3cb42
--- /dev/null
+++ b/example/SseDemo.hpp
@@ -0,0 +1,60 @@
+#ifndef KHTTPD_EXAMPLE_SSE_DEMO_HPP
+#define KHTTPD_EXAMPLE_SSE_DEMO_HPP
+
+#include <boost/asio/steady_timer.hpp>
+#include <chrono>
+#include <memory>
+#include <optional>
+
+#include <fmt/format.h>
+
+#include "io_context_pool.hpp"
+#include "router/http_router.hpp"
+#include "sse/sse_session.hpp"
+
+class SseDemo
+{
+public:
+  static void register_routes(khttpd::framework::HttpRouter& router)
+  {
+    router.sse("/events", [](khttpd::framework::HttpContext&,
+                              std::shared_ptr<khttpd::framework::sse::SseSession> session)
+    {
+      auto state = std::make_shared<State>(khttpd::framework::IoContextPool::instance().get_io_context());
+      session->on_close([timer = state->timer](boost::system::error_code)
+      {
+        timer->cancel();
+      });
+      session->send({"welcome", R"({"message":"Connected to the khttpd SSE demo"})", "1", 1000});
+      publish_tick(state, session);
+    });
+  }
+
+private:
+  struct State
+  {
+    explicit State(boost::asio::io_context& ioc)
+      : timer(std::make_shared<boost::asio::steady_timer>(ioc)) {}
+
+    std::shared_ptr<boost::asio::steady_timer> timer;
+    unsigned int sequence = 1;
+  };
+
+  static void publish_tick(const std::shared_ptr<State>& state,
+                           const std::shared_ptr<khttpd::framework::sse::SseSession>& session)
+  {
+    if (!session->is_open()) return;
+    if (!session->send({"tick", fmt::format(R"({{"sequence":{}}})", state->sequence++), "", std::nullopt}))
+    {
+      session->close();
+      return;
+    }
+    state->timer->expires_after(std::chrono::seconds(1));
+    state->timer->async_wait([state, session](boost::system::error_code ec)
+    {
+      if (!ec) publish_tick(state, session);
+    });
+  }
+};
+
+#endif
diff --git a/example/export_openapi_test.sh b/example/export_openapi_test.sh
index 232311b..2b1ef27 100755
--- a/example/export_openapi_test.sh
+++ b/example/export_openapi_test.sh
@@ -1,7 +1,7 @@
 #!/usr/bin/env bash
 set -euo pipefail
 
-app="${TEST_SRCDIR}/${TEST_WORKSPACE}/example/app"
+app="${TEST_SRCDIR}/${TEST_WORKSPACE}/app"
 ready="${TEST_TMPDIR}/port-ready"
 output="${TEST_TMPDIR}/example-openapi.json"
 
diff --git a/example/homepage_docs_link_test.sh b/example/homepage_docs_link_test.sh
index ca4246a..178b1fb 100755
--- a/example/homepage_docs_link_test.sh
+++ b/example/homepage_docs_link_test.sh
@@ -1,7 +1,7 @@
 #!/usr/bin/env bash
 set -euo pipefail
 
-app="${TEST_SRCDIR}/${TEST_WORKSPACE}/example/app"
+app="${TEST_SRCDIR}/${TEST_WORKSPACE}/app"
 log="${TEST_TMPDIR}/app.log"
 
 "${app}" --enable-openapi-docs >"${log}" 2>&1 &
diff --git a/example/main.cpp b/example/main.cpp
index d497282..9544fc0 100644
--- a/example/main.cpp
+++ b/example/main.cpp
@@ -16,6 +16,7 @@
 #include "HelloController.hpp"
 #include "HelloStreamController.hpp"
 #include "HelloWsController.hpp"
+#include "SseDemo.hpp"
 #include "TypedHelloController.hpp"
 
 namespace net = boost::asio;
@@ -44,13 +45,14 @@ namespace
   HelloStreamController::create()->register_routes(http_router)->register_routes(ws_router);
   HelloWsController::create()->register_routes(http_router)->register_routes(ws_router);
   TypedHelloController::create()->register_routes(http_router)->register_routes(ws_router);
+  SseDemo::register_routes(http_router);
 
   http_router.get("/", [](khttpd::framework::HttpContext& ctx)
   {
     ctx.set_status(beast::http::status::ok);
     ctx.set_content_type("text/html");
     ctx.set_body(
-      R"(<h1>Hello from khttpd!</h1><p><a href="/docs">API documentation</a></p><p>Try <a href="/hello?name=World">/hello?name=World</a> or <a href="/info">/info</a></p><p>Dynamic paths: <a href="/users/123">/users/123</a>, <a href="/users/profile">/users/profile</a>, <a href="/items/book/id/456">/items/book/id/456</a>, <a href="/files/a/b/c.txt">/files/a/b/c.txt</a></p><p>POST examples: /api/json, /api/form, /api/upload</p><p>Or connect to <a href="/ws">WebSocket</a></p><p>Or connect to <a href="/chat">WebSocket Chat</a></p>)");
+      R"(<h1>Hello from khttpd!</h1><p><a href="/docs">API documentation</a></p><p>Try <a href="/hello?name=World">/hello?name=World</a> or <a href="/info">/info</a></p><p>Dynamic paths: <a href="/users/123">/users/123</a>, <a href="/users/profile">/users/profile</a>, <a href="/items/book/id/456">/items/book/id/456</a>, <a href="/files/a/b/c.txt">/files/a/b/c.txt</a></p><p>POST examples: /api/json, /api/form, /api/upload</p><p>Open <a href="/events">the SSE event stream</a> with an SSE client or <code>curl -N</code>.</p><p>Or connect to <a href="/ws">WebSocket</a></p><p>Or connect to <a href="/chat">WebSocket Chat</a></p>)");
   }, {"Example service home", "Links to the sample HTTP, streaming, WebSocket, and API documentation endpoints."});
 
   http_router.get("/hello", [](khttpd::framework::HttpContext& ctx)
diff --git a/example/runtime_docs_toggle_test.sh b/example/runtime_docs_toggle_test.sh
index d2743b3..196cb9d 100755
--- a/example/runtime_docs_toggle_test.sh
+++ b/example/runtime_docs_toggle_test.sh
@@ -1,7 +1,7 @@
 #!/usr/bin/env bash
 set -euo pipefail
 
-app="${TEST_SRCDIR}/${TEST_WORKSPACE}/example/app"
+app="${TEST_SRCDIR}/${TEST_WORKSPACE}/app"
 log="${TEST_TMPDIR}/app.log"
 
 "${app}" --disable-openapi-docs >"${log}" 2>&1 &
diff --git a/example/sse_demo_test.sh b/example/sse_demo_test.sh
new file mode 100755
index 0000000..7a7bd11
--- /dev/null
+++ b/example/sse_demo_test.sh
@@ -0,0 +1,39 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+app="${TEST_SRCDIR}/${TEST_WORKSPACE}/app"
+log="${TEST_TMPDIR}/app.log"
+
+"${app}" --enable-openapi-docs >"${log}" 2>&1 &
+app_pid=$!
+trap 'kill "${app_pid}" 2>/dev/null || true; wait "${app_pid}" 2>/dev/null || true' EXIT
+
+python3 - <<'PY'
+import socket
+import time
+
+for _ in range(100):
+    try:
+        with socket.create_connection(("127.0.0.1", 8080), timeout=0.1):
+            break
+    except OSError:
+        time.sleep(0.02)
+else:
+    raise SystemExit("example server did not listen on port 8080")
+
+with socket.create_connection(("127.0.0.1", 8080), timeout=1) as sock:
+    sock.settimeout(2)
+    sock.sendall(b"GET /events HTTP/1.1\r\nHost: localhost\r\nAccept: text/event-stream\r\n\r\n")
+    response = bytearray()
+    while b"event: welcome" not in response or b"event: tick" not in response:
+        chunk = sock.recv(4096)
+        if not chunk:
+            raise AssertionError("SSE connection closed before the demo events arrived")
+        response.extend(chunk)
+
+wire = bytes(response)
+assert wire.startswith(b"HTTP/1.1 200"), wire
+assert b"Content-Type: text/event-stream" in wire, wire
+assert b"event: welcome" in wire and b"Connected to the khttpd SSE demo" in wire, wire
+assert b"event: tick" in wire and b'"sequence":1' in wire, wire
+PY
diff --git a/example/typed_request_error_test.sh b/example/typed_request_error_test.sh
index 9232f07..774e5da 100755
--- a/example/typed_request_error_test.sh
+++ b/example/typed_request_error_test.sh
@@ -1,7 +1,7 @@
 #!/usr/bin/env bash
 set -euo pipefail
 
-app="${TEST_SRCDIR}/${TEST_WORKSPACE}/example/app"
+app="${TEST_SRCDIR}/${TEST_WORKSPACE}/app"
 log="${TEST_TMPDIR}/app.log"
 
 "${app}" --disable-openapi-docs >"${log}" 2>&1 &
diff --git a/framework/client/sse_client.cpp b/framework/client/sse_client.cpp
index 8e6f705..6273931 100644
--- a/framework/client/sse_client.cpp
+++ b/framework/client/sse_client.cpp
@@ -6,10 +6,13 @@
 #include <algorithm>
 #include <array>
 #include <cctype>
+#include <exception>
 #include <mutex>
 #include <stdexcept>
 #include <utility>
 
+#include <spdlog/spdlog.h>
+
 #include "client/http_client_stream.hpp"
 #include "sse/sse_parser.hpp"
 
@@ -71,7 +74,7 @@ namespace khttpd::framework::client
       }
       if (duplicate)
       {
-        if (on_close) on_close(make_error_code(boost::system::errc::operation_in_progress));
+        invoke_close_handler(on_close, make_error_code(boost::system::errc::operation_in_progress));
         return;
       }
 
@@ -138,7 +141,21 @@ namespace khttpd::framework::client
         if (terminal) return;
         handler = event_handler;
       }
-      if (handler) handler(event);
+      if (!handler) return;
+      try
+      {
+        handler(event);
+      }
+      catch (const std::exception& error)
+      {
+        spdlog::error("SSE event callback failed: {}", error.what());
+        finish(net::error::operation_aborted);
+      }
+      catch (...)
+      {
+        spdlog::error("SSE event callback failed with a non-standard exception");
+        finish(net::error::operation_aborted);
+      }
     }
 
     bool is_terminal()
@@ -158,7 +175,7 @@ namespace khttpd::framework::client
         event_handler = {};
       }
       if (ec) stream->cancel();
-      if (handler) handler(ec);
+      invoke_close_handler(handler, ec);
     }
 
     void cancel()
@@ -177,6 +194,23 @@ namespace khttpd::framework::client
       }
       stream->cancel();
     }
+
+    static void invoke_close_handler(const CloseHandler& handler, const boost::system::error_code ec)
+    {
+      if (!handler) return;
+      try
+      {
+        handler(ec);
+      }
+      catch (const std::exception& error)
+      {
+        spdlog::error("SSE close callback failed: {}", error.what());
+      }
+      catch (...)
+      {
+        spdlog::error("SSE close callback failed with a non-standard exception");
+      }
+    }
   };
 
   SseClient::SseClient(net::io_context& ioc, const std::size_t max_event_bytes)
diff --git a/framework/sse/sse_session.cpp b/framework/sse/sse_session.cpp
index 0240dd3..c76663d 100644
--- a/framework/sse/sse_session.cpp
+++ b/framework/sse/sse_session.cpp
@@ -2,9 +2,12 @@
 
 #include <boost/asio/buffer.hpp>
 #include <deque>
+#include <exception>
 #include <mutex>
 #include <utility>
 
+#include <spdlog/spdlog.h>
+
 namespace khttpd::framework::sse
 {
   struct SseSession::Impl : std::enable_shared_from_this<Impl>
@@ -116,7 +119,24 @@ namespace khttpd::framework::sse
         queue.clear();
         handler = std::move(close_handler);
       }
-      if (handler) handler(ec);
+      invoke_close_handler(handler, ec);
+    }
+
+    static void invoke_close_handler(const CloseHandler& handler, const boost::system::error_code ec)
+    {
+      if (!handler) return;
+      try
+      {
+        handler(ec);
+      }
+      catch (const std::exception& error)
+      {
+        spdlog::error("SSE session close callback failed: {}", error.what());
+      }
+      catch (...)
+      {
+        spdlog::error("SSE session close callback failed with a non-standard exception");
+      }
     }
   };
 
@@ -138,7 +158,7 @@ namespace khttpd::framework::sse
       error = impl_->close_error;
       if (!already_closed) impl_->close_handler = std::move(handler);
     }
-    if (already_closed && handler) handler(error);
+    if (already_closed) Impl::invoke_close_handler(handler, error);
   }
   bool SseSession::is_open() const { std::lock_guard lock(impl_->mutex); return !impl_->closing && !impl_->closed; }
 }
diff --git a/framework/tests/sse_test.cpp b/framework/tests/sse_test.cpp
index 4458b4b..238c727 100644
--- a/framework/tests/sse_test.cpp
+++ b/framework/tests/sse_test.cpp
@@ -20,6 +20,24 @@ namespace net = boost::asio;
 namespace client = khttpd::framework::client;
 using tcp = net::ip::tcp;
 
+namespace
+{
+  class ImmediateResponseStream final : public fw::HttpResponseStream
+  {
+  public:
+    bool finished = false;
+
+    void async_start(ResponseHead, Callback callback) override { callback({}); }
+    void async_write_some(net::const_buffer, Callback callback) override { callback({}); }
+    void async_finish(Callback callback) override
+    {
+      finished = true;
+      callback({});
+    }
+    void cancel() override {}
+  };
+}
+
 TEST(SseParserTest, PreservesAFragmentedMultilineEvent)
 {
   sse::SseParser parser;
@@ -226,6 +244,17 @@ TEST(SseSessionTest, ExplicitStartInsideRouteHandlerIsIdempotent)
   EXPECT_EQ(response.body(), "event: message\ndata: ready\n\n");
 }
 
+TEST(SseSessionTest, ContainsExceptionsThrownByCloseCallback)
+{
+  auto response = std::make_shared<ImmediateResponseStream>();
+  sse::SseSession session(response, 11, false);
+  session.on_close([](boost::system::error_code) { throw std::runtime_error("close callback failure"); });
+  session.close();
+
+  EXPECT_NO_THROW(session.start());
+  EXPECT_TRUE(response->finished);
+}
+
 TEST(SseClientTest, DeliversEventsFromAnAsyncEventStream)
 {
   net::io_context server_ioc;
@@ -384,3 +413,70 @@ TEST(SseClientTest, ClosesAnEventStreamThatExceedsTheConfiguredLimit)
   EXPECT_EQ(close_count, 1);
   EXPECT_EQ(closed_with, net::error::message_size);
 }
+
+TEST(SseClientTest, ContainsExceptionsThrownByUserCallbacks)
+{
+  net::io_context server_ioc;
+  tcp::acceptor acceptor(server_ioc, {net::ip::address_v4::loopback(), 0});
+  const auto port = acceptor.local_endpoint().port();
+  std::thread server([&]
+  {
+    tcp::socket socket(server_ioc);
+    acceptor.accept(socket);
+    boost::beast::flat_buffer buffer;
+    http::request<http::empty_body> request;
+    http::read(socket, buffer, request);
+    const std::string wire =
+      "HTTP/1.1 200 OK\r\n"
+      "Content-Type: text/event-stream\r\n"
+      "Connection: close\r\n\r\n"
+      "data: ready\n\n";
+    boost::system::error_code ignored;
+    net::write(socket, net::buffer(wire), ignored);
+  });
+
+  net::io_context ioc;
+  auto stream = std::make_shared<client::SseClient>(ioc);
+  int close_count = 0;
+  boost::system::error_code closed_with;
+  stream->connect(
+    "http://127.0.0.1:" + std::to_string(port) + "/events", {},
+    [](const sse::SseEvent&) { throw std::runtime_error("application callback failure"); },
+    [&](boost::system::error_code ec) { ++close_count; closed_with = ec; });
+
+  EXPECT_NO_THROW(ioc.run());
+  server.join();
+
+  EXPECT_EQ(close_count, 1);
+  EXPECT_EQ(closed_with, net::error::operation_aborted);
+}
+
+TEST(SseClientTest, ContainsExceptionsThrownByCloseCallback)
+{
+  net::io_context server_ioc;
+  tcp::acceptor acceptor(server_ioc, {net::ip::address_v4::loopback(), 0});
+  const auto port = acceptor.local_endpoint().port();
+  std::thread server([&]
+  {
+    tcp::socket socket(server_ioc);
+    acceptor.accept(socket);
+    boost::beast::flat_buffer buffer;
+    http::request<http::empty_body> request;
+    http::read(socket, buffer, request);
+    http::response<http::empty_body> response{http::status::ok, 11};
+    response.set(http::field::content_type, "application/json");
+    response.content_length(0);
+    response.keep_alive(false);
+    http::write(socket, response);
+  });
+
+  net::io_context ioc;
+  auto stream = std::make_shared<client::SseClient>(ioc);
+  stream->connect(
+    "http://127.0.0.1:" + std::to_string(port) + "/events", {},
+    [](const sse::SseEvent&) {},
+    [](boost::system::error_code) { throw std::runtime_error("close callback failure"); });
+
+  EXPECT_NO_THROW(ioc.run());
+  server.join();
+}

From 03719071f1fb8373f550ec8cd34b123cfbdc3cbf Mon Sep 17 00:00:00 2001
From: kekxv <caesar@kekxv.com>
Date: Sat, 22 Aug 2026 00:52:29 +0000
Subject: [PATCH 5/6] feat: add visual SSE example page

---
 README.md                          |   5 +-
 doc/server-sent-events.md          |   7 +-
 example/SseDemo.hpp                | 135 +++++++++++++++++++++++++++++
 example/homepage_docs_link_test.sh |   8 ++
 example/main.cpp                   |   2 +-
 5 files changed, 151 insertions(+), 6 deletions(-)

diff --git a/README.md b/README.md
index 1b721de..059f581 100644
--- a/README.md
+++ b/README.md
@@ -459,8 +459,9 @@ any event-stream response headers are written.
 complete event as soon as it arrives. Arbitrarily split lines, CRLF/LF,
 multiline `data`, comments, `id`, and numeric `retry` fields are supported.
 
-The starter `example` exposes a running `/events` stream; after `bazel run //:app`,
-inspect it with `curl -N http://127.0.0.1:8080/events`.
+The starter `example` exposes a visual `/events-demo` page and the raw `/events`
+stream. After `bazel run //:app`, open the page in a browser or inspect the stream
+with `curl -N http://127.0.0.1:8080/events`.
 
 ```cpp
 auto events = std::make_shared<client::SseClient>(ioc); // 1 MiB event limit
diff --git a/doc/server-sent-events.md b/doc/server-sent-events.md
index f77c209..82049d9 100644
--- a/doc/server-sent-events.md
+++ b/doc/server-sent-events.md
@@ -51,9 +51,10 @@ router.sse("/events", handler, 256 * 1024);
 业务服务决定。
 
 仓库的 `example` 包含可直接运行的 `/events` 演示：连接后立即发送 `welcome` 和
-首个 `tick` 事件，之后每秒继续推送。运行 `bazel run //:app` 后可使用
-`curl -N http://127.0.0.1:8080/events` 查看输出。定时器在连接关闭时取消，避免为
-已断开的客户端保留后台任务。
+首个 `tick` 事件，之后每秒继续推送。运行 `bazel run //:app` 后打开
+`http://127.0.0.1:8080/events-demo` 可查看浏览器实时界面，也可使用
+`curl -N http://127.0.0.1:8080/events` 查看原始事件流。定时器在连接关闭时取消，
+避免为已断开的客户端保留后台任务。
 
 ## 客户端
 
diff --git a/example/SseDemo.hpp b/example/SseDemo.hpp
index 1d3cb42..4fbba29 100644
--- a/example/SseDemo.hpp
+++ b/example/SseDemo.hpp
@@ -17,6 +17,13 @@ class SseDemo
 public:
   static void register_routes(khttpd::framework::HttpRouter& router)
   {
+    router.get("/events-demo", [](khttpd::framework::HttpContext& ctx)
+    {
+      ctx.set_status(boost::beast::http::status::ok);
+      ctx.set_content_type("text/html; charset=utf-8");
+      ctx.set_body(demo_page());
+    }, {"SSE browser demo", "Displays connection state and events received from the example SSE stream."});
+
     router.sse("/events", [](khttpd::framework::HttpContext&,
                               std::shared_ptr<khttpd::framework::sse::SseSession> session)
     {
@@ -31,6 +38,134 @@ class SseDemo
   }
 
 private:
+  static const char* demo_page()
+  {
+    return R"HTML(<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1">
+  <title>khttpd SSE demo</title>
+  <style>
+    :root { color-scheme: light; font-family: Inter, ui-sans-serif, system-ui, sans-serif; }
+    * { box-sizing: border-box; }
+    body { margin: 0; background: #f5f7f8; color: #182026; }
+    main { width: min(760px, calc(100% - 32px)); margin: 48px auto; }
+    header { display: flex; align-items: flex-start; justify-content: space-between; gap: 24px; margin-bottom: 24px; }
+    h1 { margin: 0 0 8px; font-size: 28px; letter-spacing: 0; }
+    p { margin: 0; color: #5b6670; line-height: 1.6; }
+    a { color: #1769aa; text-decoration: none; }
+    a:hover { text-decoration: underline; }
+    .status { display: inline-flex; align-items: center; gap: 8px; min-width: 132px; padding: 8px 12px;
+      border: 1px solid #d8dee3; border-radius: 6px; background: #fff; font-size: 14px; }
+    .status-dot { width: 9px; height: 9px; border-radius: 50%; background: #9aa4ad; }
+    .status.connected .status-dot { background: #16834b; }
+    .status.disconnected .status-dot { background: #c43d3d; }
+    .toolbar { display: flex; align-items: center; justify-content: space-between; gap: 16px; padding: 12px 0;
+      border-top: 1px solid #d8dee3; border-bottom: 1px solid #d8dee3; }
+    .actions { display: flex; gap: 8px; }
+    button { min-height: 36px; padding: 0 14px; border: 1px solid #bdc6cd; border-radius: 6px;
+      background: #fff; color: #182026; font: inherit; cursor: pointer; }
+    button:hover { background: #eef2f4; }
+    code { color: #34404a; }
+    #event-list { min-height: 260px; margin: 0; padding: 0; list-style: none; background: #fff; }
+    .event { display: grid; grid-template-columns: 90px 1fr auto; gap: 16px; align-items: start;
+      padding: 14px 16px; border-bottom: 1px solid #e5e9ec; }
+    .event-type { color: #1769aa; font-weight: 650; }
+    .event-data { margin: 0; overflow-wrap: anywhere; white-space: pre-wrap; font: 13px/1.5 ui-monospace, monospace; }
+    .event-time { color: #77828c; font-size: 12px; }
+    .empty { padding: 72px 16px; color: #77828c; text-align: center; }
+    @media (max-width: 560px) {
+      main { margin: 24px auto; }
+      header { flex-direction: column; }
+      .event { grid-template-columns: 72px 1fr; }
+      .event-time { grid-column: 2; }
+      .toolbar { align-items: flex-start; flex-direction: column; }
+    }
+  </style>
+</head>
+<body>
+  <main>
+    <header>
+      <div>
+        <h1>Server-Sent Events</h1>
+        <p>Live events from <code>/events</code>. <a href="/">Back to examples</a></p>
+      </div>
+      <div id="connection-status" class="status" role="status" aria-live="polite">
+        <span class="status-dot" aria-hidden="true"></span><span id="status-label">Connecting</span>
+      </div>
+    </header>
+    <div class="toolbar">
+      <span id="event-count">0 events</span>
+      <div class="actions">
+        <button id="reconnect" type="button">Reconnect</button>
+        <button id="clear" type="button">Clear</button>
+      </div>
+    </div>
+    <ul id="event-list" aria-live="polite"><li class="empty">Waiting for events...</li></ul>
+  </main>
+  <script>
+    const statusNode = document.getElementById('connection-status');
+    const statusLabel = document.getElementById('status-label');
+    const eventList = document.getElementById('event-list');
+    const eventCount = document.getElementById('event-count');
+    let source;
+    let count = 0;
+
+    function setStatus(label, state) {
+      statusLabel.textContent = label;
+      statusNode.className = 'status ' + state;
+    }
+
+    function appendEvent(event) {
+      const empty = eventList.querySelector('.empty');
+      if (empty) empty.remove();
+      const row = document.createElement('li');
+      row.className = 'event';
+      const type = document.createElement('span');
+      type.className = 'event-type';
+      type.textContent = event.type;
+      const data = document.createElement('pre');
+      data.className = 'event-data';
+      data.textContent = event.data;
+      const time = document.createElement('time');
+      time.className = 'event-time';
+      time.textContent = new Date().toLocaleTimeString();
+      row.append(type, data, time);
+      eventList.prepend(row);
+      while (eventList.children.length > 100) eventList.lastElementChild.remove();
+      count += 1;
+      eventCount.textContent = count + (count === 1 ? ' event' : ' events');
+    }
+
+    function connect() {
+      if (source) source.close();
+      setStatus('Connecting', '');
+      source = new EventSource('/events');
+      source.onopen = () => setStatus('Connected', 'connected');
+      source.onerror = () => setStatus('Reconnecting', 'disconnected');
+      source.addEventListener('welcome', appendEvent);
+      source.addEventListener('tick', appendEvent);
+      source.onmessage = appendEvent;
+    }
+
+    document.getElementById('reconnect').addEventListener('click', connect);
+    document.getElementById('clear').addEventListener('click', () => {
+      eventList.replaceChildren();
+      count = 0;
+      eventCount.textContent = '0 events';
+      const empty = document.createElement('li');
+      empty.className = 'empty';
+      empty.textContent = 'Waiting for events...';
+      eventList.append(empty);
+    });
+    window.addEventListener('beforeunload', () => source && source.close());
+    connect();
+  </script>
+</body>
+</html>)HTML";
+  }
+
   struct State
   {
     explicit State(boost::asio::io_context& ioc)
diff --git a/example/homepage_docs_link_test.sh b/example/homepage_docs_link_test.sh
index 178b1fb..bdf7fbc 100755
--- a/example/homepage_docs_link_test.sh
+++ b/example/homepage_docs_link_test.sh
@@ -35,5 +35,13 @@ def request(path):
 homepage = request("/")
 assert homepage.startswith("HTTP/1.1 200")
 assert 'href="/docs"' in homepage, "homepage must link to API documentation"
+assert 'href="/events-demo"' in homepage, "homepage must link to the visual SSE demo"
+
+sse_demo = request("/events-demo")
+assert sse_demo.startswith("HTTP/1.1 200")
+assert "new EventSource('/events')" in sse_demo, "SSE demo must subscribe to the event stream"
+assert 'id="connection-status"' in sse_demo, "SSE demo must show connection state"
+assert 'id="event-list"' in sse_demo, "SSE demo must render received events"
+assert "eventList.children.length > 100" in sse_demo, "SSE demo must bound retained event rows"
 assert request("/docs").startswith("HTTP/1.1 200")
 PY
diff --git a/example/main.cpp b/example/main.cpp
index 9544fc0..721f146 100644
--- a/example/main.cpp
+++ b/example/main.cpp
@@ -52,7 +52,7 @@ namespace
     ctx.set_status(beast::http::status::ok);
     ctx.set_content_type("text/html");
     ctx.set_body(
-      R"(<h1>Hello from khttpd!</h1><p><a href="/docs">API documentation</a></p><p>Try <a href="/hello?name=World">/hello?name=World</a> or <a href="/info">/info</a></p><p>Dynamic paths: <a href="/users/123">/users/123</a>, <a href="/users/profile">/users/profile</a>, <a href="/items/book/id/456">/items/book/id/456</a>, <a href="/files/a/b/c.txt">/files/a/b/c.txt</a></p><p>POST examples: /api/json, /api/form, /api/upload</p><p>Open <a href="/events">the SSE event stream</a> with an SSE client or <code>curl -N</code>.</p><p>Or connect to <a href="/ws">WebSocket</a></p><p>Or connect to <a href="/chat">WebSocket Chat</a></p>)");
+      R"(<h1>Hello from khttpd!</h1><p><a href="/docs">API documentation</a></p><p>Try <a href="/hello?name=World">/hello?name=World</a> or <a href="/info">/info</a></p><p>Dynamic paths: <a href="/users/123">/users/123</a>, <a href="/users/profile">/users/profile</a>, <a href="/items/book/id/456">/items/book/id/456</a>, <a href="/files/a/b/c.txt">/files/a/b/c.txt</a></p><p>POST examples: /api/json, /api/form, /api/upload</p><p>Watch the <a href="/events-demo">interactive SSE demo</a>, or inspect the raw <code>/events</code> stream with <code>curl -N</code>.</p><p>Or connect to <a href="/ws">WebSocket</a></p><p>Or connect to <a href="/chat">WebSocket Chat</a></p>)");
   }, {"Example service home", "Links to the sample HTTP, streaming, WebSocket, and API documentation endpoints."});
 
   http_router.get("/hello", [](khttpd::framework::HttpContext& ctx)

From a72cd5214fef006e4828545aa94b20dabeb47be5 Mon Sep 17 00:00:00 2001
From: kekxv <caesar@kekxv.com>
Date: Sat, 22 Aug 2026 01:29:12 +0000
Subject: [PATCH 6/6] fix: detect passive SSE disconnects

---
 README.md                                  |  4 +
 doc/server-sent-events.md                  |  6 +-
 framework/context/http_response_stream.hpp |  4 +
 framework/session/http_session.cpp         | 39 ++++++++++
 framework/session/http_session.hpp         |  7 ++
 framework/sse/sse_session.cpp              | 26 ++++++-
 framework/tests/sse_test.cpp               | 87 ++++++++++++++++++++++
 7 files changed, 169 insertions(+), 4 deletions(-)

diff --git a/README.md b/README.md
index 059f581..fc1375c 100644
--- a/README.md
+++ b/README.md
@@ -455,6 +455,10 @@ SSE routes run the ordinary pre-request interceptor chain before their handler;
 an authentication or authorization interceptor can reject the request before
 any event-stream response headers are written.
 
+The server also monitors the connection's read side while an SSE response is
+open. A passive client disconnect completes the session and invokes `on_close`
+even when the application has no event or heartbeat waiting to be written.
+
 `SseClient` uses the same fixed-buffer streaming transport and delivers each
 complete event as soon as it arrives. Arbitrarily split lines, CRLF/LF,
 multiline `data`, comments, `id`, and numeric `retry` fields are supported.
diff --git a/doc/server-sent-events.md b/doc/server-sent-events.md
index 82049d9..f1470d6 100644
--- a/doc/server-sent-events.md
+++ b/doc/server-sent-events.md
@@ -47,8 +47,10 @@ router.sse("/events", handler, 256 * 1024);
 `std::shared_ptr<SseSession>`。`close()` 会等待已排队事件写完再发送结束块；
 `cancel()` 会立即中断连接。两者最终都只触发一次 `on_close`。
 
-心跳建议使用注释帧，例如 `send_comment("heartbeat")`；心跳周期和空闲连接管理由
-业务服务决定。
+khttpd 会在响应流期间独立监听 TCP 读侧。客户端主动关闭连接时，即使服务端没有
+后续事件写入，也会立即结束 `SseSession` 并触发一次 `on_close`，因此资源回收不依赖
+业务心跳。心跳仍建议使用注释帧，例如 `send_comment("heartbeat")`，用于避免代理、
+负载均衡器或 NAT 将空闲连接回收；心跳周期由业务服务决定。
 
 仓库的 `example` 包含可直接运行的 `/events` 演示：连接后立即发送 `welcome` 和
 首个 `tick` 事件，之后每秒继续推送。运行 `bazel run //:app` 后打开
diff --git a/framework/context/http_response_stream.hpp b/framework/context/http_response_stream.hpp
index 7406709..82c654d 100644
--- a/framework/context/http_response_stream.hpp
+++ b/framework/context/http_response_stream.hpp
@@ -16,6 +16,10 @@ namespace khttpd::framework
     virtual void async_start(ResponseHead head, Callback callback) = 0;
     virtual void async_write_some(boost::asio::const_buffer buffer, Callback callback) = 0;
     virtual void async_finish(Callback callback) = 0;
+    // Completes when the peer closes its read side or sends unexpected data while a response stream is active.
+    // The default no-op disables disconnect observation for custom response-stream implementations.
+    virtual void async_wait_disconnect(Callback callback) { (void)callback; }
+    virtual void cancel_disconnect_wait() {}
     // Stops an inbound request-body read while preserving this response stream.
     virtual void cancel_request_body() { cancel(); }
     virtual void cancel() = 0;
diff --git a/framework/session/http_session.cpp b/framework/session/http_session.cpp
index 411127c..848c850 100644
--- a/framework/session/http_session.cpp
+++ b/framework/session/http_session.cpp
@@ -1,6 +1,7 @@
 #include "http_session.hpp"
 
 #include "context/http_context.hpp"
+#include <boost/asio/bind_cancellation_slot.hpp>
 #include <fmt/core.h>
 #include <spdlog/spdlog.h>
 #include <sstream>
@@ -55,6 +56,10 @@ class HttpSession::ResponseStreamImpl final : public HttpResponseStream
   { if (session_) session_->write_stream_response(b, std::move(cb)); else cb(net::error::operation_aborted); }
   void async_finish(Callback cb) override
   { if (session_) session_->finish_stream_response(std::move(cb)); else cb(net::error::operation_aborted); }
+  void async_wait_disconnect(Callback cb) override
+  { if (session_) session_->wait_stream_disconnect(std::move(cb)); }
+  void cancel_disconnect_wait() override
+  { if (session_) session_->cancel_stream_disconnect_wait(); }
   void cancel_request_body() override { if (session_) session_->cancel_stream_body(); }
   void cancel() override { if (session_) session_->cancel_session(); }
 };
@@ -286,6 +291,7 @@ void HttpSession::start_stream_response(HttpResponseStream::ResponseHead head, H
   auto self = shared_from_this();
   net::post(stream_.get_executor(), [self, head = std::move(head), callback = std::move(callback)]() mutable
   {
+    self->disconnect_wait_cancelled_ = false;
     self->streaming_response_ = {};
     self->streaming_response_.result(head.result()); self->streaming_response_.version(head.version());
     self->streaming_response_.keep_alive(head.keep_alive());
@@ -341,6 +347,39 @@ void HttpSession::finish_stream_response(HttpResponseStream::Callback callback)
   });
 }
 
+void HttpSession::wait_stream_disconnect(HttpResponseStream::Callback callback)
+{
+  auto self = shared_from_this();
+  net::post(stream_.get_executor(), [self, callback = std::move(callback)]() mutable
+  {
+    // A request body is still readable on this connection and must not be mistaken for a peer disconnect.
+    if (self->disconnect_wait_cancelled_)
+      return callback(net::error::operation_aborted);
+    if ((self->request_parser_ && !self->request_parser_->is_done()) || self->disconnect_wait_active_) return;
+    self->disconnect_wait_active_ = true;
+    self->stream_.socket().async_receive(net::buffer(self->disconnect_probe_),
+      net::bind_cancellation_slot(self->disconnect_wait_cancellation_.slot(),
+        [self, callback = std::move(callback)](beast::error_code ec, const std::size_t bytes) mutable
+        {
+          self->disconnect_wait_active_ = false;
+          if (!ec)
+            ec = bytes == 0 ? net::error::eof : make_error_code(boost::system::errc::protocol_error);
+          callback(ec);
+        }));
+  });
+}
+
+void HttpSession::cancel_stream_disconnect_wait()
+{
+  auto self = shared_from_this();
+  net::dispatch(stream_.get_executor(), [self]
+  {
+    self->disconnect_wait_cancelled_ = true;
+    if (self->disconnect_wait_active_)
+      self->disconnect_wait_cancellation_.emit(net::cancellation_type::terminal);
+  });
+}
+
 void HttpSession::on_read(const beast::error_code& ec, std::size_t bytes_transferred)
 {
   boost::ignore_unused(bytes_transferred);
diff --git a/framework/session/http_session.hpp b/framework/session/http_session.hpp
index 1abc502..b8db901 100644
--- a/framework/session/http_session.hpp
+++ b/framework/session/http_session.hpp
@@ -4,6 +4,7 @@
 #include <boost/filesystem.hpp>
 #include <boost/beast.hpp>
 #include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/cancellation_signal.hpp>
 #include <memory>
 #include <queue>
 #include <sstream>
@@ -63,6 +64,10 @@ namespace khttpd::framework
     bool stream_completed_ = false;
     std::optional<tcp::endpoint> peer_endpoint_;
     bool request_body_cancelled_ = false;
+    std::array<char, 1> disconnect_probe_{};
+    net::cancellation_signal disconnect_wait_cancellation_;
+    bool disconnect_wait_active_ = false;
+    bool disconnect_wait_cancelled_ = false;
 
     // Chunked streaming support
     std::shared_ptr<std::queue<std::string>> chunk_queue_;
@@ -83,6 +88,8 @@ namespace khttpd::framework
     void start_stream_response(HttpResponseStream::ResponseHead head, HttpResponseStream::Callback callback);
     void write_stream_response(net::const_buffer source, HttpResponseStream::Callback callback);
     void finish_stream_response(HttpResponseStream::Callback callback);
+    void wait_stream_disconnect(HttpResponseStream::Callback callback);
+    void cancel_stream_disconnect_wait();
     void on_read(const beast::error_code& ec, std::size_t bytes_transferred);
 
     void handle_request();
diff --git a/framework/sse/sse_session.cpp b/framework/sse/sse_session.cpp
index c76663d..7902b0f 100644
--- a/framework/sse/sse_session.cpp
+++ b/framework/sse/sse_session.cpp
@@ -47,11 +47,27 @@ namespace khttpd::framework::sse
       response->async_start(std::move(head), [self = shared_from_this()](boost::system::error_code ec)
       {
         if (ec) return self->finish(ec);
-        { std::lock_guard lock(self->mutex); self->started = true; }
+        bool watch_disconnect = false;
+        {
+          std::lock_guard lock(self->mutex);
+          self->started = true;
+          watch_disconnect = !self->closing && !self->closed;
+        }
+        if (watch_disconnect) self->wait_for_disconnect();
         self->write_next();
       });
     }
 
+    void wait_for_disconnect()
+    {
+      response->async_wait_disconnect([self = shared_from_this()](boost::system::error_code ec)
+      {
+        if (ec == boost::asio::error::operation_aborted) return;
+        self->response->cancel();
+        self->finish(ec ? ec : boost::asio::error::eof);
+      });
+    }
+
     bool enqueue(std::string value)
     {
       {
@@ -119,6 +135,7 @@ namespace khttpd::framework::sse
         queue.clear();
         handler = std::move(close_handler);
       }
+      response->cancel_disconnect_wait();
       invoke_close_handler(handler, ec);
     }
 
@@ -146,7 +163,12 @@ namespace khttpd::framework::sse
   void SseSession::start() { impl_->start(); }
   bool SseSession::send(SseEvent event) { return impl_->enqueue(format_sse_event(event)); }
   bool SseSession::send_comment(std::string comment) { return impl_->enqueue(format_sse_comment(comment)); }
-  void SseSession::close() { { std::lock_guard lock(impl_->mutex); impl_->closing = true; } impl_->write_next(); }
+  void SseSession::close()
+  {
+    impl_->response->cancel_disconnect_wait();
+    { std::lock_guard lock(impl_->mutex); impl_->closing = true; }
+    impl_->write_next();
+  }
   void SseSession::cancel() { impl_->response->cancel(); impl_->finish(boost::asio::error::operation_aborted); }
   void SseSession::on_close(CloseHandler handler)
   {
diff --git a/framework/tests/sse_test.cpp b/framework/tests/sse_test.cpp
index 238c727..d464e71 100644
--- a/framework/tests/sse_test.cpp
+++ b/framework/tests/sse_test.cpp
@@ -3,6 +3,10 @@
 #include <boost/asio.hpp>
 #include <boost/beast/http.hpp>
 #include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <future>
+#include <mutex>
 #include <stdexcept>
 #include <thread>
 
@@ -26,6 +30,7 @@ namespace
   {
   public:
     bool finished = false;
+    bool disconnect_wait_started = false;
 
     void async_start(ResponseHead, Callback callback) override { callback({}); }
     void async_write_some(net::const_buffer, Callback callback) override { callback({}); }
@@ -34,6 +39,7 @@ namespace
       finished = true;
       callback({});
     }
+    void async_wait_disconnect(Callback) override { disconnect_wait_started = true; }
     void cancel() override {}
   };
 }
@@ -255,6 +261,87 @@ TEST(SseSessionTest, ContainsExceptionsThrownByCloseCallback)
   EXPECT_TRUE(response->finished);
 }
 
+TEST(SseSessionTest, GracefulCloseBeforeStartDoesNotArmADisconnectRead)
+{
+  auto response = std::make_shared<ImmediateResponseStream>();
+  sse::SseSession session(response, 11, true);
+  session.close();
+
+  session.start();
+
+  EXPECT_TRUE(response->finished);
+  EXPECT_FALSE(response->disconnect_wait_started);
+}
+
+TEST(SseSessionTest, DetectsAPassiveClientDisconnectWithoutAnotherWrite)
+{
+  test::TempWebRoot web_root;
+  fw::HttpRouter router;
+  fw::WebsocketRouter websocket_router;
+  std::mutex mutex;
+  std::condition_variable closed_condition;
+  std::shared_ptr<sse::SseSession> server_session;
+  bool closed = false;
+  router.sse("/events", [&](fw::HttpContext&, std::shared_ptr<sse::SseSession> session)
+  {
+    {
+      std::lock_guard lock(mutex);
+      server_session = session;
+    }
+    session->on_close([&](boost::system::error_code)
+    {
+      {
+        std::lock_guard lock(mutex);
+        closed = true;
+      }
+      closed_condition.notify_one();
+    });
+  });
+
+  net::io_context server_ioc;
+  auto guard = net::make_work_guard(server_ioc);
+  tcp::acceptor acceptor(server_ioc, {net::ip::address_v4::loopback(), 0});
+  const auto endpoint = acceptor.local_endpoint();
+  acceptor.async_accept([&](boost::system::error_code ec, tcp::socket socket)
+  {
+    if (ec) return;
+    std::make_shared<fw::HttpSession>(std::move(socket), router, websocket_router,
+                                      web_root.path.string(), boost::filesystem::canonical(web_root.path))->run();
+  });
+  std::thread server_thread([&] { server_ioc.run(); });
+
+  net::io_context client_ioc;
+  tcp::socket client(client_ioc);
+  client.connect(endpoint);
+  http::request<http::empty_body> request{http::verb::get, "/events", 11};
+  request.set(http::field::host, "localhost");
+  request.keep_alive(true);
+  http::write(client, request);
+  boost::beast::flat_buffer response_buffer;
+  http::response_parser<http::empty_body> response_parser;
+  http::read_header(client, response_buffer, response_parser);
+  boost::system::error_code ignored;
+  client.shutdown(tcp::socket::shutdown_both, ignored);
+  client.close(ignored);
+
+  std::unique_lock lock(mutex);
+  const bool detected = closed_condition.wait_for(lock, std::chrono::milliseconds(500), [&] { return closed; });
+  auto session = server_session;
+  server_session.reset();
+  lock.unlock();
+  if (session && session->is_open()) session->cancel();
+  session.reset();
+
+  std::promise<void> cleanup_complete;
+  auto cleanup_future = cleanup_complete.get_future();
+  net::post(server_ioc, [&cleanup_complete] { cleanup_complete.set_value(); });
+  cleanup_future.wait();
+
+  guard.reset();
+  server_thread.join();
+  EXPECT_TRUE(detected);
+}
+
 TEST(SseClientTest, DeliversEventsFromAnAsyncEventStream)
 {
   net::io_context server_ioc;