api/transformerlab/routers/compute_provider.py (2)

2497-2505: ⚠️ Potential issue | 🟠 Major

resume_from_checkpoint inconsistently mutates org context and bypasses the serialization queue for LOCAL provider.

Two issues:

1. This block still uses the old set_organization_id(team_id) / try/finally / set_organization_id(None) pattern that was deliberately removed from launch_template_on_provider (Lines 1508-1515). Since the middleware has already set org context for this request, the mutation is unnecessary and inconsistent.

2. For LOCAL provider, resume_from_checkpoint calls provider_instance.launch_cluster directly via asyncio.to_thread (Line 2584) instead of going through enqueue_local_launch. This bypasses _worker_lock and the serialization queue, potentially running multiple local launches concurrently — contradicting the queue's invariant.

♻️ Proposed fix — align with launch_template_on_provider pattern

-    # For local provider, set TFL_WORKSPACE_DIR so the lab SDK in the subprocess finds the job dir
-    if provider.type == ProviderType.LOCAL.value and team_id:
-        set_organization_id(team_id)
-        try:
-            workspace_dir = await get_workspace_dir()
-            if workspace_dir and not storage.is_remote_path(workspace_dir):
-                env_vars["TFL_WORKSPACE_DIR"] = workspace_dir
-        finally:
-            set_organization_id(None)
+    # For local provider, set TFL_WORKSPACE_DIR so the lab SDK in the subprocess finds the job dir.
+    # Org context is already set by authentication middleware.
+    if provider.type == ProviderType.LOCAL.value and team_id:
+        workspace_dir = await get_workspace_dir()
+        if workspace_dir and not storage.is_remote_path(workspace_dir):
+            env_vars["TFL_WORKSPACE_DIR"] = workspace_dir

For the queue bypass, route through enqueue_local_launch for LOCAL provider (similar to launch_template_on_provider Lines 1657-1679) instead of calling launch_cluster directly.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@api/transformerlab/routers/compute_provider.py` around lines 2497 - 2505, The
resume_from_checkpoint path mutates org context with
set_organization_id(team_id)/try/finally and bypasses the local-launch
serialization by calling provider_instance.launch_cluster via asyncio.to_thread;
remove the unnecessary set_organization_id(...) calls (the request middleware
already sets org context) so resume_from_checkpoint matches
launch_template_on_provider, and for ProviderType.LOCAL.value change the direct
launch to use the existing enqueue_local_launch flow (instead of
asyncio.to_thread(provider_instance.launch_cluster...)) so the call is
serialized under _worker_lock/serialization queue just like
launch_template_on_provider; update any surrounding logic to pass the same
arguments used by enqueue_local_launch.

---

1069-1076: ⚠️ Potential issue | 🟠 Major

_launch_sweep_jobs redundantly re-sets org context and LOCAL sweep child jobs bypass the serialization queue.

set_current_org_id(team_id) and lab_set_org_id(team_id) are already called at Lines 979-981 at the top of _launch_sweep_jobs. The set_organization_id(team_id) block at Line 1070 redundantly re-sets the same value and then immediately clears it in finally, transiently resetting the context to None mid-loop if an exception is thrown in get_workspace_dir.

More critically, LOCAL provider sweep child jobs call provider_instance.launch_cluster directly via asyncio.to_thread (Line 1209) rather than enqueue_local_launch, bypassing _worker_lock and allowing concurrent local launches that the queue was introduced to serialize.

♻️ Proposed fix

-                if provider.type == ProviderType.LOCAL.value and team_id:
-                    set_organization_id(team_id)
-                    try:
-                        workspace_dir = await get_workspace_dir()
-                        if workspace_dir and not storage.is_remote_path(workspace_dir):
-                            env_vars["TFL_WORKSPACE_DIR"] = workspace_dir
-                    finally:
-                        set_organization_id(None)
+                if provider.type == ProviderType.LOCAL.value and team_id:
+                    workspace_dir = await get_workspace_dir()
+                    if workspace_dir and not storage.is_remote_path(workspace_dir):
+                        env_vars["TFL_WORKSPACE_DIR"] = workspace_dir

For the queue bypass, LOCAL provider child sweep jobs should be routed through enqueue_local_launch to respect the serialization invariant.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@api/transformerlab/routers/compute_provider.py` around lines 1069 - 1076,
Remove the redundant org-context toggling in _launch_sweep_jobs by deleting the
set_organization_id(team_id)/finally block around get_workspace_dir (since
set_current_org_id(team_id) and lab_set_org_id(team_id) are already applied at
the top); instead, ensure get_workspace_dir reads the already-set context. Also
fix the LOCAL provider child job path so it uses enqueue_local_launch rather
than calling provider_instance.launch_cluster via asyncio.to_thread (which
bypasses _worker_lock); replace the direct
asyncio.to_thread(provider_instance.launch_cluster, ...) call with a call that
enqueues the launch through enqueue_local_launch so local launches are
serialized by _worker_lock.

api/transformerlab/compute_providers/local.py (2)

244-259: ⚠️ Potential issue | 🔴 Critical

Pre-existing logic inversion in get_cluster_status defeats the PR's fix.

os.kill(pid, 0) returns None on success (process alive) and raises an exception when the process is gone. The current branch if os_killed is not None is therefore always False, making ClusterState.UP dead code and every running local job appear as DOWN. This causes check-status to immediately mark all local jobs as COMPLETE regardless of actual process state.

🐛 Proposed fix

-        try:
-            pid = int(pid_file.read_text().strip())
-            os_killed = os.kill(pid, 0)
-            # Return up only if the process is not running
-            if os_killed is not None:
-                return ClusterStatus(
-                    cluster_name=cluster_name,
-                    state=ClusterState.UP,
-                    status_message="Process running",
-                )
-            else:
-                return ClusterStatus(
-                    cluster_name=cluster_name,
-                    state=ClusterState.DOWN,
-                    status_message="Process not running",
-                )
+        try:
+            pid = int(pid_file.read_text().strip())
+            os.kill(pid, 0)  # raises if process is gone
+            return ClusterStatus(
+                cluster_name=cluster_name,
+                state=ClusterState.UP,
+                status_message="Process running",
+            )

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@api/transformerlab/compute_providers/local.py` around lines 244 - 259, In
get_cluster_status, os.kill(pid, 0) returns None on success (process exists) and
raises on failure, so the current if os_killed is not None branch is inverted
and marks running processes as DOWN; change the logic to call os.kill(pid, 0)
inside a try and treat the no-exception path as ClusterState.UP (status_message
"Process running"), catch ProcessLookupError (and/or OSError with errno ESRCH)
to return ClusterState.DOWN ("Process not running"), treat PermissionError as UP
(process exists but not permitted), and handle any other exceptions by logging
and returning an appropriate DOWN status; update references to
pid_file.read_text(), ClusterStatus and ClusterState.UP/DOWN accordingly.

---

185-192: ⚠️ Potential issue | 🟠 Major

File descriptor leak in Popen call.

The open() handles for stdout.log and stderr.log are passed directly and never closed. The parent process will hold these file descriptors open until garbage-collected.

🔒 Proposed fix

-        proc = subprocess.Popen(
-            ["/bin/bash", "-c", config.command or "true"],
-            cwd=str(job_dir),
-            env=env,
-            stdout=open(job_dir / "stdout.log", "w"),
-            stderr=open(job_dir / "stderr.log", "w"),
-            start_new_session=True,
-        )
+        stdout_log = open(job_dir / "stdout.log", "w")
+        stderr_log = open(job_dir / "stderr.log", "w")
+        proc = subprocess.Popen(
+            ["/bin/bash", "-c", config.command or "true"],
+            cwd=str(job_dir),
+            env=env,
+            stdout=stdout_log,
+            stderr=stderr_log,
+            start_new_session=True,
+        )
+        stdout_log.close()
+        stderr_log.close()

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@api/transformerlab/compute_providers/local.py` around lines 185 - 192, The
Popen call leaks file descriptors because open(job_dir / "stdout.log", "w") and
stderr are passed directly; fix by opening those files into local variables
(e.g., stdout_f, stderr_f) before calling subprocess.Popen, call proc =
subprocess.Popen(..., stdout=stdout_f, stderr=stderr_f, ...), and then close the
parent-side file objects in a finally block (or immediately after Popen returns)
to ensure they are closed even on exceptions; reference the subprocess.Popen
invocation and the job_dir/stdout.log and stderr.log file opens when making the
change.