fix

Signed-off-by: oliver könig <[email protected]>

Author: ko3n1g

nemo_run/core/execution/dgxcloud.py

@@ -352,10 +352,13 @@ def _stream_url_sync(self, url: str, headers: Dict[str, str], q: queue.Queue):
         try:
             with requests.get(url, stream=True, headers=headers, verify=False) as response:
                 for line in response.iter_lines(decode_unicode=True):
-                    q.put(f"{line}\n")
+                    q.put(url, f"{line}\n")
         except Exception as e:
             logger.error(f"Error streaming URL {url}: {e}")
 
+        finally:
+            q.put(url, None)
+
     def fetch_logs(
         self,
         job_id: str,
@@ -411,7 +414,11 @@ def fetch_logs(
 
         # Yield chunks as they arrive
         while active_urls:
-            yield q.get()
+            url, item = q.get()
+            if item is None:
+                active_urls.discard(url)
+            else:
+                yield item
 
         # Wait for threads
         for t in threads: