From 4b8152079cfd5d4f069ad103e439d99eb120764f Mon Sep 17 00:00:00 2001 From: CCG Date: Thu, 28 May 2026 18:22:44 -0300 Subject: [PATCH 1/3] Add initial SystemParameters logics --- config/client.json | 39 -------- config/das.json | 67 +++++++++++-- src/agents/BaseQueryProxy.cc | 11 +-- .../context_broker/ContextBrokerProxy.cc | 16 +-- src/agents/evolution/QueryEvolutionProxy.cc | 6 +- src/agents/inference_agent/InferenceProxy.cc | 8 +- .../LinkCreationRequestProxy.cc | 12 +-- .../query_engine/PatternMatchingQueryProxy.cc | 6 +- src/commons/BUILD | 4 + src/commons/JsonConfig.h | 2 +- src/commons/JsonConfigParser.cc | 37 ++----- src/commons/JsonConfigParser.h | 12 +-- src/commons/Properties.h | 16 +++ src/commons/SystemParameters.cc | 99 +++++++++++++++++++ src/commons/SystemParameters.h | 33 +++++++ src/commons/SystemParametersSingleton.cc | 40 ++++++++ src/commons/SystemParametersSingleton.h | 27 +++++ src/main/README.md | 10 +- src/main/bus_client.cc | 64 ++++++------ src/main/bus_node.cc | 2 + src/main/helpers/Helper.cc | 43 +++++--- src/main/helpers/Helper.h | 5 +- src/tests/cpp/BUILD | 20 ++++ src/tests/cpp/config_parser_test.cc | 42 ++------ src/tests/cpp/context_test.cc | 8 +- src/tests/cpp/link_creation_agent_test.cc | 4 + src/tests/cpp/pattern_matching_query_test.cc | 5 +- src/tests/cpp/query_evolution_test.cc | 4 + src/tests/cpp/system_parameters_test.cc | 79 +++++++++++++++ src/tests/cpp/test_commons/BUILD | 19 +++- .../cpp/test_commons/TestSystemParams.cc | 82 +++++++++++++++ src/tests/cpp/test_commons/TestSystemParams.h | 21 ++++ 32 files changed, 615 insertions(+), 228 deletions(-) delete mode 100644 config/client.json create mode 100644 src/commons/SystemParameters.cc create mode 100644 src/commons/SystemParameters.h create mode 100644 src/commons/SystemParametersSingleton.cc create mode 100644 src/commons/SystemParametersSingleton.h create mode 100644 src/tests/cpp/system_parameters_test.cc create mode 100644 src/tests/cpp/test_commons/TestSystemParams.cc create mode 100644 src/tests/cpp/test_commons/TestSystemParams.h diff --git a/config/client.json b/config/client.json deleted file mode 100644 index 0d603bc0..00000000 --- a/config/client.json +++ /dev/null @@ -1,39 +0,0 @@ -{ - "schema_version": "1.0", - "params": { - "endpoint": "localhost:40000", - "ports-range": "52000:52999", - "das-config-file": "config/das.json", - "query": { - "max-answers": 100, - "max-bundle-size": 1000, - "count-flag": false, - "attention-update": 0, - "attention-correlation": 0, - "unique-assignment-flag": true, - "positive-importance-flag": false, - "populate-metta-mapping": true, - "use-metta-as-query-tokens": true - }, - "link_creation": { - "repeat-count": 1, - "interval": 0, - "timeout": 0 - }, - "evolution": { - "elitism-rate": 0.08, - "max-generations": 10, - "population-size": 50, - "selection-rate": 0.1, - "total-attention-tokens": 100000 - }, - "context": { - "context": "context", - "use-context-cache": true, - "enforce-cache-recreation": false, - "initial-rent-rate": 0.25, - "initial-spreading-rate-lowerbound": 0.5, - "initial-spreading-rate-upperbound": 0.7 - } - } -} diff --git a/config/das.json b/config/das.json index 9a39a685..fd4e9ef0 100644 --- a/config/das.json +++ b/config/das.json @@ -122,30 +122,77 @@ } }, "agents": { + "attention": { + "endpoint": "localhost:40001" + }, + "base_query": { + "params": { + "unique_assignment_flag": false, + "attention_update": 0, + "attention_correlation": 0, + "max_bundle_size": 1000, + "max_answers": 0, + "use_link_template_cache": false, + "populate_metta_mapping": false, + "use_metta_as_query_tokens": false, + "allow_incomplete_chain_path": false + } + }, "query": { "endpoint": "localhost:40002", - "ports_range": "42000:42999" + "ports_range": "42000:42999", + "params": { + "positive_importance_flag": false, + "disregard_importance_flag": false, + "unique_value_flag": false, + "count_flag": false + } }, "link_creation": { "endpoint": "localhost:40003", - "ports_range": "43000:43999" + "ports_range": "43000:43999", + "params": { + "max_answers": 10, + "repeat_count": 1, + "context": "context", + "attention_update": 0, + "attention_correlation": 0, + "positive_importance_flag": true, + "query_interval": 0, + "query_timeout": 0, + "use_metta_as_query_tokens": false + } }, "inference": { "endpoint": "localhost:40004", - "ports_range": "44000:44999" + "ports_range": "44000:44999", + "params": { + "inference_request_timeout": 86400, + "repeat_count": 5, + "max_answers": 150 + } }, "evolution": { "endpoint": "localhost:40005", - "ports_range": "45000:45999" - } - }, - "brokers": { - "attention": { - "endpoint": "localhost:40001" + "ports_range": "45000:45999", + "params": { + "population_size": 1000, + "max_generations": 100, + "elitism_rate": 0.01, + "selection_rate": 0.1 + } }, "context": { "endpoint": "localhost:40006", - "ports_range": "46000:46999" + "ports_range": "46000:46999", + "params": { + "context": "context", + "use_cache": true, + "enforce_cache_recreation": false, + "initial_rent_rate": 0.75, + "initial_spreading_rate_lowerbound": 0.1, + "initial_spreading_rate_upperbound": 0.1 + } }, "atomdb": { "endpoint": "localhost:40007", diff --git a/src/agents/BaseQueryProxy.cc b/src/agents/BaseQueryProxy.cc index 3a01fc84..5e8ad227 100644 --- a/src/agents/BaseQueryProxy.cc +++ b/src/agents/BaseQueryProxy.cc @@ -2,6 +2,7 @@ #include "Logger.h" #include "ServiceBus.h" +#include "SystemParametersSingleton.h" using namespace agents; @@ -39,15 +40,7 @@ BaseQueryProxy::BaseQueryProxy(const vector& tokens, const string& conte void BaseQueryProxy::init() { this->atomdb = AtomDBSingleton::get_instance(); this->answer_count = 0; - this->parameters[UNIQUE_ASSIGNMENT_FLAG] = false; - this->parameters[ATTENTION_UPDATE] = (unsigned int) NONE; - this->parameters[ATTENTION_CORRELATION] = (unsigned int) NONE; - this->parameters[MAX_BUNDLE_SIZE] = (unsigned int) 1000; - this->parameters[MAX_ANSWERS] = (unsigned int) 0; // No limit - this->parameters[USE_LINK_TEMPLATE_CACHE] = false; - this->parameters[POPULATE_METTA_MAPPING] = false; - this->parameters[USE_METTA_AS_QUERY_TOKENS] = false; - this->parameters[ALLOW_INCOMPLETE_CHAIN_PATH] = false; + this->parameters = SystemParametersSingleton::get_instance()->get_base_query_params(); } BaseQueryProxy::~BaseQueryProxy() {} diff --git a/src/agents/context_broker/ContextBrokerProxy.cc b/src/agents/context_broker/ContextBrokerProxy.cc index c5bb0a3d..1e6b971c 100644 --- a/src/agents/context_broker/ContextBrokerProxy.cc +++ b/src/agents/context_broker/ContextBrokerProxy.cc @@ -1,18 +1,17 @@ #include "ContextBrokerProxy.h" #include "AtomDBSingleton.h" -#include "AttentionBrokerServer.h" #include "Hasher.h" #include "RedisMongoDB.h" #include "ServiceBus.h" #include "ServiceBusSingleton.h" +#include "SystemParametersSingleton.h" #define LOG_LEVEL INFO_LEVEL #include "Logger.h" using namespace context_broker; using namespace atomdb; -using namespace attention_broker; using namespace query_engine; using namespace service_bus; using namespace commons; @@ -36,13 +35,6 @@ string ContextBrokerProxy::INITIAL_RENT_RATE = "initial_rent_rate"; string ContextBrokerProxy::INITIAL_SPREADING_RATE_LOWERBOUND = "initial_spreading_rate_lowerbound"; string ContextBrokerProxy::INITIAL_SPREADING_RATE_UPPERBOUND = "initial_spreading_rate_upperbound"; -// Default values for AttentionBrokerClient::set_parameters() -double ContextBrokerProxy::DEFAULT_RENT_RATE = AttentionBrokerServer::RENT_RATE; -double ContextBrokerProxy::DEFAULT_SPREADING_RATE_LOWERBOUND = - AttentionBrokerServer::SPREADING_RATE_LOWERBOUND; -double ContextBrokerProxy::DEFAULT_SPREADING_RATE_UPPERBOUND = - AttentionBrokerServer::SPREADING_RATE_UPPERBOUND; - // ------------------------------------------------------------------------------------------------- // Constructors, destructors and initialization @@ -196,11 +188,7 @@ void ContextBrokerProxy::init(const string& name) { } void ContextBrokerProxy::set_default_query_parameters() { - this->parameters[USE_CACHE] = (bool) true; - this->parameters[ENFORCE_CACHE_RECREATION] = (bool) false; - this->parameters[INITIAL_RENT_RATE] = (double) DEFAULT_RENT_RATE; - this->parameters[INITIAL_SPREADING_RATE_LOWERBOUND] = (double) DEFAULT_SPREADING_RATE_LOWERBOUND; - this->parameters[INITIAL_SPREADING_RATE_UPPERBOUND] = (double) DEFAULT_SPREADING_RATE_UPPERBOUND; + this->parameters = SystemParametersSingleton::get_instance()->get_context_agent_params(); } // --------------------------------------------------------------------------------------------- diff --git a/src/agents/evolution/QueryEvolutionProxy.cc b/src/agents/evolution/QueryEvolutionProxy.cc index 5797ac61..9e8e2853 100644 --- a/src/agents/evolution/QueryEvolutionProxy.cc +++ b/src/agents/evolution/QueryEvolutionProxy.cc @@ -2,6 +2,7 @@ #include "FitnessFunctionRegistry.h" #include "ServiceBus.h" +#include "SystemParametersSingleton.h" #define LOG_LEVEL INFO_LEVEL #include "Logger.h" @@ -55,10 +56,7 @@ void QueryEvolutionProxy::init() { } void QueryEvolutionProxy::set_default_query_parameters() { - this->parameters[POPULATION_SIZE] = (unsigned int) 1000; - this->parameters[MAX_GENERATIONS] = (unsigned int) 100; - this->parameters[ELITISM_RATE] = (double) 0.01; - this->parameters[SELECTION_RATE] = (double) 0.1; + this->parameters = SystemParametersSingleton::get_instance()->get_evolution_agent_params(); } string QueryEvolutionProxy::to_string() { diff --git a/src/agents/inference_agent/InferenceProxy.cc b/src/agents/inference_agent/InferenceProxy.cc index 6e6ee3e4..9fadb3bf 100644 --- a/src/agents/inference_agent/InferenceProxy.cc +++ b/src/agents/inference_agent/InferenceProxy.cc @@ -1,6 +1,7 @@ #include "InferenceProxy.h" #include "ServiceBus.h" +#include "SystemParametersSingleton.h" using namespace inference_agent; @@ -21,12 +22,7 @@ InferenceProxy::~InferenceProxy() {} void InferenceProxy::pack_command_line_args() { tokenize(this->args); } void InferenceProxy::set_default_parameters() { - this->parameters[INFERENCE_REQUEST_TIMEOUT] = - (unsigned int) 24 * 60 * 60; // Default timeout is 24 hours - this->parameters[ATTENTION_CORRELATION] = (unsigned int) BaseQueryProxy::NONE; - this->parameters[ATTENTION_UPDATE] = (unsigned int) BaseQueryProxy::NONE; - this->parameters[REPEAT_COUNT] = (unsigned int) 5; - this->parameters[MAX_ANSWERS] = (unsigned int) 150; + this->parameters = SystemParametersSingleton::get_instance()->get_inference_agent_params(); } void InferenceProxy::set_parameter(const string& key, const PropertyValue& value) { diff --git a/src/agents/link_creation_agent/LinkCreationRequestProxy.cc b/src/agents/link_creation_agent/LinkCreationRequestProxy.cc index d6a69af8..1f41eb33 100644 --- a/src/agents/link_creation_agent/LinkCreationRequestProxy.cc +++ b/src/agents/link_creation_agent/LinkCreationRequestProxy.cc @@ -2,6 +2,7 @@ #include "BaseQueryProxy.h" #include "ServiceBus.h" +#include "SystemParametersSingleton.h" using namespace link_creation_agent; @@ -29,16 +30,7 @@ LinkCreationRequestProxy::~LinkCreationRequestProxy() {} void LinkCreationRequestProxy::pack_command_line_args() { tokenize(this->args); } void LinkCreationRequestProxy::set_default_parameters() { - this->parameters[LinkCreationRequestProxy::MAX_ANSWERS] = (unsigned int) 10; - this->parameters[LinkCreationRequestProxy::REPEAT_COUNT] = (unsigned int) 1; - this->parameters[LinkCreationRequestProxy::CONTEXT] = string(""); - this->parameters[LinkCreationRequestProxy::ATTENTION_UPDATE] = (unsigned int) BaseQueryProxy::NONE; - this->parameters[LinkCreationRequestProxy::ATTENTION_CORRELATION] = - (unsigned int) BaseQueryProxy::NONE; - this->parameters[LinkCreationRequestProxy::POSITIVE_IMPORTANCE_FLAG] = true; - this->parameters[LinkCreationRequestProxy::QUERY_INTERVAL] = (unsigned int) 0; - this->parameters[LinkCreationRequestProxy::QUERY_TIMEOUT] = (unsigned int) 0; - this->parameters[LinkCreationRequestProxy::USE_METTA_AS_QUERY_TOKENS] = false; + this->parameters = SystemParametersSingleton::get_instance()->get_link_creation_agent_params(); } void LinkCreationRequestProxy::set_parameter(const string& key, const PropertyValue& value) { diff --git a/src/agents/query_engine/PatternMatchingQueryProxy.cc b/src/agents/query_engine/PatternMatchingQueryProxy.cc index cdcc94b9..575d29a9 100644 --- a/src/agents/query_engine/PatternMatchingQueryProxy.cc +++ b/src/agents/query_engine/PatternMatchingQueryProxy.cc @@ -1,6 +1,7 @@ #include "PatternMatchingQueryProxy.h" #include "ServiceBus.h" +#include "SystemParametersSingleton.h" #define LOG_LEVEL INFO_LEVEL #include "Logger.h" @@ -32,10 +33,7 @@ PatternMatchingQueryProxy::PatternMatchingQueryProxy(const vector& token } void PatternMatchingQueryProxy::set_default_parameters() { - this->parameters[POSITIVE_IMPORTANCE_FLAG] = false; - this->parameters[DISREGARD_IMPORTANCE_FLAG] = false; - this->parameters[UNIQUE_VALUE_FLAG] = false; - this->parameters[COUNT_FLAG] = false; + this->parameters = SystemParametersSingleton::get_instance()->get_query_agent_params(); } PatternMatchingQueryProxy::~PatternMatchingQueryProxy() {} diff --git a/src/commons/BUILD b/src/commons/BUILD index c7278311..fdb70cbb 100644 --- a/src/commons/BUILD +++ b/src/commons/BUILD @@ -10,6 +10,8 @@ cc_library( "JsonConfigParser.cc", "SharedQueue.cc", "StoppableThread.cc", + "SystemParameters.cc", + "SystemParametersSingleton.cc", "Utils.cc", ], hdrs = [ @@ -25,6 +27,8 @@ cc_library( "SharedQueue.h", "Stoppable.h", "StoppableThread.h", + "SystemParameters.h", + "SystemParametersSingleton.h", "ThreadPool.h", "ThreadSafeHashmap.h", "ThreadSafeHeap.h", diff --git a/src/commons/JsonConfig.h b/src/commons/JsonConfig.h index 968843f0..a0b092d9 100644 --- a/src/commons/JsonConfig.h +++ b/src/commons/JsonConfig.h @@ -72,7 +72,7 @@ class JsonConfig : public nlohmann::json { const nlohmann::json& get_json() const { return *this; } /** - * Get a nested value by dotted path (e.g. "brokers.attention.endpoint"). + * Get a nested value by dotted path (e.g. "agents.attention.endpoint"). * @return JsonPathValue at that path; use .get() or .get_or(default) on the result. * Example: config.at_path("atomdb.type").get_or("NOT_FOUND"); */ diff --git a/src/commons/JsonConfigParser.cc b/src/commons/JsonConfigParser.cc index e5ea218a..4c8e0e5a 100644 --- a/src/commons/JsonConfigParser.cc +++ b/src/commons/JsonConfigParser.cc @@ -17,18 +17,9 @@ namespace commons { namespace { -// bus_node: required fields. -const unordered_map>& required_node_fields_by_version() { +const unordered_map>& required_fields_by_version() { static const unordered_map> m = { - {"1.0", {"atomdb", "atomdb.type", "loaders", "agents", "brokers"}}, - }; - return m; -} - -// bus_client: required fields. -const unordered_map>& required_client_fields_by_version() { - static const unordered_map> m = { - {"1.0", {"params", "params.das-config-file", "params.endpoint", "params.ports-range"}}, + {"1.0", {"atomdb", "atomdb.type", "loaders", "agents"}}, }; return m; } @@ -57,7 +48,7 @@ void validate_schema_version(const JsonConfig& config, vector& required) } } -JsonConfig parse_and_validate(const string& json_str, bool is_node) { +JsonConfig parse_and_validate(const string& json_str) { JsonConfig config; try { config = JsonConfig(json::parse(json_str)); @@ -66,8 +57,7 @@ JsonConfig parse_and_validate(const string& json_str, bool is_node) { } string version = config.get_schema_version(); try { - vector required = is_node ? required_node_fields_by_version().at(version) - : required_client_fields_by_version().at(version); + vector required = required_fields_by_version().at(version); validate_schema_version(config, required); } catch (const exception& e) { RAISE_ERROR("Invalid schema version: " + version + " " + string(e.what())); @@ -85,26 +75,11 @@ JsonConfig JsonConfigParser::load(const string& file_path, bool throw_flag) { } stringstream buf; buf << f.rdbuf(); - return parse_and_validate(buf.str(), true); + return parse_and_validate(buf.str()); } JsonConfig JsonConfigParser::load_from_string(const string& json_content) { - return parse_and_validate(json_content, true); -} - -JsonConfig JsonConfigParser::load_client_config(const string& file_path, bool throw_flag) { - ifstream f(file_path); - if (!f.good()) { - Utils::error("JsonConfigParser: Cannot open client config file: " + file_path, throw_flag); - return JsonConfig(); - } - stringstream buf; - buf << f.rdbuf(); - return parse_and_validate(buf.str(), false); -} - -JsonConfig JsonConfigParser::load_client_config_from_string(const string& json_content) { - return parse_and_validate(json_content, false); + return parse_and_validate(json_content); } } // namespace commons diff --git a/src/commons/JsonConfigParser.h b/src/commons/JsonConfigParser.h index 858bceb2..356a4294 100644 --- a/src/commons/JsonConfigParser.h +++ b/src/commons/JsonConfigParser.h @@ -16,7 +16,7 @@ class JsonConfigParser { public: /** * Load and validate config from a JSON file. - * @param file_path Path to the JSON file (e.g. ~/.das/config.json). + * @param file_path Path to the JSON file (e.g. config/das.json). * @return JsonConfig instance with validated schema. * @param throw_flag Whether to throw an exception on file not found. */ @@ -29,16 +29,6 @@ class JsonConfigParser { * @throws std::runtime_error on invalid JSON or schema validation failure. */ static JsonConfig load_from_string(const string& json_content); - - /** - * Load and validate a bus_client config (schema 1.0): requires params and params.das_config_file - * (path to `das.json`, same schema as bus `busnode` config). Atomdb may be supplied inline or merged - * from that file in bus_client. - */ - static JsonConfig load_client_config(const string& file_path, bool throw_flag = true); - - /** Same as load_client_config for inline JSON (tests). */ - static JsonConfig load_client_config_from_string(const string& json_content); }; } // namespace commons diff --git a/src/commons/Properties.h b/src/commons/Properties.h index 33a13a0c..92d3ce06 100644 --- a/src/commons/Properties.h +++ b/src/commons/Properties.h @@ -246,4 +246,20 @@ class Properties : public unordered_map { } }; +/** @return Copy of @p lhs with keys from @p rhs merged in; @p rhs wins on duplicates. */ +inline Properties operator+(Properties lhs, const Properties& rhs) { + for (const auto& entry : rhs) { + lhs[entry.first] = entry.second; + } + return lhs; +} + +/** Merges @p rhs into @p lhs in place; @p rhs wins on duplicate keys. */ +inline Properties& operator+=(Properties& lhs, const Properties& rhs) { + for (const auto& entry : rhs) { + lhs[entry.first] = entry.second; + } + return lhs; +} + } // namespace commons diff --git a/src/commons/SystemParameters.cc b/src/commons/SystemParameters.cc new file mode 100644 index 00000000..29b93eac --- /dev/null +++ b/src/commons/SystemParameters.cc @@ -0,0 +1,99 @@ +#include "SystemParameters.h" + +#include "Utils.h" + +using namespace commons; +using nlohmann::json; +using namespace std; + +namespace { + +PropertyValue json_scalar_to_property(const json& value, const string& key) { + if (value.is_string()) return value.get(); + if (value.is_number_unsigned()) return value.get(); + if (value.is_number_integer()) return static_cast(value.get()); + if (value.is_number_float()) return value.get(); + if (value.is_boolean()) return value.get(); + if (value.is_null()) return string(""); + RAISE_ERROR("Invalid non-scalar parameter value for key '" + key + "'"); + return string(""); +} + +void load_agent_params(const json& agent_params, + const string& agent, + map& params_by_agent) { + if (!agent_params.is_object()) { + RAISE_ERROR("Parameters for agent '" + agent + "' must be an object"); + } + Properties agent_props; + for (auto pit = agent_params.begin(); pit != agent_params.end(); ++pit) { + const string key = pit.key(); + agent_props[key] = json_scalar_to_property(pit.value(), agent + ".params." + key); + } + params_by_agent[agent] = agent_props; +} + +void load_from_agents(const json& agents, map& params_by_agent) { + if (!agents.is_object()) { + RAISE_ERROR("'agents' must be a JSON object"); + } + for (auto it = agents.begin(); it != agents.end(); ++it) { + const string agent = it.key(); + const json& agent_json = it.value(); + if (!agent_json.is_object()) { + continue; + } + auto agent_params_it = agent_json.find("params"); + if (agent_params_it == agent_json.end() || agent_params_it->is_null()) { + continue; + } + load_agent_params(*agent_params_it, agent + ".params", params_by_agent); + } +} + +} // namespace + +SystemParameters::SystemParameters(const json& root) { + if (!root.is_object() || !root.contains("agents")) { + RAISE_ERROR("Missing 'agents' section in parameters JSON"); + } + load_from_agents(root["agents"], this->params_by_agent); +} + +SystemParameters::SystemParameters(const JsonConfig& das_config) { + auto agents = das_config.at_path("agents"); + if (agents.is_null()) { + RAISE_ERROR("Missing 'agents' section in system config"); + } + load_from_agents(*agents, this->params_by_agent); +} + +Properties SystemParameters::get_agent_params(const string& agent) const { + auto agent_it = this->params_by_agent.find(agent + ".params"); + if (agent_it == this->params_by_agent.end()) { + RAISE_ERROR("Unknown agent: '" + agent + "'"); + } + return agent_it->second; +} + +Properties SystemParameters::get_base_query_params() const { return get_agent_params("base_query"); } + +Properties SystemParameters::get_query_agent_params() const { + return get_base_query_params() + get_agent_params("query"); +} + +Properties SystemParameters::get_link_creation_agent_params() const { + return get_agent_params("link_creation"); +} + +Properties SystemParameters::get_inference_agent_params() const { + return get_base_query_params() + get_agent_params("inference"); +} + +Properties SystemParameters::get_evolution_agent_params() const { + return get_base_query_params() + get_agent_params("evolution"); +} + +Properties SystemParameters::get_context_agent_params() const { + return get_base_query_params() + get_agent_params("context"); +} diff --git a/src/commons/SystemParameters.h b/src/commons/SystemParameters.h new file mode 100644 index 00000000..d9c83e6e --- /dev/null +++ b/src/commons/SystemParameters.h @@ -0,0 +1,33 @@ +#pragma once + +#include +#include +#include +#include + +#include "JsonConfig.h" +#include "Properties.h" + +using namespace std; + +namespace commons { + +class SystemParameters { + public: + SystemParameters(const nlohmann::json& params); + SystemParameters(const JsonConfig& das_config); + + // Agents parameters getters + Properties get_base_query_params() const; + Properties get_query_agent_params() const; + Properties get_link_creation_agent_params() const; + Properties get_inference_agent_params() const; + Properties get_evolution_agent_params() const; + Properties get_context_agent_params() const; + + private: + Properties get_agent_params(const string& agent) const; + map params_by_agent; +}; + +} // namespace commons diff --git a/src/commons/SystemParametersSingleton.cc b/src/commons/SystemParametersSingleton.cc new file mode 100644 index 00000000..dd1ee446 --- /dev/null +++ b/src/commons/SystemParametersSingleton.cc @@ -0,0 +1,40 @@ +#include "SystemParametersSingleton.h" + +#include "JsonConfigParser.h" + +using namespace commons; +using namespace std; + +bool SystemParametersSingleton::INITIALIZED = false; +shared_ptr SystemParametersSingleton::PARAMS = shared_ptr(nullptr); +mutex SystemParametersSingleton::API_MUTEX; + +// ------------------------------------------------------------------------------------------------- +// Public methods + +void SystemParametersSingleton::init(const JsonConfig& das_config) { + lock_guard semaphore(API_MUTEX); + PARAMS = make_shared(SystemParameters(das_config)); + INITIALIZED = true; +} + +void SystemParametersSingleton::init_from_file(const string& path) { + init(JsonConfigParser::load(path)); +} + +shared_ptr SystemParametersSingleton::get_instance() { + lock_guard semaphore(API_MUTEX); + if (!INITIALIZED || PARAMS == nullptr) { + RAISE_ERROR("SystemParametersSingleton not initialized"); + } + return PARAMS; +} + +void SystemParametersSingleton::provide(shared_ptr parameters) { + if (parameters == nullptr) { + RAISE_ERROR("SystemParametersSingleton::provide(): parameters cannot be nullptr"); + } + lock_guard semaphore(API_MUTEX); + PARAMS = parameters; + INITIALIZED = true; +} diff --git a/src/commons/SystemParametersSingleton.h b/src/commons/SystemParametersSingleton.h new file mode 100644 index 00000000..a5411699 --- /dev/null +++ b/src/commons/SystemParametersSingleton.h @@ -0,0 +1,27 @@ +#pragma once + +#include +#include + +#include "JsonConfig.h" +#include "SystemParameters.h" + +using namespace std; + +namespace commons { + +class SystemParametersSingleton { + public: + static void init(const JsonConfig& das_config); + static void init_from_file(const string& path); + static shared_ptr get_instance(); + static void provide(shared_ptr parameters); + + private: + SystemParametersSingleton() = default; + static bool INITIALIZED; + static shared_ptr PARAMS; + static mutex API_MUTEX; +}; + +} // namespace commons diff --git a/src/main/README.md b/src/main/README.md index 73b3031f..277a0f6d 100644 --- a/src/main/README.md +++ b/src/main/README.md @@ -67,12 +67,12 @@ make run-busnode OPTIONS="--service=atomdb-broker --endpoint=localhost:9004 --po Run the `bus_client` binary with the required parameters using make: ```bash -make run-client OPTIONS="--client= [--endpoint=...] --bus-endpoint= --ports-range= [--config=]" +make run-client OPTIONS="--client= [--config=config/das.json] [--endpoint=...] [--bus-endpoint=...] [--ports-range=...]" ``` **`--client`** is required and must be a valid bus node service. -Pass **`--config=`** (schema 1.0) so **`atomdb`** can be merged from **`params.das-config-file`** (path to **`das.json`**). Without it, **`bus_client`** cannot initialize AtomDB with the slim client schema. With config, **`--bus-endpoint`**, **`--ports-range`** and **`--endpoint`** may default from **`params`** field and **`das.json`**; optional **`--attention-broker-endpoint`** from merged **`brokers.attention.endpoint`**. Command-line arguments always win. +**`bus_client`** loads **`config/das.json`** by default (same file as **`busnode`**). **`client.endpoint`** and **`client.ports_range`** default **`--endpoint`** and **`--ports-range`**; **`agents`** supply **`--bus-endpoint`** from **`--client`**; optional **`--attention-broker-endpoint`** from **`agents.attention.endpoint`**; scalar **`params.*`** merge into CLI args. Command-line arguments always win. ### Examples #### AtomDB: @@ -81,11 +81,11 @@ make run-client OPTIONS="--client=atomdb-broker --endpoint=localhost:8887 --bus- ``` #### Query Engine: ``` -make run-client OPTIONS="--config=config/client.json --client=query-engine --query=LINK_TEMPLATE Expression 2 NODE Symbol Predicate VARIABLE V1 --context=test" +make run-client OPTIONS="--client=query-engine --query=LINK_TEMPLATE Expression 2 NODE Symbol Predicate VARIABLE V1 --context=test" ``` -(With **`config/client.json`**, **`--endpoint`**, **`--bus-endpoint`**, and **`--ports-range`** can be omitted when **`das.json`** and **`params`** supply them.) +(With **`config/das.json`**, **`--endpoint`**, **`--bus-endpoint`**, and **`--ports-range`** can be omitted.) -**If you see `Bus: no owner is defined for command `:** the resolved **`--endpoint`** (from **`agents..endpoint`** / **`brokers..endpoint`** in **`das.json`**) must match the **`--endpoint`** used when starting that agent’s **`busnode`** (same `host:port` string). +**If you see `Bus: no owner is defined for command `:** the resolved **`--endpoint`** (from **`agents..endpoint`** in **`das.json`**) must match the **`--endpoint`** used when starting that agent’s **`busnode`** (same `host:port` string). #### LCA: ``` diff --git a/src/main/bus_client.cc b/src/main/bus_client.cc index 94ffcb87..6809f37e 100644 --- a/src/main/bus_client.cc +++ b/src/main/bus_client.cc @@ -9,12 +9,19 @@ #include "ProxyFactory.h" #include "RemoteAtomDB.h" #include "ServiceBusSingleton.h" +#include "SystemParametersSingleton.h" #include "Utils.h" using namespace commons; using namespace mains; using namespace std; +namespace { + +const char* kDefaultConfigPath = "config/das.json"; + +} // namespace + int main(int argc, char* argv[]) { try { auto required_cmd_args = { @@ -31,39 +38,41 @@ int main(int argc, char* argv[]) { RAISE_ERROR("Required argument missing: " + Helper::CLIENT); } - ///////// Optional JsonConfig + string config_path = kDefaultConfigPath; auto it_config = cmd_args.find("config"); - JsonConfig json_config; - const bool has_client_config = it_config != cmd_args.end() && !it_config->second.empty(); - if (has_client_config) { - json_config = JsonConfigParser::load_client_config(it_config->second); - auto das_config_file = json_config.at_path("params.das-config-file"); - if (!das_config_file.is_null()) { - JsonConfig das_config = JsonConfigParser::load(das_config_file.get()); - // Merge atomdb config from das.json - json_config["atomdb"] = das_config["atomdb"]; - - // Get bus node endpoint from das.json - auto it_known_peer = Helper::arg_to_json_config_key.find(cmd_args[Helper::CLIENT]); - if (it_known_peer != Helper::arg_to_json_config_key.end()) { - cmd_args[Helper::BUS_ENDPOINT] = - das_config.at_path(it_known_peer->second + ".endpoint").get(); - } else { - RAISE_ERROR("Required argument missing: " + cmd_args[Helper::CLIENT] + - " is not a bus node."); - } + if (it_config != cmd_args.end() && !it_config->second.empty()) { + config_path = it_config->second; + } - } else { - RAISE_ERROR("params.das-config-file is missing"); - } + JsonConfig json_config = JsonConfigParser::load(config_path); + SystemParametersSingleton::init(json_config); - cmd_args[Helper::ENDPOINT] = json_config.at_path("params.endpoint").get(); - cmd_args[Helper::PORTS_RANGE] = json_config.at_path("params.ports-range").get(); + if (cmd_args.find(Helper::ENDPOINT) == cmd_args.end()) { + cmd_args[Helper::ENDPOINT] = "localhost:40000"; + } + if (cmd_args.find(Helper::PORTS_RANGE) == cmd_args.end()) { + cmd_args[Helper::PORTS_RANGE] = "52000:52999"; + } - // Merge all params from client json config to cmd_args, existing cmd_args have precedence - Helper::merge_params_from_client_json_config(cmd_args, json_config); + auto it_known_peer = Helper::arg_to_json_config_key.find(cmd_args[Helper::CLIENT]); + if (it_known_peer != Helper::arg_to_json_config_key.end()) { + if (cmd_args.find(Helper::BUS_ENDPOINT) == cmd_args.end()) { + cmd_args[Helper::BUS_ENDPOINT] = + json_config.at_path(it_known_peer->second + ".endpoint").get(); + } + } else { + RAISE_ERROR("Unknown client: " + cmd_args[Helper::CLIENT]); } + if (cmd_args.find(Helper::ATTENTION_BROKER_ENDPOINT) == cmd_args.end()) { + auto attention = json_config.at_path("agents.attention.endpoint"); + if (!attention.is_null()) { + cmd_args[Helper::ATTENTION_BROKER_ENDPOINT] = attention.get(); + } + } + + Helper::merge_params_from_config(cmd_args, json_config); + for (auto req_arg : required_cmd_args) { if (cmd_args.find(req_arg) == cmd_args.end()) { RAISE_ERROR("Required argument missing: " + string(req_arg)); @@ -138,7 +147,6 @@ int main(int argc, char* argv[]) { cmd_args.find(Helper::USE_METTA_AS_QUERY_TOKENS) != cmd_args.end() && cmd_args[Helper::USE_METTA_AS_QUERY_TOKENS] == "true"; - // Default case for other clients while (proxy->finished() == false && Helper::is_running) { if (dynamic_cast(proxy.get()) != nullptr) { auto query_proxy = dynamic_cast(proxy.get()); diff --git a/src/main/bus_node.cc b/src/main/bus_node.cc index 12314544..41c02f4d 100644 --- a/src/main/bus_node.cc +++ b/src/main/bus_node.cc @@ -8,6 +8,7 @@ #include "ProcessorFactory.h" #include "Properties.h" #include "RemoteAtomDB.h" +#include "SystemParametersSingleton.h" #include "Utils.h" using namespace commons; @@ -31,6 +32,7 @@ int main(int argc, char* argv[]) { JsonConfig json_config; if (it_config != cmd_args.end() && !it_config->second.empty()) { json_config = JsonConfigParser::load(it_config->second); + SystemParametersSingleton::init(json_config); } // Map service name (e.g. "query-engine") to config section path (e.g. "agents.query") string service_name = cmd_args[Helper::SERVICE]; diff --git a/src/main/helpers/Helper.cc b/src/main/helpers/Helper.cc index e47a4d2a..8133e6b4 100644 --- a/src/main/helpers/Helper.cc +++ b/src/main/helpers/Helper.cc @@ -57,34 +57,47 @@ map Helper::arg_to_json_config_key = { {"evolution-agent", "agents.evolution"}, {"link-creation-agent", "agents.link_creation"}, {"inference-agent", "agents.inference"}, - {"atomdb-broker", "brokers.atomdb"}, - {"context-broker", "brokers.context"}, - {"attention-broker", "brokers.attention"}, - {Helper::ATTENTION_BROKER_ENDPOINT, "brokers.attention.endpoint"}, + {"atomdb-broker", "agents.atomdb"}, + {"context-broker", "agents.context"}, + {"attention-broker", "agents.attention"}, + {Helper::ATTENTION_BROKER_ENDPOINT, "agents.attention.endpoint"}, }; +string param_key_to_cmd_arg(const string& key) { + string cmd_key = key; + replace(cmd_key.begin(), cmd_key.end(), '_', '-'); + return cmd_key; +} + void merge_params(map& cmd_args, JsonConfig& params) { for (const auto& entry : params.items()) { const nlohmann::json& v = entry.value(); if (v.is_object() || v.is_array()) { continue; } - // If the param is not already set, set it - if (cmd_args.find(entry.key()) == cmd_args.end()) { - cmd_args[entry.key()] = v.dump(); + string cmd_key = param_key_to_cmd_arg(entry.key()); + if (cmd_args.find(cmd_key) == cmd_args.end()) { + if (v.is_string()) { + cmd_args[cmd_key] = v.get(); + } else if (v.is_boolean()) { + cmd_args[cmd_key] = v.get() ? "true" : "false"; + } else { + cmd_args[cmd_key] = v.dump(); + } } } } -void Helper::merge_params_from_client_json_config(map& cmd_args, - JsonConfig& json_config) { - auto query_params = json_config.at_path("params.query").get(); +void Helper::merge_params_from_config(map& cmd_args, JsonConfig& json_config) { + auto base_query_params = json_config.at_path("agents.base_query.params").get(); + merge_params(cmd_args, base_query_params); + auto query_params = json_config.at_path("agents.query.params").get(); merge_params(cmd_args, query_params); - auto link_creation_params = json_config.at_path("params.link_creation").get(); + auto link_creation_params = json_config.at_path("agents.link_creation.params").get(); merge_params(cmd_args, link_creation_params); - auto evolution_params = json_config.at_path("params.evolution").get(); + auto evolution_params = json_config.at_path("agents.evolution.params").get(); merge_params(cmd_args, evolution_params); - auto context_params = json_config.at_path("params.context").get(); + auto context_params = json_config.at_path("agents.context.params").get(); merge_params(cmd_args, context_params); } @@ -215,8 +228,8 @@ This client interacts with the AtomDB Broker via the service bus. )")}, {ProcessorType::UNKNOWN, string(R"( Usage: -busclient --client= --endpoint= --bus-endpoint= --ports-range= [--config=] [--use-mork=true|false] -With --config=client.json (schema 1.0), params.das_config_file points at das.json (DAS bus node config); --endpoint may be omitted when --client is set (reads agents..endpoint or brokers..endpoint); optional attention-broker-endpoint from merged brokers.attention.endpoint; bus-endpoint and ports-range from params. Internal bus proxy is chosen from --client (query -> query-engine, etc.). +busclient --client= [--config=] [--endpoint=] [--bus-endpoint=] [--ports-range=] [--use-mork=true|false] +Defaults load config/das.json (schema 1.0): client.endpoint and client.ports_range for this client; agents..endpoint for --bus-endpoint when --client is set; optional attention-broker-endpoint from agents.attention.endpoint; scalar params.* merged into CLI args. Command-line arguments always win. )")}}; static map string_to_processor_type = { diff --git a/src/main/helpers/Helper.h b/src/main/helpers/Helper.h index 03c561b0..82cd8831 100644 --- a/src/main/helpers/Helper.h +++ b/src/main/helpers/Helper.h @@ -72,9 +72,8 @@ class Helper { /** Maps CLI arg names to dotted JSON config paths (e.g. "query-engine" -> "agents.query"). */ static map arg_to_json_config_key; - /** Merges params from client json config to cmd_args. */ - static void merge_params_from_client_json_config(map& cmd_args, - JsonConfig& json_config); + /** Merges scalar entries from params.* sections into cmd_args (CLI wins). */ + static void merge_params_from_config(map& cmd_args, JsonConfig& json_config); static bool is_running; diff --git a/src/tests/cpp/BUILD b/src/tests/cpp/BUILD index 360df702..a77aa625 100644 --- a/src/tests/cpp/BUILD +++ b/src/tests/cpp/BUILD @@ -494,6 +494,7 @@ cc_test( "//atomdb:atomdb_singleton", "//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_main", ], ) @@ -524,6 +525,7 @@ cc_test( "//atomdb", "//attention_broker:attention_broker_lib", "//tests/cpp/test_commons:test_atomdb_json_config", + "//tests/cpp/test_commons:test_system_params", "@com_github_google_googletest//:gtest_main", ], ) @@ -577,6 +579,7 @@ cc_test( "//agents/evolution:evolution_lib", "//agents/query_engine:pattern_matching_query_processor", "//tests/cpp/test_commons:test_atomdb_json_config", + "//tests/cpp/test_commons:test_system_params", "@com_github_google_googletest//:gtest_main", ], ) @@ -594,6 +597,7 @@ cc_test( "//agents/link_creation_agent:link_creation_agent_lib", "//commons:commons_lib", "//service_bus:service_bus_lib", + "//tests/cpp/test_commons:test_system_params", "//tests/cpp/test_commons/mocks:mock_atom_db_lib", "//tests/cpp/test_commons/mocks:mock_service_bus_lib", "@com_github_google_googletest//:gtest_main", @@ -686,6 +690,22 @@ cc_test( ], ) +cc_test( + name = "system_parameters_test", + size = "small", + srcs = ["system_parameters_test.cc"], + copts = [ + "-Iexternal/gtest/googletest/include", + "-Iexternal/gtest/googletest", + ], + linkstatic = 1, + deps = [ + "//commons:commons_lib", + "//tests/cpp/test_commons:test_system_params", + "@com_github_google_googletest//:gtest_main", + ], +) + cc_test( name = "processor_test", size = "small", diff --git a/src/tests/cpp/config_parser_test.cc b/src/tests/cpp/config_parser_test.cc index a35adb28..e4e2433a 100644 --- a/src/tests/cpp/config_parser_test.cc +++ b/src/tests/cpp/config_parser_test.cc @@ -35,23 +35,13 @@ const char* kValidConfigV1 = R"({ "metta": { "image": "trueagi/das:metta-parser" } }, "agents": { - "query": { "endpoint": "localhost:40002" } + "query": { "endpoint": "localhost:40002", "params": { "max-answers": 100 } } }, "brokers": { "attention": { "endpoint": "localhost:40001" } } })"; -const char* kValidClientConfigV1 = R"({ - "schema_version": "1.0", - "params": { - "endpoint": "localhost:9000", - "ports-range": "41000:41999", - "das-config-file": "config/das.json", - "query": { "max-answers": 100 } - } -})"; - } // namespace TEST(ConfigParserTest, LoadFromStringValidSchemaV1) { @@ -102,9 +92,9 @@ TEST(ConfigParserTest, GetNestedAgentsAndParams) { EXPECT_EQ(query, "localhost:40002"); } -TEST(ConfigParserTest, GetParamsFromClientConfig) { - JsonConfig config = JsonConfigParser::load_client_config_from_string(kValidClientConfigV1); - long max_answers = config.at_path("params.query.max-answers").get_or(0); +TEST(ConfigParserTest, GetParamsFromDasConfig) { + JsonConfig config = JsonConfigParser::load_from_string(kValidConfigV1); + long max_answers = config.at_path("agents.query.params.max-answers").get_or(0); EXPECT_EQ(max_answers, 100); } @@ -115,14 +105,14 @@ TEST(ConfigParserTest, GetNestedMissingKeyReturnsEmptyString) { } TEST(ConfigParserTest, MissingSchemaVersionThrows) { - const char* no_version = R"({ "atomdb": {}, "loaders": {}, "agents": {}, "brokers": {} })"; + const char* no_version = R"({ "atomdb": {}, "loaders": {}, "agents": {} })"; EXPECT_THROW(JsonConfigParser::load_from_string(no_version), runtime_error); } TEST(ConfigParserTest, UnsupportedSchemaVersionThrows) { const char* bad_version = R"({ "schema_version": "99.0", - "atomdb": {}, "loaders": {}, "agents": {}, "brokers": {} + "atomdb": {}, "loaders": {}, "agents": {} })"; EXPECT_THROW(JsonConfigParser::load_from_string(bad_version), runtime_error); } @@ -130,7 +120,7 @@ TEST(ConfigParserTest, UnsupportedSchemaVersionThrows) { TEST(ConfigParserTest, MissingRequiredFieldThrows) { const char* no_atomdb = R"({ "schema_version": "1.0", - "loaders": {}, "agents": {}, "brokers": {} + "loaders": {}, "agents": {} })"; EXPECT_THROW(JsonConfigParser::load_from_string(no_atomdb), runtime_error); } @@ -139,27 +129,11 @@ TEST(ConfigParserTest, NullRequiredFieldThrows) { const char* null_atomdb = R"({ "schema_version": "1.0", "atomdb": null, - "loaders": {}, "agents": {}, "brokers": {} + "loaders": {}, "agents": {} })"; EXPECT_THROW(JsonConfigParser::load_from_string(null_atomdb), runtime_error); } -TEST(ConfigParserTest, ClientProfileMissingParamsThrows) { - const char* no_params = R"({ - "schema_version": "1.0", - "atomdb": { "type": "redismongodb", "redis": {}, "mongodb": {}, "morkdb": {} } - })"; - EXPECT_THROW(JsonConfigParser::load_client_config_from_string(no_params), runtime_error); -} - -TEST(ConfigParserTest, ClientProfileMissingDasConfigFileThrows) { - const char* no_das = R"({ - "schema_version": "1.0", - "params": { "query": {} } - })"; - EXPECT_THROW(JsonConfigParser::load_client_config_from_string(no_das), runtime_error); -} - TEST(ConfigParserTest, InvalidJsonThrows) { EXPECT_THROW(JsonConfigParser::load_from_string("{ invalid json }"), runtime_error); } diff --git a/src/tests/cpp/context_test.cc b/src/tests/cpp/context_test.cc index 7528dcdb..a62ad9e6 100644 --- a/src/tests/cpp/context_test.cc +++ b/src/tests/cpp/context_test.cc @@ -10,6 +10,7 @@ #include "ServiceBus.h" #include "ServiceBusSingleton.h" #include "TestAtomDBJsonConfig.h" +#include "TestSystemParams.h" #include "UntypedVariable.h" #include "Utils.h" #include "gtest/gtest.h" @@ -18,6 +19,8 @@ #define LOG_LEVEL INFO_LEVEL #include "Logger.h" +using das_test::init_test_system_parameters_singleton; + using namespace atomdb; using namespace attention_broker; using namespace context_broker; @@ -25,7 +28,10 @@ using namespace query_engine; class ContextTestEnvironment : public ::testing::Environment { public: - void SetUp() override { AtomDBSingleton::init(test_atomdb_json_config()); } + void SetUp() override { + AtomDBSingleton::init(test_atomdb_json_config()); + init_test_system_parameters_singleton(); + } void TearDown() override {} }; diff --git a/src/tests/cpp/link_creation_agent_test.cc b/src/tests/cpp/link_creation_agent_test.cc index 24c8f7ef..55413e82 100644 --- a/src/tests/cpp/link_creation_agent_test.cc +++ b/src/tests/cpp/link_creation_agent_test.cc @@ -12,8 +12,11 @@ #include "MockAtomDB.h" #include "MockServiceBus.h" #include "TemplateProcessor.h" +#include "TestSystemParams.h" #include "Utils.h" +using das_test::init_test_system_parameters_singleton; + using namespace std; using namespace link_creation_agent; using namespace commons; @@ -49,6 +52,7 @@ class LinkCreationAgentTest : public ::testing::Test { this->save_links_to_db = false; this->server_id = "localhost:40040"; AtomDBSingleton::provide(move(make_shared())); + init_test_system_parameters_singleton(); ServiceBusSingleton::provide( move(make_shared("localhost:40038", "localhost:40039"))); } diff --git a/src/tests/cpp/pattern_matching_query_test.cc b/src/tests/cpp/pattern_matching_query_test.cc index af86ca68..353c3130 100644 --- a/src/tests/cpp/pattern_matching_query_test.cc +++ b/src/tests/cpp/pattern_matching_query_test.cc @@ -6,6 +6,7 @@ #include "PatternMatchingQueryProxy.h" #include "ServiceBus.h" #include "TestAtomDBJsonConfig.h" +#include "TestSystemParams.h" #include "Utils.h" #include "gtest/gtest.h" @@ -15,6 +16,7 @@ using namespace query_engine; using namespace query_element; using namespace atomdb; +using das_test::init_test_system_parameters_singleton; string handle_to_atom(const string& handle) { shared_ptr db = AtomDBSingleton::get_instance(); @@ -250,6 +252,7 @@ void check_query_chain(const string& query_tag, TEST(PatternMatchingQuery, queries) { AtomDBSingleton::init(test_atomdb_json_config()); + init_test_system_parameters_singleton(); ServiceBus::initialize_statics({}, 40200, 40299); string peer1_id = "localhost:40041"; @@ -581,7 +584,7 @@ TEST(PatternMatchingQuery, queries) { EXPECT_EQ(answer->metta_expression[answer->assignment.get("v1")], "\"ent\""); } } - EXPECT_EQ(count, 1); + EXPECT_EQ(count, 1U); // clang-format on } diff --git a/src/tests/cpp/query_evolution_test.cc b/src/tests/cpp/query_evolution_test.cc index 39fcd474..4cb313df 100644 --- a/src/tests/cpp/query_evolution_test.cc +++ b/src/tests/cpp/query_evolution_test.cc @@ -8,9 +8,12 @@ #include "ServiceBus.h" #include "ServiceBusSingleton.h" #include "TestAtomDBJsonConfig.h" +#include "TestSystemParams.h" #include "Utils.h" #include "gtest/gtest.h" +using das_test::init_test_system_parameters_singleton; + using namespace atomdb; using namespace evolution; @@ -32,6 +35,7 @@ class TestFitnessFunction : public FitnessFunction { TEST(QueryEvolution, protected_methods) { AtomDBSingleton::init(test_atomdb_json_config()); + init_test_system_parameters_singleton(); string peer1_id = "localhost:40043"; string peer2_id = "localhost:40044"; diff --git a/src/tests/cpp/system_parameters_test.cc b/src/tests/cpp/system_parameters_test.cc new file mode 100644 index 00000000..7d3d3d43 --- /dev/null +++ b/src/tests/cpp/system_parameters_test.cc @@ -0,0 +1,79 @@ +#include + +#include "SystemParameters.h" +#include "SystemParametersSingleton.h" +#include "TestSystemParams.h" + +using namespace commons; +using namespace das_test; +using namespace std; + +TEST(SystemParametersTest, get_base_query_params) { + auto params = make_test_parameters().get_base_query_params(); + EXPECT_EQ(params.get("max_answers"), 0U); + EXPECT_EQ(params.get("max_bundle_size"), 1000U); + EXPECT_FALSE(params.get("unique_assignment_flag")); +} + +TEST(SystemParametersTest, get_query_agent_params_merges_base_and_query) { + auto params = make_test_parameters().get_query_agent_params(); + EXPECT_FALSE(params.get("count_flag")); + EXPECT_FALSE(params.get("unique_assignment_flag")); + EXPECT_FALSE(params.get("positive_importance_flag")); + EXPECT_EQ(params.get("max_bundle_size"), 1000U); + // Must be merged with base query params + EXPECT_EQ(params.get("max_answers"), 0U); +} + +TEST(SystemParametersTest, get_link_creation_agent_params_standalone) { + auto params = make_test_parameters().get_link_creation_agent_params(); + EXPECT_EQ(params.get("max_answers"), 10U); + EXPECT_TRUE(params.get("positive_importance_flag")); + EXPECT_EQ(params.find("unique_assignment_flag"), params.end()); +} + +TEST(SystemParametersTest, get_inference_agent_params_overrides_base_max_answers) { + auto params = make_test_parameters().get_inference_agent_params(); + EXPECT_EQ(params.get("max_answers"), 150U); + EXPECT_EQ(params.get("repeat_count"), 5U); + EXPECT_EQ(params.get("inference_request_timeout"), 86400U); +} + +TEST(SystemParametersTest, get_evolution_agent_params_merges_base) { + auto params = make_test_parameters().get_evolution_agent_params(); + EXPECT_EQ(params.get("population_size"), 1000U); + EXPECT_EQ(params.get("max_generations"), 100U); + EXPECT_FALSE(params.get("unique_assignment_flag")); +} + +TEST(SystemParametersTest, get_context_agent_params_merges_base) { + auto params = make_test_parameters().get_context_agent_params(); + EXPECT_TRUE(params.get("use_cache")); + EXPECT_DOUBLE_EQ(params.get("initial_rent_rate"), 0.75); + EXPECT_EQ(params.get("context"), "context"); +} + +TEST(SystemParametersTest, singleton_matches_direct_instance) { + init_test_system_parameters_singleton(); + auto from_singleton = SystemParametersSingleton::get_instance()->get_query_agent_params(); + auto direct = make_test_parameters().get_query_agent_params(); + EXPECT_EQ(from_singleton.get("max_answers"), direct.get("max_answers")); + EXPECT_EQ(from_singleton.get("count_flag"), direct.get("count_flag")); +} + +TEST(SystemParametersTest, properties_merge_prioritizes_right_hand_side) { + Properties base; + base["max_answers"] = 0U; + base["count_flag"] = false; + + Properties agent; + agent["max_answers"] = 100U; + agent["count_flag"] = true; + agent["unique_value_flag"] = true; + + Properties merged = base + agent; + EXPECT_EQ(merged.get("max_answers"), 100U); + EXPECT_TRUE(merged.get("count_flag")); + EXPECT_TRUE(merged.get("unique_value_flag")); + EXPECT_EQ(merged.size(), 3U); +} diff --git a/src/tests/cpp/test_commons/BUILD b/src/tests/cpp/test_commons/BUILD index 8e38bd8f..78838098 100644 --- a/src/tests/cpp/test_commons/BUILD +++ b/src/tests/cpp/test_commons/BUILD @@ -2,6 +2,16 @@ load("@rules_cc//cc:cc_library.bzl", "cc_library") package(default_visibility = ["//visibility:public"]) +cc_library( + name = "test_atomdb_json_config", + hdrs = ["TestAtomDBJsonConfig.h"], + includes = ["."], + deps = [ + "//commons:commons_lib", + "@nlohmann_json//:json", + ], +) + cc_library( name = "mock_animals_data_lib", srcs = ["MockAnimalsData.cc"], @@ -10,14 +20,19 @@ cc_library( deps = [ "//atomdb:atomdb_singleton", "//commons:commons_lib", + "//commons/atoms:atoms_lib", + "//tests/cpp/test_commons:test_atomdb_json_config", ], ) cc_library( - name = "test_atomdb_json_config", - hdrs = ["TestAtomDBJsonConfig.h"], + name = "test_system_params", + srcs = ["TestSystemParams.cc"], + hdrs = ["TestSystemParams.h"], includes = ["."], deps = [ "//commons:commons_lib", + "@com_github_google_googletest//:gtest", + "@nlohmann_json//:json", ], ) diff --git a/src/tests/cpp/test_commons/TestSystemParams.cc b/src/tests/cpp/test_commons/TestSystemParams.cc new file mode 100644 index 00000000..0a7e6885 --- /dev/null +++ b/src/tests/cpp/test_commons/TestSystemParams.cc @@ -0,0 +1,82 @@ +#include "TestSystemParams.h" + +#include + +using namespace commons; +using nlohmann::json; + +namespace das_test { + +const char kAgentsJson[] = R"({ + "agents": { + "base_query": { + "params": { + "unique_assignment_flag": false, + "attention_update": 0, + "attention_correlation": 0, + "max_bundle_size": 1000, + "max_answers": 0, + "use_link_template_cache": false, + "populate_metta_mapping": false, + "use_metta_as_query_tokens": false, + "allow_incomplete_chain_path": false + } + }, + "query": { + "endpoint": "localhost:40002", + "ports_range": "42000:42999", + "params": { + "positive_importance_flag": false, + "disregard_importance_flag": false, + "unique_value_flag": false, + "count_flag": false + } + }, + "link_creation": { + "params": { + "max_answers": 10, + "repeat_count": 1, + "context": "context", + "attention_update": 0, + "attention_correlation": 0, + "positive_importance_flag": true, + "query_interval": 0, + "query_timeout": 0, + "use_metta_as_query_tokens": false + } + }, + "inference": { + "params": { + "inference_request_timeout": 86400, + "repeat_count": 5, + "max_answers": 150 + } + }, + "evolution": { + "params": { + "population_size": 1000, + "max_generations": 100, + "elitism_rate": 0.01, + "selection_rate": 0.1 + } + }, + "context": { + "params": { + "context": "context", + "use_cache": true, + "enforce_cache_recreation": false, + "initial_rent_rate": 0.75, + "initial_spreading_rate_lowerbound": 0.1, + "initial_spreading_rate_upperbound": 0.1 + } + } + } +})"; + +SystemParameters make_test_parameters() { return SystemParameters(json::parse(kAgentsJson)); } + +void init_test_system_parameters_singleton() { + SystemParametersSingleton::provide(std::make_shared(make_test_parameters())); +} + +} // namespace das_test diff --git a/src/tests/cpp/test_commons/TestSystemParams.h b/src/tests/cpp/test_commons/TestSystemParams.h new file mode 100644 index 00000000..3676cc16 --- /dev/null +++ b/src/tests/cpp/test_commons/TestSystemParams.h @@ -0,0 +1,21 @@ +#pragma once + +#include + +#include + +#include "SystemParameters.h" +#include "SystemParametersSingleton.h" + +namespace das_test { + +/** JSON fixture aligned with agents.*.params sections in config/das.json. */ +extern const char kAgentsJson[]; + +/** Build a SystemParameters instance from {@link kAgentsJson}. */ +commons::SystemParameters make_test_parameters(); + +/** Install test parameters into {@link commons::SystemParametersSingleton}. */ +void init_test_system_parameters_singleton(); + +} // namespace das_test From 7c6ee8e0daf1797aa760a5a86d825bd00185bc84 Mon Sep 17 00:00:00 2001 From: CCG Date: Thu, 28 May 2026 19:02:37 -0300 Subject: [PATCH 2/3] Add SystemParametersValidation --- src/commons/BUILD | 2 + src/commons/SystemParameters.cc | 54 ++--- src/commons/SystemParametersValidation.cc | 187 ++++++++++++++++++ src/commons/SystemParametersValidation.h | 33 ++++ src/tests/cpp/system_parameters_test.cc | 38 ++++ .../cpp/test_commons/TestSystemParams.cc | 1 + 6 files changed, 272 insertions(+), 43 deletions(-) create mode 100644 src/commons/SystemParametersValidation.cc create mode 100644 src/commons/SystemParametersValidation.h diff --git a/src/commons/BUILD b/src/commons/BUILD index fdb70cbb..0597a1f3 100644 --- a/src/commons/BUILD +++ b/src/commons/BUILD @@ -12,6 +12,7 @@ cc_library( "StoppableThread.cc", "SystemParameters.cc", "SystemParametersSingleton.cc", + "SystemParametersValidation.cc", "Utils.cc", ], hdrs = [ @@ -29,6 +30,7 @@ cc_library( "StoppableThread.h", "SystemParameters.h", "SystemParametersSingleton.h", + "SystemParametersValidation.h", "ThreadPool.h", "ThreadSafeHashmap.h", "ThreadSafeHeap.h", diff --git a/src/commons/SystemParameters.cc b/src/commons/SystemParameters.cc index 29b93eac..718e3afe 100644 --- a/src/commons/SystemParameters.cc +++ b/src/commons/SystemParameters.cc @@ -1,5 +1,7 @@ #include "SystemParameters.h" +#include "JsonConfig.h" +#include "SystemParametersValidation.h" #include "Utils.h" using namespace commons; @@ -8,47 +10,11 @@ using namespace std; namespace { -PropertyValue json_scalar_to_property(const json& value, const string& key) { - if (value.is_string()) return value.get(); - if (value.is_number_unsigned()) return value.get(); - if (value.is_number_integer()) return static_cast(value.get()); - if (value.is_number_float()) return value.get(); - if (value.is_boolean()) return value.get(); - if (value.is_null()) return string(""); - RAISE_ERROR("Invalid non-scalar parameter value for key '" + key + "'"); - return string(""); -} - -void load_agent_params(const json& agent_params, - const string& agent, - map& params_by_agent) { - if (!agent_params.is_object()) { - RAISE_ERROR("Parameters for agent '" + agent + "' must be an object"); - } - Properties agent_props; - for (auto pit = agent_params.begin(); pit != agent_params.end(); ++pit) { - const string key = pit.key(); - agent_props[key] = json_scalar_to_property(pit.value(), agent + ".params." + key); - } - params_by_agent[agent] = agent_props; -} - -void load_from_agents(const json& agents, map& params_by_agent) { - if (!agents.is_object()) { - RAISE_ERROR("'agents' must be a JSON object"); - } - for (auto it = agents.begin(); it != agents.end(); ++it) { - const string agent = it.key(); - const json& agent_json = it.value(); - if (!agent_json.is_object()) { - continue; - } - auto agent_params_it = agent_json.find("params"); - if (agent_params_it == agent_json.end() || agent_params_it->is_null()) { - continue; - } - load_agent_params(*agent_params_it, agent + ".params", params_by_agent); +string resolve_schema_version(const json& root) { + if (root.contains("schema_version") && !root["schema_version"].is_null()) { + return root["schema_version"].get(); } + return SystemParametersValidation::SCHEMA_VERSION_1_0; } } // namespace @@ -57,7 +23,8 @@ SystemParameters::SystemParameters(const json& root) { if (!root.is_object() || !root.contains("agents")) { RAISE_ERROR("Missing 'agents' section in parameters JSON"); } - load_from_agents(root["agents"], this->params_by_agent); + SystemParametersValidation::load_from_agents( + root["agents"], resolve_schema_version(root), this->params_by_agent); } SystemParameters::SystemParameters(const JsonConfig& das_config) { @@ -65,11 +32,12 @@ SystemParameters::SystemParameters(const JsonConfig& das_config) { if (agents.is_null()) { RAISE_ERROR("Missing 'agents' section in system config"); } - load_from_agents(*agents, this->params_by_agent); + SystemParametersValidation::load_from_agents( + *agents, das_config.get_schema_version(), this->params_by_agent); } Properties SystemParameters::get_agent_params(const string& agent) const { - auto agent_it = this->params_by_agent.find(agent + ".params"); + auto agent_it = this->params_by_agent.find(agent); if (agent_it == this->params_by_agent.end()) { RAISE_ERROR("Unknown agent: '" + agent + "'"); } diff --git a/src/commons/SystemParametersValidation.cc b/src/commons/SystemParametersValidation.cc new file mode 100644 index 00000000..3e45e18e --- /dev/null +++ b/src/commons/SystemParametersValidation.cc @@ -0,0 +1,187 @@ +#include "SystemParametersValidation.h" + +#include + +#include "Utils.h" + +using namespace commons; +using nlohmann::json; +using namespace std; + +namespace { + +using FieldTypeSchema = unordered_map; +using AgentParamsSchema = unordered_map; + +const AgentParamsSchema& schema_v1_0() { + static const AgentParamsSchema schema = { + {"base_query", + {{"unique_assignment_flag", "bool"}, + {"attention_update", "long"}, + {"attention_correlation", "long"}, + {"max_bundle_size", "unsigned_int"}, + {"max_answers", "unsigned_int"}, + {"use_link_template_cache", "bool"}, + {"populate_metta_mapping", "bool"}, + {"use_metta_as_query_tokens", "bool"}, + {"allow_incomplete_chain_path", "bool"}}}, + {"query", + {{"positive_importance_flag", "bool"}, + {"disregard_importance_flag", "bool"}, + {"unique_value_flag", "bool"}, + {"count_flag", "bool"}}}, + {"link_creation", + {{"max_answers", "unsigned_int"}, + {"repeat_count", "unsigned_int"}, + {"context", "string"}, + {"attention_update", "long"}, + {"attention_correlation", "long"}, + {"positive_importance_flag", "bool"}, + {"query_interval", "long"}, + {"query_timeout", "long"}, + {"use_metta_as_query_tokens", "bool"}}}, + {"inference", + {{"inference_request_timeout", "long"}, + {"repeat_count", "unsigned_int"}, + {"max_answers", "unsigned_int"}}}, + {"evolution", + {{"population_size", "unsigned_int"}, + {"max_generations", "unsigned_int"}, + {"elitism_rate", "double"}, + {"selection_rate", "double"}}}, + {"context", + {{"context", "string"}, + {"use_cache", "bool"}, + {"enforce_cache_recreation", "bool"}, + {"initial_rent_rate", "double"}, + {"initial_spreading_rate_lowerbound", "double"}, + {"initial_spreading_rate_upperbound", "double"}}}, + }; + return schema; +} + +const unordered_map& schemas_by_version() { + static const AgentParamsSchema v1 = schema_v1_0(); + static const unordered_map registry = { + {SystemParametersValidation::SCHEMA_VERSION_1_0, &v1}, + }; + return registry; +} + +const AgentParamsSchema* agent_schema_for_version(const string& schema_version) { + auto registry_it = schemas_by_version().find(schema_version); + if (registry_it == schemas_by_version().end()) { + RAISE_ERROR("Unsupported system parameters schema version: '" + schema_version + "'"); + } + return registry_it->second; +} + +PropertyValue json_scalar_to_property(const json& value, const string& key) { + if (value.is_string()) return value.get(); + if (value.is_number_unsigned()) return value.get(); + if (value.is_number_integer()) return static_cast(value.get()); + if (value.is_number_float()) return value.get(); + if (value.is_boolean()) return value.get(); + if (value.is_null()) return string(""); + RAISE_ERROR("Invalid non-scalar parameter value for key '" + key + "'"); + return string(""); +} + +bool property_matches_type(const PropertyValue& value, const string& expected_type) { + if (expected_type == "string") { + return holds_alternative(value); + } + if (expected_type == "bool") { + return holds_alternative(value); + } + if (expected_type == "double") { + return holds_alternative(value); + } + if (expected_type == "long") { + return holds_alternative(value) || holds_alternative(value); + } + if (expected_type == "unsigned_int") { + return holds_alternative(value) || holds_alternative(value); + } + return false; +} + +Properties parse_agent_params_with_schema(const json& agent_params, + const string& agent, + const AgentParamsSchema& agent_schema) { + if (!agent_params.is_object()) { + RAISE_ERROR("Parameters for agent '" + agent + "' must be an object"); + } + + const auto schema_it = agent_schema.find(agent); + const bool has_schema = schema_it != agent_schema.end(); + + if (has_schema) { + for (const auto& field : schema_it->second) { + if (!agent_params.contains(field.first)) { + RAISE_ERROR("Missing required parameter '" + field.first + "' for agent '" + agent + + "'"); + } + } + } + + Properties agent_props; + for (auto pit = agent_params.begin(); pit != agent_params.end(); ++pit) { + const string key = pit.key(); + const string path = agent + ".params." + key; + + if (has_schema) { + auto field_it = schema_it->second.find(key); + if (field_it == schema_it->second.end()) { + RAISE_ERROR("Unknown parameter '" + key + "' for agent '" + agent + "'"); + } + PropertyValue value = json_scalar_to_property(pit.value(), path); + if (!property_matches_type(value, field_it->second)) { + RAISE_ERROR("Invalid type for parameter '" + key + "' for agent '" + agent + + "' (expected " + field_it->second + ")"); + } + agent_props[key] = value; + } else { + agent_props[key] = json_scalar_to_property(pit.value(), path); + } + } + return agent_props; +} + +} // namespace + +void SystemParametersValidation::validate_schema_version(const string& schema_version) { + agent_schema_for_version(schema_version); +} + +bool SystemParametersValidation::is_supported_schema_version(const string& schema_version) { + return schemas_by_version().find(schema_version) != schemas_by_version().end(); +} + +Properties SystemParametersValidation::parse_agent_params(const json& agent_params, + const string& agent, + const string& schema_version) { + return parse_agent_params_with_schema( + agent_params, agent, *agent_schema_for_version(schema_version)); +} + +void SystemParametersValidation::load_from_agents(const json& agents, + const string& schema_version, + map& params_by_agent) { + validate_schema_version(schema_version); + if (!agents.is_object()) { + RAISE_ERROR("'agents' must be a JSON object"); + } + for (auto it = agents.begin(); it != agents.end(); ++it) { + const string agent = it.key(); + const json& agent_json = it.value(); + if (!agent_json.is_object()) { + continue; + } + auto agent_params_it = agent_json.find("params"); + if (agent_params_it == agent_json.end() || agent_params_it->is_null()) { + continue; + } + params_by_agent[agent] = parse_agent_params(*agent_params_it, agent, schema_version); + } +} diff --git a/src/commons/SystemParametersValidation.h b/src/commons/SystemParametersValidation.h new file mode 100644 index 00000000..afcc371f --- /dev/null +++ b/src/commons/SystemParametersValidation.h @@ -0,0 +1,33 @@ +#pragma once + +#include +#include +#include + +#include "Properties.h" + +using namespace std; + +namespace commons { + +/** + * Validates and parses agent params from das.json against a versioned field schema. + * Schema versions align with the top-level config schema_version (e.g. "1.0"). + */ +class SystemParametersValidation { + public: + static constexpr const char* SCHEMA_VERSION_1_0 = "1.0"; + + static void validate_schema_version(const string& schema_version); + static bool is_supported_schema_version(const string& schema_version); + + static Properties parse_agent_params(const nlohmann::json& agent_params, + const string& agent, + const string& schema_version); + + static void load_from_agents(const nlohmann::json& agents, + const string& schema_version, + map& params_by_agent); +}; + +} // namespace commons diff --git a/src/tests/cpp/system_parameters_test.cc b/src/tests/cpp/system_parameters_test.cc index 7d3d3d43..1f34909a 100644 --- a/src/tests/cpp/system_parameters_test.cc +++ b/src/tests/cpp/system_parameters_test.cc @@ -2,6 +2,7 @@ #include "SystemParameters.h" #include "SystemParametersSingleton.h" +#include "SystemParametersValidation.h" #include "TestSystemParams.h" using namespace commons; @@ -61,6 +62,43 @@ TEST(SystemParametersTest, singleton_matches_direct_instance) { EXPECT_EQ(from_singleton.get("count_flag"), direct.get("count_flag")); } +TEST(SystemParametersValidationTest, rejects_unsupported_schema_version) { + EXPECT_FALSE(SystemParametersValidation::is_supported_schema_version("99.0")); + EXPECT_THROW(SystemParametersValidation::validate_schema_version("99.0"), runtime_error); +} + +TEST(SystemParametersValidationTest, rejects_unknown_parameter_key) { + const char* json_with_extra = R"({ + "schema_version": "1.0", + "agents": { + "query": { + "params": { + "positive_importance_flag": false, + "disregard_importance_flag": false, + "unique_value_flag": false, + "count_flag": false, + "unknown_param": true + } + } + } + })"; + EXPECT_THROW(SystemParameters(nlohmann::json::parse(json_with_extra)), runtime_error); +} + +TEST(SystemParametersValidationTest, rejects_missing_required_parameter) { + const char* json_missing = R"({ + "schema_version": "1.0", + "agents": { + "query": { + "params": { + "count_flag": false + } + } + } + })"; + EXPECT_THROW(SystemParameters(nlohmann::json::parse(json_missing)), runtime_error); +} + TEST(SystemParametersTest, properties_merge_prioritizes_right_hand_side) { Properties base; base["max_answers"] = 0U; diff --git a/src/tests/cpp/test_commons/TestSystemParams.cc b/src/tests/cpp/test_commons/TestSystemParams.cc index 0a7e6885..ce27ea4e 100644 --- a/src/tests/cpp/test_commons/TestSystemParams.cc +++ b/src/tests/cpp/test_commons/TestSystemParams.cc @@ -8,6 +8,7 @@ using nlohmann::json; namespace das_test { const char kAgentsJson[] = R"({ + "schema_version": "1.0", "agents": { "base_query": { "params": { From ad8057818e3b718e0d0fb667d0d78cfe27a65acf Mon Sep 17 00:00:00 2001 From: CCG Date: Mon, 1 Jun 2026 19:09:25 -0300 Subject: [PATCH 3/3] Move schema_version to agents --- config/das.json | 2 +- src/atomdb/adapterdb/README.md | 134 ++++++++++++------ src/commons/JsonConfig.cc | 8 -- src/commons/JsonConfig.h | 3 - src/commons/JsonConfigParser.cc | 43 +----- src/commons/SystemParameters.cc | 17 +-- src/commons/SystemParametersValidation.cc | 73 +++++----- src/commons/SystemParametersValidation.h | 17 +-- src/main/bus_client.cc | 18 +-- src/main/bus_node.cc | 32 ++--- src/main/helpers/Helper.cc | 1 + src/main/helpers/Helper.h | 2 + src/tests/cpp/config_parser_test.cc | 41 ------ src/tests/cpp/system_parameters_test.cc | 20 ++- .../cpp/test_commons/TestSystemParams.cc | 2 +- 15 files changed, 181 insertions(+), 232 deletions(-) diff --git a/config/das.json b/config/das.json index fd4e9ef0..d39c20f6 100644 --- a/config/das.json +++ b/config/das.json @@ -1,5 +1,4 @@ { - "schema_version": "1.0", "atomdb": { "type": "redismongodb", "redis": { @@ -122,6 +121,7 @@ } }, "agents": { + "schema_version": "1.0", "attention": { "endpoint": "localhost:40001" }, diff --git a/src/atomdb/adapterdb/README.md b/src/atomdb/adapterdb/README.md index c27a1031..13552814 100644 --- a/src/atomdb/adapterdb/README.md +++ b/src/atomdb/adapterdb/README.md @@ -43,9 +43,35 @@ feature as f WHERE f.feature_id<=500; ```json { - "schema_version": "1.0", "atomdb": { "type": "adapterdb", + "redis": { + "image": "redis:7.2.3-alpine", + "endpoint": "localhost:40020", + "cluster": false, + "nodes": [ + { + "context": "default", + "ip": "localhost", + "username": "arturgontijo" + } + ] + }, + "mongodb": { + "image": "mongodb/mongodb-community-server:8.2-ubuntu2204", + "endpoint": "localhost:40021", + "username": "admin", + "password": "admin", + "cluster": false, + "cluster_secret_key": "8UDJSgpUCaVOTQG", + "nodes": [ + { + "context": "default", + "ip": "localhost", + "username": "arturgontijo" + } + ] + }, "adapterdb": { "endpoint": "localhost:40023", "type": "postgres", @@ -87,6 +113,10 @@ feature as f WHERE f.feature_id<=500; } } }, + "morkdb": { + "image": "trueagi/das:mork-server-1.0.4", + "endpoint": "localhost:40022" + }, "remote_peers": [ { "uid": "peer1", @@ -134,67 +164,83 @@ feature as f WHERE f.feature_id<=500; } }, "agents": { + "schema_version": "1.0", + "attention": { + "endpoint": "localhost:40001" + }, + "base_query": { + "params": { + "unique_assignment_flag": false, + "attention_update": 0, + "attention_correlation": 0, + "max_bundle_size": 1000, + "max_answers": 0, + "use_link_template_cache": false, + "populate_metta_mapping": false, + "use_metta_as_query_tokens": false, + "allow_incomplete_chain_path": false + } + }, "query": { "endpoint": "localhost:40002", - "ports_range": "42000:42999" + "ports_range": "42000:42999", + "params": { + "positive_importance_flag": false, + "disregard_importance_flag": false, + "unique_value_flag": false, + "count_flag": false + } }, "link_creation": { "endpoint": "localhost:40003", - "ports_range": "43000:43999" + "ports_range": "43000:43999", + "params": { + "max_answers": 10, + "repeat_count": 1, + "context": "context", + "attention_update": 0, + "attention_correlation": 0, + "positive_importance_flag": true, + "query_interval": 0, + "query_timeout": 0, + "use_metta_as_query_tokens": false + } }, "inference": { "endpoint": "localhost:40004", - "ports_range": "44000:44999" + "ports_range": "44000:44999", + "params": { + "inference_request_timeout": 86400, + "repeat_count": 5, + "max_answers": 150 + } }, "evolution": { "endpoint": "localhost:40005", - "ports_range": "45000:45999" - } - }, - "brokers": { - "attention": { - "endpoint": "localhost:40001" + "ports_range": "45000:45999", + "params": { + "population_size": 1000, + "max_generations": 100, + "elitism_rate": 0.01, + "selection_rate": 0.1 + } }, "context": { "endpoint": "localhost:40006", - "ports_range": "46000:46999" + "ports_range": "46000:46999", + "params": { + "context": "context", + "use_cache": true, + "enforce_cache_recreation": false, + "initial_rent_rate": 0.75, + "initial_spreading_rate_lowerbound": 0.1, + "initial_spreading_rate_upperbound": 0.1 + } }, "atomdb": { "endpoint": "localhost:40007", "ports_range": "47000:47999" } - }, - "params": { - "query": { - "max_answers": 100, - "max_bundle_size": 1000, - "count_flag": false, - "attention_update_flag": false, - "unique_assignment_flag": true, - "positive_importance_flag": false, - "populate_metta_mapping": true, - "use_metta_as_query_tokens": true - }, - "link_creation": { - "repeat_count": 1, - "query_interval": 0, - "query_timeout": 0 - }, - "evolution": { - "elitism_rate": 0.08, - "max_generations": 10, - "population_size": 50, - "selection_rate": 0.1, - "total_attention_tokens": 100000 - }, - "context": { - "context": "context", - "use_cache": true, - "enforce_cache_recreation": false, - "initial_rent_rate": 0.25, - "initial_spreading_rate_lowerbound": 0.5, - "initial_spreading_rate_upperbound": 0.7 - } } } -``` \ No newline at end of file +``` diff --git a/src/commons/JsonConfig.cc b/src/commons/JsonConfig.cc index daaf3311..a852c43b 100644 --- a/src/commons/JsonConfig.cc +++ b/src/commons/JsonConfig.cc @@ -67,14 +67,6 @@ JsonConfig::JsonConfig() : nlohmann::json() {} JsonConfig::JsonConfig(nlohmann::json root) : nlohmann::json(std::move(root)) {} -string JsonConfig::get_schema_version() const { - string version = value("schema_version", string("")); - if (version.empty()) { - RAISE_ERROR("schema_version is missing"); - } - return version; -} - JsonPathValue JsonConfig::at_path(const string& dotted_path) const { vector keys = Utils::split(dotted_path, '.'); const nlohmann::json* j = this; diff --git a/src/commons/JsonConfig.h b/src/commons/JsonConfig.h index a0b092d9..060b5d5b 100644 --- a/src/commons/JsonConfig.h +++ b/src/commons/JsonConfig.h @@ -65,9 +65,6 @@ class JsonConfig : public nlohmann::json { JsonConfig(); - /** @return Schema version string (e.g. "2.0"). */ - string get_schema_version() const; - /** @return The root JSON (same as *this; for API compatibility). */ const nlohmann::json& get_json() const { return *this; } diff --git a/src/commons/JsonConfigParser.cc b/src/commons/JsonConfigParser.cc index 4c8e0e5a..7b22fb16 100644 --- a/src/commons/JsonConfigParser.cc +++ b/src/commons/JsonConfigParser.cc @@ -5,7 +5,6 @@ #include #include #include -#include #include #include "Utils.h" @@ -17,35 +16,9 @@ namespace commons { namespace { -const unordered_map>& required_fields_by_version() { - static const unordered_map> m = { - {"1.0", {"atomdb", "atomdb.type", "loaders", "agents"}}, - }; - return m; -} - -string required_fields_error(const string& version, const vector& missing) { - string msg = "Config schema version \"" + version + - "\" requires the following fields that are missing or null: "; - for (size_t i = 0; i < missing.size(); ++i) { - if (i > 0) msg += ", "; - msg += "\"" + missing[i] + "\""; - } - msg += ". Please add them to your config file or use a compatible schema version."; - return msg; -} - -void validate_schema_version(const JsonConfig& config, vector& required) { - vector missing; - for (const string& key : required) { - if (config.at_path(key).is_null()) { - missing.push_back(key); - } - } - if (!missing.empty()) { - string version = config.get_schema_version(); - RAISE_ERROR(required_fields_error(version, missing)); - } +const vector required_fields_by_version() { + static const vector required = {"atomdb", "atomdb.type", "loaders", "agents"}; + return required; } JsonConfig parse_and_validate(const string& json_str) { @@ -55,12 +28,10 @@ JsonConfig parse_and_validate(const string& json_str) { } catch (const exception& e) { RAISE_ERROR("Invalid JSON in config: " + string(e.what())); } - string version = config.get_schema_version(); - try { - vector required = required_fields_by_version().at(version); - validate_schema_version(config, required); - } catch (const exception& e) { - RAISE_ERROR("Invalid schema version: " + version + " " + string(e.what())); + for (const string& key : required_fields_by_version()) { + if (config.at_path(key).is_null()) { + RAISE_ERROR("Missing required field: " + key); + } } return config; } diff --git a/src/commons/SystemParameters.cc b/src/commons/SystemParameters.cc index 718e3afe..4788c4a5 100644 --- a/src/commons/SystemParameters.cc +++ b/src/commons/SystemParameters.cc @@ -8,23 +8,11 @@ using namespace commons; using nlohmann::json; using namespace std; -namespace { - -string resolve_schema_version(const json& root) { - if (root.contains("schema_version") && !root["schema_version"].is_null()) { - return root["schema_version"].get(); - } - return SystemParametersValidation::SCHEMA_VERSION_1_0; -} - -} // namespace - SystemParameters::SystemParameters(const json& root) { if (!root.is_object() || !root.contains("agents")) { RAISE_ERROR("Missing 'agents' section in parameters JSON"); } - SystemParametersValidation::load_from_agents( - root["agents"], resolve_schema_version(root), this->params_by_agent); + SystemParametersValidation::load_from_agents(root["agents"], this->params_by_agent); } SystemParameters::SystemParameters(const JsonConfig& das_config) { @@ -32,8 +20,7 @@ SystemParameters::SystemParameters(const JsonConfig& das_config) { if (agents.is_null()) { RAISE_ERROR("Missing 'agents' section in system config"); } - SystemParametersValidation::load_from_agents( - *agents, das_config.get_schema_version(), this->params_by_agent); + SystemParametersValidation::load_from_agents(*agents, this->params_by_agent); } Properties SystemParameters::get_agent_params(const string& agent) const { diff --git a/src/commons/SystemParametersValidation.cc b/src/commons/SystemParametersValidation.cc index 3e45e18e..e51e91ec 100644 --- a/src/commons/SystemParametersValidation.cc +++ b/src/commons/SystemParametersValidation.cc @@ -8,12 +8,14 @@ using namespace commons; using nlohmann::json; using namespace std; +string SystemParametersValidation::SCHEMA_VERSION = "1.0"; + namespace { using FieldTypeSchema = unordered_map; using AgentParamsSchema = unordered_map; -const AgentParamsSchema& schema_v1_0() { +const AgentParamsSchema& params_schema() { static const AgentParamsSchema schema = { {"base_query", {{"unique_assignment_flag", "bool"}, @@ -60,22 +62,6 @@ const AgentParamsSchema& schema_v1_0() { return schema; } -const unordered_map& schemas_by_version() { - static const AgentParamsSchema v1 = schema_v1_0(); - static const unordered_map registry = { - {SystemParametersValidation::SCHEMA_VERSION_1_0, &v1}, - }; - return registry; -} - -const AgentParamsSchema* agent_schema_for_version(const string& schema_version) { - auto registry_it = schemas_by_version().find(schema_version); - if (registry_it == schemas_by_version().end()) { - RAISE_ERROR("Unsupported system parameters schema version: '" + schema_version + "'"); - } - return registry_it->second; -} - PropertyValue json_scalar_to_property(const json& value, const string& key) { if (value.is_string()) return value.get(); if (value.is_number_unsigned()) return value.get(); @@ -117,11 +103,19 @@ Properties parse_agent_params_with_schema(const json& agent_params, const bool has_schema = schema_it != agent_schema.end(); if (has_schema) { + vector missing_fields; for (const auto& field : schema_it->second) { if (!agent_params.contains(field.first)) { - RAISE_ERROR("Missing required parameter '" + field.first + "' for agent '" + agent + - "'"); + missing_fields.push_back(field.first); + } + } + if (!missing_fields.empty()) { + string err_msg = "Missing required parameters for agent '" + agent + "': "; + for (const auto& field : missing_fields) { + err_msg += field + ", "; } + err_msg.substr(0, err_msg.size() - 2); + RAISE_ERROR(err_msg + "\n\n" + SystemParametersValidation::params_schema_to_json_string()); } } @@ -150,28 +144,37 @@ Properties parse_agent_params_with_schema(const json& agent_params, } // namespace -void SystemParametersValidation::validate_schema_version(const string& schema_version) { - agent_schema_for_version(schema_version); -} - -bool SystemParametersValidation::is_supported_schema_version(const string& schema_version) { - return schemas_by_version().find(schema_version) != schemas_by_version().end(); -} - -Properties SystemParametersValidation::parse_agent_params(const json& agent_params, - const string& agent, - const string& schema_version) { - return parse_agent_params_with_schema( - agent_params, agent, *agent_schema_for_version(schema_version)); +string SystemParametersValidation::params_schema_to_json_string() { + json schema_json = json::object(); + for (const auto& agent : params_schema()) { + schema_json[agent.first]["params"] = json::object(); + for (const auto& param : agent.second) { + schema_json[agent.first]["params"][param.first] = param.second; + } + } + return schema_json.dump(4); } void SystemParametersValidation::load_from_agents(const json& agents, - const string& schema_version, map& params_by_agent) { - validate_schema_version(schema_version); if (!agents.is_object()) { RAISE_ERROR("'agents' must be a JSON object"); } + + if (!agents.contains("schema_version") || !agents["schema_version"].is_string()) { + RAISE_ERROR("Missing or invalid 'agents.schema_version' (expected \"" + + SystemParametersValidation::SCHEMA_VERSION + "\")"); + } + + const string schema_version = agents["schema_version"].get(); + if (schema_version != SystemParametersValidation::SCHEMA_VERSION) { + RAISE_ERROR("Unsupported system parameters schema version: \"" + schema_version + + "\" (expected \"" + SystemParametersValidation::SCHEMA_VERSION + "\")\n\n" + + params_schema_to_json_string()); + } + + const AgentParamsSchema& schema = params_schema(); + for (auto it = agents.begin(); it != agents.end(); ++it) { const string agent = it.key(); const json& agent_json = it.value(); @@ -182,6 +185,6 @@ void SystemParametersValidation::load_from_agents(const json& agents, if (agent_params_it == agent_json.end() || agent_params_it->is_null()) { continue; } - params_by_agent[agent] = parse_agent_params(*agent_params_it, agent, schema_version); + params_by_agent[agent] = parse_agent_params_with_schema(*agent_params_it, agent, schema); } } diff --git a/src/commons/SystemParametersValidation.h b/src/commons/SystemParametersValidation.h index afcc371f..9abbc686 100644 --- a/src/commons/SystemParametersValidation.h +++ b/src/commons/SystemParametersValidation.h @@ -12,22 +12,13 @@ namespace commons { /** * Validates and parses agent params from das.json against a versioned field schema. - * Schema versions align with the top-level config schema_version (e.g. "1.0"). + * Schema version is read from agents.schema_version in das.json (e.g. "1.0"). */ class SystemParametersValidation { public: - static constexpr const char* SCHEMA_VERSION_1_0 = "1.0"; - - static void validate_schema_version(const string& schema_version); - static bool is_supported_schema_version(const string& schema_version); - - static Properties parse_agent_params(const nlohmann::json& agent_params, - const string& agent, - const string& schema_version); - - static void load_from_agents(const nlohmann::json& agents, - const string& schema_version, - map& params_by_agent); + static string SCHEMA_VERSION; + static string params_schema_to_json_string(); + static void load_from_agents(const nlohmann::json& agents, map& params_by_agent); }; } // namespace commons diff --git a/src/main/bus_client.cc b/src/main/bus_client.cc index 6809f37e..7e26bb65 100644 --- a/src/main/bus_client.cc +++ b/src/main/bus_client.cc @@ -16,16 +16,9 @@ using namespace commons; using namespace mains; using namespace std; -namespace { - -const char* kDefaultConfigPath = "config/das.json"; - -} // namespace - int main(int argc, char* argv[]) { try { - auto required_cmd_args = { - Helper::CLIENT, Helper::ENDPOINT, Helper::BUS_ENDPOINT, Helper::PORTS_RANGE}; + auto required_cmd_args = {Helper::ENDPOINT, Helper::BUS_ENDPOINT, Helper::PORTS_RANGE}; auto cmd_args = Utils::parse_command_line(argc, argv); if (cmd_args.find("help") != cmd_args.end()) { cout << Helper::help(Helper::processor_type_from_string(cmd_args[Helper::CLIENT]), @@ -38,13 +31,11 @@ int main(int argc, char* argv[]) { RAISE_ERROR("Required argument missing: " + Helper::CLIENT); } - string config_path = kDefaultConfigPath; - auto it_config = cmd_args.find("config"); - if (it_config != cmd_args.end() && !it_config->second.empty()) { - config_path = it_config->second; + if (cmd_args.find(Helper::CONFIG) == cmd_args.end()) { + RAISE_ERROR("Required argument missing: " + Helper::CONFIG); } - JsonConfig json_config = JsonConfigParser::load(config_path); + JsonConfig json_config = JsonConfigParser::load(cmd_args[Helper::CONFIG]); SystemParametersSingleton::init(json_config); if (cmd_args.find(Helper::ENDPOINT) == cmd_args.end()) { @@ -162,7 +153,6 @@ int main(int argc, char* argv[]) { } } catch (const std::exception& e) { - LOG_ERROR("Exception in bus_client: " + string(e.what())); return 1; } return 0; diff --git a/src/main/bus_node.cc b/src/main/bus_node.cc index 41c02f4d..a8245b3c 100644 --- a/src/main/bus_node.cc +++ b/src/main/bus_node.cc @@ -24,16 +24,24 @@ void ctrl_c_handler(int) { int main(int argc, char* argv[]) { try { - auto required_cmd_args = {Helper::SERVICE}; + auto required_cmd_args = {Helper::SERVICE, Helper::CONFIG}; auto cmd_args = Utils::parse_command_line(argc, argv); - auto it_config = cmd_args.find("config"); - ///////// Loading JSON config - JsonConfig json_config; - if (it_config != cmd_args.end() && !it_config->second.empty()) { - json_config = JsonConfigParser::load(it_config->second); - SystemParametersSingleton::init(json_config); + ///////// Checking args + if (cmd_args.find("help") != cmd_args.end()) { + cout << Helper::help(Helper::processor_type_from_string(cmd_args[Helper::SERVICE])) << endl; + return 0; + } + for (auto req_arg : required_cmd_args) { + if (cmd_args.find(req_arg) == cmd_args.end()) { + RAISE_ERROR("Required argument missing: " + string(req_arg)); + } } + + ///////// Loading JSON config + JsonConfig json_config = JsonConfigParser::load(cmd_args[Helper::CONFIG]); + SystemParametersSingleton::init(json_config); + // Map service name (e.g. "query-engine") to config section path (e.g. "agents.query") string service_name = cmd_args[Helper::SERVICE]; auto it_section = Helper::arg_to_json_config_key.find(service_name); @@ -49,16 +57,6 @@ int main(int argc, char* argv[]) { } } - ///////// Checking args - if (cmd_args.find("help") != cmd_args.end()) { - cout << Helper::help(Helper::processor_type_from_string(cmd_args[Helper::SERVICE])) << endl; - return 0; - } - for (auto req_arg : required_cmd_args) { - if (cmd_args.find(req_arg) == cmd_args.end()) { - RAISE_ERROR("Required argument missing: " + string(req_arg)); - } - } auto required_args = Helper::get_required_arguments(cmd_args[Helper::SERVICE]); for (auto req_arg : required_args) { if (cmd_args.find(req_arg) == cmd_args.end()) { diff --git a/src/main/helpers/Helper.cc b/src/main/helpers/Helper.cc index 8133e6b4..28b0f59d 100644 --- a/src/main/helpers/Helper.cc +++ b/src/main/helpers/Helper.cc @@ -13,6 +13,7 @@ using namespace commons; bool Helper::is_running = true; // Args names +string Helper::CONFIG = "config"; string Helper::SERVICE = "service"; string Helper::CLIENT = "client"; string Helper::ENDPOINT = "endpoint"; diff --git a/src/main/helpers/Helper.h b/src/main/helpers/Helper.h index 82cd8831..64c97e3e 100644 --- a/src/main/helpers/Helper.h +++ b/src/main/helpers/Helper.h @@ -25,6 +25,8 @@ class Helper { Helper() = default; ~Helper() = default; + // Path to the DAS JSON configuration file. + static string CONFIG; // Args names (CLI keys). bus_client: required --client; SERVICE is derived (not a CLI flag). // bus_node vs bus_client use ENDPOINT / BUS_ENDPOINT differently; see merge_client_json_defaults. static string SERVICE; diff --git a/src/tests/cpp/config_parser_test.cc b/src/tests/cpp/config_parser_test.cc index e4e2433a..cbf43fd1 100644 --- a/src/tests/cpp/config_parser_test.cc +++ b/src/tests/cpp/config_parser_test.cc @@ -12,7 +12,6 @@ using namespace commons; namespace { const char* kValidConfigV1 = R"({ - "schema_version": "1.0", "atomdb": { "type": "redismongodb", "redis": { @@ -44,18 +43,9 @@ const char* kValidConfigV1 = R"({ } // namespace -TEST(ConfigParserTest, LoadFromStringValidSchemaV1) { - JsonConfig config = JsonConfigParser::load_from_string(kValidConfigV1); - EXPECT_EQ(config.get_schema_version(), "1.0"); -} - TEST(ConfigParserTest, GetNestedStructure) { JsonConfig config = JsonConfigParser::load_from_string(kValidConfigV1); - // Top-level scalar - string schema = config.at_path("schema_version").get_or(""); - EXPECT_EQ(schema, "1.0"); - // Nested sections auto atomdb = config.at_path("atomdb").get_or(JsonConfig()); string type = atomdb.at_path("type").get_or(""); @@ -104,36 +94,6 @@ TEST(ConfigParserTest, GetNestedMissingKeyReturnsEmptyString) { EXPECT_EQ(config.at_path("atomdb.nonexistent").get_or(""), ""); } -TEST(ConfigParserTest, MissingSchemaVersionThrows) { - const char* no_version = R"({ "atomdb": {}, "loaders": {}, "agents": {} })"; - EXPECT_THROW(JsonConfigParser::load_from_string(no_version), runtime_error); -} - -TEST(ConfigParserTest, UnsupportedSchemaVersionThrows) { - const char* bad_version = R"({ - "schema_version": "99.0", - "atomdb": {}, "loaders": {}, "agents": {} - })"; - EXPECT_THROW(JsonConfigParser::load_from_string(bad_version), runtime_error); -} - -TEST(ConfigParserTest, MissingRequiredFieldThrows) { - const char* no_atomdb = R"({ - "schema_version": "1.0", - "loaders": {}, "agents": {} - })"; - EXPECT_THROW(JsonConfigParser::load_from_string(no_atomdb), runtime_error); -} - -TEST(ConfigParserTest, NullRequiredFieldThrows) { - const char* null_atomdb = R"({ - "schema_version": "1.0", - "atomdb": null, - "loaders": {}, "agents": {} - })"; - EXPECT_THROW(JsonConfigParser::load_from_string(null_atomdb), runtime_error); -} - TEST(ConfigParserTest, InvalidJsonThrows) { EXPECT_THROW(JsonConfigParser::load_from_string("{ invalid json }"), runtime_error); } @@ -142,6 +102,5 @@ TEST(ConfigParserTest, GetJsonReturnsRoot) { JsonConfig config = JsonConfigParser::load_from_string(kValidConfigV1); const auto& j = config.get_json(); EXPECT_TRUE(j.is_object()); - EXPECT_EQ(j["schema_version"].get(), "1.0"); EXPECT_EQ(j["atomdb"]["redis"]["port"].get(), 40020); } diff --git a/src/tests/cpp/system_parameters_test.cc b/src/tests/cpp/system_parameters_test.cc index 1f34909a..f1c8f3fa 100644 --- a/src/tests/cpp/system_parameters_test.cc +++ b/src/tests/cpp/system_parameters_test.cc @@ -63,14 +63,26 @@ TEST(SystemParametersTest, singleton_matches_direct_instance) { } TEST(SystemParametersValidationTest, rejects_unsupported_schema_version) { - EXPECT_FALSE(SystemParametersValidation::is_supported_schema_version("99.0")); - EXPECT_THROW(SystemParametersValidation::validate_schema_version("99.0"), runtime_error); + const char* json_with_extra = R"({ + "agents": { + "schema_version": "99.0", + "query": { + "params": { + "positive_importance_flag": false, + "disregard_importance_flag": false, + "unique_value_flag": false, + "count_flag": false + } + } + } + })"; + EXPECT_THROW(SystemParameters(nlohmann::json::parse(json_with_extra)), runtime_error); } TEST(SystemParametersValidationTest, rejects_unknown_parameter_key) { const char* json_with_extra = R"({ - "schema_version": "1.0", "agents": { + "schema_version": "1.0", "query": { "params": { "positive_importance_flag": false, @@ -87,8 +99,8 @@ TEST(SystemParametersValidationTest, rejects_unknown_parameter_key) { TEST(SystemParametersValidationTest, rejects_missing_required_parameter) { const char* json_missing = R"({ - "schema_version": "1.0", "agents": { + "schema_version": "1.0", "query": { "params": { "count_flag": false diff --git a/src/tests/cpp/test_commons/TestSystemParams.cc b/src/tests/cpp/test_commons/TestSystemParams.cc index ce27ea4e..a9d91a45 100644 --- a/src/tests/cpp/test_commons/TestSystemParams.cc +++ b/src/tests/cpp/test_commons/TestSystemParams.cc @@ -8,8 +8,8 @@ using nlohmann::json; namespace das_test { const char kAgentsJson[] = R"({ - "schema_version": "1.0", "agents": { + "schema_version": "1.0", "base_query": { "params": { "unique_assignment_flag": false,