[#1195] Simplify Command Router HTTP query execution

src/agents/command_router/BusCommandRouterProcessor.cc

@@ -63,7 +63,6 @@ void BusCommandRouterProcessor::dispatch_http_command(
 
     caller_proxy->issued = true;
     caller_proxy->requestor_id = http_requestor_id;
-    caller_proxy->parameters = parameters_for_peer(http_requestor_id);
     caller_proxy->serial = serial;
     caller_proxy->proxy_port = PortPool::get_port();
     if (caller_proxy->proxy_port == 0) {
@@ -86,8 +85,9 @@ void BusCommandRouterProcessor::dispatch_http_command(
     processor_proxy->setup_proxy_node(processor_proxy_node_id, caller_proxy->my_id());
     processor_proxy->command = std::move(caller_proxy->command);
     processor_proxy->args = std::move(caller_proxy->args);
+    processor_proxy->parameters = caller_proxy->parameters;
 
-    this->run_command(processor_proxy);
+    this->run_command_internal(processor_proxy, false);
 }
 
 void BusCommandRouterProcessor::run_command(shared_ptr<BusCommandProxy> proxy) {
@@ -96,7 +96,14 @@ void BusCommandRouterProcessor::run_command(shared_ptr<BusCommandProxy> proxy) {
         proxy->raise_error_on_peer("Invalid proxy type for BUS_COMMAND_ROUTER");
         return;
     }
-    router_proxy->parameters = parameters_for_peer(router_proxy->get_requestor_id());
+    this->run_command_internal(router_proxy, true);
+}
+
+void BusCommandRouterProcessor::run_command_internal(shared_ptr<BusCommandRouterProxy> router_proxy,
+                                                     bool load_peer_parameters) {
+    if (load_peer_parameters) {
+        router_proxy->parameters = parameters_for_peer(router_proxy->get_requestor_id());
+    }
     try {
         if (router_proxy->get_args().size() < 2) {
             RAISE_ERROR("Invalid bus_command_router args: expected {COMMAND, ARG}");

src/agents/command_router/BusCommandRouterProcessor.h

@@ -36,6 +36,8 @@ class BusCommandRouterProcessor : public BusCommandProcessor {
                                const string& http_requestor_id);
 
    private:
+    void run_command_internal(shared_ptr<BusCommandRouterProxy> router_proxy, bool load_peer_parameters);
+
     void handle_get(shared_ptr<BusCommandRouterProxy> proxy, const string& arg);
     void handle_set(shared_ptr<BusCommandRouterProxy> proxy, const string& arg);
     void handle_query(shared_ptr<BusCommandRouterProxy> proxy, const string& arg);

src/agents/command_router/http_api/BUILD

@@ -24,6 +24,30 @@ cc_library(
     ],
 )
 
+cc_library(
+    name = "proxy_parameters_from_json",
+    srcs = ["ProxyParametersFromJson.cc"],
+    hdrs = ["ProxyParametersFromJson.h"],
+    includes = ["."],
+    deps = [
+        "//commons:commons_lib",
+        "@nlohmann_json//:json",
+    ],
+)
+
+cc_library(
+    name = "http_command_proxy_factory",
+    srcs = ["HttpCommandProxyFactory.cc"],
+    hdrs = ["HttpCommandProxyFactory.h"],
+    includes = ["."],
+    deps = [
+        ":proxy_parameters_from_json",
+        "//agents/command_router:bus_command_router_proxy",
+        "//commons:commons_lib",
+        "@nlohmann_json//:json",
+    ],
+)
+
 cc_library(
     name = "command_router_http_api_config",
     srcs = ["CommandRouterHttpAPIConfig.cc"],
@@ -44,6 +68,7 @@ cc_library(
         ":bus_command_router_proxy_stream_poller",
         ":command_execution",
         ":command_router_http_api_config",
+        ":http_command_proxy_factory",
         "//agents/command_router:bus_command_router_processor",
         "//agents/command_router:bus_command_router_proxy",
         "//commons:commons_lib",

src/agents/command_router/http_api/CommandExecution.cc

@@ -5,13 +5,10 @@ using namespace commons;
 using namespace command_router;
 
 CommandExecution::CommandExecution(const string& execution_id,
-                                   const string& command_type,
-                                   const string& command_text,
+                                   const string& command,
+                                   const json& params,
                                    size_t max_events)
-    : execution_id(execution_id),
-      command_type(command_type),
-      command_text(command_text),
-      max_events(max_events) {
+    : execution_id(execution_id), command(command), params(params), max_events(max_events) {
     if (this->max_events == 0) {
         RAISE_ERROR("max_events must be greater than 0");
     }
@@ -125,7 +122,11 @@ void CommandExecution::publish_event_locked(const json& payload) {
     this->cv_.notify_all();
 }
 
-json CommandExecution::lifecycle_event_locked() const {
+json CommandExecution::make_envelope_locked(const string& command, json params) const {
+    return {{"command", command}, {"params", std::move(params)}};
+}
+
+json CommandExecution::status_params_locked() const {
     return {{"execution_id", this->execution_id}, {"status", status_to_string(this->status_)}};
 }
 
@@ -138,7 +139,8 @@ void CommandExecution::stamp_finished_at_locked() {
 void CommandExecution::mark_running() {
     lock_guard<mutex> lock(this->mtx_);
     this->status_ = ExecutionStatus::RUNNING;
-    this->publish_event_locked(this->lifecycle_event_locked());
+    this->publish_event_locked(
+        this->make_envelope_locked(COMMAND_EXECUTION_STATUS, this->status_params_locked()));
 }
 
 void CommandExecution::publish_chunk(int seq, const json& data) {
@@ -147,11 +149,11 @@ void CommandExecution::publish_chunk(int seq, const json& data) {
         RAISE_ERROR("Chunk data must be a JSON array");
     }
     this->received_count_ += static_cast<int>(data.size());
-    this->publish_event_locked({{"execution_id", this->execution_id},
-                                {"type", "chunk"},
-                                {"seq", seq},
-                                {"data", data},
-                                {"received_count", this->received_count_}});
+    json params = {{"execution_id", this->execution_id},
+                   {"seq", seq},
+                   {"answers", data},
+                   {"received_count", this->received_count_}};
+    this->publish_event_locked(this->make_envelope_locked(COMMAND_QUERY_ANSWERS, std::move(params)));
 }
 
 void CommandExecution::mark_completed(unsigned long duration_ms, int total_items) {
@@ -160,27 +162,28 @@ void CommandExecution::mark_completed(unsigned long duration_ms, int total_items
     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_locked(event);
+    json params = this->status_params_locked();
+    params["duration_ms"] = duration_ms;
+    params["total_items"] = total_items;
+    this->publish_event_locked(this->make_envelope_locked(COMMAND_EXECUTION_STATUS, std::move(params)));
 }
 
 void CommandExecution::mark_error(const string& message) {
     lock_guard<mutex> 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_locked(event);
+    json params = this->status_params_locked();
+    params["message"] = message;
+    this->publish_event_locked(this->make_envelope_locked(COMMAND_EXECUTION_STATUS, std::move(params)));
 }
 
 void CommandExecution::mark_aborted() {
     lock_guard<mutex> lock(this->mtx_);
     this->status_ = ExecutionStatus::ABORTED;
     this->stamp_finished_at_locked();
-    this->publish_event_locked(this->lifecycle_event_locked());
+    this->publish_event_locked(
+        this->make_envelope_locked(COMMAND_EXECUTION_STATUS, this->status_params_locked()));
 }
 
 void CommandExecution::mark_error_unless_terminal(const string& message) {
@@ -191,9 +194,9 @@ void CommandExecution::mark_error_unless_terminal(const string& message) {
     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);
+    json params = this->status_params_locked();
+    params["message"] = message;
+    this->publish_event_locked(this->make_envelope_locked(COMMAND_EXECUTION_STATUS, std::move(params)));
 }
 
 void CommandExecution::mark_aborted_unless_terminal() {
@@ -203,5 +206,6 @@ void CommandExecution::mark_aborted_unless_terminal() {
     }
     this->status_ = ExecutionStatus::ABORTED;
     this->stamp_finished_at_locked();
-    this->publish_event_locked(this->lifecycle_event_locked());
+    this->publish_event_locked(
+        this->make_envelope_locked(COMMAND_EXECUTION_STATUS, this->status_params_locked()));
 }

src/agents/command_router/http_api/CommandExecution.h

@@ -22,18 +22,22 @@ enum ExecutionStatus { PENDING, RUNNING, COMPLETED, ERROR, ABORTED };
  * @brief In-memory state for one asynchronous command run.
  *
  * Holds status, progress counters, and a JSON event log (events) consumed by
- * GET /executions/{id} and WebSocket replay. Status transitions always emit a
- * lifecycle event so clients can rely on status in the stream.
+ * GET /executions/{id} and WebSocket replay. Stream events use the same envelope
+ * as HTTP requests: { "command": ..., "params": ... }.
  *
  * Thread-safe: callers do not need to lock mtx; public methods synchronize internally.
  */
 class CommandExecution {
    public:
     static constexpr size_t DEFAULT_MAX_EVENTS = 10000;
 
+    /** WebSocket / stream command names (same envelope as HTTP requests). */
+    static constexpr const char* COMMAND_QUERY_ANSWERS = "query_answers";
+    static constexpr const char* COMMAND_EXECUTION_STATUS = "execution_status";
+
     CommandExecution(const string& execution_id,
-                     const string& command_type,
-                     const string& command_text,
+                     const string& command,
+                     const json& params,
                      size_t max_events = DEFAULT_MAX_EVENTS);
     ~CommandExecution() = default;
 
@@ -44,8 +48,8 @@ class CommandExecution {
     static bool is_terminal(ExecutionStatus status);
 
     string execution_id;
-    string command_type;
-    string command_text;
+    string command;
+    json params;
     size_t max_events;
 
     ExecutionStatus status() const;
@@ -75,19 +79,19 @@ class CommandExecution {
     /** @brief True when terminal and finished_at_ms is older than retention_ms. */
     bool is_retention_expired(unsigned long now_ms, unsigned long retention_ms) const;
 
-    /** @brief Append a chunk event and update received_count. @p data must be a JSON array. */
+    /** @brief Append a query_answers event and update received_count. @p data must be a JSON array. */
     void publish_chunk(int seq, const json& data);
 
-    /** @brief PENDING -> RUNNING; emits a lifecycle event. */
+    /** @brief PENDING -> RUNNING; emits an execution_status event. */
     void mark_running();
 
-    /** @brief -> COMPLETED; sets duration/totals and emits a lifecycle event. */
+    /** @brief -> COMPLETED; sets duration/totals and emits an execution_status event. */
     void mark_completed(unsigned long duration_ms, int total_items);
 
-    /** @brief -> ERROR; sets error_message and emits a lifecycle event. */
+    /** @brief -> ERROR; sets error_message and emits an execution_status event. */
     void mark_error(const string& message);
 
-    /** @brief -> ABORTED; emits a lifecycle event. */
+    /** @brief -> ABORTED; emits an execution_status event. */
     void mark_aborted();
 
     void mark_error_unless_terminal(const string& message);
@@ -108,7 +112,8 @@ class CommandExecution {
 
     void publish_event_locked(const json& payload);
     void stamp_finished_at_locked();
-    json lifecycle_event_locked() const;
+    json make_envelope_locked(const string& command, json params) const;
+    json status_params_locked() const;
 };
 
 }  // namespace command_router