Slai Agent Runtime：Support opt-in deterministic Linux evaluation clock

launch/core/platforms/linux.py

@@ -16,16 +16,25 @@
     BaseRuntime
 )
 
-import os, json
+import json
+import os
+import posixpath
 from typing import Any, Optional
 import queue
+import shlex
 import threading
 import time
 import uuid
 
 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 +104,35 @@ def _read_raw_output(self, timeout:int=30) -> tuple[str, Optional[CmdOutputMetad
         '''
         timeout: in seconds
         '''
-
-        accumulated_output = ""
+
+        # Do not repeatedly normalize and regex-scan the complete PTY transcript
+        # after every chunk. Large build/test commands can emit tens of megabytes;
+        # rescanning the growing string turns one command read into quadratic host
+        # work after the command itself has already completed.
+        output_chunks: list[str] = []
+        prompt_probe = ""
+        prompt_end = CMD_OUTPUT_PS1_END.strip()
         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:
+                normalized_chunk = ANSI_ESCAPE.sub(
+                    "", chunk.decode("utf-8", errors="ignore")
+                ).replace("\r", "")
+                output_chunks.append(normalized_chunk)
+
+                # The delimiter can straddle two socket reads, so retain only the
+                # short suffix needed to recognize it incrementally. Parse the
+                # complete metadata exactly once after the delimiter is observed.
+                prompt_probe += normalized_chunk
+                if prompt_end in prompt_probe:
                     break
+                prompt_probe = prompt_probe[-(len(prompt_end) - 1):]
             except queue.Empty:
                 continue
-        accumulated_output = ANSI_ESCAPE.sub("", accumulated_output).replace("\r", "")
+
+        accumulated_output = "".join(output_chunks)
         ps1_matches = CmdOutputMetadata.matches_ps1_metadata(accumulated_output)
         metadata = (
             CmdOutputMetadata.from_ps1_match(ps1_matches[-1]) if ps1_matches else None
@@ -194,6 +216,18 @@ def send_command(self, command: str, timeout: int|None = None) -> CommandResult:
             kill_metadata.exit_code = TIMEOUT_EXIT_CODE
             return CommandResult(output=output, metadata=kill_metadata)
 
+        # If Ctrl-C did not produce a prompt, the timed-out foreground process
+        # may still be running in this isolated evaluation container. Do not
+        # report a timeout while leaving that process alive: terminate the
+        # container so the command has a reliable cancellation boundary.
+        try:
+            self.container.kill()
+        finally:
+            try:
+                self.container.remove(force=True)
+            finally:
+                self.stopped = True
+
         fallback_metadata = CmdOutputMetadata(
             exit_code=TIMEOUT_EXIT_CODE,
         )
@@ -205,9 +239,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)
+        # Patch text may originate from a Windows-hosted model output.  Keep
+        # the mounted file byte-for-byte LF/UTF-8 so Linux git apply does not
+        # treat every context line as CRLF text inside the evaluation container.
+        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)
@@ -256,9 +294,7 @@ def _start_container(
             stdin_open=True,
             tty=True,
             detach=True,
-            environment={
-                "TERM": "xterm-mono",
-            },
+            environment={"TERM": "xterm-mono"},
             working_dir=working_dir,
             extra_hosts=extra_hosts,
             volumes={
@@ -275,6 +311,31 @@ def _start_container(
                     command_timeout=command_timeout,
                 )
 
+        # The task image is intentionally immutable. When the evaluator has an
+        # explicit snapshot clock, provision faketime before exporting it in the
+        # persistent shell so certificate/timestamp checks use that clock without
+        # changing normal tasks.  Export faketime only for test commands: keeping
+        # git, patch application, and rebuild tools on real filesystem timestamps
+        # prevents a snapshot-clock run from making generated binaries appear older
+        # than their patched sources and accidentally reusing a stale build.
+        evaluation_clock = os.environ.get("SWE_BENCH_LIVE_EVALUATION_CLOCK", "").strip()
+        if evaluation_clock:
+            faketime_library = "/usr/lib/x86_64-linux-gnu/faketime/libfaketime.so.1"
+            setup_clock = (
+                f"if [ ! -f {shlex.quote(faketime_library)} ]; then "
+                "DEBIAN_FRONTEND=noninteractive apt-get update -qq && "
+                "DEBIAN_FRONTEND=noninteractive apt-get install -y -qq libfaketime || exit $?; "
+                "fi; "
+                f"export SWE_BENCH_LIVE_FAKETIME_LIBRARY={shlex.quote(faketime_library)}; "
+                f"export SWE_BENCH_LIVE_FAKETIME_VALUE={shlex.quote('@' + evaluation_clock)}; "
+                "swe_bench_live_with_clock() { "
+                "FAKETIME=$SWE_BENCH_LIVE_FAKETIME_VALUE "
+                "FAKETIME_DONT_RESET=1 FAKETIME_DONT_FAKE_MONOTONIC=1 "
+                "LD_PRELOAD=$SWE_BENCH_LIVE_FAKETIME_LIBRARY \"$@\"; "
+                "}"
+            )
+            session.send_command(setup_clock)
+
         return session
 
     @classmethod

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/linux_deterministic_clock_test.py

@@ -0,0 +1,68 @@
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from launch.core.platforms.linux import LinuxRuntime
+
+
+class LinuxDeterministicClockContractTests(unittest.TestCase):
+    def _start(self, clock):
+        captured = {}
+
+        class FakeContainer:
+            def attach_socket(self, **_kwargs):
+                return object()
+
+        class FakeContainers:
+            def run(self, *_args, **kwargs):
+                captured.update(kwargs)
+                return FakeContainer()
+
+        class FakeClient:
+            containers = FakeContainers()
+
+            def version(self):
+                return {"Os": "linux"}
+
+        class FakeDocker:
+            def ping(self):
+                return True
+
+        environment = {} if not clock else {"SWE_BENCH_LIVE_EVALUATION_CLOCK": clock}
+        with patch("launch.core.platforms.linux.docker.from_env", side_effect=[FakeDocker(), FakeClient()]), patch.object(
+            LinuxRuntime, "pull_image", return_value=None
+        ), patch.object(LinuxRuntime, "_clear_initial_prompt", return_value=None), patch.object(
+            LinuxRuntime, "send_command", return_value=MagicMock()
+        ) as send_command, patch.dict(os.environ, environment, clear=not bool(clock)):
+            runtime = LinuxRuntime._start_container("task-image", "task-id", 30, 1)
+        return runtime, captured, send_command
+
+    def test_clock_env_provisions_and_exports_faketime(self):
+        _runtime, captured, send_command = self._start("2026-04-10 12:00:00")
+        self.assertEqual(captured["environment"], {"TERM": "xterm-mono"})
+        clock_commands = [
+            call.args[0] for call in send_command.call_args_list if "libfaketime" in call.args[0]
+        ]
+        self.assertEqual(len(clock_commands), 1)
+        command = clock_commands[0]
+        self.assertIn("apt-get install -y -qq libfaketime", command)
+        self.assertIn("export SWE_BENCH_LIVE_FAKETIME_VALUE='@2026-04-10 12:00:00'", command)
+        self.assertIn("swe_bench_live_with_clock()", command)
+        self.assertIn("FAKETIME_DONT_RESET=1", command)
+        self.assertIn("FAKETIME_DONT_FAKE_MONOTONIC=1", command)
+        self.assertIn("LD_PRELOAD=$SWE_BENCH_LIVE_FAKETIME_LIBRARY", command)
+
+    def test_no_clock_preserves_default_environment_and_skips_setup(self):
+        _runtime, captured, send_command = self._start("")
+        self.assertEqual(captured["environment"], {"TERM": "xterm-mono"})
+        self.assertFalse(any("libfaketime" in call.args[0] for call in send_command.call_args_list))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/linux_output_drain_test.py

@@ -0,0 +1,52 @@
+import queue
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from launch.core.platforms.base import CmdOutputMetadata
+from launch.core.platforms.linux import LinuxRuntime
+
+
+class LinuxOutputDrainTests(unittest.TestCase):
+    def test_large_output_scans_for_prompt_once_and_accepts_split_delimiter(self):
+        runtime = object.__new__(LinuxRuntime)
+        runtime.stopped = True
+        runtime.output_queue = queue.Queue()
+        for _ in range(256):
+            runtime.output_queue.put(b"x" * 4096)
+        runtime.output_queue.put(b'\n###PS1JSON###\n{"exit_code": "0", "username": "root"')
+        runtime.output_queue.put(
+            b', "hostname": "test", "working_dir": "/testbed", '
+            b'"py_interpreter_path": ""}\n###PS1END###\n'
+        )
+
+        original_matches = CmdOutputMetadata.matches_ps1_metadata
+        calls = 0
+
+        def counted_matches(output_text):
+            nonlocal calls
+            calls += 1
+            return original_matches(output_text)
+
+        with patch.object(
+            CmdOutputMetadata,
+            "matches_ps1_metadata",
+            new=counted_matches,
+        ):
+            output, metadata = runtime._read_raw_output(timeout=1)
+
+        self.assertEqual(calls, 1)
+        self.assertIsNotNone(metadata)
+        self.assertEqual(metadata.exit_code, 0)
+        self.assertEqual(metadata.working_dir, "/testbed")
+        self.assertTrue(output.startswith("x" * 4096))
+        self.assertEqual(output.count("x"), 256 * 4096)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/linux_timeout_test.py

@@ -0,0 +1,57 @@
+import sys
+import unittest
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from launch.core.platforms.linux import LinuxRuntime
+from launch.core.platforms.base import TIMEOUT_EXIT_CODE
+
+
+LINUX_RUNTIME = ROOT / "launch" / "core" / "platforms" / "linux.py"
+
+
+class LinuxTimeoutContractTests(unittest.TestCase):
+    def test_timeout_without_prompt_kills_isolated_container(self):
+        source = LINUX_RUNTIME.read_text(encoding="utf-8")
+        self.assertIn("self.container.kill()", source)
+        self.assertIn("self.container.remove(force=True)", source)
+        self.assertIn("self.stopped = True", source)
+
+        class FakeContainer:
+            def __init__(self):
+                self.killed = False
+                self.removed = False
+
+            def kill(self):
+                self.killed = True
+
+            def remove(self, force=False):
+                self.removed = force
+
+        runtime = object.__new__(LinuxRuntime)
+        runtime.container = FakeContainer()
+        runtime.command_timeout = 1
+        runtime.stopped = False
+        runtime._clear_initial_prompt = lambda: None
+        runtime._send_bytes = lambda data: None
+        reads = iter([("partial output", None), ("", None)])
+        runtime._read_raw_output = lambda timeout: next(reads)
+
+        result = runtime.send_command("long-running build", timeout=0)
+
+        self.assertEqual(result.metadata.exit_code, TIMEOUT_EXIT_CODE)
+        self.assertTrue(runtime.container.killed)
+        self.assertTrue(runtime.container.removed)
+        self.assertTrue(runtime.stopped)
+
+    def test_linux_runtime_keeps_minute_constructor_contract(self):
+        source = LINUX_RUNTIME.read_text(encoding="utf-8")
+        self.assertIn("command_timeout: int = 30", source)
+        self.assertIn("timeout = self.command_timeout * 60", source)
+
+
+if __name__ == "__main__":
+    unittest.main()