agents: support bounded replay and backward history paging on /stream

examples/agents/page_history.py

@@ -0,0 +1,60 @@
+"""Page backwards through a session's older event history.
+
+A plain attach replays only the newest events the server keeps within its
+replay budget, so long-lived sessions have history that never arrives on the
+live feed. This walks further back a page at a time using ``before``, the same
+way a UI would build scrollback.
+
+Required env:
+  DIGITALOCEAN_TOKEN
+  SESSION_ID
+  PYDO_AGENTS_ENDPOINT  (stage2: https://api.s2r1.internal.digitalocean.com)
+
+Optional env:
+  PAGE_SIZE   events per page (server default is 200)
+  MAX_PAGES   stop after this many pages (default 5)
+"""
+
+import os
+import sys
+
+from pydo import Client
+
+SESSION_ID = os.environ["SESSION_ID"]
+PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "50"))
+MAX_PAGES = int(os.environ.get("MAX_PAGES", "5"))
+
+client = Client(
+    token=os.environ["DIGITALOCEAN_TOKEN"],
+    agents_endpoint=os.environ.get("PYDO_AGENTS_ENDPOINT"),
+)
+sessions = client.agents.sessions
+
+# Start from the oldest event of a bounded replay: that is the cursor for the
+# page before it.
+with sessions.stream(SESSION_ID, replay_only=True) as replay:
+    recent = list(replay)
+
+if not recent:
+    print("session has no events to page back from", file=sys.stderr)
+    raise SystemExit(0)
+
+cursor = replay.oldest_event_id
+print(f"replay returned {len(recent)} events, oldest={cursor}", file=sys.stderr)
+
+older = []
+for page_number in range(1, MAX_PAGES + 1):
+    page = sessions.history_page(SESSION_ID, before=cursor, limit=PAGE_SIZE)
+    older = page.events + older
+    print(
+        f"page {page_number}: {len(page.events)} events "
+        f"(has_more={page.has_more}, next before={page.next_before})",
+        file=sys.stderr,
+    )
+    if not page.has_more or not page.next_before:
+        break
+    cursor = page.next_before
+
+print(f"\nfetched {len(older)} older events, oldest first:", file=sys.stderr)
+for event in older:
+    print(f"  {event.get('event_id')}  {event.get('type')}")

src/pydo/agents/__init__.py

@@ -32,6 +32,7 @@
 from .custom_sessions import (
     HarnessEventStream,
     HarnessStreamError,
+    HistoryPage,
     SessionsOperations,
     WorkspaceDownload,
     WorkspaceTransferError,
@@ -129,6 +130,7 @@ def attach_by_name(self, name: str) -> AgentSession:
     "TriggersOperations",
     "HarnessEventStream",
     "HarnessStreamError",
+    "HistoryPage",
     "WorkspaceDownload",
     "WorkspaceTransferError",
     "DEFAULT_AGENTS_BASE_URL",

src/pydo/agents/custom_sessions.py

@@ -11,7 +11,7 @@
 import os
 import time
 import warnings
-from typing import Any, BinaryIO, Dict, Iterator, List, Optional, Union
+from typing import Any, BinaryIO, Dict, Iterator, List, NamedTuple, Optional, Union
 from urllib.error import HTTPError, URLError
 from urllib.parse import quote
 from urllib.request import Request, urlopen
@@ -269,11 +269,33 @@ def _raise_agents_http_error(response) -> None:
     raise HttpResponseError(message=message, response=response)
 
 
+class HistoryPage(NamedTuple):
+    """One backward page of session history.
+
+    ``next_before`` is the cursor to pass as ``before`` for the page before
+    this one; it is ``None`` when the page came back empty.
+    """
+
+    events: List[Any]
+    has_more: Optional[bool]
+    next_before: Optional[str]
+
+
 class HarnessEventStream:
     """Unwraps grpc-gateway SSE envelopes ``{result, error}`` into harness Events."""
 
     def __init__(self, sse_stream: SSEStream):
         self._sse = sse_stream
+        self.oldest_event_id: Optional[str] = None
+
+    @property
+    def has_more(self) -> Optional[bool]:
+        """Whether older history remains, per the server's trailing comment.
+
+        Only history pages (``before=``) carry this; ``None`` until the
+        ``: has_more=...`` frame arrives, so read it after iterating.
+        """
+        return getattr(self._sse, "has_more", None)
 
     def __iter__(self) -> Iterator[Any]:
         for chunk in self._sse:
@@ -290,6 +312,10 @@ def __iter__(self) -> Iterator[Any]:
                 )
             event = _unwrap_harness_sse_chunk(chunk)
             if event is not None:
+                if self.oldest_event_id is None:
+                    event_id = _field(event, "event_id")
+                    if event_id:
+                        self.oldest_event_id = str(event_id)
                 yield event
 
     def close(self) -> None:
@@ -491,10 +517,35 @@ def stream(
         *,
         replay_from: Optional[str] = None,
         replay_only: bool = False,
+        before: Optional[str] = None,
+        limit: Optional[int] = None,
     ) -> HarnessEventStream:
+        """Attach to a session's SSE event feed.
+
+        A cursorless attach replays only the newest events the server keeps
+        within its replay budget, then goes live — it is not the session's
+        full history. Older history is read a page at a time with ``before``,
+        an ``event_id`` to page backwards from (exclusive): the server sends
+        up to ``limit`` older events, oldest-first, then closes without going
+        live. ``before`` implies ``replay_only``, which the server requires.
+
+        Prefer :meth:`history_page` for scrollback; it drains one page and
+        hands back the next cursor.
+        """
+        if limit is not None:
+            if before is None:
+                raise ValueError("limit is only meaningful together with before")
+            if int(limit) < 1:
+                raise ValueError("limit must be a positive integer")
+
         params: Dict[str, Any] = {}
         if replay_from:
             params["replay_from"] = replay_from
+        if before:
+            params["before"] = before
+            replay_only = True
+        if limit is not None:
+            params["limit"] = int(limit)
         if replay_only:
             params["replay_only"] = "true"
 
@@ -511,6 +562,34 @@ def stream(
             _raise_agents_http_error(response)
         return HarnessEventStream(SSEStream(response))
 
+    def history_page(
+        self,
+        session_id: str,
+        *,
+        before: str,
+        limit: Optional[int] = None,
+    ) -> HistoryPage:
+        """Read one page of history older than ``before``, oldest-first.
+
+        Walk backwards by feeding ``next_before`` into the next call::
+
+            cursor = oldest_event_id_you_hold
+            while cursor:
+                page = sessions.history_page(session_id, before=cursor)
+                older = page.events + older
+                cursor = page.next_before if page.has_more else None
+        """
+        if not before:
+            raise ValueError("before is required")
+        stream = self.stream(session_id, before=before, limit=limit)
+        with stream:
+            events = list(stream)
+        return HistoryPage(
+            events=events,
+            has_more=stream.has_more,
+            next_before=stream.oldest_event_id,
+        )
+
     def _transfers_path(self, session_id: str, *parts: str) -> str:
         path = f"{_BASE_PATH}/{_quote(session_id)}/{_TRANSFERS_SUFFIX}"
         for part in parts:
@@ -934,6 +1013,7 @@ def __exit__(self, *args: Any) -> None:
     "SessionsOperations",
     "HarnessEventStream",
     "HarnessStreamError",
+    "HistoryPage",
     "WorkspaceDownload",
     "WorkspaceTransferError",
     "UploadData",

src/pydo/agents/session.py

@@ -378,6 +378,13 @@ def resume(self) -> Any:
     def stream(self, **kwargs: Any) -> Any:
         return self._sessions.stream(self.session_id, **kwargs)
 
+    def history(self, *, before: str, limit: Optional[int] = None) -> Any:
+        """Read one page of history older than ``before``.
+
+        See :meth:`SessionsOperations.history_page`.
+        """
+        return self._sessions.history_page(self.session_id, before=before, limit=limit)
+
     def resolve_hitl(
         self,
         request_id: str,

src/pydo/aio/agents/custom_sessions.py

@@ -26,6 +26,7 @@
     _TRANSFERS_SUFFIX,
     _YAML_MEDIA_TYPE,
     HarnessStreamError,
+    HistoryPage,
     UploadData,
     WorkspaceTransferError,
     _coerce_upload_content,
@@ -95,6 +96,16 @@ async def _aio_http_get_iter(url: str) -> AsyncIterator[bytes]:
 class AsyncHarnessEventStream:
     def __init__(self, sse_stream: AsyncSSEStream):
         self._sse = sse_stream
+        self.oldest_event_id: Optional[str] = None
+
+    @property
+    def has_more(self) -> Optional[bool]:
+        """Whether older history remains, per the server's trailing comment.
+
+        Only history pages (``before=``) carry this; ``None`` until the
+        ``: has_more=...`` frame arrives, so read it after iterating.
+        """
+        return getattr(self._sse, "has_more", None)
 
     def __aiter__(self) -> AsyncIterator[Any]:
         return self._iter()
@@ -114,6 +125,10 @@ async def _iter(self) -> AsyncIterator[Any]:
                 )
             event = _unwrap_harness_sse_chunk(chunk)
             if event is not None:
+                if self.oldest_event_id is None:
+                    event_id = _field(event, "event_id")
+                    if event_id:
+                        self.oldest_event_id = str(event_id)
                 yield event
 
     async def close(self) -> None:
@@ -294,10 +309,35 @@ async def stream(
         *,
         replay_from: Optional[str] = None,
         replay_only: bool = False,
+        before: Optional[str] = None,
+        limit: Optional[int] = None,
     ) -> AsyncHarnessEventStream:
+        """Attach to a session's SSE event feed.
+
+        A cursorless attach replays only the newest events the server keeps
+        within its replay budget, then goes live — it is not the session's
+        full history. Older history is read a page at a time with ``before``,
+        an ``event_id`` to page backwards from (exclusive): the server sends
+        up to ``limit`` older events, oldest-first, then closes without going
+        live. ``before`` implies ``replay_only``, which the server requires.
+
+        Prefer :meth:`history_page` for scrollback; it drains one page and
+        hands back the next cursor.
+        """
+        if limit is not None:
+            if before is None:
+                raise ValueError("limit is only meaningful together with before")
+            if int(limit) < 1:
+                raise ValueError("limit must be a positive integer")
+
         params: Dict[str, Any] = {}
         if replay_from:
             params["replay_from"] = replay_from
+        if before:
+            params["before"] = before
+            replay_only = True
+        if limit is not None:
+            params["limit"] = int(limit)
         if replay_only:
             params["replay_only"] = "true"
 
@@ -315,6 +355,33 @@ async def stream(
             _raise_agents_http_error(response)
         return AsyncHarnessEventStream(AsyncSSEStream(response))
 
+    async def history_page(
+        self,
+        session_id: str,
+        *,
+        before: str,
+        limit: Optional[int] = None,
+    ) -> HistoryPage:
+        """Read one page of history older than ``before``, oldest-first.
+
+        Walk backwards by feeding ``next_before`` into the next call::
+
+            cursor = oldest_event_id_you_hold
+            while cursor:
+                page = await sessions.history_page(session_id, before=cursor)
+                cursor = page.next_before if page.has_more else None
+        """
+        if not before:
+            raise ValueError("before is required")
+        stream = await self.stream(session_id, before=before, limit=limit)
+        async with stream:
+            events = [event async for event in stream]
+        return HistoryPage(
+            events=events,
+            has_more=stream.has_more,
+            next_before=stream.oldest_event_id,
+        )
+
     def _transfers_path(self, session_id: str, *parts: str) -> str:
         path = f"{_BASE_PATH}/{_quote(session_id)}/{_TRANSFERS_SUFFIX}"
         for part in parts:

src/pydo/aio/agents/session.py

@@ -216,6 +216,15 @@ async def resume(self) -> Any:
     async def stream(self, **kwargs: Any) -> Any:
         return await self._sessions.stream(self.session_id, **kwargs)
 
+    async def history(self, *, before: str, limit: Optional[int] = None) -> Any:
+        """Read one page of history older than ``before``.
+
+        See :meth:`AsyncSessionsOperations.history_page`.
+        """
+        return await self._sessions.history_page(
+            self.session_id, before=before, limit=limit
+        )
+
     async def resolve_hitl(
         self,
         request_id: str,