[#1146] Route HTTP commands through BusCommandRouterProcessor in-process

config/das.json

@@ -207,7 +207,8 @@
                 "thread_pool_size": 4,
                 "max_concurrent_executions": 100,
                 "max_queued_executions": 500,
-                "max_events_per_execution": 10000,
+                "max_events_per_execution": 100000,
+                "stream_items_per_chunk": 100,
                 "execution_retention_ms": 900000
             }
         }

src/agents/command_router/BUILD

@@ -48,7 +48,6 @@ cc_library(
     deps = [
         ":bus_command_router_proxy",
         ":evolution_metta_parser",
-        "//agents/command_router/http_api:command_router_http_api_singleton",
         "//agents/evolution:query_evolution_proxy",
         "//agents/query_engine:pattern_matching_query_proxy",
         "//agents/query_engine:query_answer",

src/agents/command_router/BusCommandRouterProcessor.cc

@@ -1,11 +1,13 @@
 #include "BusCommandRouterProcessor.h"
 
 #include <algorithm>
+#include <atomic>
 #include <variant>
 
 #include "BaseQueryProxy.h"
 #include "EvolutionMettaParser.h"
 #include "PatternMatchingQueryProxy.h"
+#include "PortPool.h"
 #include "QueryEvolutionProxy.h"
 #include "ServiceBusSingleton.h"
 #include "Utils.h"
@@ -21,6 +23,7 @@ using namespace service_bus;
 namespace {
 
 const string CONTEXT_KEY = "context";
+atomic<unsigned int> HTTP_REQUEST_SERIAL{1};
 
 }  // namespace
 
@@ -44,6 +47,48 @@ Properties& BusCommandRouterProcessor::parameters_for_peer(const string& peer_id
     return iterator->second;
 }
 
+void BusCommandRouterProcessor::dispatch_http_command(
+    const shared_ptr<BusCommandRouterProxy>& caller_proxy, const string& http_requestor_id) {
+    if (caller_proxy == nullptr) {
+        RAISE_ERROR("HTTP command requires a caller proxy");
+    }
+    if (caller_proxy->issued) {
+        RAISE_ERROR("Attempt to dispatch the same HTTP caller proxy twice");
+    }
+    if (http_requestor_id.empty()) {
+        RAISE_ERROR("HTTP command dispatch requires a non-empty http_requestor_id");
+    }
+
+    const unsigned int serial = HTTP_REQUEST_SERIAL.fetch_add(1);
+
+    caller_proxy->issued = true;
+    caller_proxy->requestor_id = http_requestor_id;
+    caller_proxy->serial = serial;
+    caller_proxy->proxy_port = PortPool::get_port();
+    if (caller_proxy->proxy_port == 0) {
+        RAISE_ERROR("No port is available to start HTTP caller proxy");
+    }
+    caller_proxy->setup_proxy_node();
+
+    auto processor_proxy = dynamic_pointer_cast<BusCommandRouterProxy>(factory_empty_proxy());
+    if (processor_proxy == nullptr) {
+        RAISE_ERROR("Invalid proxy type for HTTP BUS_COMMAND_ROUTER dispatch");
+    }
+    processor_proxy->proxy_port = PortPool::get_port();
+    if (processor_proxy->proxy_port == 0) {
+        RAISE_ERROR("No port is available to start HTTP processor proxy");
+    }
+    processor_proxy->requestor_id = http_requestor_id;
+    processor_proxy->serial = serial;
+    const string requestor_host = http_requestor_id.substr(0, http_requestor_id.find(':'));
+    const string processor_proxy_node_id = requestor_host + ":" + to_string(processor_proxy->proxy_port);
+    processor_proxy->setup_proxy_node(processor_proxy_node_id, caller_proxy->my_id());
+    processor_proxy->command = std::move(caller_proxy->command);
+    processor_proxy->args = std::move(caller_proxy->args);
+
+    this->run_command(processor_proxy);
+}
+
 void BusCommandRouterProcessor::run_command(shared_ptr<BusCommandProxy> proxy) {
     auto router_proxy = dynamic_pointer_cast<BusCommandRouterProxy>(proxy);
     if (router_proxy == nullptr) {

src/agents/command_router/BusCommandRouterProcessor.h

@@ -31,6 +31,10 @@ class BusCommandRouterProcessor : public BusCommandProcessor {
     shared_ptr<BusCommandProxy> factory_empty_proxy() override;
     void run_command(shared_ptr<BusCommandProxy> proxy) override;
 
+    /** Run a router command in-process for HTTP; responses are written to caller_proxy. */
+    void dispatch_http_command(const shared_ptr<BusCommandRouterProxy>& caller_proxy,
+                               const string& http_requestor_id);
+
    private:
     void handle_get(shared_ptr<BusCommandRouterProxy> proxy, const string& arg);
     void handle_set(shared_ptr<BusCommandRouterProxy> proxy, const string& arg);

src/agents/command_router/http_api/BUILD

@@ -12,6 +12,18 @@ cc_library(
     ],
 )
 
+cc_library(
+    name = "bus_command_router_proxy_stream_poller",
+    srcs = ["BusCommandRouterProxyStreamPoller.cc"],
+    hdrs = ["BusCommandRouterProxyStreamPoller.h"],
+    includes = ["."],
+    deps = [
+        "//agents:base_query_proxy",
+        "//agents/command_router:bus_command_router_proxy",
+        "//commons:commons_lib",
+    ],
+)
+
 cc_library(
     name = "command_router_http_api_config",
     srcs = ["CommandRouterHttpAPIConfig.cc"],
@@ -29,8 +41,11 @@ cc_library(
     hdrs = ["CommandRouterHttpAPI.h"],
     includes = ["."],
     deps = [
+        ":bus_command_router_proxy_stream_poller",
         ":command_execution",
         ":command_router_http_api_config",
+        "//agents/command_router:bus_command_router_processor",
+        "//agents/command_router:bus_command_router_proxy",
         "//commons:commons_lib",
         "//commons/processor:processor_lib",
         "//hasher:hasher_lib",
@@ -45,7 +60,8 @@ cc_library(
     deps = [
         ":command_router_http_api",
         ":command_router_http_api_config",
-        "//commons:commons_lib",
+        "//agents/command_router:bus_command_router_processor",
         "//commons/processor:processor_lib",
+        "//service_bus:bus_command_processor",
     ],
 )

src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc

@@ -0,0 +1,171 @@
+#include "BusCommandRouterProxyStreamPoller.h"
+
+#include "BaseQueryProxy.h"
+#include "Utils.h"
+
+using namespace command_router;
+using namespace commons;
+using namespace agents;
+
+namespace {
+
+void emit_chunk(const function<void(const vector<string>& chunk)>& on_chunk,
+                vector<string>& chunk_data) {
+    if (!chunk_data.empty() && on_chunk) {
+        on_chunk(chunk_data);
+        chunk_data.clear();
+    }
+}
+
+void append_answer_chunk(const shared_ptr<BusCommandRouterProxy>& router_proxy,
+                         bool use_metta_as_query_tokens,
+                         size_t items_per_chunk,
+                         const function<void(const vector<string>& chunk)>& on_chunk,
+                         vector<string>& chunk_data) {
+    shared_ptr<QueryAnswer> answer;
+    while ((answer = router_proxy->pop()) != nullptr) {
+        chunk_data.push_back(answer->to_string(use_metta_as_query_tokens));
+        if (chunk_data.size() >= items_per_chunk) {
+            emit_chunk(on_chunk, chunk_data);
+        }
+    }
+    emit_chunk(on_chunk, chunk_data);
+}
+
+}  // namespace
+
+bool BusCommandRouterProxyStreamPoller::poll_stream(
+    const shared_ptr<BusCommandRouterProxy>& router_proxy,
+    const string& command_type,
+    size_t items_per_chunk,
+    const function<bool()>& should_abort,
+    const function<void(const vector<string>& chunk)>& on_chunk,
+    const function<void(const string& error)>& on_error,
+    const function<void()>& on_aborted) {
+    if (items_per_chunk == 0) {
+        if (on_error) {
+            on_error("items_per_chunk must be at least 1");
+        }
+        return false;
+    }
+
+    if (!router_proxy) {
+        if (on_error) {
+            on_error("router_proxy must not be null");
+        }
+        return false;
+    }
+
+    auto finished_or_error = [&]() { return router_proxy->finished() || router_proxy->error_flag; };
+
+    auto handle_abort = [&]() -> bool {
+        if (should_abort && should_abort()) {
+            router_proxy->abort();
+            if (on_aborted) {
+                on_aborted();
+            }
+            return true;
+        }
+        return false;
+    };
+
+    if (command_type == "get") {
+        while (router_proxy->params_response.empty() && !finished_or_error()) {
+            if (handle_abort()) {
+                return false;
+            }
+            Utils::sleep(100);
+        }
+        if (router_proxy->error_flag) {
+            if (on_error) {
+                on_error(router_proxy->error_message);
+            }
+            return false;
+        }
+        if (router_proxy->params_response.empty()) {
+            if (on_error) {
+                on_error("GET command finished without params response");
+            }
+            return false;
+        }
+        if (on_chunk) {
+            on_chunk({router_proxy->params_response});
+        }
+        return true;
+    }
+
+    if (command_type == "set") {
+        while (router_proxy->set_param_ack.empty() && !finished_or_error()) {
+            if (handle_abort()) {
+                return false;
+            }
+            Utils::sleep(100);
+        }
+        if (router_proxy->error_flag) {
+            if (on_error) {
+                on_error(router_proxy->error_message);
+            }
+            return false;
+        }
+        if (router_proxy->set_param_ack.empty()) {
+            if (on_error) {
+                on_error("SET command finished without parameter ack");
+            }
+            return false;
+        }
+        if (on_chunk) {
+            on_chunk({router_proxy->set_param_ack});
+        }
+        return true;
+    }
+
+    if (command_type == "query" || command_type == "evolution") {
+        while (!router_proxy->routed_flag && !finished_or_error()) {
+            if (handle_abort()) {
+                return false;
+            }
+            Utils::sleep(100);
+        }
+        if (router_proxy->error_flag) {
+            if (on_error) {
+                on_error(router_proxy->error_message);
+            }
+            return false;
+        }
+        if (!router_proxy->routed_flag) {
+            if (on_error) {
+                on_error("Command finished without being routed to a downstream service");
+            }
+            return false;
+        }
+
+        const bool use_metta_as_query_tokens =
+            router_proxy->parameters.get_or<bool>(BaseQueryProxy::USE_METTA_AS_QUERY_TOKENS, true);
+
+        vector<string> chunk_data;
+        while (!finished_or_error()) {
+            if (handle_abort()) {
+                return false;
+            }
+
+            append_answer_chunk(
+                router_proxy, use_metta_as_query_tokens, items_per_chunk, on_chunk, chunk_data);
+            Utils::sleep(100);
+        }
+        append_answer_chunk(
+            router_proxy, use_metta_as_query_tokens, items_per_chunk, on_chunk, chunk_data);
+
+        if (router_proxy->error_flag) {
+            if (on_error) {
+                on_error(router_proxy->error_message);
+            }
+            return false;
+        }
+        return true;
+    }
+
+    if (on_error) {
+        on_error("Unknown command_type: " + command_type);
+    }
+    return false;
+}

src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.h

@@ -0,0 +1,53 @@
+#pragma once
+
+#include <functional>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "BusCommandRouterProxy.h"
+
+using namespace std;
+
+namespace command_router {
+
+/**
+ * Polls a BusCommandRouterProxy and delivers results in HTTP-friendly chunks.
+ *
+ * Used by CommandRouterHttpAPI to stream command output over WebSocket. The terminal
+ * bus client keeps its own inline polling loop instead of this helper.
+ */
+class BusCommandRouterProxyStreamPoller {
+   public:
+    /**
+     * Poll router_proxy until the command finishes, is aborted, or fails.
+     *
+     * For command_type "query" and "evolution", answers are popped from the proxy
+     * and forwarded in batches of at most items_per_chunk strings. For "get" and
+     * "set", a single chunk is emitted once the proxy response is ready.
+     *
+     * @param router_proxy Proxy already issued on the service bus.
+     * @param command_type Router command: "get", "set", "query", or "evolution".
+     * @param items_per_chunk Maximum answers per on_chunk call for query/evolution.
+     *                        Must be at least 1.
+     * @param should_abort Optional callback; when it returns true, the proxy is aborted
+     *                     and on_aborted is invoked.
+     * @param on_chunk Called with each batch of serialized answers or response payload.
+     * @param on_error Called with an error message on validation, proxy, or unknown
+     *                 command failures.
+     * @param on_aborted Called when polling stops because should_abort returned true.
+     * @return true when the command completed without error; false otherwise.
+     */
+    static bool poll_stream(const shared_ptr<BusCommandRouterProxy>& router_proxy,
+                            const string& command_type,
+                            size_t items_per_chunk,
+                            const function<bool()>& should_abort,
+                            const function<void(const vector<string>& chunk)>& on_chunk,
+                            const function<void(const string& error)>& on_error,
+                            const function<void()>& on_aborted);
+
+   private:
+    BusCommandRouterProxyStreamPoller() = delete;
+};
+
+}  // namespace command_router

src/agents/command_router/http_api/CommandExecution.cc

@@ -57,7 +57,7 @@ int CommandExecution::received_count() const {
     return this->received_count_;
 }
 
-long long CommandExecution::finished_at_ms() const {
+unsigned long CommandExecution::finished_at_ms() const {
     lock_guard<mutex> lock(this->mtx_);
     return this->finished_at_ms_;
 }
@@ -111,7 +111,7 @@ optional<string> CommandExecution::wait_next_event(size_t& next_index,
     return payload;
 }
 
-bool CommandExecution::is_retention_expired(long long now_ms, long long retention_ms) const {
+bool CommandExecution::is_retention_expired(unsigned long now_ms, unsigned long retention_ms) const {
     lock_guard<mutex> lock(this->mtx_);
     return is_terminal(this->status_) && this->finished_at_ms_ > 0 &&
            (now_ms - this->finished_at_ms_) > retention_ms;
@@ -151,7 +151,7 @@ void CommandExecution::publish_chunk(int seq, const vector<string>& data) {
                                 {"received_count", this->received_count_}});
 }
 
-void CommandExecution::mark_completed(long long duration_ms, int total_items) {
+void CommandExecution::mark_completed(unsigned long duration_ms, int total_items) {
     lock_guard<mutex> lock(this->mtx_);
     this->duration_ms_ = duration_ms;
     this->total_items_ = total_items;