From be8d2583b78108fb3c1562ef1cd0ca960e05d7e0 Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Thu, 25 Jun 2026 21:16:50 -0300 Subject: [PATCH 01/16] Working, but modifying the Service Bus --- src/agents/command_router/BUILD | 2 + .../command_router/CommandRouterHttpAPI.cc | 116 +++++++++++++++--- src/service_bus/ServiceBus.cc | 44 ++++--- 3 files changed, 131 insertions(+), 31 deletions(-) diff --git a/src/agents/command_router/BUILD b/src/agents/command_router/BUILD index a1a108e2..bfaa30e0 100644 --- a/src/agents/command_router/BUILD +++ b/src/agents/command_router/BUILD @@ -74,10 +74,12 @@ cc_library( hdrs = ["CommandRouterHttpAPI.h"], includes = ["."], deps = [ + ":bus_command_router_proxy", ":command_execution", "//commons:commons_lib", "//commons/processor:processor_lib", "//hasher:hasher_lib", + "//service_bus:service_bus_singleton", ], ) diff --git a/src/agents/command_router/CommandRouterHttpAPI.cc b/src/agents/command_router/CommandRouterHttpAPI.cc index b0dce6a9..f30f7817 100644 --- a/src/agents/command_router/CommandRouterHttpAPI.cc +++ b/src/agents/command_router/CommandRouterHttpAPI.cc @@ -5,13 +5,18 @@ #include #define LOG_LEVEL INFO_LEVEL +#include "BaseQueryProxy.h" +#include "BusCommandRouterProxy.h" #include "Logger.h" +#include "ServiceBusSingleton.h" #include "Utils.h" #include "expression_hasher.h" using namespace command_router; using namespace commons; using namespace processor; +using namespace agents; +using namespace service_bus; using json = nlohmann::json; @@ -349,32 +354,109 @@ shared_ptr CommandRouterHttpAPI::find_execution(const string& } void CommandRouterHttpAPI::run_execution_inner(const shared_ptr& exec) { - // POC code to simulate event streaming. - - const int TOTAL = 1000; - const int CHUNK_SIZE = 10; - auto started_at = Utils::get_current_time_millis(); + const auto started_at = Utils::get_current_time_millis(); int seq = 0; - for (int i = 0; i < TOTAL; i += CHUNK_SIZE) { - { - lock_guard lock(exec->mtx); - if (exec->cancel_requested || this->shutting_down.load()) { - exec->mark_aborted(); - LOG_INFO("CommandRouter HTTP API execution aborted id=" << exec->execution_id); + auto should_stop = [&]() -> bool { + lock_guard lock(exec->mtx); + return exec->cancel_requested || this->shutting_down.load(); + }; + + auto abort_execution = [&](BusCommandRouterProxy& router_proxy) { + router_proxy.abort(); + lock_guard lock(exec->mtx); + exec->mark_aborted(); + LOG_INFO("CommandRouter HTTP API execution aborted id=" << exec->execution_id); + }; + + auto mark_proxy_error = [&](BusCommandRouterProxy& router_proxy) { + lock_guard lock(exec->mtx); + exec->mark_error(router_proxy.error_message); + }; + + string router_arg = exec->command_text; + Utils::replace_all(router_arg, "%", "$"); + + auto proxy = make_shared(exec->command_type, router_arg); + auto router_proxy = dynamic_pointer_cast(proxy); + ServiceBusSingleton::get_instance()->issue_bus_command(proxy); + + auto finished_or_error = [&]() { + return router_proxy->finished() || router_proxy->error_flag; + }; + + const string& command = exec->command_type; + + if (command == "get") { + while (router_proxy->params_response.empty() && !finished_or_error()) { + if (should_stop()) { + abort_execution(*router_proxy); + return; + } + Utils::sleep(100); + } + if (router_proxy->error_flag) { + mark_proxy_error(*router_proxy); + return; + } + lock_guard lock(exec->mtx); + exec->publish_chunk(++seq, {router_proxy->params_response}); + } else if (command == "set") { + while (router_proxy->set_param_ack.empty() && !finished_or_error()) { + if (should_stop()) { + abort_execution(*router_proxy); return; } + Utils::sleep(100); + } + if (router_proxy->error_flag) { + mark_proxy_error(*router_proxy); + return; + } + lock_guard lock(exec->mtx); + exec->publish_chunk(++seq, {router_proxy->set_param_ack}); + } else if (command == "query" || command == "evolution") { + while (!router_proxy->routed_flag && !finished_or_error()) { + if (should_stop()) { + abort_execution(*router_proxy); + return; + } + Utils::sleep(100); + } + if (router_proxy->error_flag) { + mark_proxy_error(*router_proxy); + return; } - Utils::sleep(200); + bool use_metta_as_query_tokens = + router_proxy->parameters.get_or(BaseQueryProxy::USE_METTA_AS_QUERY_TOKENS, true); - vector chunk_data; - for (int k = i; k < min(i + CHUNK_SIZE, TOTAL); ++k) { - chunk_data.push_back("result-" + std::to_string(k)); - } + while (!router_proxy->finished()) { + if (should_stop()) { + abort_execution(*router_proxy); + return; + } + vector chunk_data; + shared_ptr answer; + while ((answer = router_proxy->pop()) != nullptr) { + chunk_data.push_back(answer->to_string(use_metta_as_query_tokens)); + } + + if (!chunk_data.empty()) { + lock_guard lock(exec->mtx); + exec->publish_chunk(++seq, chunk_data); + } + Utils::sleep(100); + } + if (router_proxy->error_flag) { + mark_proxy_error(*router_proxy); + return; + } + } else { lock_guard lock(exec->mtx); - exec->publish_chunk(++seq, chunk_data); + exec->mark_error("Unknown command_type: " + command); + return; } const auto duration_ms = Utils::get_current_time_millis() - started_at; diff --git a/src/service_bus/ServiceBus.cc b/src/service_bus/ServiceBus.cc index 762bd16d..65875626 100644 --- a/src/service_bus/ServiceBus.cc +++ b/src/service_bus/ServiceBus.cc @@ -75,20 +75,25 @@ void ServiceBus::register_processor(shared_ptr processor) { } void ServiceBus::issue_bus_command(shared_ptr proxy) { - lock_guard semaphore(this->api_mutex); - LOG_DEBUG(bus_node->node_id() << " is issuing BUS command <" << proxy->command << ">"); - if (proxy->issued) { - RAISE_ERROR("Attempt to issue the same proxy twice"); - } - proxy->issued = true; - proxy->requestor_id = this->bus_node->node_id(); - proxy->serial = this->next_request_serial++; - proxy->proxy_port = PortPool::get_port(); - if (proxy->proxy_port == 0) { - RAISE_ERROR("No port is available to start bus command proxy"); - } else { + string command; + vector args; + bool dispatch_locally = false; + shared_ptr node = this->bus_node; + + { + lock_guard semaphore(this->api_mutex); + LOG_DEBUG(bus_node->node_id() << " is issuing BUS command <" << proxy->command << ">"); + if (proxy->issued) { + RAISE_ERROR("Attempt to issue the same proxy twice"); + } + proxy->issued = true; + proxy->requestor_id = this->bus_node->node_id(); + proxy->serial = this->next_request_serial++; + proxy->proxy_port = PortPool::get_port(); + if (proxy->proxy_port == 0) { + RAISE_ERROR("No port is available to start bus command proxy"); + } proxy->setup_proxy_node(); - vector args; args.push_back(proxy->requestor_id); args.push_back(to_string(proxy->serial)); args.push_back(proxy->proxy_node->node_id()); @@ -96,7 +101,18 @@ void ServiceBus::issue_bus_command(shared_ptr proxy) { for (auto arg : proxy->args) { args.push_back(arg); } - this->bus_node->send_bus_command(proxy->command, args); + command = proxy->command; + dispatch_locally = this->bus_node->get_ownership(command) == this->bus_node->node_id(); + if (!dispatch_locally) { + this->bus_node->send_bus_command(command, args); + } + } + + if (dispatch_locally) { + LOG_DEBUG(bus_node->node_id() << " is dispatching BUS command <" << command << "> locally"); + shared_ptr message = + shared_ptr(new BusCommandMessage(command, args)); + message->act(node); } } From 46215352a912d5103f70f0b20fe084b4246179f5 Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Fri, 26 Jun 2026 12:04:46 -0300 Subject: [PATCH 02/16] rollback --- src/service_bus/ServiceBus.cc | 44 +++++++++++------------------------ 1 file changed, 14 insertions(+), 30 deletions(-) diff --git a/src/service_bus/ServiceBus.cc b/src/service_bus/ServiceBus.cc index 65875626..762bd16d 100644 --- a/src/service_bus/ServiceBus.cc +++ b/src/service_bus/ServiceBus.cc @@ -75,25 +75,20 @@ void ServiceBus::register_processor(shared_ptr processor) { } void ServiceBus::issue_bus_command(shared_ptr proxy) { - string command; - vector args; - bool dispatch_locally = false; - shared_ptr node = this->bus_node; - - { - lock_guard semaphore(this->api_mutex); - LOG_DEBUG(bus_node->node_id() << " is issuing BUS command <" << proxy->command << ">"); - if (proxy->issued) { - RAISE_ERROR("Attempt to issue the same proxy twice"); - } - proxy->issued = true; - proxy->requestor_id = this->bus_node->node_id(); - proxy->serial = this->next_request_serial++; - proxy->proxy_port = PortPool::get_port(); - if (proxy->proxy_port == 0) { - RAISE_ERROR("No port is available to start bus command proxy"); - } + lock_guard semaphore(this->api_mutex); + LOG_DEBUG(bus_node->node_id() << " is issuing BUS command <" << proxy->command << ">"); + if (proxy->issued) { + RAISE_ERROR("Attempt to issue the same proxy twice"); + } + proxy->issued = true; + proxy->requestor_id = this->bus_node->node_id(); + proxy->serial = this->next_request_serial++; + proxy->proxy_port = PortPool::get_port(); + if (proxy->proxy_port == 0) { + RAISE_ERROR("No port is available to start bus command proxy"); + } else { proxy->setup_proxy_node(); + vector args; args.push_back(proxy->requestor_id); args.push_back(to_string(proxy->serial)); args.push_back(proxy->proxy_node->node_id()); @@ -101,18 +96,7 @@ void ServiceBus::issue_bus_command(shared_ptr proxy) { for (auto arg : proxy->args) { args.push_back(arg); } - command = proxy->command; - dispatch_locally = this->bus_node->get_ownership(command) == this->bus_node->node_id(); - if (!dispatch_locally) { - this->bus_node->send_bus_command(command, args); - } - } - - if (dispatch_locally) { - LOG_DEBUG(bus_node->node_id() << " is dispatching BUS command <" << command << "> locally"); - shared_ptr message = - shared_ptr(new BusCommandMessage(command, args)); - message->act(node); + this->bus_node->send_bus_command(proxy->command, args); } } From fc99c7747b90f7ca34fd49ed4a6857f0a927c7fa Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Fri, 26 Jun 2026 12:13:33 -0300 Subject: [PATCH 03/16] rollback --- src/tests/cpp/bus_command_router_test.cc | 1 - 1 file changed, 1 deletion(-) diff --git a/src/tests/cpp/bus_command_router_test.cc b/src/tests/cpp/bus_command_router_test.cc index 19f35dd1..4c7acd76 100644 --- a/src/tests/cpp/bus_command_router_test.cc +++ b/src/tests/cpp/bus_command_router_test.cc @@ -1,7 +1,6 @@ #include "AtomDBSingleton.h" #include "BusCommandRouterProcessor.h" #include "BusCommandRouterProxy.h" -#include "CommandRouterHttpAPISingleton.h" #include "EvolutionMettaParser.h" #include "ServiceBus.h" #include "TestAtomDBJsonConfig.h" From a0a03e117b736e0e8319fb207976c94f2544a3d7 Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Fri, 26 Jun 2026 12:21:48 -0300 Subject: [PATCH 04/16] Add HTTP API streaming and config modules for command router. Extract BusCommandRouterProxyClient and CommandRouterHttpAPIConfig so the HTTP API uses a dedicated issuer ServiceBus with configurable WebSocket chunk size, while bus_client keeps its original terminal behavior. --- config/das.json | 7 +- src/agents/command_router/BUILD | 31 ++- .../BusCommandRouterProxyClient.cc | 144 ++++++++++++ .../BusCommandRouterProxyClient.h | 32 +++ .../command_router/CommandRouterHttpAPI.cc | 168 ++++---------- .../command_router/CommandRouterHttpAPI.h | 21 +- .../CommandRouterHttpAPIConfig.cc | 68 ++++++ .../CommandRouterHttpAPIConfig.h | 33 +++ .../CommandRouterHttpAPISingleton.cc | 40 +--- src/main/BUILD | 1 + src/tests/cpp/BUILD | 19 +- src/tests/cpp/command_router_http_api_test.cc | 210 +++++++++++++++++- 12 files changed, 600 insertions(+), 174 deletions(-) create mode 100644 src/agents/command_router/BusCommandRouterProxyClient.cc create mode 100644 src/agents/command_router/BusCommandRouterProxyClient.h create mode 100644 src/agents/command_router/CommandRouterHttpAPIConfig.cc create mode 100644 src/agents/command_router/CommandRouterHttpAPIConfig.h diff --git a/config/das.json b/config/das.json index 14875d0e..0d044343 100644 --- a/config/das.json +++ b/config/das.json @@ -134,8 +134,8 @@ "max_bundle_size": 1000, "max_answers": 0, "use_link_template_cache": false, - "populate_metta_mapping": false, - "use_metta_as_query_tokens": false, + "populate_metta_mapping": true, + "use_metta_as_query_tokens": true, "allow_incomplete_chain_path": false } }, @@ -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 } } diff --git a/src/agents/command_router/BUILD b/src/agents/command_router/BUILD index bfaa30e0..65399111 100644 --- a/src/agents/command_router/BUILD +++ b/src/agents/command_router/BUILD @@ -27,6 +27,29 @@ cc_library( ], ) +cc_library( + name = "bus_command_router_proxy_client", + srcs = ["BusCommandRouterProxyClient.cc"], + hdrs = ["BusCommandRouterProxyClient.h"], + includes = ["."], + deps = [ + ":bus_command_router_proxy", + "//agents:base_query_proxy", + "//commons:commons_lib", + ], +) + +cc_library( + name = "command_router_http_api_config", + srcs = ["CommandRouterHttpAPIConfig.cc"], + hdrs = ["CommandRouterHttpAPIConfig.h"], + includes = ["."], + deps = [ + ":command_execution", + "//commons:commons_lib", + ], +) + cc_library( name = "evolution_metta_parser", srcs = ["EvolutionMettaParser.cc"], @@ -47,7 +70,6 @@ cc_library( includes = ["."], deps = [ ":bus_command_router_proxy", - ":command_router_http_api_singleton", ":evolution_metta_parser", "//agents/evolution:query_evolution_proxy", "//agents/query_engine:pattern_matching_query_proxy", @@ -75,11 +97,13 @@ cc_library( includes = ["."], deps = [ ":bus_command_router_proxy", + ":bus_command_router_proxy_client", ":command_execution", + ":command_router_http_api_config", "//commons:commons_lib", "//commons/processor:processor_lib", "//hasher:hasher_lib", - "//service_bus:service_bus_singleton", + "//service_bus:service_bus_lib", ], ) @@ -90,7 +114,8 @@ cc_library( includes = ["."], deps = [ ":command_router_http_api", - "//commons:commons_lib", + ":command_router_http_api_config", "//commons/processor:processor_lib", + "//service_bus:service_bus_lib", ], ) diff --git a/src/agents/command_router/BusCommandRouterProxyClient.cc b/src/agents/command_router/BusCommandRouterProxyClient.cc new file mode 100644 index 00000000..1697fd18 --- /dev/null +++ b/src/agents/command_router/BusCommandRouterProxyClient.cc @@ -0,0 +1,144 @@ +#include "BusCommandRouterProxyClient.h" + +#include "BaseQueryProxy.h" +#include "Utils.h" + +using namespace command_router; +using namespace commons; +using namespace agents; + +namespace { + +void emit_chunk(const function& chunk)>& on_chunk, vector& chunk_data) { + if (!chunk_data.empty() && on_chunk) { + on_chunk(chunk_data); + chunk_data.clear(); + } +} + +void append_answer_chunk(const shared_ptr& router_proxy, + bool use_metta_as_query_tokens, + size_t items_per_chunk, + const function& chunk)>& on_chunk, + vector& chunk_data) { + shared_ptr 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 command_router::poll_router_proxy_stream(const shared_ptr& router_proxy, + const string& command_type, + size_t items_per_chunk, + const function& should_abort, + const function& chunk)>& on_chunk, + const function& on_error, + const function& on_aborted) { + if (items_per_chunk == 0) { + if (on_error) { + on_error("items_per_chunk must be at least 1"); + } + 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 (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 (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; + } + + const bool use_metta_as_query_tokens = + router_proxy->parameters.get_or(BaseQueryProxy::USE_METTA_AS_QUERY_TOKENS, true); + + vector chunk_data; + while (!router_proxy->finished()) { + 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; +} diff --git a/src/agents/command_router/BusCommandRouterProxyClient.h b/src/agents/command_router/BusCommandRouterProxyClient.h new file mode 100644 index 00000000..6426e99f --- /dev/null +++ b/src/agents/command_router/BusCommandRouterProxyClient.h @@ -0,0 +1,32 @@ +#pragma once + +#include +#include +#include +#include + +#include "BusCommandRouterProxy.h" + +using namespace std; + +namespace command_router { + +/** + * Poll a BusCommandRouterProxy for HTTP streaming consumers. + * + * Emits at most @p items_per_chunk answers per on_chunk callback so clients can + * iterate results incrementally (1-by-1 or N-by-N). Terminal bus_client does + * not use this; it keeps its own polling loop and query-engine limits. + * + * @param items_per_chunk Minimum 1. Each callback receives at most this many items. + * @return true on success, false on error or abort. + */ +bool poll_router_proxy_stream(const shared_ptr& router_proxy, + const string& command_type, + size_t items_per_chunk, + const function& should_abort, + const function& chunk)>& on_chunk, + const function& on_error, + const function& on_aborted); + +} // namespace command_router diff --git a/src/agents/command_router/CommandRouterHttpAPI.cc b/src/agents/command_router/CommandRouterHttpAPI.cc index f30f7817..78f79fb1 100644 --- a/src/agents/command_router/CommandRouterHttpAPI.cc +++ b/src/agents/command_router/CommandRouterHttpAPI.cc @@ -4,55 +4,38 @@ #include #include -#define LOG_LEVEL INFO_LEVEL -#include "BaseQueryProxy.h" #include "BusCommandRouterProxy.h" +#include "BusCommandRouterProxyClient.h" + +#define LOG_LEVEL INFO_LEVEL #include "Logger.h" -#include "ServiceBusSingleton.h" #include "Utils.h" #include "expression_hasher.h" using namespace command_router; using namespace commons; using namespace processor; -using namespace agents; using namespace service_bus; +using namespace agents; using json = nlohmann::json; const unordered_set CommandRouterHttpAPI::VALID_COMMAND_TYPES = { "query", "evolution", "get", "set"}; -namespace { - -HttpAPISettings load_http_api_settings(const JsonConfig& command_router_config) { - HttpAPISettings settings; - settings.max_concurrent_executions = - command_router_config.at_path("http_api.max_concurrent_executions") - .get_or(settings.max_concurrent_executions); - settings.max_queued_executions = command_router_config.at_path("http_api.max_queued_executions") - .get_or(settings.max_queued_executions); - settings.max_events_per_execution = - command_router_config.at_path("http_api.max_events_per_execution") - .get_or(settings.max_events_per_execution); - settings.execution_retention_ms = command_router_config.at_path("http_api.execution_retention_ms") - .get_or(settings.execution_retention_ms); - return settings; -} - -} // namespace - // ------------------------------------------------------------------------------------------------- // Constructors, destructors CommandRouterHttpAPI::CommandRouterHttpAPI(const string& host, int port, shared_ptr thread_pool, - HttpAPISettings settings) + HttpAPISettings settings, + shared_ptr issuer_bus) : Processor("command_router_http_api"), host(host), port(port), thread_pool(thread_pool), + issuer_bus(issuer_bus), settings(std::move(settings)) { if (this->thread_pool == nullptr) { RAISE_ERROR("CommandRouterHttpAPI requires a non-null thread pool"); @@ -75,10 +58,6 @@ void CommandRouterHttpAPI::initialize(shared_ptr instance, << " listening on " << instance->host << ":" << instance->port); } -HttpAPISettings CommandRouterHttpAPI::settings_from_config(const JsonConfig& command_router_config) { - return load_http_api_settings(command_router_config); -} - bool CommandRouterHttpAPI::thread_one_step() { if (this->shutting_down.load()) return false; @@ -354,118 +333,61 @@ shared_ptr CommandRouterHttpAPI::find_execution(const string& } void CommandRouterHttpAPI::run_execution_inner(const shared_ptr& exec) { + if (this->issuer_bus == nullptr) { + lock_guard lock(exec->mtx); + exec->mark_error("Issuer ServiceBus is not configured for command execution"); + return; + } + const auto started_at = Utils::get_current_time_millis(); int seq = 0; + int received_count = 0; - auto should_stop = [&]() -> bool { + auto should_abort = [&]() { lock_guard lock(exec->mtx); return exec->cancel_requested || this->shutting_down.load(); }; - - auto abort_execution = [&](BusCommandRouterProxy& router_proxy) { - router_proxy.abort(); + auto on_chunk = [&](const vector& chunk) { + lock_guard lock(exec->mtx); + received_count += static_cast(chunk.size()); + exec->publish_chunk(++seq, chunk); + }; + auto on_error = [&](const string& message) { + lock_guard lock(exec->mtx); + exec->mark_error(message); + }; + auto on_aborted = [&]() { lock_guard lock(exec->mtx); exec->mark_aborted(); LOG_INFO("CommandRouter HTTP API execution aborted id=" << exec->execution_id); }; - - auto mark_proxy_error = [&](BusCommandRouterProxy& router_proxy) { + auto on_complete = [&](long long duration_ms, int total_items) { lock_guard lock(exec->mtx); - exec->mark_error(router_proxy.error_message); - }; - - string router_arg = exec->command_text; - Utils::replace_all(router_arg, "%", "$"); - - auto proxy = make_shared(exec->command_type, router_arg); - auto router_proxy = dynamic_pointer_cast(proxy); - ServiceBusSingleton::get_instance()->issue_bus_command(proxy); - - auto finished_or_error = [&]() { - return router_proxy->finished() || router_proxy->error_flag; + exec->mark_completed(duration_ms, total_items); + LOG_INFO("CommandRouter HTTP API execution completed id=" + << exec->execution_id << " duration_ms=" << duration_ms << " items=" << total_items); }; - const string& command = exec->command_type; - - if (command == "get") { - while (router_proxy->params_response.empty() && !finished_or_error()) { - if (should_stop()) { - abort_execution(*router_proxy); - return; - } - Utils::sleep(100); - } - if (router_proxy->error_flag) { - mark_proxy_error(*router_proxy); - return; - } - lock_guard lock(exec->mtx); - exec->publish_chunk(++seq, {router_proxy->params_response}); - } else if (command == "set") { - while (router_proxy->set_param_ack.empty() && !finished_or_error()) { - if (should_stop()) { - abort_execution(*router_proxy); - return; - } - Utils::sleep(100); - } - if (router_proxy->error_flag) { - mark_proxy_error(*router_proxy); - return; - } - lock_guard lock(exec->mtx); - exec->publish_chunk(++seq, {router_proxy->set_param_ack}); - } else if (command == "query" || command == "evolution") { - while (!router_proxy->routed_flag && !finished_or_error()) { - if (should_stop()) { - abort_execution(*router_proxy); - return; - } - Utils::sleep(100); - } - if (router_proxy->error_flag) { - mark_proxy_error(*router_proxy); + try { + string router_arg = exec->command_text; + Utils::replace_all(router_arg, "%", "$"); + auto router_proxy = make_shared(exec->command_type, router_arg); + this->issuer_bus->issue_bus_command(router_proxy); + + if (!poll_router_proxy_stream(router_proxy, + exec->command_type, + this->settings.stream_items_per_chunk, + should_abort, + on_chunk, + on_error, + on_aborted)) { return; } - bool use_metta_as_query_tokens = - router_proxy->parameters.get_or(BaseQueryProxy::USE_METTA_AS_QUERY_TOKENS, true); - - while (!router_proxy->finished()) { - if (should_stop()) { - abort_execution(*router_proxy); - return; - } - - vector chunk_data; - shared_ptr answer; - while ((answer = router_proxy->pop()) != nullptr) { - chunk_data.push_back(answer->to_string(use_metta_as_query_tokens)); - } - - if (!chunk_data.empty()) { - lock_guard lock(exec->mtx); - exec->publish_chunk(++seq, chunk_data); - } - Utils::sleep(100); - } - if (router_proxy->error_flag) { - mark_proxy_error(*router_proxy); - return; - } - } else { - lock_guard lock(exec->mtx); - exec->mark_error("Unknown command_type: " + command); - return; + on_complete(Utils::get_current_time_millis() - started_at, received_count); + } catch (const exception& e) { + on_error(e.what()); } - - const auto duration_ms = Utils::get_current_time_millis() - started_at; - lock_guard lock(exec->mtx); - exec->mark_completed(duration_ms, exec->received_count); - - LOG_INFO("CommandRouter HTTP API execution completed id=" << exec->execution_id - << " duration_ms=" << duration_ms - << " items=" << exec->received_count); } void CommandRouterHttpAPI::run_execution(const shared_ptr& exec) { diff --git a/src/agents/command_router/CommandRouterHttpAPI.h b/src/agents/command_router/CommandRouterHttpAPI.h index 95c68fc1..620a304b 100644 --- a/src/agents/command_router/CommandRouterHttpAPI.h +++ b/src/agents/command_router/CommandRouterHttpAPI.h @@ -10,27 +10,22 @@ #include #include "CommandExecution.h" +#include "CommandRouterHttpAPIConfig.h" #include "DedicatedThread.h" -#include "JsonConfig.h" #include "Processor.h" +#include "ServiceBus.h" #include "httplib.h" #include "nlohmann/json.hpp" #include "processor/ThreadPool.h" using namespace std; using namespace commons; +using namespace service_bus; using json = nlohmann::json; namespace command_router { -struct HttpAPISettings { - size_t max_concurrent_executions = 100; - size_t max_queued_executions = 500; - size_t max_events_per_execution = CommandExecution::DEFAULT_MAX_EVENTS; - long long execution_retention_ms = 15 * 60 * 1000; -}; - /** * @brief HTTP + WebSocket server for asynchronous command execution. * @@ -40,6 +35,9 @@ struct HttpAPISettings { * POST /command-router/executions/{id}/cancel — request cancel * WS /command-router/ws/{id} — stream JSON events * + * Command execution uses a dedicated issuer ServiceBus (same path as bus_client) because + * the router node cannot issue bus_command_router to itself on ServiceBusSingleton. + * * Runs on a DedicatedThread: thread_one_step() blocks in listen() until stop(). * Each accepted command is enqueued on thread_pool so the listener stays free. * Terminal executions stay in executions until execution_retention_ms expires. @@ -58,7 +56,8 @@ class CommandRouterHttpAPI : public processor::Processor, public processor::Thre CommandRouterHttpAPI(const string& host, int port, shared_ptr thread_pool, - HttpAPISettings settings = {}); + HttpAPISettings settings = {}, + shared_ptr issuer_bus = nullptr); /** @brief Calls stop(). */ ~CommandRouterHttpAPI() override; @@ -70,9 +69,6 @@ class CommandRouterHttpAPI : public processor::Processor, public processor::Thre static void initialize(shared_ptr instance, vector> additional_subprocessors); - /** @brief Load HttpAPISettings from command_router.http_api.* config paths. */ - static HttpAPISettings settings_from_config(const JsonConfig& command_router_config); - /** @brief DedicatedThread entry point; blocks in server.listen() until stop(). */ bool thread_one_step() override; @@ -87,6 +83,7 @@ class CommandRouterHttpAPI : public processor::Processor, public processor::Thre int port; httplib::Server server; shared_ptr thread_pool; + shared_ptr issuer_bus; HttpAPISettings settings; atomic shutting_down{false}; unordered_map> executions; diff --git a/src/agents/command_router/CommandRouterHttpAPIConfig.cc b/src/agents/command_router/CommandRouterHttpAPIConfig.cc new file mode 100644 index 00000000..02fa008c --- /dev/null +++ b/src/agents/command_router/CommandRouterHttpAPIConfig.cc @@ -0,0 +1,68 @@ +#include "CommandRouterHttpAPIConfig.h" + +#include "Utils.h" + +using namespace command_router; +using namespace commons; + +static pair parse_host_port(const string& endpoint, const string& config_path) { + auto tokens = Utils::split(endpoint, ':'); + if (tokens.size() != 2) { + RAISE_ERROR("Invalid " + config_path + " configuration: endpoint must be :"); + } + + int port; + try { + port = stoi(tokens[1]); + } catch (const exception&) { + RAISE_ERROR("Invalid " + config_path + " configuration: port must be an integer"); + } + + return {tokens[0], port}; +} + +static string resolve_issuer_bus_endpoint(const JsonConfig& command_router_config) { + auto configured = command_router_config.at_path("http_api.bus_client_endpoint").get_or(""); + if (!configured.empty()) { + return configured; + } + + const string router_bus_endpoint = command_router_config.at_path("endpoint").get(); + auto [host, router_port] = parse_host_port(router_bus_endpoint, "command_router.endpoint"); + return host + ":" + to_string(router_port + 10); +} + +static HttpAPISettings load_http_api_settings(const JsonConfig& command_router_config) { + HttpAPISettings settings; + settings.max_concurrent_executions = + command_router_config.at_path("http_api.max_concurrent_executions") + .get_or(settings.max_concurrent_executions); + settings.max_queued_executions = command_router_config.at_path("http_api.max_queued_executions") + .get_or(settings.max_queued_executions); + settings.max_events_per_execution = + command_router_config.at_path("http_api.max_events_per_execution") + .get_or(settings.max_events_per_execution); + settings.execution_retention_ms = command_router_config.at_path("http_api.execution_retention_ms") + .get_or(settings.execution_retention_ms); + settings.stream_items_per_chunk = + command_router_config.at_path("http_api.stream_items_per_chunk") + .get_or(settings.stream_items_per_chunk); + if (settings.stream_items_per_chunk == 0) { + RAISE_ERROR("command_router.http_api.stream_items_per_chunk must be at least 1"); + } + return settings; +} + +CommandRouterHttpAPIConfig CommandRouterHttpAPIConfig::from_config( + const JsonConfig& command_router_config) { + CommandRouterHttpAPIConfig config; + tie(config.host, config.port) = + parse_host_port(command_router_config.at_path("http_api.endpoint").get(), + "command_router.http_api.endpoint"); + config.settings = load_http_api_settings(command_router_config); + config.thread_pool_size = + command_router_config.at_path("http_api.thread_pool_size").get_or(4); + config.router_bus_endpoint = command_router_config.at_path("endpoint").get(); + config.issuer_bus_endpoint = resolve_issuer_bus_endpoint(command_router_config); + return config; +} diff --git a/src/agents/command_router/CommandRouterHttpAPIConfig.h b/src/agents/command_router/CommandRouterHttpAPIConfig.h new file mode 100644 index 00000000..100614e4 --- /dev/null +++ b/src/agents/command_router/CommandRouterHttpAPIConfig.h @@ -0,0 +1,33 @@ +#pragma once + +#include + +#include "CommandExecution.h" +#include "JsonConfig.h" + +using namespace std; +using namespace commons; + +namespace command_router { + +struct HttpAPISettings { + size_t max_concurrent_executions = 100; + size_t max_queued_executions = 500; + size_t max_events_per_execution = CommandExecution::DEFAULT_MAX_EVENTS; + long long execution_retention_ms = 15 * 60 * 1000; + /** Max query/evolution answers per WebSocket chunk event (1 = one item per event). */ + size_t stream_items_per_chunk = 1; +}; + +struct CommandRouterHttpAPIConfig { + string host; + int port = 0; + HttpAPISettings settings; + unsigned int thread_pool_size = 4; + string router_bus_endpoint; + string issuer_bus_endpoint; + + static CommandRouterHttpAPIConfig from_config(const JsonConfig& command_router_config); +}; + +} // namespace command_router diff --git a/src/agents/command_router/CommandRouterHttpAPISingleton.cc b/src/agents/command_router/CommandRouterHttpAPISingleton.cc index 790b4164..ed237a02 100644 --- a/src/agents/command_router/CommandRouterHttpAPISingleton.cc +++ b/src/agents/command_router/CommandRouterHttpAPISingleton.cc @@ -1,38 +1,17 @@ #include "CommandRouterHttpAPISingleton.h" -#include "Utils.h" +#include "CommandRouterHttpAPIConfig.h" +#include "ServiceBus.h" using namespace command_router; using namespace commons; +using namespace service_bus; bool CommandRouterHttpAPISingleton::INITIALIZED = false; shared_ptr CommandRouterHttpAPISingleton::HTTP_API = shared_ptr(nullptr); mutex CommandRouterHttpAPISingleton::API_MUTEX; -namespace { - -pair parse_http_api_endpoint(const JsonConfig& command_router_config) { - auto endpoint = command_router_config.at_path("http_api.endpoint").get(); - auto tokens = Utils::split(endpoint, ':'); - if (tokens.size() != 2) { - RAISE_ERROR( - "Invalid command_router.http_api configuration: endpoint must be in the format " - ":"); - } - - int port; - try { - port = stoi(tokens[1]); - } catch (const exception&) { - RAISE_ERROR("Invalid command_router.http_api configuration: port must be an integer"); - } - - return {tokens[0], port}; -} - -} // namespace - void CommandRouterHttpAPISingleton::init(const JsonConfig& command_router_config) { lock_guard semaphore(API_MUTEX); if (INITIALIZED) { @@ -40,18 +19,17 @@ void CommandRouterHttpAPISingleton::init(const JsonConfig& command_router_config "CommandRouterHttpAPISingleton already initialized. " "CommandRouterHttpAPISingleton::init() should be called only once."); } - CommandRouterHttpAPISingleton::create_and_start(command_router_config); + create_and_start(command_router_config); } void CommandRouterHttpAPISingleton::create_and_start(const JsonConfig& command_router_config) { - auto [host, port] = parse_http_api_endpoint(command_router_config); - auto settings = CommandRouterHttpAPI::settings_from_config(command_router_config); - unsigned int num_threads = - command_router_config.at_path("http_api.thread_pool_size").get_or(4); + const auto config = CommandRouterHttpAPIConfig::from_config(command_router_config); + auto issuer_bus = make_shared(config.issuer_bus_endpoint, config.router_bus_endpoint); auto thread_pool_executor = - make_shared("http_api_thread_pool_executor", num_threads); - HTTP_API = make_shared(host, port, thread_pool_executor, settings); + make_shared("http_api_thread_pool_executor", config.thread_pool_size); + HTTP_API = make_shared( + config.host, config.port, thread_pool_executor, config.settings, issuer_bus); auto http_api_thread = make_shared("http_api_thread", HTTP_API.get()); diff --git a/src/main/BUILD b/src/main/BUILD index 43fd2c11..3a1c9a22 100644 --- a/src/main/BUILD +++ b/src/main/BUILD @@ -32,6 +32,7 @@ cc_library( srcs = ["bus_node.cc"], deps = [ "//agents/atomdb_broker:atomdb_broker_lib", + "//agents/command_router:command_router_http_api_singleton", "//agents/command_router:command_router_lib", "//commons:commons_lib", "//main/helpers:main_helper_processor_factory_lib", diff --git a/src/tests/cpp/BUILD b/src/tests/cpp/BUILD index 6908daba..198931ec 100644 --- a/src/tests/cpp/BUILD +++ b/src/tests/cpp/BUILD @@ -1030,7 +1030,7 @@ cc_test( cc_test( name = "command_router_http_api_test", - size = "small", + size = "medium", srcs = [ "command_router_http_api_test.cc", ], @@ -1038,12 +1038,29 @@ cc_test( "-Iexternal/gtest/googletest/include", "-Iexternal/gtest/googletest", ], + linkopts = [ + "-L/usr/local/lib", + "-lpqxx", + "-lpq", + "-lhiredis_cluster", + "-lhiredis", + "-lmongocxx", + "-lbsoncxx", + ], linkstatic = 1, deps = [ + "//agents/command_router:bus_command_router_processor", + "//agents/command_router:bus_command_router_proxy_client", "//agents/command_router:command_execution", "//agents/command_router:command_router_http_api", + "//agents/command_router:command_router_http_api_config", "//agents/command_router:command_router_http_api_singleton", + "//agents/query_engine:query_answer", + "//atomdb:atomdb_singleton", "//commons/processor:processor_lib", + "//service_bus:service_bus_lib", + "//tests/cpp/test_commons:test_atomdb_json_config", + "//tests/cpp/test_commons:test_system_params", "@com_github_google_googletest//:gtest", "@mbedtls", ], diff --git a/src/tests/cpp/command_router_http_api_test.cc b/src/tests/cpp/command_router_http_api_test.cc index 81274c82..947ae000 100644 --- a/src/tests/cpp/command_router_http_api_test.cc +++ b/src/tests/cpp/command_router_http_api_test.cc @@ -2,11 +2,23 @@ #include #include +#include "AtomDBSingleton.h" +#include "BaseProxy.h" +#include "BaseQueryProxy.h" +#include "BusCommandRouterProcessor.h" +#include "BusCommandRouterProxy.h" +#include "BusCommandRouterProxyClient.h" #include "CommandExecution.h" #include "CommandRouterHttpAPI.h" +#include "CommandRouterHttpAPIConfig.h" #include "CommandRouterHttpAPISingleton.h" #include "DedicatedThread.h" #include "JsonConfig.h" +#include "PortPool.h" +#include "QueryAnswer.h" +#include "ServiceBus.h" +#include "TestAtomDBJsonConfig.h" +#include "TestSystemParams.h" #include "gtest/gtest.h" #include "httplib.h" #include "processor/ThreadPool.h" @@ -14,6 +26,11 @@ using namespace command_router; using namespace processor; using namespace commons; +using namespace agents; +using namespace atomdb; +using namespace query_engine; +using namespace service_bus; +using das_test::init_test_system_parameters_singleton; using json = nlohmann::json; namespace { @@ -29,11 +46,32 @@ json make_execution_body(const string& command_type = "query", return {{"command_type", command_type}, {"command_text", command_text}}; } +/** + * HTTP API fixture with an in-process command router bus so executions reach "running". + */ class HttpAPIServerFixture { public: void start(int port, const HttpAPISettings& settings = {}, unsigned int num_threads = 8) { + if (!service_bus_statics_initialized) { + ServiceBus::initialize_statics({ServiceBus::BUS_COMMAND_ROUTER}, 49500, 49999); + service_bus_statics_initialized = true; + } + + const unsigned int router_port = PortPool::get_port(); + const unsigned int issuer_port = PortPool::get_port(); + const string router_id = TEST_HOST + ":" + std::to_string(router_port); + const string issuer_id = TEST_HOST + ":" + std::to_string(issuer_port); + + this->router_bus = make_shared(router_id); + this->router_bus->register_processor(make_shared()); + Utils::sleep(500); + + this->issuer_bus = make_shared(issuer_id, router_id); + Utils::sleep(500); + this->thread_pool = make_shared("test_thread_pool", num_threads); - this->api = make_shared(TEST_HOST, port, this->thread_pool, settings); + this->api = make_shared( + TEST_HOST, port, this->thread_pool, settings, this->issuer_bus); this->api_thread = make_shared("test_api_thread", this->api.get()); CommandRouterHttpAPI::initialize(this->api, {this->api_thread, this->thread_pool}); Utils::sleep(300); @@ -42,7 +80,10 @@ class HttpAPIServerFixture { void stop() { if (this->api != nullptr) { this->api->stop(); + this->api = nullptr; } + this->issuer_bus = nullptr; + this->router_bus = nullptr; } httplib::Client make_client(int port) const { @@ -53,11 +94,71 @@ class HttpAPIServerFixture { } private: + inline static bool service_bus_statics_initialized = false; + shared_ptr router_bus; + shared_ptr issuer_bus; shared_ptr thread_pool; shared_ptr api; shared_ptr api_thread; }; +class CommandRouterStreamTestEnvironment : public ::testing::Environment { + public: + void SetUp() override { + AtomDBSingleton::init(test_atomdb_json_config()); + init_test_system_parameters_singleton(); + } +}; + +class StreamTestProxy : public BusCommandRouterProxy { + public: + StreamTestProxy() : BusCommandRouterProxy("query", "(Similarity %V1 %V2)") {} + + void enqueue_answers(unsigned int count) { + vector bundle; + for (unsigned int i = 0; i < count; ++i) { + QueryAnswer answer("answer-" + std::to_string(i), 0.0); + bundle.push_back(answer.tokenize()); + } + this->from_remote_peer(BaseQueryProxy::ANSWER_BUNDLE, bundle); + } + + void mark_routed() { this->from_remote_peer(BusCommandRouterProxy::ROUTED, {}); } + + void mark_finished() { this->from_remote_peer(BaseProxy::FINISHED, {}); } +}; + +size_t total_items(const vector>& chunks) { + size_t count = 0; + for (const auto& chunk : chunks) { + count += chunk.size(); + } + return count; +} + +vector chunk_sizes(const vector>& chunks) { + vector sizes; + for (const auto& chunk : chunks) { + sizes.push_back(chunk.size()); + } + return sizes; +} + +JsonConfig make_command_router_config(const json& overrides = json::object()) { + json root = {{"endpoint", "localhost:40008"}, + {"ports_range", "48000:48999"}, + {"http_api", + {{"endpoint", "localhost:40009"}, + {"thread_pool_size", 8}, + {"max_concurrent_executions", 50}, + {"max_queued_executions", 200}, + {"max_events_per_execution", 5000}, + {"stream_items_per_chunk", 25}, + {"execution_retention_ms", 123456}}}}; + root.merge_patch(overrides); + return JsonConfig(root); +} + } // namespace class CommandRouterHttpAPITest : public ::testing::Test { @@ -137,6 +238,112 @@ TEST(CommandExecutionTest, event_buffer_overflow_raises) { EXPECT_THROW(exec.publish_chunk(2, {"b", "c"}), runtime_error); } +// ----------------------------------------------------------------------------- +// CommandRouterHttpAPIConfig + +TEST(CommandRouterHttpAPIConfigTest, from_config_loads_http_api_fields) { + const auto config = CommandRouterHttpAPIConfig::from_config(make_command_router_config()); + + EXPECT_EQ(config.host, "localhost"); + EXPECT_EQ(config.port, 40009); + EXPECT_EQ(config.thread_pool_size, 8u); + EXPECT_EQ(config.router_bus_endpoint, "localhost:40008"); + EXPECT_EQ(config.issuer_bus_endpoint, "localhost:40018"); + EXPECT_EQ(config.settings.max_concurrent_executions, 50u); + EXPECT_EQ(config.settings.max_queued_executions, 200u); + EXPECT_EQ(config.settings.max_events_per_execution, 5000u); + EXPECT_EQ(config.settings.execution_retention_ms, 123456); + EXPECT_EQ(config.settings.stream_items_per_chunk, 25u); +} + +TEST(CommandRouterHttpAPIConfigTest, from_config_uses_explicit_bus_client_endpoint) { + const auto config = CommandRouterHttpAPIConfig::from_config(make_command_router_config( + {{"http_api", {{"bus_client_endpoint", "localhost:50000"}}}})); + + EXPECT_EQ(config.issuer_bus_endpoint, "localhost:50000"); +} + +TEST(CommandRouterHttpAPIConfigTest, from_config_rejects_zero_stream_items_per_chunk) { + EXPECT_THROW(CommandRouterHttpAPIConfig::from_config( + make_command_router_config({{"http_api", {{"stream_items_per_chunk", 0}}}})), + runtime_error); +} + +TEST(CommandRouterHttpAPIConfigTest, from_config_rejects_invalid_http_api_endpoint) { + EXPECT_THROW(CommandRouterHttpAPIConfig::from_config(make_command_router_config( + {{"http_api", {{"endpoint", "not-a-valid-endpoint"}}}})), + runtime_error); +} + +// ----------------------------------------------------------------------------- +// BusCommandRouterProxyClient (HTTP stream polling) + +TEST(BusCommandRouterProxyClientTest, stream_emits_one_item_per_chunk_by_default) { + auto proxy = make_shared(); + proxy->mark_routed(); + proxy->enqueue_answers(5); + proxy->mark_finished(); + + vector> chunks; + auto on_chunk = [&](const vector& chunk) { chunks.push_back(chunk); }; + + ASSERT_TRUE(poll_router_proxy_stream(proxy, "query", 1, nullptr, on_chunk, nullptr, nullptr)); + EXPECT_EQ(chunks.size(), 5u); + EXPECT_EQ(chunk_sizes(chunks), (vector{1, 1, 1, 1, 1})); + EXPECT_EQ(total_items(chunks), 5u); + EXPECT_NE(chunks.front().front().find("answer-0"), string::npos); +} + +TEST(BusCommandRouterProxyClientTest, stream_groups_items_per_chunk) { + auto proxy = make_shared(); + proxy->mark_routed(); + proxy->enqueue_answers(7); + proxy->mark_finished(); + + vector> chunks; + auto on_chunk = [&](const vector& chunk) { chunks.push_back(chunk); }; + + ASSERT_TRUE(poll_router_proxy_stream(proxy, "query", 3, nullptr, on_chunk, nullptr, nullptr)); + EXPECT_EQ(chunks.size(), 3u); + EXPECT_EQ(chunk_sizes(chunks), (vector{3, 3, 1})); + EXPECT_EQ(total_items(chunks), 7u); +} + +TEST(BusCommandRouterProxyClientTest, rejects_zero_items_per_chunk) { + auto proxy = make_shared(); + proxy->mark_routed(); + proxy->enqueue_answers(1); + proxy->mark_finished(); + + string error_message; + auto on_error = [&](const string& message) { error_message = message; }; + + EXPECT_FALSE(poll_router_proxy_stream(proxy, "query", 0, nullptr, nullptr, on_error, nullptr)); + EXPECT_NE(error_message.find("items_per_chunk"), string::npos); +} + +TEST(BusCommandRouterProxyClientTest, get_and_set_emit_single_chunk) { + auto get_proxy = make_shared(); + get_proxy->from_remote_peer(BusCommandRouterProxy::PARAMS_RESPONSE, {"params-body"}); + + vector> get_chunks; + ASSERT_TRUE(poll_router_proxy_stream( + get_proxy, "get", 1, nullptr, [&](const vector& chunk) { get_chunks.push_back(chunk); }, + nullptr, nullptr)); + ASSERT_EQ(get_chunks.size(), 1u); + EXPECT_EQ(get_chunks[0], vector{"params-body"}); + + auto set_proxy = make_shared(); + set_proxy->from_remote_peer(BusCommandRouterProxy::SET_PARAM_ACK, {"ack-body"}); + + vector> set_chunks; + ASSERT_TRUE(poll_router_proxy_stream( + set_proxy, "set", 1, nullptr, [&](const vector& chunk) { set_chunks.push_back(chunk); }, + nullptr, nullptr)); + ASSERT_EQ(set_chunks.size(), 1u); + EXPECT_EQ(set_chunks[0], vector{"ack-body"}); +} + // ----------------------------------------------------------------------------- // HTTP API @@ -369,5 +576,6 @@ TEST_F(CommandRouterHttpAPISingletonTest, init_after_provide_throws) { int main(int argc, char** argv) { ::testing::InitGoogleTest(&argc, argv); + ::testing::AddGlobalTestEnvironment(new CommandRouterStreamTestEnvironment()); return RUN_ALL_TESTS(); } From b716e962812d0a15952027a7a0fddbbb2c455660 Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Fri, 26 Jun 2026 12:29:59 -0300 Subject: [PATCH 05/16] Move command router HTTP API into http_api subdirectory. Extract HTTP-related sources and Bazel targets into agents/command_router/http_api and update main and test BUILD deps to the new package path. --- src/agents/command_router/BUILD | 63 ------------------ src/agents/command_router/http_api/BUILD | 66 +++++++++++++++++++ .../BusCommandRouterProxyClient.cc | 0 .../BusCommandRouterProxyClient.h | 0 .../{ => http_api}/CommandExecution.cc | 0 .../{ => http_api}/CommandExecution.h | 0 .../{ => http_api}/CommandRouterHttpAPI.cc | 0 .../{ => http_api}/CommandRouterHttpAPI.h | 0 .../CommandRouterHttpAPIConfig.cc | 0 .../CommandRouterHttpAPIConfig.h | 0 .../CommandRouterHttpAPISingleton.cc | 0 .../CommandRouterHttpAPISingleton.h | 0 src/main/BUILD | 2 +- src/tests/cpp/BUILD | 10 +-- 14 files changed, 72 insertions(+), 69 deletions(-) create mode 100644 src/agents/command_router/http_api/BUILD rename src/agents/command_router/{ => http_api}/BusCommandRouterProxyClient.cc (100%) rename src/agents/command_router/{ => http_api}/BusCommandRouterProxyClient.h (100%) rename src/agents/command_router/{ => http_api}/CommandExecution.cc (100%) rename src/agents/command_router/{ => http_api}/CommandExecution.h (100%) rename src/agents/command_router/{ => http_api}/CommandRouterHttpAPI.cc (100%) rename src/agents/command_router/{ => http_api}/CommandRouterHttpAPI.h (100%) rename src/agents/command_router/{ => http_api}/CommandRouterHttpAPIConfig.cc (100%) rename src/agents/command_router/{ => http_api}/CommandRouterHttpAPIConfig.h (100%) rename src/agents/command_router/{ => http_api}/CommandRouterHttpAPISingleton.cc (100%) rename src/agents/command_router/{ => http_api}/CommandRouterHttpAPISingleton.h (100%) diff --git a/src/agents/command_router/BUILD b/src/agents/command_router/BUILD index 65399111..87858001 100644 --- a/src/agents/command_router/BUILD +++ b/src/agents/command_router/BUILD @@ -27,29 +27,6 @@ cc_library( ], ) -cc_library( - name = "bus_command_router_proxy_client", - srcs = ["BusCommandRouterProxyClient.cc"], - hdrs = ["BusCommandRouterProxyClient.h"], - includes = ["."], - deps = [ - ":bus_command_router_proxy", - "//agents:base_query_proxy", - "//commons:commons_lib", - ], -) - -cc_library( - name = "command_router_http_api_config", - srcs = ["CommandRouterHttpAPIConfig.cc"], - hdrs = ["CommandRouterHttpAPIConfig.h"], - includes = ["."], - deps = [ - ":command_execution", - "//commons:commons_lib", - ], -) - cc_library( name = "evolution_metta_parser", srcs = ["EvolutionMettaParser.cc"], @@ -79,43 +56,3 @@ cc_library( "//service_bus:service_bus_singleton", ], ) - -cc_library( - name = "command_execution", - srcs = ["CommandExecution.cc"], - hdrs = ["CommandExecution.h"], - includes = ["."], - deps = [ - "//commons:commons_lib", - ], -) - -cc_library( - name = "command_router_http_api", - srcs = ["CommandRouterHttpAPI.cc"], - hdrs = ["CommandRouterHttpAPI.h"], - includes = ["."], - deps = [ - ":bus_command_router_proxy", - ":bus_command_router_proxy_client", - ":command_execution", - ":command_router_http_api_config", - "//commons:commons_lib", - "//commons/processor:processor_lib", - "//hasher:hasher_lib", - "//service_bus:service_bus_lib", - ], -) - -cc_library( - name = "command_router_http_api_singleton", - srcs = ["CommandRouterHttpAPISingleton.cc"], - hdrs = ["CommandRouterHttpAPISingleton.h"], - includes = ["."], - deps = [ - ":command_router_http_api", - ":command_router_http_api_config", - "//commons/processor:processor_lib", - "//service_bus:service_bus_lib", - ], -) diff --git a/src/agents/command_router/http_api/BUILD b/src/agents/command_router/http_api/BUILD new file mode 100644 index 00000000..b5b56446 --- /dev/null +++ b/src/agents/command_router/http_api/BUILD @@ -0,0 +1,66 @@ +load("@rules_cc//cc:cc_library.bzl", "cc_library") + +package(default_visibility = ["//visibility:public"]) + +cc_library( + name = "command_execution", + srcs = ["CommandExecution.cc"], + hdrs = ["CommandExecution.h"], + includes = ["."], + deps = [ + "//commons:commons_lib", + ], +) + +cc_library( + name = "bus_command_router_proxy_client", + srcs = ["BusCommandRouterProxyClient.cc"], + hdrs = ["BusCommandRouterProxyClient.h"], + includes = ["."], + deps = [ + "//agents/command_router:bus_command_router_proxy", + "//agents:base_query_proxy", + "//commons:commons_lib", + ], +) + +cc_library( + name = "command_router_http_api_config", + srcs = ["CommandRouterHttpAPIConfig.cc"], + hdrs = ["CommandRouterHttpAPIConfig.h"], + includes = ["."], + deps = [ + ":command_execution", + "//commons:commons_lib", + ], +) + +cc_library( + name = "command_router_http_api", + srcs = ["CommandRouterHttpAPI.cc"], + hdrs = ["CommandRouterHttpAPI.h"], + includes = ["."], + deps = [ + ":bus_command_router_proxy_client", + ":command_execution", + ":command_router_http_api_config", + "//agents/command_router:bus_command_router_proxy", + "//commons:commons_lib", + "//commons/processor:processor_lib", + "//hasher:hasher_lib", + "//service_bus:service_bus_lib", + ], +) + +cc_library( + name = "command_router_http_api_singleton", + srcs = ["CommandRouterHttpAPISingleton.cc"], + hdrs = ["CommandRouterHttpAPISingleton.h"], + includes = ["."], + deps = [ + ":command_router_http_api", + ":command_router_http_api_config", + "//commons/processor:processor_lib", + "//service_bus:service_bus_lib", + ], +) diff --git a/src/agents/command_router/BusCommandRouterProxyClient.cc b/src/agents/command_router/http_api/BusCommandRouterProxyClient.cc similarity index 100% rename from src/agents/command_router/BusCommandRouterProxyClient.cc rename to src/agents/command_router/http_api/BusCommandRouterProxyClient.cc diff --git a/src/agents/command_router/BusCommandRouterProxyClient.h b/src/agents/command_router/http_api/BusCommandRouterProxyClient.h similarity index 100% rename from src/agents/command_router/BusCommandRouterProxyClient.h rename to src/agents/command_router/http_api/BusCommandRouterProxyClient.h diff --git a/src/agents/command_router/CommandExecution.cc b/src/agents/command_router/http_api/CommandExecution.cc similarity index 100% rename from src/agents/command_router/CommandExecution.cc rename to src/agents/command_router/http_api/CommandExecution.cc diff --git a/src/agents/command_router/CommandExecution.h b/src/agents/command_router/http_api/CommandExecution.h similarity index 100% rename from src/agents/command_router/CommandExecution.h rename to src/agents/command_router/http_api/CommandExecution.h diff --git a/src/agents/command_router/CommandRouterHttpAPI.cc b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc similarity index 100% rename from src/agents/command_router/CommandRouterHttpAPI.cc rename to src/agents/command_router/http_api/CommandRouterHttpAPI.cc diff --git a/src/agents/command_router/CommandRouterHttpAPI.h b/src/agents/command_router/http_api/CommandRouterHttpAPI.h similarity index 100% rename from src/agents/command_router/CommandRouterHttpAPI.h rename to src/agents/command_router/http_api/CommandRouterHttpAPI.h diff --git a/src/agents/command_router/CommandRouterHttpAPIConfig.cc b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc similarity index 100% rename from src/agents/command_router/CommandRouterHttpAPIConfig.cc rename to src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc diff --git a/src/agents/command_router/CommandRouterHttpAPIConfig.h b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h similarity index 100% rename from src/agents/command_router/CommandRouterHttpAPIConfig.h rename to src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h diff --git a/src/agents/command_router/CommandRouterHttpAPISingleton.cc b/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.cc similarity index 100% rename from src/agents/command_router/CommandRouterHttpAPISingleton.cc rename to src/agents/command_router/http_api/CommandRouterHttpAPISingleton.cc diff --git a/src/agents/command_router/CommandRouterHttpAPISingleton.h b/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.h similarity index 100% rename from src/agents/command_router/CommandRouterHttpAPISingleton.h rename to src/agents/command_router/http_api/CommandRouterHttpAPISingleton.h diff --git a/src/main/BUILD b/src/main/BUILD index 3a1c9a22..70f23e2c 100644 --- a/src/main/BUILD +++ b/src/main/BUILD @@ -32,7 +32,7 @@ cc_library( srcs = ["bus_node.cc"], deps = [ "//agents/atomdb_broker:atomdb_broker_lib", - "//agents/command_router:command_router_http_api_singleton", + "//agents/command_router/http_api:command_router_http_api_singleton", "//agents/command_router:command_router_lib", "//commons:commons_lib", "//main/helpers:main_helper_processor_factory_lib", diff --git a/src/tests/cpp/BUILD b/src/tests/cpp/BUILD index 198931ec..35122ed0 100644 --- a/src/tests/cpp/BUILD +++ b/src/tests/cpp/BUILD @@ -1050,11 +1050,11 @@ cc_test( linkstatic = 1, deps = [ "//agents/command_router:bus_command_router_processor", - "//agents/command_router:bus_command_router_proxy_client", - "//agents/command_router:command_execution", - "//agents/command_router:command_router_http_api", - "//agents/command_router:command_router_http_api_config", - "//agents/command_router:command_router_http_api_singleton", + "//agents/command_router/http_api:bus_command_router_proxy_client", + "//agents/command_router/http_api:command_execution", + "//agents/command_router/http_api:command_router_http_api", + "//agents/command_router/http_api:command_router_http_api_config", + "//agents/command_router/http_api:command_router_http_api_singleton", "//agents/query_engine:query_answer", "//atomdb:atomdb_singleton", "//commons/processor:processor_lib", From 768e00d17d0701b4bab3f8d7ec1a4077241718a9 Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Fri, 26 Jun 2026 13:31:52 -0300 Subject: [PATCH 06/16] Change BusCommandRouterProxyClient to BusCommandRouterStreamPoller --- src/agents/command_router/http_api/BUILD | 10 ++-- .../http_api/BusCommandRouterProxyClient.h | 32 ----------- ...c => BusCommandRouterProxyStreamPoller.cc} | 20 +++---- .../BusCommandRouterProxyStreamPoller.h | 53 +++++++++++++++++++ .../http_api/CommandRouterHttpAPI.cc | 16 +++--- .../http_api/CommandRouterHttpAPIConfig.cc | 5 +- src/main/BUILD | 2 +- src/tests/cpp/BUILD | 2 +- src/tests/cpp/command_router_http_api_test.cc | 51 +++++++++++------- 9 files changed, 113 insertions(+), 78 deletions(-) delete mode 100644 src/agents/command_router/http_api/BusCommandRouterProxyClient.h rename src/agents/command_router/http_api/{BusCommandRouterProxyClient.cc => BusCommandRouterProxyStreamPoller.cc} (84%) create mode 100644 src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.h diff --git a/src/agents/command_router/http_api/BUILD b/src/agents/command_router/http_api/BUILD index b5b56446..b8a4d560 100644 --- a/src/agents/command_router/http_api/BUILD +++ b/src/agents/command_router/http_api/BUILD @@ -13,13 +13,13 @@ cc_library( ) cc_library( - name = "bus_command_router_proxy_client", - srcs = ["BusCommandRouterProxyClient.cc"], - hdrs = ["BusCommandRouterProxyClient.h"], + name = "bus_command_router_proxy_stream_poller", + srcs = ["BusCommandRouterProxyStreamPoller.cc"], + hdrs = ["BusCommandRouterProxyStreamPoller.h"], includes = ["."], deps = [ - "//agents/command_router:bus_command_router_proxy", "//agents:base_query_proxy", + "//agents/command_router:bus_command_router_proxy", "//commons:commons_lib", ], ) @@ -41,7 +41,7 @@ cc_library( hdrs = ["CommandRouterHttpAPI.h"], includes = ["."], deps = [ - ":bus_command_router_proxy_client", + ":bus_command_router_proxy_stream_poller", ":command_execution", ":command_router_http_api_config", "//agents/command_router:bus_command_router_proxy", diff --git a/src/agents/command_router/http_api/BusCommandRouterProxyClient.h b/src/agents/command_router/http_api/BusCommandRouterProxyClient.h deleted file mode 100644 index 6426e99f..00000000 --- a/src/agents/command_router/http_api/BusCommandRouterProxyClient.h +++ /dev/null @@ -1,32 +0,0 @@ -#pragma once - -#include -#include -#include -#include - -#include "BusCommandRouterProxy.h" - -using namespace std; - -namespace command_router { - -/** - * Poll a BusCommandRouterProxy for HTTP streaming consumers. - * - * Emits at most @p items_per_chunk answers per on_chunk callback so clients can - * iterate results incrementally (1-by-1 or N-by-N). Terminal bus_client does - * not use this; it keeps its own polling loop and query-engine limits. - * - * @param items_per_chunk Minimum 1. Each callback receives at most this many items. - * @return true on success, false on error or abort. - */ -bool poll_router_proxy_stream(const shared_ptr& router_proxy, - const string& command_type, - size_t items_per_chunk, - const function& should_abort, - const function& chunk)>& on_chunk, - const function& on_error, - const function& on_aborted); - -} // namespace command_router diff --git a/src/agents/command_router/http_api/BusCommandRouterProxyClient.cc b/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc similarity index 84% rename from src/agents/command_router/http_api/BusCommandRouterProxyClient.cc rename to src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc index 1697fd18..8d6c1bec 100644 --- a/src/agents/command_router/http_api/BusCommandRouterProxyClient.cc +++ b/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc @@ -1,4 +1,4 @@ -#include "BusCommandRouterProxyClient.h" +#include "BusCommandRouterProxyStreamPoller.h" #include "BaseQueryProxy.h" #include "Utils.h" @@ -9,7 +9,8 @@ using namespace agents; namespace { -void emit_chunk(const function& chunk)>& on_chunk, vector& chunk_data) { +void emit_chunk(const function& chunk)>& on_chunk, + vector& chunk_data) { if (!chunk_data.empty() && on_chunk) { on_chunk(chunk_data); chunk_data.clear(); @@ -33,13 +34,14 @@ void append_answer_chunk(const shared_ptr& router_proxy, } // namespace -bool command_router::poll_router_proxy_stream(const shared_ptr& router_proxy, - const string& command_type, - size_t items_per_chunk, - const function& should_abort, - const function& chunk)>& on_chunk, - const function& on_error, - const function& on_aborted) { +bool BusCommandRouterProxyStreamPoller::poll_stream( + const shared_ptr& router_proxy, + const string& command_type, + size_t items_per_chunk, + const function& should_abort, + const function& chunk)>& on_chunk, + const function& on_error, + const function& on_aborted) { if (items_per_chunk == 0) { if (on_error) { on_error("items_per_chunk must be at least 1"); diff --git a/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.h b/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.h new file mode 100644 index 00000000..047a9862 --- /dev/null +++ b/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.h @@ -0,0 +1,53 @@ +#pragma once + +#include +#include +#include +#include + +#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 @p router_proxy until the command finishes, is aborted, or fails. + * + * For @p command_type "query" and "evolution", answers are popped from the proxy + * and forwarded in batches of at most @p 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 @p 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 @p 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 @p should_abort returned true. + * @return true when the command completed without error; false otherwise. + */ + static bool poll_stream(const shared_ptr& router_proxy, + const string& command_type, + size_t items_per_chunk, + const function& should_abort, + const function& chunk)>& on_chunk, + const function& on_error, + const function& on_aborted); + + private: + BusCommandRouterProxyStreamPoller() = delete; +}; + +} // namespace command_router diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPI.cc b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc index 78f79fb1..69c8ca04 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPI.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc @@ -5,7 +5,7 @@ #include #include "BusCommandRouterProxy.h" -#include "BusCommandRouterProxyClient.h" +#include "BusCommandRouterProxyStreamPoller.h" #define LOG_LEVEL INFO_LEVEL #include "Logger.h" @@ -374,13 +374,13 @@ void CommandRouterHttpAPI::run_execution_inner(const shared_ptr(exec->command_type, router_arg); this->issuer_bus->issue_bus_command(router_proxy); - if (!poll_router_proxy_stream(router_proxy, - exec->command_type, - this->settings.stream_items_per_chunk, - should_abort, - on_chunk, - on_error, - on_aborted)) { + if (!BusCommandRouterProxyStreamPoller::poll_stream(router_proxy, + exec->command_type, + this->settings.stream_items_per_chunk, + should_abort, + on_chunk, + on_error, + on_aborted)) { return; } diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc index 02fa008c..41819ccb 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc @@ -44,9 +44,8 @@ static HttpAPISettings load_http_api_settings(const JsonConfig& command_router_c .get_or(settings.max_events_per_execution); settings.execution_retention_ms = command_router_config.at_path("http_api.execution_retention_ms") .get_or(settings.execution_retention_ms); - settings.stream_items_per_chunk = - command_router_config.at_path("http_api.stream_items_per_chunk") - .get_or(settings.stream_items_per_chunk); + settings.stream_items_per_chunk = command_router_config.at_path("http_api.stream_items_per_chunk") + .get_or(settings.stream_items_per_chunk); if (settings.stream_items_per_chunk == 0) { RAISE_ERROR("command_router.http_api.stream_items_per_chunk must be at least 1"); } diff --git a/src/main/BUILD b/src/main/BUILD index 70f23e2c..02894867 100644 --- a/src/main/BUILD +++ b/src/main/BUILD @@ -32,8 +32,8 @@ cc_library( srcs = ["bus_node.cc"], deps = [ "//agents/atomdb_broker:atomdb_broker_lib", - "//agents/command_router/http_api:command_router_http_api_singleton", "//agents/command_router:command_router_lib", + "//agents/command_router/http_api:command_router_http_api_singleton", "//commons:commons_lib", "//main/helpers:main_helper_processor_factory_lib", ], diff --git a/src/tests/cpp/BUILD b/src/tests/cpp/BUILD index 35122ed0..cd3f88b2 100644 --- a/src/tests/cpp/BUILD +++ b/src/tests/cpp/BUILD @@ -1050,7 +1050,7 @@ cc_test( linkstatic = 1, deps = [ "//agents/command_router:bus_command_router_processor", - "//agents/command_router/http_api:bus_command_router_proxy_client", + "//agents/command_router/http_api:bus_command_router_proxy_stream_poller", "//agents/command_router/http_api:command_execution", "//agents/command_router/http_api:command_router_http_api", "//agents/command_router/http_api:command_router_http_api_config", diff --git a/src/tests/cpp/command_router_http_api_test.cc b/src/tests/cpp/command_router_http_api_test.cc index 947ae000..bc1260a0 100644 --- a/src/tests/cpp/command_router_http_api_test.cc +++ b/src/tests/cpp/command_router_http_api_test.cc @@ -7,7 +7,7 @@ #include "BaseQueryProxy.h" #include "BusCommandRouterProcessor.h" #include "BusCommandRouterProxy.h" -#include "BusCommandRouterProxyClient.h" +#include "BusCommandRouterProxyStreamPoller.h" #include "CommandExecution.h" #include "CommandRouterHttpAPI.h" #include "CommandRouterHttpAPIConfig.h" @@ -257,8 +257,8 @@ TEST(CommandRouterHttpAPIConfigTest, from_config_loads_http_api_fields) { } TEST(CommandRouterHttpAPIConfigTest, from_config_uses_explicit_bus_client_endpoint) { - const auto config = CommandRouterHttpAPIConfig::from_config(make_command_router_config( - {{"http_api", {{"bus_client_endpoint", "localhost:50000"}}}})); + const auto config = CommandRouterHttpAPIConfig::from_config( + make_command_router_config({{"http_api", {{"bus_client_endpoint", "localhost:50000"}}}})); EXPECT_EQ(config.issuer_bus_endpoint, "localhost:50000"); } @@ -270,15 +270,15 @@ TEST(CommandRouterHttpAPIConfigTest, from_config_rejects_zero_stream_items_per_c } TEST(CommandRouterHttpAPIConfigTest, from_config_rejects_invalid_http_api_endpoint) { - EXPECT_THROW(CommandRouterHttpAPIConfig::from_config(make_command_router_config( - {{"http_api", {{"endpoint", "not-a-valid-endpoint"}}}})), + EXPECT_THROW(CommandRouterHttpAPIConfig::from_config( + make_command_router_config({{"http_api", {{"endpoint", "not-a-valid-endpoint"}}}})), runtime_error); } // ----------------------------------------------------------------------------- -// BusCommandRouterProxyClient (HTTP stream polling) +// BusCommandRouterProxyStreamPoller (HTTP stream polling) -TEST(BusCommandRouterProxyClientTest, stream_emits_one_item_per_chunk_by_default) { +TEST(BusCommandRouterProxyStreamPollerTest, stream_emits_one_item_per_chunk_by_default) { auto proxy = make_shared(); proxy->mark_routed(); proxy->enqueue_answers(5); @@ -287,14 +287,15 @@ TEST(BusCommandRouterProxyClientTest, stream_emits_one_item_per_chunk_by_default vector> chunks; auto on_chunk = [&](const vector& chunk) { chunks.push_back(chunk); }; - ASSERT_TRUE(poll_router_proxy_stream(proxy, "query", 1, nullptr, on_chunk, nullptr, nullptr)); + ASSERT_TRUE(BusCommandRouterProxyStreamPoller::poll_stream( + proxy, "query", 1, nullptr, on_chunk, nullptr, nullptr)); EXPECT_EQ(chunks.size(), 5u); EXPECT_EQ(chunk_sizes(chunks), (vector{1, 1, 1, 1, 1})); EXPECT_EQ(total_items(chunks), 5u); EXPECT_NE(chunks.front().front().find("answer-0"), string::npos); } -TEST(BusCommandRouterProxyClientTest, stream_groups_items_per_chunk) { +TEST(BusCommandRouterProxyStreamPollerTest, stream_groups_items_per_chunk) { auto proxy = make_shared(); proxy->mark_routed(); proxy->enqueue_answers(7); @@ -303,13 +304,14 @@ TEST(BusCommandRouterProxyClientTest, stream_groups_items_per_chunk) { vector> chunks; auto on_chunk = [&](const vector& chunk) { chunks.push_back(chunk); }; - ASSERT_TRUE(poll_router_proxy_stream(proxy, "query", 3, nullptr, on_chunk, nullptr, nullptr)); + ASSERT_TRUE(BusCommandRouterProxyStreamPoller::poll_stream( + proxy, "query", 3, nullptr, on_chunk, nullptr, nullptr)); EXPECT_EQ(chunks.size(), 3u); EXPECT_EQ(chunk_sizes(chunks), (vector{3, 3, 1})); EXPECT_EQ(total_items(chunks), 7u); } -TEST(BusCommandRouterProxyClientTest, rejects_zero_items_per_chunk) { +TEST(BusCommandRouterProxyStreamPollerTest, rejects_zero_items_per_chunk) { auto proxy = make_shared(); proxy->mark_routed(); proxy->enqueue_answers(1); @@ -318,18 +320,24 @@ TEST(BusCommandRouterProxyClientTest, rejects_zero_items_per_chunk) { string error_message; auto on_error = [&](const string& message) { error_message = message; }; - EXPECT_FALSE(poll_router_proxy_stream(proxy, "query", 0, nullptr, nullptr, on_error, nullptr)); + EXPECT_FALSE(BusCommandRouterProxyStreamPoller::poll_stream( + proxy, "query", 0, nullptr, nullptr, on_error, nullptr)); EXPECT_NE(error_message.find("items_per_chunk"), string::npos); } -TEST(BusCommandRouterProxyClientTest, get_and_set_emit_single_chunk) { +TEST(BusCommandRouterProxyStreamPollerTest, get_and_set_emit_single_chunk) { auto get_proxy = make_shared(); get_proxy->from_remote_peer(BusCommandRouterProxy::PARAMS_RESPONSE, {"params-body"}); vector> get_chunks; - ASSERT_TRUE(poll_router_proxy_stream( - get_proxy, "get", 1, nullptr, [&](const vector& chunk) { get_chunks.push_back(chunk); }, - nullptr, nullptr)); + ASSERT_TRUE(BusCommandRouterProxyStreamPoller::poll_stream( + get_proxy, + "get", + 1, + nullptr, + [&](const vector& chunk) { get_chunks.push_back(chunk); }, + nullptr, + nullptr)); ASSERT_EQ(get_chunks.size(), 1u); EXPECT_EQ(get_chunks[0], vector{"params-body"}); @@ -337,9 +345,14 @@ TEST(BusCommandRouterProxyClientTest, get_and_set_emit_single_chunk) { set_proxy->from_remote_peer(BusCommandRouterProxy::SET_PARAM_ACK, {"ack-body"}); vector> set_chunks; - ASSERT_TRUE(poll_router_proxy_stream( - set_proxy, "set", 1, nullptr, [&](const vector& chunk) { set_chunks.push_back(chunk); }, - nullptr, nullptr)); + ASSERT_TRUE(BusCommandRouterProxyStreamPoller::poll_stream( + set_proxy, + "set", + 1, + nullptr, + [&](const vector& chunk) { set_chunks.push_back(chunk); }, + nullptr, + nullptr)); ASSERT_EQ(set_chunks.size(), 1u); EXPECT_EQ(set_chunks[0], vector{"ack-body"}); } From a3bd115e643df6b37c139759cab248b4d1932f8d Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Fri, 26 Jun 2026 20:40:07 -0300 Subject: [PATCH 07/16] WIP --- .../BusCommandRouterProcessor.cc | 45 +++++++++++++++++ .../BusCommandRouterProcessor.h | 9 ++++ src/agents/command_router/http_api/BUILD | 5 +- .../BusCommandRouterProxyStreamPoller.cc | 9 +++- .../BusCommandRouterProxyStreamPoller.h | 12 ++--- .../http_api/CommandRouterHttpAPI.cc | 16 +++--- .../http_api/CommandRouterHttpAPI.h | 14 +++--- .../http_api/CommandRouterHttpAPIConfig.cc | 23 +++------ .../http_api/CommandRouterHttpAPIConfig.h | 4 +- .../http_api/CommandRouterHttpAPISingleton.cc | 29 +++++++---- .../http_api/CommandRouterHttpAPISingleton.h | 8 ++- src/main/bus_node.cc | 2 +- src/service_bus/BusCommandProxy.h | 5 ++ src/tests/cpp/command_router_http_api_test.cc | 49 ++++++++++++------- 14 files changed, 160 insertions(+), 70 deletions(-) diff --git a/src/agents/command_router/BusCommandRouterProcessor.cc b/src/agents/command_router/BusCommandRouterProcessor.cc index 24f4521b..fbf1b029 100644 --- a/src/agents/command_router/BusCommandRouterProcessor.cc +++ b/src/agents/command_router/BusCommandRouterProcessor.cc @@ -1,11 +1,13 @@ #include "BusCommandRouterProcessor.h" #include +#include #include #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 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& 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(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 = caller_proxy->command; + processor_proxy->args = caller_proxy->args; + + run_command(processor_proxy); +} + void BusCommandRouterProcessor::run_command(shared_ptr proxy) { auto router_proxy = dynamic_pointer_cast(proxy); if (router_proxy == nullptr) { diff --git a/src/agents/command_router/BusCommandRouterProcessor.h b/src/agents/command_router/BusCommandRouterProcessor.h index ab8219f6..c0208baf 100644 --- a/src/agents/command_router/BusCommandRouterProcessor.h +++ b/src/agents/command_router/BusCommandRouterProcessor.h @@ -31,6 +31,15 @@ class BusCommandRouterProcessor : public BusCommandProcessor { shared_ptr factory_empty_proxy() override; void run_command(shared_ptr proxy) override; + /** + * Run a router command for an HTTP execution without issuing bus_command_router on the mesh. + * + * Sets up caller and processor proxy RPC peers in-process (same pattern as ServiceBus::act), + * then invokes run_command(). The caller proxy receives responses for polling/WebSocket. + */ + void dispatch_http_command(const shared_ptr& caller_proxy, + const string& http_requestor_id); + private: void handle_get(shared_ptr proxy, const string& arg); void handle_set(shared_ptr proxy, const string& arg); diff --git a/src/agents/command_router/http_api/BUILD b/src/agents/command_router/http_api/BUILD index b8a4d560..9758010d 100644 --- a/src/agents/command_router/http_api/BUILD +++ b/src/agents/command_router/http_api/BUILD @@ -44,11 +44,11 @@ cc_library( ":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", - "//service_bus:service_bus_lib", ], ) @@ -60,7 +60,8 @@ cc_library( deps = [ ":command_router_http_api", ":command_router_http_api_config", + "//agents/command_router:bus_command_router_processor", "//commons/processor:processor_lib", - "//service_bus:service_bus_lib", + "//service_bus:bus_command_processor", ], ) diff --git a/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc b/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc index 8d6c1bec..9e096b31 100644 --- a/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc +++ b/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc @@ -49,6 +49,13 @@ bool BusCommandRouterProxyStreamPoller::poll_stream( 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 { @@ -118,7 +125,7 @@ bool BusCommandRouterProxyStreamPoller::poll_stream( router_proxy->parameters.get_or(BaseQueryProxy::USE_METTA_AS_QUERY_TOKENS, true); vector chunk_data; - while (!router_proxy->finished()) { + while (!finished_or_error()) { if (handle_abort()) { return false; } diff --git a/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.h b/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.h index 047a9862..526a8419 100644 --- a/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.h +++ b/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.h @@ -20,22 +20,22 @@ namespace command_router { class BusCommandRouterProxyStreamPoller { public: /** - * Poll @p router_proxy until the command finishes, is aborted, or fails. + * Poll router_proxy until the command finishes, is aborted, or fails. * - * For @p command_type "query" and "evolution", answers are popped from the proxy - * and forwarded in batches of at most @p items_per_chunk strings. For "get" and + * 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 @p on_chunk call for query/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 @p on_aborted is invoked. + * 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 @p should_abort returned true. + * @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& router_proxy, diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPI.cc b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc index 69c8ca04..ebd05aaa 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPI.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc @@ -4,6 +4,7 @@ #include #include +#include "BusCommandRouterProcessor.h" #include "BusCommandRouterProxy.h" #include "BusCommandRouterProxyStreamPoller.h" @@ -15,7 +16,6 @@ using namespace command_router; using namespace commons; using namespace processor; -using namespace service_bus; using namespace agents; using json = nlohmann::json; @@ -30,12 +30,14 @@ CommandRouterHttpAPI::CommandRouterHttpAPI(const string& host, int port, shared_ptr thread_pool, HttpAPISettings settings, - shared_ptr issuer_bus) + shared_ptr router_processor, + const string& bus_host) : Processor("command_router_http_api"), host(host), port(port), thread_pool(thread_pool), - issuer_bus(issuer_bus), + router_processor(std::move(router_processor)), + bus_host(bus_host), settings(std::move(settings)) { if (this->thread_pool == nullptr) { RAISE_ERROR("CommandRouterHttpAPI requires a non-null thread pool"); @@ -333,9 +335,9 @@ shared_ptr CommandRouterHttpAPI::find_execution(const string& } void CommandRouterHttpAPI::run_execution_inner(const shared_ptr& exec) { - if (this->issuer_bus == nullptr) { + if (this->router_processor == nullptr) { lock_guard lock(exec->mtx); - exec->mark_error("Issuer ServiceBus is not configured for command execution"); + exec->mark_error("Command router processor is not configured for HTTP execution"); return; } @@ -372,7 +374,9 @@ void CommandRouterHttpAPI::run_execution_inner(const shared_ptrcommand_text; Utils::replace_all(router_arg, "%", "$"); auto router_proxy = make_shared(exec->command_type, router_arg); - this->issuer_bus->issue_bus_command(router_proxy); + + string http_requestor_id = this->bus_host + ":http-" + exec->execution_id; + this->router_processor->dispatch_http_command(router_proxy, http_requestor_id); if (!BusCommandRouterProxyStreamPoller::poll_stream(router_proxy, exec->command_type, diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPI.h b/src/agents/command_router/http_api/CommandRouterHttpAPI.h index 620a304b..a09ff367 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPI.h +++ b/src/agents/command_router/http_api/CommandRouterHttpAPI.h @@ -13,19 +13,19 @@ #include "CommandRouterHttpAPIConfig.h" #include "DedicatedThread.h" #include "Processor.h" -#include "ServiceBus.h" #include "httplib.h" #include "nlohmann/json.hpp" #include "processor/ThreadPool.h" using namespace std; using namespace commons; -using namespace service_bus; using json = nlohmann::json; namespace command_router { +class BusCommandRouterProcessor; + /** * @brief HTTP + WebSocket server for asynchronous command execution. * @@ -35,8 +35,8 @@ namespace command_router { * POST /command-router/executions/{id}/cancel — request cancel * WS /command-router/ws/{id} — stream JSON events * - * Command execution uses a dedicated issuer ServiceBus (same path as bus_client) because - * the router node cannot issue bus_command_router to itself on ServiceBusSingleton. + * Command execution uses the registered BusCommandRouterProcessor via in-process proxy + * dispatch (dispatch_http_command). Each HTTP execution owns a caller proxy for streaming. * * Runs on a DedicatedThread: thread_one_step() blocks in listen() until stop(). * Each accepted command is enqueued on thread_pool so the listener stays free. @@ -57,7 +57,8 @@ class CommandRouterHttpAPI : public processor::Processor, public processor::Thre int port, shared_ptr thread_pool, HttpAPISettings settings = {}, - shared_ptr issuer_bus = nullptr); + shared_ptr router_processor = nullptr, + const string& bus_host = "localhost"); /** @brief Calls stop(). */ ~CommandRouterHttpAPI() override; @@ -83,7 +84,8 @@ class CommandRouterHttpAPI : public processor::Processor, public processor::Thre int port; httplib::Server server; shared_ptr thread_pool; - shared_ptr issuer_bus; + shared_ptr router_processor; + string bus_host; HttpAPISettings settings; atomic shutting_down{false}; unordered_map> executions; diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc index 41819ccb..88862c18 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc @@ -1,11 +1,14 @@ #include "CommandRouterHttpAPIConfig.h" +#include + #include "Utils.h" using namespace command_router; using namespace commons; -static pair parse_host_port(const string& endpoint, const string& config_path) { +static pair parse_host_port(const string& endpoint) { + string config_path = "command_router." + endpoint; auto tokens = Utils::split(endpoint, ':'); if (tokens.size() != 2) { RAISE_ERROR("Invalid " + config_path + " configuration: endpoint must be :"); @@ -21,17 +24,6 @@ static pair parse_host_port(const string& endpoint, const string& c return {tokens[0], port}; } -static string resolve_issuer_bus_endpoint(const JsonConfig& command_router_config) { - auto configured = command_router_config.at_path("http_api.bus_client_endpoint").get_or(""); - if (!configured.empty()) { - return configured; - } - - const string router_bus_endpoint = command_router_config.at_path("endpoint").get(); - auto [host, router_port] = parse_host_port(router_bus_endpoint, "command_router.endpoint"); - return host + ":" + to_string(router_port + 10); -} - static HttpAPISettings load_http_api_settings(const JsonConfig& command_router_config) { HttpAPISettings settings; settings.max_concurrent_executions = @@ -56,12 +48,11 @@ CommandRouterHttpAPIConfig CommandRouterHttpAPIConfig::from_config( const JsonConfig& command_router_config) { CommandRouterHttpAPIConfig config; tie(config.host, config.port) = - parse_host_port(command_router_config.at_path("http_api.endpoint").get(), - "command_router.http_api.endpoint"); + parse_host_port(command_router_config.at_path("http_api.endpoint").get()); config.settings = load_http_api_settings(command_router_config); config.thread_pool_size = command_router_config.at_path("http_api.thread_pool_size").get_or(4); - config.router_bus_endpoint = command_router_config.at_path("endpoint").get(); - config.issuer_bus_endpoint = resolve_issuer_bus_endpoint(command_router_config); + tie(config.bus_host, std::ignore) = + parse_host_port(command_router_config.at_path("endpoint").get()); return config; } diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h index 100614e4..4aea1f41 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h +++ b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h @@ -22,10 +22,10 @@ struct HttpAPISettings { struct CommandRouterHttpAPIConfig { string host; int port = 0; + /** Hostname of the command router bus node (from command_router.endpoint). */ + string bus_host; HttpAPISettings settings; unsigned int thread_pool_size = 4; - string router_bus_endpoint; - string issuer_bus_endpoint; static CommandRouterHttpAPIConfig from_config(const JsonConfig& command_router_config); }; diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.cc b/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.cc index ed237a02..1dff5f62 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.cc @@ -1,42 +1,53 @@ #include "CommandRouterHttpAPISingleton.h" +#include "BusCommandRouterProcessor.h" #include "CommandRouterHttpAPIConfig.h" -#include "ServiceBus.h" using namespace command_router; using namespace commons; -using namespace service_bus; bool CommandRouterHttpAPISingleton::INITIALIZED = false; shared_ptr CommandRouterHttpAPISingleton::HTTP_API = shared_ptr(nullptr); mutex CommandRouterHttpAPISingleton::API_MUTEX; -void CommandRouterHttpAPISingleton::init(const JsonConfig& command_router_config) { +void CommandRouterHttpAPISingleton::init(const JsonConfig& command_router_config, + const shared_ptr& processor) { lock_guard semaphore(API_MUTEX); if (INITIALIZED) { RAISE_ERROR( "CommandRouterHttpAPISingleton already initialized. " "CommandRouterHttpAPISingleton::init() should be called only once."); } - create_and_start(command_router_config); + create_and_start(command_router_config, processor); } -void CommandRouterHttpAPISingleton::create_and_start(const JsonConfig& command_router_config) { +void CommandRouterHttpAPISingleton::create_and_start(const JsonConfig& command_router_config, + const shared_ptr& processor) { + auto router_processor = dynamic_pointer_cast(processor); + + if (router_processor == nullptr) { + RAISE_ERROR("CommandRouterHttpAPISingleton requires a non-null BusCommandRouterProcessor"); + } + const auto config = CommandRouterHttpAPIConfig::from_config(command_router_config); - auto issuer_bus = make_shared(config.issuer_bus_endpoint, config.router_bus_endpoint); auto thread_pool_executor = make_shared("http_api_thread_pool_executor", config.thread_pool_size); - HTTP_API = make_shared( - config.host, config.port, thread_pool_executor, config.settings, issuer_bus); + + HTTP_API = make_shared(config.host, + config.port, + thread_pool_executor, + config.settings, + router_processor, + config.bus_host); auto http_api_thread = make_shared("http_api_thread", HTTP_API.get()); try { CommandRouterHttpAPI::initialize(HTTP_API, {http_api_thread, thread_pool_executor}); INITIALIZED = true; - } catch (const std::exception& e) { + } catch (const exception& e) { CommandRouterHttpAPISingleton::shutdown_on_startup_failure(); RAISE_ERROR( "CommandRouterHttpAPISingleton::init(): CommandRouterHttpAPI::initialize() failed: " + diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.h b/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.h index 17ad5a1c..503e3863 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.h +++ b/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.h @@ -4,15 +4,18 @@ #include #include "CommandRouterHttpAPI.h" +#include "BusCommandProcessor.h" #include "JsonConfig.h" using namespace std; +using namespace service_bus; namespace command_router { class CommandRouterHttpAPISingleton { public: - static void init(const JsonConfig& command_router_config); + static void init(const JsonConfig& command_router_config, + const shared_ptr& processor); static shared_ptr get_instance(); static void provide(shared_ptr http_api); @@ -21,7 +24,8 @@ class CommandRouterHttpAPISingleton { /** @brief Build and start HTTP API from command_router.http_api config. Sets HTTP_API and * INITIALIZED. */ - static void create_and_start(const JsonConfig& command_router_config); + static void create_and_start(const JsonConfig& command_router_config, + const shared_ptr& processor); /** @brief Stop and clear HTTP_API after a failed startup. */ static void shutdown_on_startup_failure(); diff --git a/src/main/bus_node.cc b/src/main/bus_node.cc index f278ff35..da865312 100644 --- a/src/main/bus_node.cc +++ b/src/main/bus_node.cc @@ -128,7 +128,7 @@ int main(int argc, char* argv[]) { mains::ProcessorType::COMMAND_ROUTER) { auto command_router_config = json_config.at_path("agents.command_router").get_or(JsonConfig()); - CommandRouterHttpAPISingleton::init(command_router_config); + CommandRouterHttpAPISingleton::init(command_router_config, service); } while (true) { diff --git a/src/service_bus/BusCommandProxy.h b/src/service_bus/BusCommandProxy.h index 2891ce89..cbdb6195 100644 --- a/src/service_bus/BusCommandProxy.h +++ b/src/service_bus/BusCommandProxy.h @@ -6,6 +6,10 @@ using namespace std; using namespace commons; using namespace distributed_algorithm_node; +namespace command_router { +class BusCommandRouterProcessor; +} // namespace command_router + namespace service_bus { class BusCommandProxy; @@ -83,6 +87,7 @@ class ProxyMessage : public Message { */ class BusCommandProxy { friend class ServiceBus; + friend class command_router::BusCommandRouterProcessor; public: // Commands allowed at the proxy level (caller <--> processor) diff --git a/src/tests/cpp/command_router_http_api_test.cc b/src/tests/cpp/command_router_http_api_test.cc index bc1260a0..f76e4e59 100644 --- a/src/tests/cpp/command_router_http_api_test.cc +++ b/src/tests/cpp/command_router_http_api_test.cc @@ -58,20 +58,16 @@ class HttpAPIServerFixture { } const unsigned int router_port = PortPool::get_port(); - const unsigned int issuer_port = PortPool::get_port(); const string router_id = TEST_HOST + ":" + std::to_string(router_port); - const string issuer_id = TEST_HOST + ":" + std::to_string(issuer_port); + this->router_processor = make_shared(); this->router_bus = make_shared(router_id); - this->router_bus->register_processor(make_shared()); - Utils::sleep(500); - - this->issuer_bus = make_shared(issuer_id, router_id); + this->router_bus->register_processor(this->router_processor); Utils::sleep(500); this->thread_pool = make_shared("test_thread_pool", num_threads); this->api = make_shared( - TEST_HOST, port, this->thread_pool, settings, this->issuer_bus); + TEST_HOST, port, this->thread_pool, settings, this->router_processor, TEST_HOST); this->api_thread = make_shared("test_api_thread", this->api.get()); CommandRouterHttpAPI::initialize(this->api, {this->api_thread, this->thread_pool}); Utils::sleep(300); @@ -82,7 +78,7 @@ class HttpAPIServerFixture { this->api->stop(); this->api = nullptr; } - this->issuer_bus = nullptr; + this->router_processor = nullptr; this->router_bus = nullptr; } @@ -96,7 +92,7 @@ class HttpAPIServerFixture { private: inline static bool service_bus_statics_initialized = false; shared_ptr router_bus; - shared_ptr issuer_bus; + shared_ptr router_processor; shared_ptr thread_pool; shared_ptr api; shared_ptr api_thread; @@ -247,8 +243,7 @@ TEST(CommandRouterHttpAPIConfigTest, from_config_loads_http_api_fields) { EXPECT_EQ(config.host, "localhost"); EXPECT_EQ(config.port, 40009); EXPECT_EQ(config.thread_pool_size, 8u); - EXPECT_EQ(config.router_bus_endpoint, "localhost:40008"); - EXPECT_EQ(config.issuer_bus_endpoint, "localhost:40018"); + EXPECT_EQ(config.bus_host, "localhost"); EXPECT_EQ(config.settings.max_concurrent_executions, 50u); EXPECT_EQ(config.settings.max_queued_executions, 200u); EXPECT_EQ(config.settings.max_events_per_execution, 5000u); @@ -256,13 +251,6 @@ TEST(CommandRouterHttpAPIConfigTest, from_config_loads_http_api_fields) { EXPECT_EQ(config.settings.stream_items_per_chunk, 25u); } -TEST(CommandRouterHttpAPIConfigTest, from_config_uses_explicit_bus_client_endpoint) { - const auto config = CommandRouterHttpAPIConfig::from_config( - make_command_router_config({{"http_api", {{"bus_client_endpoint", "localhost:50000"}}}})); - - EXPECT_EQ(config.issuer_bus_endpoint, "localhost:50000"); -} - TEST(CommandRouterHttpAPIConfigTest, from_config_rejects_zero_stream_items_per_chunk) { EXPECT_THROW(CommandRouterHttpAPIConfig::from_config( make_command_router_config({{"http_api", {{"stream_items_per_chunk", 0}}}})), @@ -275,6 +263,27 @@ TEST(CommandRouterHttpAPIConfigTest, from_config_rejects_invalid_http_api_endpoi runtime_error); } +// ----------------------------------------------------------------------------- +// BusCommandRouterProcessor HTTP dispatch + +TEST(BusCommandRouterProcessorTest, dispatch_http_command_get_returns_params) { + set commands = {ServiceBus::BUS_COMMAND_ROUTER}; + ServiceBus::initialize_statics(commands, 49400, 49499); + + auto router_processor = make_shared(); + const string router_id = TEST_HOST + ":" + std::to_string(PortPool::get_port()); + ServiceBus router_bus(router_id); + router_bus.register_processor(router_processor); + Utils::sleep(500); + + auto caller_proxy = make_shared("get", "params"); + router_processor->dispatch_http_command(caller_proxy, "exec-http-get-test"); + Utils::sleep(500); + + EXPECT_FALSE(caller_proxy->params_response.empty()); + EXPECT_TRUE(caller_proxy->finished()); +} + // ----------------------------------------------------------------------------- // BusCommandRouterProxyStreamPoller (HTTP stream polling) @@ -584,7 +593,9 @@ TEST_F(CommandRouterHttpAPISingletonTest, init_after_provide_throws) { CommandRouterHttpAPISingleton::provide(make_shared("localhost", 19005, pool)); json raw = {{"http_api", {{"endpoint", "localhost:19005"}}}}; - EXPECT_THROW(CommandRouterHttpAPISingleton::init(JsonConfig(raw)), runtime_error); + EXPECT_THROW( + CommandRouterHttpAPISingleton::init(JsonConfig(raw), make_shared()), + runtime_error); } int main(int argc, char** argv) { From c985d176c8f787c9a26af745871fe6d42ead9272 Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Mon, 29 Jun 2026 12:41:46 -0300 Subject: [PATCH 08/16] Change parameters value --- config/das.json | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/config/das.json b/config/das.json index 0d044343..5d72f169 100644 --- a/config/das.json +++ b/config/das.json @@ -134,8 +134,8 @@ "max_bundle_size": 1000, "max_answers": 0, "use_link_template_cache": false, - "populate_metta_mapping": true, - "use_metta_as_query_tokens": true, + "populate_metta_mapping": false, + "use_metta_as_query_tokens": false, "allow_incomplete_chain_path": false } }, From 8ee4d500ca2db358be9a3181f297855ebddb68d2 Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Mon, 29 Jun 2026 13:31:05 -0300 Subject: [PATCH 09/16] change num threads --- src/tests/cpp/command_router_http_api_test.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/tests/cpp/command_router_http_api_test.cc b/src/tests/cpp/command_router_http_api_test.cc index f76e4e59..4ce1ec18 100644 --- a/src/tests/cpp/command_router_http_api_test.cc +++ b/src/tests/cpp/command_router_http_api_test.cc @@ -194,7 +194,7 @@ class CommandRouterHttpAPIQueuedConcurrencyTest : public ::testing::Test { HttpAPISettings settings; settings.max_concurrent_executions = 2; settings.max_queued_executions = 10; - server.start(19007, settings, 1); + server.start(19007, settings, 4); } static void TearDownTestSuite() { server.stop(); } From 560cda138079fc0b05ee0020ba9acfa20a630adf Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Tue, 30 Jun 2026 11:04:43 -0300 Subject: [PATCH 10/16] Make CommandExecution thread-safe internally, simplify HTTP execution admission/cancel flow, require router processor at construction, and tighten stream poller error handling. --- .../BusCommandRouterProcessor.cc | 2 +- .../BusCommandRouterProxyStreamPoller.cc | 18 ++ .../http_api/CommandExecution.cc | 176 ++++++++++++++---- .../http_api/CommandExecution.h | 68 ++++--- .../http_api/CommandRouterHttpAPI.cc | 170 +++++------------ .../http_api/CommandRouterHttpAPI.h | 2 +- .../http_api/CommandRouterHttpAPIConfig.cc | 23 ++- .../http_api/CommandRouterHttpAPIConfig.h | 4 +- .../http_api/CommandRouterHttpAPISingleton.cc | 2 +- .../http_api/CommandRouterHttpAPISingleton.h | 2 +- 10 files changed, 269 insertions(+), 198 deletions(-) diff --git a/src/agents/command_router/BusCommandRouterProcessor.cc b/src/agents/command_router/BusCommandRouterProcessor.cc index fbf1b029..1d6aeadf 100644 --- a/src/agents/command_router/BusCommandRouterProcessor.cc +++ b/src/agents/command_router/BusCommandRouterProcessor.cc @@ -86,7 +86,7 @@ void BusCommandRouterProcessor::dispatch_http_command( processor_proxy->command = caller_proxy->command; processor_proxy->args = caller_proxy->args; - run_command(processor_proxy); + this->run_command(processor_proxy); } void BusCommandRouterProcessor::run_command(shared_ptr proxy) { diff --git a/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc b/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc index 9e096b31..9c78ae7d 100644 --- a/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc +++ b/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc @@ -82,6 +82,12 @@ bool BusCommandRouterProxyStreamPoller::poll_stream( } 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}); } @@ -101,6 +107,12 @@ bool BusCommandRouterProxyStreamPoller::poll_stream( } 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}); } @@ -120,6 +132,12 @@ bool BusCommandRouterProxyStreamPoller::poll_stream( } 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(BaseQueryProxy::USE_METTA_AS_QUERY_TOKENS, true); diff --git a/src/agents/command_router/http_api/CommandExecution.cc b/src/agents/command_router/http_api/CommandExecution.cc index bbf1d34d..d5b7d6fa 100644 --- a/src/agents/command_router/http_api/CommandExecution.cc +++ b/src/agents/command_router/http_api/CommandExecution.cc @@ -38,60 +38,168 @@ bool CommandExecution::is_terminal(ExecutionStatus status) { status == ExecutionStatus::ABORTED; } -void CommandExecution::publish_event(const json& payload) { - if (this->events.size() >= this->max_events && !this->is_terminal(this->status)) { - RAISE_ERROR("Execution event buffer full"); +ExecutionStatus CommandExecution::status() const { + lock_guard lock(this->mtx_); + return this->status_; +} + +string CommandExecution::status_string() const { return status_to_string(this->status()); } + +bool CommandExecution::is_terminal_status() const { return is_terminal(this->status()); } + +bool CommandExecution::is_cancel_requested() const { + lock_guard lock(this->mtx_); + return this->cancel_requested_; +} + +int CommandExecution::received_count() const { + lock_guard lock(this->mtx_); + return this->received_count_; +} + +long long CommandExecution::finished_at_ms() const { + lock_guard lock(this->mtx_); + return this->finished_at_ms_; +} + +json CommandExecution::to_status_json() const { + lock_guard lock(this->mtx_); + json body = {{"execution_id", this->execution_id}, + {"status", status_to_string(this->status_)}, + {"received_count", this->received_count_}, + {"total_items", this->total_items_}, + {"duration_ms", this->duration_ms_}}; + if (!this->error_message_.empty()) { + body["error_message"] = this->error_message_; } - this->events.push_back(payload.dump()); - this->cv.notify_all(); + return body; } -json CommandExecution::lifecycle_event() const { - return {{"execution_id", this->execution_id}, {"status", this->status_to_string(this->status)}}; +bool CommandExecution::try_request_cancel(ExecutionStatus* already_terminal) { + lock_guard lock(this->mtx_); + if (is_terminal(this->status_)) { + if (already_terminal != nullptr) { + *already_terminal = this->status_; + } + return false; + } + this->cancel_requested_ = true; + this->cv_.notify_all(); + return true; } -void CommandExecution::mark_running() { - this->status = ExecutionStatus::RUNNING; - this->publish_event(this->lifecycle_event()); +optional CommandExecution::wait_next_event(size_t& next_index, + chrono::milliseconds timeout, + bool& stream_finished) { + unique_lock lock(this->mtx_); + stream_finished = false; + + const bool ready = this->cv_.wait_for(lock, timeout, [&] { + return next_index < this->events_.size() || is_terminal(this->status_); + }); + + if (!ready) { + return nullopt; + } + + if (next_index >= this->events_.size()) { + stream_finished = is_terminal(this->status_); + return nullopt; + } + + string payload = this->events_[next_index++]; + stream_finished = is_terminal(this->status_) && next_index >= this->events_.size(); + return payload; } -void CommandExecution::publish_chunk(int seq, const vector& data) { - this->received_count += static_cast(data.size()); - this->publish_event({{"execution_id", this->execution_id}, - {"type", "chunk"}, - {"seq", seq}, - {"data", data}, - {"received_count", this->received_count}}); +bool CommandExecution::is_retention_expired(long long now_ms, long long retention_ms) const { + lock_guard lock(this->mtx_); + return is_terminal(this->status_) && this->finished_at_ms_ > 0 && + (now_ms - this->finished_at_ms_) > retention_ms; } -void CommandExecution::stamp_finished_at() { - if (this->finished_at_ms == 0) { - this->finished_at_ms = Utils::get_current_time_millis(); +void CommandExecution::publish_event_locked(const json& payload) { + if (this->events_.size() >= this->max_events && !is_terminal(this->status_)) { + RAISE_ERROR("Execution event buffer full"); + } + this->events_.push_back(payload.dump()); + this->cv_.notify_all(); +} + +json CommandExecution::lifecycle_event_locked() const { + return {{"execution_id", this->execution_id}, {"status", status_to_string(this->status_)}}; +} + +void CommandExecution::stamp_finished_at_locked() { + if (this->finished_at_ms_ == 0) { + this->finished_at_ms_ = Utils::get_current_time_millis(); } } +void CommandExecution::mark_running() { + lock_guard lock(this->mtx_); + this->status_ = ExecutionStatus::RUNNING; + this->publish_event_locked(this->lifecycle_event_locked()); +} + +void CommandExecution::publish_chunk(int seq, const vector& data) { + lock_guard lock(this->mtx_); + this->received_count_ += static_cast(data.size()); + this->publish_event_locked({{"execution_id", this->execution_id}, + {"type", "chunk"}, + {"seq", seq}, + {"data", data}, + {"received_count", this->received_count_}}); +} + void CommandExecution::mark_completed(long long duration_ms, int total_items) { - this->duration_ms = duration_ms; - this->total_items = total_items; - this->status = ExecutionStatus::COMPLETED; - this->stamp_finished_at(); - auto event = this->lifecycle_event(); + lock_guard lock(this->mtx_); + this->duration_ms_ = duration_ms; + this->total_items_ = total_items; + this->status_ = ExecutionStatus::COMPLETED; + this->stamp_finished_at_locked(); + auto event = this->lifecycle_event_locked(); event["duration_ms"] = duration_ms; event["total_items"] = total_items; - this->publish_event(event); + this->publish_event_locked(event); } void CommandExecution::mark_error(const string& message) { - this->error_message = message; - this->status = ExecutionStatus::ERROR; - this->stamp_finished_at(); - auto event = this->lifecycle_event(); + lock_guard lock(this->mtx_); + this->error_message_ = message; + this->status_ = ExecutionStatus::ERROR; + this->stamp_finished_at_locked(); + auto event = this->lifecycle_event_locked(); event["message"] = message; - this->publish_event(event); + this->publish_event_locked(event); } void CommandExecution::mark_aborted() { - this->status = ExecutionStatus::ABORTED; - this->stamp_finished_at(); - this->publish_event(this->lifecycle_event()); + lock_guard lock(this->mtx_); + this->status_ = ExecutionStatus::ABORTED; + this->stamp_finished_at_locked(); + this->publish_event_locked(this->lifecycle_event_locked()); +} + +void CommandExecution::mark_error_unless_terminal(const string& message) { + lock_guard lock(this->mtx_); + if (is_terminal(this->status_)) { + return; + } + this->error_message_ = message; + this->status_ = ExecutionStatus::ERROR; + this->stamp_finished_at_locked(); + auto event = this->lifecycle_event_locked(); + event["message"] = message; + this->publish_event_locked(event); +} + +void CommandExecution::mark_aborted_unless_terminal() { + lock_guard lock(this->mtx_); + if (is_terminal(this->status_)) { + return; + } + this->status_ = ExecutionStatus::ABORTED; + this->stamp_finished_at_locked(); + this->publish_event_locked(this->lifecycle_event_locked()); } diff --git a/src/agents/command_router/http_api/CommandExecution.h b/src/agents/command_router/http_api/CommandExecution.h index dd81dd2f..fe1c8c1e 100644 --- a/src/agents/command_router/http_api/CommandExecution.h +++ b/src/agents/command_router/http_api/CommandExecution.h @@ -1,7 +1,9 @@ #pragma once +#include #include #include +#include #include #include @@ -23,7 +25,7 @@ enum ExecutionStatus { PENDING, RUNNING, COMPLETED, ERROR, ABORTED }; * GET /executions/{id} and WebSocket replay. Status transitions always emit a * lifecycle event so clients can rely on status in the stream. * - * All mutable fields are protected by mtx. cv is notified on each new event. + * Thread-safe: callers do not need to lock mtx; public methods synchronize internally. */ class CommandExecution { public: @@ -46,30 +48,37 @@ class CommandExecution { string command_text; size_t max_events; - mutex mtx; - condition_variable cv; + ExecutionStatus status() const; + string status_string() const; + bool is_terminal_status() const; + bool is_cancel_requested() const; + int received_count() const; + long long finished_at_ms() const; - ExecutionStatus status = ExecutionStatus::PENDING; - int received_count = 0; - int total_items = 0; - long long duration_ms = 0; - long long finished_at_ms = 0; - string error_message; - bool cancel_requested = false; + /** @brief JSON body for GET /executions/{id}. */ + json to_status_json() const; - /** Serialised JSON payloads, in order, for WebSocket replay. */ - vector events; + /** + * @brief Request cancellation. + * @return false if already terminal; sets already_terminal when non-null. + */ + bool try_request_cancel(ExecutionStatus* already_terminal = nullptr); /** - * @brief Append one serialised JSON event and notify cv. - * Caller must hold mtx. Throws if the buffer is full while still running. + * @brief Wait for the next stream event or terminal state. + * @return payload when next_index points at a new event; nullopt on timeout or terminal drain. */ - void publish_event(const json& payload); + optional wait_next_event(size_t& next_index, + chrono::milliseconds timeout, + bool& stream_finished); + + /** @brief True when terminal and finished_at_ms is older than retention_ms. */ + bool is_retention_expired(long long now_ms, long long retention_ms) const; - /** @brief Append a chunk event and update received_count */ + /** @brief Append a chunk event and update received_count. */ void publish_chunk(int seq, const vector& data); - /** @brief PENDING -> RUNNING; emits a lifecycle event */ + /** @brief PENDING -> RUNNING; emits a lifecycle event. */ void mark_running(); /** @brief -> COMPLETED; sets duration/totals and emits a lifecycle event. */ @@ -78,15 +87,28 @@ class CommandExecution { /** @brief -> ERROR; sets error_message and emits a lifecycle event. */ void mark_error(const string& message); - /** @brief -> ABORTED; emits a lifecycle event (e.g. after cancel_requested). */ + /** @brief -> ABORTED; emits a lifecycle event. */ void mark_aborted(); - private: - /** @brief Set finished_at_ms once, on first transition to a terminal status. */ - void stamp_finished_at(); + void mark_error_unless_terminal(const string& message); + void mark_aborted_unless_terminal(); - /** @brief Base lifecycle payload: execution_id + current status. */ - json lifecycle_event() const; + private: + mutable mutex mtx_; + condition_variable cv_; + + ExecutionStatus status_ = ExecutionStatus::PENDING; + int received_count_ = 0; + int total_items_ = 0; + long long duration_ms_ = 0; + long long finished_at_ms_ = 0; + string error_message_; + bool cancel_requested_ = false; + vector events_; + + void publish_event_locked(const json& payload); + void stamp_finished_at_locked(); + json lifecycle_event_locked() const; }; } // namespace command_router diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPI.cc b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc index ebd05aaa..e6957cf9 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPI.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc @@ -29,16 +29,16 @@ const unordered_set CommandRouterHttpAPI::VALID_COMMAND_TYPES = { CommandRouterHttpAPI::CommandRouterHttpAPI(const string& host, int port, shared_ptr thread_pool, - HttpAPISettings settings, shared_ptr router_processor, + HttpAPISettings settings, const string& bus_host) : Processor("command_router_http_api"), host(host), port(port), thread_pool(thread_pool), - router_processor(std::move(router_processor)), - bus_host(bus_host), - settings(std::move(settings)) { + router_processor(router_processor), + settings(settings), + bus_host(bus_host) { if (this->thread_pool == nullptr) { RAISE_ERROR("CommandRouterHttpAPI requires a non-null thread pool"); } @@ -166,11 +166,7 @@ void CommandRouterHttpAPI::setup_routes() { break; } - string status_for_response; - { - lock_guard exec_semaphore(exec->mtx); - status_for_response = CommandExecution::status_to_string(exec->status); - } + string status_for_response = exec->status_string(); LOG_INFO("CommandRouter HTTP API execution scheduled id=" << exec->execution_id << " type=" << command_type); @@ -218,29 +214,12 @@ void CommandRouterHttpAPI::setup_routes() { bool stream_finished = false; while (!stream_finished && ws.is_open()) { - string payload; - - { - unique_lock exec_semaphore(exec->mtx); - - bool has_new_event = exec->cv.wait_for(exec_semaphore, chrono::seconds(1), [&] { - return next_index < exec->events.size() || - CommandExecution::is_terminal(exec->status); - }); - - if (!has_new_event) continue; - - if (next_index >= exec->events.size()) { - stream_finished = CommandExecution::is_terminal(exec->status); - continue; - } - - payload = exec->events[next_index++]; - stream_finished = - CommandExecution::is_terminal(exec->status) && next_index >= exec->events.size(); + auto payload = exec->wait_next_event(next_index, chrono::seconds(1), stream_finished); + if (!payload.has_value()) { + continue; } - bool send_ok = ws.send(payload); + bool send_ok = ws.send(*payload); if (!send_ok) { LOG_INFO("CommandRouter HTTP API WebSocket send failed id=" << execution_id); @@ -266,25 +245,10 @@ void CommandRouterHttpAPI::setup_routes() { return; } - lock_guard exec_semaphore(exec->mtx); - LOG_DEBUG("CommandRouter HTTP API GET /command-router/executions/" - << execution_id - << " status=" << CommandExecution::status_to_string(exec->status)); - - json success_body = { - {"execution_id", exec->execution_id}, - {"status", CommandExecution::status_to_string(exec->status)}, - {"received_count", exec->received_count}, - {"total_items", exec->total_items}, - {"duration_ms", exec->duration_ms}, - }; - - if (!exec->error_message.empty()) { - success_body["error_message"] = exec->error_message; - } + << execution_id << " status=" << exec->status_string()); - this->set_json_response(response, 200, success_body); + this->set_json_response(response, 200, exec->to_status_json()); }); // POST /command-router/executions/:id/cancel @@ -302,25 +266,15 @@ void CommandRouterHttpAPI::setup_routes() { } ExecutionStatus current_status; - { - lock_guard exec_semaphore(exec->mtx); - current_status = exec->status; - - if (CommandExecution::is_terminal(current_status)) { - this->set_json_response( - response, - 409, - {{"execution_id", execution_id}, - {"status", CommandExecution::status_to_string(current_status)}, - {"error", "Execution is already in a terminal state"}}); - return; - } - - exec->cancel_requested = true; + if (!exec->try_request_cancel(¤t_status)) { + this->set_json_response(response, + 409, + {{"execution_id", execution_id}, + {"status", CommandExecution::status_to_string(current_status)}, + {"error", "Execution is already in a terminal state"}}); + return; } - exec->cv.notify_all(); - LOG_INFO("CommandRouter HTTP API execution cancel requested id=" << execution_id); this->set_json_response( @@ -336,35 +290,21 @@ shared_ptr CommandRouterHttpAPI::find_execution(const string& void CommandRouterHttpAPI::run_execution_inner(const shared_ptr& exec) { if (this->router_processor == nullptr) { - lock_guard lock(exec->mtx); exec->mark_error("Command router processor is not configured for HTTP execution"); return; } const auto started_at = Utils::get_current_time_millis(); int seq = 0; - int received_count = 0; - auto should_abort = [&]() { - lock_guard lock(exec->mtx); - return exec->cancel_requested || this->shutting_down.load(); - }; - auto on_chunk = [&](const vector& chunk) { - lock_guard lock(exec->mtx); - received_count += static_cast(chunk.size()); - exec->publish_chunk(++seq, chunk); - }; - auto on_error = [&](const string& message) { - lock_guard lock(exec->mtx); - exec->mark_error(message); - }; + auto should_abort = [&]() { return exec->is_cancel_requested() || this->shutting_down.load(); }; + auto on_chunk = [&](const vector& chunk) { exec->publish_chunk(++seq, chunk); }; + auto on_error = [&](const string& message) { exec->mark_error(message); }; auto on_aborted = [&]() { - lock_guard lock(exec->mtx); exec->mark_aborted(); LOG_INFO("CommandRouter HTTP API execution aborted id=" << exec->execution_id); }; auto on_complete = [&](long long duration_ms, int total_items) { - lock_guard lock(exec->mtx); exec->mark_completed(duration_ms, total_items); LOG_INFO("CommandRouter HTTP API execution completed id=" << exec->execution_id << " duration_ms=" << duration_ms << " items=" << total_items); @@ -388,56 +328,49 @@ void CommandRouterHttpAPI::run_execution_inner(const shared_ptrreceived_count()); } catch (const exception& e) { on_error(e.what()); } } void CommandRouterHttpAPI::run_execution(const shared_ptr& exec) { + bool acquired_running_slot = false; + { - unique_lock semaphore(this->executions_mtx); - this->execution_slots_cv.wait(semaphore, [&] { + unique_lock lock(this->executions_mtx); + + // Block until a concurrent slot is free or the server is shutting down. + this->execution_slots_cv.wait(lock, [&] { return this->shutting_down.load() || this->running_executions < this->settings.max_concurrent_executions; }); - if (this->shutting_down.load()) { - lock_guard exec_semaphore(exec->mtx); - if (!CommandExecution::is_terminal(exec->status)) { - exec->mark_error("Server is shutting down"); + this->pending_executions--; + + if (!this->shutting_down.load()) { + if (!exec->is_cancel_requested()) { + this->running_executions++; + acquired_running_slot = true; } - this->pending_executions--; - return; } + } - { - lock_guard exec_semaphore(exec->mtx); - exec->mark_running(); - } - this->pending_executions--; - this->running_executions++; + if (this->shutting_down.load()) { + exec->mark_error_unless_terminal("Server is shutting down"); + return; } - try { - this->run_execution_inner(exec); - } catch (const exception& e) { - LOG_ERROR("CommandRouter HTTP API execution failed id=" << exec->execution_id << ": " - << e.what()); - lock_guard lock(exec->mtx); - if (!CommandExecution::is_terminal(exec->status)) { - exec->mark_error(string("Execution failed: ") + e.what()); - } - } catch (...) { - LOG_ERROR("CommandRouter HTTP API execution failed id=" << exec->execution_id - << ": unknown error"); - lock_guard lock(exec->mtx); - if (!CommandExecution::is_terminal(exec->status)) { - exec->mark_error("Execution failed: unknown error"); - } + if (!acquired_running_slot) { + exec->mark_aborted_unless_terminal(); + return; } - lock_guard semaphore(this->executions_mtx); + exec->mark_running(); + + this->run_execution_inner(exec); + + lock_guard lock(this->executions_mtx); this->running_executions--; this->execution_slots_cv.notify_one(); } @@ -450,16 +383,7 @@ void CommandRouterHttpAPI::cleanup_finished_executions() { const auto now = Utils::get_current_time_millis(); lock_guard semaphore(this->executions_mtx); for (auto it = this->executions.begin(); it != this->executions.end();) { - auto exec = it->second; - bool should_erase = false; - { - lock_guard exec_semaphore(exec->mtx); - const bool retention_expired = - exec->finished_at_ms > 0 && - (now - exec->finished_at_ms) > this->settings.execution_retention_ms; - should_erase = CommandExecution::is_terminal(exec->status) && retention_expired; - } - if (should_erase) { + if (it->second->is_retention_expired(now, this->settings.execution_retention_ms)) { it = this->executions.erase(it); } else { ++it; diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPI.h b/src/agents/command_router/http_api/CommandRouterHttpAPI.h index a09ff367..19fed3d9 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPI.h +++ b/src/agents/command_router/http_api/CommandRouterHttpAPI.h @@ -56,8 +56,8 @@ class CommandRouterHttpAPI : public processor::Processor, public processor::Thre CommandRouterHttpAPI(const string& host, int port, shared_ptr thread_pool, + shared_ptr router_processor, HttpAPISettings settings = {}, - shared_ptr router_processor = nullptr, const string& bus_host = "localhost"); /** @brief Calls stop(). */ diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc index 88862c18..2b8b894f 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc @@ -7,21 +7,17 @@ using namespace command_router; using namespace commons; -static pair parse_host_port(const string& endpoint) { +static pair parse_host_port(const string& endpoint) { string config_path = "command_router." + endpoint; auto tokens = Utils::split(endpoint, ':'); if (tokens.size() != 2) { RAISE_ERROR("Invalid " + config_path + " configuration: endpoint must be :"); } - int port; - try { - port = stoi(tokens[1]); - } catch (const exception&) { - RAISE_ERROR("Invalid " + config_path + " configuration: port must be an integer"); - } + string host = tokens[0]; + string port = tokens[1]; - return {tokens[0], port}; + return {host, port}; } static HttpAPISettings load_http_api_settings(const JsonConfig& command_router_config) { @@ -46,13 +42,16 @@ static HttpAPISettings load_http_api_settings(const JsonConfig& command_router_c CommandRouterHttpAPIConfig CommandRouterHttpAPIConfig::from_config( const JsonConfig& command_router_config) { - CommandRouterHttpAPIConfig config; - tie(config.host, config.port) = + auto [host, port] = parse_host_port(command_router_config.at_path("http_api.endpoint").get()); + auto [bus_host, _] = parse_host_port(command_router_config.at_path("endpoint").get()); + + CommandRouterHttpAPIConfig config; + config.host = host; + config.port = stoi(port); config.settings = load_http_api_settings(command_router_config); config.thread_pool_size = command_router_config.at_path("http_api.thread_pool_size").get_or(4); - tie(config.bus_host, std::ignore) = - parse_host_port(command_router_config.at_path("endpoint").get()); + config.bus_host = bus_host; return config; } diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h index 4aea1f41..fdb198da 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h +++ b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h @@ -15,14 +15,14 @@ struct HttpAPISettings { size_t max_queued_executions = 500; size_t max_events_per_execution = CommandExecution::DEFAULT_MAX_EVENTS; long long execution_retention_ms = 15 * 60 * 1000; - /** Max query/evolution answers per WebSocket chunk event (1 = one item per event). */ + // Max query/evolution answers per WebSocket chunk event (1 = one item per event). size_t stream_items_per_chunk = 1; }; struct CommandRouterHttpAPIConfig { string host; int port = 0; - /** Hostname of the command router bus node (from command_router.endpoint). */ + // Hostname of the command router bus node (from command_router.endpoint). string bus_host; HttpAPISettings settings; unsigned int thread_pool_size = 4; diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.cc b/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.cc index 1dff5f62..65751f33 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.cc @@ -38,8 +38,8 @@ void CommandRouterHttpAPISingleton::create_and_start(const JsonConfig& command_r HTTP_API = make_shared(config.host, config.port, thread_pool_executor, - config.settings, router_processor, + config.settings, config.bus_host); auto http_api_thread = make_shared("http_api_thread", HTTP_API.get()); diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.h b/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.h index 503e3863..fc387146 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.h +++ b/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.h @@ -3,8 +3,8 @@ #include #include -#include "CommandRouterHttpAPI.h" #include "BusCommandProcessor.h" +#include "CommandRouterHttpAPI.h" #include "JsonConfig.h" using namespace std; From fd29915f2edc719ea2a072319e5f23017fd7ec18 Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Tue, 30 Jun 2026 12:08:02 -0300 Subject: [PATCH 11/16] Fix tests --- config/das.json | 4 +- src/tests/cpp/command_router_http_api_test.cc | 69 +++++++++++++++---- 2 files changed, 58 insertions(+), 15 deletions(-) diff --git a/config/das.json b/config/das.json index 5d72f169..0d044343 100644 --- a/config/das.json +++ b/config/das.json @@ -134,8 +134,8 @@ "max_bundle_size": 1000, "max_answers": 0, "use_link_template_cache": false, - "populate_metta_mapping": false, - "use_metta_as_query_tokens": false, + "populate_metta_mapping": true, + "use_metta_as_query_tokens": true, "allow_incomplete_chain_path": false } }, diff --git a/src/tests/cpp/command_router_http_api_test.cc b/src/tests/cpp/command_router_http_api_test.cc index 4ce1ec18..10844f32 100644 --- a/src/tests/cpp/command_router_http_api_test.cc +++ b/src/tests/cpp/command_router_http_api_test.cc @@ -46,6 +46,23 @@ json make_execution_body(const string& command_type = "query", return {{"command_type", command_type}, {"command_text", command_text}}; } +class HangingQueryForwardProxy : public BusCommandProxy { + public: + void pack_command_line_args() override {} +}; + +/** Accepts forwarded queries but never responds, so HTTP executions stay running. */ +class HangingQueryForwardProcessor : public BusCommandProcessor { + public: + HangingQueryForwardProcessor() : BusCommandProcessor({ServiceBus::PATTERN_MATCHING_QUERY}) {} + + shared_ptr factory_empty_proxy() override { + return make_shared(); + } + + void run_command(shared_ptr /*proxy*/) override {} +}; + /** * HTTP API fixture with an in-process command router bus so executions reach "running". */ @@ -53,21 +70,28 @@ class HttpAPIServerFixture { public: void start(int port, const HttpAPISettings& settings = {}, unsigned int num_threads = 8) { if (!service_bus_statics_initialized) { - ServiceBus::initialize_statics({ServiceBus::BUS_COMMAND_ROUTER}, 49500, 49999); + ServiceBus::initialize_statics( + {ServiceBus::BUS_COMMAND_ROUTER, ServiceBus::PATTERN_MATCHING_QUERY}, 49500, 49999); service_bus_statics_initialized = true; } + const unsigned int query_port = PortPool::get_port(); + const string query_id = TEST_HOST + ":" + std::to_string(query_port); const unsigned int router_port = PortPool::get_port(); const string router_id = TEST_HOST + ":" + std::to_string(router_port); - this->router_processor = make_shared(); - this->router_bus = make_shared(router_id); + this->query_bus = make_shared(query_id); + this->query_bus->register_processor(make_shared()); + Utils::sleep(300); + + this->router_bus = make_shared(router_id, query_id); + this->router_processor = make_shared(this->router_bus); this->router_bus->register_processor(this->router_processor); Utils::sleep(500); this->thread_pool = make_shared("test_thread_pool", num_threads); this->api = make_shared( - TEST_HOST, port, this->thread_pool, settings, this->router_processor, TEST_HOST); + TEST_HOST, port, this->thread_pool, this->router_processor, settings, TEST_HOST); this->api_thread = make_shared("test_api_thread", this->api.get()); CommandRouterHttpAPI::initialize(this->api, {this->api_thread, this->thread_pool}); Utils::sleep(300); @@ -80,6 +104,7 @@ class HttpAPIServerFixture { } this->router_processor = nullptr; this->router_bus = nullptr; + this->query_bus = nullptr; } httplib::Client make_client(int port) const { @@ -91,6 +116,7 @@ class HttpAPIServerFixture { private: inline static bool service_bus_statics_initialized = false; + shared_ptr query_bus; shared_ptr router_bus; shared_ptr router_processor; shared_ptr thread_pool; @@ -220,15 +246,13 @@ TEST(CommandExecutionTest, status_and_terminal_flags) { TEST(CommandExecutionTest, terminal_marks_finished_at) { CommandExecution exec("exec-abc", "query", "(Similarity %V1 %V2)"); - lock_guard lock(exec.mtx); exec.mark_completed(100, 5); - EXPECT_GT(exec.finished_at_ms, 0); + EXPECT_GT(exec.finished_at_ms(), 0); } TEST(CommandExecutionTest, event_buffer_overflow_raises) { CommandExecution exec("exec-abc", "query", "(Similarity %V1 %V2)", 2); - lock_guard lock(exec.mtx); exec.mark_running(); exec.publish_chunk(1, {"a"}); EXPECT_THROW(exec.publish_chunk(2, {"b", "c"}), runtime_error); @@ -270,14 +294,14 @@ TEST(BusCommandRouterProcessorTest, dispatch_http_command_get_returns_params) { set commands = {ServiceBus::BUS_COMMAND_ROUTER}; ServiceBus::initialize_statics(commands, 49400, 49499); - auto router_processor = make_shared(); const string router_id = TEST_HOST + ":" + std::to_string(PortPool::get_port()); - ServiceBus router_bus(router_id); - router_bus.register_processor(router_processor); + auto router_bus = make_shared(router_id); + auto router_processor = make_shared(router_bus); + router_bus->register_processor(router_processor); Utils::sleep(500); auto caller_proxy = make_shared("get", "params"); - router_processor->dispatch_http_command(caller_proxy, "exec-http-get-test"); + router_processor->dispatch_http_command(caller_proxy, TEST_HOST + ":http-get-test"); Utils::sleep(500); EXPECT_FALSE(caller_proxy->params_response.empty()); @@ -366,6 +390,24 @@ TEST(BusCommandRouterProxyStreamPollerTest, get_and_set_emit_single_chunk) { EXPECT_EQ(set_chunks[0], vector{"ack-body"}); } +TEST(BusCommandRouterProxyStreamPollerTest, get_and_set_reject_empty_response) { + auto get_proxy = make_shared(); + get_proxy->mark_finished(); + + string get_error; + EXPECT_FALSE(BusCommandRouterProxyStreamPoller::poll_stream( + get_proxy, "get", 1, nullptr, nullptr, [&](const string& msg) { get_error = msg; }, nullptr)); + EXPECT_NE(get_error.find("params response"), string::npos); + + auto set_proxy = make_shared(); + set_proxy->mark_finished(); + + string set_error; + EXPECT_FALSE(BusCommandRouterProxyStreamPoller::poll_stream( + set_proxy, "set", 1, nullptr, nullptr, [&](const string& msg) { set_error = msg; }, nullptr)); + EXPECT_NE(set_error.find("parameter ack"), string::npos); +} + // ----------------------------------------------------------------------------- // HTTP API @@ -583,14 +625,15 @@ TEST_F(CommandRouterHttpAPITest, websocket_streams_lifecycle_events) { TEST_F(CommandRouterHttpAPISingletonTest, provide_and_get_instance) { auto pool = make_shared("test_pool", 1); - auto api = make_shared("localhost", 19002, pool); + auto api = make_shared("localhost", 19002, pool, nullptr); CommandRouterHttpAPISingleton::provide(api); EXPECT_EQ(CommandRouterHttpAPISingleton::get_instance().get(), api.get()); } TEST_F(CommandRouterHttpAPISingletonTest, init_after_provide_throws) { auto pool = make_shared("double_init_pool", 1); - CommandRouterHttpAPISingleton::provide(make_shared("localhost", 19005, pool)); + CommandRouterHttpAPISingleton::provide( + make_shared("localhost", 19005, pool, nullptr)); json raw = {{"http_api", {{"endpoint", "localhost:19005"}}}}; EXPECT_THROW( From 4ec241d6db0ae7704e0f0a7844b47970bf01ad98 Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Tue, 30 Jun 2026 12:25:58 -0300 Subject: [PATCH 12/16] apply lint --- src/agents/command_router/http_api/CommandExecution.cc | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/agents/command_router/http_api/CommandExecution.cc b/src/agents/command_router/http_api/CommandExecution.cc index d5b7d6fa..e5d762dc 100644 --- a/src/agents/command_router/http_api/CommandExecution.cc +++ b/src/agents/command_router/http_api/CommandExecution.cc @@ -94,9 +94,8 @@ optional CommandExecution::wait_next_event(size_t& next_index, unique_lock lock(this->mtx_); stream_finished = false; - const bool ready = this->cv_.wait_for(lock, timeout, [&] { - return next_index < this->events_.size() || is_terminal(this->status_); - }); + const bool ready = this->cv_.wait_for( + lock, timeout, [&] { return next_index < this->events_.size() || is_terminal(this->status_); }); if (!ready) { return nullopt; From fdc79813d5a9d9604f9bfcfe07d7a23035cdcc7b Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Wed, 1 Jul 2026 11:05:17 -0300 Subject: [PATCH 13/16] change params to false --- config/das.json | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/config/das.json b/config/das.json index 0d044343..5d72f169 100644 --- a/config/das.json +++ b/config/das.json @@ -134,8 +134,8 @@ "max_bundle_size": 1000, "max_answers": 0, "use_link_template_cache": false, - "populate_metta_mapping": true, - "use_metta_as_query_tokens": true, + "populate_metta_mapping": false, + "use_metta_as_query_tokens": false, "allow_incomplete_chain_path": false } }, From aa30a30d8c4ed5aa252e6dbd95436e6b6be147b9 Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Thu, 2 Jul 2026 09:37:45 -0300 Subject: [PATCH 14/16] Validate HTTP API config endpoints and fix command router tests Reject non-numeric ports and zero stream_items_per_chunk during config load, and repair the HTTP API test fixture with coverage for invalid endpoint values. --- .../BusCommandRouterProcessor.cc | 4 +-- .../http_api/CommandRouterHttpAPI.cc | 1 - .../http_api/CommandRouterHttpAPIConfig.cc | 20 ++++++++--- src/tests/cpp/command_router_http_api_test.cc | 35 ++++++++++++++----- 4 files changed, 44 insertions(+), 16 deletions(-) diff --git a/src/agents/command_router/BusCommandRouterProcessor.cc b/src/agents/command_router/BusCommandRouterProcessor.cc index 1d6aeadf..b3fe35d4 100644 --- a/src/agents/command_router/BusCommandRouterProcessor.cc +++ b/src/agents/command_router/BusCommandRouterProcessor.cc @@ -83,8 +83,8 @@ void BusCommandRouterProcessor::dispatch_http_command( 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 = caller_proxy->command; - processor_proxy->args = caller_proxy->args; + processor_proxy->command = std::move(caller_proxy->command); + processor_proxy->args = std::move(caller_proxy->args); this->run_command(processor_proxy); } diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPI.cc b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc index 8b9ae083..e6957cf9 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPI.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc @@ -166,7 +166,6 @@ void CommandRouterHttpAPI::setup_routes() { break; } - string status_for_response = exec->status_string(); string status_for_response = exec->status_string(); LOG_INFO("CommandRouter HTTP API execution scheduled id=" << exec->execution_id diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc index d08826aa..7a1c2a4e 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc @@ -7,7 +7,7 @@ using namespace command_router; using namespace commons; -static pair parse_host_port(const string& endpoint) { +static pair parse_host_port(const string& endpoint) { string config_path = "command_router." + endpoint; auto tokens = Utils::split(endpoint, ':'); if (tokens.size() != 2) { @@ -17,7 +17,11 @@ static pair parse_host_port(const string& endpoint) { string host = tokens[0]; string port = tokens[1]; - return {host, port}; + if (!Utils::is_number(port)) { + RAISE_ERROR("Invalid " + config_path + " configuration: port must be numeric"); + } + + return {host, stoi(port)}; } static HttpAPISettings load_http_api_settings(const JsonConfig& command_router_config) { @@ -32,6 +36,14 @@ static HttpAPISettings load_http_api_settings(const JsonConfig& command_router_c .get_or(settings.max_events_per_execution); settings.execution_retention_ms = command_router_config.at_path("http_api.execution_retention_ms") .get_or(settings.execution_retention_ms); + settings.stream_items_per_chunk = + command_router_config.at_path("http_api.stream_items_per_chunk") + .get_or(settings.stream_items_per_chunk); + if (settings.stream_items_per_chunk == 0) { + RAISE_ERROR( + "Invalid command_router.http_api.stream_items_per_chunk configuration: value must be at " + "least 1"); + } return settings; } @@ -43,10 +55,10 @@ CommandRouterHttpAPIConfig CommandRouterHttpAPIConfig::from_config( CommandRouterHttpAPIConfig config; config.host = host; - config.port = stoi(port); + config.port = port; + config.bus_host = bus_host; config.settings = load_http_api_settings(command_router_config); config.thread_pool_size = command_router_config.at_path("http_api.thread_pool_size").get_or(4); - config.bus_host = bus_host; return config; } diff --git a/src/tests/cpp/command_router_http_api_test.cc b/src/tests/cpp/command_router_http_api_test.cc index 14c9a299..64a724dc 100644 --- a/src/tests/cpp/command_router_http_api_test.cc +++ b/src/tests/cpp/command_router_http_api_test.cc @@ -1,4 +1,3 @@ -#include #include #include "AtomDBSingleton.h" @@ -25,7 +24,6 @@ using namespace command_router; using namespace processor; using namespace commons; -using namespace agents; using namespace atomdb; using namespace query_engine; using namespace service_bus; @@ -62,17 +60,22 @@ class HangingQueryForwardProcessor : public BusCommandProcessor { void run_command(shared_ptr /*proxy*/) override {} }; +void initialize_test_service_bus_statics_once() { + static bool initialized = false; + if (!initialized) { + ServiceBus::initialize_statics( + {ServiceBus::BUS_COMMAND_ROUTER, ServiceBus::PATTERN_MATCHING_QUERY}, 49400, 49999); + initialized = true; + } +} + /** * HTTP API fixture with an in-process command router bus so executions reach "running". */ class HttpAPIServerFixture { public: void start(int port, const HttpAPISettings& settings = {}, unsigned int num_threads = 8) { - if (!service_bus_statics_initialized) { - ServiceBus::initialize_statics( - {ServiceBus::BUS_COMMAND_ROUTER, ServiceBus::PATTERN_MATCHING_QUERY}, 49500, 49999); - service_bus_statics_initialized = true; - } + initialize_test_service_bus_statics_once(); const unsigned int query_port = PortPool::get_port(); const string query_id = TEST_HOST + ":" + std::to_string(query_port); @@ -99,8 +102,10 @@ class HttpAPIServerFixture { void stop() { if (this->api != nullptr) { this->api->stop(); - this->api = nullptr; } + this->api_thread = nullptr; + this->thread_pool = nullptr; + this->api = nullptr; this->router_processor = nullptr; this->router_bus = nullptr; this->query_bus = nullptr; @@ -114,7 +119,6 @@ class HttpAPIServerFixture { } private: - inline static bool service_bus_statics_initialized = false; shared_ptr query_bus; shared_ptr router_bus; shared_ptr router_processor; @@ -286,12 +290,25 @@ TEST(CommandRouterHttpAPIConfigTest, from_config_rejects_invalid_http_api_endpoi runtime_error); } +TEST(CommandRouterHttpAPIConfigTest, from_config_rejects_non_numeric_http_api_port) { + EXPECT_THROW(CommandRouterHttpAPIConfig::from_config( + make_command_router_config({{"http_api", {{"endpoint", "localhost:abc"}}}})), + runtime_error); +} + +TEST(CommandRouterHttpAPIConfigTest, from_config_rejects_trailing_junk_http_api_port) { + EXPECT_THROW(CommandRouterHttpAPIConfig::from_config( + make_command_router_config({{"http_api", {{"endpoint", "localhost:8080abc"}}}})), + runtime_error); +} + // ----------------------------------------------------------------------------- // BusCommandRouterProcessor HTTP dispatch TEST(BusCommandRouterProcessorTest, dispatch_http_command_get_returns_params) { set commands = {ServiceBus::BUS_COMMAND_ROUTER}; ServiceBus::initialize_statics(commands, 49400, 49499); + initialize_test_service_bus_statics_once(); const string router_id = TEST_HOST + ":" + std::to_string(PortPool::get_port()); auto router_bus = make_shared(router_id); From 81807aee6ce6f13c4e6ebb541fa06972bea8f696 Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Thu, 2 Jul 2026 12:18:41 -0300 Subject: [PATCH 15/16] Change doc --- src/agents/command_router/BusCommandRouterProcessor.h | 7 +------ .../command_router/http_api/CommandRouterHttpAPIConfig.cc | 5 ++--- 2 files changed, 3 insertions(+), 9 deletions(-) diff --git a/src/agents/command_router/BusCommandRouterProcessor.h b/src/agents/command_router/BusCommandRouterProcessor.h index c0208baf..f8b8b33f 100644 --- a/src/agents/command_router/BusCommandRouterProcessor.h +++ b/src/agents/command_router/BusCommandRouterProcessor.h @@ -31,12 +31,7 @@ class BusCommandRouterProcessor : public BusCommandProcessor { shared_ptr factory_empty_proxy() override; void run_command(shared_ptr proxy) override; - /** - * Run a router command for an HTTP execution without issuing bus_command_router on the mesh. - * - * Sets up caller and processor proxy RPC peers in-process (same pattern as ServiceBus::act), - * then invokes run_command(). The caller proxy receives responses for polling/WebSocket. - */ + /** Run a router command in-process for HTTP; responses are written to caller_proxy. */ void dispatch_http_command(const shared_ptr& caller_proxy, const string& http_requestor_id); diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc index 7a1c2a4e..619def40 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc @@ -36,9 +36,8 @@ static HttpAPISettings load_http_api_settings(const JsonConfig& command_router_c .get_or(settings.max_events_per_execution); settings.execution_retention_ms = command_router_config.at_path("http_api.execution_retention_ms") .get_or(settings.execution_retention_ms); - settings.stream_items_per_chunk = - command_router_config.at_path("http_api.stream_items_per_chunk") - .get_or(settings.stream_items_per_chunk); + settings.stream_items_per_chunk = command_router_config.at_path("http_api.stream_items_per_chunk") + .get_or(settings.stream_items_per_chunk); if (settings.stream_items_per_chunk == 0) { RAISE_ERROR( "Invalid command_router.http_api.stream_items_per_chunk configuration: value must be at " From d095efda3db56cc8717fe4aca3f9d29403945af1 Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Thu, 9 Jul 2026 13:56:34 -0300 Subject: [PATCH 16/16] Align HTTP API execution timing types and clarify slot-release locking --- .../command_router/http_api/CommandExecution.cc | 6 +++--- .../command_router/http_api/CommandExecution.h | 10 +++++----- .../http_api/CommandRouterHttpAPI.cc | 16 +++++++++++----- .../http_api/CommandRouterHttpAPIConfig.cc | 2 +- .../http_api/CommandRouterHttpAPIConfig.h | 2 +- 5 files changed, 21 insertions(+), 15 deletions(-) diff --git a/src/agents/command_router/http_api/CommandExecution.cc b/src/agents/command_router/http_api/CommandExecution.cc index e5d762dc..77b4157d 100644 --- a/src/agents/command_router/http_api/CommandExecution.cc +++ b/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 lock(this->mtx_); return this->finished_at_ms_; } @@ -111,7 +111,7 @@ optional 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 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& 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 lock(this->mtx_); this->duration_ms_ = duration_ms; this->total_items_ = total_items; diff --git a/src/agents/command_router/http_api/CommandExecution.h b/src/agents/command_router/http_api/CommandExecution.h index fe1c8c1e..d878575c 100644 --- a/src/agents/command_router/http_api/CommandExecution.h +++ b/src/agents/command_router/http_api/CommandExecution.h @@ -53,7 +53,7 @@ class CommandExecution { bool is_terminal_status() const; bool is_cancel_requested() const; int received_count() const; - long long finished_at_ms() const; + unsigned long finished_at_ms() const; /** @brief JSON body for GET /executions/{id}. */ json to_status_json() const; @@ -73,7 +73,7 @@ class CommandExecution { bool& stream_finished); /** @brief True when terminal and finished_at_ms is older than retention_ms. */ - bool is_retention_expired(long long now_ms, long long retention_ms) const; + bool is_retention_expired(unsigned long now_ms, unsigned long retention_ms) const; /** @brief Append a chunk event and update received_count. */ void publish_chunk(int seq, const vector& data); @@ -82,7 +82,7 @@ class CommandExecution { void mark_running(); /** @brief -> COMPLETED; sets duration/totals and emits a lifecycle event. */ - void mark_completed(long long duration_ms, int total_items); + void mark_completed(unsigned long duration_ms, int total_items); /** @brief -> ERROR; sets error_message and emits a lifecycle event. */ void mark_error(const string& message); @@ -100,8 +100,8 @@ class CommandExecution { ExecutionStatus status_ = ExecutionStatus::PENDING; int received_count_ = 0; int total_items_ = 0; - long long duration_ms_ = 0; - long long finished_at_ms_ = 0; + unsigned long duration_ms_ = 0; + unsigned long finished_at_ms_ = 0; string error_message_; bool cancel_requested_ = false; vector events_; diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPI.cc b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc index e6957cf9..103ff585 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPI.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc @@ -294,7 +294,9 @@ void CommandRouterHttpAPI::run_execution_inner(const shared_ptris_cancel_requested() || this->shutting_down.load(); }; @@ -304,7 +306,7 @@ void CommandRouterHttpAPI::run_execution_inner(const shared_ptrmark_aborted(); LOG_INFO("CommandRouter HTTP API execution aborted id=" << exec->execution_id); }; - auto on_complete = [&](long long duration_ms, int total_items) { + auto on_complete = [&](unsigned long duration_ms, int total_items) { exec->mark_completed(duration_ms, total_items); LOG_INFO("CommandRouter HTTP API execution completed id=" << exec->execution_id << " duration_ms=" << duration_ms << " items=" << total_items); @@ -328,7 +330,9 @@ void CommandRouterHttpAPI::run_execution_inner(const shared_ptrreceived_count()); + timer.stop(); + + on_complete(timer.milliseconds(), exec->received_count()); } catch (const exception& e) { on_error(e.what()); } @@ -370,8 +374,10 @@ void CommandRouterHttpAPI::run_execution(const shared_ptr& exe this->run_execution_inner(exec); - lock_guard lock(this->executions_mtx); - this->running_executions--; + { + lock_guard lock(this->executions_mtx); + this->running_executions--; + } this->execution_slots_cv.notify_one(); } diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc index 619def40..d3cfec73 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc @@ -35,7 +35,7 @@ static HttpAPISettings load_http_api_settings(const JsonConfig& command_router_c command_router_config.at_path("http_api.max_events_per_execution") .get_or(settings.max_events_per_execution); settings.execution_retention_ms = command_router_config.at_path("http_api.execution_retention_ms") - .get_or(settings.execution_retention_ms); + .get_or(settings.execution_retention_ms); settings.stream_items_per_chunk = command_router_config.at_path("http_api.stream_items_per_chunk") .get_or(settings.stream_items_per_chunk); if (settings.stream_items_per_chunk == 0) { diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h index fdb198da..f4a8871a 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h +++ b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h @@ -14,7 +14,7 @@ struct HttpAPISettings { size_t max_concurrent_executions = 100; size_t max_queued_executions = 500; size_t max_events_per_execution = CommandExecution::DEFAULT_MAX_EVENTS; - long long execution_retention_ms = 15 * 60 * 1000; + unsigned long execution_retention_ms = 15 * 60 * 1000; // Max query/evolution answers per WebSocket chunk event (1 = one item per event). size_t stream_items_per_chunk = 1; };