external_bridge.cpp

Go to the documentation of this file.

// SPDX-License-Identifier: Apache-2.0
#include <entropic/mcp/external_bridge.h>
#include <entropic/entropic.h>
#include <entropic/types/logging.h>
 
#include "engine_handle.h"
#include "final_text.h"  // gh#130 (v2.10.2)
 
#include <nlohmann/json.hpp>
 
#include <atomic>
#include <cerrno>
#include <chrono>
#include <cstring>
#include <filesystem>
#include <fstream>
#include <functional>
#include <mutex>
#include <sstream>
#include <thread>
#include <unordered_map>
#include <unordered_set>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/un.h>
#include <poll.h>
 
using json = nlohmann::json;
 
namespace entropic {
 
static auto logger = entropic::log::get("mcp.external_bridge");
 
// gh#58: process-wide set of socket paths currently bound by an
// ExternalBridge in this process. Without this, a second handle whose
// project_dir hashes to the same socket path as a first handle would
// unlink the live socket out from under handle #1 (see the
// std::filesystem::remove call in create_listen_socket). With this
// guard, the second handle declines to start its bridge instead.
static std::mutex s_bound_sockets_mu;
static std::unordered_set<std::string> s_bound_sockets;
 
// Forward declaration
std::filesystem::path compute_socket_path(
    const std::filesystem::path& project_dir);
 
// ── JSON-RPC helpers ─────────────────────────────────────
 
static std::string rpc_ok(const json& id, const json& result) {
    json r = {{"jsonrpc", "2.0"}, {"id", id}, {"result", result}};
    return r.dump();
}
 
static std::string rpc_err(const json& id, int code,
                           const std::string& msg) {
    json r = {{"jsonrpc", "2.0"}, {"id", id},
              {"error", {{"code", code}, {"message", msg}}}};
    return r.dump();
}
 
static json tool_text(const std::string& text) {
    return {{"content", json::array({{{"type", "text"}, {"text", text}}})}};
}
 
// ── Tool definitions ─────────────────────────────────────
 
static json tool_definitions() {
    return json::array({
        {{"name", "entropic.ask"},
         {"description",
          "Submit a prompt to the running entropic engine. "
          "Set async=true to return immediately with a task_id; "
          "the engine pushes a notification when done."},
         {"inputSchema", {
            {"type", "object"},
            {"properties", {
                {"prompt", {{"type", "string"},
                            {"description", "User message"}}},
                {"async", {{"type", "boolean"},
                           {"description", "Run asynchronously"},
                           {"default", false}}}
            }},
            {"required", json::array({"prompt"})}
         }}},
        {{"name", "entropic.ask_status"},
         {"description",
          "Check status of an async entropic.ask task."},
         {"inputSchema", {
            {"type", "object"},
            {"properties", {{"task_id", {
                {"type", "string"},
                {"description", "Task ID from async ask"}
            }}}},
            {"required", json::array({"task_id"})}
         }}},
        {{"name", "entropic.status"},
         {"description", "Engine version and message count."},
         {"inputSchema", {{"type", "object"},
                          {"properties", json::object()}}}},
        {{"name", "entropic.context_clear"},
         {"description", "Clear conversation history."},
         {"inputSchema", {{"type", "object"},
                          {"properties", json::object()}}}},
        {{"name", "entropic.context_count"},
         {"description", "Return the message count."},
         {"inputSchema", {{"type", "object"},
                          {"properties", json::object()}}}},
    });
}
 
// ── Tool handlers ────────────────────────────────────────
 
// gh#130 (v2.10.2): extract_final_text moved to the private facade header
// final_text.h so its selection rule is unit-testable — this was a `static`
// helper with no reachable test, and it shipped a scanning bug that returned
// "(no response)" to operators while the answer sat earlier in the same
// conversation. See facade_text::extract_final_text for the rule.
 
static void write_json_line(int fd, const json& msg) {
    auto s = msg.dump() + "\n";
    ::write(fd, s.c_str(), s.size());
}
 
static void send_progress(int fd, const std::string& token_text,
                          const std::string& progress_token) {
    json notif = {
        {"jsonrpc", "2.0"},
        {"method", "notifications/progress"},
        {"params", {
            {"progressToken", progress_token},
            {"progress", token_text}
        }}
    };
    write_json_line(fd, notif);
}
 
static json final_answer_from_context(entropic_handle_t handle) {
    char* msgs_json = nullptr;
    entropic_context_get(handle, &msgs_json);
    auto answer = facade_text::final_text_or_reason(msgs_json);
    entropic_free(msgs_json);
    return tool_text(answer);
}
 
static json handle_ask_plain(entropic_handle_t handle, const json& args) {
    auto it = args.find("prompt");
    if (it == args.end() || !it->is_string()) {
        return tool_text("error: missing 'prompt' argument");
    }
    char* result_json = nullptr;
    auto err = entropic_run(handle, it->get<std::string>().c_str(),
                            &result_json);
    if (err != ENTROPIC_OK) {
        const char* msg = entropic_last_error(handle);
        entropic_free(result_json);
        return tool_text(std::string("error: ") + (msg ? msg : "unknown"));
    }
    // Resolve before the free — final_text_or_reason reads that buffer.
    auto answer = facade_text::final_text_or_reason(result_json);
    entropic_free(result_json);
    return tool_text(answer);
}
 
static json handle_ask(entropic_handle_t handle, const json& args,
                       int client_fd, const std::string& call_id) {
    auto it = args.find("prompt");
    if (it == args.end() || !it->is_string()) {
        return tool_text("error: missing 'prompt' argument");
    }
    std::string prompt = it->get<std::string>();
 
    // Stream tokens as progress notifications
    struct StreamCtx { int fd; std::string token_id; };
    StreamCtx sctx{client_fd, call_id};
    auto on_token = [](const char* tok, size_t len, void* ud) {
        auto* ctx = static_cast<StreamCtx*>(ud);
        send_progress(ctx->fd, std::string(tok, len), ctx->token_id);
    };
    auto err = entropic_run_streaming(
        handle, prompt.c_str(), on_token, &sctx, nullptr);
    if (err != ENTROPIC_OK) {
        const char* msg = entropic_last_error(handle);
        return tool_text(std::string("error: ") + (msg ? msg : "unknown"));
    }
 
    return final_answer_from_context(handle);
}
 
static json handle_status(entropic_handle_t handle) {
    size_t count = 0;
    entropic_context_count(handle, &count);
    std::ostringstream os;
    os << "entropic " << entropic_version()
       << "\nmessages: " << count;
    // Metrics + per-tier breakdown (P2-15 follow-up, 2.0.6-rc16.2)
    char* mjson = nullptr;
    if (entropic_metrics_json(handle, &mjson) == ENTROPIC_OK
        && mjson != nullptr) {
        os << "\nmetrics: " << mjson;
        entropic_free(mjson);
    }
    return tool_text(os.str());
}
 
struct AsyncFinalState {
    std::string status;   
    std::string phase;    
    std::string text;     
};
 
static AsyncFinalState derive_async_final_state(
    entropic_handle_t handle,
    entropic_error_t err,
    char* result_json) {
    AsyncFinalState s;
    if (err == ENTROPIC_ERROR_CANCELLED
        || err == ENTROPIC_ERROR_INTERRUPTED) {
        const char* msg = entropic_last_error(handle);
        s.text = msg ? msg : "cancelled";
        s.status = "cancelled";
        s.phase = "cancelled";
    } else if (err != ENTROPIC_OK) {
        const char* msg = entropic_last_error(handle);
        s.text = msg ? msg : "unknown error";
        s.status = "error";
        s.phase = "failed";
    } else {
        // gh#130 (v2.10.2): same selection rule as the sync paths. This one
        // had no fallback at all, so a stalled async ask reported
        // status="done" with empty text — even less diagnosable than the
        // sync path's "(no response)".
        s.text = facade_text::final_text_or_reason(result_json);
        entropic_free(result_json);
        s.status = "done";
        s.phase = "done";
    }
    return s;
}
 
static bool mark_tasks_cancelling(ExternalBridge* bridge) {
    std::lock_guard<std::mutex> lock(bridge->tasks_mutex_);
    bool any = false;
    for (auto& [_, task] : bridge->tasks_for_cancel()) {
        if (task.status == "queued" || task.status == "running") {
            task.status = "cancelled";
            task.phase = "cancelling";
            any = true;
        }
    }
    return any;
}
 
static bool any_cancelling_left(ExternalBridge* bridge) {
    std::lock_guard<std::mutex> lock(bridge->tasks_mutex_);
    for (auto& [_, task] : bridge->tasks_for_cancel()) {
        if (task.phase == "cancelling") { return true; }
    }
    return false;
}
 
static void cancel_inflight_async_tasks(
    entropic_handle_t handle, ExternalBridge* bridge) {
    if (bridge == nullptr) { return; }
    entropic_interrupt(handle);  // idempotent, cheap
    if (!mark_tasks_cancelling(bridge)) { return; }
    for (int i = 0; i < 20 && any_cancelling_left(bridge); ++i) {
        std::this_thread::sleep_for(std::chrono::milliseconds(50));
    }
    bridge->detach_phase_observer();  // bump gen; silences any stale observer
}
 
static json handle_clear(entropic_handle_t handle,
                         ExternalBridge* bridge) {
    cancel_inflight_async_tasks(handle, bridge);
    auto err = entropic_context_clear(handle);
    if (err != ENTROPIC_OK) {
        return tool_text("error: clear failed");
    }
    return tool_text("conversation cleared");
}
 
static json handle_count(entropic_handle_t handle) {
    size_t count = 0;
    entropic_context_count(handle, &count);
    return tool_text(std::to_string(count));
}
 
// ── Async ask status ─────────────────────────────────────
 
json ExternalBridge::handle_ask_status(const json& args) {
    auto tid = args.value("task_id", std::string{});
    std::lock_guard<std::mutex> lock(tasks_mutex_);
    auto it = tasks_.find(tid);
    if (it == tasks_.end()) {
        return tool_text("error: unknown task_id");
    }
    json status = {{"status", it->second.status},
                   {"phase",  it->second.phase}};
    if (!it->second.result.empty()) {
        auto key = (it->second.status == "error") ? "error" : "result";
        status[key] = it->second.result;
    }
    return tool_text(status.dump());
}
 
// ── UUID generation ──────────────────────────────────────
 
static std::string generate_task_id() {
    static std::atomic<uint64_t> counter{0};
    auto n = counter.fetch_add(1);
    auto t = std::chrono::steady_clock::now().time_since_epoch().count();
    std::ostringstream ss;
    ss << std::hex << (t ^ (n * 2654435761ULL));
    return "task-" + ss.str();
}
 
// ── Dispatch ─────────────────────────────────────────────
 
static json dispatch_ask(entropic_handle_t handle,
                         ExternalBridge* bridge,
                         const json& args,
                         int client_fd,
                         const std::string& call_id) {
    if (args.value("async", false)) {
        auto task_id = generate_task_id();
        bridge->run_async_ask(
            args.value("prompt", ""), task_id, client_fd);
        return tool_text("async task started: " + task_id);
    }
    if (!bridge->ask_streaming()) { return handle_ask_plain(handle, args); }
    return handle_ask(handle, args, client_fd, call_id);
}
 
static json dispatch_tool(entropic_handle_t handle,
                          ExternalBridge* bridge,
                          const json& params,
                          int client_fd,
                          const std::string& call_id) {
    std::string name = params.value("name", std::string{});
    json args = params.value("arguments", json::object());
    if (name == "entropic.ask") {
        return dispatch_ask(handle, bridge, args, client_fd, call_id);
    }
    json result;
    if      (name == "entropic.ask_status")    { result = bridge->handle_ask_status(args); }
    else if (name == "entropic.status")        { result = handle_status(handle); }
    else if (name == "entropic.context_clear") { result = handle_clear(handle, bridge); }
    else if (name == "entropic.context_count") { result = handle_count(handle); }
    else { result = tool_text("error: unknown tool '" + name + "'"); }
    return result;
}
 
// ── ExternalBridge ───────────────────────────────────────
 
ExternalBridge::ExternalBridge(
    entropic_handle_t handle,
    const ExternalMCPConfig& config,
    const std::filesystem::path& project_dir)
    : handle_(handle), config_(config) {
    socket_path_ = config.socket_path.has_value()
        ? config.socket_path.value()
        : compute_socket_path(project_dir);
}
 
ExternalBridge::~ExternalBridge() {
    stop();
}
 
static void prepare_socket_dir(const std::filesystem::path& parent) {
    std::error_code ec;
    std::filesystem::create_directories(parent, ec);
    ::chmod(parent.c_str(), S_IRWXU);  // 0700
}
 
static bool socket_path_safe(const std::filesystem::path& path) {
    struct stat st{};
    if (::lstat(path.c_str(), &st) != 0) {
        return errno == ENOENT;  // absent is fine
    }
    bool is_symlink = S_ISLNK(st.st_mode);
    bool is_socket = S_ISSOCK(st.st_mode);
    if (!is_socket || is_symlink) {
        logger->error(
            "Refusing to bind: {} is a {} (expected unix socket)",
            path.string(),
            is_symlink ? "symlink" : "non-socket file");
    }
    return is_socket && !is_symlink;
}
 
static bool bind_and_listen(int fd, const std::filesystem::path& path) {
    auto s = path.string();
    if (s.size() >= sizeof(sockaddr_un::sun_path)) { return false; }
    struct sockaddr_un addr{};
    addr.sun_family = AF_UNIX;
    std::strncpy(addr.sun_path, s.c_str(),
                 sizeof(addr.sun_path) - 1);
    if (bind(fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) != 0) {
        return false;
    }
    ::chmod(s.c_str(), S_IRUSR | S_IWUSR);  // 0600 — owner only
    return listen(fd, 1) == 0;
}
 
static int create_listen_socket(const std::filesystem::path& path) {
    prepare_socket_dir(path.parent_path());
    if (!socket_path_safe(path)) { return -1; }
    std::filesystem::remove(path);
 
    int fd = socket(AF_UNIX, SOCK_STREAM, 0);
    bool ok = (fd >= 0) && bind_and_listen(fd, path);
    if (!ok) {
        logger->error("Socket setup failed for {}: {}",
                      path.string(), std::strerror(errno));
        if (fd >= 0) { ::close(fd); }
        return -1;
    }
    logger->info(
        "External MCP bridge ready: project_dir(canonical)={} socket={}",
        path.parent_path().parent_path().string(), path.string());
    return fd;
}
 
static bool peer_uid_matches(int client_fd) {
    struct ucred cred{};
    socklen_t len = sizeof(cred);
    if (getsockopt(client_fd, SOL_SOCKET, SO_PEERCRED,
                   &cred, &len) != 0) {
        logger->warn("SO_PEERCRED failed on fd={}: {}",
                     client_fd, std::strerror(errno));
        return false;
    }
    if (cred.uid != ::geteuid()) {
        logger->warn(
            "Rejecting MCP client on fd={}: peer uid {} != engine uid {}",
            client_fd, cred.uid, ::geteuid());
        return false;
    }
    return true;
}
 
bool ExternalBridge::start() {
    // Refuse to start if another handle in this process already owns
    // this socket path. The pre-fix create_listen_socket call unlinks
    // any existing socket file before binding — that's correct for
    // crashed-prior-process recovery, but if a live in-process bridge
    // owned that file, the unlink stole its binding silently. The
    // claim must happen *before* create_listen_socket so a concurrent
    // start() on a colliding path cannot pass the check and then race
    // into the unlink.
    std::error_code ec;
    auto canonical =
        std::filesystem::weakly_canonical(socket_path_, ec).string();
    if (ec) { canonical = socket_path_.string(); }
    {
        std::lock_guard lk(s_bound_sockets_mu);
        if (!s_bound_sockets.insert(canonical).second) {
            logger->warn(
                "External MCP bridge: socket {} already bound by another "
                "handle in this process; declining to start. Set "
                "external.socket_path to a distinct path per handle "
                "if you need per-handle bridges.",
                canonical);
            return false;
        }
    }
 
    listen_fd_ = create_listen_socket(socket_path_);
    if (listen_fd_ < 0) {
        std::lock_guard lk(s_bound_sockets_mu);
        s_bound_sockets.erase(canonical);
        return false;
    }
    bound_canonical_ = canonical;
 
    running_.store(true);
    // gh#59 (v2.3.1): the accept thread (and the per-client serve
    // threads it spawns) need to log on this handle's behalf — install
    // a HandleLogScope at the top of each thread body so spdlog lines
    // route through HandleAwareSink to this handle's session.log.
    int log_id = handle_ ? handle_->log_id : 0;
    accept_thread_ = std::thread([this, log_id]() {
        entropic::log::HandleLogScope scope(log_id);
        accept_loop();
    });
 
    logger->info("External MCP bridge listening on {}",
                 socket_path_.string());
    return true;
}
 
void ExternalBridge::stop() {
    running_.store(false);
    if (listen_fd_ >= 0) {
        ::close(listen_fd_);
        listen_fd_ = -1;
    }
    if (accept_thread_.joinable()) {
        accept_thread_.join();
    }
 
    // Wake every connected client's blocking read so the per-client
    // thread can observe the disconnect and exit. shutdown(SHUT_RDWR)
    // on a connected socket returns EOF to the peer; the read in
    // serve_client returns 0 and read_line returns "" → exits.
    std::vector<std::unique_ptr<ClientThread>> drained;
    {
        std::lock_guard<std::mutex> lock(client_threads_mutex_);
        drained = std::move(client_threads_);
        client_threads_.clear();
    }
    for (auto& ct : drained) {
        if (ct->fd >= 0) { ::shutdown(ct->fd, SHUT_RDWR); }
    }
    for (auto& ct : drained) {
        if (ct->thread.joinable()) { ct->thread.join(); }
    }
 
    // Clean up socket file
    std::error_code ec;
    std::filesystem::remove(socket_path_, ec);
 
    // Release our claim on the canonical path so a later handle in
    // this process can re-bind. Use the exact string we inserted in
    // start(); recomputing weakly_canonical here could differ if the
    // path was deleted mid-run.
    if (!bound_canonical_.empty()) {
        std::lock_guard lk(s_bound_sockets_mu);
        s_bound_sockets.erase(bound_canonical_);
        bound_canonical_.clear();
    }
}
 
void ExternalBridge::reap_finished_clients_locked() {
    auto it = client_threads_.begin();
    while (it != client_threads_.end()) {
        if ((*it)->finished.load()) {
            if ((*it)->thread.joinable()) { (*it)->thread.join(); }
            it = client_threads_.erase(it);
        } else {
            ++it;
        }
    }
}
 
void ExternalBridge::accept_loop() {
    while (running_.load()) {
        struct pollfd pfd{};
        pfd.fd = listen_fd_;
        pfd.events = POLLIN;
 
        int rc = poll(&pfd, 1, 500);  // 500ms timeout for shutdown check
        if (rc <= 0) { continue; }
 
        int client_fd = accept(listen_fd_, nullptr, nullptr);
        if (client_fd < 0) { continue; }
 
        if (!peer_uid_matches(client_fd)) {
            ::close(client_fd);  // v2.1.7 (gh#34): cross-uid attempt
            continue;
        }
 
        logger->info("External MCP client connected (fd={})", client_fd);
 
        auto ct = std::make_unique<ClientThread>();
        ct->fd = client_fd;
        // Capture a raw pointer for the thread body so the unique_ptr
        // can stay in client_threads_ without aliasing.
        ClientThread* raw = ct.get();
        // gh#59 (v2.3.1): propagate the handle log scope into the
        // per-client thread so its `serve_client` logs route to the
        // owning handle's session.log.
        int log_id = handle_ ? handle_->log_id : 0;
        ct->thread = std::thread([this, raw, log_id]() {
            entropic::log::HandleLogScope scope(log_id);
            serve_client(raw->fd);
            ::close(raw->fd);
            raw->fd = -1;
            logger->info("External MCP client disconnected");
            raw->finished.store(true);
        });
 
        std::lock_guard<std::mutex> lock(client_threads_mutex_);
        client_threads_.push_back(std::move(ct));
        reap_finished_clients_locked();
    }
}
 
static std::string read_line(int fd) {
    std::string line;
    char c;
    while (true) {
        ssize_t n = read(fd, &c, 1);
        if (n <= 0) { return {}; }
        if (c == '\n') { return line; }
        line += c;
    }
}
 
void ExternalBridge::serve_client(int client_fd) {
    subscribe(client_fd);
    struct Unsub {
        ExternalBridge* self;
        int fd;
        ~Unsub() { self->unsubscribe(fd); }
    } guard{this, client_fd};
 
    while (running_.load()) {
        auto line = read_line(client_fd);
        if (line.empty()) { break; }
 
        auto response = dispatch(line, client_fd);
        if (response.empty()) { continue; }  // notification — no reply
        response += '\n';
 
        ssize_t written = write(client_fd, response.c_str(),
                                response.size());
        if (written < 0) { break; }
    }
}
 
static json initialize_result() {
    return {
        {"protocolVersion", "2025-06-18"},
        {"serverInfo", {{"name", "entropic"},
                        {"version", entropic_version()}}},
        {"capabilities", {{"tools", json::object()}}}
    };
}
 
std::string ExternalBridge::dispatch(
    const std::string& request, int client_fd) {
    auto req = json::parse(request, nullptr, false);
    // Parse error or notification (no id) → no dispatch
    if (req.is_discarded() || !req.contains("id")) {
        return req.is_discarded()
            ? rpc_err(nullptr, -32700, "Parse error")
            : std::string{};
    }
 
    json id = req["id"];
    std::string method = req.value("method", std::string{});
    json params = req.value("params", json::object());
    json result;
 
    if      (method == "initialize")  { result = initialize_result(); }
    else if (method == "tools/list")  { result = {{"tools", tool_definitions()}}; }
    else if (method == "tools/call")  {
        auto id_str = id.is_string() ? id.get<std::string>()
                                     : id.dump();
        result = dispatch_tool(handle_, this, params, client_fd, id_str);
    }
    else if (method == "shutdown" || method == "exit") { result = json::object(); }
    else { return rpc_err(id, -32601, "Unknown method: " + method); }
    return rpc_ok(id, result);
}
 
// ── Async task support ───────────────────────────────────
 
static void phase_observer_cb(int state, void* ud) {
    auto* self = static_cast<ExternalBridge*>(ud);
    if (state != ENTROPIC_AGENT_STATE_VERIFYING) { return; }
    std::lock_guard<std::mutex> lock(self->tasks_mutex_);
    // E5+E6 (2.1.0): discard stale callbacks fired after detach_phase_observer
    // incremented observer_gen_. attached_gen_ was captured at attach time.
    if (self->observer_call_is_stale()) { return; }
    auto it = self->tasks_for_cancel().find(self->active_task_id_for_observer());
    if (it == self->tasks_for_cancel().end()) { return; }
    // First VERIFYING = "validating"; subsequent VERIFYING transitions
    // on the same task indicate revision retries → "revising".
    it->second.phase = (it->second.phase == "validating"
                         || it->second.phase == "revising")
        ? "revising" : "validating";
}
 
void ExternalBridge::attach_phase_observer(const std::string& task_id) {
    {
        std::lock_guard<std::mutex> lock(tasks_mutex_);
        active_task_id_ = task_id;
        attached_gen_ = ++observer_gen_;
    }
    entropic_set_state_observer(handle_, phase_observer_cb, this);
}
 
void ExternalBridge::detach_phase_observer() {
    {
        std::lock_guard<std::mutex> lock(tasks_mutex_);
        ++observer_gen_;
        active_task_id_.clear();
    }
    entropic_set_state_observer(handle_, nullptr, nullptr);
}
 
void ExternalBridge::run_async_ask(
    const std::string& prompt,
    const std::string& task_id,
    int client_fd) {
    {
        std::lock_guard<std::mutex> lock(tasks_mutex_);
        AsyncTask t;
        t.status = "queued";
        t.phase = "queued";
        t.created = std::chrono::steady_clock::now();
        tasks_[task_id] = std::move(t);
    }
 
    // gh#59 (v2.3.1): the async-ask worker runs entropic_run on this
    // handle's behalf — scope its logs to the owning handle.
    int log_id = handle_ ? handle_->log_id : 0;
    std::thread([this, prompt, task_id, client_fd, log_id]() {
        entropic::log::HandleLogScope scope(log_id);
        update_task_phase(task_id, "running", "running");
        attach_phase_observer(task_id);
 
        char* result_json = nullptr;
        auto err = entropic_run(handle_, prompt.c_str(), &result_json);
 
        detach_phase_observer();
 
        auto final_state = derive_async_final_state(
            handle_, err, result_json);
 
        {
            std::lock_guard<std::mutex> lock(tasks_mutex_);
            auto it = tasks_.find(task_id);
            if (it != tasks_.end()) {
                it->second.status = final_state.status;
                it->second.phase = final_state.phase;
                it->second.result = final_state.text;
            }
            // Issue #12 (v2.1.4): write sentinel UNDER tasks_mutex_,
            // before the MCP notification fires. Any external monitor
            // reacting to the sentinel can immediately call
            // entropic.ask_status and see the same terminal state.
            write_sentinel(task_id, final_state.status);
        }
        auto status = final_state.status;
 
        // Issue #4 (v2.1.2, parts A+B): emit the spec-defined
        // ``notifications/progress`` method instead of the previous
        // non-spec ``notifications/ask_complete``. MCP-compliant
        // clients are obligated to drain ``notifications/progress``
        // (it's in the documented set); some clients silently
        // buffered or stalled on the unknown method name, which
        // combined with the bridge's blocking broadcast (fixed in
        // part C of this release) produced the deadlock observed
        // against entropic-explorer ↔ Claude Code in the field.
        //
        // ``progressToken`` is the ``task_id`` so the consumer can
        // correlate the notification back to the originating
        // ``entropic.ask`` response (which carried the same
        // ``task_id``). The result body is NO LONGER shipped inline
        // — consumers fetch via ``entropic.ask_status``. This caps
        // notification size at ~200 bytes regardless of generated
        // output, eliminating a real DoS surface (a 50KB result
        // would otherwise flood the broadcast write path on every
        // subscriber). ``status`` rides in ``message`` so consumers
        // can branch on done / error / cancelled without an extra
        // round-trip just to learn which result kind to fetch.
        json notif = {
            {"jsonrpc", "2.0"},
            {"method", "notifications/progress"},
            {"params", {
                {"progressToken", task_id},
                {"progress", 100},
                {"total", 100},
                {"message", status}
            }}
        };
        broadcast_notification(notif);
 
        cleanup_expired_tasks();
        logger->info("Async task {} completed: {}", task_id, status);
    }).detach();
}
 
void ExternalBridge::subscribe(int fd) {
    std::lock_guard<std::mutex> lock(subscribers_mutex_);
    subscribers_.insert(fd);
}
 
void ExternalBridge::unsubscribe(int fd) {
    std::lock_guard<std::mutex> lock(subscribers_mutex_);
    subscribers_.erase(fd);
}
 
void ExternalBridge::broadcast_notification(const json& notif) {
    auto payload = notif.dump() + "\n";
 
    std::vector<int> snapshot;
    {
        std::lock_guard<std::mutex> lock(subscribers_mutex_);
        snapshot.assign(subscribers_.begin(), subscribers_.end());
    }
 
    std::vector<int> dead;
    for (int fd : snapshot) {
        ssize_t rc = ::send(fd, payload.c_str(), payload.size(),
                            MSG_DONTWAIT | MSG_NOSIGNAL);
        if (rc < 0) {
            // EAGAIN / EWOULDBLOCK: peer recv buffer full (slow consumer).
            // EBADF / EPIPE / ECONNRESET: peer closed or otherwise dead.
            // All collapse to "drop" — the long-term per-subscriber
            // queue lives in proposal P2-20260429-001.
            logger->warn("Subscriber fd {} send failed (errno={}) — dropping",
                         fd, errno);
            dead.push_back(fd);
        } else if (static_cast<size_t>(rc) < payload.size()) {
            logger->warn("Subscriber fd {} partial send ({}/{}) — dropping",
                         fd, rc, payload.size());
            dead.push_back(fd);
        }
    }
 
    if (!dead.empty()) {
        std::lock_guard<std::mutex> lock(subscribers_mutex_);
        for (int fd : dead) { subscribers_.erase(fd); }
    }
}
 
void ExternalBridge::update_task_phase(const std::string& task_id,
                                       const std::string& status,
                                       const std::string& phase) {
    std::lock_guard<std::mutex> lock(tasks_mutex_);
    auto it = tasks_.find(task_id);
    if (it == tasks_.end()) { return; }
    it->second.status = status;
    it->second.phase = phase;
}
 
void ExternalBridge::cleanup_expired_tasks() {
    auto cutoff = std::chrono::steady_clock::now()
        - std::chrono::minutes(15);
    auto sentinel_dir = async_sentinel_dir();
    std::lock_guard<std::mutex> lock(tasks_mutex_);
    for (auto it = tasks_.begin(); it != tasks_.end(); ) {
        if (it->second.created < cutoff) {
            if (!sentinel_dir.empty()) {
                for (const char* suffix :
                     {".done", ".failed", ".cancelled"}) {
                    std::error_code ec;
                    std::filesystem::remove(
                        sentinel_dir / (it->first + suffix), ec);
                }
            }
            it = tasks_.erase(it);
        } else {
            ++it;
        }
    }
}
 
std::filesystem::path ExternalBridge::async_sentinel_dir() const {
    std::filesystem::path root = async_sentinel_root_override_;
    if (root.empty() && handle_ != nullptr) {
        root = handle_->config.log_dir;
    }
    return root.empty() ? std::filesystem::path{} : (root / "async");
}
 
void ExternalBridge::set_async_sentinel_root(
    const std::filesystem::path& root) {
    async_sentinel_root_override_ = root;
}
 
static const char* sentinel_suffix_for_status(
    const std::string& status) {
    const char* suffix = ".done";
    if (status == "error") {
        suffix = ".failed";
    } else if (status == "cancelled") {
        suffix = ".cancelled";
    }
    return suffix;
}
 
void ExternalBridge::write_sentinel(const std::string& task_id,
                                    const std::string& status) {
    auto dir = async_sentinel_dir();
    if (dir.empty()) { return; }
    std::error_code ec;
    std::filesystem::create_directories(dir, ec);
    if (ec) {
        logger->warn("write_sentinel: mkdir {} failed: {}",
                     dir.string(), ec.message());
        return;
    }
    auto path = dir / (task_id + sentinel_suffix_for_status(status));
    std::ofstream out(path);
    if (!out.is_open()) {
        logger->warn("write_sentinel: open {} failed", path.string());
        return;
    }
    out << status << '\n';
}
 
} // namespace entropic