Sprint 291: Add HTTP/SSE daemon mode to whetstone_mcp (Steps 1983-1987)
Eliminates the dual-process bug where multiple Claude Code sessions each spawn their own whetstone_mcp stdio process with split in-memory state. All sessions now share a single persistent daemon via MCP HTTP+SSE transport. Changes: - MCPHttpServer.h: HTTP+SSE transport (cpp-httplib via FetchContent). GET /sse streams events per session, POST /message routes JSON-RPC, GET /health for status. Thread-safe session map with mutex/cv/queue. - MCPBridge.h: adds runHttp(port, workspace) method - mcp_main.cpp: --daemon/--port flags, daemon.pid lockfile, duplicate guard - CMakeLists.txt: FetchContent for cpp-httplib, step1983-1987 targets - tools/start-whetstone-daemon.sh: idempotent startup, health-check poll - 5/5 tests passing (step1987_test) Note: Documents/.mcp.json switched to SSE transport in working directory (not tracked in this repo). Daemon must be started before Claude Code sessions. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Bill
280
editor/src/MCPHttpServer.h
Normal file
280
editor/src/MCPHttpServer.h
Normal file
@@ -0,0 +1,280 @@
|
||||
#pragma once
|
||||
// Step 1983: MCPHttpServer — HTTP+SSE transport for MCP daemon mode.
|
||||
//
|
||||
// Implements MCP 2024-11-05 HTTP+SSE transport so whetstone_mcp can run as
|
||||
// a persistent shared daemon instead of a per-session stdio process:
|
||||
//
|
||||
// GET /sse — open SSE stream; sends endpoint event
|
||||
// POST /message?sessionId=<id> — receive JSON-RPC; push response to SSE
|
||||
// GET /health — status JSON
|
||||
//
|
||||
// Multiple Claude Code sessions all connect to the same daemon and share
|
||||
// a single in-memory HeadlessEditorState (recording, workspace, AST, etc.).
|
||||
//
|
||||
// Usage:
|
||||
// MCPHttpServer srv;
|
||||
// srv.setRequestHandler([&state](const json& req) {
|
||||
// return state.processAgentRequest(req, "mcp-session");
|
||||
// });
|
||||
// srv.setWorkspace("/path/to/workspace");
|
||||
// srv.listen(7600); // blocking
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <condition_variable>
|
||||
#include <cstdlib>
|
||||
#include <filesystem>
|
||||
#include <fstream>
|
||||
#include <functional>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <queue>
|
||||
#include <random>
|
||||
#include <sstream>
|
||||
#include <string>
|
||||
#include <unordered_map>
|
||||
|
||||
#include <nlohmann/json.hpp>
|
||||
#include <httplib.h>
|
||||
|
||||
using json = nlohmann::json;
|
||||
|
||||
class MCPHttpServer {
|
||||
public:
|
||||
using RpcHandler = std::function<json(const json&)>;
|
||||
|
||||
void setRequestHandler(RpcHandler h) { handler_ = std::move(h); }
|
||||
void setWorkspace(const std::string& root) { workspaceRoot_ = root; }
|
||||
|
||||
// Start HTTP server on given port (blocking). Returns when stopped.
|
||||
void listen(int port) {
|
||||
port_ = port;
|
||||
_writePidFile();
|
||||
|
||||
httplib::Server svr;
|
||||
svr_ = &svr;
|
||||
|
||||
svr.Get("/sse", [this](const httplib::Request& req, httplib::Response& res) {
|
||||
_sseHandler(req, res);
|
||||
});
|
||||
svr.Post("/message", [this](const httplib::Request& req, httplib::Response& res) {
|
||||
_messageHandler(req, res);
|
||||
});
|
||||
svr.Get("/health", [this](const httplib::Request& req, httplib::Response& res) {
|
||||
_healthHandler(req, res);
|
||||
});
|
||||
|
||||
std::cerr << "[whetstone-mcp] HTTP+SSE daemon on port " << port << "\n";
|
||||
svr.listen("0.0.0.0", port);
|
||||
|
||||
_removePidFile();
|
||||
svr_ = nullptr;
|
||||
}
|
||||
|
||||
void stop() {
|
||||
if (svr_) svr_->stop();
|
||||
std::lock_guard<std::mutex> lk(sessionsMtx_);
|
||||
for (auto& [id, sess] : sessions_) {
|
||||
std::lock_guard<std::mutex> slk(sess->mtx);
|
||||
sess->closed = true;
|
||||
sess->cv.notify_all();
|
||||
}
|
||||
sessions_.clear();
|
||||
}
|
||||
|
||||
// Generate a unique session ID (hex, 128-bit). Also used by tests.
|
||||
std::string newSessionId() {
|
||||
std::uniform_int_distribution<uint64_t> dist;
|
||||
std::ostringstream oss;
|
||||
oss << std::hex << dist(rng_) << dist(rng_);
|
||||
return oss.str();
|
||||
}
|
||||
|
||||
private:
|
||||
// -----------------------------------------------------------------------
|
||||
// Per-connection session state
|
||||
// -----------------------------------------------------------------------
|
||||
struct Session {
|
||||
std::string id;
|
||||
std::mutex mtx;
|
||||
std::queue<std::string> pending; // outbound JSON-RPC response strings
|
||||
std::condition_variable cv;
|
||||
bool closed = false;
|
||||
bool endpointSent = false; // did we send the "endpoint" event yet?
|
||||
};
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// GET /sse
|
||||
// -----------------------------------------------------------------------
|
||||
void _sseHandler(const httplib::Request& /*req*/, httplib::Response& res) {
|
||||
auto sess = std::make_shared<Session>();
|
||||
sess->id = newSessionId();
|
||||
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(sessionsMtx_);
|
||||
sessions_[sess->id] = sess;
|
||||
}
|
||||
|
||||
// The endpoint URL the client will POST messages to
|
||||
std::string endpointEvent =
|
||||
"event: endpoint\ndata: /message?sessionId=" + sess->id + "\n\n";
|
||||
|
||||
res.set_header("Cache-Control", "no-cache");
|
||||
res.set_header("X-Accel-Buffering","no");
|
||||
|
||||
// Chunked SSE stream — provider is called in a loop until it returns false
|
||||
res.set_chunked_content_provider(
|
||||
"text/event-stream",
|
||||
|
||||
// Content provider: called repeatedly by httplib's send loop
|
||||
[sess, endpointEvent](size_t /*offset*/, httplib::DataSink& sink) -> bool {
|
||||
// First call: send the endpoint event
|
||||
{
|
||||
std::unique_lock<std::mutex> lk(sess->mtx);
|
||||
if (!sess->endpointSent) {
|
||||
sess->endpointSent = true;
|
||||
lk.unlock();
|
||||
return sink.write(endpointEvent.data(), endpointEvent.size());
|
||||
}
|
||||
}
|
||||
|
||||
// Subsequent calls: wait for outbound messages or keepalive timeout
|
||||
std::unique_lock<std::mutex> lk(sess->mtx);
|
||||
bool got = sess->cv.wait_for(lk, std::chrono::seconds(25),
|
||||
[&sess]{ return !sess->pending.empty() || sess->closed; });
|
||||
|
||||
if (sess->closed) return false;
|
||||
|
||||
if (!got) {
|
||||
// Timeout — send SSE comment as keepalive to prevent proxy timeouts
|
||||
static const std::string kKeepalive = ": keepalive\n\n";
|
||||
lk.unlock();
|
||||
return sink.write(kKeepalive.data(), kKeepalive.size());
|
||||
}
|
||||
|
||||
// Drain the pending queue
|
||||
while (!sess->pending.empty()) {
|
||||
std::string frame = "event: message\ndata: " +
|
||||
sess->pending.front() + "\n\n";
|
||||
sess->pending.pop();
|
||||
lk.unlock();
|
||||
if (!sink.write(frame.data(), frame.size())) return false;
|
||||
lk.lock();
|
||||
}
|
||||
return true;
|
||||
},
|
||||
|
||||
// Releaser: remove session when the SSE connection closes
|
||||
[this, sess](bool /*success*/) {
|
||||
std::lock_guard<std::mutex> lk(sessionsMtx_);
|
||||
sessions_.erase(sess->id);
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// POST /message?sessionId=<id>
|
||||
// -----------------------------------------------------------------------
|
||||
void _messageHandler(const httplib::Request& req, httplib::Response& res) {
|
||||
std::string sessionId = req.get_param_value("sessionId");
|
||||
if (sessionId.empty()) {
|
||||
res.status = 400;
|
||||
res.set_content("{\"error\":\"missing sessionId\"}", "application/json");
|
||||
return;
|
||||
}
|
||||
|
||||
std::shared_ptr<Session> sess;
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(sessionsMtx_);
|
||||
auto it = sessions_.find(sessionId);
|
||||
if (it == sessions_.end()) {
|
||||
res.status = 404;
|
||||
res.set_content("{\"error\":\"session not found\"}", "application/json");
|
||||
return;
|
||||
}
|
||||
sess = it->second;
|
||||
}
|
||||
|
||||
// Parse and dispatch JSON-RPC request
|
||||
json response;
|
||||
try {
|
||||
json request = json::parse(req.body);
|
||||
if (handler_) {
|
||||
response = handler_(request);
|
||||
} else {
|
||||
response = {
|
||||
{"jsonrpc","2.0"},
|
||||
{"id", request.value("id", json(nullptr))},
|
||||
{"error", {{"code",-32603},{"message","No handler registered"}}}
|
||||
};
|
||||
}
|
||||
} catch (const json::exception& /*e*/) {
|
||||
response = {
|
||||
{"jsonrpc","2.0"}, {"id", json(nullptr)},
|
||||
{"error", {{"code",-32700},{"message","Parse error"}}}
|
||||
};
|
||||
}
|
||||
|
||||
// Push response onto the session's SSE queue
|
||||
if (!response.is_null()) {
|
||||
std::lock_guard<std::mutex> lk(sess->mtx);
|
||||
sess->pending.push(response.dump());
|
||||
sess->cv.notify_one();
|
||||
}
|
||||
|
||||
// 202 Accepted — actual response delivered via SSE
|
||||
res.status = 202;
|
||||
res.set_content("", "application/json");
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// GET /health
|
||||
// -----------------------------------------------------------------------
|
||||
void _healthHandler(const httplib::Request& /*req*/, httplib::Response& res) {
|
||||
size_t sessionCount;
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(sessionsMtx_);
|
||||
sessionCount = sessions_.size();
|
||||
}
|
||||
json body = {
|
||||
{"status", "ok"},
|
||||
{"sessions", (int)sessionCount},
|
||||
{"port", port_}
|
||||
};
|
||||
res.set_content(body.dump(), "application/json");
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// PID file — written on listen(), removed on exit
|
||||
// -----------------------------------------------------------------------
|
||||
void _writePidFile() {
|
||||
if (workspaceRoot_.empty()) return;
|
||||
std::filesystem::path dir =
|
||||
std::filesystem::path(workspaceRoot_) / ".whetstone";
|
||||
std::error_code ec;
|
||||
std::filesystem::create_directories(dir, ec);
|
||||
std::ofstream f(dir / "daemon.pid");
|
||||
if (f) f << ::getpid();
|
||||
}
|
||||
|
||||
void _removePidFile() {
|
||||
if (workspaceRoot_.empty()) return;
|
||||
std::filesystem::path p =
|
||||
std::filesystem::path(workspaceRoot_) / ".whetstone" / "daemon.pid";
|
||||
std::error_code ec;
|
||||
std::filesystem::remove(p, ec);
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// State
|
||||
// -----------------------------------------------------------------------
|
||||
RpcHandler handler_;
|
||||
std::string workspaceRoot_;
|
||||
int port_ = 7600;
|
||||
httplib::Server* svr_ = nullptr;
|
||||
|
||||
std::mt19937_64 rng_{std::random_device{}()};
|
||||
|
||||
std::mutex sessionsMtx_;
|
||||
std::unordered_map<std::string, std::shared_ptr<Session>> sessions_;
|
||||
};
|
||||
Reference in New Issue
Block a user