Unify workspace root resolution across services

backend/api/podcast/constants.py

@@ -10,9 +10,7 @@ from pathlib import Path
 from typing import Literal
 from loguru import logger
 from services.story_writer.audio_generation_service import StoryAudioGenerationService
-from utils.storage_paths import get_repo_root, sanitize_user_id as _sanitize_user_id
+from services.workspace_paths import get_workspace_root, get_user_workspace_dir
-
-ROOT_DIR = get_repo_root()
 
 # Video subdirectory (relative to workspace media dir)
 AI_VIDEO_SUBDIR = Path("AI_Videos")
@@ -45,15 +43,10 @@ def get_podcast_media_dir(
     }[media_type]
 
     if user_id:
-        sanitized = _sanitize_user_id(user_id)
+        resolved_dir = (get_user_workspace_dir(user_id) / "media" / media_subdir).resolve()
-        resolved_dir = (
-            ROOT_DIR / "workspace" / f"workspace_{sanitized}" / "media" / media_subdir
-        ).resolve()
     else:
         logger.warning(f"[Podcast] get_podcast_media_dir called without user_id for {media_type} — using default workspace. This should not happen in production.")
-        resolved_dir = (
+        resolved_dir = (get_workspace_root() / "workspace_alwrity" / "media" / media_subdir).resolve()
-            ROOT_DIR / "workspace" / "workspace_alwrity" / "media" / media_subdir
-        ).resolve()
 
     if ensure_exists:
         resolved_dir.mkdir(parents=True, exist_ok=True)

backend/services/database.py

@@ -46,10 +46,10 @@ import models.platform_insights_monitoring_models
 import models.agent_activity_models
 import models.daily_workflow_models
 
+from services.workspace_paths import get_workspace_root, get_user_workspace_dir
+
 # Database configuration
-# Get project root (3 levels up from services/database.py: services -> backend -> root)
+WORKSPACE_DIR = str(get_workspace_root())
-ROOT_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-WORKSPACE_DIR = os.path.join(ROOT_DIR, 'workspace')
 
 # Engine cache for multi-tenant support
 _user_engines = {}
@@ -95,7 +95,7 @@ def _sanitize_user_id(user_id: str) -> str:
 def ensure_user_workspace_db_directory(user_id: str) -> str:
     """Ensure modern `db/` directory exists, migrating legacy `database/` when safe."""
     safe_user_id = _sanitize_user_id(user_id)
-    user_workspace = os.path.join(WORKSPACE_DIR, f"workspace_{safe_user_id}")
+    user_workspace = str(get_user_workspace_dir(user_id))
     db_dir = os.path.join(user_workspace, 'db')
     legacy_db_dir = os.path.join(user_workspace, 'database')
 
@@ -126,7 +126,7 @@ def ensure_user_workspace_db_directory(user_id: str) -> str:
 def get_user_db_path(user_id: str) -> str:
     """Get the database path for a specific user."""
     safe_user_id = _sanitize_user_id(user_id)
-    user_workspace = os.path.join(WORKSPACE_DIR, f"workspace_{safe_user_id}")
+    user_workspace = str(get_user_workspace_dir(user_id))
     db_dir = ensure_user_workspace_db_directory(user_id)
     
     # Check for legacy naming convention first (to support existing data)

backend/services/user_workspace_manager.py

@@ -15,23 +15,16 @@ from sqlalchemy import text
 
 from services.database import init_user_database, ensure_user_workspace_db_directory
 from services.workspace_dirs import ensure_user_workspace_dirs
+from services.workspace_paths import get_workspace_root, get_user_workspace_dir
 
 class UserWorkspaceManager:
     """Manages user-specific workspaces and progressive setup."""
     
     def __init__(self, db_session: Session):
         self.db = db_session
-        # Use environment-safe paths for production
+        # Use shared workspace root authority for all environments.
-        if os.getenv("RENDER") or os.getenv("RAILWAY") or os.getenv("HEROKU"):
+        self.base_workspace_dir = get_workspace_root()
-            # In production, use temp directories or skip file operations
+        self.user_workspaces_dir = self.base_workspace_dir
-            self.base_workspace_dir = Path("/tmp/alwrity_workspace")
-            self.user_workspaces_dir = self.base_workspace_dir / "users"
-        else:
-            # In development, use project root 'workspace' directory
-            # services/user_workspace_manager.py -> services -> backend -> root
-            root_dir = Path(__file__).parent.parent.parent
-            self.base_workspace_dir = root_dir / "workspace"
-            self.user_workspaces_dir = self.base_workspace_dir
         
     def _sanitize_user_id(self, user_id: str) -> str:
         """Sanitize user_id to be safe for filesystem (matches database.py logic)."""
@@ -63,7 +56,7 @@ class UserWorkspaceManager:
             
             # Create user-specific directories
             # Format: workspaces/workspace_{user_id}
-            user_dir = self.user_workspaces_dir / f"workspace_{safe_user_id}"
+            user_dir = get_user_workspace_dir(user_id)
             user_dir.mkdir(parents=True, exist_ok=True)
 
             # Ensure canonical DB directory and migrate legacy layout if needed
@@ -161,7 +154,7 @@ class UserWorkspaceManager:
     def get_user_workspace(self, user_id: str) -> Optional[Dict[str, Any]]:
         """Get user workspace information."""
         safe_user_id = self._sanitize_user_id(user_id)
-        user_dir = self.user_workspaces_dir / f"workspace_{safe_user_id}"
+        user_dir = get_user_workspace_dir(user_id)
         
         if not user_dir.exists():
             return None
@@ -181,7 +174,7 @@ class UserWorkspaceManager:
         """Update user configuration."""
         try:
             safe_user_id = self._sanitize_user_id(user_id)
-            user_dir = self.user_workspaces_dir / f"workspace_{safe_user_id}"
+            user_dir = get_user_workspace_dir(user_id)
             config_file = user_dir / "config" / "user_config.json"
             
             if config_file.exists():
@@ -331,7 +324,7 @@ class UserWorkspaceManager:
         """Clean up user workspace (for account deletion)."""
         try:
             safe_user_id = self._sanitize_user_id(user_id)
-            user_dir = self.user_workspaces_dir / f"workspace_{safe_user_id}"
+            user_dir = get_user_workspace_dir(user_id)
             if user_dir.exists():
                 shutil.rmtree(user_dir)
             

backend/services/workspace_paths.py

@@ -0,0 +1,19 @@
+"""Shared workspace path helpers.
+
+Single authority for workspace root and per-user workspace paths.
+"""
+
+from pathlib import Path
+
+from utils.storage_paths import get_repo_root, sanitize_user_id
+
+
+def get_workspace_root() -> Path:
+    """Return absolute workspace root directory under repo root."""
+    return (get_repo_root() / "workspace").resolve()
+
+
+def get_user_workspace_dir(user_id: str) -> Path:
+    """Return absolute workspace directory for the given user."""
+    safe_user_id = sanitize_user_id(user_id)
+    return (get_workspace_root() / f"workspace_{safe_user_id}").resolve()