From 7df3b513bcd9f50adea66e22bde9050dc9b9e0ff Mon Sep 17 00:00:00 2001 From: marcocapozzoli Date: Tue, 7 Jul 2026 17:29:43 -0300 Subject: [PATCH] Add sync command flow to Get and Set --- .../http_api/CommandRouterHttpAPI.cc | 93 +++++++++++++++---- .../http_api/CommandRouterHttpAPI.h | 19 +++- src/tests/cpp/command_router_http_api_test.cc | 59 ++++++++++++ 3 files changed, 150 insertions(+), 21 deletions(-) diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPI.cc b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc index e6957cf9..943da1a0 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPI.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc @@ -38,7 +38,8 @@ CommandRouterHttpAPI::CommandRouterHttpAPI(const string& host, thread_pool(thread_pool), router_processor(router_processor), settings(settings), - bus_host(bus_host) { + bus_host(bus_host), + http_requestor_id(bus_host + ":http-api:" + std::to_string(port)) { if (this->thread_pool == nullptr) { RAISE_ERROR("CommandRouterHttpAPI requires a non-null thread pool"); } @@ -149,6 +150,38 @@ void CommandRouterHttpAPI::setup_routes() { return; } + if (this->is_sync_command_type(command_type)) { + LOG_INFO("CommandRouter HTTP API sync execution type=" << command_type); + + vector chunks; + string error_message; + if (!this->execute_router_command( + command_type, + command_text, + nullptr, + [&](const vector& chunk) { + chunks.insert(chunks.end(), chunk.begin(), chunk.end()); + }, + [&](const string& message) { error_message = message; }, + nullptr)) { + if (error_message.empty()) { + error_message = "Command failed"; + } + this->set_json_response(response, 500, {{"error", error_message}}); + return; + } + + if (chunks.empty()) { + this->set_json_response( + response, 500, {{"error", "Command finished without a response"}}); + return; + } + + this->set_json_response( + response, 200, {{"command_type", command_type}, {"result", chunks.front()}}); + return; + } + auto exec = make_shared(this->generate_execution_id(), command_type, command_text, @@ -310,27 +343,51 @@ void CommandRouterHttpAPI::run_execution_inner(const shared_ptrexecution_id << " duration_ms=" << duration_ms << " items=" << total_items); }; + if (!this->execute_router_command( + exec->command_type, exec->command_text, should_abort, on_chunk, on_error, on_aborted)) { + return; + } + + on_complete(Utils::get_current_time_millis() - started_at, exec->received_count()); +} + +bool CommandRouterHttpAPI::is_sync_command_type(const string& command_type) { + return command_type == "get" || command_type == "set"; +} + +bool CommandRouterHttpAPI::execute_router_command( + const string& command_type, + const string& command_text, + const function& should_abort, + const function& chunk)>& on_chunk, + const function& on_error, + const function& on_aborted) { + if (this->router_processor == nullptr) { + if (on_error) { + on_error("Command router processor is not configured for HTTP execution"); + } + return false; + } + try { - string router_arg = exec->command_text; + string router_arg = command_text; Utils::replace_all(router_arg, "%", "$"); - auto router_proxy = make_shared(exec->command_type, router_arg); - - 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, - this->settings.stream_items_per_chunk, - should_abort, - on_chunk, - on_error, - on_aborted)) { - return; - } + auto router_proxy = make_shared(command_type, router_arg); + + this->router_processor->dispatch_http_command(router_proxy, this->http_requestor_id); - on_complete(Utils::get_current_time_millis() - started_at, exec->received_count()); + return BusCommandRouterProxyStreamPoller::poll_stream(router_proxy, + command_type, + this->settings.stream_items_per_chunk, + should_abort, + on_chunk, + on_error, + on_aborted); } catch (const exception& e) { - on_error(e.what()); + if (on_error) { + on_error(e.what()); + } + return false; } } diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPI.h b/src/agents/command_router/http_api/CommandRouterHttpAPI.h index 19fed3d9..02491a49 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPI.h +++ b/src/agents/command_router/http_api/CommandRouterHttpAPI.h @@ -2,6 +2,7 @@ #include #include +#include #include #include #include @@ -27,16 +28,17 @@ namespace command_router { class BusCommandRouterProcessor; /** - * @brief HTTP + WebSocket server for asynchronous command execution. + * @brief HTTP + WebSocket server for command execution. * * Routes: - * POST /command-router/executions — schedule a command + * POST /command-router/executions — run get/set synchronously, or schedule query/evolution * GET /command-router/executions/{id} — poll status * POST /command-router/executions/{id}/cancel — request cancel * WS /command-router/ws/{id} — stream JSON events * * Command execution uses the registered BusCommandRouterProcessor via in-process proxy - * dispatch (dispatch_http_command). Each HTTP execution owns a caller proxy for streaming. + * dispatch (dispatch_http_command). All HTTP requests share one router parameter store + * (http_requestor_id), matching how busclient reuses the same endpoint across commands. * * 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. @@ -86,6 +88,7 @@ class CommandRouterHttpAPI : public processor::Processor, public processor::Thre shared_ptr thread_pool; shared_ptr router_processor; string bus_host; + string http_requestor_id; HttpAPISettings settings; atomic shutting_down{false}; unordered_map> executions; @@ -105,6 +108,16 @@ class CommandRouterHttpAPI : public processor::Processor, public processor::Thre /** @brief Run command_type/command_text and publish chunk/lifecycle events. */ void run_execution_inner(const shared_ptr& exec); + /** @brief Dispatch a router command and poll its response stream. */ + bool execute_router_command(const string& command_type, + const string& command_text, + const function& should_abort, + const function& chunk)>& on_chunk, + const function& on_error, + const function& on_aborted); + + static bool is_sync_command_type(const string& command_type); + /** @brief Remove finished executions from the executions map. */ void cleanup_finished_executions(); diff --git a/src/tests/cpp/command_router_http_api_test.cc b/src/tests/cpp/command_router_http_api_test.cc index 64a724dc..8989658e 100644 --- a/src/tests/cpp/command_router_http_api_test.cc +++ b/src/tests/cpp/command_router_http_api_test.cc @@ -448,6 +448,65 @@ TEST_F(CommandRouterHttpAPITest, create_execution_returns_202) { EXPECT_EQ(payload["status"], "pending"); } +TEST_F(CommandRouterHttpAPITest, get_params_returns_sync_result) { + auto res = client().Post( + "/command-router/executions", make_execution_body("get", "params").dump(), "application/json"); + ASSERT_TRUE(res); + EXPECT_EQ(res->status, 200); + + auto payload = json::parse(res->body); + EXPECT_EQ(payload["command_type"], "get"); + EXPECT_TRUE(payload.contains("result")); + EXPECT_NE(payload["result"].get().find("use_cache"), string::npos); +} + +TEST_F(CommandRouterHttpAPITest, set_param_returns_sync_ack) { + auto res = client().Post("/command-router/executions", + make_execution_body("set", "param context test-context").dump(), + "application/json"); + ASSERT_TRUE(res); + EXPECT_EQ(res->status, 200); + + auto payload = json::parse(res->body); + EXPECT_EQ(payload["command_type"], "set"); + EXPECT_NE(payload["result"].get().find("context"), string::npos); + + auto get_res = client().Post( + "/command-router/executions", make_execution_body("get", "params").dump(), "application/json"); + ASSERT_TRUE(get_res); + ASSERT_EQ(get_res->status, 200); + EXPECT_NE(json::parse(get_res->body)["result"].get().find("context: test-context"), + string::npos); +} + +TEST_F(CommandRouterHttpAPITest, set_param_persists_for_later_get) { + auto set_res = client().Post("/command-router/executions", + make_execution_body("set", "param populate_metta_mapping true").dump(), + "application/json"); + ASSERT_TRUE(set_res); + ASSERT_EQ(set_res->status, 200); + + auto get_res = client().Post( + "/command-router/executions", make_execution_body("get", "params").dump(), "application/json"); + ASSERT_TRUE(get_res); + ASSERT_EQ(get_res->status, 200); + + const string params = json::parse(get_res->body)["result"].get(); + EXPECT_NE(params.find("populate_metta_mapping: true"), string::npos); +} + +TEST_F(CommandRouterHttpAPITest, set_param_rejects_unknown_key) { + auto res = client().Post("/command-router/executions", + make_execution_body("set", "param unknown_key value").dump(), + "application/json"); + ASSERT_TRUE(res); + EXPECT_EQ(res->status, 500); + + auto payload = json::parse(res->body); + EXPECT_TRUE(payload.contains("error")); + EXPECT_NE(payload["error"].get().find("Unknown parameter"), string::npos); +} + TEST_F(CommandRouterHttpAPITest, create_execution_rejects_invalid_requests) { auto bad_json = client().Post("/command-router/executions", "{bad", "application/json"); ASSERT_TRUE(bad_json);