Fix UnicodeDecodeError When Reading Job Log Files (#45)

Author: AryanAhadinia

src/swiss_ai_model_launch/launchers/firecrest_launcher.py

@@ -9,7 +9,11 @@
 from swiss_ai_model_launch.launchers.launch_args import LaunchArgs
 from swiss_ai_model_launch.launchers.launch_request import LaunchRequest
 from swiss_ai_model_launch.launchers.launcher import JobStatus, Launcher
-from swiss_ai_model_launch.launchers.utils import create_salt, render_job_script
+from swiss_ai_model_launch.launchers.utils import (
+    create_salt,
+    decode_log,
+    render_job_script,
+)
 
 _REMOTE_MODEL_REGISTRY = Path("/capstor/store/cscs/swissai/infra01/hf_models/models/")
 
@@ -211,8 +215,8 @@ async def get_job_logs(self, job_id: int) -> tuple[str, str]:
                     account=self.account,
                     blocking=True,
                 )
-                with open(target_dir_path / "log.out") as out_f:
-                    out_log = out_f.read()
+                with open(target_dir_path / "log.out", "rb") as out_f:
+                    out_log = decode_log(out_f.read())
             except FileNotFoundError:
                 out_log = ""
 
@@ -224,8 +228,8 @@ async def get_job_logs(self, job_id: int) -> tuple[str, str]:
                     account=self.account,
                     blocking=True,
                 )
-                with open(target_dir_path / "log.err") as err_f:
-                    err_log = err_f.read()
+                with open(target_dir_path / "log.err", "rb") as err_f:
+                    err_log = decode_log(err_f.read())
             except FileNotFoundError:
                 err_log = ""
 

src/swiss_ai_model_launch/launchers/slurm_launcher.py

@@ -8,7 +8,11 @@
 from swiss_ai_model_launch.launchers.launch_args import LaunchArgs
 from swiss_ai_model_launch.launchers.launch_request import LaunchRequest
 from swiss_ai_model_launch.launchers.launcher import JobStatus, Launcher
-from swiss_ai_model_launch.launchers.utils import create_salt, render_job_script
+from swiss_ai_model_launch.launchers.utils import (
+    create_salt,
+    decode_log,
+    render_job_script,
+)
 
 _REMOTE_MODEL_REGISTRY = Path("/capstor/store/cscs/swissai/infra01/hf_models/models/")
 
@@ -198,12 +202,12 @@ async def get_job_logs(self, job_id: int) -> tuple[str, str]:
         log_dir = self._get_working_dir() / "logs" / str(job_id)
 
         try:
-            out_log = (log_dir / "log.out").read_text()
+            out_log = decode_log((log_dir / "log.out").read_bytes())
         except FileNotFoundError:
             out_log = ""
 
         try:
-            err_log = (log_dir / "log.err").read_text()
+            err_log = decode_log((log_dir / "log.err").read_bytes())
         except FileNotFoundError:
             err_log = ""
 

src/swiss_ai_model_launch/launchers/utils.py

@@ -16,3 +16,34 @@ def create_salt(length: int) -> str:
 def render_job_script(launch_args: LaunchArgs) -> str:
     template = Template(_TEMPLATE_PATH.read_text())
     return str(template.render(**launch_args.model_dump()))
+
+
+def decode_log(data: bytes) -> str:
+    """Decode log bytes to string, tolerating partial UTF-8 sequences at the tail.
+
+    Log files may be read while the writer is mid-flush, leaving an incomplete
+    multi-byte UTF-8 sequence at the end. This function strips any such trailing
+    incomplete sequence before decoding, then falls back to errors="replace" for
+    any other malformed bytes in the content.
+    """
+    # Strip trailing incomplete multi-byte UTF-8 sequence.
+    # Walk backwards past continuation bytes (10xxxxxx), then check whether the
+    # leading byte at that position expects more continuation bytes than are present.
+    i = len(data) - 1
+    num_continuation = 0
+    while i >= 0 and (data[i] & 0xC0) == 0x80:
+        num_continuation += 1
+        i -= 1
+    if i >= 0:
+        lead = data[i]
+        if lead & 0xE0 == 0xC0:
+            expected = 1
+        elif lead & 0xF0 == 0xE0:
+            expected = 2
+        elif lead & 0xF8 == 0xF0:
+            expected = 3
+        else:
+            expected = num_continuation  # single-byte or already complete
+        if num_continuation < expected:
+            data = data[:i]
+    return data.decode("utf-8", errors="replace")