feat(paths): Add support for project-specific conductor directories
This commit is contained in:
69
src/paths.py
69
src/paths.py
@@ -18,17 +18,18 @@ Configuration (config.toml):
|
|||||||
|
|
||||||
Path Functions:
|
Path Functions:
|
||||||
get_config_path() -> Path to config.toml
|
get_config_path() -> Path to config.toml
|
||||||
get_conductor_dir() -> Path to conductor directory
|
get_conductor_dir(project_path=None) -> Path to conductor directory
|
||||||
get_logs_dir() -> Path to logs/sessions
|
get_logs_dir() -> Path to logs/sessions
|
||||||
get_scripts_dir() -> Path to scripts/generated
|
get_scripts_dir() -> Path to scripts/generated
|
||||||
get_tracks_dir() -> Path to conductor/tracks
|
get_tracks_dir(project_path=None) -> Path to conductor/tracks
|
||||||
get_track_state_dir(track_id) -> Path to conductor/tracks/<track_id>
|
get_track_state_dir(track_id, project_path=None) -> Path to conductor/tracks/<track_id>
|
||||||
get_archive_dir() -> Path to conductor/archive
|
get_archive_dir(project_path=None) -> Path to conductor/archive
|
||||||
|
|
||||||
Resolution Order:
|
Resolution Order:
|
||||||
1. Check environment variable
|
1. Check project-specific manual_slop.toml (for conductor paths)
|
||||||
2. Check config.toml [paths] section
|
2. Check environment variable
|
||||||
3. Fall back to default
|
3. Check config.toml [paths] section
|
||||||
|
4. Fall back to default
|
||||||
|
|
||||||
Usage:
|
Usage:
|
||||||
from src.paths import get_logs_dir, get_scripts_dir
|
from src.paths import get_logs_dir, get_scripts_dir
|
||||||
@@ -51,9 +52,11 @@ _RESOLVED: dict[str, Path] = {}
|
|||||||
def get_config_path() -> Path:
|
def get_config_path() -> Path:
|
||||||
root_dir = Path(__file__).resolve().parent.parent
|
root_dir = Path(__file__).resolve().parent.parent
|
||||||
return Path(os.environ.get("SLOP_CONFIG", root_dir / "config.toml"))
|
return Path(os.environ.get("SLOP_CONFIG", root_dir / "config.toml"))
|
||||||
|
|
||||||
def get_global_presets_path() -> Path:
|
def get_global_presets_path() -> Path:
|
||||||
root_dir = Path(__file__).resolve().parent.parent
|
root_dir = Path(__file__).resolve().parent.parent
|
||||||
return Path(os.environ.get("SLOP_GLOBAL_PRESETS", root_dir / "presets.toml"))
|
return Path(os.environ.get("SLOP_GLOBAL_PRESETS", root_dir / "presets.toml"))
|
||||||
|
|
||||||
def get_project_presets_path(project_root: Path) -> Path:
|
def get_project_presets_path(project_root: Path) -> Path:
|
||||||
return project_root / "project_presets.toml"
|
return project_root / "project_presets.toml"
|
||||||
|
|
||||||
@@ -72,18 +75,47 @@ def get_project_personas_path(project_root: Path) -> Path:
|
|||||||
return project_root / "project_personas.toml"
|
return project_root / "project_personas.toml"
|
||||||
|
|
||||||
def _resolve_path(env_var: str, config_key: str, default: str) -> Path:
|
def _resolve_path(env_var: str, config_key: str, default: str) -> Path:
|
||||||
|
root_dir = Path(__file__).resolve().parent.parent
|
||||||
|
p = None
|
||||||
if env_var in os.environ:
|
if env_var in os.environ:
|
||||||
return Path(os.environ[env_var])
|
p = Path(os.environ[env_var])
|
||||||
|
else:
|
||||||
try:
|
try:
|
||||||
with open(get_config_path(), "rb") as f:
|
with open(get_config_path(), "rb") as f:
|
||||||
cfg = tomllib.load(f)
|
cfg = tomllib.load(f)
|
||||||
if "paths" in cfg and config_key in cfg["paths"]:
|
if "paths" in cfg and config_key in cfg["paths"]:
|
||||||
return Path(cfg["paths"][config_key])
|
p = Path(cfg["paths"][config_key])
|
||||||
except FileNotFoundError:
|
except (FileNotFoundError, tomllib.TOMLDecodeError):
|
||||||
pass
|
pass
|
||||||
return Path(default)
|
if p is None:
|
||||||
|
p = Path(default)
|
||||||
|
if not p.is_absolute():
|
||||||
|
return root_dir / p
|
||||||
|
return p
|
||||||
|
|
||||||
|
def _get_project_conductor_dir_from_toml(project_root: Path) -> Optional[Path]:
|
||||||
|
# Look for manual_slop.toml in project_root
|
||||||
|
toml_path = project_root / 'manual_slop.toml'
|
||||||
|
if not toml_path.exists(): return None
|
||||||
|
try:
|
||||||
|
with open(toml_path, 'rb') as f:
|
||||||
|
data = tomllib.load(f)
|
||||||
|
# Check [conductor] dir = '...'
|
||||||
|
c_dir = data.get('conductor', {}).get('dir')
|
||||||
|
if c_dir:
|
||||||
|
p = Path(c_dir)
|
||||||
|
if not p.is_absolute(): p = project_root / p
|
||||||
|
return p
|
||||||
|
except: pass
|
||||||
|
return None
|
||||||
|
|
||||||
|
def get_conductor_dir(project_path: Optional[str] = None) -> Path:
|
||||||
|
if project_path:
|
||||||
|
project_root = Path(project_path)
|
||||||
|
p = _get_project_conductor_dir_from_toml(project_root)
|
||||||
|
if p: return p
|
||||||
|
return project_root / 'conductor'
|
||||||
|
|
||||||
def get_conductor_dir() -> Path:
|
|
||||||
if "conductor_dir" not in _RESOLVED:
|
if "conductor_dir" not in _RESOLVED:
|
||||||
_RESOLVED["conductor_dir"] = _resolve_path("SLOP_CONDUCTOR_DIR", "conductor_dir", "conductor")
|
_RESOLVED["conductor_dir"] = _resolve_path("SLOP_CONDUCTOR_DIR", "conductor_dir", "conductor")
|
||||||
return _RESOLVED["conductor_dir"]
|
return _RESOLVED["conductor_dir"]
|
||||||
@@ -98,16 +130,15 @@ def get_scripts_dir() -> Path:
|
|||||||
_RESOLVED["scripts_dir"] = _resolve_path("SLOP_SCRIPTS_DIR", "scripts_dir", "scripts/generated")
|
_RESOLVED["scripts_dir"] = _resolve_path("SLOP_SCRIPTS_DIR", "scripts_dir", "scripts/generated")
|
||||||
return _RESOLVED["scripts_dir"]
|
return _RESOLVED["scripts_dir"]
|
||||||
|
|
||||||
def get_tracks_dir() -> Path:
|
def get_tracks_dir(project_path: Optional[str] = None) -> Path:
|
||||||
return get_conductor_dir() / "tracks"
|
return get_conductor_dir(project_path) / "tracks"
|
||||||
|
|
||||||
def get_track_state_dir(track_id: str) -> Path:
|
def get_track_state_dir(track_id: str, project_path: Optional[str] = None) -> Path:
|
||||||
return get_tracks_dir() / track_id
|
return get_tracks_dir(project_path) / track_id
|
||||||
|
|
||||||
def get_archive_dir() -> Path:
|
def get_archive_dir(project_path: Optional[str] = None) -> Path:
|
||||||
return get_conductor_dir() / "archive"
|
return get_conductor_dir(project_path) / "archive"
|
||||||
|
|
||||||
def reset_resolved() -> None:
|
def reset_resolved() -> None:
|
||||||
"""For testing only - clear cached resolutions."""
|
"""For testing only - clear cached resolutions."""
|
||||||
_RESOLVED.clear()
|
_RESOLVED.clear()
|
||||||
|
|
||||||
|
|||||||
55
tests/test_project_paths.py
Normal file
55
tests/test_project_paths.py
Normal file
@@ -0,0 +1,55 @@
|
|||||||
|
import os
|
||||||
|
import pytest
|
||||||
|
import tomllib
|
||||||
|
import tomli_w
|
||||||
|
from pathlib import Path
|
||||||
|
from src import paths
|
||||||
|
|
||||||
|
def test_get_conductor_dir_default():
|
||||||
|
paths.reset_resolved()
|
||||||
|
# Should return default "conductor" relative to root
|
||||||
|
expected = Path(__file__).resolve().parent.parent / "conductor"
|
||||||
|
assert paths.get_conductor_dir() == expected
|
||||||
|
|
||||||
|
def test_get_conductor_dir_project_specific_no_toml(tmp_path):
|
||||||
|
paths.reset_resolved()
|
||||||
|
project_root = tmp_path / "my_project"
|
||||||
|
project_root.mkdir()
|
||||||
|
|
||||||
|
# Should default to project_root / "conductor" if no manual_slop.toml
|
||||||
|
res = paths.get_conductor_dir(project_path=str(project_root))
|
||||||
|
assert res == project_root / "conductor"
|
||||||
|
|
||||||
|
def test_get_conductor_dir_project_specific_with_toml(tmp_path):
|
||||||
|
paths.reset_resolved()
|
||||||
|
project_root = tmp_path / "my_project"
|
||||||
|
project_root.mkdir()
|
||||||
|
|
||||||
|
# Create manual_slop.toml with custom conductor dir
|
||||||
|
toml_path = project_root / "manual_slop.toml"
|
||||||
|
config = {
|
||||||
|
"conductor": {
|
||||||
|
"dir": "custom_tracks"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
with open(toml_path, "wb") as f:
|
||||||
|
f.write(tomli_w.dumps(config).encode())
|
||||||
|
|
||||||
|
res = paths.get_conductor_dir(project_path=str(project_root))
|
||||||
|
assert res == project_root / "custom_tracks"
|
||||||
|
|
||||||
|
def test_get_tracks_dir_project_specific(tmp_path):
|
||||||
|
paths.reset_resolved()
|
||||||
|
project_root = tmp_path / "my_project"
|
||||||
|
project_root.mkdir()
|
||||||
|
|
||||||
|
res = paths.get_tracks_dir(project_path=str(project_root))
|
||||||
|
assert res == project_root / "conductor" / "tracks"
|
||||||
|
|
||||||
|
def test_get_track_state_dir_project_specific(tmp_path):
|
||||||
|
paths.reset_resolved()
|
||||||
|
project_root = tmp_path / "my_project"
|
||||||
|
project_root.mkdir()
|
||||||
|
|
||||||
|
res = paths.get_track_state_dir("track-123", project_path=str(project_root))
|
||||||
|
assert res == project_root / "conductor" / "tracks" / "track-123"
|
||||||
Reference in New Issue
Block a user