external_bridge.h

Go to the documentation of this file.

// SPDX-License-Identifier: Apache-2.0
#pragma once
 
#include <entropic/entropic_export.h>
#include <entropic/types/config.h>
 
#include <nlohmann/json_fwd.hpp>
 
#include <atomic>
#include <chrono>
#include <filesystem>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <unordered_map>
#include <unordered_set>
#include <vector>
 
// Forward declaration — bridge holds a raw pointer to the opaque handle.
struct entropic_engine;
using entropic_handle_t = entropic_engine*;
 
namespace entropic {
 
class ENTROPIC_EXPORT ExternalBridge {
public:
    ExternalBridge(
        entropic_handle_t handle,
        const ExternalMCPConfig& config,
        const std::filesystem::path& project_dir);
 
    ~ExternalBridge();
 
    bool start();
 
    void stop();
 
    const std::filesystem::path& socket_path() const { return socket_path_; }
 
    bool ask_streaming() const { return config_.ask_streaming; }
 
    nlohmann::json handle_ask_status(const nlohmann::json& args);
 
    void attach_phase_observer(const std::string& task_id);
 
    void detach_phase_observer();
 
    void run_async_ask(const std::string& prompt,
                       const std::string& task_id,
                       int client_fd);
 
    void write_sentinel(const std::string& task_id,
                        const std::string& status);
 
    std::filesystem::path async_sentinel_dir() const;
 
    void set_async_sentinel_root(const std::filesystem::path& root);
 
    void cleanup_expired_tasks();
 
    mutable std::mutex tasks_mutex_;
 
    struct AsyncTask {
        std::string status = "queued";   
        std::string phase = "queued";    
        std::string result;              
        std::chrono::steady_clock::time_point created; 
    };
 
    std::unordered_map<std::string, AsyncTask>& tasks_for_cancel() {
        return tasks_;
    }
 
    const std::string& active_task_id_for_observer() const {
        return active_task_id_;
    }
 
    void subscribe(int fd);
 
    void unsubscribe(int fd);
 
    void broadcast_notification(const nlohmann::json& notif);
 
    size_t subscriber_count() const {
        std::lock_guard<std::mutex> lock(subscribers_mutex_);
        return subscribers_.size();
    }
 
    void update_task_phase(const std::string& task_id,
                           const std::string& status,
                           const std::string& phase);
 
    bool observer_call_is_stale() const {
        return observer_gen_.load() != attached_gen_;
    }
 
private:
    void accept_loop();
 
    void serve_client(int client_fd);
 
    std::string dispatch(const std::string& request, int client_fd);
 
    void reap_finished_clients_locked();
 
    entropic_handle_t handle_;                 
    ExternalMCPConfig config_;                 
    std::filesystem::path socket_path_;        
    std::string bound_canonical_;              
    int listen_fd_ = -1;                       
    std::atomic<bool> running_{false};         
    std::thread accept_thread_;                
 
    struct ClientThread {
        int fd;                          
        std::atomic<bool> finished{false}; 
        std::thread thread;
    };
    std::vector<std::unique_ptr<ClientThread>> client_threads_;
    mutable std::mutex client_threads_mutex_;
 
    std::unordered_map<std::string, AsyncTask> tasks_;
 
    std::unordered_set<int> subscribers_;
    mutable std::mutex subscribers_mutex_; 
 
    std::string active_task_id_;
 
    std::atomic<uint64_t> observer_gen_{0};
 
    uint64_t attached_gen_ = 0;
 
    std::filesystem::path async_sentinel_root_override_;
};
 
} // namespace entropic