diff --git a/CHANGELOG.md b/CHANGELOG.md index 872625786..1d3c1b353 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,171 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +## [0.1.14] - 2026-07-22 + +### Fixed + +- Fix Streamable HTTP tool-call body parsing for pretty-printed JSON, newline-delimited JSON fallback, and malformed body error reporting. +- Propagate `Mcp-Session-Id` from Streamable HTTP requests as the transport session id. +- Bound reconnect readiness polling so it cannot consume the full request timeout. + +## [0.1.13] - 2026-07-08 + +### Added + +- Add MCP HTTP header passthrough support (#250). + +### Fixed + +- Fix SSL transport use-after-free in posted lambdas (#245). +- Null-guard OpenSSL 3.x error-string accessors to prevent crash (#246). +- Fix SSL transport data loss on full network BIO (#247). +- Route MCP responses to the originating connection (#248). +- Improve transport and MCP flow logging (#249). +- Skip TLS peer metadata when verification is disabled (#251). +- Avoid retaining streamed HTTP client bodies (#260). +- Guard oversized HTTP body callbacks (#255). +- Defer HTTP parser callback errors (#256). +- Post HTTP parser error callbacks (#252). +- Defer active connection destruction on close (#257). +- Bind llhttp symbols inside the shared library (#259). +- Honor preferred HTTP transport (#253). + +## [0.1.12] - 2026-07-03 + +### Added + +- Add MCP HTTP header passthrough support. +- Add MCP client and HTTP transport invoke logging controlled by `GOPHER_LOG_LEVEL`. + +## [0.1.11] - 2026-07-02 + +### Changed + +- Bind llhttp symbols inside the shared library. +- Defer active connection destruction on close. +- Post and defer HTTP parser error callbacks safely. +- Guard oversized HTTP body callbacks. +- Avoid retaining streamed HTTP client bodies. +- Skip TLS peer metadata when verification is disabled. + +### Fixed + +- Fix SSL transport data loss on full network BIO. +- Fix SSL transport use-after-free in posted lambdas. + +## [0.1.10] - 2026-06-30 + +### Added + +- Add MCP HTTP header passthrough support. +- Add MCP client and HTTP transport invoke logging controlled by `GOPHER_LOG_LEVEL`. + +## [0.1.9] - 2026-06-30 + +### Added + +- Add MCP client and HTTP transport invoke logging controlled by `GOPHER_LOG_LEVEL`. + +### Changed + +- Route MCP responses to the originating connection to fix concurrent request hangs. + +### Fixed + +- Fix SSL transport data loss on full network BIO. +- Fix SSL transport use-after-free in posted lambdas. + +## [0.1.8] - 2026-06-24 + +### Added + +- Add MCP client and HTTP transport invoke logging controlled by `GOPHER_LOG_LEVEL`. + +### Changed + +- Scope MCP flow logging to a dedicated `GOPHER_MCP_LOG_FLOW` switch. +- Demote filter registry initialization and registration logs to debug level. +- Surface SSL errors through warning logs. +- Null-guard OpenSSL 3.x error-string accessors. + +### Fixed + +- Fix SSL transport data loss on full network BIO. +- Fix SSL transport use-after-free in posted lambdas. + +## [0.1.7] - 2026-06-22 + +### Changed + +- Demote filter registry initialization and registration logs to debug level. +- Surface SSL errors through warning logs. +- Null-guard OpenSSL 3.x error-string accessors. + +### Fixed + +- Fix SSL transport use-after-free in posted lambdas. + +## [0.1.6] - 2026-06-11 + +### Added + +- Add request-scoped `_meta` storage to `SessionContext` for tool handlers. +- Add client-side notification handler registration to `McpClient`. +- Add server and client SSE state machines with integration coverage. +- Add runtime formatting support for dynamic format strings. + +### Changed + +- Populate `ReadResourceResult` contents in `McpClient::readResource`. +- Integrate SSE connection state machines into `HttpSseJsonRpcProtocolFilter`. +- Remove obsolete request-stream and SSE mode state from the HTTP/SSE filter. +- Improve MSVC build configuration. +- Update README architecture documentation. + +### Fixed + +- Fix `ConnectionPoolImpl` timeout crash from premature write events. + +## [0.1.5] - 2026-04-21 + +### Added + +- Add idle-read timeout handling to `ConnectionImpl` and `McpServer`. +- Add SSE server transport support with a per-factory session registry. +- Add configurable SSE/RPC paths and external URL parameters to the filter chain factory. +- Add `HttpAsyncClient` built on `HttpCodecFilter`. +- Add integration and lifecycle tests for HTTP, SSE, and connection cleanup behavior. + +### Changed + +- Match `/callback/{id}` requests under reverse-proxy path prefixes. +- Rename the default SSE path to `/sse`. +- Surface numeric `:status` pseudo-headers from the HTTP client codec. +- Disable body timeout for client-mode HTTP codec. +- Defer closed connection destruction through dispatcher cleanup. +- Drain active server connections during shutdown. + +### Fixed + +- Fix background-task timer lifetime in `McpServer`. +- Fix callback scheduling so deferred callbacks run after caller stack unwinds. + +## [0.1.4] - 2026-04-08 + +### Added + +- Add resource read handlers to example server registrations. +- Add `ResourceReadHandler` callback support to `ResourceManager`. +- Add tests for `resources/read` responses and resource manager handlers. + +### Changed + +- Run formatting on resource read implementation files. + +### Fixed + +- Fix `resources/read` responses to match the MCP schema. ## [0.1.1] - 2026-03-03 diff --git a/CMakeLists.txt b/CMakeLists.txt index 2da2179e9..002f20e37 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -15,7 +15,7 @@ Please create a build directory and run cmake from there: You may need to remove CMakeCache.txt and CMakeFiles/") endif() -project(gopher-mcp VERSION 0.1.1 LANGUAGES C CXX) +project(gopher-mcp VERSION 0.1.14 LANGUAGES C CXX) # Set library version for shared libraries set(GOPHER_MCP_VERSION_MAJOR ${PROJECT_VERSION_MAJOR}) diff --git a/include/mcp/client/mcp_client.h b/include/mcp/client/mcp_client.h index ef818bdf5..a9c5ed965 100644 --- a/include/mcp/client/mcp_client.h +++ b/include/mcp/client/mcp_client.h @@ -433,6 +433,8 @@ class McpClient : public application::ApplicationBase { void disconnect(); bool isConnected() const { return connected_; } bool isConnectionOpen() const; // Check actual connection state + static std::chrono::milliseconds reconnectWaitBudgetForRequestTimeout( + std::chrono::milliseconds request_timeout); // Shutdown the client (stops workers and event loop) void shutdown() override; diff --git a/include/mcp/filter/http_routing_filter.h b/include/mcp/filter/http_routing_filter.h index 4e4b0f606..a7bc7450d 100644 --- a/include/mcp/filter/http_routing_filter.h +++ b/include/mcp/filter/http_routing_filter.h @@ -137,6 +137,7 @@ class HttpRoutingFilter : public HttpCodecFilter::MessageCallbacks { // State for POST requests that need body bool pending_post_request_ = false; + bool suppress_current_request_ = false; RequestContext pending_context_; HandlerFunc pending_handler_; std::string accumulated_body_; @@ -164,4 +165,4 @@ class HttpRoutingFilterFactory { } // namespace filter } // namespace mcp -#endif // MCP_FILTER_HTTP_ROUTING_FILTER_H \ No newline at end of file +#endif // MCP_FILTER_HTTP_ROUTING_FILTER_H diff --git a/include/mcp/filter/json_rpc_protocol_filter.h b/include/mcp/filter/json_rpc_protocol_filter.h index 32f06650f..3ec877252 100644 --- a/include/mcp/filter/json_rpc_protocol_filter.h +++ b/include/mcp/filter/json_rpc_protocol_filter.h @@ -192,6 +192,13 @@ class JsonRpcProtocolFilter : public network::Filter { */ bool parseMessage(const std::string& json_str); + /** + * Dispatch a parsed JSON-RPC message + * @param json_val Parsed JSON value + * @return True if message shape was valid + */ + bool dispatchMessage(const json::JsonValue& json_val); + /** * Frame outgoing message with length prefix if needed * @param data Buffer containing message to frame diff --git a/src/client/mcp_client.cc b/src/client/mcp_client.cc index c574d7c8a..a7bb08722 100644 --- a/src/client/mcp_client.cc +++ b/src/client/mcp_client.cc @@ -311,6 +311,12 @@ bool McpClient::isConnectionOpen() const { return connection_manager_->isConnected(); } +std::chrono::milliseconds McpClient::reconnectWaitBudgetForRequestTimeout( + std::chrono::milliseconds request_timeout) { + return std::min(std::max(request_timeout / 3, std::chrono::milliseconds(250)), + std::chrono::milliseconds(5000)); +} + // Reconnect using stored URI VoidResult McpClient::reconnect() { if (current_uri_.empty()) { @@ -744,10 +750,16 @@ void McpClient::sendRequestInternal(std::shared_ptr context) { "is_stale={}", idle_seconds, kConnectionIdleTimeoutSec, is_stale); - // Check if connection is stale or not open - need to reconnect - // Maximum retries to wait for connection after reconnect (50 * 10ms = 500ms - // max) - static constexpr int kMaxReconnectRetries = 50; + // Check if connection is stale or not open - need to reconnect. + // + // Reconnect readiness is driven by dispatcher I/O and can take several + // seconds for remote HTTPS/SSE backends, but it must leave request-deadline + // headroom for the actual send and response. + static constexpr int kReconnectRetryDelayMs = 10; + const auto reconnect_wait_budget = + reconnectWaitBudgetForRequestTimeout(config_.request_timeout); + const auto max_reconnect_retries = static_cast(std::max( + 1, reconnect_wait_budget.count() / kReconnectRetryDelayMs)); // THREAD SAFETY: Use atomic connected_ flag instead of isConnectionOpen() // isConnectionOpen() reads McpConnectionManager::active_connection_ without @@ -756,7 +768,7 @@ void McpClient::sendRequestInternal(std::shared_ptr context) { if (is_stale || !connected_) { // Track if this is a retry after reconnect if (context->retry_count > 0 && - context->retry_count <= kMaxReconnectRetries) { + context->retry_count <= max_reconnect_retries) { // This is a retry - check if we're connected now if (!connected_) { // Still not connected, schedule another retry with timer delay @@ -765,11 +777,12 @@ void McpClient::sendRequestInternal(std::shared_ptr context) { context->retry_count++; context->retry_timer = main_dispatcher_->createTimer( [this, context]() { sendRequestInternal(context); }); - context->retry_timer->enableTimer(std::chrono::milliseconds(10)); + context->retry_timer->enableTimer( + std::chrono::milliseconds(kReconnectRetryDelayMs)); return; } // Connected now, proceed with send below - } else if (context->retry_count > kMaxReconnectRetries) { + } else if (context->retry_count > max_reconnect_retries) { // Too many retries context->promise.set_value(Response::make_error( context->id, Error(::mcp::jsonrpc::INTERNAL_ERROR, diff --git a/src/filter/http_routing_filter.cc b/src/filter/http_routing_filter.cc index a757d1d43..e3873fce1 100644 --- a/src/filter/http_routing_filter.cc +++ b/src/filter/http_routing_filter.cc @@ -109,18 +109,41 @@ void HttpRoutingFilter::onHeaders( if (resp.status_code != 0) { // Handler wants to handle this - send response immediately // This is appropriate for endpoints that don't need the body + suppress_current_request_ = true; sendResponse(resp); return; // Don't forward to next layer } } - // No handler or handler returned 0 - pass through + // No registered handler matched, or the matched handler explicitly returned + // status 0. Give the default handler a chance to answer before forwarding to + // the protocol layer. Transport paths can still opt into pass-through by + // returning status 0 from the default handler. + { + RequestContext ctx; + ctx.method = method; + ctx.path = full_url; + ctx.headers = headers; + ctx.keep_alive = keep_alive; + Response resp = default_handler_(ctx); + if (resp.status_code != 0) { + suppress_current_request_ = true; + sendResponse(resp); + return; // Handled or rejected; do not forward to next layer. + } + } + + // Default handler signalled pass-through - forward to next layer if (next_callbacks_) { next_callbacks_->onHeaders(headers, keep_alive); } } void HttpRoutingFilter::onBody(const std::string& data, bool end_stream) { + if (suppress_current_request_) { + return; + } + // If we're accumulating body for a POST handler, buffer it if (pending_post_request_) { accumulated_body_ += data; @@ -136,6 +159,11 @@ void HttpRoutingFilter::onBody(const std::string& data, bool end_stream) { void HttpRoutingFilter::onMessageComplete() { GOPHER_LOG_DEBUG("HttpRoutingFilter::onMessageComplete called"); + if (suppress_current_request_) { + suppress_current_request_ = false; + return; + } + // If we have a pending POST request, now we have the complete body if (pending_post_request_) { pending_context_.body = accumulated_body_; @@ -292,4 +320,4 @@ std::shared_ptr HttpRoutingFilterFactory::createWithHandlers( } } // namespace filter -} // namespace mcp \ No newline at end of file +} // namespace mcp diff --git a/src/filter/http_sse_filter_chain_factory.cc b/src/filter/http_sse_filter_chain_factory.cc index 60c398a9f..9d5077a35 100644 --- a/src/filter/http_sse_filter_chain_factory.cc +++ b/src/filter/http_sse_filter_chain_factory.cc @@ -771,6 +771,11 @@ class HttpSseJsonRpcProtocolFilter server_mode_->currentMode() == ServerConnMode::Undetermined) { server_mode_->handleEvent(ServerConnEvent::PlainHttpDetected); } + streamable_http_session_id_.clear(); + auto session_it = headers.find("mcp-session-id"); + if (session_it != headers.end()) { + streamable_http_session_id_ = session_it->second; + } auto accept = headers.find("accept"); if (accept != headers.end() && accept->second.find("text/event-stream") != std::string::npos) { @@ -797,11 +802,19 @@ class HttpSseJsonRpcProtocolFilter } void onBody(const std::string& data, bool end_stream) override { + GOPHER_LOG_FLOW_DEBUG( + "HTTP/SSE server body received mode={} bytes={} end_stream={}", + server_mode_ + ? ServerConnectionMode::getModeName(server_mode_->currentMode()) + : "", + data.size(), end_stream ? "true" : "false"); + // The long-lived GET /sse request has no request body — but if the // codec surfaces any trailing bytes we don't want to push them down // into the JSON-RPC parser. Ignore bodies on the SSE stream // connection entirely. if (server_mode_ && server_mode_->isSseStream()) { + GOPHER_LOG_FLOW_DEBUG("HTTP/SSE server body ignored on SSE stream"); return; } // Server receives JSON-RPC in request body regardless of SSE mode @@ -810,6 +823,9 @@ class HttpSseJsonRpcProtocolFilter // Server always receives JSON-RPC in request body pending_json_data_.add(data); if (end_stream) { + GOPHER_LOG_FLOW_DEBUG( + "HTTP/SSE server forwarding JSON-RPC body bytes={} to parser", + pending_json_data_.length()); jsonrpc_filter_->onData(pending_json_data_, true); pending_json_data_.drain(pending_json_data_.length()); } @@ -842,11 +858,18 @@ class HttpSseJsonRpcProtocolFilter } void onMessageComplete() override { + GOPHER_LOG_FLOW_DEBUG( + "HTTP/SSE message complete sse_active={} pending_json_bytes={}", + isSseActive() ? "true" : "false", pending_json_data_.length()); + // HTTP message complete — flush any remaining JSON-RPC data that // was not yet processed. In SSE mode the data flows through the // SSE codec instead, so we only flush for non-SSE connections. if (!isSseActive() && pending_json_data_.length() > 0) { // Process any remaining JSON-RPC data + GOPHER_LOG_FLOW_DEBUG( + "HTTP/SSE message complete forwarding pending JSON-RPC bytes={}", + pending_json_data_.length()); jsonrpc_filter_->onData(pending_json_data_, true); pending_json_data_.drain(pending_json_data_.length()); } @@ -933,11 +956,11 @@ 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. + * Per-message dispatch context for messages decoded on this composite chain. + * Origin is the connection the message physically arrived on. The transport + * session id is the durable client identity for transports where logical MCP + * sessions span short-lived HTTP connections: the callback id parsed from + * POST /callback/{id}, or Mcp-Session-Id from Streamable HTTP POST /mcp. * * The reply sink writes the bare JSON to the origin connection, exactly * the bytes the server used to write to its ambient current-connection @@ -957,7 +980,10 @@ class HttpSseJsonRpcProtocolFilter } const std::string& transportSessionId() const override { - return parent_.sse_callback_session_id_; + if (!parent_.sse_callback_session_id_.empty()) { + return parent_.sse_callback_session_id_; + } + return parent_.streamable_http_session_id_; } VoidResult sendResponse(const jsonrpc::Response& response) override { @@ -1180,10 +1206,16 @@ class HttpSseJsonRpcProtocolFilter return resp; }); - // Default handler - handle OPTIONS for CORS preflight on any path, - // pass through other methods to MCP protocol handling + // Default handler - handle OPTIONS for CORS preflight on any path, pass + // through the real MCP transport paths to protocol handling, and return a + // definitive 404 for everything else. Unknown non-RPC paths must get an + // immediate response; otherwise they fall through to a protocol layer that + // has no request to answer and the connection can wait until client + // timeout. + const std::string rpc_path = configured_rpc_path_; + const std::string sse_path = configured_sse_path_; routing_filter_->registerDefaultHandler( - [](const HttpRoutingFilter::RequestContext& req) { + [rpc_path, sse_path](const HttpRoutingFilter::RequestContext& req) { // Handle OPTIONS for CORS preflight on any path if (req.method == "OPTIONS") { HttpRoutingFilter::Response resp; @@ -1197,9 +1229,30 @@ class HttpSseJsonRpcProtocolFilter resp.headers["Content-Length"] = "0"; return resp; } - // Return status 0 to indicate pass-through for MCP endpoints + + std::string path = req.path; + auto query_start = path.find('?'); + if (query_start != std::string::npos) { + path = path.substr(0, query_start); + } + + const bool is_transport_path = + path == rpc_path || path == sse_path || path == "/rpc" || + path == "/events" || path == "/mcp/events" || + (req.method == "POST" && + path.find("/callback/") != std::string::npos); + if (is_transport_path) { + HttpRoutingFilter::Response resp; + resp.status_code = 0; + return resp; + } + HttpRoutingFilter::Response resp; - resp.status_code = 0; + resp.status_code = 404; + resp.headers["content-type"] = "application/json"; + resp.headers["Access-Control-Allow-Origin"] = "*"; + resp.body = R"({"error":"not_found"})"; + resp.headers["content-length"] = std::to_string(resp.body.length()); return resp; }); @@ -1263,6 +1316,10 @@ class HttpSseJsonRpcProtocolFilter // registered under this ID instead of writing back to the POST // connection. std::string sse_callback_session_id_; + // Session ID from Mcp-Session-Id on Streamable HTTP requests. This is the + // durable request-session identity for POST /mcp clients that do not use the + // SSE callback path. + std::string streamable_http_session_id_; // Messages queued during SSE endpoint negotiation (client mode only). // Drained once the state machine reaches EndpointReceived. diff --git a/src/filter/json_rpc_protocol_filter.cc b/src/filter/json_rpc_protocol_filter.cc index 349ec86e9..b5c7fb2a7 100644 --- a/src/filter/json_rpc_protocol_filter.cc +++ b/src/filter/json_rpc_protocol_filter.cc @@ -9,6 +9,9 @@ #include "mcp/filter/json_rpc_protocol_filter.h" +#include +#include + #include "mcp/json/json_serialization.h" #include "mcp/logging/log_macros.h" #include "mcp/mcp_connection_manager.h" @@ -298,6 +301,77 @@ network::FilterStatus JsonRpcProtocolFilter::onData(Buffer& data, bool end_stream) { GOPHER_LOG_TRACE("onData called - buffer size: {}, end_stream: {}", data.length(), end_stream); + GOPHER_LOG_FLOW_DEBUG( + "JSON-RPC parser received bytes={} end_stream={} partial_before={}", + data.length(), end_stream ? "true" : "false", partial_message_.size()); + + // Streamable HTTP POST bodies are complete JSON documents. They may be + // pretty-printed and contain whitespace newlines, which are not message + // delimiters in HTTP body mode. Prefer parsing the full end-of-stream body as + // one JSON value; fall back to newline-delimited parsing when it is not a + // single JSON document, preserving stream/NDJSON behavior. + if (end_stream && !use_framing_) { + std::string body = partial_message_ + data.toString(); + data.drain(data.length()); + + const auto first = body.find_first_not_of(" \t\r\n"); + const auto last = body.find_last_not_of(" \t\r\n"); + if (first != std::string::npos) { + std::string trimmed = body.substr(first, last - first + 1); + try { + auto json_val = json::JsonValue::parse(trimmed); + GOPHER_LOG_FLOW_DEBUG( + "JSON-RPC parser treating end_stream body as single message " + "bytes={}", + trimmed.size()); + dispatchMessage(json_val); + partial_message_.clear(); + return network::FilterStatus::Continue; + } catch (const json::JsonException& e) { + std::vector ndjson_messages; + bool valid_ndjson = true; + std::istringstream lines(trimmed); + std::string line; + while (std::getline(lines, line)) { + const auto line_first = line.find_first_not_of(" \t\r"); + if (line_first == std::string::npos) { + continue; + } + const auto line_last = line.find_last_not_of(" \t\r"); + const std::string line_trimmed = + line.substr(line_first, line_last - line_first + 1); + try { + ndjson_messages.push_back(json::JsonValue::parse(line_trimmed)); + } catch (const json::JsonException&) { + valid_ndjson = false; + break; + } + } + + if (valid_ndjson && !ndjson_messages.empty()) { + GOPHER_LOG_FLOW_DEBUG( + "JSON-RPC parser treating end_stream body as NDJSON messages " + "count={}", + ndjson_messages.size()); + for (const auto& json_val : ndjson_messages) { + dispatchMessage(json_val); + } + } else { + Error error; + error.code = jsonrpc::PARSE_ERROR; + error.message = "JSON parse error: " + std::string(e.what()); + protocol_errors_++; + handler_.onProtocolError(error); + } + partial_message_.clear(); + return network::FilterStatus::Continue; + } + } else { + partial_message_.clear(); + return network::FilterStatus::Continue; + } + } + // Parse JSON-RPC messages from the data buffer // This data has already been processed by lower protocol layers (HTTP/SSE) parseMessages(data); @@ -306,6 +380,9 @@ network::FilterStatus JsonRpcProtocolFilter::onData(Buffer& data, // message This handles HTTP requests where the body doesn't end with a // newline if (end_stream && !partial_message_.empty() && !use_framing_) { + GOPHER_LOG_FLOW_DEBUG( + "JSON-RPC parser flushing partial message bytes={} on end_stream", + partial_message_.size()); parseMessage(partial_message_); partial_message_.clear(); } @@ -382,15 +459,42 @@ void JsonRpcProtocolFilter::parseMessages(Buffer& buffer) { bool JsonRpcProtocolFilter::parseMessage(const std::string& json_str) { GOPHER_LOG_TRACE("JsonRpcProtocolFilter attempting to parse: {}", json_str); + GOPHER_LOG_FLOW_DEBUG("JSON-RPC parser parsing message bytes={}", + json_str.size()); try { // Parse JSON string auto json_val = json::JsonValue::parse(json_str); + return dispatchMessage(json_val); + + } catch (const json::JsonException& e) { + // JSON parse error + Error error; + error.code = jsonrpc::PARSE_ERROR; + error.message = "JSON parse error: " + std::string(e.what()); + protocol_errors_++; + handler_.onProtocolError(error); + return false; + } catch (const std::exception& e) { + // Other errors + Error error; + error.code = jsonrpc::INTERNAL_ERROR; + error.message = "Internal error: " + std::string(e.what()); + protocol_errors_++; + handler_.onProtocolError(error); + return false; + } +} + +bool JsonRpcProtocolFilter::dispatchMessage(const json::JsonValue& json_val) { + try { // Determine message type and dispatch to callbacks if (json_val.contains("method")) { if (json_val.contains("id")) { // JSON-RPC Request jsonrpc::Request request = json::from_json(json_val); + GOPHER_LOG_FLOW_DEBUG("JSON-RPC parser dispatch request method={}", + request.method); GOPHER_LOG_DEBUG("JsonRpcFilter dispatching request for method: {}", request.method); requests_received_++; @@ -400,6 +504,8 @@ bool JsonRpcProtocolFilter::parseMessage(const std::string& json_str) { // JSON-RPC Notification jsonrpc::Notification notification = json::from_json(json_val); + GOPHER_LOG_FLOW_DEBUG("JSON-RPC parser dispatch notification method={}", + notification.method); notifications_received_++; DispatchContextImpl context(*this); handler_.onNotificationWithContext(notification, context); @@ -424,7 +530,6 @@ bool JsonRpcProtocolFilter::parseMessage(const std::string& json_str) { return true; } catch (const json::JsonException& e) { - // JSON parse error Error error; error.code = jsonrpc::PARSE_ERROR; error.message = "JSON parse error: " + std::string(e.what()); @@ -433,7 +538,6 @@ bool JsonRpcProtocolFilter::parseMessage(const std::string& json_str) { return false; } catch (const std::exception& e) { - // Other errors Error error; error.code = jsonrpc::INTERNAL_ERROR; error.message = "Internal error: " + std::string(e.what()); diff --git a/src/filter/request_logger_filter.cc b/src/filter/request_logger_filter.cc index 2da7bdeb2..4b0f10158 100644 --- a/src/filter/request_logger_filter.cc +++ b/src/filter/request_logger_filter.cc @@ -128,7 +128,8 @@ void RequestLoggerFilter::onRequest(const jsonrpc::Request& request) { if (next_callbacks_) { next_callbacks_->onRequest(request); } else { - std::cout << "⚠️ [RequestLogger] No next handler registered!" << std::endl; + std::cout << "⚠️ [RequestLogger] No next handler registered!" + << std::endl; } } @@ -139,7 +140,8 @@ void RequestLoggerFilter::onRequestWithContext( if (next_callbacks_) { next_callbacks_->onRequestWithContext(request, context); } else { - std::cout << "⚠️ [RequestLogger] No next handler registered!" << std::endl; + std::cout << "⚠️ [RequestLogger] No next handler registered!" + << std::endl; } } @@ -197,7 +199,8 @@ void RequestLoggerFilter::onResponse(const jsonrpc::Response& response) { if (next_callbacks_) { next_callbacks_->onResponse(response); } else { - std::cout << "⚠️ [RequestLogger] No next handler registered!" << std::endl; + std::cout << "⚠️ [RequestLogger] No next handler registered!" + << std::endl; } } @@ -221,7 +224,8 @@ void RequestLoggerFilter::onProtocolError(const Error& error) { if (next_callbacks_) { next_callbacks_->onProtocolError(error); } else { - std::cout << "⚠️ [RequestLogger] No next handler registered!" << std::endl; + std::cout << "⚠️ [RequestLogger] No next handler registered!" + << std::endl; } } diff --git a/src/server/mcp_server.cc b/src/server/mcp_server.cc index 867ac47f4..e3619664e 100644 --- a/src/server/mcp_server.cc +++ b/src/server/mcp_server.cc @@ -870,6 +870,11 @@ void McpServer::onRequestWithContext(const jsonrpc::Request& request, MessageDispatchContext& context) { GOPHER_LOG_DEBUG("McpServer::onRequest called with method: {}", request.method); + GOPHER_LOG_FLOW_DEBUG( + "MCP server dispatch request method={} transport_session={}", + request.method, + context.transportSessionId().empty() ? "" + : context.transportSessionId()); // Handle request in dispatcher context - already in dispatcher server_stats_.requests_total++; @@ -1016,6 +1021,12 @@ void McpServer::onNotification(const jsonrpc::Notification& notification) { void McpServer::onNotificationWithContext( const jsonrpc::Notification& notification, MessageDispatchContext& context) { + GOPHER_LOG_FLOW_DEBUG( + "MCP server dispatch notification method={} transport_session={}", + notification.method, + context.transportSessionId().empty() ? "" + : context.transportSessionId()); + // Handle notification in dispatcher context server_stats_.notifications_total++; diff --git a/tests/client/test_client_reconnection_and_logging.cc b/tests/client/test_client_reconnection_and_logging.cc index ee3fe6fe5..1ccaca9bf 100644 --- a/tests/client/test_client_reconnection_and_logging.cc +++ b/tests/client/test_client_reconnection_and_logging.cc @@ -297,7 +297,7 @@ TEST_F(ClientReconnectionTest, RetryCountCanBeIncremented) { context.retry_count++; EXPECT_EQ(context.retry_count, 2); - // Can increment up to kMaxReconnectRetries (50) + // Retry count remains ordinary mutable request context state. context.retry_count = 50; EXPECT_EQ(context.retry_count, 50); } @@ -365,22 +365,29 @@ TEST_F(ClientReconnectionTest, RetryTimerDelayIs10Milliseconds) { EXPECT_LE(elapsed, 50); // But not more than 50ms } -// Test maximum retry count is 50 -TEST_F(ClientReconnectionTest, MaximumRetryCountIs50) { - // The code defines kMaxReconnectRetries = 50 - // This gives 50 * 10ms = 500ms maximum retry time - - using namespace jsonrpc; - RequestId id = 1; - RequestContext context(id, "test.method"); - - // Simulate reaching max retries - context.retry_count = 50; - EXPECT_EQ(context.retry_count, 50); +TEST_F(ClientReconnectionTest, ReconnectWaitBudgetLeavesRequestHeadroom) { + EXPECT_EQ(McpClient::reconnectWaitBudgetForRequestTimeout( + std::chrono::milliseconds(30000)), + std::chrono::milliseconds(5000)) + << "Default request timeout should not be fully consumed by reconnect " + "readiness polling"; + EXPECT_EQ(McpClient::reconnectWaitBudgetForRequestTimeout( + std::chrono::milliseconds(15000)), + std::chrono::milliseconds(5000)); + EXPECT_EQ(McpClient::reconnectWaitBudgetForRequestTimeout( + std::chrono::milliseconds(5000)), + std::chrono::milliseconds(1666)); +} - // At 51 retries, the code should fail the request - context.retry_count = 51; - EXPECT_GT(context.retry_count, 50); +TEST_F(ClientReconnectionTest, ReconnectWaitBudgetHonorsShortTimeouts) { + EXPECT_EQ(McpClient::reconnectWaitBudgetForRequestTimeout( + std::chrono::milliseconds(1000)), + std::chrono::milliseconds(333)); + EXPECT_EQ(McpClient::reconnectWaitBudgetForRequestTimeout( + std::chrono::milliseconds(100)), + std::chrono::milliseconds(250)) + << "Very small request timeouts get only a small readiness floor, not " + "the old multi-second minimum"; } // ============================================================================ diff --git a/tests/filter/test_http_routing_filter_simple.cc b/tests/filter/test_http_routing_filter_simple.cc index eae739b10..9e9353320 100644 --- a/tests/filter/test_http_routing_filter_simple.cc +++ b/tests/filter/test_http_routing_filter_simple.cc @@ -182,6 +182,132 @@ TEST_F(HttpRoutingFilterSimpleTest, CustomDefaultHandler) { EXPECT_FALSE(default_handler_called); } +TEST_F(HttpRoutingFilterSimpleTest, UnmatchedRequestUsesDefaultHandler) { + std::atomic default_handler_called(false); + + executeInDispatcher([this, &default_handler_called]() { + filter_->registerDefaultHandler( + [&default_handler_called]( + const HttpRoutingFilter::RequestContext& req) { + default_handler_called = true; + EXPECT_EQ(req.method, "GET"); + EXPECT_EQ(req.path, "/missing?source=test"); + HttpRoutingFilter::Response resp; + resp.status_code = 404; + resp.body = "not found"; + resp.headers["content-length"] = std::to_string(resp.body.length()); + return resp; + }); + + EXPECT_CALL(*next_callbacks_, onHeaders(_, _)).Times(0); + + std::map headers; + headers[":method"] = "GET"; + headers[":path"] = "/missing?source=test"; + + filter_->onHeaders(headers, true); + }); + + EXPECT_TRUE(default_handler_called); +} + +TEST_F(HttpRoutingFilterSimpleTest, DefaultHandlerStatusZeroPassesThrough) { + std::atomic default_handler_called(false); + + executeInDispatcher([this, &default_handler_called]() { + filter_->registerDefaultHandler( + [&default_handler_called]( + const HttpRoutingFilter::RequestContext& req) { + default_handler_called = true; + EXPECT_EQ(req.path, "/mcp"); + HttpRoutingFilter::Response resp; + resp.status_code = 0; + return resp; + }); + + EXPECT_CALL(*next_callbacks_, onHeaders(_, true)).Times(1); + + std::map headers; + headers[":method"] = "POST"; + headers[":path"] = "/mcp"; + + filter_->onHeaders(headers, true); + }); + + EXPECT_TRUE(default_handler_called); +} + +TEST_F(HttpRoutingFilterSimpleTest, HandlerStatusZeroFallsBackToDefault) { + std::atomic handler_called(false); + std::atomic default_handler_called(false); + + executeInDispatcher([this, &handler_called, &default_handler_called]() { + filter_->registerHandler( + "GET", "/passthrough", + [&handler_called](const HttpRoutingFilter::RequestContext& req) { + handler_called = true; + HttpRoutingFilter::Response resp; + resp.status_code = 0; + return resp; + }); + filter_->registerDefaultHandler( + [&default_handler_called]( + const HttpRoutingFilter::RequestContext& req) { + default_handler_called = true; + EXPECT_EQ(req.path, "/passthrough"); + HttpRoutingFilter::Response resp; + resp.status_code = 404; + resp.body = "not found"; + resp.headers["content-length"] = std::to_string(resp.body.length()); + return resp; + }); + + EXPECT_CALL(*next_callbacks_, onHeaders(_, _)).Times(0); + + std::map headers; + headers[":method"] = "GET"; + headers[":path"] = "/passthrough"; + + filter_->onHeaders(headers, true); + }); + + EXPECT_TRUE(handler_called); + EXPECT_TRUE(default_handler_called); +} + +TEST_F(HttpRoutingFilterSimpleTest, HandledDefaultResponseConsumesRequestBody) { + std::atomic default_handler_called(false); + + executeInDispatcher([this, &default_handler_called]() { + filter_->registerDefaultHandler( + [&default_handler_called]( + const HttpRoutingFilter::RequestContext& req) { + default_handler_called = true; + EXPECT_EQ(req.method, "POST"); + EXPECT_EQ(req.path, "/unknown"); + HttpRoutingFilter::Response resp; + resp.status_code = 404; + resp.body = "not found"; + resp.headers["content-length"] = std::to_string(resp.body.length()); + return resp; + }); + + EXPECT_CALL(*next_callbacks_, onHeaders(_, _)).Times(0); + EXPECT_CALL(*next_callbacks_, onBody(_, _)).Times(0); + EXPECT_CALL(*next_callbacks_, onMessageComplete()).Times(0); + + std::map headers; + headers[":method"] = "POST"; + headers[":path"] = "/unknown"; + + filter_->onHeaders(headers, true); + filter_->onBody(R"({"jsonrpc":"2.0","method":"ping","id":1})", true); + filter_->onMessageComplete(); + }); + + EXPECT_TRUE(default_handler_called); +} + // Test request context structure TEST_F(HttpRoutingFilterSimpleTest, RequestContext) { HttpRoutingFilter::RequestContext ctx; @@ -456,4 +582,4 @@ TEST_F(HttpRoutingFilterSimpleTest, GetRequestImmediateExecution) { } // namespace } // namespace filter -} // namespace mcp \ No newline at end of file +} // namespace mcp diff --git a/tests/filter/test_json_rpc_protocol_filter.cc b/tests/filter/test_json_rpc_protocol_filter.cc index 7e34a6ec3..ad7eb4c13 100644 --- a/tests/filter/test_json_rpc_protocol_filter.cc +++ b/tests/filter/test_json_rpc_protocol_filter.cc @@ -230,6 +230,81 @@ TEST_F(JsonRpcProtocolFilterTest, InvalidJson) { EXPECT_EQ(network::FilterStatus::Continue, status); } +TEST_F(JsonRpcProtocolFilterTest, EndStreamParsesPrettyPrintedBodyAsOneJson) { + EXPECT_CALL(*callbacks_, onRequest(_)) + .WillOnce([](const jsonrpc::Request& req) { + EXPECT_EQ("test.pretty", req.method); + EXPECT_TRUE(holds_alternative(req.id)); + EXPECT_EQ(7, get(req.id)); + }); + EXPECT_CALL(*callbacks_, onProtocolError(_)).Times(0); + + const std::string json_str = + "{\n" + " \"jsonrpc\": \"2.0\",\n" + " \"id\": 7,\n" + " \"method\": \"test.pretty\",\n" + " \"params\": {\n" + " \"message\": \"line one\\nline two\"\n" + " }\n" + "}"; + + auto status = processData(json_str, true); + EXPECT_EQ(network::FilterStatus::Continue, status); +} + +TEST_F(JsonRpcProtocolFilterTest, EndStreamFallsBackToNdjsonBodies) { + InSequence seq; + EXPECT_CALL(*callbacks_, onNotification(_)) + .WillOnce([](const jsonrpc::Notification& notification) { + EXPECT_EQ("first", notification.method); + }); + EXPECT_CALL(*callbacks_, onNotification(_)) + .WillOnce([](const jsonrpc::Notification& notification) { + EXPECT_EQ("second", notification.method); + }); + EXPECT_CALL(*callbacks_, onProtocolError(_)).Times(0); + + const std::string json_str = R"({"jsonrpc":"2.0","method":"first"})" + "\n" + R"({"jsonrpc":"2.0","method":"second"})" + "\n"; + + auto status = processData(json_str, true); + EXPECT_EQ(network::FilterStatus::Continue, status); +} + +TEST_F(JsonRpcProtocolFilterTest, EndStreamMalformedBodyReportsParseError) { + EXPECT_CALL(*callbacks_, onRequest(_)).Times(0); + EXPECT_CALL(*callbacks_, onNotification(_)).Times(0); + EXPECT_CALL(*callbacks_, onResponse(_)).Times(0); + EXPECT_CALL(*callbacks_, onProtocolError(_)).WillOnce([](const Error& error) { + EXPECT_EQ(jsonrpc::PARSE_ERROR, error.code); + }); + + auto status = processData("{ invalid json }", true); + EXPECT_EQ(network::FilterStatus::Continue, status); +} + +TEST_F(JsonRpcProtocolFilterTest, + EndStreamMalformedPrettyBodyReportsOneParseError) { + EXPECT_CALL(*callbacks_, onRequest(_)).Times(0); + EXPECT_CALL(*callbacks_, onNotification(_)).Times(0); + EXPECT_CALL(*callbacks_, onResponse(_)).Times(0); + EXPECT_CALL(*callbacks_, onProtocolError(_)).WillOnce([](const Error& error) { + EXPECT_EQ(jsonrpc::PARSE_ERROR, error.code); + }); + + const std::string json_str = + "{\n" + " \"jsonrpc\": \"2.0\",\n" + " \"id\": 9,\n" + " \"method\": \"broken\"\n"; + + auto status = processData(json_str, true); + EXPECT_EQ(network::FilterStatus::Continue, status); +} + /** * Test message framing mode */ diff --git a/tests/integration/test_http_sse_filter_server_mode.cc b/tests/integration/test_http_sse_filter_server_mode.cc index 1c8a76438..a216854d4 100644 --- a/tests/integration/test_http_sse_filter_server_mode.cc +++ b/tests/integration/test_http_sse_filter_server_mode.cc @@ -248,6 +248,35 @@ TEST_F(ServerModeFilterTest, GetSseWithQueryString_StillSseStream) { closeOnDispatcher(std::move(conn), std::move(factory)); } +TEST_F(ServerModeFilterTest, UnknownHttpPath_ReturnsNotFoundImmediately) { + ServerModeCallbacks callbacks; + std::unique_ptr conn; + network::IoHandlePtr peer; + std::shared_ptr factory; + + executeInDispatcher([&]() { + auto h = makeServerHarness(callbacks); + conn = std::move(h.conn); + peer = std::move(h.peer); + factory = std::move(h.factory); + + writeClientBytes(*peer, + "GET /.well-known/oauth-protected-resource HTTP/1.1\r\n" + "Host: localhost\r\n" + "\r\n"); + }); + + std::string wire = drainPeer(*peer, 500ms); + EXPECT_NE(wire.find("HTTP/1.1 404"), std::string::npos) + << "Expected immediate 404, got: " << wire; + EXPECT_NE(wire.find(R"({"error":"not_found"})"), std::string::npos) + << "Expected JSON not_found body, got: " << wire; + EXPECT_TRUE(callbacks.requests_.empty()) + << "Unknown HTTP paths must not reach JSON-RPC dispatch"; + + closeOnDispatcher(std::move(conn), std::move(factory)); +} + // ── POST /mcp → PlainHttp mode ──────────────────────────────────── TEST_F(ServerModeFilterTest, PostMcp_PlainHttpMode) { @@ -431,6 +460,43 @@ TEST_F(ServerModeFilterTest, PlainPost_AnnouncesEmptyBinding) { closeOnDispatcher(std::move(conn), std::move(factory)); } +TEST_F(ServerModeFilterTest, StreamablePost_AnnouncesMcpSessionIdBinding) { + ServerModeCallbacks callbacks; + std::unique_ptr conn; + network::IoHandlePtr peer; + std::shared_ptr factory; + + executeInDispatcher([&]() { + auto h = makeServerHarness(callbacks); + conn = std::move(h.conn); + peer = std::move(h.peer); + factory = std::move(h.factory); + + std::string body = R"({"jsonrpc":"2.0","method":"ping","id":3})"; + std::string request = + "POST /mcp HTTP/1.1\r\n" + "Host: localhost\r\n" + "Content-Type: application/json\r\n" + "Mcp-Session-Id: streamable-client-1\r\n" + "Content-Length: " + + std::to_string(body.size()) + + "\r\n" + "\r\n" + + body; + writeClientBytes(*peer, request); + }); + + std::this_thread::sleep_for(200ms); + + ASSERT_EQ(callbacks.requests_.size(), 1u); + EXPECT_EQ(callbacks.requests_[0].method, "ping"); + EXPECT_EQ(callbacks.binding_at_request_[0], "streamable-client-1") + << "Streamable HTTP dispatch must propagate Mcp-Session-Id as the " + "transport session id"; + + closeOnDispatcher(std::move(conn), std::move(factory)); +} + // ── Callback-proxy notification must not leak a 202 onto the SSE stream ── // // A notification POSTed to /callback/{id} is already answered with a 202