[DAS-Dashboard#1225] Update usage of query params on command router.

das-dashboard/backend/controllers/query_controllers.py

@@ -36,9 +36,9 @@ def proxy_health_check():
 
 @router.post("/executions")
 def create_execution_on_proxy(body: QueryExecutionDto):
-    response = QUERY_SERVICES.execute_proxy_command(
-        body.command_type,
-        body.command_text,
+    response = QUERY_SERVICES.create_query_execution(
+        body.query_text,
+        body.parameters,
     )
 
     return JSONResponse(
@@ -132,8 +132,22 @@ async def _safe_send_error(
         pass
 
 
+def _stringify_proxy_error(error_value: Any) -> str:
+    if isinstance(error_value, str):
+        return error_value
+    return json.dumps(error_value)
+
+
 def _proxy_json_content(response: Response) -> Any:
     try:
-        return response.json()
+        content = response.json()
     except ValueError:
         return {"content": response.text}
+
+    if isinstance(content, dict) and "error" in content and "message" not in content:
+        error_value = content["error"]
+        content = {**content, "message": _stringify_proxy_error(error_value)}
+        if not isinstance(error_value, str) and "details" not in content:
+            content["details"] = error_value
+
+    return content

das-dashboard/backend/services/query_services.py

@@ -1,7 +1,6 @@
 import json
 import logging
 from collections.abc import AsyncIterator
-from concurrent.futures import ThreadPoolExecutor, as_completed
 
 import requests
 from requests import Response
@@ -13,9 +12,9 @@
 from shared.exceptions.custom_exceptions import CommandRouterConnectionError, CustomValueError
 from shared.internal.constants import LOCAL_HOSTS
 from shared.internal.web_configuration import WebConfiguration
+from shared.utils.command_router_payload import build_query_execution_payload
 from shared.utils.parse_query_answer import transform_stream_event
 
-VALID_COMMAND_TYPES = ("get", "set", "query")  # Evolution will be disconsidered for now.
 ROUTE_PREFIX = "/command-router"
 
 logger = logging.getLogger(__name__)
@@ -64,56 +63,21 @@ def get_execution_answers(
     def cancel_query_execution(self, execution_id: str) -> Response:
         return self._call_http_proxy("POST", f"{ROUTE_PREFIX}/executions/{execution_id}/cancel")
 
-    def execute_proxy_command(self, command_type: str, command_text: str) -> Response:
-        handlers = {
-            "get": self._get_query_parameters,
-            "set": self._set_query_parameters,
-            "query": self._create_query_execution,
-        }
-
-        if command_type not in VALID_COMMAND_TYPES:
-            raise CustomValueError(
-                "This proxy command does not exist/is not permitted in the current context."
-            )
-
-        return handlers[command_type](command_type, command_text)
-
-    def _get_query_parameters(self, command_type: str, command_text: str) -> Response:
-        return self._post_execution(command_type, command_text)
-
-    def _set_query_parameters(self, command_type: str, command_text: str) -> Response:
+    def create_query_execution(
+        self,
+        query_text: str,
+        parameters: dict | None = None,
+    ) -> Response:
         try:
-            custom_parameters_dict = json.loads(command_text)
-        except json.JSONDecodeError as error:
-            raise CustomValueError(f"Invalid query parameters payload: {error}") from error
-
-        if not custom_parameters_dict:
-            raise CustomValueError("No query parameters were provided.")
-
-        responses = []
-        max_workers = min(len(custom_parameters_dict), 8)
-
-        with ThreadPoolExecutor(max_workers=max_workers) as executor:
-            futures = [
-                executor.submit(
-                    self._post_execution,
-                    command_type,
-                    f"param {key} {self._format_param_value(value)}",
-                )
-                for key, value in custom_parameters_dict.items()
-            ]
-
-            for future in as_completed(futures):
-                responses.append(future.result())
-
-        failed_response = next(
-            (response for response in responses if response.status_code >= 400),
-            None,
-        )
-        if failed_response is not None:
-            return failed_response
+            payload = build_query_execution_payload(query_text, parameters)
+        except ValueError as error:
+            raise CustomValueError(str(error)) from error
 
-        return responses[-1]
+        return self._call_http_proxy(
+            "POST",
+            f"{ROUTE_PREFIX}/executions",
+            json=payload,
+        )
 
     def get_default_params_from_config(self) -> dict:
         config = self.web_config.load_raw_configuration()
@@ -131,22 +95,6 @@ def get_default_params_from_config(self) -> dict:
 
         return defaults
 
-    @staticmethod
-    def _format_param_value(value) -> str:
-        if isinstance(value, bool):
-            return "true" if value else "false"
-        return str(value)
-
-    def _create_query_execution(self, command_type: str, command_text: str) -> Response:
-        return self._post_execution(command_type, command_text)
-
-    def _post_execution(self, command_type: str, command_text: str) -> Response:
-        return self._call_http_proxy(
-            "POST",
-            f"{ROUTE_PREFIX}/executions",
-            json={"command_type": command_type, "command_text": command_text},
-        )
-
     def _call_http_proxy(self, method: str, path: str, **request_kwargs) -> Response:
         command_proxy_url = self._find_command_router_http_url()
         url = f"http://{command_proxy_url}{path}"

das-dashboard/backend/shared/dtos/query_execution_dto.py

@@ -1,5 +1,8 @@
-from pydantic import BaseModel
+from typing import Any
+
+from pydantic import BaseModel, Field
+
 
 class QueryExecutionDto(BaseModel):
-    command_type: str
-    command_text: str
+    query_text: str = Field(min_length=1)
+    parameters: dict[str, Any] | None = None

das-dashboard/backend/shared/utils/command_router_payload.py

@@ -0,0 +1,45 @@
+from typing import Any
+
+RESERVED_ROUTER_PARAM_KEYS = frozenset({"query"})
+
+
+def build_query_execution_payload(
+    query_text: str,
+    parameters: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    trimmed = query_text.strip()
+    if not trimmed:
+        raise ValueError("Query text must not be empty.")
+
+    params: dict[str, Any] = {
+        "query": {
+            "syntax": "metta",
+            "tokens": [trimmed],
+        }
+    }
+
+    if parameters:
+        params.update(normalize_router_parameters(parameters))
+
+    return {"command": "query", "params": params}
+
+
+def normalize_router_parameters(parameters: dict[str, Any]) -> dict[str, Any]:
+    normalized: dict[str, Any] = {}
+
+    for key, value in parameters.items():
+        if key in RESERVED_ROUTER_PARAM_KEYS:
+            raise ValueError(f"Reserved parameter '{key}' cannot be overridden.")
+
+        if isinstance(value, bool):
+            normalized[key] = value
+        elif isinstance(value, int):
+            normalized[key] = value
+        elif isinstance(value, float):
+            normalized[key] = value
+        elif isinstance(value, str):
+            normalized[key] = value
+        else:
+            normalized[key] = value
+
+    return normalized

das-dashboard/backend/shared/utils/parse_query_answer.py

@@ -37,23 +37,41 @@ def normalize_query_answer(item: Any) -> dict[str, Any] | None:
 
 
 def transform_stream_event(event: dict[str, Any]) -> dict[str, Any]:
-    if event.get("type") != "chunk":
-        return event
-
-    raw_items = event.get("data")
-    if not isinstance(raw_items, list):
-        return event
-
-    data = []
-    for item in raw_items:
-        normalized = normalize_query_answer(item)
-        if normalized is not None:
-            data.append(normalized)
-
-    return {
-        "execution_id": event.get("execution_id"),
-        "type": "chunk",
-        "seq": event.get("seq"),
-        "received_count": event.get("received_count"),
-        "data": data,
-    }
+    command = event.get("command")
+    params = event.get("params")
+
+    if command == "query_answers" and isinstance(params, dict):
+        raw_items = params.get("answers")
+        if not isinstance(raw_items, list):
+            return event
+
+        data = []
+        for item in raw_items:
+            normalized = normalize_query_answer(item)
+            if normalized is not None:
+                data.append(normalized)
+
+        return {
+            "execution_id": params.get("execution_id") or event.get("execution_id"),
+            "type": "chunk",
+            "seq": params.get("seq"),
+            "received_count": params.get("received_count"),
+            "data": data,
+        }
+
+    if command == "execution_status" and isinstance(params, dict):
+        transformed: dict[str, Any] = {
+            "execution_id": params.get("execution_id"),
+            "status": params.get("status"),
+        }
+
+        if params.get("message"):
+            transformed["message"] = params["message"]
+        if params.get("total_items") is not None:
+            transformed["received_count"] = params["total_items"]
+        elif params.get("received_count") is not None:
+            transformed["received_count"] = params["received_count"]
+
+        return transformed
+
+    return event

das-dashboard/backend/tests/test_command_router_payload.py

@@ -0,0 +1,35 @@
+import pytest
+
+from shared.utils.command_router_payload import build_query_execution_payload
+
+
+def test_build_payload_uses_query_text_for_query_tokens():
+    payload = build_query_execution_payload(
+        '(Similarity "human" %C)',
+        {"max_answers": 1, "populate_metta_mapping": True},
+    )
+
+    assert payload == {
+        "command": "query",
+        "params": {
+            "query": {
+                "syntax": "metta",
+                "tokens": ['(Similarity "human" %C)'],
+            },
+            "max_answers": 1,
+            "populate_metta_mapping": True,
+        },
+    }
+
+
+def test_build_payload_rejects_reserved_query_parameter():
+    with pytest.raises(ValueError, match="Reserved parameter 'query'"):
+        build_query_execution_payload(
+            '(Similarity "human" %C)',
+            {"query": {"syntax": "metta", "tokens": ["(evil query)"]}},
+        )
+
+
+def test_build_payload_requires_non_empty_query_text():
+    with pytest.raises(ValueError, match="Query text must not be empty"):
+        build_query_execution_payload("   ")

das-dashboard/src/api/APIUtils.js

@@ -11,7 +11,13 @@ export function extractErrorMessage(err, fallback = "An unexpected error occurre
     }
 
     if (data?.message) {
-      return data.message;
+      return typeof data.message === "string"
+        ? data.message
+        : JSON.stringify(data.message);
+    }
+
+    if (data?.error) {
+      return typeof data.error === "string" ? data.error : JSON.stringify(data.error);
     }
   }
 
@@ -38,6 +44,10 @@ export function extractErrorDetails(err) {
       return data.exceptionMessage;
     }
 
+    if (data?.error) {
+      return typeof data.error === "string" ? data.error : JSON.stringify(data.error);
+    }
+
     if (data?.detail) {
       return typeof data.detail === "string" ? data.detail : JSON.stringify(data.detail);
     }

das-dashboard/src/api/QueryAPI.js

@@ -5,19 +5,16 @@ export async function getQueryParamDefaults() {
   return response.data;
 }
 
-export async function setQueryParameters(params) {
-  const response = await api.post("/query/executions", {
-    command_type: "set",
-    command_text: JSON.stringify(params)
-  });
-  return response.data;
-}
+export async function startQueryExecution(queryText, parameters = null) {
+  const payload = {
+    query_text: queryText
+  };
 
-export async function startQueryExecution(queryText) {
-  const response = await api.post("/query/executions", {
-    command_type: "query",
-    command_text: queryText
-  });
+  if (parameters && Object.keys(parameters).length > 0) {
+    payload.parameters = parameters;
+  }
+
+  const response = await api.post("/query/executions", payload);
   return response.data;
 }
 

das-dashboard/src/hooks/useQueryExecution.js

@@ -1,7 +1,6 @@
 import { useCallback, useEffect, useMemo, useRef, useState } from "react";
 import {
   cancelQueryExecution,
-  setQueryParameters,
   startQueryExecution
 } from "../api/QueryAPI";
 import { extractApiError } from "../api/APIUtils";
@@ -179,12 +178,11 @@ export function useQueryExecution(parameters) {
       startedAtRef.current = Date.now();
 
       try {
-        const pendingParams = parameters.consumeQueryRunParameters();
-        if (Object.keys(pendingParams).length > 0) {
-          await setQueryParameters(pendingParams);
-        }
-
-        const { execution_id: nextExecutionId } = await startQueryExecution(trimmedQuery);
+        const runParameters = parameters.collectParameters();
+        const { execution_id: nextExecutionId } = await startQueryExecution(
+          trimmedQuery,
+          runParameters
+        );
         if (!nextExecutionId) {
           throw new Error("Query execution did not return an execution id.");
         }