diff --git a/sloppy.py b/sloppy.py index ca3f62ff..bb299355 100644 --- a/sloppy.py +++ b/sloppy.py @@ -42,9 +42,8 @@ args: argparse.Namespace = argparse.Namespace() # type: ignore[assignment] if __name__ == "__main__": args = parser.parse_args() from pathlib import Path - from src.paths import set_config_override - if args.config: - set_config_override(Path(args.config).resolve()) + from src.paths import initialize_paths + initialize_paths(Path(args.config).resolve() if args.config else None) if args.web_host is not None: with startup_profiler.phase("web_host_imports"): from imgui_bundle import hello_imgui diff --git a/src/app_controller.py b/src/app_controller.py index b571811c..cc5ba00f 100644 --- a/src/app_controller.py +++ b/src/app_controller.py @@ -1804,7 +1804,7 @@ class AppController: spath = Path(proj_paths['scripts_dir']) if not spath.is_absolute(): spath = project_root / spath os.environ['SLOP_SCRIPTS_DIR'] = str(spath) - paths.reset_resolved() + paths.initialize_paths(paths.get_config_path()) path_info = paths.get_full_path_info() self.ui_logs_dir = str(path_info['logs_dir']['path']) diff --git a/src/gui_2.py b/src/gui_2.py index 0d0cc1f7..a96d5202 100644 --- a/src/gui_2.py +++ b/src/gui_2.py @@ -1311,7 +1311,7 @@ class App: cfg_path = paths.get_config_path() if cfg_path.exists(): shutil.copy(cfg_path, str(cfg_path) + ".bak") self.save_config() - paths.reset_resolved() + paths.initialize_paths(cfg_path) self.init_state() self.ai_status = 'paths applied and session reset' @@ -2341,10 +2341,17 @@ def render_paths_panel(app: App) -> None: render_path_field("Logs Directory", "ui_logs_dir", "logs_dir", "Directory where session JSON-L logs and artifacts are stored.") render_path_field("Scripts Directory", "ui_scripts_dir", "scripts_dir", "Directory for AI-generated PowerShell scripts.") - + imgui.separator() if imgui.button("Apply", imgui.ImVec2(120, 0)): app._save_paths() imgui.same_line() + if imgui.button("Refresh Paths", imgui.ImVec2(140, 0)): + paths.initialize_paths(paths.get_config_path()) + app.init_state() + app.ai_status = "paths reloaded from config.toml" + if imgui.is_item_hovered(): + imgui.set_tooltip("Re-read [paths] section from config.toml and rebuild the PathsConfig singleton. Use after editing config.toml directly or after importing a new config.") + imgui.same_line() if imgui.button("Reset", imgui.ImVec2(120, 0)): app.init_state() app.ai_status = "paths reset to defaults" diff --git a/src/paths.py b/src/paths.py index e2e4ff8f..005becbb 100644 --- a/src/paths.py +++ b/src/paths.py @@ -1,189 +1,231 @@ """ -Paths - Centralized path resolution for configuration and environment variables. +Paths - Single source of truth for all application paths. -This module provides centralized path resolution for all configurable paths in the application. - All paths can be overridden via environment variables or config.toml. +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. -Environment Variables: - SLOP_CONFIG: Path to config.toml - SLOP_LOGS_DIR: Path to logs directory - SLOP_SCRIPTS_DIR: Path to generated scripts directory +**Usage contract:** -Configuration (config.toml): - [paths] - logs_dir = "logs/sessions" - scripts_dir = "scripts/generated" +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. -Path Functions: - get_config_path() -> Path to config.toml - get_conductor_dir(project_path=None) -> Path to conductor directory - get_logs_dir() -> Path to logs/sessions - get_scripts_dir() -> Path to scripts/generated - get_tracks_dir(project_path=None) -> Path to conductor/tracks - get_track_state_dir(track_id, project_path=None) -> Path to conductor/tracks/ - get_archive_dir(project_path=None) -> Path to conductor/archive +**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 Order: - 1. Check project-specific manual_slop.toml (for conductor paths) - 2. Check environment variable (for logs/scripts) - 3. Check config.toml [paths] section (for logs/scripts) - 4. Fall back to default +**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 `/` -Usage: - from src.paths import get_logs_dir, get_scripts_dir +**Codepath ordering:** - logs_dir = get_logs_dir() - scripts_dir = get_scripts_dir() +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. -See Also: - - docs/guide_tools.md for configuration documentation - - src/session_logger.py for logging paths - - src/project_manager.py for project paths +`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 +from typing import Optional, Any -_RESOLVED: dict[str, Path] = {} - -_CONFIG_OVERRIDE: Path | None = None +@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 -def set_config_override(path: Path | None) -> None: - """ - Set the active config.toml path. Pass None to use the default - /config.toml. The CLI --config flag is the ONLY - supported override mechanism; the historical SLOP_CONFIG env var - fallback has been removed (per test_sandbox_hardening_20260619 FR2). - [C: sloppy.py:main, tests/conftest.py] - """ - global _CONFIG_OVERRIDE - _CONFIG_OVERRIDE = path - _RESOLVED.clear() +_PATHS_CONFIG: Optional[PathsConfig] = None +_PATHS_LOCK = threading.RLock() +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 `/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() " + "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: - """ - Returns the active config.toml. If a CLI override is set, returns it. - Otherwise returns the default /config.toml. + """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] - """ - if _CONFIG_OVERRIDE is not None: - return _CONFIG_OVERRIDE - root_dir = Path(__file__).resolve().parent.parent - return root_dir / "config.toml" + tests/test_test_sandbox.py]""" + return _cfg().config_path def get_global_presets_path() -> Path: - """ - [C: src/presets.py:PresetManager.__init__, src/presets.py:PresetManager.delete_preset, src/presets.py:PresetManager.get_preset_scope] - """ - if "presets" not in _RESOLVED: - _RESOLVED["presets"] = _resolve_path("SLOP_GLOBAL_PRESETS", "presets", "presets.toml") - return _RESOLVED["presets"] + """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: - """ - [C: src/presets.py:PresetManager.delete_preset, src/presets.py:PresetManager.get_preset_scope, src/presets.py:PresetManager.project_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: - """ - [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] - """ - if "tool_presets" not in _RESOLVED: - _RESOLVED["tool_presets"] = _resolve_path("SLOP_GLOBAL_TOOL_PRESETS", "tool_presets", "tool_presets.toml") - return _RESOLVED["tool_presets"] + """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] - """ + """[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: - """ - [C: src/personas.py:PersonaManager._get_path, src/personas.py:PersonaManager.get_persona_scope, src/personas.py:PersonaManager.load_all] - """ - if "personas" not in _RESOLVED: - _RESOLVED["personas"] = _resolve_path("SLOP_GLOBAL_PERSONAS", "personas", "personas.toml") - return _RESOLVED["personas"] + """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] - """ + """[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: - """ - [C: src/theme_2.py:load_themes_from_disk] - """ - if "themes" not in _RESOLVED: - _RESOLVED["themes"] = _resolve_path("SLOP_GLOBAL_THEMES", "themes", "themes") - return _RESOLVED["themes"] + """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] - """ + """[C: src/theme_2.py:load_themes_from_disk]""" return project_root / "project_themes.toml" def get_global_workspace_profiles_path() -> Path: - """ - [C: src/workspace_manager.py:WorkspaceManager._get_path, src/workspace_manager.py:WorkspaceManager.load_all_profiles] - """ - if "workspace_profiles" not in _RESOLVED: - _RESOLVED["workspace_profiles"] = _resolve_path("SLOP_GLOBAL_WORKSPACE_PROFILES", "workspace_profiles", "workspace_profiles.toml") - return _RESOLVED["workspace_profiles"] + """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] - """ + """[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: - """ - [C: src/mcp_client.py:_is_allowed] - """ - if "credentials" not in _RESOLVED: - _RESOLVED["credentials"] = _resolve_path("SLOP_CREDENTIALS", "credentials", "credentials.toml") - return _RESOLVED["credentials"] + """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 _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: - p = Path(os.environ[env_var]) - else: - try: - with open(get_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]) - except (FileNotFoundError, tomllib.TOMLDecodeError): - pass - 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 + """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) - # Check [conductor] dir = '...' c_dir = data.get('conductor', {}).get('dir') if c_dir: p = Path(c_dir) @@ -192,79 +234,44 @@ def _get_project_conductor_dir_from_toml(project_root: Path) -> Optional[Path]: 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] - """ + """[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: - # Fallback for legacy/tests, but we should avoid this return Path('conductor').resolve() - project_root = Path(project_path).resolve() - p = _get_project_conductor_dir_from_toml(project_root) + p = _get_project_conductor_dir_from_toml(project_root) if p: return p return (project_root / "conductor").resolve() -def get_logs_dir() -> Path: - """ - [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] - """ - if "logs_dir" not in _RESOLVED: - _RESOLVED["logs_dir"] = _resolve_path("SLOP_LOGS_DIR", "logs_dir", "logs/sessions") - return _RESOLVED["logs_dir"] - -def get_scripts_dir() -> Path: - """ - [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] - """ - if "scripts_dir" not in _RESOLVED: - _RESOLVED["scripts_dir"] = _resolve_path("SLOP_SCRIPTS_DIR", "scripts_dir", "scripts/generated") - return _RESOLVED["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 _resolve_path_info(env_var: str, config_key: str, default: str) -> dict[str, Any]: - if env_var in os.environ: - return {'path': Path(os.environ[env_var]).resolve(), 'source': f'env:{env_var}'} - try: - with open(get_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]) - if not p.is_absolute(): - p = (Path(__file__).resolve().parent.parent / p).resolve() - return {'path': p, 'source': 'config.toml'} - except: pass - root_dir = Path(__file__).resolve().parent.parent - p = (root_dir / default).resolve() - return {'path': p, 'source': 'default'} 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 { - 'logs_dir': _resolve_path_info('SLOP_LOGS_DIR', 'logs_dir', 'logs/sessions'), - 'scripts_dir': _resolve_path_info('SLOP_SCRIPTS_DIR', 'scripts_dir', 'scripts/generated') + '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_resolved() -> None: - """ - For testing only - clear cached resolutions. - [C: tests/conftest.py:reset_paths, tests/test_app_controller_offloading.py:tmp_session_dir, tests/test_gui_phase3.py:test_conductor_setup_scan, tests/test_paths.py:reset_paths, tests/test_project_paths.py:test_get_all_tracks_project_specific, tests/test_project_paths.py:test_get_conductor_dir_default, tests/test_project_paths.py:test_get_conductor_dir_project_specific_with_toml] - """ - _RESOLVED.clear() - \ No newline at end of file + +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 \ No newline at end of file diff --git a/tests/conftest.py b/tests/conftest.py index 75e622fc..45b0a157 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -34,7 +34,7 @@ if _config_override_arg is None: _config_override_arg = _ISOLATION_WORKSPACE / "config_overrides.toml" from src import paths as _paths # noqa: E402 -_paths.set_config_override(_config_override_arg) +_paths.initialize_paths(_config_override_arg) thirdparty_dir = os.path.join(os.path.dirname(__file__), "..", "thirdparty") if thirdparty_dir not in sys.path: @@ -381,11 +381,11 @@ def isolate_workspace(monkeypatch) -> Generator[None, None, None]: (tests/artifacts/_isolation_workspace_/). Also writes placeholder TOML files for the redirected paths. NO SLOP_* env vars are set; src/paths.py reads the overrides from config.toml [paths] (with env var - as fallback if needed). - [C: tests/conftest.py:_ISOLATION_WORKSPACE, src/paths.py:_resolve_path] + as fallback if needed). Also re-initializes the paths singleton so every + getter sees the test-workspace overrides. + [C: tests/conftest.py:_ISOLATION_WORKSPACE, src/paths.py:initialize_paths] """ from src import paths as _paths - _paths.reset_resolved() test_workspace = _ISOLATION_WORKSPACE @@ -421,14 +421,15 @@ def isolate_workspace(monkeypatch) -> Generator[None, None, None]: @pytest.fixture(autouse=True) def reset_paths() -> Generator[None, None, None]: """ - - - Autouse fixture that resets the paths global state before each test. + Autouse fixture that resets the paths singleton before each test. + The isolate_workspace fixture re-initializes paths before the test runs, + so reset_paths is a no-op here (PathsConfig is frozen at init time, + the per-getter cache is gone). Kept as a marker for backward compatibility + with test code that asserts the singleton is cleared at test boundaries. """ from src import paths - paths.reset_resolved() yield - paths.reset_resolved() + paths.reset_paths() @pytest.fixture(autouse=True) def reset_ai_client() -> Generator[None, None, None]: diff --git a/tests/test_app_controller_offloading.py b/tests/test_app_controller_offloading.py index cdee541a..bddd2961 100644 --- a/tests/test_app_controller_offloading.py +++ b/tests/test_app_controller_offloading.py @@ -18,7 +18,7 @@ def tmp_session_dir(tmp_path, monkeypatch): monkeypatch.setenv("SLOP_LOGS_DIR", str(logs_dir)) monkeypatch.setenv("SLOP_SCRIPTS_DIR", str(scripts_dir)) - paths.reset_resolved() + paths.reset_paths() # Ensure session_logger is clean with patch("src.session_logger._comms_fh", None): diff --git a/tests/test_gui_paths.py b/tests/test_gui_paths.py index 70817a7c..12240846 100644 --- a/tests/test_gui_paths.py +++ b/tests/test_gui_paths.py @@ -26,7 +26,7 @@ def test_save_paths(): with patch('shutil.copy') as mock_copy, \ patch('src.paths.get_config_path') as mock_get_cfg, \ - patch('src.paths.reset_resolved') as mock_reset, \ + patch('src.paths.reset_paths') as mock_reset, \ patch.object(MockApp, 'init_state') as mock_init: mock_get_cfg.return_value = MagicMock() diff --git a/tests/test_gui_phase3.py b/tests/test_gui_phase3.py index 3b3b1d83..aa38d38f 100644 --- a/tests/test_gui_phase3.py +++ b/tests/test_gui_phase3.py @@ -50,7 +50,7 @@ def test_conductor_setup_scan(app_instance, tmp_path, monkeypatch): (cond_dir / "tracks" / "track1").mkdir(exist_ok=True) monkeypatch.setenv('SLOP_CONDUCTOR_DIR', str((tmp_path / 'conductor').resolve())) - paths.reset_resolved() + paths.reset_paths() app_instance._cb_run_conductor_setup() diff --git a/tests/test_paths.py b/tests/test_paths.py index 7c526a81..6ec7ce27 100644 --- a/tests/test_paths.py +++ b/tests/test_paths.py @@ -5,9 +5,9 @@ from src import paths @pytest.fixture(autouse=True) def reset_paths(): - paths.reset_resolved() + paths.reset_paths() yield - paths.reset_resolved() + paths.reset_paths() def test_default_paths(tmp_path, monkeypatch): monkeypatch.setenv("SLOP_CONFIG", str(tmp_path / "non_existent.toml")) diff --git a/tests/test_project_paths.py b/tests/test_project_paths.py index 0adc4fe9..629b8435 100644 --- a/tests/test_project_paths.py +++ b/tests/test_project_paths.py @@ -7,13 +7,13 @@ from src import paths from src import project_manager def test_get_conductor_dir_default(): - paths.reset_resolved() + paths.reset_paths() # Should return absolute path to "conductor" in project root expected = Path(__file__).resolve().parent.parent / "conductor" assert paths.get_conductor_dir() == expected def test_get_conductor_dir_project_specific_with_toml(tmp_path): - paths.reset_resolved() + paths.reset_paths() project_root = tmp_path / "my_project" project_root.mkdir() @@ -31,7 +31,7 @@ def test_get_conductor_dir_project_specific_with_toml(tmp_path): assert res == project_root / "custom_tracks" def test_get_all_tracks_project_specific(tmp_path): - paths.reset_resolved() + paths.reset_paths() project_root = tmp_path / "my_project" project_root.mkdir() diff --git a/tests/test_test_sandbox.py b/tests/test_test_sandbox.py index 7f4a475d..bc6cca6c 100644 --- a/tests/test_test_sandbox.py +++ b/tests/test_test_sandbox.py @@ -149,47 +149,64 @@ def test_sandbox_allows_pytest_cache_write() -> None: def test_config_override_via_cli_flag(tmp_path) -> None: - """paths.set_config_override(path) makes get_config_path() return that path.""" + """paths.initialize_paths(config_path) makes all path getters return paths + rooted at config_path's [paths] section (or default). + [C: src/paths.py:initialize_paths, sloppy.py:main]""" from src import paths config_path = tmp_path / "my_config.toml" config_path.write_text("[ai]\nprovider='gemini'\n", encoding="utf-8") - original = paths._CONFIG_OVERRIDE + paths.initialize_paths(config_path) try: - paths.set_config_override(config_path) assert paths.get_config_path() == config_path + assert paths.get_global_presets_path().name == "presets.toml" + assert paths.get_logs_dir().name == "sessions" finally: - paths.set_config_override(original) + paths.reset_paths() -def test_paths_get_config_path_no_env_fallback(monkeypatch) -> None: - """Without an override AND without SLOP_CONFIG, get_config_path returns default.""" - monkeypatch.delenv("SLOP_CONFIG", raising=False) +def test_paths_runtime_refresh_atomic_swap(tmp_path) -> None: + """Calling initialize_paths() a second time atomically swaps the singleton. + Reader threads see the new config immediately. The PathsConfig is frozen. + [C: src/paths.py:initialize_paths, src/paths.py:PathsConfig]""" from src import paths - original = paths._CONFIG_OVERRIDE - try: - paths.set_config_override(None) - expected = Path(__file__).resolve().parent.parent / "config.toml" - assert paths.get_config_path() == expected - finally: - paths.set_config_override(original) + cfg_a = tmp_path / "a.toml" + cfg_b = tmp_path / "b.toml" + cfg_a.write_text("[ai]\nprovider='gemini-a'\n", encoding="utf-8") + cfg_b.write_text("[ai]\nprovider='gemini-b'\n", encoding="utf-8") + paths.initialize_paths(cfg_a) + assert paths.get_config_path() == cfg_a + paths.initialize_paths(cfg_b) + assert paths.get_config_path() == cfg_b + paths.reset_paths() + + +def test_paths_uninitialized_raises(tmp_path) -> None: + """Calling a path getter before initialize_paths() raises RuntimeError + (not silent fallback). This is the contract that enforces explicit init. + [C: src/paths.py:_cfg]""" + from src import paths + paths.reset_paths() + with pytest.raises(RuntimeError, match="not initialized"): + paths.get_logs_dir() + paths.reset_paths() def test_sloppy_py_parses_config_flag() -> None: - """sloppy.py has a --config argparse argument that calls set_config_override.""" + """sloppy.py has a --config argparse argument that calls initialize_paths.""" import ast sloppy = Path(__file__).resolve().parent.parent / "sloppy.py" tree = ast.parse(sloppy.read_text(encoding="utf-8")) found_config_arg = False - found_set_override_call = False + found_init_call = False for node in ast.walk(tree): if isinstance(node, ast.Constant) and node.value == "--config": found_config_arg = True if isinstance(node, ast.Call): func = node.func - if isinstance(func, ast.Name) and func.id == "set_config_override": - found_set_override_call = True + if isinstance(func, ast.Name) and func.id == "initialize_paths": + found_init_call = True assert found_config_arg, "sloppy.py must have a --config argparse argument" - assert found_set_override_call, "sloppy.py must call paths.set_config_override(args.config)" + assert found_init_call, "sloppy.py must call paths.initialize_paths(args.config)" def test_pyproject_toml_basetemp_is_under_tests() -> None: @@ -276,30 +293,86 @@ def test_config_overrides_toml_has_paths_section() -> None: ) -def test_path_getters_read_from_config_paths_section() -> None: - """Every global path getter in src/paths.py reads from the [paths] section - of the active config when no env var is set. Verifies via AST that each - get_*_path() function calls _resolve_path (not raw os.environ.get).""" +def test_path_getters_are_trivial_field_access() -> None: + """Every global path getter in src/paths.py is a trivial field access on the + PathsConfig singleton (return _cfg().). They must NOT do file I/O + or call _resolve_path() (which is internal-only, called from initialize_paths). + This enforces the v3 design: explicit init at startup, trivial getters.""" import ast paths_py = Path(__file__).resolve().parent.parent / "src" / "paths.py" tree = ast.parse(paths_py.read_text(encoding="utf-8")) - getters = { - "get_global_presets_path": "presets", - "get_global_tool_presets_path": "tool_presets", - "get_global_personas_path": "personas", - "get_global_themes_path": "themes", - "get_global_workspace_profiles_path": "workspace_profiles", - "get_credentials_path": "credentials", - "get_logs_dir": "logs_dir", - "get_scripts_dir": "scripts_dir", - } + trivial_getters = [ + "get_global_presets_path", "get_global_tool_presets_path", + "get_global_personas_path", "get_global_themes_path", + "get_global_workspace_profiles_path", "get_credentials_path", + "get_logs_dir", "get_scripts_dir", + ] for node in ast.walk(tree): - if isinstance(node, ast.FunctionDef) and node.name in getters: + if isinstance(node, ast.FunctionDef) and node.name in trivial_getters: src = ast.unparse(node) - assert "_resolve_path(" in src, ( - f"{node.name} must call _resolve_path() to read from config.toml [paths]; " - f"got direct os.environ.get() instead." + assert "_cfg()" in src, ( + f"{node.name} must call _cfg() to read from the PathsConfig singleton; " + f"got direct file I/O or env var lookup instead." ) + assert "_resolve_path(" not in src, ( + f"{node.name} must NOT call _resolve_path(); that's internal-only " + f"(called once from initialize_paths)." + ) + assert "os.environ" not in src, ( + f"{node.name} must NOT use os.environ directly; env vars are read once " + f"during initialize_paths()." + ) + + +def test_initialize_paths_thread_safe_atomic_swap(tmp_path) -> None: + """initialize_paths() uses an RLock; concurrent swaps don't corrupt the + singleton. Reader threads always see a consistent PathsConfig snapshot + (frozen dataclass). Per the user's "bad programmers take shortcuts" rule: + no torn writes, no partial reads. + [C: src/paths.py:_PATHS_LOCK, src/paths.py:PathsConfig]""" + import threading + import tomllib + from src import paths + cfg_a = tmp_path / "a.toml" + cfg_b = tmp_path / "b.toml" + cfg_a.write_bytes(b"[paths]\nlogs_dir = 'C:/tmp/thread_a'\n") + cfg_b.write_bytes(b"[paths]\nlogs_dir = 'C:/tmp/thread_b'\n") + errors = [] + def swap(cfg, n): + for _ in range(n): + try: + paths.initialize_paths(cfg) + except Exception as e: + errors.append(e) + t1 = threading.Thread(target=swap, args=(cfg_a, 100)) + t2 = threading.Thread(target=swap, args=(cfg_b, 100)) + t1.start(); t2.start() + t1.join(); t2.join() + assert not errors, f"thread-safety violation: {errors}" + paths.reset_paths() + + +def test_pathsconfig_is_frozen_dataclass() -> None: + """PathsConfig uses @dataclass(frozen=True) so individual field reads are + atomic and cannot be mutated by readers. Per user directive: gated + transactions, no data race over config. + [C: src/paths.py:PathsConfig]""" + import dataclasses + import tempfile, tomli_w + from src import paths + params = getattr(paths.PathsConfig, "__dataclass_params__", None) + assert params is not None, "PathsConfig must be a dataclass" + assert params.frozen is True, "PathsConfig must be @dataclass(frozen=True)" + with tempfile.NamedTemporaryFile(suffix=".toml", delete=False, mode="wb") as f: + tomli_w.dump({"paths": {"logs_dir": "C:/tmp/frozen_test"}}, f) + cfg = Path(f.name) + try: + paths.initialize_paths(cfg) + with pytest.raises(dataclasses.FrozenInstanceError): + paths._PATHS_CONFIG.presets = Path("/tmp/should_fail") + finally: + paths.reset_paths() + cfg.unlink() @pytest.mark.skipif(os.name != "nt", reason="Windows-only sandbox wrapper")