move it to the IsolateSettings

src/isolate/backends/settings.py

@@ -11,7 +11,7 @@
 from platformdirs import user_cache_dir
 
 from isolate.backends.common import lock_build_path
-from isolate.logs import Log
+from isolate.logs import Log, LogLevel, LogSource
 
 if TYPE_CHECKING:
     from isolate.backends import BaseEnvironment
@@ -28,7 +28,34 @@ class IsolateSettings:
     strict_cache: bool = _STRICT_CACHE
 
     def log(self, log: Log) -> None:
-        self.log_hook(log)
+        self.log_hook(self._infer_log_level(log))
+
+    def _infer_log_level(self, log: Log) -> Log:
+        """Infer the log level if it's correctly set."""
+        if log.level not in (LogLevel.STDOUT, LogLevel.STDERR):
+            # We should only infer the log level for stdout/stderr logs.
+            return log
+
+        if log.source in (LogSource.BUILDER, LogSource.BRIDGE):
+            return replace(log, level=LogLevel.TRACE)
+
+        line = log.message.lower()
+
+        if "[error]" in line:
+            return replace(log, level=LogLevel.ERROR)
+        if "[warning]" in line:
+            return replace(log, level=LogLevel.WARNING)
+        if "[warn]" in line:
+            return replace(log, level=LogLevel.WARNING)
+        if "[info]" in line:
+            return replace(log, level=LogLevel.INFO)
+        if "[debug]" in line:
+            return replace(log, level=LogLevel.DEBUG)
+        if "[trace]" in line:
+            return replace(log, level=LogLevel.TRACE)
+
+        # Default to INFO level
+        return replace(log, level=LogLevel.INFO)
 
     def _get_temp_base(self) -> Path:
         """Return the base path for creating temporary files/directories.

src/isolate/connections/_local/_base.py

@@ -12,7 +12,6 @@
     Any,
     Generic,
     Iterator,
-    Optional,
     TypeVar,
 )
 
@@ -116,9 +115,7 @@ def start_process(
         python_executable = get_executable_path(self.environment_path, "python")
         with logged_io(
             partial(
-                self.handle_agent_log,
-                source=LogSource.USER,
-                level=None,  # Will be inferred
+                self.handle_agent_log, source=LogSource.USER, level=LogLevel.STDOUT
             ),
             partial(
                 self.handle_agent_log, source=LogSource.USER, level=LogLevel.STDERR
@@ -176,29 +173,8 @@ def get_python_cmd(
         """Return the command to run the agent process with."""
         raise NotImplementedError
 
-    def infer_log_level(self, line: str, source: LogSource) -> LogLevel:
-        """Infer the log level of the given line."""
-        if source in (LogSource.BUILDER, LogSource.BRIDGE):
-            return LogLevel.TRACE
-        else:
-            if "[error]" in line.lower():
-                return LogLevel.ERROR
-            elif "[warning]" in line.lower():
-                return LogLevel.WARNING
-            elif "[warn]" in line.lower():
-                return LogLevel.WARNING
-            elif "[info]" in line.lower():
-                return LogLevel.INFO
-            elif "[debug]" in line.lower():
-                return LogLevel.DEBUG
-            elif "[trace]" in line.lower():
-                return LogLevel.TRACE
-
-        # Default
-        return LogLevel.INFO
-
     def handle_agent_log(
-        self, line: str, *, level: Optional[LogLevel], source: LogSource
+        self, line: str, *, level: LogLevel, source: LogSource
     ) -> None:
         """Handle a log line emitted by the agent process. The level will be either
         STDOUT or STDERR."""

src/isolate/connections/grpc/_base.py

@@ -2,7 +2,7 @@
 from contextlib import contextmanager
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, ContextManager, Iterator, List, Optional, Tuple, Union, cast
+from typing import Any, ContextManager, Iterator, List, Tuple, Union, cast
 
 import grpc
 
@@ -148,6 +148,6 @@ def get_python_cmd(
         ]
 
     def handle_agent_log(
-        self, line: str, *, level: Optional[LogLevel], source: LogSource
+        self, line: str, *, level: LogLevel, source: LogSource
     ) -> None:
-        self.log(line, level=level or self.infer_log_level(line, source), source=source)
+        self.log(line, level=level, source=source)

src/isolate/connections/ipc/_base.py

@@ -13,7 +13,6 @@
     Any,
     Callable,
     ContextManager,
-    Optional,
 )
 
 from isolate.backends import (
@@ -221,6 +220,6 @@ def get_python_cmd(
         ]
 
     def handle_agent_log(
-        self, line: str, *, level: Optional[LogLevel], source: LogSource
+        self, line: str, *, level: LogLevel, source: LogSource
     ) -> None:
-        self.log(line, level=level or self.infer_log_level(line, source), source=source)
+        self.log(line, level=level, source=source)