[#1146] Add synchronous get/set execution on POST /command-router/executions

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<string> chunks;
+                string error_message;
+                if (!this->execute_router_command(
+                        command_type,
+                        command_text,
+                        nullptr,
+                        [&](const vector<string>& 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<CommandExecution>(this->generate_execution_id(),
                                                       command_type,
                                                       command_text,
@@ -312,29 +345,54 @@ void CommandRouterHttpAPI::run_execution_inner(const shared_ptr<CommandExecution
                  << exec->execution_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;
+    }
+
+    timer.stop();
+
+    on_complete(timer.milliseconds(), 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<bool()>& should_abort,
+    const function<void(const vector<string>& chunk)>& on_chunk,
+    const function<void(const string& error)>& on_error,
+    const function<void()>& on_aborted) {
+    if (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<BusCommandRouterProxy>(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<BusCommandRouterProxy>(command_type, router_arg);
+
+        this->router_processor->dispatch_http_command(router_proxy, this->http_requestor_id);
 
-        timer.stop();
+        return BusCommandRouterProxyStreamPoller::poll_stream(router_proxy,
+                                                              command_type,
+                                                              this->settings.stream_items_per_chunk,
+                                                              should_abort,
+                                                              on_chunk,
+                                                              on_error,
+                                                              on_aborted);
 
-        on_complete(timer.milliseconds(), exec->received_count());
     } catch (const exception& e) {
-        on_error(e.what());
+        if (on_error) {
+            on_error(e.what());
+        }
+        return false;
     }
 }
 

src/agents/command_router/http_api/CommandRouterHttpAPI.h

@@ -2,6 +2,7 @@
 
 #include <atomic>
 #include <condition_variable>
+#include <functional>
 #include <memory>
 #include <mutex>
 #include <string>
@@ -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<processor::ThreadPool> thread_pool;
     shared_ptr<BusCommandRouterProcessor> router_processor;
     string bus_host;
+    string http_requestor_id;
     HttpAPISettings settings;
     atomic<bool> shutting_down{false};
     unordered_map<string, shared_ptr<CommandExecution>> 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<CommandExecution>& 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<bool()>& should_abort,
+                                const function<void(const vector<string>& chunk)>& on_chunk,
+                                const function<void(const string& error)>& on_error,
+                                const function<void()>& on_aborted);
+
+    static bool is_sync_command_type(const string& command_type);
+
     /** @brief Remove finished executions from the executions map. */
     void cleanup_finished_executions();
 

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<string>().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<string>().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<string>().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<string>();
+    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<string>().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);