release: bump version to 0.39.0

backend/app/services/system_logs.py

@@ -72,6 +72,7 @@ EMBEDDED_LEVEL_PATTERN = re.compile(
     r"\b(CRITICAL|FATAL|ERROR|WARNING|WARN|INFO|DEBUG|TRACE)\b",
     re.IGNORECASE,
 )
+CONTROL_CHAR_PATTERN = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
 
 
 @dataclass(frozen=True)
@@ -260,19 +261,19 @@ def parse_prefixed_timestamp(line: str) -> tuple[datetime | None, str]:
     return None, stripped
 
 
-def infer_log_level_from_text(text: str) -> str | None:
+def infer_log_level_from_text(text: str, *, allow_embedded: bool = True) -> str | None:
     leading_match = LEADING_LEVEL_PATTERN.match(text)
     if leading_match:
         return normalize_log_level(leading_match.group(1))
 
-    embedded_match = EMBEDDED_LEVEL_PATTERN.search(text)
-    if embedded_match:
-        return normalize_log_level(embedded_match.group(1))
-
-    upper_text = text.upper()
-    for pattern, normalized in LEVEL_PATTERNS:
-        if f"{pattern}:" in upper_text or f"{pattern} " in upper_text:
-            return normalized
+    if allow_embedded:
+        embedded_match = EMBEDDED_LEVEL_PATTERN.search(text)
+        if embedded_match:
+            return normalize_log_level(embedded_match.group(1))
+        upper_text = text.upper()
+        for pattern, normalized in LEVEL_PATTERNS:
+            if f"{pattern}:" in upper_text or f"{pattern} " in upper_text:
+                return normalized
     return None
 
 
@@ -288,10 +289,15 @@ def build_display_line(timestamp: datetime | None, level: str | None, message: s
     return " ".join(parts).strip()
 
 
+def sanitize_text_log_line(line: str) -> str:
+    return CONTROL_CHAR_PATTERN.sub("", line)
+
+
 def parse_text_log_entry(line: str) -> StructuredLogEntry:
-    timestamp, remainder = parse_prefixed_timestamp(line)
-    level = infer_log_level_from_text(remainder or line)
-    display_line = line.rstrip("\n")
+    sanitized_line = sanitize_text_log_line(line).rstrip("\n")
+    timestamp, remainder = parse_prefixed_timestamp(sanitized_line)
+    level = infer_log_level_from_text(remainder or sanitized_line, allow_embedded=False)
+    display_line = sanitized_line
     return StructuredLogEntry(
         timestamp=timestamp,
         level=level,
@@ -338,7 +344,11 @@ def read_file_entries(source: LogSource, scan_limit: int) -> list[StructuredLogE
         return []
     with path.open("r", encoding="utf-8", errors="replace") as handle:
         recent_lines = deque(handle, maxlen=scan_limit)
-    return [parse_text_log_entry(line) for line in recent_lines if line.strip()]
+    return [
+        parse_text_log_entry(line)
+        for line in recent_lines
+        if sanitize_text_log_line(line).strip()
+    ]
 
 
 def read_docker_entries(source: LogSource, scan_limit: int) -> list[StructuredLogEntry]: