docs(conductor): Synchronize docs for track 'Project-Specific Conductor Directory'

conductor/product.md

@@ -33,7 +33,7 @@ For deep implementation details when planning or implementing tracks, consult `d
     - **Track Browser:** Real-time visualization of all implementation tracks with status indicators and progress bars. Includes a dedicated **Active Track Summary** featuring a color-coded progress bar, precise ticket status breakdown (Completed, In Progress, Blocked, Todo), and dynamic **ETA estimation** based on historical completion times.
     - **Visual Task DAG:** An interactive, node-based visualizer for the active track's task dependencies using `imgui-node-editor`. Features color-coded state tracking (Ready, Running, Blocked, Done), drag-and-drop dependency creation, and right-click deletion.
     - **Strategy Visualization:** Dedicated real-time output streams for Tier 1 (Strategic Planning) and Tier 2/3 (Execution) agents, allowing the user to follow the agent's reasoning chains alongside the task DAG.
-  - **Track-Scoped State Management:** Segregates discussion history and task progress into per-track state files (e.g., `conductor/tracks/<track_id>/state.toml`). This prevents global context pollution and ensures the Tech Lead session is isolated to the specific track's objective.
+  - **Track-Scoped State Management:** Segregates discussion history and task progress into per-track state files. Supports **Project-Specific Conductor Directories**, allowing projects to define their own conductor path in `manual_slop.toml` (`[conductor].dir`) for isolated track management. This prevents global context pollution and ensures the Tech Lead session is isolated to the specific track's objective.
   **Native DAG Execution Engine:** Employs a Python-based Directed Acyclic Graph (DAG) engine to manage complex task dependencies. Supports automated topological sorting, robust cycle detection, and **transitive blocking propagation** (cascading `blocked` status to downstream dependents to prevent execution stalls).
 
   - **Programmable Execution State machine:** Governing the transition between "Auto-Queue" (autonomous worker spawning) and "Step Mode" (explicit manual approval for each task transition).

conductor/tech-stack.md

@@ -30,7 +30,7 @@
 
 - **ai_style_formatter.py:** Custom Python formatter specifically designed to enforce 1-space indentation and ultra-compact whitespace to minimize token consumption.
 
-- **src/paths.py:** Centralized module for path resolution, allowing directory paths (logs, conductor, scripts) to be configured via `config.toml` or environment variables, eliminating hardcoded filesystem dependencies.
+- **src/paths.py:** Centralized module for path resolution. Supports project-specific conductor directory overrides via project TOML (`[conductor].dir`), enabling isolated track management per project. All paths are resolved to absolute objects. Path configuration (logs, conductor, scripts) can also be configured via `config.toml` or environment variables, eliminating hardcoded filesystem dependencies.
 
 - **src/presets.py:** Implements `PresetManager` for high-performance CRUD operations on system prompt presets stored in TOML format (`presets.toml`, `project_presets.toml`). Supports dynamic path resolution and scope-based inheritance.
 

src/app_controller.py

@@ -430,6 +430,12 @@ class AppController:
   if hasattr(self, 'perf_monitor'):
    self.perf_monitor.enabled = value
 
+ @property
+ def active_project_root(self) -> str:
+  if self.active_project_path:
+   return str(Path(self.active_project_path).parent)
+  return self.ui_files_base_dir
+
  def _update_inject_preview(self) -> None:
   """Updates the preview content based on the selected file and injection mode."""
   if not self._inject_file_path:
@@ -1859,7 +1865,7 @@ class AppController:
   agent_tools_cfg = proj.get("agent", {}).get("tools", {})
   self.ui_agent_tools = {t: agent_tools_cfg.get(t, True) for t in models.AGENT_TOOL_NAMES}
   # MMA Tracks
-  self.tracks = project_manager.get_all_tracks(self.ui_files_base_dir)
+  self.tracks = project_manager.get_all_tracks(self.active_project_root)
   # Restore MMA state
   mma_sec = proj.get("mma", {})
   self.ui_epic_input = mma_sec.get("epic", "")
@@ -1889,7 +1895,7 @@ class AppController:
    self.active_tickets = []
    # Load track-scoped history if track is active
   if self.active_track:
-   track_history = project_manager.load_track_history(self.active_track.id, self.ui_files_base_dir)
+   track_history = project_manager.load_track_history(self.active_track.id, self.active_project_root)
    if track_history:
     with self._disc_entries_lock:
      self.disc_entries = models.parse_history_entries(track_history, self.disc_roles)
@@ -1962,7 +1968,7 @@ class AppController:
 
 
  def _cb_load_track(self, track_id: str) -> None:
-  state = project_manager.load_track_state(track_id, self.ui_files_base_dir)
+  state = project_manager.load_track_state(track_id, self.active_project_root)
   if state:
    try:
    # Convert list[Ticket] or list[dict] to list[Ticket] for Track object
@@ -1980,7 +1986,7 @@ class AppController:
     # Keep dicts for UI table (or convert models.Ticket objects back to dicts if needed)
     self.active_tickets = [asdict(t) if not isinstance(t, dict) else t for t in tickets]
     # Load track-scoped history
-    history = project_manager.load_track_history(track_id, self.ui_files_base_dir)
+    history = project_manager.load_track_history(track_id, self.active_project_root)
     with self._disc_entries_lock:
      if history:
       self.disc_entries = models.parse_history_entries(history, self.disc_roles)
@@ -2650,7 +2656,7 @@ class AppController:
   if not name: return
   date_suffix = datetime.now().strftime("%Y%m%d")
   track_id = f"{name.lower().replace(' ', '_')}_{date_suffix}"
-  track_dir = paths.get_tracks_dir() / track_id
+  track_dir = paths.get_track_state_dir(track_id, project_path=self.active_project_root)
   track_dir.mkdir(parents=True, exist_ok=True)
   spec_file = track_dir / "spec.md"
   with open(spec_file, "w", encoding="utf-8") as f:
@@ -2669,7 +2675,7 @@ class AppController:
      "progress": 0.0
     }, f, indent=1)
    # Refresh tracks from disk
-  self.tracks = project_manager.get_all_tracks(self.ui_files_base_dir)
+  self.tracks = project_manager.get_all_tracks(self.active_project_root)
 
  def _push_mma_state_update(self) -> None:
   if not self.active_track:

src/paths.py

@@ -18,17 +18,18 @@ Configuration (config.toml):
 
 Path Functions:
  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_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:
- 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:
  from src.paths import get_logs_dir, get_scripts_dir
@@ -51,9 +52,11 @@ _RESOLVED: dict[str, Path] = {}
 def get_config_path() -> Path:
  root_dir = Path(__file__).resolve().parent.parent
  return Path(os.environ.get("SLOP_CONFIG", root_dir / "config.toml"))
+
 def get_global_presets_path() -> Path:
  root_dir = Path(__file__).resolve().parent.parent
  return Path(os.environ.get("SLOP_GLOBAL_PRESETS", root_dir / "presets.toml"))
+
 def get_project_presets_path(project_root: Path) -> Path:
  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"
 
 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:
-  return Path(os.environ[env_var])
+  p = Path(os.environ[env_var])
- try:
+ else:
-  with open(get_config_path(), "rb") as f:
+  try:
-   cfg = tomllib.load(f)
+   with open(get_config_path(), "rb") as f:
-  if "paths" in cfg and config_key in cfg["paths"]:
+    cfg = tomllib.load(f)
-   return Path(cfg["paths"][config_key])
+   if "paths" in cfg and config_key in cfg["paths"]:
- except FileNotFoundError:
+    p = Path(cfg["paths"][config_key])
-  pass
+  except (FileNotFoundError, tomllib.TOMLDecodeError):
- return Path(default)
+   pass
+ if p is None:
+  p = Path(default)
+ if not p.is_absolute():
+  return root_dir / p
+ return p
 
-def get_conductor_dir() -> Path:
+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'
+ 
  if "conductor_dir" not in _RESOLVED:
   _RESOLVED["conductor_dir"] = _resolve_path("SLOP_CONDUCTOR_DIR", "conductor_dir", "conductor")
  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")
  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:
  """For testing only - clear cached resolutions."""
  _RESOLVED.clear()
-

src/project_manager.py

@@ -245,7 +245,7 @@ def save_track_state(track_id: str, state: 'TrackState', base_dir: Union[str, Pa
  """
     Saves a TrackState object to conductor/tracks/<track_id>/state.toml.
     """
- track_dir = Path(base_dir) / paths.get_track_state_dir(track_id)
+ track_dir = paths.get_track_state_dir(track_id, project_path=str(base_dir))
  track_dir.mkdir(parents=True, exist_ok=True)
  state_file = track_dir / "state.toml"
  data = clean_nones(state.to_dict())
@@ -257,7 +257,7 @@ def load_track_state(track_id: str, base_dir: Union[str, Path] = ".") -> Optiona
     Loads a TrackState object from conductor/tracks/<track_id>/state.toml.
     """
  from src.models import TrackState
- state_file = Path(base_dir) / paths.get_track_state_dir(track_id) / "state.toml"
+ state_file = paths.get_track_state_dir(track_id, project_path=str(base_dir)) / 'state.toml'
  if not state_file.exists():
   return None
  with open(state_file, "rb") as f:
@@ -302,7 +302,7 @@ def get_all_tracks(base_dir: Union[str, Path] = ".") -> list[dict[str, Any]]:
     Handles missing or malformed metadata.json or state.toml by falling back
     to available info or defaults.
     """
- tracks_dir = Path(base_dir) / paths.get_tracks_dir()
+ tracks_dir = paths.get_tracks_dir(project_path=str(base_dir))
  if not tracks_dir.exists():
   return []
  results: list[dict[str, Any]] = []

tests/test_project_paths.py

@@ -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"