Merge branch 'main' into add/chunk-file-uploads

Author: deep1401

api/localprovider_pyproject.toml

@@ -32,7 +32,7 @@ dependencies = [
   "soundfile==0.13.1",
   "tensorboardX==2.6.2.2",
   "timm==1.0.15",
-  "transformerlab==0.1.24",
+  "transformerlab==0.1.25",
   "transformerlab-inference==0.2.52",
   "transformers==4.57.1",
   "wandb==0.23.1",

api/pyproject.toml

@@ -24,7 +24,7 @@ dependencies = [
   "python-dotenv==1.1.0",
   "python-multipart==0.0.20",
   "sqlalchemy[asyncio]==2.0.40",
-  "transformerlab==0.1.24",
+  "transformerlab==0.1.25",
   "transformerlab-inference==0.2.52",
   "uvicorn==0.35.0",
   "watchfiles==1.0.5",

api/test/api/test_job_save_to_registry.py

@@ -6,6 +6,18 @@
 from transformerlab.shared.models.models import ProviderType, AcceleratorType
 
 
+class ProviderResourceGroup(BaseModel):
+    """Schema for provider resource group configuration."""
+
+    id: str = Field(..., min_length=1, max_length=100)
+    name: str = Field(..., min_length=1, max_length=100)
+    cpus: Optional[str] = None
+    memory: Optional[str] = None
+    disk_space: Optional[str] = None
+    accelerators: Optional[str] = None
+    num_nodes: Optional[int] = None
+
+
 class ProviderConfigBase(BaseModel):
     """Base schema for provider configuration."""
 
@@ -34,6 +46,7 @@ class ProviderConfigBase(BaseModel):
 
     # Accelerators supported by this provider
     supported_accelerators: Optional[List[AcceleratorType]] = Field(default=None)
+    resource_groups: Optional[List[ProviderResourceGroup]] = Field(default=None)
 
     # Additional provider-specific config
     extra_config: Dict[str, Any] = Field(default_factory=dict)

api/transformerlab/routers/experiment/jobs.py

@@ -3,10 +3,32 @@
 from typing import Optional
 
 
+# Cap cluster names to 41 characters to stay withing the limit of
+# the strictest provider (dstack).
+# This means capping the basename portion to 28 characters (use 25 to be safe)
+# to leave room for a ``-job-<short_id>`` suffix (13 chars).
+_MAX_BASENAME_LENGTH = 25
+
+
 def sanitize_cluster_basename(base_name: Optional[str]) -> str:
-    """Return a filesystem-safe cluster base name."""
+    """Return a cluster base name that is safe across compute providers.
+
+    Some providers have restrictions on resource names.
+    Sanitize generated name so the result is:
+     - lowercased
+     - underscores are replaced with hyphens
+     - name is guaranteed to start with a letter
+     - no longer than 41 characters
+    """
     if not base_name:
         return "remote-template"
-    normalized = "".join(ch if ch.isalnum() or ch in ("-", "_") else "-" for ch in base_name.strip())
-    normalized = normalized.strip("-_")
-    return normalized or "remote-template"
+    lowered = base_name.strip().lower()
+    normalized = "".join(ch if (ch.isalnum() and ch.isascii()) or ch == "-" else "-" for ch in lowered)
+    normalized = normalized.strip("-")
+    if not normalized:
+        return "remote-template"
+    if not normalized[0].isalpha():
+        normalized = f"t-{normalized}"
+    if len(normalized) > _MAX_BASENAME_LENGTH:
+        normalized = normalized[:_MAX_BASENAME_LENGTH].rstrip("-")
+    return normalized

api/transformerlab/schemas/compute_providers.py

@@ -278,12 +278,18 @@ async def launch_sweep_jobs(
                             if azure_sas:
                                 env_vars["AZURE_STORAGE_SAS_TOKEN"] = azure_sas
 
-                if request.file_mounts is True and request.task_id:
-                    setup_commands.append(COPY_FILE_MOUNTS_SETUP)
-
                 if provider.type == ProviderType.RUNPOD.value:
                     setup_commands.append("curl -LsSf https://astral.sh/uv/install.sh | sh")
 
+                if provider.type != ProviderType.LOCAL.value:
+                    setup_commands.append("pip install -q transformerlab")
+
+                    if request.enable_profiling_torch:
+                        setup_commands.append("pip install -q torch")
+
+                if request.file_mounts is True and request.task_id:
+                    setup_commands.append(COPY_FILE_MOUNTS_SETUP)
+
                 if request.github_repo_url:
                     workspace_dir = await get_workspace_dir()
                     github_pat = await read_github_pat_from_workspace(workspace_dir, user_id=user_id)

api/transformerlab/services/compute_provider/cluster_naming.py

@@ -254,8 +254,6 @@ async def launch_template_on_provider(
                 if azure_sas:
                     env_vars["AZURE_STORAGE_SAS_TOKEN"] = azure_sas
 
-    if request.file_mounts is True and request.task_id:
-        setup_commands.append(COPY_FILE_MOUNTS_SETUP)
     # Ensure transformerlab SDK is available on remote machines for live_status tracking and other helpers.
     # This runs after AWS credentials are configured so we have access to any remote storage if needed.
     if provider.type != ProviderType.LOCAL.value:
@@ -264,6 +262,8 @@ async def launch_template_on_provider(
         # Install torch as well if torch profiler is enabled
         if request.enable_profiling_torch:
             setup_commands.append("pip install -q torch")
+    if request.file_mounts is True and request.task_id:
+        setup_commands.append(COPY_FILE_MOUNTS_SETUP)
     # For RunPod providers, ensure uv is available and configured to use the
     # system Python. This allows user commands to invoke `uv` directly.
     if provider.type == ProviderType.RUNPOD.value:

api/transformerlab/services/compute_provider/launch_sweep.py

@@ -3,7 +3,7 @@
 import logging
 import os
 import uuid
-from datetime import datetime, timezone
+from datetime import datetime
 from typing import Optional
 
 from pydantic import BaseModel
@@ -98,23 +98,15 @@ async def _poll_pending_remote_entries() -> list[JobQueue]:
 async def _mark_entry_dispatched(entry_id: str) -> None:
     """Transition a job_queue row from PENDING to DISPATCHED."""
     async with async_session() as session:
-        stmt = (
-            update(JobQueue)
-            .where(JobQueue.id == entry_id)
-            .values(status="DISPATCHED", updated_at=datetime.now(timezone.utc))
-        )
+        stmt = update(JobQueue).where(JobQueue.id == entry_id).values(status="DISPATCHED", updated_at=datetime.utcnow())
         await session.execute(stmt)
         await session.commit()
 
 
 async def _mark_entry_failed(entry_id: str) -> None:
     """Transition a job_queue row to FAILED (could not reconstruct work item)."""
     async with async_session() as session:
-        stmt = (
-            update(JobQueue)
-            .where(JobQueue.id == entry_id)
-            .values(status="FAILED", updated_at=datetime.now(timezone.utc))
-        )
+        stmt = update(JobQueue).where(JobQueue.id == entry_id).values(status="FAILED", updated_at=datetime.utcnow())
         await session.execute(stmt)
         await session.commit()