Slai Agent Runtime： linux pty drain

launch/core/platforms/base.py

@@ -4,6 +4,7 @@
 import json
 import os
 import re
+import subprocess
 import tarfile
 from dataclasses import dataclass
 from pathlib import Path
@@ -258,29 +259,62 @@ def commit(self, image_name: str, tag: str = "latest", push: bool = False) -> st
         # currently restart from stopped container is not enabled. you need to launch a new instance from the new image you committed.
         return f"{image_name}:{tag}"
 
+    @staticmethod
+    def _image_pull_timeout_seconds() -> int:
+        """Return a bounded image-pull deadline from the environment."""
+        value = os.getenv("REPOLAUNCH_IMAGE_PULL_TIMEOUT_SECONDS", "1800")
+        try:
+            timeout = int(value)
+        except ValueError as exc:
+            raise ValueError(
+                "REPOLAUNCH_IMAGE_PULL_TIMEOUT_SECONDS must be a positive integer"
+            ) from exc
+        if timeout <= 0:
+            raise ValueError(
+                "REPOLAUNCH_IMAGE_PULL_TIMEOUT_SECONDS must be a positive integer"
+            )
+        return timeout
+
     @staticmethod
     def pull_image(image_name: str) -> bool:
-        """
-        Pull Docker image from registry.
-
-        Args:
-            image_name (str): Name of the Docker image to pull
-
-        Returns:
-            bool: True if successful, False if image not found
-        """
-        client = docker.from_env(timeout=3600) # pull should finish in 1 hour
+        """Ensure an image is local, with a terminal deadline for registry pulls."""
+        client = docker.from_env(timeout=30)
         try:
-            # Check if image exists locally
             client.images.get(image_name)
             return True
         except docker.errors.ImageNotFound:
-            # Image doesn't exist locally, try to pull it
-            try:
-                client.images.pull(image_name)
-                return True
-            except docker.errors.ImageNotFound:
-                raise ValueError(f"Image {image_name} not found in registry")
+            pass
+
+        timeout = BaseRuntime._image_pull_timeout_seconds()
+        command = ["docker", "pull", image_name]
+        try:
+            completed = subprocess.run(
+                command,
+                capture_output=True,
+                text=True,
+                timeout=timeout,
+                check=False,
+            )
+        except subprocess.TimeoutExpired as exc:
+            raise RuntimeError(
+                f"Docker image pull for {image_name} timed out after {timeout} seconds"
+            ) from exc
+
+        if completed.returncode != 0:
+            detail = (completed.stderr or completed.stdout).strip()
+            suffix = f": {detail}" if detail else ""
+            raise RuntimeError(
+                f"Docker image pull for {image_name} failed with exit code "
+                f"{completed.returncode}{suffix}"
+            )
+
+        try:
+            client.images.get(image_name)
+        except docker.errors.ImageNotFound as exc:
+            raise RuntimeError(
+                f"Docker image pull for {image_name} completed without a local image"
+            ) from exc
+        return True
 
     def cleanup(self, prune_dangling: bool = True) -> None:
         if self.stopped:

launch/core/platforms/linux.py

@@ -16,7 +16,9 @@
     BaseRuntime
 )
 
-import os, json
+import json
+import os
+import posixpath
 from typing import Any, Optional
 import queue
 import threading
@@ -26,6 +28,12 @@
 import docker
 from docker.models.containers import Container
 
+
+def _container_mount_path(mount_path: str, filename: str) -> str:
+    """Build a path for a Linux container regardless of the host platform."""
+    return posixpath.join(mount_path, filename)
+
+
 class LinuxRuntime(BaseRuntime):
 
     def __init__(
@@ -95,22 +103,61 @@ def _read_raw_output(self, timeout:int=30) -> tuple[str, Optional[CmdOutputMetad
         '''
         timeout: in seconds
         '''
-
-        accumulated_output = ""
+
+        # Keep all output for the command result, but do not run the full prompt
+        # regex against that growing buffer after every socket chunk.  Large test
+        # logs arrive in many small chunks; rescanning the whole log per chunk is
+        # quadratic and can leave the evaluator apparently hung after a command
+        # has already completed.
+        output_chunks: list[str] = []
+        prompt_start = CMD_OUTPUT_PS1_BEGIN.strip()
+        prompt_end = CMD_OUTPUT_PS1_END.strip()
+        marker_tail = ""
+        prompt_candidate: str | None = None
         start_time = time.time()
 
         while time.time() - start_time < timeout:
             try:
                 chunk = self.output_queue.get(timeout=0.1)
-                accumulated_output += chunk.decode("utf-8", errors="ignore")
-                # PSReadLine injects ANSI + cursor control; normalize before matching
-                accumulated_clean = ANSI_ESCAPE.sub("", accumulated_output).replace("\r", "")
-                ps1_matches = CmdOutputMetadata.matches_ps1_metadata(accumulated_clean)
-                if ps1_matches:
+                decoded_chunk = chunk.decode("utf-8", errors="ignore")
+                output_chunks.append(decoded_chunk)
+                # Prompt markers are plain text.  Normalize only the bounded
+                # detection buffer while streaming, then normalize the complete
+                # output once below before extracting the final command output.
+                clean_chunk = ANSI_ESCAPE.sub("", decoded_chunk).replace("\r", "")
+
+                if prompt_candidate is None:
+                    probe = marker_tail + clean_chunk
+                    start = probe.rfind(prompt_start)
+                    if start < 0:
+                        marker_tail = probe[-(len(prompt_start) - 1):]
+                        continue
+                    prompt_candidate = probe[start:]
+                    marker_tail = ""
+                else:
+                    prompt_candidate += clean_chunk
+
+                if prompt_end not in prompt_candidate:
+                    continue
+
+                # Parse only the small prompt candidate while streaming.  The
+                # complete output is parsed once after the loop so callers keep
+                # the existing output/metadata semantics.
+                if CmdOutputMetadata.matches_ps1_metadata(prompt_candidate):
                     break
+
+                # Command output can contain marker-like text.  Discard an
+                # invalid completed candidate and resume bounded marker search.
+                next_start = prompt_candidate.rfind(prompt_start, 1)
+                if next_start > 0:
+                    prompt_candidate = prompt_candidate[next_start:]
+                else:
+                    marker_tail = prompt_candidate[-(len(prompt_start) - 1):]
+                    prompt_candidate = None
             except queue.Empty:
                 continue
-        accumulated_output = ANSI_ESCAPE.sub("", accumulated_output).replace("\r", "")
+
+        accumulated_output = ANSI_ESCAPE.sub("", "".join(output_chunks)).replace("\r", "")
         ps1_matches = CmdOutputMetadata.matches_ps1_metadata(accumulated_output)
         metadata = (
             CmdOutputMetadata.from_ps1_match(ps1_matches[-1]) if ps1_matches else None
@@ -205,9 +252,13 @@ def apply_patch(self, patch: str, verbose: bool = False) -> bool:
 
         filename = f"{uuid.uuid4()}.diff"
         hostpath = os.path.join(self.mnt_host, filename)
-        with open(hostpath, "w") as f:
-            f.write(patch)
-        containerpath =  os.path.join(self.mnt_container, filename)
+        # The file is bind-mounted into a Linux container. Normalize model
+        # output to UTF-8, no BOM, and LF-only newlines so git apply sees
+        # stable patch context bytes regardless of the Windows host.
+        normalized_patch = patch.lstrip("\ufeff").replace("\r\n", "\n").replace("\r", "\n")
+        with open(hostpath, "w", encoding="utf-8", newline="\n") as f:
+            f.write(normalized_patch)
+        containerpath = _container_mount_path(self.mnt_container, filename)
 
         cmd = f"""git apply --reject  --whitespace=nowarn  {containerpath} """
         res = self.send_command(cmd)

launch/utilities/tools/str_replace_editor.py

@@ -1,11 +1,18 @@
 import os
+import posixpath
 import time
 from typing import Any, Optional
 
 from launch.core.runtime import SetupRuntime
 from launch.agent.action_parser import ActionParser
 
 
+def _join_container_path(mount_path: str, filename: str, platform: str) -> str:
+    if platform == "linux":
+        return posixpath.join(mount_path, filename)
+    return os.path.join(mount_path, filename)
+
+
 class Editor(ActionParser):
     action_prompt = """
 # String_replace: replace a sub-string with a new sub-string in an existing file. 
@@ -187,7 +194,7 @@ def _get_file_from_container(self, container_path: str) -> dict[str, Any]:
                 "error": f"{container_path} is directory. Cannot edit on directory. Check again!"
             }
         swap_file = self.temp_file()
-        container_mnt_path = os.path.join(self.mnt_container, swap_file)
+        container_mnt_path = _join_container_path(self.mnt_container, swap_file, self.container.platform)
         host_mnt_path = os.path.join(self.mnt_host, swap_file)
         self.container.send_command(f"cp {container_path} {container_mnt_path}")
         return {
@@ -219,7 +226,7 @@ def str_replace(self, path: str, old_str: str, new_str: str) -> str:
         content = content.replace(old_str, new_str)
         new_swap = self.temp_file()
         new_host_path = os.path.join(self.mnt_host, new_swap)
-        new_container_path = os.path.join(self.mnt_container, new_swap)
+        new_container_path = _join_container_path(self.mnt_container, new_swap, self.container.platform)
         with open(new_host_path, "w", encoding="utf-8") as f:
             f.write(content)
         self._write_back(new_container_path, path, old_swap=res["container_path"])
@@ -243,7 +250,7 @@ def create_file(self, path: str, content: str) -> str:
         swap_file = self.temp_file()
         with open(os.path.join(self.mnt_host,swap_file), "w", encoding="utf-8") as f:
             f.write(content)
-        self._write_back(os.path.join(self.mnt_container,swap_file), path)
+        self._write_back(_join_container_path(self.mnt_container, swap_file, self.container.platform), path)
         observation = f"File created successfully at {path}.\n"
         if ("testbed" in path) or ((not path.strip().startswith("C:")) and (not path.strip().startswith("/"))):
             self.container.send_command(f"git add {path}")

tests/image_pull_deadline_test.py

@@ -0,0 +1,78 @@
+import os
+import subprocess
+import sys
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+import docker
+from launch.core.platforms.base import BaseRuntime
+
+
+class _Images:
+    def __init__(self):
+        self.get_calls = 0
+
+    def get(self, image_name):
+        self.get_calls += 1
+        if self.get_calls == 1:
+            raise docker.errors.ImageNotFound("not local")
+        return object()
+
+
+class _Client:
+    def __init__(self):
+        self.images = _Images()
+
+
+class ImagePullDeadlineTests(unittest.TestCase):
+    def test_missing_image_uses_bounded_docker_cli_pull(self):
+        client = _Client()
+        completed = SimpleNamespace(returncode=0, stdout="pulled", stderr="")
+        with patch.dict(os.environ, {"REPOLAUNCH_IMAGE_PULL_TIMEOUT_SECONDS": "7"}, clear=False), \
+             patch("launch.core.platforms.base.docker.from_env", return_value=client), \
+             patch("launch.core.platforms.base.subprocess.run", return_value=completed) as run:
+            self.assertTrue(BaseRuntime.pull_image("example/image"))
+
+        run.assert_called_once_with(
+            ["docker", "pull", "example/image"],
+            capture_output=True,
+            text=True,
+            timeout=7,
+            check=False,
+        )
+        self.assertEqual(client.images.get_calls, 2)
+
+    def test_pull_timeout_has_a_terminal_error(self):
+        client = _Client()
+        timed_out = subprocess.TimeoutExpired(
+            cmd=["docker", "pull", "example/image"], timeout=3,
+        )
+        with patch.dict(os.environ, {"REPOLAUNCH_IMAGE_PULL_TIMEOUT_SECONDS": "3"}, clear=False), \
+             patch("launch.core.platforms.base.docker.from_env", return_value=client), \
+             patch("launch.core.platforms.base.subprocess.run", side_effect=timed_out):
+            with self.assertRaisesRegex(RuntimeError, "timed out after 3 seconds"):
+                BaseRuntime.pull_image("example/image")
+
+    def test_nonzero_pull_exposes_the_docker_error(self):
+        client = _Client()
+        completed = SimpleNamespace(returncode=1, stdout="", stderr="registry unavailable")
+        with patch.dict(os.environ, {"REPOLAUNCH_IMAGE_PULL_TIMEOUT_SECONDS": "3"}, clear=False), \
+             patch("launch.core.platforms.base.docker.from_env", return_value=client), \
+             patch("launch.core.platforms.base.subprocess.run", return_value=completed):
+            with self.assertRaisesRegex(RuntimeError, "registry unavailable"):
+                BaseRuntime.pull_image("example/image")
+
+    def test_invalid_timeout_configuration_is_rejected(self):
+        with patch.dict(os.environ, {"REPOLAUNCH_IMAGE_PULL_TIMEOUT_SECONDS": "zero"}, clear=False):
+            with self.assertRaisesRegex(ValueError, "positive integer"):
+                BaseRuntime._image_pull_timeout_seconds()
+
+
+if __name__ == "__main__":
+    unittest.main()