Move tasks and make them experiment scoped

Author: deep1401

api/api.py

@@ -123,8 +123,10 @@ async def lifespan(app: FastAPI):
         # One-time migration: legacy workspace/jobs -> workspace/experiments/<exp_id>/jobs
         # Runs in the background so it doesn't delay the API startup.
         from transformerlab.services.migrate_jobs_to_experiment_dirs import start_jobs_migration_worker
+        from transformerlab.services.migrate_tasks_to_experiment_dirs import start_tasks_migration_worker
 
         await start_jobs_migration_worker()
+        await start_tasks_migration_worker()
 
         # Start background sweep status updater after all startup steps succeed.
         await start_sweep_status_worker()
@@ -152,11 +154,13 @@ async def lifespan(app: FastAPI):
     # Do the following at API Shutdown:
     if is_leader():
         from transformerlab.services.migrate_jobs_to_experiment_dirs import stop_jobs_migration_worker
+        from transformerlab.services.migrate_tasks_to_experiment_dirs import stop_tasks_migration_worker
 
         await stop_sweep_status_worker()
         await stop_remote_job_status_worker()
         await stop_notification_worker()
         await stop_remote_job_queue_worker()
+        await stop_tasks_migration_worker()
         await stop_jobs_migration_worker()
     from transformerlab.services.process_registry import get_registry
 

api/transformerlab/routers/experiment/task.py

@@ -117,7 +117,7 @@ async def task_get_all(experimentId: str):
     tags=["tasks:{experimentId}"],
 )
 async def task_get_by_id(experimentId: str, task_id: str):
-    task = await task_service.task_get_by_id(task_id)
+    task = await task_service.task_get_by_id(task_id, experiment_id=experimentId)
     if task is None:
         return {"message": "NOT FOUND"}
     return task
@@ -153,7 +153,7 @@ async def task_get_by_type_in_experiment(experimentId: str, type: str):
     response_model=TaskFilesResponse,
     summary="List files associated with a task template (GitHub + local mounts)",
 )
-async def task_list_files(task_id: str) -> TaskFilesResponse:
+async def task_list_files(experimentId: str, task_id: str) -> TaskFilesResponse:
     """
     Return a lightweight list of files associated with a task template.
 
@@ -163,7 +163,7 @@ async def task_list_files(task_id: str) -> TaskFilesResponse:
       readily available.
     - If file_mounts is set, it will be returned as-is as a list of local paths.
     """
-    task = await task_service.task_get_by_id(task_id)
+    task = await task_service.task_get_by_id(task_id, experiment_id=experimentId)
     if not task:
         raise HTTPException(status_code=404, detail="Task not found")
 
@@ -207,7 +207,7 @@ async def task_list_files(task_id: str) -> TaskFilesResponse:
     # Always list files from the canonical per-task directory.
     # This directory contains at minimum task.yaml and may include uploaded files.
     try:
-        task_dir = await task_service.get_task_dir(task_id)
+        task_dir = await task_service.get_task_dir(task_id, experiment_id=experimentId)
         if await storage.exists(task_dir):
             entries = await storage.ls(task_dir)
             # Build a set of basenames already in local_files for dedup.
@@ -353,7 +353,7 @@ async def task_update_file(experimentId: str, task_id: str, file_path: str, requ
     .py, .md, .yaml/.yml, .json, .txt). Request bodies are decoded as UTF-8 and
     written in text mode. Binary uploads should use the file-upload endpoint.
     """
-    task = await task_service.task_get_by_id(task_id)
+    task = await task_service.task_get_by_id(task_id, experiment_id=experimentId)
     if not task:
         raise HTTPException(status_code=404, detail="Task not found")
 
@@ -376,7 +376,7 @@ async def task_update_file(experimentId: str, task_id: str, file_path: str, requ
         await f.write(body)
 
     if not task.get("file_mounts"):
-        await task_service.update_task(task_id, {"file_mounts": True})
+        await task_service.update_task(task_id, {"file_mounts": True}, experiment_id=experimentId)
 
     await cache.invalidate(f"tasks:{experimentId}")
     return {"message": "OK"}
@@ -387,7 +387,7 @@ async def task_update_file(experimentId: str, task_id: str, file_path: str, requ
     summary="Delete a file from a task's local workspace directory",
 )
 async def task_delete_file(experimentId: str, task_id: str, file_path: str):
-    task = await task_service.task_get_by_id(task_id)
+    task = await task_service.task_get_by_id(task_id, experiment_id=experimentId)
     if not task:
         raise HTTPException(status_code=404, detail="Task not found")
 
@@ -421,7 +421,7 @@ async def task_upload_file(
     task_id: str,
     files: list[UploadFile] = File(...),
 ):
-    task = await task_service.task_get_by_id(task_id)
+    task = await task_service.task_get_by_id(task_id, experiment_id=experimentId)
     if not task:
         raise HTTPException(status_code=404, detail="Task not found")
 
@@ -450,7 +450,7 @@ async def task_upload_file(
         raise HTTPException(status_code=400, detail="No valid files uploaded")
 
     if not task.get("file_mounts"):
-        await task_service.update_task(task_id, {"file_mounts": True})
+        await task_service.update_task(task_id, {"file_mounts": True}, experiment_id=experimentId)
 
     await cache.invalidate(f"tasks:{experimentId}")
     return {"status": "success", "files": saved_files}
@@ -460,14 +460,14 @@ async def task_upload_file(
     "/{task_id}/github_file/{file_path:path}",
     summary="Serve a file from the task's associated GitHub repository for preview",
 )
-async def task_get_github_file(task_id: str, file_path: str):
+async def task_get_github_file(experimentId: str, task_id: str, file_path: str):
     """
     Serve a file from the GitHub repository configured on the task (github_repo_url).
 
     This endpoint uses the same GitHub PAT resolution logic as other GitHub helpers
     and is intended for lightweight previews in the UI.
     """
-    task = await task_service.task_get_by_id(task_id)
+    task = await task_service.task_get_by_id(task_id, experiment_id=experimentId)
     if not task:
         raise HTTPException(status_code=404, detail="Task not found")
 
@@ -583,7 +583,7 @@ async def update_task(experimentId: str, task_id: str, new_task: dict = Body()):
     if "name" in new_task:
         new_task["name"] = secure_filename(new_task["name"])
 
-    success = await task_service.update_task(task_id, new_task)
+    success = await task_service.update_task(task_id, new_task, experiment_id=experimentId)
     if success:
         # Best-effort invalidation: task detail + this experiment's task lists.
         await cache.invalidate(f"tasks:{experimentId}")
@@ -594,7 +594,7 @@ async def update_task(experimentId: str, task_id: str, new_task: dict = Body()):
 
 @router.get("/{task_id}/delete", summary="Deletes a task")
 async def delete_task(experimentId: str, task_id: str):
-    success = await task_service.delete_task(task_id)
+    success = await task_service.delete_task(task_id, experiment_id=experimentId)
     if success:
         # Best-effort invalidation: task detail + this experiment's task lists.
         await cache.invalidate(f"tasks:{experimentId}")
@@ -899,23 +899,23 @@ async def create_task(
 
 @router.get("/{task_id}/yaml", summary="Get task.yaml from the task directory")
 async def get_task_yaml(experimentId: str, task_id: str):
-    task = await task_service.task_get_by_id(task_id)
+    task = await task_service.task_get_by_id(task_id, experiment_id=experimentId)
     if task is None:
         raise HTTPException(status_code=404, detail="Task not found")
-    content = await task_service.read_task_yaml(task_id)
+    content = await task_service.read_task_yaml(task_id, experiment_id=experimentId)
     return PlainTextResponse(content, media_type="text/plain")
 
 
 @router.put("/{task_id}/yaml", summary="Save task.yaml and sync index.json")
 async def update_task_yaml(experimentId: str, task_id: str, request: Request):
     body = (await request.body()).decode("utf-8")
-    task = await task_service.task_get_by_id(task_id)
+    task = await task_service.task_get_by_id(task_id, experiment_id=experimentId)
     if task is None:
         raise HTTPException(status_code=404, detail="Task not found")
-    await task_service.write_task_yaml(task_id, body)
+    await task_service.write_task_yaml(task_id, body, experiment_id=experimentId)
     task_data = _parse_yaml_to_task_data(body)
     task_data.pop("id", None)
-    success = await task_service.update_task_from_yaml(task_id, task_data)
+    success = await task_service.update_task_from_yaml(task_id, task_data, experiment_id=experimentId)
     if not success:
         raise HTTPException(status_code=404, detail="Task not found")
     await cache.invalidate(f"tasks:{experimentId}")
@@ -1120,7 +1120,7 @@ async def import_task_from_gallery(
             await storage.makedirs(task_dir_path, exist_ok=True)
             dest_subdir = storage.join(task_dir_path, os.path.basename(local_task_dir.rstrip("/")))
             await storage.copy_dir(local_task_dir, dest_subdir)
-            await task_service.update_task(task_id, {"file_mounts": True})
+            await task_service.update_task(task_id, {"file_mounts": True}, experiment_id=experimentId)
 
         return {"status": "success", "message": f"Interactive task '{task_name}' imported successfully", "id": task_id}
 
@@ -1532,7 +1532,7 @@ async def import_task_from_team_gallery(
                     await f.write(json.dumps(data, indent=2))
             else:
                 # No index.json in the copied directory; write at least minimal metadata.
-                await task_service.update_task(task_id, {"id": task_id})
+                await task_service.update_task(task_id, {"id": task_id}, experiment_id=experimentId)
         except Exception as e:
             raise HTTPException(
                 status_code=500, detail=f"Failed to overwrite id in index.json for imported team task: {e}"
@@ -1708,7 +1708,7 @@ async def export_task_to_team_gallery(
     Export a task into the team-specific gallery stored in workspace_dir.
     Tasks store all fields directly (not nested in config).
     """
-    task = await task_service.task_get_by_id(request.task_id)
+    task = await task_service.task_get_by_id(request.task_id, experiment_id=request.experiment_id)
     if not task:
         raise HTTPException(status_code=404, detail="Task not found")
 

api/transformerlab/services/compute_provider/launch_template.py

@@ -42,7 +42,13 @@
 from transformerlab.shared.models.models import ProviderType
 from transformerlab.shared.secret_utils import load_team_secrets, replace_secrets_in_dict, replace_secret_placeholders
 from lab import storage
-from lab.dirs import get_job_dir, get_local_provider_job_dir, get_task_dir, get_workspace_dir
+from lab.dirs import (
+    get_experiment_task_dir,
+    get_job_dir,
+    get_local_provider_job_dir,
+    get_task_dir,
+    get_workspace_dir,
+)
 from lab.job_status import JobStatus
 from lab.storage import STORAGE_PROVIDER
 from werkzeug.utils import secure_filename
@@ -274,7 +280,7 @@ async def launch_template_on_provider(
     # This handles GitHub-sourced interactive tasks where the CLI/TUI doesn't
     # send these fields and relies on the backend to resolve them from the task.
     if not request.github_repo_url and request.task_id:
-        task_data = await task_service.task_get_by_id(request.task_id)
+        task_data = await task_service.task_get_by_id(request.task_id, experiment_id=request.experiment_id)
         if task_data:
             request.github_repo_url = task_data.get("github_repo_url", "") or ""
             request.github_repo_dir = task_data.get("github_repo_dir", "") or ""
@@ -436,7 +442,7 @@ async def launch_template_on_provider(
     # This handles GitHub-sourced interactive tasks where the command/setup
     # are in task.yaml and were stored in the task at import time.
     if not base_command.strip() and request.task_id:
-        fallback_task = await task_service.task_get_by_id(request.task_id)
+        fallback_task = await task_service.task_get_by_id(request.task_id, experiment_id=request.experiment_id)
         if fallback_task:
             base_command = fallback_task.get("run", "") or fallback_task.get("command", "")
             # Also pick up setup from the task if not already added
@@ -557,8 +563,11 @@ async def launch_template_on_provider(
     # for metadata and overwriting it with the task's index.json would break
     # job status tracking.
     if request.task_id:
-        task_dir_root = await get_task_dir()
-        task_src = storage.join(task_dir_root, secure_filename(str(request.task_id)))
+        task_src = await get_experiment_task_dir(request.experiment_id, request.task_id)
+        if not await storage.isdir(task_src):
+            # Legacy fallback for tasks that have not been migrated yet.
+            task_dir_root = await get_task_dir()
+            task_src = storage.join(task_dir_root, secure_filename(str(request.task_id)))
         if await storage.isdir(task_src):
             workspace_job_dir = await get_job_dir(job_id, request.experiment_id)
             await copy_task_files_to_dir(task_src, workspace_job_dir)