Skip to content
Merged
Show file tree
Hide file tree
Changes from 9 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion das-cli/src/commands/config/config_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,9 +73,20 @@ def __init__(

def _set_file_path(self, save_path) -> None:
self._settings.set_path(save_path)

self._settings.rewind()

load_error = self._settings.get_load_error()
if load_error is not None:
raise ValueError(
f"Could not load configuration from '{save_path}': {load_error}. "
"The file was left unchanged. Fix the JSON and try again."
) from load_error

if not self._settings.exists():
raise ValueError(
f"Configuration file at '{save_path}' is empty. " "The file was left unchanged."
)

verify_populate_missing_values(self._settings, save_path)

self.stdout(
Expand Down
6 changes: 6 additions & 0 deletions das-cli/src/commands/config/config_sections/agents_params.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
"unique_assignment_flag": False,
"attention_update": 0,
"attention_correlation": 0,
"attention_focus_strictness": 0.0,
"max_bundle_size": 1000,
"max_answers": 0,
"use_link_template_cache": False,
Expand Down Expand Up @@ -81,6 +82,11 @@ def setup_base_query_params(settings: Settings):
default=BASE_QUERY_DEFAULTS["attention_correlation"],
type=int,
),
"attention_focus_strictness": Command.prompt(
"Attention focus strictness",
default=BASE_QUERY_DEFAULTS["attention_focus_strictness"],
type=float,
),
"max_bundle_size": Command.prompt(
"Max bundle size",
default=BASE_QUERY_DEFAULTS["max_bundle_size"],
Expand Down
66 changes: 66 additions & 0 deletions das-cli/src/commands/config/config_sections/normalize_file.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
import getpass
from copy import deepcopy
from typing import Any, Dict, List

from common.config.core import get_core_defaults_dict
from common.docker import RemoteContextManager
from common.docker.remote_context_manager import Server
from common.settings import Settings
Expand Down Expand Up @@ -44,8 +46,72 @@ def normalize_servers(
return updated_nodes


def _fill_missing_values(current: Any, defaults: Any) -> Any:
"""Add keys present in defaults but missing in current; never overwrite user values."""
if not isinstance(defaults, dict):
return current if current is not None else deepcopy(defaults)

result = dict(current) if isinstance(current, dict) else {}
for key, default_value in defaults.items():
if key not in result:
result[key] = deepcopy(default_value)
else:
result[key] = _fill_missing_values(result[key], default_value)
return result


def _defaults_for_config(content: Dict[str, Any]) -> Dict[str, Any]:
"""Return schema defaults trimmed to the active atomdb type (same rules as validation)."""
expected = deepcopy(get_core_defaults_dict())
default_atomdb_type = expected.get("atomdb", {}).get("type", "redismongodb")
atomdb_type = content.get("atomdb", {}).get("type") or default_atomdb_type
atomdb_section = expected["atomdb"]

if atomdb_type != "adapterdb":
atomdb_section.pop("adapterdb", None)

if atomdb_type != "remotedb":
atomdb_section.pop("remote_peers", None)

if atomdb_type != "morkdb":
atomdb_section.pop("mongodb", None)
atomdb_section.pop("morkdb", None)

if atomdb_type != "redismongodb":
atomdb_section.pop("mongodb", None)
atomdb_section.pop("redis", None)

adapterdb = atomdb_section.get("adapterdb")
if adapterdb:
backend = adapterdb.get("atomdb_backend")
if backend:
backend_type = (
content.get("atomdb", {}).get("adapterdb", {}).get("atomdb_backend", {}).get("type")
) or backend.get("type")

if backend_type != "redismongodb":
backend.pop("redis", None)
backend.pop("mongodb", None)

if backend_type != "morkdb":
backend.pop("morkdb", None)

if backend_type != "inmemorydb":
backend.pop("inmemorydb", None)

return expected
Comment thread
coderabbitai[bot] marked this conversation as resolved.


def verify_populate_missing_values(settings: Settings, path: str) -> None:
content: Dict[str, Any] = settings.get_content()
content = _fill_missing_values(content, _defaults_for_config(content))

base_query_params = content.get("agents", {}).get("base_query", {}).get("params")
if isinstance(base_query_params, dict) and "attention_focus_strictness" in base_query_params:
base_query_params["attention_focus_strictness"] = float(
base_query_params["attention_focus_strictness"]
)

current_user = getpass.getuser()

context_manager = RemoteContextManager()
Expand Down
4 changes: 1 addition & 3 deletions das-cli/src/commands/system/system_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,10 +44,9 @@ class SystemStatus(Command):
help="Sets how many seconds of cooldown before updating the metrics again.",
default=2,
required=False,
)
),
]


@inject
def __init__(
self,
Expand Down Expand Up @@ -218,7 +217,6 @@ def machine_loop():
latest_machine.clear()
latest_machine.update(data)


except Exception as e:
print(f"[machine_loop] {e}")

Expand Down
4 changes: 2 additions & 2 deletions das-cli/src/common/command.py
Original file line number Diff line number Diff line change
Expand Up @@ -331,11 +331,11 @@ def _remote_run(self, kwargs, remote_kwargs):
command = f"{prefix} {command_path} {extra_args} {remote_context}".strip()

try:

if "config" not in command_path:
self._check_remote_config(remote_kwargs)

# Ignores this check when a config command is called, prevents command from breaking when user is setting up configuration across multiple remote machines.
# Ignores this check when a config command is called, prevents command from breaking when user is setting up configuration across multiple remote machines.
Connection(**remote_kwargs).run(command, pty=False)

except Exception as e:
Expand Down
15 changes: 14 additions & 1 deletion das-cli/src/common/config/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,7 @@ def get_core_defaults_dict() -> Dict[str, Any]:
"unique_assignment_flag": False,
"attention_update": 0,
"attention_correlation": 0,
"attention_focus_strictness": 0.0,
"max_bundle_size": 1000,
"max_answers": 0,
"use_link_template_cache": False,
Expand Down Expand Up @@ -159,7 +160,19 @@ def get_core_defaults_dict() -> Dict[str, Any]:
},
},
"atomdb": {"endpoint": "localhost:40007", "ports_range": "47000:47999"},
"command_router": {"endpoint": "localhost:40008", "ports_range": "48000:48999"},
"command_router": {
"endpoint": "localhost:40008",
"ports_range": "48000:48999",
"http_api": {
"endpoint": "localhost:40009",
"thread_pool_size": 4,
"max_concurrent_executions": 100,
"max_queued_executions": 500,
"max_events_per_execution": 100000,
"stream_items_per_chunk": 100,
"execution_retention_ms": 900000,
},
},
},
"environment": {"jupyter": {"endpoint": "localhost:40019"}},
}
Expand Down
36 changes: 30 additions & 6 deletions das-cli/src/common/config/store.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@ def __init__(self, env_file_path: str):
self._content: Dict[str, Any] = {}
self._new_content: Dict[str, Any] = {}
self._overwrite_mode = False
self._load_error: Exception | None = None
self.rewind()

def get_content(self) -> dict:
Expand All @@ -102,24 +103,47 @@ def set_path(self, new_file_path: str) -> None:
self._file_path = new_file_path

def save_path(self) -> None:
env_file = open(self._env_path, "w")
env_file.write(f"configpath={self._file_path}\n")
os.makedirs(os.path.dirname(self._env_path), exist_ok=True)
with open(self._env_path, "w", encoding="utf-8") as env_file:
env_file.write(f"configpath={self._file_path}\n")

def get_dir_path(self) -> str:
return os.path.dirname(self._file_path)

def file_exists(self) -> bool:
return bool(self._file_path) and os.path.isfile(self._file_path)

def exists(self) -> bool:
return len(self.get_content().items()) > 0
return isinstance(self.get_content(), dict) and len(self.get_content()) > 0

def rewind(self):
self._new_content = {}
try:
with open(self._file_path, "r") as f:
self._content = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
with open(self._file_path, "r", encoding="utf-8") as f:
loaded = json.load(f)
if not isinstance(loaded, dict):
raise json.JSONDecodeError(
"Configuration root must be a JSON object",
doc=str(self._file_path),
pos=0,
)
self._content = loaded
self._load_error = None
except FileNotFoundError as error:
self._content = {}
self._load_error = error
except (json.JSONDecodeError, UnicodeDecodeError) as error:
self._content = {}
self._load_error = error
except OSError as error:
self._content = {}
self._load_error = error
Comment thread
coderabbitai[bot] marked this conversation as resolved.

return self

def get_load_error(self) -> Exception | None:
return self._load_error

def enable_overwrite_mode(self):
self._overwrite_mode = True
self._content = {}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -145,9 +145,9 @@ def _parse_container_stats(self, stats: dict) -> dict:
cpu_percent = self._calculate_cpu_percent(stats)

memory_usage = stats.get("memory_stats", {}).get("usage", 0)

memory_mb = round(
memory_usage / (1024 ** 3),
memory_usage / (1024**3),
2,
)

Expand Down
1 change: 1 addition & 0 deletions das-cli/src/common/prompt_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ def _is_remote_invocation(ctx) -> bool:
params = ctx.params
return bool(params.get("remote") or params.get("host") or params.get("user"))


class ReachableIpAddress(ParamType):
name = "reachable ip address"

Expand Down
29 changes: 27 additions & 2 deletions das-cli/src/common/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,10 +80,35 @@ def validate_configuration_file(self):
self.raise_on_missing_file()
self.raise_on_version_mismatch()

def get_load_error(self) -> Exception | None:
store = self._store
if hasattr(store, "get_load_error"):
return store.get_load_error()
return getattr(store, "_load_error", None)

def raise_on_missing_file(self):
if not self.exists():
path = self.get_path()
store = self._store
load_error = self.get_load_error()

if hasattr(store, "file_exists") and not store.file_exists():
raise FileNotFoundError(
"Configuration file not found. You can run the command 'config set' to create a configuration file or point to an existing file."
f"Configuration file not found at '{path}'. "
"Run 'das-cli config set --file <path>' to point to an existing file, "
"or 'das-cli config set' to create one."
)

if load_error is not None:
raise ValueError(
f"Configuration file at '{path}' could not be loaded: {load_error}. "
"Fix the JSON and try again."
) from load_error

if not self.exists():
raise ValueError(
f"Configuration file at '{path}' is empty. "
"Restore a valid DAS config JSON, then run "
"'das-cli config set --file <path>'."
)

def raise_on_version_mismatch(self):
Expand Down
14 changes: 12 additions & 2 deletions das-cli/tests/integration/fixtures/config/simple.json
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@
}
},
"agents": {
"schema_version": "1.0",
"schema_version": "1.0.1",
"attention": {
"endpoint": "localhost:40001"
},
Expand All @@ -47,6 +47,7 @@
"unique_assignment_flag": false,
"attention_update": 0,
"attention_correlation": 0,
"attention_focus_strictness": 0.0,
"max_bundle_size": 1000,
"max_answers": 0,
"use_link_template_cache": false,
Expand Down Expand Up @@ -117,7 +118,16 @@
},
"command_router": {
"endpoint": "localhost:40008",
"ports_range": "48000:48999"
"ports_range": "48000:48999",
"http_api": {
"endpoint": "localhost:40009",
"thread_pool_size": 4,
"max_concurrent_executions": 100,
"max_queued_executions": 500,
"max_events_per_execution": 100000,
"stream_items_per_chunk": 100,
"execution_retention_ms": 900000
}
}
},
"environment": {
Expand Down
3 changes: 3 additions & 0 deletions das-dashboard/backend/services/config_services.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,9 @@ async def save_config(self, configuration_entries: ConfigurationEntriesDto) -> d
return {
"message": message,
"content": nested_config,
# Pre-serialized with Python so floats like 0.0 survive browser download
# (JS JSON.stringify turns 0.0 into 0, which DAS rejects for doubles).
"content_text": json.dumps(nested_config, indent=2),
"remote_hosts": remote_hosts,
"hosts": self.web_config.map_dashboard_hosts(),
}
Expand Down
Loading
Loading