paginated stdout/stderr in ExecutionResult (#670)

* added pagination logic to stdout/stderr

* unit test adjustments

* added smoke tests and todos for fixing and testing output line counting logic

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,85 @@ 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."""
+        # TODO: Fix inconsistency - _count_non_empty_lines counts non-empty lines but
+        # _get_last_n_lines returns N lines (may include empty ones). This means
+        # num_lines=50 might return fewer than 50 non-empty lines. Should either:
+        # 1. Make _get_last_n_lines return N non-empty lines, OR
+        # 2. Make _count_non_empty_lines count all lines
+        # This affects both Python and TypeScript SDKs - fix together.
+        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,85 @@ 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."""
+        # TODO: Fix inconsistency - _count_non_empty_lines counts non-empty lines but
+        # _get_last_n_lines returns N lines (may include empty ones). This means
+        # num_lines=50 might return fewer than 50 non-empty lines. Should either:
+        # 1. Make _get_last_n_lines return N non-empty lines, OR
+        # 2. Make _count_non_empty_lines count all lines
+        # This affects both Python and TypeScript SDKs - fix together.
+        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

tests/sdk/async_devbox/test_core.py

@@ -137,11 +137,9 @@ async def test_shutdown(self, mock_async_client: AsyncMock, devbox_view: MockDev
     async def test_suspend(self, mock_async_client: AsyncMock, devbox_view: MockDevboxView) -> None:
         """Test suspend method."""
         mock_async_client.devboxes.suspend = AsyncMock(return_value=devbox_view)
-        polling_config = PollingConfig(timeout_seconds=60.0)
 
         devbox = AsyncDevbox(mock_async_client, "dev_123")
         result = await devbox.suspend(
-            polling_config=polling_config,
             extra_headers={"X-Custom": "value"},
             extra_query={"param": "value"},
             extra_body={"key": "value"},
@@ -152,7 +150,6 @@ async def test_suspend(self, mock_async_client: AsyncMock, devbox_view: MockDevb
         assert result == devbox_view
         mock_async_client.devboxes.suspend.assert_called_once_with(
             "dev_123",
-            polling_config=polling_config,
             extra_headers={"X-Custom": "value"},
             extra_query={"param": "value"},
             extra_body={"key": "value"},
@@ -164,11 +161,9 @@ async def test_suspend(self, mock_async_client: AsyncMock, devbox_view: MockDevb
     async def test_resume(self, mock_async_client: AsyncMock, devbox_view: MockDevboxView) -> None:
         """Test resume method."""
         mock_async_client.devboxes.resume = AsyncMock(return_value=devbox_view)
-        polling_config = PollingConfig(timeout_seconds=60.0)
 
         devbox = AsyncDevbox(mock_async_client, "dev_123")
         result = await devbox.resume(
-            polling_config=polling_config,
             extra_headers={"X-Custom": "value"},
             extra_query={"param": "value"},
             extra_body={"key": "value"},
@@ -179,7 +174,6 @@ async def test_resume(self, mock_async_client: AsyncMock, devbox_view: MockDevbo
         assert result == devbox_view
         mock_async_client.devboxes.resume.assert_called_once_with(
             "dev_123",
-            polling_config=polling_config,
             extra_headers={"X-Custom": "value"},
             extra_query={"param": "value"},
             extra_body={"key": "value"},

tests/sdk/conftest.py

@@ -55,6 +55,8 @@ class MockExecutionView:
     exit_status: int = 0
     stdout: str = "output"
     stderr: str = ""
+    stdout_truncated: bool = False
+    stderr_truncated: bool = False
 
 
 @dataclass

tests/sdk/test_async_clients.py

@@ -215,20 +215,6 @@ async def test_create(self, mock_async_client: AsyncMock, object_view: MockObjec
             metadata={"key": "value"},
         )
 
-    @pytest.mark.asyncio
-    async def test_create_auto_detect_content_type(
-        self, mock_async_client: AsyncMock, object_view: MockObjectView
-    ) -> None:
-        """Test create auto-detects content type."""
-        mock_async_client.objects.create = AsyncMock(return_value=object_view)
-
-        client = AsyncStorageObjectClient(mock_async_client)
-        obj = await client.create(name="test.txt")
-
-        assert isinstance(obj, AsyncStorageObject)
-        call_kwargs = mock_async_client.objects.create.call_args[1]
-        assert "content_type" not in call_kwargs
-
     def test_from_id(self, mock_async_client: AsyncMock) -> None:
         """Test from_id method."""
         client = AsyncStorageObjectClient(mock_async_client)

tests/sdk/test_async_execution.py

@@ -159,6 +159,8 @@ async def test_result_needs_polling(self, mock_async_client: AsyncMock) -> None:
             exit_status=0,
             stdout="output",
             stderr="",
+            stdout_truncated=False,
+            stderr_truncated=False,
         )
 
         mock_async_client.devboxes.wait_for_command = AsyncMock(return_value=completed_execution)
@@ -263,19 +265,3 @@ async def test_kill(self, mock_async_client: AsyncMock, execution_view: MockExec
             "exec_123",
             devbox_id="dev_123",
         )
-
-    @pytest.mark.asyncio
-    async def test_kill_with_process_group(
-        self, mock_async_client: AsyncMock, execution_view: MockExecutionView
-    ) -> None:
-        """Test kill with kill_process_group."""
-        mock_async_client.devboxes.executions.kill = AsyncMock(return_value=None)
-
-        execution = AsyncExecution(mock_async_client, "dev_123", execution_view)  # type: ignore[arg-type]
-        await execution.kill(kill_process_group=True)
-
-        mock_async_client.devboxes.executions.kill.assert_awaited_once_with(
-            "exec_123",
-            devbox_id="dev_123",
-            kill_process_group=True,
-        )