Private
Public Access
0
0
Files
manual_slop/src/paths.py
T

310 lines
13 KiB
Python

"""
Paths - Single source of truth for all application paths.
All paths are resolved ONCE at startup via `initialize_paths(config_path)`,
which reads the active config.toml's `[paths]` section (with env-var overrides)
and builds an immutable `PathsConfig` snapshot. Path getters are trivial
lookups into this snapshot.
**Usage contract:**
1. Call `initialize_paths(config_path)` ONCE at process startup, BEFORE any
path getter is invoked. This is the only correct entry point.
2. After init, all `get_*_path()` functions return cached `Path` objects.
3. To change paths (e.g., in tests), call `initialize_paths(new_config_path)`
again — atomic swap under lock. Do not mutate `PathsConfig` instances;
they are frozen.
**Thread safety:** The singleton swap is guarded by an RLock. `PathsConfig`
is a `@dataclass(frozen=True)`, so reads of individual fields are atomic.
Reader threads see a consistent snapshot; writer threads serialize through
the lock. No partial writes.
**Resolution priority** (per key, in `initialize_paths`):
1. Env var (e.g., `SLOP_GLOBAL_PRESETS`) if set
2. `config.toml [paths]` entry if present
3. Default `<project_root>/<default_filename>`
**Codepath ordering:**
The major codepaths that consume paths are:
- `sloppy.py` (production GUI entry point)
- `src/app_controller.py:AppController.__init__`
- `src/presets.py`, `src/tool_presets.py`, `src/personas.py`, etc.
`initialize_paths()` must run BEFORE any of these. In sloppy.py, it runs
at the top of `__main__`. In tests, it runs at conftest module body (before
any src/ import). In other contexts (e.g., direct library use), the caller
is responsible.
If a path getter is called before `initialize_paths()`, a `RuntimeError`
is raised. This catches the "bad programmer" case where ordering is wrong.
"""
import os
import threading
import tomllib
from dataclasses import dataclass
from pathlib import Path
from typing import Optional, Any
@dataclass(frozen=True)
class PathsConfig:
"""Immutable snapshot of resolved paths. Created ONCE per process.
[C: src/paths.py:initialize_paths, src/paths.py:_cfg]"""
config_path: Path
presets: Path
tool_presets: Path
personas: Path
themes: Path
workspace_profiles: Path
credentials: Path
logs_dir: Path
scripts_dir: Path
_PATHS_CONFIG: Optional[PathsConfig] = None
_PATHS_LOCK = threading.RLock()
def _default_paths_config() -> PathsConfig:
"""Build the default PathsConfig (no [paths] overrides, just defaults).
Called once at module load to ensure _PATHS_CONFIG is never None for
callers that don't explicitly initialize (e.g., subprocess imports).
[C: src/paths.py:initialize_paths, src/paths.py:_module_init_default]"""
root_dir = Path(__file__).resolve().parent.parent
config_path = root_dir / "config.toml"
cfg = PathsConfig(
config_path = config_path,
presets = root_dir / "presets.toml",
tool_presets = root_dir / "tool_presets.toml",
personas = root_dir / "personas.toml",
themes = root_dir / "themes",
workspace_profiles = root_dir / "workspace_profiles.toml",
credentials = root_dir / "credentials.toml",
logs_dir = root_dir / "logs" / "sessions",
scripts_dir = root_dir / "scripts" / "generated",
)
return cfg
def _module_init_default() -> None:
"""Initialize _PATHS_CONFIG with defaults at module load.
Idempotent. Subsequent calls to initialize_paths(<custom>) override this.
[C: src/paths.py:initialize_paths, src/paths.py:reset_paths]"""
global _PATHS_CONFIG
if _PATHS_CONFIG is None:
_PATHS_CONFIG = _default_paths_config()
_module_init_default()
def _resolve_path(env_var: str, config_key: str, default: Path, config_path: Path) -> Path:
"""Internal: resolve one path from env var -> config [paths] -> default.
Called only from initialize_paths(). Not thread-safe; caller holds lock."""
root_dir = Path(__file__).resolve().parent.parent
if env_var in os.environ:
return Path(os.environ[env_var])
try:
with open(config_path, "rb") as f:
cfg = tomllib.load(f)
if "paths" in cfg and config_key in cfg["paths"]:
p = Path(cfg["paths"][config_key])
return p if p.is_absolute() else root_dir / p
except (FileNotFoundError, tomllib.TOMLDecodeError):
pass
return default if default.is_absolute() else root_dir / default
def initialize_paths(config_path: Optional[Path] = None) -> PathsConfig:
"""Initialize the global paths singleton. Call this ONCE at startup,
BEFORE any path getter is invoked. Atomic swap under RLock.
If config_path is None, uses the default `<project_root>/config.toml`.
This is the SOLE entry point for setting the path graph at runtime.
Tests re-init to reset.
Raises:
OSError: if the config_path cannot be opened (other than FileNotFoundError
which is treated as "no [paths] overrides, use defaults")
TypeError: if config_path is not a Path
Returns:
The newly installed PathsConfig snapshot.
[C: src/paths.py:_cfg, tests/conftest.py:_setup_test_paths, sloppy.py:main]"""
global _PATHS_CONFIG
if config_path is None:
root_dir = Path(__file__).resolve().parent.parent
config_path = root_dir / "config.toml"
config_path = Path(config_path).resolve()
root_dir = Path(__file__).resolve().parent.parent
cfg = PathsConfig(
config_path = config_path,
presets = _resolve_path("SLOP_GLOBAL_PRESETS", "presets", root_dir / "presets.toml", config_path),
tool_presets = _resolve_path("SLOP_GLOBAL_TOOL_PRESETS", "tool_presets", root_dir / "tool_presets.toml", config_path),
personas = _resolve_path("SLOP_GLOBAL_PERSONAS", "personas", root_dir / "personas.toml", config_path),
themes = _resolve_path("SLOP_GLOBAL_THEMES", "themes", root_dir / "themes", config_path),
workspace_profiles = _resolve_path("SLOP_GLOBAL_WORKSPACE_PROFILES", "workspace_profiles", root_dir / "workspace_profiles.toml", config_path),
credentials = _resolve_path("SLOP_CREDENTIALS", "credentials", root_dir / "credentials.toml", config_path),
logs_dir = _resolve_path("SLOP_LOGS_DIR", "logs_dir", root_dir / "logs" / "sessions", config_path),
scripts_dir = _resolve_path("SLOP_SCRIPTS_DIR", "scripts_dir", root_dir / "scripts" / "generated", config_path),
)
with _PATHS_LOCK:
_PATHS_CONFIG = cfg
return cfg
def _cfg() -> PathsConfig:
"""Internal: get the current singleton, raising if uninitialized."""
if _PATHS_CONFIG is None:
raise RuntimeError(
"src.paths not initialized. Call paths.initialize_paths(<config.toml>) "
"BEFORE any path getter. See src/paths.py docstring for codepath ordering."
)
return _PATHS_CONFIG
# === Trivial getters (single source of truth) ===
def get_config_path() -> Path:
"""Active config.toml path. Frozen at initialize_paths() time.
[C: src/app_controller.py:AppController.load_config,
src/app_controller.py:AppController.init_state,
src/models.py:_load_config_from_disk,
tests/test_test_sandbox.py]"""
return _cfg().config_path
def get_global_presets_path() -> Path:
"""Global presets file. Frozen at initialize_paths() time.
[C: src/presets.py:PresetManager.__init__, src/presets.py:PresetManager.delete_preset, src/presets.py:PresetManager.get_preset_scope]"""
return _cfg().presets
def get_project_presets_path(project_root: Path) -> Path:
"""Project-specific presets file. Computed from project_root (no cache).
[C: src/presets.py:PresetManager.delete_preset, src/presets.py:PresetManager.get_preset_scope, src/presets.py:PresetManager.project_path]"""
return project_root / "project_presets.toml"
def get_global_tool_presets_path() -> Path:
"""Global tool presets file. Frozen at initialize_paths() time.
[C: src/tool_presets.py:ToolPresetManager._get_path, src/tool_presets.py:ToolPresetManager.load_all_bias_profiles, src/tool_presets.py:ToolPresetManager.load_all_presets]"""
return _cfg().tool_presets
def get_project_tool_presets_path(project_root: Path) -> Path:
"""[C: src/tool_presets.py:ToolPresetManager._get_path, src/tool_presets.py:ToolPresetManager.load_all_bias_profiles, src/tool_presets.py:ToolPresetManager.load_all_presets]"""
return project_root / "project_tool_presets.toml"
def get_global_personas_path() -> Path:
"""Global personas file. Frozen at initialize_paths() time.
[C: src/personas.py:PersonaManager._get_path, src/personas.py:PersonaManager.get_persona_scope, src/personas.py:PersonaManager.load_all]"""
return _cfg().personas
def get_project_personas_path(project_root: Path) -> Path:
"""[C: src/personas.py:PersonaManager._get_path, src/personas.py:PersonaManager.get_persona_scope, src/personas.py:PersonaManager.load_all]"""
return project_root / "project_personas.toml"
def get_global_themes_path() -> Path:
"""Global themes directory. Frozen at initialize_paths() time.
[C: src/theme_2.py:load_themes_from_disk]"""
return _cfg().themes
def get_project_themes_path(project_root: Path) -> Path:
"""[C: src/theme_2.py:load_themes_from_disk]"""
return project_root / "project_themes.toml"
def get_global_workspace_profiles_path() -> Path:
"""Global workspace profiles file. Frozen at initialize_paths() time.
[C: src/workspace_manager.py:WorkspaceManager._get_path, src/workspace_manager.py:WorkspaceManager.load_all_profiles]"""
return _cfg().workspace_profiles
def get_project_workspace_profiles_path(project_root: Path) -> Path:
"""[C: src/workspace_manager.py:WorkspaceManager._get_path, src/workspace_manager.py:WorkspaceManager.load_all_profiles]"""
return project_root / ".ai" / "workspace_profiles.toml"
def get_credentials_path() -> Path:
"""Global credentials file. Frozen at initialize_paths() time.
[C: src/mcp_client.py:_is_allowed]"""
return _cfg().credentials
def get_logs_dir() -> Path:
"""Logs directory (contains session subdirs). Frozen at initialize_paths() time.
[C: src/session_logger.py:close_session, src/session_logger.py:open_session, tests/test_paths.py:test_config_overrides, tests/test_paths.py:test_default_paths, tests/test_paths.py:test_env_var_overrides, tests/test_paths.py:test_precedence]"""
return _cfg().logs_dir
def get_scripts_dir() -> Path:
"""Generated scripts directory. Frozen at initialize_paths() time.
[C: src/session_logger.py:log_tool_call, src/session_logger.py:open_session, tests/test_paths.py:test_config_overrides, tests/test_paths.py:test_default_paths]"""
return _cfg().scripts_dir
def get_tracks_dir(project_path: Optional[str] = None) -> Path:
"""[C: src/project_manager.py:get_all_tracks, tests/test_paths.py:test_conductor_dir_project_relative]"""
return get_conductor_dir(project_path) / "tracks"
def get_track_state_dir(track_id: str, project_path: Optional[str] = None) -> Path:
"""[C: src/project_manager.py:load_track_state, src/project_manager.py:save_track_state, tests/test_paths.py:test_conductor_dir_project_relative]"""
return get_tracks_dir(project_path) / track_id
def get_archive_dir(project_path: Optional[str] = None) -> Path:
"""[C: tests/test_paths.py:test_conductor_dir_project_relative]"""
return get_conductor_dir(project_path) / "archive"
def _get_project_conductor_dir_from_toml(project_root: Path) -> Optional[Path]:
"""Look for manual_slop.toml in project_root for [conductor] dir override."""
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)
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.resolve()
except: pass
return None
def get_conductor_dir(project_path: Optional[str] = None) -> Path:
"""[C: tests/test_paths.py:test_conductor_dir_project_relative, tests/test_project_paths.py:test_get_conductor_dir_default, tests/test_project_paths.py:test_get_conductor_dir_project_specific_with_toml]"""
if not project_path:
return Path('conductor').resolve()
project_root = Path(project_path).resolve()
p = _get_project_conductor_dir_from_toml(project_root)
if p: return p
return (project_root / "conductor").resolve()
def get_full_path_info() -> dict[str, dict[str, Any]]:
"""Return the resolved paths + their source (env / config / default).
For diagnostic UIs (e.g., the Session Hub's "show resolved paths" panel).
[C: src/gui_2.py:App._render_path_field]"""
cfg = _cfg()
def info(value: Path) -> dict[str, Any]:
return {'path': str(value), 'source': 'frozen_at_init'}
return {
'config_path': info(cfg.config_path),
'presets': info(cfg.presets),
'tool_presets': info(cfg.tool_presets),
'personas': info(cfg.personas),
'themes': info(cfg.themes),
'workspace_profiles': info(cfg.workspace_profiles),
'credentials': info(cfg.credentials),
'logs_dir': info(cfg.logs_dir),
'scripts_dir': info(cfg.scripts_dir),
}
def reset_paths() -> None:
"""Clear the singleton. FOR TESTS ONLY — production code should never
call this. After reset, the next path getter raises RuntimeError until
initialize_paths() is called again.
[C: tests/conftest.py:reset_paths, tests/test_paths.py:reset_paths,
tests/test_app_controller_offloading.py:setup_function,
tests/test_gui_phase3.py:setup]"""
global _PATHS_CONFIG
with _PATHS_LOCK:
_PATHS_CONFIG = None