diff --git a/kilostar/utils/config_loader.py b/kilostar/utils/config_loader.py index fbed5ee..34b1a18 100644 --- a/kilostar/utils/config_loader.py +++ b/kilostar/utils/config_loader.py @@ -1,23 +1,22 @@ -"""Workflow 配置文件管理：读取、缓存、热重载。 +"""KiloStar 统一配置管理：多 YAML 文件加载、组合校验、缓存、热重载。 -配置文件路径：``config/workflow.yaml``（相对于项目根目录）。 -采用模块级单例 + 文件修改时间检测，保证： -- 首次调用时懒加载 -- reload_workflow_config() 显式触发重载 -- 工作流引擎调 get_workflow_config() 始终拿到最新生效值 +配置目录：``config/``（项目根），含 config.yml、workflow.yaml、sandbox.yaml。 +各文件独立维护，启动时统一加载到 ``AppConfig`` 并做 schema 校验。 """ from __future__ import annotations import os from pathlib import Path -from typing import Any +from typing import Any, List, Optional import yaml from pydantic import BaseModel, Field _CONFIG_DIR = Path(__file__).resolve().parent.parent.parent / "config" _WORKFLOW_YAML = _CONFIG_DIR / "workflow.yaml" +_CONFIG_YML = _CONFIG_DIR / "config.yml" +_SANDBOX_YAML = _CONFIG_DIR / "sandbox.yaml" class RetryConfig(BaseModel): @@ -28,28 +27,121 @@ class WorkflowConfig(BaseModel): retry: RetryConfig = Field(default_factory=RetryConfig) -_current: WorkflowConfig | None = None +# ─── Sandbox Models (镜像 sandbox.py 中的定义，供统一加载使用) ─── -def _load_from_disk() -> WorkflowConfig: - if not _WORKFLOW_YAML.exists(): - return WorkflowConfig() - with open(_WORKFLOW_YAML, "r", encoding="utf-8") as f: - data = yaml.safe_load(f) or {} - return WorkflowConfig.model_validate(data) +class FilesystemPolicy(BaseModel): + workspace_root: str = "/tmp/kilostar_workspace" + allowed_read_paths: List[str] = Field(default_factory=lambda: ["/tmp"]) + denied_paths: List[str] = Field(default_factory=list) + + +class ShellPolicy(BaseModel): + enabled: bool = True + blocked_commands: List[str] = Field(default_factory=list) + blocked_operators: List[str] = Field(default_factory=list) + max_timeout: int = 60 + + +class PythonExecutorPolicy(BaseModel): + enabled: bool = True + max_timeout: int = 30 + blocked_imports: List[str] = Field(default_factory=list) + blocked_builtins: List[str] = Field(default_factory=list) + + +class SandboxConfig(BaseModel): + enabled: bool = True + filesystem: FilesystemPolicy = Field(default_factory=FilesystemPolicy) + shell: ShellPolicy = Field(default_factory=ShellPolicy) + python_executor: PythonExecutorPolicy = Field(default_factory=PythonExecutorPolicy) + + +# ─── App-level Models ─── + + +class AppInfo(BaseModel): + version: str = "0.1.1-alpha" + name: str = "Kilostar" + + +class AppConfig(BaseModel): + app: AppInfo = Field(default_factory=AppInfo) + workflow: WorkflowConfig = Field(default_factory=WorkflowConfig) + sandbox: SandboxConfig = Field(default_factory=SandboxConfig) + + +# ─── Unified AppConfig Loading ─── + +_app_current: AppConfig | None = None + + +def _read_yaml(path: Path) -> dict: + if not path.exists(): + return {} + with open(path, "r", encoding="utf-8") as f: + return yaml.safe_load(f) or {} + + +def load_app_config(config_dir: Path | None = None) -> AppConfig: + """从 config/ 下多个 YAML 文件加载并组合校验。""" + d = config_dir or _CONFIG_DIR + + app_raw = _read_yaml(d / "config.yml") + workflow_raw = _read_yaml(d / "workflow.yaml") + sandbox_raw = _read_yaml(d / "sandbox.yaml") + + sandbox_section = sandbox_raw.get("sandbox", sandbox_raw) + + return AppConfig.model_validate({ + "app": app_raw, + "workflow": workflow_raw, + "sandbox": sandbox_section, + }) + + +def get_app_config() -> AppConfig: + global _app_current + if _app_current is None: + _app_current = load_app_config() + return _app_current + + +def reload_app_config(section: str | None = None) -> AppConfig: + """热重载配置。section=None 全量重载，指定 section 只刷新对应部分。""" + global _app_current + if section is None or _app_current is None: + _app_current = load_app_config() + return _app_current + + d = _CONFIG_DIR + if section == "workflow": + raw = _read_yaml(d / "workflow.yaml") + _app_current = _app_current.model_copy( + update={"workflow": WorkflowConfig.model_validate(raw)} + ) + elif section == "sandbox": + raw = _read_yaml(d / "sandbox.yaml") + sandbox_section = raw.get("sandbox", raw) + _app_current = _app_current.model_copy( + update={"sandbox": SandboxConfig.model_validate(sandbox_section)} + ) + else: + _app_current = load_app_config() + + return _app_current + + +# ─── Backward-Compatible Workflow Accessors ─── def get_workflow_config() -> WorkflowConfig: - global _current - if _current is None: - _current = _load_from_disk() - return _current + return get_app_config().workflow def reload_workflow_config() -> WorkflowConfig: - global _current - _current = _load_from_disk() - return _current + reload_app_config(section="workflow") + return get_app_config().workflow def save_workflow_config(config: WorkflowConfig) -> None: @@ -57,4 +149,4 @@ def save_workflow_config(config: WorkflowConfig) -> None: data = config.model_dump() with open(_WORKFLOW_YAML, "w", encoding="utf-8") as f: yaml.dump(data, f, default_flow_style=False, allow_unicode=True) - reload_workflow_config() + reload_app_config(section="workflow") diff --git a/kilostar/utils/sandbox.py b/kilostar/utils/sandbox.py index 3d3102a..728b87f 100644 --- a/kilostar/utils/sandbox.py +++ b/kilostar/utils/sandbox.py @@ -4,66 +4,25 @@ from __future__ import annotations import os import re -from pathlib import Path from typing import List, Optional -import yaml -from pydantic import BaseModel, Field - -_CONFIG_DIR = Path(__file__).resolve().parent.parent.parent / "config" -_SANDBOX_YAML = _CONFIG_DIR / "sandbox.yaml" - - -class FilesystemPolicy(BaseModel): - workspace_root: str = "/tmp/kilostar_workspace" - allowed_read_paths: List[str] = Field(default_factory=lambda: ["/tmp"]) - denied_paths: List[str] = Field(default_factory=list) - - -class ShellPolicy(BaseModel): - enabled: bool = True - blocked_commands: List[str] = Field(default_factory=list) - blocked_operators: List[str] = Field(default_factory=list) - max_timeout: int = 60 - - -class PythonExecutorPolicy(BaseModel): - enabled: bool = True - max_timeout: int = 30 - blocked_imports: List[str] = Field(default_factory=list) - blocked_builtins: List[str] = Field(default_factory=list) - - -class SandboxConfig(BaseModel): - enabled: bool = True - filesystem: FilesystemPolicy = Field(default_factory=FilesystemPolicy) - shell: ShellPolicy = Field(default_factory=ShellPolicy) - python_executor: PythonExecutorPolicy = Field(default_factory=PythonExecutorPolicy) - - -_current: Optional[SandboxConfig] = None - - -def _load_sandbox_config() -> SandboxConfig: - if not _SANDBOX_YAML.exists(): - return SandboxConfig() - with open(_SANDBOX_YAML, "r", encoding="utf-8") as f: - data = yaml.safe_load(f) or {} - raw = data.get("sandbox", data) - return SandboxConfig.model_validate(raw) +from kilostar.utils.config_loader import ( + SandboxConfig, + FilesystemPolicy, + ShellPolicy, + PythonExecutorPolicy, + get_app_config, + reload_app_config, +) def get_sandbox_config() -> SandboxConfig: - global _current - if _current is None: - _current = _load_sandbox_config() - return _current + return get_app_config().sandbox def reload_sandbox_config() -> SandboxConfig: - global _current - _current = _load_sandbox_config() - return _current + reload_app_config(section="sandbox") + return get_app_config().sandbox # ─── Exceptions ─── diff --git a/main.py b/main.py index 301daff..0061f44 100644 --- a/main.py +++ b/main.py @@ -22,6 +22,14 @@ if not _secret_key or _secret_key in _INSECURE_SECRETS: ) sys.exit(1) +from kilostar.utils.config_loader import get_app_config + +try: + _app_cfg = get_app_config() +except Exception as e: + print(f"❌ [致命错误] 配置文件校验失败：{e}") + sys.exit(1) + import asyncio import ray from ray import serve diff --git a/tests/unit/test_api_onebot.py b/tests/unit/test_api_onebot.py index bf754bd..8dad600 100644 --- a/tests/unit/test_api_onebot.py +++ b/tests/unit/test_api_onebot.py @@ -16,6 +16,7 @@ from kilostar.core.individual.regulatory_node.template import MessageResponse def test_verify_token_skipped_when_env_missing(monkeypatch): monkeypatch.delenv("ONEBOT_ACCESS_TOKEN", raising=False) + monkeypatch.setenv("KILOSTAR_ENV", "dev") onebot_mod._verify_token(None) onebot_mod._verify_token("anything") diff --git a/tests/unit/test_config_loader.py b/tests/unit/test_config_loader.py index c59e4e7..5d3ce1c 100644 --- a/tests/unit/test_config_loader.py +++ b/tests/unit/test_config_loader.py @@ -8,13 +8,15 @@ import pytest @pytest.fixture(autouse=True) def isolated_yaml(tmp_path, monkeypatch): - """每个用例用独立的临时 yaml，避免污染真实 config/workflow.yaml。""" + """每个用例用独立的临时目录作为 config 目录，避免污染真实配置文件。""" from kilostar.utils import config_loader - fake_yaml = tmp_path / "workflow.yaml" - monkeypatch.setattr(config_loader, "_WORKFLOW_YAML", fake_yaml) - monkeypatch.setattr(config_loader, "_current", None) - return fake_yaml + monkeypatch.setattr(config_loader, "_CONFIG_DIR", tmp_path) + monkeypatch.setattr(config_loader, "_WORKFLOW_YAML", tmp_path / "workflow.yaml") + monkeypatch.setattr(config_loader, "_CONFIG_YML", tmp_path / "config.yml") + monkeypatch.setattr(config_loader, "_SANDBOX_YAML", tmp_path / "sandbox.yaml") + monkeypatch.setattr(config_loader, "_app_current", None) + return tmp_path / "workflow.yaml" def test_get_workflow_config_returns_default_when_file_absent(): diff --git a/tests/unit/test_sandbox.py b/tests/unit/test_sandbox.py index cc2dc3e..b750afa 100644 --- a/tests/unit/test_sandbox.py +++ b/tests/unit/test_sandbox.py @@ -4,11 +4,16 @@ import os import pytest from unittest.mock import patch -from kilostar.utils.sandbox import ( +from kilostar.utils.config_loader import ( SandboxConfig, FilesystemPolicy, ShellPolicy, PythonExecutorPolicy, + AppConfig, + AppInfo, + WorkflowConfig, +) +from kilostar.utils.sandbox import ( validate_path, validate_shell_command, validate_python_code, @@ -24,47 +29,49 @@ from kilostar.utils.sandbox import ( @pytest.fixture(autouse=True) def reset_sandbox_config(): - """每个测试前重置沙箱配置缓存。""" - import kilostar.utils.sandbox as mod - mod._current = None + """每个测试前重置配置缓存。""" + import kilostar.utils.config_loader as loader + loader._app_current = None yield - mod._current = None + loader._app_current = None @pytest.fixture def mock_config(): """注入测试用的沙箱配置。""" - import kilostar.utils.sandbox as mod - cfg = SandboxConfig( - enabled=True, - filesystem=FilesystemPolicy( - workspace_root="/tmp/kilostar_workspace", - allowed_read_paths=["/tmp"], - denied_paths=["/etc/shadow", "/root"], - ), - shell=ShellPolicy( + import kilostar.utils.config_loader as loader + cfg = AppConfig( + sandbox=SandboxConfig( enabled=True, - blocked_commands=["rm -rf /", "mkfs", "shutdown"], - blocked_operators=["&&", "||", ";", "`", "$("], - max_timeout=60, - ), - python_executor=PythonExecutorPolicy( - enabled=True, - max_timeout=30, - blocked_imports=["os", "subprocess", "shutil"], - blocked_builtins=["exec", "eval", "__import__"], + filesystem=FilesystemPolicy( + workspace_root="/tmp/kilostar_workspace", + allowed_read_paths=["/tmp"], + denied_paths=["/etc/shadow", "/root"], + ), + shell=ShellPolicy( + enabled=True, + blocked_commands=["rm -rf /", "mkfs", "shutdown"], + blocked_operators=["&&", "||", ";", "`", "$("], + max_timeout=60, + ), + python_executor=PythonExecutorPolicy( + enabled=True, + max_timeout=30, + blocked_imports=["os", "subprocess", "shutil"], + blocked_builtins=["exec", "eval", "__import__"], + ), ), ) - mod._current = cfg + loader._app_current = cfg return cfg @pytest.fixture def disabled_config(): """沙箱关闭时的配置。""" - import kilostar.utils.sandbox as mod - cfg = SandboxConfig(enabled=False) - mod._current = cfg + import kilostar.utils.config_loader as loader + cfg = AppConfig(sandbox=SandboxConfig(enabled=False)) + loader._app_current = cfg return cfg @@ -140,10 +147,12 @@ class TestValidateShellCommand: assert validate_shell_command("rm -rf /") == "rm -rf /" def test_shell_disabled_in_policy(self): - import kilostar.utils.sandbox as mod - mod._current = SandboxConfig( - enabled=True, - shell=ShellPolicy(enabled=False), + import kilostar.utils.config_loader as loader + loader._app_current = AppConfig( + sandbox=SandboxConfig( + enabled=True, + shell=ShellPolicy(enabled=False), + ), ) with pytest.raises(CommandViolation, match="已被沙箱策略禁用"): validate_shell_command("ls") @@ -190,10 +199,12 @@ class TestValidatePythonCode: assert validate_python_code("import os") == "import os" def test_python_disabled_in_policy(self): - import kilostar.utils.sandbox as mod - mod._current = SandboxConfig( - enabled=True, - python_executor=PythonExecutorPolicy(enabled=False), + import kilostar.utils.config_loader as loader + loader._app_current = AppConfig( + sandbox=SandboxConfig( + enabled=True, + python_executor=PythonExecutorPolicy(enabled=False), + ), ) with pytest.raises(CodeViolation, match="已被沙箱策略禁用"): validate_python_code("print(1)") diff --git a/tests/unit/test_unified_config.py b/tests/unit/test_unified_config.py new file mode 100644 index 0000000..3429b49 --- /dev/null +++ b/tests/unit/test_unified_config.py @@ -0,0 +1,168 @@ +"""统一配置加载 (AppConfig) 的测试：多文件联合加载、缺失文件默认值、schema 校验、热重载。""" + +from __future__ import annotations + +import yaml +import pytest + +from kilostar.utils.config_loader import ( + AppConfig, + AppInfo, + WorkflowConfig, + RetryConfig, + SandboxConfig, + load_app_config, + get_app_config, + reload_app_config, + get_workflow_config, + reload_workflow_config, + save_workflow_config, +) + + +@pytest.fixture(autouse=True) +def isolated_config(tmp_path, monkeypatch): + """隔离配置目录，每个测试用独立临时目录。""" + from kilostar.utils import config_loader + + monkeypatch.setattr(config_loader, "_CONFIG_DIR", tmp_path) + monkeypatch.setattr(config_loader, "_WORKFLOW_YAML", tmp_path / "workflow.yaml") + monkeypatch.setattr(config_loader, "_CONFIG_YML", tmp_path / "config.yml") + monkeypatch.setattr(config_loader, "_SANDBOX_YAML", tmp_path / "sandbox.yaml") + monkeypatch.setattr(config_loader, "_app_current", None) + return tmp_path + + +# ─── 多文件联合加载 ─── + + +class TestLoadAppConfig: + def test_all_files_present(self, isolated_config): + (isolated_config / "config.yml").write_text( + "version: v0.2.0\nname: TestStar\n", encoding="utf-8" + ) + (isolated_config / "workflow.yaml").write_text( + "retry:\n max_attempts: 10\n", encoding="utf-8" + ) + (isolated_config / "sandbox.yaml").write_text( + "sandbox:\n enabled: false\n", encoding="utf-8" + ) + + cfg = load_app_config(isolated_config) + assert cfg.app.version == "v0.2.0" + assert cfg.app.name == "TestStar" + assert cfg.workflow.retry.max_attempts == 10 + assert cfg.sandbox.enabled is False + + def test_no_files_returns_defaults(self, isolated_config): + cfg = load_app_config(isolated_config) + assert cfg.app.version == "0.1.1-alpha" + assert cfg.app.name == "Kilostar" + assert cfg.workflow.retry.max_attempts == 5 + assert cfg.sandbox.enabled is True + + def test_partial_files_fills_defaults(self, isolated_config): + (isolated_config / "workflow.yaml").write_text( + "retry:\n max_attempts: 8\n", encoding="utf-8" + ) + cfg = load_app_config(isolated_config) + assert cfg.workflow.retry.max_attempts == 8 + assert cfg.app.version == "0.1.1-alpha" + assert cfg.sandbox.enabled is True + + def test_sandbox_without_wrapper_key(self, isolated_config): + """sandbox.yaml 不带顶层 'sandbox:' key 也能正常解析。""" + (isolated_config / "sandbox.yaml").write_text( + "enabled: false\n", encoding="utf-8" + ) + cfg = load_app_config(isolated_config) + assert cfg.sandbox.enabled is False + + +# ─── Schema 校验 ─── + + +class TestSchemaValidation: + def test_invalid_retry_max_attempts_type(self, isolated_config): + (isolated_config / "workflow.yaml").write_text( + "retry:\n max_attempts: not_a_number\n", encoding="utf-8" + ) + with pytest.raises(Exception): + load_app_config(isolated_config) + + def test_retry_max_attempts_out_of_range(self, isolated_config): + (isolated_config / "workflow.yaml").write_text( + "retry:\n max_attempts: 999\n", encoding="utf-8" + ) + with pytest.raises(Exception): + load_app_config(isolated_config) + + def test_config_yml_extra_fields_ignored(self, isolated_config): + (isolated_config / "config.yml").write_text( + "version: v1.0.0\nname: X\nunknown_field: hello\n", encoding="utf-8" + ) + cfg = load_app_config(isolated_config) + assert cfg.app.version == "v1.0.0" + + +# ─── 单例与热重载 ─── + + +class TestSingletonAndReload: + def test_get_app_config_is_singleton(self, isolated_config): + cfg1 = get_app_config() + cfg2 = get_app_config() + assert cfg1 is cfg2 + + def test_reload_full(self, isolated_config): + get_app_config() + (isolated_config / "workflow.yaml").write_text( + "retry:\n max_attempts: 42\n", encoding="utf-8" + ) + cfg = reload_app_config() + assert cfg.workflow.retry.max_attempts == 42 + + def test_reload_section_sandbox(self, isolated_config): + get_app_config() + (isolated_config / "sandbox.yaml").write_text( + "sandbox:\n enabled: false\n", encoding="utf-8" + ) + cfg = reload_app_config(section="sandbox") + assert cfg.sandbox.enabled is False + assert cfg.workflow.retry.max_attempts == 5 + + def test_reload_section_workflow(self, isolated_config): + get_app_config() + (isolated_config / "workflow.yaml").write_text( + "retry:\n max_attempts: 77\n", encoding="utf-8" + ) + cfg = reload_app_config(section="workflow") + assert cfg.workflow.retry.max_attempts == 77 + + +# ─── 向后兼容性 ─── + + +class TestBackwardCompat: + def test_get_workflow_config_delegates(self, isolated_config): + (isolated_config / "workflow.yaml").write_text( + "retry:\n max_attempts: 15\n", encoding="utf-8" + ) + wf = get_workflow_config() + assert wf.retry.max_attempts == 15 + + def test_reload_workflow_config_refreshes(self, isolated_config): + get_workflow_config() + (isolated_config / "workflow.yaml").write_text( + "retry:\n max_attempts: 33\n", encoding="utf-8" + ) + wf = reload_workflow_config() + assert wf.retry.max_attempts == 33 + + def test_save_workflow_config_persists(self, isolated_config): + save_workflow_config(WorkflowConfig(retry=RetryConfig(max_attempts=25))) + on_disk = yaml.safe_load( + (isolated_config / "workflow.yaml").read_text(encoding="utf-8") + ) + assert on_disk == {"retry": {"max_attempts": 25}} + assert get_workflow_config().retry.max_attempts == 25 diff --git a/tests/unit/test_workflow_graph.py b/tests/unit/test_workflow_graph.py index 574e337..708c07d 100644 --- a/tests/unit/test_workflow_graph.py +++ b/tests/unit/test_workflow_graph.py @@ -295,9 +295,10 @@ def test_workflow_graph_state_defaults(): async def test_loop_retry_exceeds_max_attempts_fails(monkeypatch): """环跳转超过 max_attempts 后，工作流应直接 FAILED 而非无限重试。""" from kilostar.utils import config_loader - from kilostar.utils.config_loader import WorkflowConfig, RetryConfig + monkeypatch.setattr(config_loader, "_app_current", None) + from kilostar.utils.config_loader import WorkflowConfig, RetryConfig, AppConfig - monkeypatch.setattr(config_loader, "_current", WorkflowConfig(retry=RetryConfig(max_attempts=2))) + monkeypatch.setattr(config_loader, "_app_current", AppConfig(workflow=WorkflowConfig(retry=RetryConfig(max_attempts=2)))) deps, sink = _make_deps( skill_outputs=[