fix(cli): bound how much log a single drain reads into memory [PC-4873]

packages/uipath/src/uipath/_cli/_server_jobs.py

@@ -189,6 +189,10 @@ def build_result_payload(job_key: str, outcome: dict[str, Any]) -> dict[str, Any
 
 LOG_POLL_SECONDS = 0.25
 LOG_BATCH_MAX_LINES = 200
+# Cap on how much log a single drain pulls into memory. Without it a job that logs
+# heavily between polls -- or the final drain after a long run -- reads the whole
+# remainder at once, and the decoded str costs more again.
+LOG_READ_CHUNK_BYTES = 256 * 1024
 LOG_FLUSH_TIMEOUT_SECONDS = 10
 
 # How long a cancelled job gets to unwind before we admit the stop did not take.
@@ -249,31 +253,48 @@ def stop(self) -> None:
         self._stop.set()
 
     async def _drain(self, final: bool = False) -> None:
-        try:
-            if not os.path.exists(self.path):
+        """Forward whatever is new, a bounded window at a time until caught up."""
+        while True:
+            try:
+                if not os.path.exists(self.path):
+                    return
+                # Binary with explicit offsets: a text-mode tell() cookie is opaque, and
+                # we need to rewind past an unterminated tail.
+                with open(self.path, "rb") as f:
+                    f.seek(self._offset)
+                    chunk = f.read(LOG_READ_CHUNK_BYTES)
+            except OSError:
                 return
-            # Binary with explicit offsets: a text-mode tell() cookie is opaque, and we
-            # need to rewind past an unterminated tail.
-            with open(self.path, "rb") as f:
-                f.seek(self._offset)
-                chunk = f.read()
-        except OSError:
-            return
 
-        if not chunk:
-            return
-
-        # A logging handler writes the record and only then flushes, so the tail can be
-        # half a line. Leave it for the next poll rather than reporting a fragment as a
-        # complete entry — except on the final drain, where nothing more is coming.
-        if not final and not chunk.endswith(b"\n"):
-            cut = chunk.rfind(b"\n")
-            if cut == -1:
+            if not chunk:
                 return
-            chunk = chunk[: cut + 1]
 
-        self._offset += len(chunk)
+            # A short read means we reached the end of the file; a full one means there
+            # is more waiting and we should come back round.
+            more_pending = len(chunk) == LOG_READ_CHUNK_BYTES
+
+            if not chunk.endswith(b"\n"):
+                # A logging handler writes the record and only then flushes, so the tail
+                # can be half a line.
+                cut = chunk.rfind(b"\n")
+                if cut != -1:
+                    # Trim to the last complete line. Safe to decode: \n never appears
+                    # inside a multi-byte UTF-8 sequence.
+                    chunk = chunk[: cut + 1]
+                elif not (more_pending or final):
+                    # Partial line still being written — wait for its newline.
+                    return
+                # Otherwise: a single line longer than the read window, or the last drain
+                # with nothing more coming. Emit the fragment rather than buffering
+                # without bound.
+
+            self._offset += len(chunk)
+            await self._emit(chunk)
+
+            if not more_pending:
+                return
 
+    async def _emit(self, chunk: bytes) -> None:
         batch: list[dict[str, Any]] = []
         for raw in chunk.decode("utf-8", errors="replace").splitlines():
             line = raw.rstrip("\r")

packages/uipath/tests/cli/test_server_logs.py

@@ -294,3 +294,52 @@ async def test_tailer_tolerates_a_missing_file(tmp_path):
     await asyncio.wait_for(tailer.run(), timeout=5)
 
     assert callback.batches == []
+
+
+# --------------------------------------------------------------------------- #
+# bounded reads                                                               #
+# --------------------------------------------------------------------------- #
+
+
+async def test_tailer_catches_up_across_multiple_read_windows(tmp_path, monkeypatch):
+    """A job that logs more than one window between polls must still be forwarded in
+    full — bounding the read must not mean dropping the remainder."""
+    import uipath._cli._server_jobs as jobs
+
+    monkeypatch.setattr(jobs, "LOG_READ_CHUNK_BYTES", 512)
+
+    log_file = tmp_path / "execution.log"
+    total = 200  # ~40 bytes each, comfortably several windows
+    log_file.write_bytes(
+        b"".join(b"[2026-07-29 17:00:00,000][INFO] line-%d\n" % i for i in range(total))
+    )
+
+    callback = RecordingCallback()
+    tailer = JobLogTailer("job-1", str(log_file), callback)
+    tailer.stop()
+    await asyncio.wait_for(tailer.run(), timeout=10)
+
+    messages = [line["message"] for line in callback.lines]
+    assert len(messages) == total
+    assert messages[0] == "line-0"
+    assert messages[-1] == f"line-{total - 1}"
+
+
+async def test_tailer_emits_a_line_longer_than_the_read_window(tmp_path, monkeypatch):
+    """A single pathologically long line must not stall the tailer forever: bounded
+    memory beats perfect line integrity here."""
+    import uipath._cli._server_jobs as jobs
+
+    monkeypatch.setattr(jobs, "LOG_READ_CHUNK_BYTES", 64)
+
+    log_file = tmp_path / "execution.log"
+    log_file.write_bytes(b"x" * 500 + b"\n[2026-07-29 17:00:00,000][INFO] after\n")
+
+    callback = RecordingCallback()
+    tailer = JobLogTailer("job-1", str(log_file), callback)
+    tailer.stop()
+    await asyncio.wait_for(tailer.run(), timeout=10)
+
+    joined = "".join(line["message"] for line in callback.lines)
+    assert "x" * 500 in joined, "the long line must be forwarded, even if fragmented"
+    assert any(line["message"] == "after" for line in callback.lines)