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 <condition_variable>
#include <filesystem>
#include <memory>
#include <mutex>
#include <map>
#include <set>
#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; }
 
    const ExternalMCPConfig& config() const { return config_; }
 
    std::string dispatch(const std::string& request, int client_fd);
 
    // ── Turn queue (gh#144, v2.12.0) ─────────────────────────
 
    bool begin_turn_wait(long timeout_ms, long* waited_ms_out,
                         const std::string& session_key = "");
 
    void end_turn_wait(const std::string& session_key = "");
 
    void fail_task_engine_busy(const std::string& task_id, long waited_ms);
 
    void commit_async_final_state(const std::string& task_id,
                                  const std::string& status,
                                  const std::string& phase,
                                  const std::string& text);
 
    size_t turn_queue_depth() const;
 
    bool turn_in_flight() const;
 
    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,
                       const std::string& session_key = "",
                       const std::string& tier = "");
 
    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::string session_key;         
        int owner_fd = -1;               
    };
 
    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);
 
    void notify_client(int fd, 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);
 
 
    void reap_finished_clients_locked();
 
    void reap_finished_async_locked();
 
    void join_async_workers();
 
    entropic_handle_t handle_;                 
    ExternalMCPConfig config_;                 
 
    // ── Turn queue (gh#144, v2.12.0; per-session lanes v2.14.0) ──
    // Serialises every path that reaches the engine — sync ask, streaming
    // ask, AND the detached async worker — PER SESSION. Missing the async path would let
    // an async ask bypass the queue and collide with a sync one, earning an
    // ALREADY_RUNNING where the caller expected to be queued.
    //
    // Nothing else on the bridge may take turn_mutex_. In particular
    // handle_clear -> cancel_inflight_async_tasks interrupts that SESSION's
    // run, and must NOT wait behind the turn it is trying to cancel —
    // that would be gh#109's failure one layer up. It reaches the engine
    // through entropic_context_clear (api_mutex), which runs never take.
    struct TurnLane {
        uint64_t next_ticket = 0;      
        uint64_t now_serving = 0;      
        size_t waiters = 0;            
        bool busy = false;             
        std::set<uint64_t> abandoned;
    };
    mutable std::mutex turn_mutex_;            
    std::condition_variable turn_cv_;          
    std::map<std::string, TurnLane> lanes_;    
 
    std::string lane_for(const std::string& session_key) const;
 
    void async_worker(const std::string& prompt, const std::string& task_id,
                      int client_fd, const std::string& session_key,
                      const std::string& tier);
    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::mutex fd_mutex;
        std::atomic<bool> finished{false}; 
        std::thread thread;
    };
    std::vector<std::unique_ptr<ClientThread>> client_threads_;
    struct AsyncWorker {
        std::atomic<bool> finished{false}; 
        std::thread thread;                
    };
    std::vector<std::unique_ptr<AsyncWorker>> async_workers_; 
    std::mutex async_workers_mutex_;                          
    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