diff --git a/examples/agents/page_history.py b/examples/agents/page_history.py new file mode 100644 index 00000000..b5f63a4f --- /dev/null +++ b/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')}") diff --git a/src/pydo/agents/__init__.py b/src/pydo/agents/__init__.py index e23e0be6..c661a4fb 100644 --- a/src/pydo/agents/__init__.py +++ b/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", diff --git a/src/pydo/agents/custom_sessions.py b/src/pydo/agents/custom_sessions.py index 6a691628..3b51b8b7 100644 --- a/src/pydo/agents/custom_sessions.py +++ b/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", diff --git a/src/pydo/agents/session.py b/src/pydo/agents/session.py index 3b0acdde..de9760c7 100644 --- a/src/pydo/agents/session.py +++ b/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, diff --git a/src/pydo/aio/agents/custom_sessions.py b/src/pydo/aio/agents/custom_sessions.py index 1cce2341..a12d6039 100644 --- a/src/pydo/aio/agents/custom_sessions.py +++ b/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: diff --git a/src/pydo/aio/agents/session.py b/src/pydo/aio/agents/session.py index e433324e..87f821de 100644 --- a/src/pydo/aio/agents/session.py +++ b/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, diff --git a/src/pydo/custom_extensions.py b/src/pydo/custom_extensions.py index dc52efe0..987eca14 100644 --- a/src/pydo/custom_extensions.py +++ b/src/pydo/custom_extensions.py @@ -462,6 +462,20 @@ async def _auto_streaming_call( # SSE stream iterators # --------------------------------------------------------------------------- +_HAS_MORE_RE = re.compile(r"^has_more\s*=\s*(true|false)$", re.IGNORECASE) + + +def _parse_has_more_comment(line: str) -> Optional[bool]: + """Read ``has_more`` out of an SSE comment line (``: has_more=true``). + + Returns ``None`` for any other comment, such as harness-api's + ``: connected to `` banner. + """ + match = _HAS_MORE_RE.match(line.lstrip(":").strip()) + if match is None: + return None + return match.group(1).lower() == "true" + class SSEStream: """Synchronous iterator over Server-Sent Events. @@ -485,10 +499,14 @@ class SSEStream: For automatic retries on **transient** transport errors **before any chunk is yielded**, see :func:`iter_sse_with_retry`. + + :ivar has_more: ``True``/``False`` once the server has sent a + ``: has_more=...`` comment (harness history pages), else ``None``. """ def __init__(self, response: Any): self._response = response + self.has_more: Optional[bool] = None def __iter__(self) -> Iterator[dict]: return self._iter_events() @@ -505,6 +523,11 @@ def _iter_events(self) -> Iterator[dict]: line = line.strip() if not line: continue + if line.startswith(":"): + has_more = _parse_has_more_comment(line) + if has_more is not None: + self.has_more = has_more + continue if line.startswith("data:"): data = line[5:].strip() if data == "[DONE]": @@ -546,10 +569,14 @@ class AsyncSSEStream: Transport and decode errors match :class:`SSEStream`. See :func:`async_iter_sse_with_retry` for retries before the first chunk. + + :ivar has_more: ``True``/``False`` once the server has sent a + ``: has_more=...`` comment (harness history pages), else ``None``. """ def __init__(self, response: Any): self._response = response + self.has_more: Optional[bool] = None def __aiter__(self) -> AsyncIterator[dict]: return self._iter_events() @@ -566,6 +593,11 @@ async def _iter_events(self) -> AsyncIterator[dict]: line = line.strip() if not line: continue + if line.startswith(":"): + has_more = _parse_has_more_comment(line) + if has_more is not None: + self.has_more = has_more + continue if line.startswith("data:"): data = line[5:].strip() if data == "[DONE]": diff --git a/tests/agents/test_async_sessions.py b/tests/agents/test_async_sessions.py index 3de3b91f..63ff5a63 100644 --- a/tests/agents/test_async_sessions.py +++ b/tests/agents/test_async_sessions.py @@ -18,7 +18,7 @@ class _FakeAsyncResponse: - def __init__(self, status_code: int, body: Any = None): + def __init__(self, status_code: int, body: Any = None, *, sse_chunks=None): self.status_code = status_code if isinstance(body, (dict, list)): self._body_bytes = json.dumps(body).encode("utf-8") @@ -28,6 +28,7 @@ def __init__(self, status_code: int, body: Any = None): self._body_bytes = body else: self._body_bytes = b"" + self._sse = sse_chunks async def read(self) -> bytes: return self._body_bytes @@ -38,6 +39,13 @@ def text(self) -> str: def body(self) -> bytes: return self._body_bytes + async def iter_bytes(self): + for chunk in self._sse or []: + yield chunk + + def close(self) -> None: + pass + class _FakeAsyncPipeline: def __init__(self, responses: List[_FakeAsyncResponse]): @@ -150,3 +158,124 @@ async def test_async_attach_by_name_raises_when_not_found(): resources = _make_async_resources([_FakeAsyncResponse(200, {"sessions": []})]) with pytest.raises(LookupError): await resources.attach_by_name("missing") + + +# --------------------------------------------------------------------------- +# Backward history paging +# --------------------------------------------------------------------------- + +_HISTORY_PAGE_SSE = ( + b": connected to s1\n\n" + b'data: {"event_id":"e10","type":"run.token_delta","data":{"text":"older "}}\n\n' + b'data: {"event_id":"e11","type":"run.token_delta","data":{"text":"newer"}}\n\n' + b": has_more=true\n\n" +) + + +async def _drain(stream) -> list: + return [event async for event in stream] + + +@pytest.mark.asyncio +async def test_async_stream_before_implies_replay_only_and_sends_limit(): + resources = _make_async_resources([_FakeAsyncResponse(200, sse_chunks=[b""])]) + + await _drain(await resources.sessions.stream("s1", before="evt-99", limit=50)) + + url = resources._proxy._original._pipeline.calls[0].request.url + assert "before=evt-99" in url + assert "limit=50" in url + assert "replay_only=true" in url + + +@pytest.mark.asyncio +async def test_async_stream_omits_paging_params_when_unset(): + resources = _make_async_resources([_FakeAsyncResponse(200, sse_chunks=[b""])]) + + await _drain(await resources.sessions.stream("s1")) + + url = resources._proxy._original._pipeline.calls[0].request.url + assert "before=" not in url + assert "limit=" not in url + assert "replay_only=" not in url + + +@pytest.mark.asyncio +async def test_async_stream_rejects_limit_without_before(): + resources = _make_async_resources([]) + with pytest.raises(ValueError, match="before"): + await resources.sessions.stream("s1", limit=10) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("limit", [0, -1]) +async def test_async_stream_rejects_non_positive_limit(limit): + resources = _make_async_resources([]) + with pytest.raises(ValueError, match="positive"): + await resources.sessions.stream("s1", before="evt-99", limit=limit) + + +@pytest.mark.asyncio +async def test_async_stream_records_has_more_comment(): + resources = _make_async_resources( + [_FakeAsyncResponse(200, sse_chunks=[_HISTORY_PAGE_SSE])] + ) + + stream = await resources.sessions.stream("s1", before="evt-12") + assert stream.has_more is None + + events = await _drain(stream) + assert [e.event_id for e in events] == ["e10", "e11"] + assert stream.has_more is True + assert stream.oldest_event_id == "e10" + + +@pytest.mark.asyncio +async def test_async_history_page_returns_events_cursor_and_has_more(): + resources = _make_async_resources( + [_FakeAsyncResponse(200, sse_chunks=[_HISTORY_PAGE_SSE])] + ) + + page = await resources.sessions.history_page("s1", before="evt-12", limit=2) + + url = resources._proxy._original._pipeline.calls[0].request.url + assert "before=evt-12" in url + assert "limit=2" in url + assert "replay_only=true" in url + + events, has_more, next_before = page + assert [e.event_id for e in events] == ["e10", "e11"] + assert has_more is True + assert next_before == "e10" + + +@pytest.mark.asyncio +async def test_async_history_page_empty_has_no_cursor(): + resources = _make_async_resources( + [_FakeAsyncResponse(200, sse_chunks=[b": has_more=false\n\n"])] + ) + + page = await resources.sessions.history_page("s1", before="e1") + assert page.events == [] + assert page.has_more is False + assert page.next_before is None + + +@pytest.mark.asyncio +async def test_async_history_page_requires_before(): + resources = _make_async_resources([]) + with pytest.raises(ValueError, match="before"): + await resources.sessions.history_page("s1", before="") + + +@pytest.mark.asyncio +async def test_async_agent_session_history_binds_session_id(): + resources = _make_async_resources( + [_FakeAsyncResponse(200, sse_chunks=[_HISTORY_PAGE_SSE])] + ) + + page = await resources.attach("s1").history(before="evt-12") + + url = resources._proxy._original._pipeline.calls[0].request.url + assert "/v2/agents/sessions/s1/stream" in url + assert page.next_before == "e10" diff --git a/tests/agents/test_sessions.py b/tests/agents/test_sessions.py index 439f0bab..87644c6f 100644 --- a/tests/agents/test_sessions.py +++ b/tests/agents/test_sessions.py @@ -330,6 +330,132 @@ def test_stream_passes_replay_query_params(): assert "replay_only=true" in call.request.url +# --------------------------------------------------------------------------- +# Backward history paging +# --------------------------------------------------------------------------- + +_HISTORY_PAGE_SSE = ( + b": connected to s1\n\n" + b'data: {"event_id":"e10","type":"run.token_delta","data":{"text":"older "}}\n\n' + b'data: {"event_id":"e11","type":"run.token_delta","data":{"text":"newer"}}\n\n' + b": has_more=true\n\n" +) + + +def test_stream_before_implies_replay_only_and_sends_limit(): + resources = _make_resources([_FakeResponse(200, sse_chunks=[b""])]) + + list(resources.sessions.stream("s1", before="evt-99", limit=50)) + + url = resources._proxy._original._pipeline.calls[0].request.url + assert "before=evt-99" in url + assert "limit=50" in url + assert "replay_only=true" in url + + +def test_stream_omits_paging_params_when_unset(): + resources = _make_resources([_FakeResponse(200, sse_chunks=[b""])]) + + list(resources.sessions.stream("s1")) + + url = resources._proxy._original._pipeline.calls[0].request.url + assert "before=" not in url + assert "limit=" not in url + assert "replay_only=" not in url + + +def test_stream_rejects_limit_without_before(): + resources = _make_resources([]) + with pytest.raises(ValueError, match="before"): + resources.sessions.stream("s1", limit=10) + + +@pytest.mark.parametrize("limit", [0, -1]) +def test_stream_rejects_non_positive_limit(limit): + resources = _make_resources([]) + with pytest.raises(ValueError, match="positive"): + resources.sessions.stream("s1", before="evt-99", limit=limit) + + +def test_stream_records_has_more_comment(): + resources = _make_resources([_FakeResponse(200, sse_chunks=[_HISTORY_PAGE_SSE])]) + + stream = resources.sessions.stream("s1", before="evt-12") + assert stream.has_more is None # not known until the trailing comment arrives + + events = list(stream) + assert [e.event_id for e in events] == ["e10", "e11"] + assert stream.has_more is True + assert stream.oldest_event_id == "e10" + + +def test_stream_ignores_non_has_more_comments(): + payload = ( + b": connected to s1\n\n" + b'data: {"event_id":"e1","type":"run.started","data":{}}\n\n' + ) + resources = _make_resources([_FakeResponse(200, sse_chunks=[payload])]) + + stream = resources.sessions.stream("s1") + assert len(list(stream)) == 1 + assert stream.has_more is None + + +def test_history_page_returns_events_cursor_and_has_more(): + resources = _make_resources([_FakeResponse(200, sse_chunks=[_HISTORY_PAGE_SSE])]) + + page = resources.sessions.history_page("s1", before="evt-12", limit=2) + + url = resources._proxy._original._pipeline.calls[0].request.url + assert "before=evt-12" in url + assert "limit=2" in url + assert "replay_only=true" in url + + events, has_more, next_before = page + assert [e.event_id for e in events] == ["e10", "e11"] + assert has_more is True + assert next_before == "e10" + + +def test_history_page_at_oldest_event_reports_no_more(): + payload = ( + b'data: {"event_id":"e1","type":"run.started","data":{}}\n\n' + b": has_more=false\n\n" + ) + resources = _make_resources([_FakeResponse(200, sse_chunks=[payload])]) + + page = resources.sessions.history_page("s1", before="e2") + assert page.has_more is False + assert page.next_before == "e1" + + +def test_history_page_empty_has_no_cursor(): + payload = b": has_more=false\n\n" + resources = _make_resources([_FakeResponse(200, sse_chunks=[payload])]) + + page = resources.sessions.history_page("s1", before="e1") + assert page.events == [] + assert page.has_more is False + assert page.next_before is None + + +def test_history_page_requires_before(): + resources = _make_resources([]) + with pytest.raises(ValueError, match="before"): + resources.sessions.history_page("s1", before="") + + +def test_agent_session_history_binds_session_id(): + resources = _make_resources([_FakeResponse(200, sse_chunks=[_HISTORY_PAGE_SSE])]) + + page = resources.attach("s1").history(before="evt-12") + + url = resources._proxy._original._pipeline.calls[0].request.url + assert "/v2/agents/sessions/s1/stream" in url + assert "before=evt-12" in url + assert page.next_before == "e10" + + def test_resolve_agents_base_url_adds_https_scheme(): assert ( resolve_agents_base_url("api.digitalocean.com")