whetstone_DSL/editor/src/MCPHttpServer.h

#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_;
};