From c7910e49162c106a932820d4c4c5d739ae489122 Mon Sep 17 00:00:00 2001
From: gophergogo <gogo@gopher.security>
Date: Mon, 6 Jul 2026 18:37:51 -0700
Subject: [PATCH 1/5] Introduce a per-message dispatch context for JSON-RPC
 messages (#264)

A message's origin (connection, transport session id) and its reply path
currently live in ambient server state stamped at accept time, which is
wrong whenever connections interleave. MessageDispatchContext makes the
origin travel with the message instead: the filter that parses a message
constructs a stack-scoped context and dispatches through new
context-carrying handler hooks, pairing decode and encode on the same
connection. Defaults forward to the existing context-free hooks, so this
commit changes no behavior; producers and the server migrate next.

Distinct method names (onRequestWithContext) rather than overloads keep
every existing implementation clear of -Woverloaded-virtual hiding.
---
 include/mcp/filter/json_rpc_protocol_filter.h | 31 +++++++
 include/mcp/mcp_connection_manager.h          | 29 +++++++
 include/mcp/message_dispatch_context.h        | 81 +++++++++++++++++++
 src/filter/json_rpc_protocol_filter.cc        | 51 +++++++++++-
 4 files changed, 190 insertions(+), 2 deletions(-)
 create mode 100644 include/mcp/message_dispatch_context.h

diff --git a/include/mcp/filter/json_rpc_protocol_filter.h b/include/mcp/filter/json_rpc_protocol_filter.h
index 5e6187bbf..32f06650f 100644
--- a/include/mcp/filter/json_rpc_protocol_filter.h
+++ b/include/mcp/filter/json_rpc_protocol_filter.h
@@ -4,6 +4,7 @@
 #include "mcp/event/event_loop.h"
 #include "mcp/filter/filter_context.h"
 #include "mcp/json/json_bridge.h"
+#include "mcp/message_dispatch_context.h"
 #include "mcp/network/filter.h"
 #include "mcp/types.h"
 
@@ -66,6 +67,30 @@ class JsonRpcProtocolFilter : public network::Filter {
      * @param error The error details
      */
     virtual void onProtocolError(const Error& error) = 0;
+
+    /**
+     * Context-carrying variants. The filter that parsed the message builds
+     * a per-message context (origin connection + a reply path through this
+     * filter's own encoder) and dispatches through these; the defaults
+     * forward to the context-free hooks so existing handlers keep working.
+     * Handlers that forward toward McpProtocolCallbacks should override
+     * these and propagate the context so origin information travels with
+     * the message all the way to the application layer. Distinct names
+     * (not overloads) keep existing handlers outside the
+     * -Woverloaded-virtual hiding trap.
+     */
+    virtual void onRequestWithContext(const jsonrpc::Request& request,
+                                      MessageDispatchContext& context) {
+      (void)context;
+      onRequest(request);
+    }
+
+    virtual void onNotificationWithContext(
+        const jsonrpc::Notification& notification,
+        MessageDispatchContext& context) {
+      (void)context;
+      onNotification(notification);
+    }
   };
 
   /**
@@ -148,6 +173,12 @@ class JsonRpcProtocolFilter : public network::Filter {
   class EncoderImpl;
   friend class EncoderImpl;  // Allow encoder to access private members
 
+  // Per-message dispatch context handed to the MessageHandler: origin is
+  // this filter's connection, reply path is this filter's encoder. Nested
+  // (like EncoderImpl) so it can read write_callbacks_.
+  class DispatchContextImpl;
+  friend class DispatchContextImpl;
+
   /**
    * Parse messages from buffer
    * @param buffer Data buffer containing JSON messages
diff --git a/include/mcp/mcp_connection_manager.h b/include/mcp/mcp_connection_manager.h
index c008a68f1..67fd1f960 100644
--- a/include/mcp/mcp_connection_manager.h
+++ b/include/mcp/mcp_connection_manager.h
@@ -7,6 +7,7 @@
 #include "mcp/core/result.h"
 #include "mcp/event/event_loop.h"
 #include "mcp/json/json_bridge.h"
+#include "mcp/message_dispatch_context.h"
 #include "mcp/network/connection_manager.h"
 #include "mcp/network/filter.h"
 #include "mcp/transport/http_sse_transport_socket.h"
@@ -70,6 +71,34 @@ class McpProtocolCallbacks {
    */
   virtual void onNotification(const jsonrpc::Notification& notification) = 0;
 
+  /**
+   * Context-carrying variants: the message arrives together with a
+   * per-message dispatch context describing its origin (connection,
+   * transport session id) and reply path. Producers that know the origin
+   * call these; the defaults forward to the context-free hooks so existing
+   * implementations keep working unchanged. Receivers that route replies or
+   * key sessions should override these instead of the context-free forms —
+   * the context makes "respond to the wrong connection" and "inherit a
+   * stale session binding" unrepresentable, where ambient
+   * current-connection state cannot.
+   *
+   * Distinct names (rather than overloads of onRequest/onNotification) keep
+   * every existing implementation outside the -Woverloaded-virtual hiding
+   * trap.
+   */
+  virtual void onRequestWithContext(const jsonrpc::Request& request,
+                                    MessageDispatchContext& context) {
+    (void)context;
+    onRequest(request);
+  }
+
+  virtual void onNotificationWithContext(
+      const jsonrpc::Notification& notification,
+      MessageDispatchContext& context) {
+    (void)context;
+    onNotification(notification);
+  }
+
   /**
    * Called when a response is received
    */
diff --git a/include/mcp/message_dispatch_context.h b/include/mcp/message_dispatch_context.h
new file mode 100644
index 000000000..cd70bafb0
--- /dev/null
+++ b/include/mcp/message_dispatch_context.h
@@ -0,0 +1,81 @@
+#pragma once
+
+#include <string>
+
+#include "mcp/core/result.h"
+#include "mcp/types.h"
+
+namespace mcp {
+
+namespace network {
+class Connection;
+}
+
+/**
+ * Per-message dispatch context.
+ *
+ * Carries where a JSON-RPC message came from and how to send a reply back
+ * along the same path. The transport producer (the filter or connection
+ * manager that parsed the message) constructs one immediately before
+ * dispatching the message and passes it through the callback chain, so the
+ * origin travels *with* the message instead of living in ambient state on
+ * the receiver. A stale binding is unrepresentable by construction: the
+ * context dies when the dispatch call returns.
+ *
+ * Lifetime contract: valid only for the duration of the dispatch call, on
+ * the dispatcher thread. Receivers must not retain a pointer or reference
+ * past the callback's return; a handler that wants to finish work
+ * asynchronously must resolve what it needs (session id, transport session
+ * id) while the context is live.
+ */
+class MessageDispatchContext {
+ public:
+  virtual ~MessageDispatchContext() = default;
+
+  /**
+   * The connection the message physically arrived on. May be null when the
+   * producer is not connection-backed (e.g. a legacy dispatch path with no
+   * origin information). Only guaranteed valid while the context is live.
+   */
+  virtual network::Connection* originConnection() const = 0;
+
+  /**
+   * Durable transport-level session id the message belongs to (e.g. the SSE
+   * stream id from a POST /callback/{id} path — the client identity that
+   * outlives the one-shot POST connection). Empty when the transport has no
+   * session concept (stdio, plain HTTP); receivers then fall back to
+   * connection identity.
+   */
+  virtual const std::string& transportSessionId() const = 0;
+
+  /**
+   * Send a JSON-RPC response back along this message's own return path.
+   * Returns an error (rather than silently dropping) when the path is gone,
+   * e.g. the origin connection already closed.
+   */
+  virtual VoidResult sendResponse(const jsonrpc::Response& response) = 0;
+};
+
+/**
+ * Context for dispatch paths that carry no origin information. Session
+ * resolution falls back to "no connection" and any attempted reply fails
+ * loudly instead of being written to an unrelated connection.
+ */
+class NullMessageDispatchContext : public MessageDispatchContext {
+ public:
+  network::Connection* originConnection() const override { return nullptr; }
+
+  const std::string& transportSessionId() const override {
+    static const std::string empty;
+    return empty;
+  }
+
+  VoidResult sendResponse(const jsonrpc::Response&) override {
+    Error err;
+    err.code = jsonrpc::INTERNAL_ERROR;
+    err.message = "no dispatch context: response has no return path";
+    return makeVoidError(err);
+  }
+};
+
+}  // namespace mcp
diff --git a/src/filter/json_rpc_protocol_filter.cc b/src/filter/json_rpc_protocol_filter.cc
index 792cbd9d7..0550bc3b7 100644
--- a/src/filter/json_rpc_protocol_filter.cc
+++ b/src/filter/json_rpc_protocol_filter.cc
@@ -165,6 +165,51 @@ class JsonRpcProtocolFilter::EncoderImpl
   JsonRpcProtocolFilter& parent_;
 };
 
+// DispatchContextImpl - per-message origin + reply path for the handler.
+//
+// Why the encoder is the reply path: the response must be framed exactly
+// like any other outbound message on the connection the request arrived on
+// (newline-delimited or length-prefixed, per this filter's configuration).
+// Routing the reply through the parsing filter's own encoder pairs decode
+// and encode on the same connection, so a handler can never answer on a
+// different connection than the one that asked.
+//
+// Stack-constructed immediately before each dispatch; dies when the
+// dispatch returns, which is what makes a stale origin unrepresentable.
+class JsonRpcProtocolFilter::DispatchContextImpl
+    : public MessageDispatchContext {
+ public:
+  explicit DispatchContextImpl(JsonRpcProtocolFilter& parent)
+      : parent_(parent) {}
+
+  network::Connection* originConnection() const override {
+    return parent_.write_callbacks_ ? &parent_.write_callbacks_->connection()
+                                    : nullptr;
+  }
+
+  const std::string& transportSessionId() const override {
+    // A bare JSON-RPC chain has no transport session concept; composite
+    // transports (HTTP+SSE) supply their own context with the stream id.
+    static const std::string empty;
+    return empty;
+  }
+
+  VoidResult sendResponse(const jsonrpc::Response& response) override {
+    // Fail loudly when the reply path is gone: a null result here would
+    // otherwise read as "sent" to the caller while nothing went out.
+    if (!parent_.write_callbacks_) {
+      Error err;
+      err.code = jsonrpc::INTERNAL_ERROR;
+      err.message = "response dropped: origin connection is gone";
+      return makeVoidError(err);
+    }
+    return parent_.encoder_->encodeResponse(response);
+  }
+
+ private:
+  JsonRpcProtocolFilter& parent_;
+};
+
 // JsonRpcProtocolFilter implementation
 
 JsonRpcProtocolFilter::JsonRpcProtocolFilter(MessageHandler& handler,
@@ -328,13 +373,15 @@ bool JsonRpcProtocolFilter::parseMessage(const std::string& json_str) {
         GOPHER_LOG_DEBUG("JsonRpcFilter dispatching request for method: {}",
                          request.method);
         requests_received_++;
-        handler_.onRequest(request);
+        DispatchContextImpl context(*this);
+        handler_.onRequestWithContext(request, context);
       } else {
         // JSON-RPC Notification
         jsonrpc::Notification notification =
             json::from_json<jsonrpc::Notification>(json_val);
         notifications_received_++;
-        handler_.onNotification(notification);
+        DispatchContextImpl context(*this);
+        handler_.onNotificationWithContext(notification, context);
       }
     } else if (json_val.contains("result") || json_val.contains("error")) {
       // JSON-RPC Response

From 97131dcb157ef254f91f2db21aa02d274de94bc1 Mon Sep 17 00:00:00 2001
From: gophergogo <gogo@gopher.security>
Date: Mon, 6 Jul 2026 19:06:11 -0700
Subject: [PATCH 2/5] Propagate the dispatch context through every message
 producer (#264)

Each adapter between the JSON-RPC filter and the application callbacks now
forwards the per-message context instead of dropping it: the stdio,
protocol-detection, enhanced and factory-header adapters, the
JSON-RPC-to-protocol bridge, the application-base adapter, and the
connection manager. The HTTP+SSE composite supplies its own richer context
carrying the SSE stream id and a reply sink that writes to the origin POST
connection, where its onWrite already decides the wire form (SSE-registry
reroute or HTTP framing).

No behavior change yet: the server still consumes the context-free hooks,
which every context-carrying default forwards to.
---
 include/mcp/filter/callback_bridges.h         |  6 ++
 include/mcp/filter/json_rpc_filter_factory.h  | 12 ++++
 include/mcp/mcp_application_base.h            | 12 ++++
 include/mcp/mcp_connection_manager.h          |  6 ++
 src/filter/callback_bridges.cc                | 18 +++++
 src/filter/enhanced_filter_chain_factory.cc   | 19 +++++
 src/filter/http_sse_filter_chain_factory.cc   | 72 ++++++++++++++++++-
 ...protocol_detection_filter_chain_factory.cc | 12 ++++
 src/filter/stdio_filter_chain_factory.cc      | 12 ++++
 src/mcp_connection_manager.cc                 | 16 +++++
 10 files changed, 183 insertions(+), 2 deletions(-)

diff --git a/include/mcp/filter/callback_bridges.h b/include/mcp/filter/callback_bridges.h
index 4a97c516e..d949ef5b8 100644
--- a/include/mcp/filter/callback_bridges.h
+++ b/include/mcp/filter/callback_bridges.h
@@ -99,6 +99,12 @@ class JsonRpcToProtocolBridge : public JsonRpcProtocolFilter::MessageHandler {
   void onNotification(const jsonrpc::Notification& notification) override;
   void onResponse(const jsonrpc::Response& response) override;
   void onProtocolError(const Error& error) override;
+  // Propagate the per-message origin built by the JSON-RPC filter so the
+  // application layer can reply on the connection the message arrived on.
+  void onRequestWithContext(const jsonrpc::Request& request,
+                            MessageDispatchContext& context) override;
+  void onNotificationWithContext(const jsonrpc::Notification& notification,
+                                 MessageDispatchContext& context) override;
 
  private:
   McpProtocolCallbacks& callbacks_;
diff --git a/include/mcp/filter/json_rpc_filter_factory.h b/include/mcp/filter/json_rpc_filter_factory.h
index 49cd68168..05cb501ed 100644
--- a/include/mcp/filter/json_rpc_filter_factory.h
+++ b/include/mcp/filter/json_rpc_filter_factory.h
@@ -33,6 +33,18 @@ class DirectJsonRpcCallbacks : public JsonRpcProtocolFilter::MessageHandler {
     mcp_callbacks_.onError(error);
   }
 
+  // Propagate the per-message origin built by the JSON-RPC filter so the
+  // application layer can reply on the connection the message arrived on.
+  void onRequestWithContext(const jsonrpc::Request& request,
+                            MessageDispatchContext& context) override {
+    mcp_callbacks_.onRequestWithContext(request, context);
+  }
+
+  void onNotificationWithContext(const jsonrpc::Notification& notification,
+                                 MessageDispatchContext& context) override {
+    mcp_callbacks_.onNotificationWithContext(notification, context);
+  }
+
  private:
   McpProtocolCallbacks& mcp_callbacks_;
 };
diff --git a/include/mcp/mcp_application_base.h b/include/mcp/mcp_application_base.h
index 4fcf093ed..bc1f2e323 100644
--- a/include/mcp/mcp_application_base.h
+++ b/include/mcp/mcp_application_base.h
@@ -143,6 +143,18 @@ class McpToJsonRpcAdapter
     callbacks_.onError(error);
   }
 
+  // Propagate the per-message origin built by the JSON-RPC filter so the
+  // application layer can reply on the connection the message arrived on.
+  void onRequestWithContext(const jsonrpc::Request& request,
+                            MessageDispatchContext& context) override {
+    callbacks_.onRequestWithContext(request, context);
+  }
+
+  void onNotificationWithContext(const jsonrpc::Notification& notification,
+                                 MessageDispatchContext& context) override {
+    callbacks_.onNotificationWithContext(notification, context);
+  }
+
  private:
   McpProtocolCallbacks& callbacks_;
 };
diff --git a/include/mcp/mcp_connection_manager.h b/include/mcp/mcp_connection_manager.h
index 67fd1f960..1a5ca0050 100644
--- a/include/mcp/mcp_connection_manager.h
+++ b/include/mcp/mcp_connection_manager.h
@@ -232,6 +232,12 @@ class McpConnectionManager : public McpProtocolCallbacks,
   void onRequest(const jsonrpc::Request& request) override;
   void onNotification(const jsonrpc::Notification& notification) override;
   void onResponse(const jsonrpc::Response& response) override;
+  // Forward the per-message origin alongside the message so the application
+  // layer can key sessions and route replies without ambient state.
+  void onRequestWithContext(const jsonrpc::Request& request,
+                            MessageDispatchContext& context) override;
+  void onNotificationWithContext(const jsonrpc::Notification& notification,
+                                 MessageDispatchContext& context) override;
   void onConnectionEvent(network::ConnectionEvent event) override;
   void onError(const Error& error) override;
   void onMessageEndpoint(const std::string& endpoint) override;
diff --git a/src/filter/callback_bridges.cc b/src/filter/callback_bridges.cc
index ab589416a..8b9654502 100644
--- a/src/filter/callback_bridges.cc
+++ b/src/filter/callback_bridges.cc
@@ -153,6 +153,24 @@ void JsonRpcToProtocolBridge::onNotification(
   callbacks_.onNotification(notification);
 }
 
+void JsonRpcToProtocolBridge::onRequestWithContext(
+    const jsonrpc::Request& request, MessageDispatchContext& context) {
+  GOPHER_LOG_DEBUG(
+      "JsonRpcToProtocolBridge::onRequestWithContext called with method='{}'",
+      request.method);
+  callbacks_.onRequestWithContext(request, context);
+}
+
+void JsonRpcToProtocolBridge::onNotificationWithContext(
+    const jsonrpc::Notification& notification,
+    MessageDispatchContext& context) {
+  GOPHER_LOG_DEBUG(
+      "JsonRpcToProtocolBridge::onNotificationWithContext called with "
+      "method='{}'",
+      notification.method);
+  callbacks_.onNotificationWithContext(notification, context);
+}
+
 void JsonRpcToProtocolBridge::onResponse(const jsonrpc::Response& response) {
   if (response.id.has_value()) {
     GOPHER_LOG_DEBUG("JsonRpcToProtocolBridge::onResponse called with id={}",
diff --git a/src/filter/enhanced_filter_chain_factory.cc b/src/filter/enhanced_filter_chain_factory.cc
index 3cf305832..a428d90b0 100644
--- a/src/filter/enhanced_filter_chain_factory.cc
+++ b/src/filter/enhanced_filter_chain_factory.cc
@@ -333,6 +333,25 @@ class EnhancedProtocolFilter : public network::Filter,
     mcp_callbacks_.onNotification(notification);
   }
 
+  // Context-carrying variants: track metrics identically, then propagate
+  // the per-message origin so the application layer can reply on the
+  // connection the message arrived on.
+  void onRequestWithContext(const jsonrpc::Request& request,
+                            MessageDispatchContext& context) override {
+    if (metrics_collector_) {
+      metrics_collector_->onRequest(request);
+    }
+    mcp_callbacks_.onRequestWithContext(request, context);
+  }
+
+  void onNotificationWithContext(const jsonrpc::Notification& notification,
+                                 MessageDispatchContext& context) override {
+    if (metrics_collector_) {
+      metrics_collector_->onNotification(notification);
+    }
+    mcp_callbacks_.onNotificationWithContext(notification, context);
+  }
+
   void onProtocolError(const Error& error) override {
     // Track errors
     if (circuit_breaker_) {
diff --git a/src/filter/http_sse_filter_chain_factory.cc b/src/filter/http_sse_filter_chain_factory.cc
index 3e129dc38..52df4562b 100644
--- a/src/filter/http_sse_filter_chain_factory.cc
+++ b/src/filter/http_sse_filter_chain_factory.cc
@@ -925,6 +925,55 @@ class HttpSseJsonRpcProtocolFilter
 
   // ===== JsonRpcProtocolFilter::MessageHandler =====
 
+  /**
+   * Per-message dispatch context for messages decoded on this composite
+   * chain. Origin is the connection the message physically arrived on (for
+   * HTTP+SSE, the short-lived POST connection); the transport session id is
+   * the SSE stream id parsed from POST /callback/{id} — the durable client
+   * identity the server keys its session on.
+   *
+   * The reply sink writes the bare JSON to the origin connection, exactly
+   * the bytes the server used to write to its ambient current-connection
+   * pointer: this composite's own onWrite then decides the wire form
+   * (reroute through the SSE registry for a callback proxy, HTTP-frame for
+   * plain HTTP). Same wire behavior, but the destination is the message's
+   * own connection by construction.
+   */
+  class DispatchContext : public MessageDispatchContext {
+   public:
+    explicit DispatchContext(HttpSseJsonRpcProtocolFilter& parent)
+        : parent_(parent) {}
+
+    network::Connection* originConnection() const override {
+      return parent_.write_callbacks_ ? &parent_.write_callbacks_->connection()
+                                      : nullptr;
+    }
+
+    const std::string& transportSessionId() const override {
+      return parent_.sse_callback_session_id_;
+    }
+
+    VoidResult sendResponse(const jsonrpc::Response& response) override {
+      // Fail loudly when the reply path is gone instead of pretending the
+      // response went out.
+      if (!parent_.write_callbacks_) {
+        Error err;
+        err.code = jsonrpc::INTERNAL_ERROR;
+        err.message = "response dropped: origin connection is gone";
+        return makeVoidError(err);
+      }
+      auto json_val = json::to_json(response);
+      std::string json_str = json_val.toString();
+      OwnedBuffer buffer;
+      buffer.add(json_str);
+      parent_.write_callbacks_->connection().write(buffer, false);
+      return makeVoidSuccess();
+    }
+
+   private:
+    HttpSseJsonRpcProtocolFilter& parent_;
+  };
+
   /**
    * Called by JsonRpcProtocolFilter when a complete JSON-RPC request is parsed
    * Creates a RequestStream to track this request-response pair
@@ -940,12 +989,31 @@ class HttpSseJsonRpcProtocolFilter
     // reads from different connections interleave; an empty id explicitly
     // clears any previous connection's binding.
     mcp_callbacks_.onTransportSessionBound(sse_callback_session_id_);
-    mcp_callbacks_.onRequest(request);
+    DispatchContext context(*this);
+    mcp_callbacks_.onRequestWithContext(request, context);
+  }
+
+  /**
+   * The JSON-RPC sub-filter dispatches through here with its own generic
+   * context, but that context knows neither the SSE stream id nor this
+   * composite's write semantics — replace it with the composite's own.
+   */
+  void onRequestWithContext(const jsonrpc::Request& request,
+                            MessageDispatchContext& context) override {
+    (void)context;
+    onRequest(request);
+  }
+
+  void onNotificationWithContext(const jsonrpc::Notification& notification,
+                                 MessageDispatchContext& context) override {
+    (void)context;
+    onNotification(notification);
   }
 
   void onNotification(const jsonrpc::Notification& notification) override {
     mcp_callbacks_.onTransportSessionBound(sse_callback_session_id_);
-    mcp_callbacks_.onNotification(notification);
+    DispatchContext context(*this);
+    mcp_callbacks_.onNotificationWithContext(notification, context);
 
     // For HTTP transport, send HTTP 202 Accepted response
     // JSON-RPC notifications don't have responses, but HTTP requires one
diff --git a/src/filter/protocol_detection_filter_chain_factory.cc b/src/filter/protocol_detection_filter_chain_factory.cc
index 28776417c..bfab2436b 100644
--- a/src/filter/protocol_detection_filter_chain_factory.cc
+++ b/src/filter/protocol_detection_filter_chain_factory.cc
@@ -34,6 +34,18 @@ class ProtocolDetectionJsonRpcCallbacks
     mcp_callbacks_.onError(error);
   }
 
+  // Propagate the per-message origin built by the JSON-RPC filter so the
+  // application layer can reply on the connection the message arrived on.
+  void onRequestWithContext(const jsonrpc::Request& request,
+                            MessageDispatchContext& context) override {
+    mcp_callbacks_.onRequestWithContext(request, context);
+  }
+
+  void onNotificationWithContext(const jsonrpc::Notification& notification,
+                                 MessageDispatchContext& context) override {
+    mcp_callbacks_.onNotificationWithContext(notification, context);
+  }
+
  private:
   McpProtocolCallbacks& mcp_callbacks_;
 };
diff --git a/src/filter/stdio_filter_chain_factory.cc b/src/filter/stdio_filter_chain_factory.cc
index 994e0064c..f8adcd3fe 100644
--- a/src/filter/stdio_filter_chain_factory.cc
+++ b/src/filter/stdio_filter_chain_factory.cc
@@ -28,6 +28,18 @@ class DirectJsonRpcCallbacks : public JsonRpcProtocolFilter::MessageHandler {
     mcp_callbacks_.onNotification(notification);
   }
 
+  // Propagate the per-message origin built by the JSON-RPC filter so the
+  // application layer can reply on the connection the message arrived on.
+  void onRequestWithContext(const jsonrpc::Request& request,
+                            MessageDispatchContext& context) override {
+    mcp_callbacks_.onRequestWithContext(request, context);
+  }
+
+  void onNotificationWithContext(const jsonrpc::Notification& notification,
+                                 MessageDispatchContext& context) override {
+    mcp_callbacks_.onNotificationWithContext(notification, context);
+  }
+
   void onResponse(const jsonrpc::Response& response) override {
     mcp_callbacks_.onResponse(response);
   }
diff --git a/src/mcp_connection_manager.cc b/src/mcp_connection_manager.cc
index 6b75169c5..15df01e5f 100644
--- a/src/mcp_connection_manager.cc
+++ b/src/mcp_connection_manager.cc
@@ -767,6 +767,22 @@ void McpConnectionManager::onRequest(const jsonrpc::Request& request) {
   }
 }
 
+void McpConnectionManager::onRequestWithContext(
+    const jsonrpc::Request& request, MessageDispatchContext& context) {
+  if (protocol_callbacks_) {
+    protocol_callbacks_->onRequestWithContext(request, context);
+  }
+}
+
+void McpConnectionManager::onNotificationWithContext(
+    const jsonrpc::Notification& notification,
+    MessageDispatchContext& context) {
+  if (protocol_callbacks_) {
+    protocol_callbacks_->onNotificationWithContext(notification, context);
+  }
+  // HTTP 202 response is sent by HttpSseJsonRpcProtocolFilter::onNotification
+}
+
 void McpConnectionManager::onNotification(
     const jsonrpc::Notification& notification) {
   if (protocol_callbacks_) {

From d73ae596b8f6a7a647e04927c8733bf9e9c34a65 Mon Sep 17 00:00:00 2001
From: gophergogo <gogo@gopher.security>
Date: Mon, 6 Jul 2026 19:11:49 -0700
Subject: [PATCH 3/5] Key sessions and route replies by the message's own
 dispatch context (#264)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The server previously resolved sessions and wrote responses against an
ambient current-connection pointer stamped at accept time, which is the
most recently accepted connection — not the one the message arrived on.
With concurrent connections that cross-wires replies (accept A, accept B,
A's request answered on B), attributes a request to another connection's
session on the connection-keyed fallback, and an unrelated close nulls the
pointer so a live connection's response is silently skipped.

onRequest/onNotification now consume the per-message context: the session
fallback keys on the origin connection, the reply goes out through the
context's return path, and a failed send is logged and counted instead of
dropped. This also retires the response broadcast to the first connected
stdio manager, which double-sent every HTTP response onto stdio when both
transports were active. The context-free hooks remain as an explicitly
degraded fallback for out-of-tree producers and warn when hit.
---
 include/mcp/server/mcp_server.h |  38 +++++++-
 src/server/mcp_server.cc        | 159 +++++++++++++++++++-------------
 2 files changed, 130 insertions(+), 67 deletions(-)

diff --git a/include/mcp/server/mcp_server.h b/include/mcp/server/mcp_server.h
index 9457085ab..fad14c70d 100644
--- a/include/mcp/server/mcp_server.h
+++ b/include/mcp/server/mcp_server.h
@@ -960,6 +960,17 @@ class McpServer : public application::ApplicationBase,
   void onRequest(const jsonrpc::Request& request) override;
   void onNotification(const jsonrpc::Notification& notification) override;
   void onResponse(const jsonrpc::Response& response) override;
+
+  // Context-carrying dispatch entry points. The context travels with the
+  // message from the filter that parsed it, so session keying and the reply
+  // both bind to the message's own origin — never to whichever connection
+  // happened to be accepted or announced most recently. The context-free
+  // overrides above remain only as a degraded fallback for producers that
+  // do not supply origin information.
+  void onRequestWithContext(const jsonrpc::Request& request,
+                            MessageDispatchContext& context);
+  void onNotificationWithContext(const jsonrpc::Notification& notification,
+                                 MessageDispatchContext& context);
   void onConnectionEvent(network::ConnectionEvent event);
   void onError(const Error& error) override;
 
@@ -996,12 +1007,18 @@ class McpServer : public application::ApplicationBase,
   // Register built-in handlers
   void registerBuiltinHandlers();
 
-  // Resolve the session for the message currently being dispatched.
-  // Prefers the transport session id (durable across the short-lived POST
-  // connections of HTTP+SSE) and falls back to connection identity for
+  // Resolve the session for the message described by `context`. Prefers
+  // the transport session id (durable across the short-lived POST
+  // connections of HTTP+SSE) and falls back to the origin connection for
   // transports without one (stdio, plain HTTP). Returns nullptr only when
   // the session limit is reached. Dispatcher thread only.
-  SessionManager::SessionPtr getOrCreateCurrentSession();
+  SessionManager::SessionPtr getOrCreateSessionFor(
+      const MessageDispatchContext& context);
+
+  // Reply path for messages that arrived without origin information (the
+  // context-free legacy hooks): no connection to key a session on, replies
+  // fall back to the first connected stdio manager. Defined in the .cc.
+  class LegacyDispatchContext;
 
   // Deliver a notification to one session's client, routed by how the
   // session is keyed: SSE stream (via the registry), owning connection,
@@ -1095,6 +1112,19 @@ class McpServer : public application::ApplicationBase,
       server_.onNotification(notification);
     }
 
+    void onRequestWithContext(const jsonrpc::Request& request,
+                              MessageDispatchContext& context) override {
+      GOPHER_LOG_DEBUG(
+          "ServerProtocolCallbacks::onRequestWithContext for method: {}",
+          request.method);
+      server_.onRequestWithContext(request, context);
+    }
+
+    void onNotificationWithContext(const jsonrpc::Notification& notification,
+                                   MessageDispatchContext& context) override {
+      server_.onNotificationWithContext(notification, context);
+    }
+
     void onResponse(const jsonrpc::Response& response) override {
       server_.onResponse(response);
     }
diff --git a/src/server/mcp_server.cc b/src/server/mcp_server.cc
index 4373eab32..d37a44ce7 100644
--- a/src/server/mcp_server.cc
+++ b/src/server/mcp_server.cc
@@ -757,41 +757,87 @@ void McpServer::setupFilterChain(application::FilterChainBuilder& builder) {
   }
 }
 
-SessionManager::SessionPtr McpServer::getOrCreateCurrentSession() {
-  // The transport session binding is strictly single-use: consume it here
-  // and clear it immediately, so it can only ever apply to the one message
-  // whose dispatch it preceded. The transport filter announces it (via
-  // onTransportSessionBound) right before onRequest/onNotification, but a
-  // producer that does NOT announce — stdio, or any non-SSE filter chain —
-  // must not inherit the previous message's id. Clearing on consume makes a
-  // stale binding unrepresentable rather than relying on every producer to
-  // remember to announce an empty id. Dispatch is synchronous per message
-  // on the dispatcher thread, so consume-and-clear is race-free.
-  std::string transport_session_id;
-  transport_session_id.swap(current_transport_session_id_);
-
+SessionManager::SessionPtr McpServer::getOrCreateSessionFor(
+    const MessageDispatchContext& context) {
+  // The context is constructed per message by the transport that parsed
+  // it and dies when the dispatch returns, so a stale binding is
+  // unrepresentable by construction — no consume-and-clear discipline
+  // needed, unlike the ambient announce-then-dispatch scheme this
+  // replaces.
+  //
   // Transport session id wins: for HTTP+SSE each request arrives on a
   // one-shot POST connection, so keying the session on the connection
   // would hand every request a fresh session and silently drop state
   // such as resource subscriptions. The SSE stream id is the identity that
   // actually spans the client's requests — and it is also what the push
   // path needs to find the client's SSE stream.
+  const std::string& transport_session_id = context.transportSessionId();
   if (!transport_session_id.empty()) {
     return session_manager_->getOrCreateSessionByTransportId(
         transport_session_id);
   }
 
   // Connection-keyed fallback for transports where the connection is
-  // long-lived (stdio) or there is no transport session concept.
-  auto session = session_manager_->getSessionByConnection(current_connection_);
+  // long-lived (stdio) or there is no transport session concept. The
+  // origin comes from the message itself, so interleaved reads from
+  // concurrent connections each land in their own session.
+  auto session =
+      session_manager_->getSessionByConnection(context.originConnection());
   if (!session) {
-    session = session_manager_->createSession(current_connection_);
+    session = session_manager_->createSession(context.originConnection());
   }
   return session;
 }
 
+// Reply path for messages that arrived through the context-free legacy
+// hooks. There is no origin connection to reply on, so fall back to the
+// first connected stdio manager — the historical degraded behavior — and
+// fail loudly when there is none, instead of writing to an unrelated
+// connection.
+class McpServer::LegacyDispatchContext : public MessageDispatchContext {
+ public:
+  explicit LegacyDispatchContext(McpServer& server) : server_(server) {}
+
+  network::Connection* originConnection() const override { return nullptr; }
+
+  const std::string& transportSessionId() const override {
+    static const std::string empty;
+    return empty;
+  }
+
+  VoidResult sendResponse(const jsonrpc::Response& response) override {
+    for (auto& conn_manager : server_.connection_managers_) {
+      if (conn_manager->isConnected()) {
+        return conn_manager->sendResponse(response);
+      }
+    }
+    Error err;
+    err.code = jsonrpc::INTERNAL_ERROR;
+    err.message = "no dispatch context and no connected transport";
+    return makeVoidError(err);
+  }
+
+ private:
+  McpServer& server_;
+};
+
 // McpProtocolCallbacks overrides
 void McpServer::onRequest(const jsonrpc::Request& request) {
+  // Context-free legacy entry: the producer did not say where the message
+  // came from. Every in-tree transport dispatches through
+  // onRequestWithContext; reaching this path means an external producer
+  // has not been migrated, so route replies through the degraded legacy
+  // fallback rather than guessing at a connection.
+  GOPHER_LOG_WARN(
+      "Request '{}' dispatched without origin context; session and reply "
+      "routing degraded to the legacy transport fallback",
+      request.method);
+  LegacyDispatchContext context(*this);
+  onRequestWithContext(request, context);
+}
+
+void McpServer::onRequestWithContext(const jsonrpc::Request& request,
+                                     MessageDispatchContext& context) {
   GOPHER_LOG_DEBUG("McpServer::onRequest called with method: {}",
                    request.method);
 
@@ -813,38 +859,25 @@ void McpServer::onRequest(const jsonrpc::Request& request) {
   }
 
   // Resolve the session for this request: transport session id first
-  // (durable across HTTP+SSE POST connections), connection identity as
+  // (durable across HTTP+SSE POST connections), origin connection as
   // fallback.
-  auto session = getOrCreateCurrentSession();
+  auto session = getOrCreateSessionFor(context);
 
   if (session) {
     pending_req->session_id = session->getId();
   }
 
   if (!session) {
-    // Max sessions reached
+    // Max sessions reached. Reply on the requester's own return path —
+    // the context pins it to the connection the request arrived on.
     server_stats_.requests_failed++;
     auto response = jsonrpc::Response::make_error(
         request.id, Error(jsonrpc::INTERNAL_ERROR, "Max sessions reached"));
 
-    // Send response through appropriate mechanism
-    // For HTTP connections, use filter chain; for stdio, use connection manager
-    GOPHER_LOG_DEBUG("Sending error response for max sessions");
-
-    // Send response through the current connection (for TCP/HTTP connections)
-    // Following production pattern: thread-local connection context
-    if (current_connection_) {
-      filter::HttpSseFilterChainFactory::sendHttpResponse(response,
-                                                          *current_connection_);
-    }
-
-    // Also try connection managers (for stdio connections)
-    for (auto& conn_manager : connection_managers_) {
-      if (conn_manager->isConnected()) {
-        GOPHER_LOG_DEBUG("Found connected manager, sending response");
-        conn_manager->sendResponse(response);
-        break;
-      }
+    auto send_result = context.sendResponse(response);
+    if (holds_alternative<Error>(send_result)) {
+      GOPHER_LOG_ERROR("Failed to send max-sessions error response: {}",
+                       get<Error>(send_result).message);
     }
     return;
   }
@@ -901,36 +934,24 @@ void McpServer::onRequest(const jsonrpc::Request& request) {
     }
   }
 
-  // Send response through appropriate channel
-  // Following proper architecture: use filter chain for HTTP, connection
-  // manager for stdio
+  // Send the response along the request's own return path. The context
+  // pins the origin connection and its transport's framing, so concurrent
+  // connections cannot cross-wire replies and a single dispatch can never
+  // fan out to an unrelated transport. A failed send is surfaced instead
+  // of silently dropped.
   GOPHER_LOG_DEBUG("Sending response for request id: {}",
                    holds_alternative<std::string>(request.id)
                        ? get<std::string>(request.id)
                        : std::to_string(get<int64_t>(request.id)));
 
-  // Send response through the current connection (for TCP/HTTP connections)
-  // Following production pattern: server sends JSON-RPC, filter handles HTTP
-  if (current_connection_) {
-    // Convert response to JSON and send through connection
-    // The filter chain will handle HTTP protocol wrapping
-    auto json_val = json::to_json(response);
-    std::string json_str = json_val.toString();
-
-    OwnedBuffer response_buffer;
-    response_buffer.add(json_str);
-
-    // Write JSON-RPC response - HTTP filter will wrap it
-    current_connection_->write(response_buffer, false);
-  }
-
-  // Also try connection managers (for stdio transport)
-  // This is the legacy path for non-HTTP transports
-  for (auto& conn_manager : connection_managers_) {
-    if (conn_manager->isConnected()) {
-      conn_manager->sendResponse(response);
-      break;
-    }
+  auto send_result = context.sendResponse(response);
+  if (holds_alternative<Error>(send_result)) {
+    server_stats_.errors_total++;
+    GOPHER_LOG_ERROR("Failed to send response for request id {}: {}",
+                     holds_alternative<std::string>(request.id)
+                         ? get<std::string>(request.id)
+                         : std::to_string(get<int64_t>(request.id)),
+                     get<Error>(send_result).message);
   }
 
   // Remove request from pending list
@@ -945,12 +966,24 @@ void McpServer::onRequest(const jsonrpc::Request& request) {
 }
 
 void McpServer::onNotification(const jsonrpc::Notification& notification) {
+  // Context-free legacy entry; see onRequest for why this is degraded.
+  GOPHER_LOG_WARN(
+      "Notification '{}' dispatched without origin context; session "
+      "resolution degraded to the legacy transport fallback",
+      notification.method);
+  LegacyDispatchContext context(*this);
+  onNotificationWithContext(notification, context);
+}
+
+void McpServer::onNotificationWithContext(
+    const jsonrpc::Notification& notification,
+    MessageDispatchContext& context) {
   // Handle notification in dispatcher context
   server_stats_.notifications_total++;
 
   // Resolve the session the same way requests do, so a notification sent
   // on a fresh POST connection still lands in the subscriber's session.
-  auto session = getOrCreateCurrentSession();
+  auto session = getOrCreateSessionFor(context);
   if (!session) {
     return;  // Can't process notification without session
   }

From 234234d57fbc238904ee59d17e93ec55806c1a72 Mon Sep 17 00:00:00 2001
From: gophergogo <gogo@gopher.security>
Date: Mon, 6 Jul 2026 19:21:45 -0700
Subject: [PATCH 4/5] Remove the ambient connection and transport-session
 members (#264)

Both request-scoped ambient members are now dead: session keying and reply
routing read the per-message dispatch context, so the accept-time
current_connection_ stamp, its close-path null-out, and the
announce-then-dispatch onTransportSessionBound protocol (interface method,
server binding member, and the composite filter's three announce sites)
all go away. The server-mode integration test now asserts the binding that
travels with each message's context instead of the last announced value,
and seeds a sentinel that would surface if dispatch ever regressed to the
context-free path.
---
 include/mcp/mcp_connection_manager.h          | 20 --------------
 include/mcp/server/mcp_server.h               | 26 -------------------
 src/filter/http_sse_filter_chain_factory.cc   | 15 ++++-------
 src/server/mcp_server.cc                      | 10 +------
 .../test_http_sse_filter_server_mode.cc       | 23 ++++++++++------
 5 files changed, 21 insertions(+), 73 deletions(-)

diff --git a/include/mcp/mcp_connection_manager.h b/include/mcp/mcp_connection_manager.h
index 1a5ca0050..550aba954 100644
--- a/include/mcp/mcp_connection_manager.h
+++ b/include/mcp/mcp_connection_manager.h
@@ -122,26 +122,6 @@ class McpProtocolCallbacks {
     (void)endpoint;  // Default implementation does nothing
   }
 
-  /**
-   * Called immediately before a parsed message is dispatched, carrying the
-   * transport-level session id the message belongs to. For HTTP+SSE servers
-   * this is the SSE stream id extracted from the POST /callback/{id} path —
-   * the durable client identity that outlives the one-shot POST connection
-   * the message physically arrived on. Empty when the transport has no
-   * session concept (stdio, plain HTTP); receivers must then fall back to
-   * connection identity.
-   *
-   * Always invoked in the dispatcher thread right before the matching
-   * onRequest/onNotification/onResponse for the same message, so an
-   * implementation may stash it as request-scoped context without locking.
-   * It is re-announced per message (not per connection) because reads from
-   * different connections interleave on the dispatcher thread.
-   */
-  virtual void onTransportSessionBound(
-      const std::string& transport_session_id) {
-    (void)transport_session_id;  // Default implementation does nothing
-  }
-
   /**
    * Send a POST request to the message endpoint
    * Used by HTTP/SSE transport to send messages on a separate connection
diff --git a/include/mcp/server/mcp_server.h b/include/mcp/server/mcp_server.h
index fad14c70d..8ea0323ef 100644
--- a/include/mcp/server/mcp_server.h
+++ b/include/mcp/server/mcp_server.h
@@ -974,18 +974,6 @@ class McpServer : public application::ApplicationBase,
   void onConnectionEvent(network::ConnectionEvent event);
   void onError(const Error& error) override;
 
-  // Request-scoped transport session binding. The transport filter announces
-  // the transport-level session id (e.g. SSE stream id from the POST
-  // /callback/{id} path) immediately before dispatching each message, on the
-  // dispatcher thread. It is single-use: getOrCreateCurrentSession() consumes
-  // and clears it, so it applies only to the one message whose dispatch it
-  // preceded and a non-announcing producer can never inherit a stale id. An
-  // empty id means the current message's transport has no session concept and
-  // session lookup falls back to connection identity.
-  void onTransportSessionBound(const std::string& transport_session_id) {
-    current_transport_session_id_ = transport_session_id;
-  }
-
   // Request tracking helpers
   bool isRequestCancelled(const RequestId& id) const {
     std::lock_guard<std::mutex> lock(pending_requests_mutex_);
@@ -1135,11 +1123,6 @@ class McpServer : public application::ApplicationBase,
 
     void onError(const Error& error) override { server_.onError(error); }
 
-    void onTransportSessionBound(
-        const std::string& transport_session_id) override {
-      server_.onTransportSessionBound(transport_session_id);
-    }
-
    private:
     McpServer& server_;
   };
@@ -1184,15 +1167,6 @@ class McpServer : public application::ApplicationBase,
   // Following production pattern: listener owns connections, not threads
   // Connections tracked by count only, ownership managed by listener
 
-  // Current connection being processed (for request context)
-  // This is set temporarily during request processing in dispatcher thread
-  network::Connection* current_connection_{nullptr};
-
-  // Transport session id of the message currently being dispatched (see
-  // onTransportSessionBound). Dispatcher thread only, like
-  // current_connection_. Empty for transports without session identity.
-  std::string current_transport_session_id_;
-
   // Active connections owned by server
   // Following production pattern: all operations in dispatcher thread, no mutex
   // needed Connections removed when they close via callbacks
diff --git a/src/filter/http_sse_filter_chain_factory.cc b/src/filter/http_sse_filter_chain_factory.cc
index 52df4562b..dacb5a864 100644
--- a/src/filter/http_sse_filter_chain_factory.cc
+++ b/src/filter/http_sse_filter_chain_factory.cc
@@ -981,14 +981,11 @@ class HttpSseJsonRpcProtocolFilter
    */
   void onRequest(const jsonrpc::Request& request) override {
     GOPHER_LOG_DEBUG("HttpSseFilter::onRequest for method: {}", request.method);
-    // Announce which transport session this message belongs to before
-    // dispatching. On a POST /callback/{id} connection this carries the SSE
-    // stream id — the durable client identity — so the server can key its
-    // MCP session on it instead of this short-lived POST connection. It is
-    // announced per message (not per connection) because dispatcher-thread
-    // reads from different connections interleave; an empty id explicitly
-    // clears any previous connection's binding.
-    mcp_callbacks_.onTransportSessionBound(sse_callback_session_id_);
+    // The context carries the transport session id (the SSE stream id from
+    // a POST /callback/{id} path — the durable client identity) with the
+    // message itself. Built fresh per message because dispatcher-thread
+    // reads from different connections interleave; a previous message's
+    // binding cannot leak because the previous context is already gone.
     DispatchContext context(*this);
     mcp_callbacks_.onRequestWithContext(request, context);
   }
@@ -1011,7 +1008,6 @@ class HttpSseJsonRpcProtocolFilter
   }
 
   void onNotification(const jsonrpc::Notification& notification) override {
-    mcp_callbacks_.onTransportSessionBound(sse_callback_session_id_);
     DispatchContext context(*this);
     mcp_callbacks_.onNotificationWithContext(notification, context);
 
@@ -1038,7 +1034,6 @@ class HttpSseJsonRpcProtocolFilter
   }
 
   void onResponse(const jsonrpc::Response& response) override {
-    mcp_callbacks_.onTransportSessionBound(sse_callback_session_id_);
     mcp_callbacks_.onResponse(response);
   }
 
diff --git a/src/server/mcp_server.cc b/src/server/mcp_server.cc
index d37a44ce7..bc535ea2d 100644
--- a/src/server/mcp_server.cc
+++ b/src/server/mcp_server.cc
@@ -1103,8 +1103,7 @@ void McpServer::onConnectionLifecycleEvent(network::Connection* connection,
     http_sse_factory_->sseRegistry().removeConnection(connection);
   }
 
-  // Tear down session state keyed by the actual closing connection, not a
-  // global current_connection_ which may point at a different session.
+  // Tear down session state keyed by the actual closing connection.
   // Transport-keyed sessions (HTTP+SSE) are untouched here by design: they
   // are created with a null connection, live across many short POST
   // connections, and are released by the SSE registry's session-closed
@@ -1145,10 +1144,6 @@ void McpServer::onConnectionLifecycleEvent(network::Connection* connection,
       lifecycle_callbacks_.erase(cb_it);
     }
   }
-
-  if (current_connection_ == connection) {
-    current_connection_ = nullptr;
-  }
 }
 
 void McpServer::onError(const Error& error) {
@@ -1600,9 +1595,6 @@ void McpServer::onNewConnection(network::ConnectionPtr&& connection) {
     connection_sessions_[conn_ptr] = session;
   }
 
-  // Set current connection for request processing context
-  current_connection_ = conn_ptr;
-
   // Update connection count.
   //
   // We bump the public stats here rather than wait for a Connected event
diff --git a/tests/integration/test_http_sse_filter_server_mode.cc b/tests/integration/test_http_sse_filter_server_mode.cc
index 49ea32742..a474b0f3c 100644
--- a/tests/integration/test_http_sse_filter_server_mode.cc
+++ b/tests/integration/test_http_sse_filter_server_mode.cc
@@ -42,25 +42,30 @@ using namespace std::chrono_literals;
 class ServerModeCallbacks : public McpProtocolCallbacks {
  public:
   void onRequest(const jsonrpc::Request& req) override {
+    // Context-free fallback: nothing traveled with the message. If the
+    // filter ever regresses to this path, the seeded sentinel below
+    // surfaces in binding_at_request_ and fails the assertions.
     requests_.push_back(req);
-    // Capture which transport session id was in effect when this request
-    // was dispatched — the filter's contract is to announce it (possibly
-    // empty) immediately beforehand, per message.
     binding_at_request_.push_back(current_binding_);
   }
+  void onRequestWithContext(const jsonrpc::Request& req,
+                            MessageDispatchContext& context) override {
+    requests_.push_back(req);
+    // Capture the transport session id that traveled WITH this message —
+    // the filter's contract is to build a fresh context per dispatched
+    // message (possibly with an empty id).
+    binding_at_request_.push_back(context.transportSessionId());
+  }
   void onNotification(const jsonrpc::Notification&) override {}
   void onResponse(const jsonrpc::Response&) override {}
   void onConnectionEvent(network::ConnectionEvent) override {}
   void onError(const Error&) override { error_count_++; }
   void onMessageEndpoint(const std::string&) override {}
   bool sendHttpPost(const std::string&) override { return true; }
-  void onTransportSessionBound(const std::string& id) override {
-    current_binding_ = id;
-  }
 
   std::vector<jsonrpc::Request> requests_;
   std::vector<std::string> binding_at_request_;
-  std::string current_binding_{"<never-announced>"};
+  std::string current_binding_{"<context-free-dispatch>"};
   int error_count_{0};
 };
 
@@ -304,7 +309,9 @@ TEST_F(ServerModeFilterTest, PlainPost_AnnouncesEmptyBinding) {
   network::IoHandlePtr peer;
   std::shared_ptr<HttpSseFilterChainFactory> factory;
 
-  // Pretend a callback POST on another connection just dispatched.
+  // Seed the context-free sentinel: if dispatch ever bypasses the
+  // per-message context, this value leaks into binding_at_request_ and
+  // the empty-id assertion below fails.
   callbacks.current_binding_ = "client_from_previous_connection";
 
   executeInDispatcher([&]() {

From ec761ed6e20c9ed25964c4a9b921e26cf02af401 Mon Sep 17 00:00:00 2001
From: gophergogo <gogo@gopher.security>
Date: Mon, 6 Jul 2026 19:37:40 -0700
Subject: [PATCH 5/5] Pin response routing and dispatch-context contracts with
 tests (#264)

New integration suite drives a real server over real TCP sockets with two
concurrent plain-HTTP clients and pins the fixed failure modes: a request
is answered on the connection it arrived on while the other connection
stays silent (the accept-time ambient pointer answered on the most
recently accepted connection instead), an unrelated close no longer drops
a live connection's response, and interleaved rounds keep every
request/response pair on its own connection.

Filter-level tests pin the context contract at the source: a parsed
message dispatches through the context-carrying hook (never the
context-free fallback), an unwired filter reports a null origin and empty
transport session id rather than stale values, and a reply attempt with
no connection surfaces an error instead of silently succeeding.
---
 tests/CMakeLists.txt                          |  11 +
 tests/filter/test_json_rpc_protocol_filter.cc | 102 ++++++
 .../test_mcp_server_request_routing.cc        | 321 ++++++++++++++++++
 3 files changed, 434 insertions(+)
 create mode 100644 tests/integration/test_mcp_server_request_routing.cc

diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 0d429398c..609fa2bf5 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -1668,6 +1668,17 @@ target_link_libraries(test_server_notification_delivery
 )
 add_test(NAME ServerNotificationDeliveryTest COMMAND test_server_notification_delivery)
 
+add_executable(test_mcp_server_request_routing integration/test_mcp_server_request_routing.cc)
+target_link_libraries(test_mcp_server_request_routing
+    gopher-mcp
+    gopher-mcp-event
+    gtest
+    gtest_main
+    gmock
+    Threads::Threads
+)
+add_test(NAME McpServerRequestRoutingTest COMMAND test_mcp_server_request_routing)
+
 add_executable(test_mcp_server_connection_lifecycle integration/test_mcp_server_connection_lifecycle.cc)
 target_link_libraries(test_mcp_server_connection_lifecycle
     gopher-mcp
diff --git a/tests/filter/test_json_rpc_protocol_filter.cc b/tests/filter/test_json_rpc_protocol_filter.cc
index 21fbf8197..024016166 100644
--- a/tests/filter/test_json_rpc_protocol_filter.cc
+++ b/tests/filter/test_json_rpc_protocol_filter.cc
@@ -39,6 +39,49 @@ class MockJsonRpcCallbacks : public JsonRpcProtocolFilter::MessageHandler {
   MOCK_METHOD(void, onProtocolError, (const Error&), (override));
 };
 
+/**
+ * Handler that records the dispatch context contract: the per-message
+ * context must arrive together with the message (context path, not the
+ * context-free fallback), report the filter's actual origin, and refuse
+ * to pretend a reply was sent when there is no connection to send it on.
+ */
+class ContextCapturingHandler : public JsonRpcProtocolFilter::MessageHandler {
+ public:
+  void onRequest(const jsonrpc::Request&) override { legacy_requests_++; }
+  void onNotification(const jsonrpc::Notification&) override {
+    legacy_notifications_++;
+  }
+  void onResponse(const jsonrpc::Response&) override {}
+  void onProtocolError(const Error&) override {}
+
+  void onRequestWithContext(const jsonrpc::Request& request,
+                            MessageDispatchContext& context) override {
+    context_requests_++;
+    last_origin_ = context.originConnection();
+    last_transport_session_id_ = context.transportSessionId();
+    // Try to reply while the context is live; record whether the filter
+    // surfaced the missing reply path as an error.
+    auto result = context.sendResponse(jsonrpc::Response::make_error(
+        request.id, Error(jsonrpc::INTERNAL_ERROR, "test reply")));
+    last_send_failed_ = holds_alternative<Error>(result);
+  }
+
+  void onNotificationWithContext(const jsonrpc::Notification&,
+                                 MessageDispatchContext& context) override {
+    context_notifications_++;
+    last_origin_ = context.originConnection();
+    last_transport_session_id_ = context.transportSessionId();
+  }
+
+  int legacy_requests_{0};
+  int legacy_notifications_{0};
+  int context_requests_{0};
+  int context_notifications_{0};
+  network::Connection* last_origin_{reinterpret_cast<network::Connection*>(1)};
+  std::string last_transport_session_id_{"<unset>"};
+  bool last_send_failed_{false};
+};
+
 /**
  * Test fixture for JsonRpcProtocolFilter using real I/O
  */
@@ -306,6 +349,65 @@ TEST_F(JsonRpcProtocolFilterTest, WriteFilterAddsFraming) {
   });
 }
 
+/**
+ * A parsed request must be dispatched through the context-carrying hook
+ * with a context that travels with the message. This filter is not wired
+ * to a connection, so the contract under test is the honest degraded
+ * form: no origin, no transport session id, and a reply attempt that
+ * FAILS instead of silently succeeding (the pre-context encoder returned
+ * success while writing nothing).
+ */
+TEST_F(JsonRpcProtocolFilterTest, RequestDispatchCarriesContext) {
+  auto handler = std::make_unique<ContextCapturingHandler>();
+  std::unique_ptr<JsonRpcProtocolFilter> filter;
+  executeInDispatcher([&]() {
+    filter = std::make_unique<JsonRpcProtocolFilter>(*handler, *dispatcher_,
+                                                     /*is_server=*/true);
+    OwnedBuffer buffer;
+    buffer.add(
+        std::string(R"({"jsonrpc":"2.0","id":7,"method":"test.method"})") +
+        "\n");
+    filter->onData(buffer, false);
+  });
+
+  EXPECT_EQ(handler->context_requests_, 1)
+      << "request must dispatch through the context-carrying hook";
+  EXPECT_EQ(handler->legacy_requests_, 0)
+      << "context-free fallback must not run when a context was built";
+  EXPECT_EQ(handler->last_origin_, nullptr)
+      << "unwired filter must report no origin, not a stale pointer";
+  EXPECT_EQ(handler->last_transport_session_id_, "")
+      << "bare JSON-RPC chain has no transport session concept";
+  EXPECT_TRUE(handler->last_send_failed_)
+      << "sendResponse with no connection must surface an error, not "
+         "silently drop the reply";
+
+  executeInDispatcher([&]() { filter.reset(); });
+}
+
+/**
+ * Same contract for notifications: context path, not the fallback.
+ */
+TEST_F(JsonRpcProtocolFilterTest, NotificationDispatchCarriesContext) {
+  auto handler = std::make_unique<ContextCapturingHandler>();
+  std::unique_ptr<JsonRpcProtocolFilter> filter;
+  executeInDispatcher([&]() {
+    filter = std::make_unique<JsonRpcProtocolFilter>(*handler, *dispatcher_,
+                                                     /*is_server=*/true);
+    OwnedBuffer buffer;
+    buffer.add(std::string(R"({"jsonrpc":"2.0","method":"notify.method"})") +
+               "\n");
+    filter->onData(buffer, false);
+  });
+
+  EXPECT_EQ(handler->context_notifications_, 1);
+  EXPECT_EQ(handler->legacy_notifications_, 0);
+  EXPECT_EQ(handler->last_origin_, nullptr);
+  EXPECT_EQ(handler->last_transport_session_id_, "");
+
+  executeInDispatcher([&]() { filter.reset(); });
+}
+
 /**
  * Integration test with real connection
  * NOTE: Disabled due to issues with real I/O test infrastructure
diff --git a/tests/integration/test_mcp_server_request_routing.cc b/tests/integration/test_mcp_server_request_routing.cc
new file mode 100644
index 000000000..f0e756675
--- /dev/null
+++ b/tests/integration/test_mcp_server_request_routing.cc
@@ -0,0 +1,321 @@
+/**
+ * Integration test: responses are routed by the request's own dispatch
+ * context, not by ambient most-recently-accepted-connection state.
+ *
+ * Historical failure modes pinned here (all real with the ambient
+ * current-connection scheme, all fixed by the per-message dispatch
+ * context):
+ *
+ *   1. Cross-wired responses. The ambient pointer was stamped at accept
+ *      time, so with two live connections (accept A, accept B) a request
+ *      arriving on A was answered on B: client B received client A's
+ *      response and A hung. The dispatch context pins the reply to the
+ *      connection the request physically arrived on.
+ *
+ *   2. Silent drop after an unrelated close. Closing B nulled the ambient
+ *      pointer even though A was alive and mid-flight, so A's next
+ *      response was skipped without a log line. With the context, A's
+ *      reply path is A's own connection regardless of what B does.
+ *
+ * The tests drive a real McpServer over real TCP sockets with plain HTTP
+ * POSTs (no SSE handshake), because the plain-HTTP path is exactly the one
+ * that used to depend on the ambient pointer for both session keying and
+ * the response write.
+ */
+
+#include <chrono>
+#include <cstdint>
+#include <memory>
+#include <stdexcept>
+#include <string>
+#include <thread>
+
+#include <gtest/gtest.h>
+
+#include "mcp/buffer.h"
+#include "mcp/network/address.h"
+#include "mcp/network/io_handle.h"
+#include "mcp/network/socket_interface.h"
+#include "mcp/server/mcp_server.h"
+#include "mcp/types.h"
+
+namespace mcp {
+namespace {
+
+using namespace std::chrono_literals;
+
+// Bind ephemeral 0 to get a port the kernel thinks is free, then let go
+// of it and hand the number to the server. Same mild TOCTOU as the other
+// integration tests -- accepted on a loopback test bed.
+uint16_t pickEphemeralPort() {
+  auto& iface = network::socketInterface();
+
+  auto fd_result =
+      iface.socket(network::SocketType::Stream, network::Address::Type::Ip,
+                   network::Address::IpVersion::v4);
+  if (!fd_result.ok()) {
+    throw std::runtime_error("pickEphemeralPort: socket() failed");
+  }
+
+  auto handle = iface.ioHandleForFd(*fd_result, /*socket_v6only=*/false);
+  handle->setBlocking(false);
+
+  auto bind_addr = network::Address::parseInternetAddress("127.0.0.1", 0);
+  auto bind_result = handle->bind(bind_addr);
+  if (!bind_result.ok()) {
+    throw std::runtime_error("pickEphemeralPort: bind() failed");
+  }
+
+  auto local_addr_result = handle->localAddress();
+  if (!local_addr_result.ok()) {
+    throw std::runtime_error("pickEphemeralPort: localAddress() failed");
+  }
+
+  const auto* ip =
+      dynamic_cast<const network::Address::Ip*>(local_addr_result->get());
+  if (ip == nullptr) {
+    throw std::runtime_error("pickEphemeralPort: not an IP address");
+  }
+  uint16_t port = ip->port();
+  handle->close();
+  return port;
+}
+
+class McpServerRequestRoutingTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    port_ = pickEphemeralPort();
+
+    server::McpServerConfig config;
+    config.server_name = "routing-test-server";
+    config.server_version = "0.0.1";
+    config.supported_transports = {TransportType::HttpSse};
+    config.num_workers = 1;
+
+    server_ = server::createMcpServer(config);
+    ASSERT_NE(server_, nullptr);
+
+    const std::string listen_address =
+        "http://127.0.0.1:" + std::to_string(port_);
+    auto listen_result = server_->listen(listen_address);
+    ASSERT_TRUE(holds_alternative<std::nullptr_t>(listen_result))
+        << "McpServer::listen failed";
+
+    server_thread_ = std::thread([this]() { server_->run(); });
+
+    ASSERT_TRUE(waitForListenerReady(port_, 5s))
+        << "Server did not begin accepting on port " << port_;
+  }
+
+  void TearDown() override {
+    if (server_) {
+      server_->shutdown();
+    }
+    if (server_thread_.joinable()) {
+      server_thread_.join();
+    }
+    server_.reset();
+  }
+
+  static bool waitForListenerReady(uint16_t port,
+                                   std::chrono::milliseconds budget) {
+    auto& iface = network::socketInterface();
+    auto addr = network::Address::parseInternetAddress("127.0.0.1", port);
+    const auto deadline = std::chrono::steady_clock::now() + budget;
+    while (std::chrono::steady_clock::now() < deadline) {
+      auto fd_result =
+          iface.socket(network::SocketType::Stream, network::Address::Type::Ip,
+                       network::Address::IpVersion::v4);
+      if (fd_result.ok()) {
+        auto handle = iface.ioHandleForFd(*fd_result, false);
+        handle->setBlocking(true);
+        auto connect_result = handle->connect(addr);
+        handle->close();
+        if (connect_result.ok()) {
+          return true;
+        }
+      }
+      std::this_thread::sleep_for(25ms);
+    }
+    return false;
+  }
+
+  network::IoHandlePtr openClient() {
+    auto& iface = network::socketInterface();
+    auto fd_result =
+        iface.socket(network::SocketType::Stream, network::Address::Type::Ip,
+                     network::Address::IpVersion::v4);
+    if (!fd_result.ok()) {
+      return nullptr;
+    }
+    auto handle = iface.ioHandleForFd(*fd_result, /*socket_v6only=*/false);
+    handle->setBlocking(true);
+    auto addr = network::Address::parseInternetAddress("127.0.0.1", port_);
+    auto connect_result = handle->connect(addr);
+    if (!connect_result.ok()) {
+      handle->close();
+      return nullptr;
+    }
+    return handle;
+  }
+
+  // POST a JSON-RPC body to the server's plain-HTTP RPC path on an
+  // already-open client socket.
+  static bool sendRpcPost(network::IoHandle& handle, const std::string& body) {
+    std::string request =
+        "POST /rpc HTTP/1.1\r\n"
+        "Host: localhost\r\n"
+        "Content-Type: application/json\r\n"
+        "Content-Length: " +
+        std::to_string(body.size()) +
+        "\r\n"
+        "\r\n" +
+        body;
+    OwnedBuffer out;
+    out.add(request);
+    return handle.write(out).ok();
+  }
+
+  // Accumulate whatever arrives on the socket until it contains `needle`
+  // or the budget elapses. Returns everything read either way.
+  static std::string readUntilContains(network::IoHandle& handle,
+                                       const std::string& needle,
+                                       std::chrono::milliseconds budget) {
+    handle.setBlocking(false);
+    std::string received;
+    const auto deadline = std::chrono::steady_clock::now() + budget;
+    while (std::chrono::steady_clock::now() < deadline) {
+      OwnedBuffer buf;
+      auto r = handle.read(buf, /*max_length=*/8192);
+      if (r.ok() && *r > 0) {
+        received += buf.toString();
+        if (received.find(needle) != std::string::npos) {
+          return received;
+        }
+      } else if (r.ok() && *r == 0) {
+        return received;  // EOF
+      } else if (!r.wouldBlock()) {
+        return received;  // hard error; caller's assertion reports it
+      }
+      std::this_thread::sleep_for(10ms);
+    }
+    return received;
+  }
+
+  uint16_t port_{0};
+  std::unique_ptr<server::McpServer> server_;
+  std::thread server_thread_;
+};
+
+// Two live connections; the request goes in on the FIRST-accepted one.
+// Under the ambient scheme the reply went out on the most recently
+// accepted connection (B); under the dispatch context it must come back
+// on A, and B must stay silent.
+TEST_F(McpServerRequestRoutingTest, ResponseReturnsOnOriginConnection) {
+  auto client_a = openClient();
+  ASSERT_NE(client_a, nullptr);
+  auto client_b = openClient();
+  ASSERT_NE(client_b, nullptr);
+
+  // Give the dispatcher a beat to accept both, in order, so B is the
+  // most recently accepted connection when A's request dispatches —
+  // exactly the interleaving the ambient pointer got wrong.
+  std::this_thread::sleep_for(100ms);
+
+  ASSERT_TRUE(
+      sendRpcPost(*client_a, R"({"jsonrpc":"2.0","id":41,"method":"ping"})"));
+
+  std::string on_a = readUntilContains(*client_a, "\"id\":41", 3s);
+  EXPECT_NE(on_a.find("\"id\":41"), std::string::npos)
+      << "Ping response did not return on the connection that sent the "
+         "request; got instead: "
+      << on_a;
+
+  // B must not have received A's response. A short read window is
+  // enough: the response already proved the round trip completed.
+  std::string on_b = readUntilContains(*client_b, "\"id\":41", 200ms);
+  EXPECT_EQ(on_b.find("\"id\":41"), std::string::npos)
+      << "Another connection received this request's response: " << on_b;
+
+  // And the reverse direction: B's own request comes back on B.
+  ASSERT_TRUE(
+      sendRpcPost(*client_b, R"({"jsonrpc":"2.0","id":42,"method":"ping"})"));
+  std::string on_b2 = readUntilContains(*client_b, "\"id\":42", 3s);
+  EXPECT_NE(on_b2.find("\"id\":42"), std::string::npos)
+      << "Second connection's response did not return to it; got: " << on_b2;
+
+  client_a->close();
+  client_b->close();
+}
+
+// Closing an unrelated connection must not affect another connection's
+// request/response cycle. Under the ambient scheme, B's close nulled the
+// shared pointer and A's response was silently skipped.
+TEST_F(McpServerRequestRoutingTest, UnrelatedCloseDoesNotDropResponse) {
+  auto client_a = openClient();
+  ASSERT_NE(client_a, nullptr);
+  auto client_b = openClient();
+  ASSERT_NE(client_b, nullptr);
+
+  // B is the most recently accepted connection; its close is the one the
+  // ambient pointer used to track.
+  std::this_thread::sleep_for(100ms);
+  client_b->close();
+  client_b.reset();
+  std::this_thread::sleep_for(100ms);
+
+  ASSERT_TRUE(
+      sendRpcPost(*client_a, R"({"jsonrpc":"2.0","id":43,"method":"ping"})"));
+
+  std::string on_a = readUntilContains(*client_a, "\"id\":43", 3s);
+  EXPECT_NE(on_a.find("\"id\":43"), std::string::npos)
+      << "Response was dropped after an unrelated connection closed; "
+         "received: "
+      << on_a;
+
+  client_a->close();
+}
+
+// Several requests interleaved across two live connections, each keeping
+// its own request/response pairing throughout — the steady-state version
+// of the two tests above.
+TEST_F(McpServerRequestRoutingTest, InterleavedRequestsKeepTheirConnections) {
+  auto client_a = openClient();
+  ASSERT_NE(client_a, nullptr);
+  auto client_b = openClient();
+  ASSERT_NE(client_b, nullptr);
+  std::this_thread::sleep_for(100ms);
+
+  for (int i = 0; i < 3; ++i) {
+    const int id_a = 100 + i;
+    const int id_b = 200 + i;
+
+    ASSERT_TRUE(sendRpcPost(*client_a, R"({"jsonrpc":"2.0","id":)" +
+                                           std::to_string(id_a) +
+                                           R"(,"method":"ping"})"));
+    ASSERT_TRUE(sendRpcPost(*client_b, R"({"jsonrpc":"2.0","id":)" +
+                                           std::to_string(id_b) +
+                                           R"(,"method":"ping"})"));
+
+    std::string needle_a = "\"id\":" + std::to_string(id_a);
+    std::string needle_b = "\"id\":" + std::to_string(id_b);
+
+    std::string on_a = readUntilContains(*client_a, needle_a, 3s);
+    EXPECT_NE(on_a.find(needle_a), std::string::npos)
+        << "round " << i << ": A's response missing on A; got: " << on_a;
+    EXPECT_EQ(on_a.find(needle_b), std::string::npos)
+        << "round " << i << ": B's response leaked onto A: " << on_a;
+
+    std::string on_b = readUntilContains(*client_b, needle_b, 3s);
+    EXPECT_NE(on_b.find(needle_b), std::string::npos)
+        << "round " << i << ": B's response missing on B; got: " << on_b;
+    EXPECT_EQ(on_b.find(needle_a), std::string::npos)
+        << "round " << i << ": A's response leaked onto B: " << on_b;
+  }
+
+  client_a->close();
+  client_b->close();
+}
+
+}  // namespace
+}  // namespace mcp