feat(paths): implement set_workspace_root function and update related paths

refactor(cli): improve workspace handling in commands and agent modules
fix(interactive): update memory directory path resolution
docs(README): clarify workspace directory structure and environment variable usage
test(paths): add tests for set_workspace_root and ensure_dirs functions
test(skills_manager): update tests to use paths module for USER_SKILLS_DIR
This commit is contained in:
X-iZhang
2026-02-15 11:27:04 +00:00
parent 8893906e94
commit 7b598dd6a6
11 changed files with 173 additions and 31 deletions
+10 -4
View File
@@ -207,14 +207,20 @@ def create_cli_agent(workspace_dir: str | None = None, checkpointer=None):
Args:
workspace_dir: Optional per-session workspace directory. If provided,
creates a fresh backend rooted at this path. If None, uses the
module-level default backend (./workspace).
module-level default backend.
checkpointer: Optional LangGraph checkpointer. If None, falls back
to ``InMemorySaver`` (non-persistent).
"""
from . import paths as _paths
if checkpointer is None:
from langgraph.checkpoint.memory import InMemorySaver # type: ignore[import-untyped]
checkpointer = InMemorySaver()
# Read paths dynamically so runtime set_workspace_root() changes are picked up
_mem_dir = str(_paths.MEMORY_DIR)
_usr_skills_dir = str(_paths.USER_SKILLS_DIR)
if workspace_dir:
set_active_workspace(workspace_dir)
ws_backend = CustomSandboxBackend(
@@ -223,12 +229,12 @@ def create_cli_agent(workspace_dir: str | None = None, checkpointer=None):
timeout=300,
)
sk_backend = MergedReadOnlyBackend(
primary_dir=USER_SKILLS_DIR,
primary_dir=_usr_skills_dir,
secondary_dir=SKILLS_DIR,
)
# Memory always uses SHARED directory (not per-session) for cross-session persistence
mem_backend = FilesystemBackend(
root_dir=MEMORY_DIR,
root_dir=_mem_dir,
virtual_mode=True,
)
be = CompositeBackend(
@@ -242,7 +248,7 @@ def create_cli_agent(workspace_dir: str | None = None, checkpointer=None):
be = backend
mw = [
create_memory_middleware(MEMORY_DIR, extraction_model=chat_model),
create_memory_middleware(_mem_dir, extraction_model=chat_model),
]
# Re-load MCP tools from current config (picks up /mcp add changes)
+7 -3
View File
@@ -4,7 +4,7 @@ import os
from datetime import datetime
from pathlib import Path
from ..paths import new_run_dir, RUNS_DIR
from ..paths import new_run_dir
def _shorten_path(path: str) -> str:
@@ -21,8 +21,11 @@ def _shorten_path(path: str) -> str:
return path
def _deduplicate_run_name(name: str, runs_dir: Path = RUNS_DIR) -> str:
def _deduplicate_run_name(name: str, runs_dir: Path | None = None) -> str:
"""Return *name* if available, otherwise *name_1*, *name_2*, etc."""
if runs_dir is None:
from ..paths import RUNS_DIR
runs_dir = RUNS_DIR
if not (runs_dir / name).exists():
return name
i = 1
@@ -40,7 +43,8 @@ def _create_session_workspace(name: str | None = None) -> str:
if *name* is None.
"""
if name:
session_id = _deduplicate_run_name(name)
from ..paths import RUNS_DIR
session_id = _deduplicate_run_name(name, RUNS_DIR)
else:
session_id = datetime.now().strftime("%Y%m%d_%H%M%S")
workspace_dir = str(new_run_dir(session_id))
+8 -2
View File
@@ -11,7 +11,7 @@ import typer # type: ignore[import-untyped]
from rich.table import Table
from ..stream.display import console
from ..paths import ensure_dirs
from ..paths import ensure_dirs, set_workspace_root
from ._app import app, config_app, mcp_app
from .agent import _deduplicate_run_name, _create_session_workspace, _load_agent
from .mcp_ui import (
@@ -361,16 +361,19 @@ def _main_callback(
# --use-cwd is kept for backward compat but is now the default behavior
if use_cwd:
workspace_dir = os.getcwd()
set_workspace_root(workspace_dir)
workspace_fixed = True
elif workdir:
workspace_dir = os.path.abspath(os.path.expanduser(workdir))
os.makedirs(workspace_dir, exist_ok=True)
set_workspace_root(workspace_dir)
workspace_fixed = True
elif mode:
# Explicit --mode overrides default_workdir
effective_mode = mode
workspace_root = config.default_workdir or os.getcwd()
workspace_root = os.path.abspath(os.path.expanduser(workspace_root))
set_workspace_root(workspace_root)
if effective_mode == "run":
runs_dir = Path(workspace_root, "runs")
session_id = _deduplicate_run_name(name, runs_dir) if name else datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -383,6 +386,7 @@ def _main_callback(
elif config.default_workdir:
# Use configured default workdir with configured mode
workspace_root = os.path.abspath(os.path.expanduser(config.default_workdir))
set_workspace_root(workspace_root)
effective_mode = config.default_mode
if effective_mode == "run":
runs_dir = Path(workspace_root, "runs")
@@ -395,11 +399,13 @@ def _main_callback(
workspace_fixed = True
else:
effective_mode = config.default_mode
workspace_root = os.getcwd()
set_workspace_root(workspace_root)
if effective_mode == "run":
workspace_dir = _create_session_workspace(name)
workspace_fixed = False
else: # daemon mode (default) — use current directory
workspace_dir = os.getcwd()
workspace_dir = workspace_root
workspace_fixed = True
# Ensure memory and skills subdirs exist in workspace
+2 -2
View File
@@ -204,8 +204,8 @@ def cmd_interactive(
import nest_asyncio
nest_asyncio.apply()
from ..EvoScientist import MEMORY_DIR
memory_dir = MEMORY_DIR
from .. import paths
memory_dir = str(paths.MEMORY_DIR)
history_file = str(os.path.expanduser("~/.EvoScientist_history"))
session = PromptSession(
+5 -1
View File
@@ -671,8 +671,12 @@ def _step_workspace(config: EvoScientistConfig) -> tuple[str, str]:
# Custom workdir (optional)
current_default = config.default_workdir or ""
if current_default:
prompt_text = f"Workspace directory (Enter to keep '{current_default}'):"
else:
prompt_text = f"Workspace directory (Enter to use ./{cwd_short}/):"
workdir = questionary.text(
f"Workspace directory (Enter to use ./{cwd_short}/):",
prompt_text,
default=current_default,
style=WIZARD_STYLE,
qmark=QMARK,
+1 -1
View File
@@ -52,7 +52,7 @@ class EvoScientistConfig:
provider: Default LLM provider ('anthropic', 'openai', 'google-genai', or 'nvidia').
model: Default model name (short name or full ID).
default_mode: Default workspace mode ('daemon' or 'run').
default_workdir: Default workspace directory (empty = use ./workspace).
default_workdir: Default workspace directory (empty = use current working directory).
max_concurrent: Maximum concurrent sub-agents.
max_iterations: Maximum delegation iterations.
show_thinking: Whether to show thinking panels in CLI.
+15
View File
@@ -26,6 +26,21 @@ MEMORY_DIR = _env_path("EVOSCIENTIST_MEMORY_DIR") or (WORKSPACE_ROOT / "memory")
USER_SKILLS_DIR = _env_path("EVOSCIENTIST_SKILLS_DIR") or (WORKSPACE_ROOT / "skills")
def set_workspace_root(path: str | Path) -> None:
"""Update workspace root and re-derive dependent directories.
Directories with an explicit environment-variable override keep their
env-var value; all others are re-derived from the new root.
Also resets ``_active_workspace`` to the new root as a safe default.
"""
global WORKSPACE_ROOT, RUNS_DIR, MEMORY_DIR, USER_SKILLS_DIR, _active_workspace
WORKSPACE_ROOT = Path(path).resolve()
_active_workspace = WORKSPACE_ROOT
RUNS_DIR = _env_path("EVOSCIENTIST_RUNS_DIR") or (WORKSPACE_ROOT / "runs")
MEMORY_DIR = _env_path("EVOSCIENTIST_MEMORY_DIR") or (WORKSPACE_ROOT / "memory")
USER_SKILLS_DIR = _env_path("EVOSCIENTIST_SKILLS_DIR") or (WORKSPACE_ROOT / "skills")
def ensure_dirs() -> None:
"""Create runtime subdirectories (memory, skills) if they do not exist.
+5 -5
View File
@@ -1,7 +1,7 @@
"""Skill installation and management for EvoScientist.
This module provides functions for installing, listing, and uninstalling user skills.
Skills are installed to USER_SKILLS_DIR (./workspace/skills/).
Skills are installed to USER_SKILLS_DIR (defaults to <workspace>/skills/).
Supported installation sources:
- Local directory paths
@@ -37,7 +37,7 @@ from pathlib import Path
import yaml
from ..paths import USER_SKILLS_DIR
from .. import paths
@dataclass
@@ -220,7 +220,7 @@ def install_skill(source: str, dest_dir: str | None = None) -> dict:
- path: installed path (if successful)
- error: error message (if failed)
"""
dest_dir = dest_dir or str(USER_SKILLS_DIR)
dest_dir = dest_dir or str(paths.USER_SKILLS_DIR)
os.makedirs(dest_dir, exist_ok=True)
if _is_github_url(source):
@@ -370,7 +370,7 @@ def list_skills(include_system: bool = False) -> list[SkillInfo]:
skills: list[SkillInfo] = []
# User skills
user_dir = Path(USER_SKILLS_DIR)
user_dir = Path(paths.USER_SKILLS_DIR)
if user_dir.exists():
for entry in sorted(user_dir.iterdir()):
if entry.is_dir() and _validate_skill_dir(entry):
@@ -421,7 +421,7 @@ def uninstall_skill(name: str) -> dict:
- success: bool
- error: error message (if failed)
"""
user_dir = Path(USER_SKILLS_DIR).resolve()
user_dir = Path(paths.USER_SKILLS_DIR).resolve()
# Validate name to prevent path traversal
clean_name = _sanitize_name(name)
+10 -8
View File
@@ -239,25 +239,27 @@ The channel can also be started manually with `/channel` in the interactive CLI.
### Runtime Directories
By default, the **workspace** is created under the current directory:
By default, the **workspace root** is the current working directory. Sub-directories
are created automatically:
```
./workspace/
<cwd>/
memory/ # shared MEMORY.md (persistent across sessions)
skills/ # user-installed skills
runs/ # per-session workspaces
runs/ # per-session workspaces (run mode only)
```
You can force workspace to be the current directory via `--use-cwd`.
Use `--workdir` to set a different workspace root, or configure it via
`EvoSci config set default_workdir /path/to/workspace`.
Override individual paths via environment variables:
| Variable | Default | Description |
|----------|---------|-------------|
| `EVOSCIENTIST_WORKSPACE_DIR` | `./workspace` | Root workspace directory |
| `EVOSCIENTIST_RUNS_DIR` | `./workspace/runs` | Per-session run directories |
| `EVOSCIENTIST_MEMORY_DIR` | `./workspace/memory` | Shared memory storage |
| `EVOSCIENTIST_SKILLS_DIR` | `./workspace/skills` | User-installed skills |
| `EVOSCIENTIST_WORKSPACE_DIR` | current directory | Root workspace directory |
| `EVOSCIENTIST_RUNS_DIR` | `<workspace>/runs` | Per-session run directories |
| `EVOSCIENTIST_MEMORY_DIR` | `<workspace>/memory` | Shared memory storage |
| `EVOSCIENTIST_SKILLS_DIR` | `<workspace>/skills` | User-installed skills |
### Script Inference
```python
+105
View File
@@ -0,0 +1,105 @@
"""Tests for EvoScientist.paths — set_workspace_root and ensure_dirs."""
import os
from unittest import mock
import pytest
from EvoScientist import paths
@pytest.fixture(autouse=True)
def _restore_paths():
"""Snapshot module-level path globals and restore after each test."""
orig = {
"WORKSPACE_ROOT": paths.WORKSPACE_ROOT,
"RUNS_DIR": paths.RUNS_DIR,
"MEMORY_DIR": paths.MEMORY_DIR,
"USER_SKILLS_DIR": paths.USER_SKILLS_DIR,
"_active_workspace": paths._active_workspace,
}
yield
paths.WORKSPACE_ROOT = orig["WORKSPACE_ROOT"]
paths.RUNS_DIR = orig["RUNS_DIR"]
paths.MEMORY_DIR = orig["MEMORY_DIR"]
paths.USER_SKILLS_DIR = orig["USER_SKILLS_DIR"]
paths._active_workspace = orig["_active_workspace"]
class TestSetWorkspaceRoot:
"""Tests for set_workspace_root()."""
def test_updates_derived_dirs(self, tmp_path):
"""set_workspace_root should update WORKSPACE_ROOT and all derived dirs."""
new_root = tmp_path / "my_workspace"
new_root.mkdir()
paths.set_workspace_root(new_root)
assert paths.WORKSPACE_ROOT == new_root.resolve()
assert paths.RUNS_DIR == new_root.resolve() / "runs"
assert paths.MEMORY_DIR == new_root.resolve() / "memory"
assert paths.USER_SKILLS_DIR == new_root.resolve() / "skills"
def test_resets_active_workspace(self, tmp_path):
"""set_workspace_root should reset _active_workspace to new root."""
new_root = tmp_path / "ws"
new_root.mkdir()
# Set active workspace to something different first
paths._active_workspace = tmp_path / "other"
paths.set_workspace_root(new_root)
assert paths._active_workspace == new_root.resolve()
def test_preserves_env_overrides(self, tmp_path):
"""Dirs set via env vars should NOT be overwritten by set_workspace_root."""
custom_mem = tmp_path / "custom_memory"
custom_skills = tmp_path / "custom_skills"
custom_runs = tmp_path / "custom_runs"
env = {
"EVOSCIENTIST_MEMORY_DIR": str(custom_mem),
"EVOSCIENTIST_SKILLS_DIR": str(custom_skills),
"EVOSCIENTIST_RUNS_DIR": str(custom_runs),
}
new_root = tmp_path / "ws"
new_root.mkdir()
with mock.patch.dict(os.environ, env):
paths.set_workspace_root(new_root)
# WORKSPACE_ROOT and _active_workspace should still update
assert paths.WORKSPACE_ROOT == new_root.resolve()
assert paths._active_workspace == new_root.resolve()
# Derived dirs should reflect the env overrides, not the new root
assert paths.MEMORY_DIR == custom_mem.expanduser()
assert paths.USER_SKILLS_DIR == custom_skills.expanduser()
assert paths.RUNS_DIR == custom_runs.expanduser()
def test_accepts_string_path(self, tmp_path):
"""set_workspace_root should accept str as well as Path."""
new_root = tmp_path / "str_ws"
new_root.mkdir()
paths.set_workspace_root(str(new_root))
assert paths.WORKSPACE_ROOT == new_root.resolve()
class TestEnsureDirsUsesUpdatedPaths:
"""ensure_dirs should create dirs at the currently set paths."""
def test_ensure_dirs_uses_updated_paths(self, tmp_path):
"""After set_workspace_root, ensure_dirs creates dirs at new location."""
new_root = tmp_path / "workspace"
new_root.mkdir()
paths.set_workspace_root(new_root)
paths.ensure_dirs()
assert (new_root / "memory").is_dir()
assert (new_root / "skills").is_dir()
+5 -5
View File
@@ -272,7 +272,7 @@ class TestListSkills:
"""Tests for list_skills function."""
def test_list_empty_dir(self, temp_skills_dir):
with patch("EvoScientist.tools.skills_manager.USER_SKILLS_DIR", temp_skills_dir):
with patch("EvoScientist.paths.USER_SKILLS_DIR", temp_skills_dir):
skills = list_skills(include_system=False)
assert skills == []
@@ -280,7 +280,7 @@ class TestListSkills:
# Install a skill
install_skill(str(sample_skill_dir), str(temp_skills_dir))
with patch("EvoScientist.tools.skills_manager.USER_SKILLS_DIR", temp_skills_dir):
with patch("EvoScientist.paths.USER_SKILLS_DIR", temp_skills_dir):
skills = list_skills(include_system=False)
assert len(skills) == 1
@@ -302,7 +302,7 @@ description: Skill number {i}
)
install_skill(str(skill_dir), str(temp_skills_dir))
with patch("EvoScientist.tools.skills_manager.USER_SKILLS_DIR", temp_skills_dir):
with patch("EvoScientist.paths.USER_SKILLS_DIR", temp_skills_dir):
skills = list_skills(include_system=False)
assert len(skills) == 3
@@ -324,7 +324,7 @@ class TestUninstallSkill:
# Install first
install_skill(str(sample_skill_dir), str(temp_skills_dir))
with patch("EvoScientist.tools.skills_manager.USER_SKILLS_DIR", temp_skills_dir):
with patch("EvoScientist.paths.USER_SKILLS_DIR", temp_skills_dir):
result = uninstall_skill("sample-skill")
assert result["success"] is True
@@ -334,7 +334,7 @@ class TestUninstallSkill:
assert not skill_path.exists()
def test_uninstall_nonexistent_skill(self, temp_skills_dir):
with patch("EvoScientist.tools.skills_manager.USER_SKILLS_DIR", temp_skills_dir):
with patch("EvoScientist.paths.USER_SKILLS_DIR", temp_skills_dir):
result = uninstall_skill("nonexistent-skill")
assert result["success"] is False