added pagination logic to stdout/stderr

Author: sid-rl

src/runloop_api_client/sdk/async_execution_result.py

@@ -2,9 +2,12 @@
 
 from __future__ import annotations
 
-from typing_extensions import Optional, override
+from typing import Callable, Optional, Awaitable
+from typing_extensions import override
 
 from .._client import AsyncRunloop
+from .._streaming import AsyncStream
+from ..types.devboxes.execution_update_chunk import ExecutionUpdateChunk
 from ..types.devbox_async_execution_detail_view import DevboxAsyncExecutionDetailView
 
 
@@ -48,32 +51,79 @@ def failed(self) -> bool:
         exit_code = self.exit_code
         return exit_code is not None and exit_code != 0
 
-    # TODO: add pagination support once we have it in the API
+    def _count_non_empty_lines(self, text: str) -> int:
+        """Count non-empty lines in text, excluding trailing empty strings."""
+        if not text:
+            return 0
+        # Remove trailing newlines, split, and count non-empty lines
+        return sum(1 for line in text.rstrip("\n").split("\n") if line)
+
+    def _get_last_n_lines(self, text: str, n: int) -> str:
+        """Extract the last N lines from text."""
+        if n <= 0 or not text:
+            return ""
+        # Remove trailing newlines before splitting and slicing
+        return "\n".join(text.rstrip("\n").split("\n")[-n:])
+
+    async def _get_output(
+        self,
+        current_output: str,
+        is_truncated: bool,
+        num_lines: Optional[int],
+        stream_fn: Callable[[], Awaitable[AsyncStream[ExecutionUpdateChunk]]],
+    ) -> str:
+        """Common logic for getting output with optional line limiting and streaming."""
+        # Check if we have enough lines already
+        if num_lines is not None and (not is_truncated or self._count_non_empty_lines(current_output) >= num_lines):
+            return self._get_last_n_lines(current_output, num_lines)
+
+        # Stream full output if truncated
+        if is_truncated:
+            stream = await stream_fn()
+            output = "".join([chunk.output async for chunk in stream])
+            return self._get_last_n_lines(output, num_lines) if num_lines is not None else output
+
+        # Return current output, optionally limited to last N lines
+        return self._get_last_n_lines(current_output, num_lines) if num_lines is not None else current_output
+
     async def stdout(self, num_lines: Optional[int] = None) -> str:
-        text = self._result.stdout or ""
-        return _tail_lines(text, num_lines)
+        """
+        Return captured standard output, streaming full output if truncated.
+
+        Args:
+            num_lines: Optional number of lines to return from the end (most recent)
+
+        Returns:
+            stdout content, optionally limited to last N lines
+        """
+        return await self._get_output(
+            self._result.stdout or "",
+            self._result.stdout_truncated is True,
+            num_lines,
+            lambda: self._client.devboxes.executions.stream_stdout_updates(
+                self.execution_id, devbox_id=self._devbox_id
+            ),
+        )
 
-    # TODO: add pagination support once we have it in the API
     async def stderr(self, num_lines: Optional[int] = None) -> str:
-        text = self._result.stderr or ""
-        return _tail_lines(text, num_lines)
+        """
+        Return captured standard error, streaming full output if truncated.
+
+        Args:
+            num_lines: Optional number of lines to return from the end (most recent)
+
+        Returns:
+            stderr content, optionally limited to last N lines
+        """
+        return await self._get_output(
+            self._result.stderr or "",
+            self._result.stderr_truncated is True,
+            num_lines,
+            lambda: self._client.devboxes.executions.stream_stderr_updates(
+                self.execution_id, devbox_id=self._devbox_id
+            ),
+        )
 
     @property
     def raw(self) -> DevboxAsyncExecutionDetailView:
         return self._result
-
-
-def _tail_lines(text: str, num_lines: Optional[int]) -> str:
-    if not text:
-        return ""
-    if num_lines is None or num_lines <= 0:
-        return text
-
-    lines = text.splitlines()
-    if not lines:
-        return text
-
-    clipped = "\n".join(lines[-num_lines:])
-    if text.endswith("\n"):
-        clipped += "\n"
-    return clipped

src/runloop_api_client/sdk/execution_result.py

@@ -2,10 +2,12 @@
 
 from __future__ import annotations
 
-from typing import Optional
+from typing import Callable, Optional
 from typing_extensions import override
 
 from .._client import Runloop
+from .._streaming import Stream
+from ..types.devboxes.execution_update_chunk import ExecutionUpdateChunk
 from ..types.devbox_async_execution_detail_view import DevboxAsyncExecutionDetailView
 
 
@@ -56,35 +58,79 @@ def failed(self) -> bool:
         exit_code = self.exit_code
         return exit_code is not None and exit_code != 0
 
-    # TODO: add pagination support once we have it in the API
+    def _count_non_empty_lines(self, text: str) -> int:
+        """Count non-empty lines in text, excluding trailing empty strings."""
+        if not text:
+            return 0
+        # Remove trailing newlines, split, and count non-empty lines
+        return sum(1 for line in text.rstrip("\n").split("\n") if line)
+
+    def _get_last_n_lines(self, text: str, n: int) -> str:
+        """Extract the last N lines from text."""
+        if n <= 0 or not text:
+            return ""
+        # Remove trailing newlines before splitting and slicing
+        return "\n".join(text.rstrip("\n").split("\n")[-n:])
+
+    def _get_output(
+        self,
+        current_output: str,
+        is_truncated: bool,
+        num_lines: Optional[int],
+        stream_fn: Callable[[], Stream[ExecutionUpdateChunk]],
+    ) -> str:
+        """Common logic for getting output with optional line limiting and streaming."""
+        # Check if we have enough lines already
+        if num_lines is not None and (not is_truncated or self._count_non_empty_lines(current_output) >= num_lines):
+            return self._get_last_n_lines(current_output, num_lines)
+
+        # Stream full output if truncated
+        if is_truncated:
+            output = "".join(chunk.output for chunk in stream_fn())
+            return self._get_last_n_lines(output, num_lines) if num_lines is not None else output
+
+        # Return current output, optionally limited to last N lines
+        return self._get_last_n_lines(current_output, num_lines) if num_lines is not None else current_output
+
     def stdout(self, num_lines: Optional[int] = None) -> str:
-        """Return captured standard output."""
-        text = self._result.stdout or ""
-        return _tail_lines(text, num_lines)
+        """
+        Return captured standard output, streaming full output if truncated.
+
+        Args:
+            num_lines: Optional number of lines to return from the end (most recent)
+
+        Returns:
+            stdout content, optionally limited to last N lines
+        """
+        return self._get_output(
+            self._result.stdout or "",
+            self._result.stdout_truncated is True,
+            num_lines,
+            lambda: self._client.devboxes.executions.stream_stdout_updates(
+                self.execution_id, devbox_id=self._devbox_id
+            ),
+        )
 
-    # TODO: add pagination support once we have it in the API
     def stderr(self, num_lines: Optional[int] = None) -> str:
-        """Return captured standard error."""
-        text = self._result.stderr or ""
-        return _tail_lines(text, num_lines)
+        """
+        Return captured standard error, streaming full output if truncated.
+
+        Args:
+            num_lines: Optional number of lines to return from the end (most recent)
+
+        Returns:
+            stderr content, optionally limited to last N lines
+        """
+        return self._get_output(
+            self._result.stderr or "",
+            self._result.stderr_truncated is True,
+            num_lines,
+            lambda: self._client.devboxes.executions.stream_stderr_updates(
+                self.execution_id, devbox_id=self._devbox_id
+            ),
+        )
 
     @property
     def raw(self) -> DevboxAsyncExecutionDetailView:
         """Access the underlying API response."""
         return self._result
-
-
-def _tail_lines(text: str, num_lines: Optional[int]) -> str:
-    if not text:
-        return ""
-    if num_lines is None or num_lines <= 0:
-        return text
-
-    lines = text.splitlines()
-    if not lines:
-        return text
-
-    clipped = "\n".join(lines[-num_lines:])
-    if text.endswith("\n"):
-        clipped += "\n"
-    return clipped