From 984eb09ab5e0915da68fc3a2c42002832ece4716 Mon Sep 17 00:00:00 2001 From: rishu685 Date: Sat, 22 Aug 2026 11:00:39 +0530 Subject: [PATCH 01/10] feat(agent_manager): add pagination and lazy loading for chat history (#116) --- src/agent_manager/api/routes.py | 19 ++- src/agent_manager/api/schemas.py | 5 + .../api/static/widget/api/AgentChatClient.ts | 15 ++- .../api/static/widget/react/AgentChatApp.tsx | 50 ++++++- .../static/widget/react/useConversation.ts | 9 +- src/agent_manager/api/static/widget/types.ts | 5 + src/agent_manager/application/service.py | 7 +- src/agent_manager/domain/__init__.py | 2 + src/agent_manager/domain/models.py | 6 + src/agent_manager/domain/repository.py | 7 +- .../persistence/memory_repository.py | 43 +++++- .../persistence/sql_repository.py | 80 +++++++++++- tests/agent_manager/test_api.py | 19 +-- tests/agent_manager/test_pagination.py | 122 ++++++++++++++++++ .../agent_manager/test_repository_contract.py | 10 +- tests/agent_manager/test_service.py | 18 ++- 16 files changed, 364 insertions(+), 53 deletions(-) create mode 100644 tests/agent_manager/test_pagination.py diff --git a/src/agent_manager/api/routes.py b/src/agent_manager/api/routes.py index d87a5b24..91a50a51 100644 --- a/src/agent_manager/api/routes.py +++ b/src/agent_manager/api/routes.py @@ -9,7 +9,7 @@ from contextlib import contextmanager from typing import Annotated, Any -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Query from fastapi.responses import StreamingResponse from agent_engine.runtime.streaming import RunStreamEvent @@ -22,6 +22,7 @@ LinkAnonymousRequest, LinkAnonymousResponse, MessageOut, + PaginatedConversationsResponse, SendMessageRequest, SendMessageResponse, StreamEventOut, @@ -108,18 +109,24 @@ async def create_conversation( return CreateConversationResponse(conversation_id=session_id, session_id=session_id) -@router.get("/conversations", response_model=list[ConversationSummary]) -async def list_conversations(service: Service, caller: Caller) -> list[ConversationSummary]: - sessions = await service.list_conversations(caller) +@router.get("/conversations", response_model=PaginatedConversationsResponse) +async def list_conversations( + service: Service, + caller: Caller, + limit: int = Query(default=20, ge=1, le=100), + cursor: str | None = Query(default=None), +) -> PaginatedConversationsResponse: + paginated = await service.list_conversations(caller, limit=limit, cursor=cursor) - return [ + items = [ ConversationSummary( conversation_id=s.session_id, title=s.title, last_message_at=s.last_message_at, ) - for s in sessions + for s in paginated.sessions ] + return PaginatedConversationsResponse(items=items, next_cursor=paginated.next_cursor) @router.get("/conversations/{conversation_id}/messages", response_model=list[MessageOut]) diff --git a/src/agent_manager/api/schemas.py b/src/agent_manager/api/schemas.py index 081b388b..4467ef07 100644 --- a/src/agent_manager/api/schemas.py +++ b/src/agent_manager/api/schemas.py @@ -45,6 +45,11 @@ class ConversationSummary(BaseModel): last_message_at: datetime | None = None +class PaginatedConversationsResponse(BaseModel): + items: list[ConversationSummary] + next_cursor: str | None = None + + class MessageOut(BaseModel): role: Role content: str diff --git a/src/agent_manager/api/static/widget/api/AgentChatClient.ts b/src/agent_manager/api/static/widget/api/AgentChatClient.ts index 3f862691..ab7764bd 100644 --- a/src/agent_manager/api/static/widget/api/AgentChatClient.ts +++ b/src/agent_manager/api/static/widget/api/AgentChatClient.ts @@ -1,6 +1,7 @@ import type { TokenSource } from "../auth/tokenSource"; import type { ChatMessage, + PaginatedThreads, TokenBudget, SendMessageResponse, StreamEvent, @@ -66,16 +67,22 @@ export class AgentChatClient { return String(data.conversation_id); } - async listConversations(): Promise { - const response = await this.request("/conversations"); + async listConversations(limit = 20, cursor?: string | null): Promise { + const params = new URLSearchParams({ limit: String(limit) }); + if (cursor) params.set("cursor", cursor); + const response = await this.request(`/conversations?${params.toString()}`); const data = await response.json(); - if (!Array.isArray(data)) return []; - return data.map((thread) => ({ + const rawItems = Array.isArray(data.items) ? data.items : []; + const items: ThreadSummary[] = rawItems.map((thread: any) => ({ conversation_id: String(thread.conversation_id), title: thread.title ?? null, last_message_at: thread.last_message_at ?? null, })); + return { + items, + next_cursor: data.next_cursor ? String(data.next_cursor) : null, + }; } async getMessages(conversationId: string): Promise { diff --git a/src/agent_manager/api/static/widget/react/AgentChatApp.tsx b/src/agent_manager/api/static/widget/react/AgentChatApp.tsx index 21d45fa8..2805621f 100644 --- a/src/agent_manager/api/static/widget/react/AgentChatApp.tsx +++ b/src/agent_manager/api/static/widget/react/AgentChatApp.tsx @@ -80,6 +80,9 @@ export function AgentChatApp({ const usage = usageById[activeId] ?? null; const [threads, setThreads] = useState([]); + const [nextCursor, setNextCursor] = useState(null); + const [loadingMoreThreads, setLoadingMoreThreads] = useState(false); + const [hasMoreThreads, setHasMoreThreads] = useState(false); const [threadsOpen, setThreadsOpen] = useState(false); const launcherRef = useRef(null); const inputRef = useRef(null); @@ -95,8 +98,8 @@ export function AgentChatApp({ const refreshUsage = useCallback( async (cid: string) => { - const next = await conversation.loadUsage(cid); - setUsageById((prev) => ({ ...prev, [cid]: next })); + const u = await conversation.loadUsage(cid); + setUsageById((prev) => ({ ...prev, [cid]: u })); }, [conversation], ); @@ -146,10 +149,29 @@ export function AgentChatApp({ }, [inline]); const openThreads = useCallback(async () => { - setThreads(await conversation.listThreads()); setThreadsOpen(true); + setLoadingMoreThreads(true); + const res = await conversation.listThreads(20, null); + setThreads(res.items); + setNextCursor(res.next_cursor); + setHasMoreThreads(res.next_cursor !== null); + setLoadingMoreThreads(false); }, [conversation]); + const loadMoreThreads = useCallback(async () => { + if (loadingMoreThreads || !hasMoreThreads || !nextCursor) return; + setLoadingMoreThreads(true); + const res = await conversation.listThreads(20, nextCursor); + setThreads((prev) => { + const existingIds = new Set(prev.map((t) => t.conversation_id)); + const newItems = res.items.filter((t) => !existingIds.has(t.conversation_id)); + return [...prev, ...newItems]; + }); + setNextCursor(res.next_cursor); + setHasMoreThreads(res.next_cursor !== null); + setLoadingMoreThreads(false); + }, [conversation, hasMoreThreads, loadingMoreThreads, nextCursor]); + const openThread = useCallback( async (conversationId: string) => { conversation.switchTo(conversationId); @@ -161,7 +183,6 @@ export function AgentChatApp({ inputRef.current?.focus({ preventScroll: true }); }, [conversation, entriesById, loadThread, refreshUsage], - ); const startNewThread = useCallback(() => { @@ -293,6 +314,9 @@ export function AgentChatApp({ open={threadsOpen} threads={threads} activeId={getStoredConversationId(config.endpoint)} + loadingMore={loadingMoreThreads} + hasMore={hasMoreThreads} + onLoadMore={() => void loadMoreThreads()} onSelect={openThread} onNew={startNewThread} onClose={() => setThreadsOpen(false)} @@ -483,6 +507,9 @@ function ThreadDrawer({ open, threads, activeId, + loadingMore, + hasMore, + onLoadMore, onSelect, onNew, onClose, @@ -490,10 +517,20 @@ function ThreadDrawer({ open: boolean; threads: ThreadSummary[]; activeId: string | null; + loadingMore: boolean; + hasMore: boolean; + onLoadMore: () => void; onSelect: (conversationId: string) => void; onNew: () => void; onClose: () => void; }) { + const handleScroll = (e: React.UIEvent) => { + const { scrollTop, clientHeight, scrollHeight } = e.currentTarget; + if (scrollHeight - scrollTop - clientHeight < 40 && hasMore && !loadingMore) { + onLoadMore(); + } + }; + return (
@@ -506,7 +543,7 @@ function ThreadDrawer({ New chat -
+
{threads.map((thread) => (
); diff --git a/src/agent_manager/api/static/widget/react/useConversation.ts b/src/agent_manager/api/static/widget/react/useConversation.ts index 6c456553..b826ff34 100644 --- a/src/agent_manager/api/static/widget/react/useConversation.ts +++ b/src/agent_manager/api/static/widget/react/useConversation.ts @@ -8,6 +8,7 @@ import { } from "../storage/conversationStorage"; import type { ChatMessage, + PaginatedThreads, TokenBudget, SendMessageResponse, StreamEvent, @@ -27,7 +28,7 @@ export interface Conversation { stream(conversationId: string, text: string): AsyncGenerator; loadHistory(conversationId: string): Promise; loadUsage(conversationId: string): Promise; - listThreads(): Promise; + listThreads(limit?: number, cursor?: string | null): Promise; switchTo(conversationId: string): void; startNew(): void; } @@ -118,7 +119,11 @@ export function useConversation( [client], ); - const listThreads = useCallback(() => client.listConversations().catch(() => []), [client]); + const listThreads = useCallback( + (limit?: number, cursor?: string | null) => + client.listConversations(limit, cursor).catch(() => ({ items: [], next_cursor: null })), + [client], + ); const switchTo = useCallback( (conversationId: string) => setStoredConversationId(endpoint, conversationId), diff --git a/src/agent_manager/api/static/widget/types.ts b/src/agent_manager/api/static/widget/types.ts index fce26e9c..e3185096 100644 --- a/src/agent_manager/api/static/widget/types.ts +++ b/src/agent_manager/api/static/widget/types.ts @@ -30,6 +30,11 @@ export interface ThreadSummary { last_message_at: string | null; } +export interface PaginatedThreads { + items: ThreadSummary[]; + next_cursor: string | null; +} + export interface ChatMessage { role: ChatRole; content: string; diff --git a/src/agent_manager/application/service.py b/src/agent_manager/application/service.py index f38eba7b..b65df3c5 100644 --- a/src/agent_manager/application/service.py +++ b/src/agent_manager/application/service.py @@ -22,6 +22,7 @@ ConversationMessage, ConversationSession, Message, + PaginatedSessions, Principal, Repository, Role, @@ -124,9 +125,9 @@ async def usage(self, conversation_id: str, principal: Principal) -> TokenBudget return TokenBudgetUsage.from_totals(used, self._max_tokens) async def list_conversations( - self, principal: Principal, *, limit: int = 50 - ) -> list[ConversationSession]: - return await self._repository.list_sessions(principal.user_id, limit=limit) + self, principal: Principal, *, limit: int = 50, cursor: str | None = None + ) -> PaginatedSessions: + return await self._repository.list_sessions(principal.user_id, limit=limit, cursor=cursor) async def send(self, conversation_id: str, text: str, principal: Principal) -> RunResult: turn = await self.prepare_turn(conversation_id, text, principal) diff --git a/src/agent_manager/domain/__init__.py b/src/agent_manager/domain/__init__.py index d5286e4c..1fe8c61b 100644 --- a/src/agent_manager/domain/__init__.py +++ b/src/agent_manager/domain/__init__.py @@ -8,6 +8,7 @@ ConversationSession, ConversationSnapshot, Message, + PaginatedSessions, Role, TokenBudgetUsage, User, @@ -23,6 +24,7 @@ "ConversationSnapshot", "IdentityNamespace", "Message", + "PaginatedSessions", "Principal", "Repository", "Role", diff --git a/src/agent_manager/domain/models.py b/src/agent_manager/domain/models.py index 9e17a6be..bc0919d7 100644 --- a/src/agent_manager/domain/models.py +++ b/src/agent_manager/domain/models.py @@ -55,6 +55,12 @@ class ConversationSession: expires_at: datetime | None = None +@dataclass(frozen=True) +class PaginatedSessions: + sessions: list[ConversationSession] + next_cursor: str | None = None + + @dataclass(frozen=True) class ConversationMessage: message_id: str diff --git a/src/agent_manager/domain/repository.py b/src/agent_manager/domain/repository.py index b942dadc..b235f29b 100644 --- a/src/agent_manager/domain/repository.py +++ b/src/agent_manager/domain/repository.py @@ -12,6 +12,7 @@ ConversationSession, ConversationSnapshot, Message, + PaginatedSessions, Role, User, ) @@ -66,8 +67,10 @@ async def create_session( async def get_session(self, session_id: str) -> ConversationSession | None: ... @abstractmethod - async def list_sessions(self, user_id: str, *, limit: int = 50) -> list[ConversationSession]: - """A user's sessions, most-recently-active first.""" + async def list_sessions( + self, user_id: str, *, limit: int = 50, cursor: str | None = None + ) -> PaginatedSessions: + """A user's sessions, most-recently-active first, with cursor pagination.""" @abstractmethod async def rename_session(self, session_id: str, title: str) -> None: ... diff --git a/src/agent_manager/infrastructure/persistence/memory_repository.py b/src/agent_manager/infrastructure/persistence/memory_repository.py index 8ffc2a4f..ee54eb03 100644 --- a/src/agent_manager/infrastructure/persistence/memory_repository.py +++ b/src/agent_manager/infrastructure/persistence/memory_repository.py @@ -14,14 +14,30 @@ ConversationSession, ConversationSnapshot, Message, + PaginatedSessions, Repository, Role, User, ) +from agent_manager.infrastructure.persistence.sql_repository import ( + decode_cursor, + encode_cursor, +) _EPOCH = datetime(1970, 1, 1, tzinfo=UTC) +def _is_after_cursor(s: ConversationSession, cursor_t: datetime | None, cursor_id: str) -> bool: + sid = s.session_id or "" + if cursor_t is not None: + if s.last_message_at is None: + return True + if s.last_message_at < cursor_t: + return True + return s.last_message_at == cursor_t and sid < cursor_id + return s.last_message_at is None and sid < cursor_id + + class MemoryRepository(Repository): def __init__(self) -> None: self._users: dict[str, User] = {} @@ -108,10 +124,31 @@ async def create_session( async def get_session(self, session_id: str) -> ConversationSession | None: return self._sessions.get(session_id) - async def list_sessions(self, user_id: str, *, limit: int = 50) -> list[ConversationSession]: + async def list_sessions( + self, user_id: str, *, limit: int = 50, cursor: str | None = None + ) -> PaginatedSessions: sessions = [s for s in self._sessions.values() if s.user_id == user_id] - sessions.sort(key=lambda s: s.last_message_at or s.created_at or _EPOCH, reverse=True) - return sessions[:limit] + sessions.sort( + key=lambda s: ( + s.last_message_at is not None, + s.last_message_at or _EPOCH, + s.session_id, + ), + reverse=True, + ) + + if cursor is not None: + cursor_t, cursor_id = decode_cursor(cursor) + sessions = [s for s in sessions if _is_after_cursor(s, cursor_t, cursor_id)] + + has_more = len(sessions) > limit + result_sessions = sessions[:limit] if has_more else sessions + next_cursor = ( + encode_cursor(result_sessions[-1].last_message_at, result_sessions[-1].session_id) + if has_more and result_sessions + else None + ) + return PaginatedSessions(sessions=result_sessions, next_cursor=next_cursor) async def rename_session(self, session_id: str, title: str) -> None: session = self._sessions.get(session_id) diff --git a/src/agent_manager/infrastructure/persistence/sql_repository.py b/src/agent_manager/infrastructure/persistence/sql_repository.py index 167313e6..de0c5ac7 100644 --- a/src/agent_manager/infrastructure/persistence/sql_repository.py +++ b/src/agent_manager/infrastructure/persistence/sql_repository.py @@ -6,11 +6,13 @@ from __future__ import annotations +import base64 +import json import uuid from datetime import UTC, datetime, timedelta from typing import Any -from sqlalchemy import delete, update +from sqlalchemy import and_, delete, or_, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import async_sessionmaker from sqlmodel import col, select @@ -22,6 +24,7 @@ ConversationSession, ConversationSnapshot, Message, + PaginatedSessions, Repository, Role, User, @@ -34,6 +37,27 @@ ) +def encode_cursor(last_message_at: datetime | None, session_id: str) -> str: + payload = { + "t": last_message_at.isoformat() if last_message_at is not None else None, + "id": session_id, + } + raw_bytes = json.dumps(payload).encode("utf-8") + return base64.urlsafe_b64encode(raw_bytes).decode("ascii") + + +def decode_cursor(cursor_str: str) -> tuple[datetime | None, str]: + try: + raw_bytes = base64.urlsafe_b64decode(cursor_str.encode("ascii")) + data = json.loads(raw_bytes.decode("utf-8")) + t_str = data.get("t") + last_message_at = datetime.fromisoformat(t_str) if t_str is not None else None + session_id = str(data["id"]) + return last_message_at, session_id + except Exception as exc: + raise ValueError(f"Invalid pagination cursor: {cursor_str}") from exc + + class SqlRepository(Repository): def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None: self._sessions = sessions @@ -148,16 +172,58 @@ async def get_session(self, session_id: str) -> ConversationSession | None: row = await session.get(ConversationSessionRow, session_id) return _session(row) if row else None - async def list_sessions(self, user_id: str, *, limit: int = 50) -> list[ConversationSession]: + async def list_sessions( + self, user_id: str, *, limit: int = 50, cursor: str | None = None + ) -> PaginatedSessions: + conditions: list[Any] = [ConversationSessionRow.user_id == user_id] + + if cursor is not None: + cursor_t, cursor_id = decode_cursor(cursor) + if cursor_t is not None: + conditions.append( + or_( + col(ConversationSessionRow.last_message_at) < cursor_t, + and_( + col(ConversationSessionRow.last_message_at) == cursor_t, + col(ConversationSessionRow.session_id) < cursor_id, + ), + col(ConversationSessionRow.last_message_at).is_(None), + ) + ) + else: + conditions.append( + and_( + col(ConversationSessionRow.last_message_at).is_(None), + col(ConversationSessionRow.session_id) < cursor_id, + ) + ) + stmt = ( select(ConversationSessionRow) - .where(ConversationSessionRow.user_id == user_id) - .order_by(col(ConversationSessionRow.last_message_at).desc()) - .limit(limit) + .where(*conditions) + .order_by( + col(ConversationSessionRow.last_message_at).desc().nullslast(), + col(ConversationSessionRow.session_id).desc(), + ) + .limit(limit + 1) ) async with self._sessions() as session: - rows = (await session.exec(stmt)).all() - return [_session(row) for row in rows] + rows = list((await session.exec(stmt)).all()) + + has_more = len(rows) > limit + if has_more: + rows = rows[:limit] + + next_cursor = ( + encode_cursor(rows[-1].last_message_at, rows[-1].session_id) + if has_more and rows + else None + ) + + return PaginatedSessions( + sessions=[_session(row) for row in rows], + next_cursor=next_cursor, + ) async def rename_session(self, session_id: str, title: str) -> None: async with self._sessions() as session: diff --git a/tests/agent_manager/test_api.py b/tests/agent_manager/test_api.py index a7657646..150f2dd6 100644 --- a/tests/agent_manager/test_api.py +++ b/tests/agent_manager/test_api.py @@ -54,12 +54,12 @@ def test_list_conversations_returns_titled_threads_scoped_to_user(client: TestCl b = client.post("/conversations", headers=u1).json()["conversation_id"] client.post(f"/conversations/{b}/messages", json={"message": "second thread"}, headers=u1) - threads = client.get("/conversations", headers=u1).json() + threads = client.get("/conversations", headers=u1).json()["items"] assert {t["conversation_id"]: t["title"] for t in threads} == { a: "first thread", b: "second thread", } - assert client.get("/conversations", headers=bearer("u2")).json() == [] + assert client.get("/conversations", headers=bearer("u2")).json()["items"] == [] def test_another_caller_cannot_touch_a_conversation_it_does_not_own(client: TestClient) -> None: @@ -91,7 +91,10 @@ def test_create_cannot_claim_a_conversation_id_owned_by_another_caller( assert taken.status_code == 409 assert client.get("/conversations/sess-1/messages", headers=bob).status_code == 403 - assert client.get("/conversations", headers=alice).json()[0]["conversation_id"] == "sess-1" + assert ( + client.get("/conversations", headers=alice).json()["items"][0]["conversation_id"] + == "sess-1" + ) @pytest.fixture @@ -144,7 +147,7 @@ def test_the_host_session_cookie_authenticates_a_same_origin_deployment() -> Non cid = created.json()["conversation_id"] dana.post(f"/conversations/{cid}/messages", json={"message": "hi"}) - assert [t["conversation_id"] for t in dana.get("/conversations").json()] == [cid] + assert [t["conversation_id"] for t in dana.get("/conversations").json()["items"]] == [cid] assert TestClient(app).get(f"/conversations/{cid}/messages").status_code == 401 @@ -161,7 +164,7 @@ def test_a_visitor_pass_is_an_identity_of_its_own(unauthenticated: TestClient) - assert client.get(f"/conversations/{cid}/messages", headers=visitor).status_code == 200 assert client.get(f"/conversations/{cid}/messages", headers=other_visitor).status_code == 403 - assert client.get("/conversations", headers=other_visitor).json() == [] + assert client.get("/conversations", headers=other_visitor).json()["items"] == [] def test_signing_in_adopts_the_conversations_a_visitor_already_started() -> None: @@ -177,9 +180,9 @@ def test_signing_in_adopts_the_conversations_a_visitor_already_started() -> None linked = client.post("/auth/link", json={"anonymous_token": pass_token}, headers=alice) assert linked.json() == {"conversations_moved": 1} - assert [t["conversation_id"] for t in client.get("/conversations", headers=alice).json()] == [ - cid - ] + assert [ + t["conversation_id"] for t in client.get("/conversations", headers=alice).json()["items"] + ] == [cid] assert client.get(f"/conversations/{cid}/messages", headers=alice).status_code == 200 assert client.get(f"/conversations/{cid}/messages", headers=visitor).status_code == 403 diff --git a/tests/agent_manager/test_pagination.py b/tests/agent_manager/test_pagination.py new file mode 100644 index 00000000..cd8d7a6f --- /dev/null +++ b/tests/agent_manager/test_pagination.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from pathlib import Path +from urllib.parse import quote + +import pytest +from fastapi.testclient import TestClient +from sqlmodel import SQLModel + +import agent_manager.infrastructure.persistence.tables # noqa: F401 +from agent_manager.application import ConversationService +from agent_manager.infrastructure.persistence.database import create_db_engine, session_factory +from agent_manager.infrastructure.persistence.memory_repository import MemoryRepository +from agent_manager.infrastructure.persistence.sql_repository import ( + SqlRepository, + decode_cursor, + encode_cursor, +) +from tests.agent_manager.conftest import RecordingEngine, bearer, build_test_app + + +@pytest.fixture +def client() -> TestClient: + app = build_test_app(ConversationService(RecordingEngine(), MemoryRepository())) + return TestClient(app, headers=bearer("default-user")) + + +def test_cursor_encode_decode_round_trip() -> None: + now = datetime.now(UTC) + cursor = encode_cursor(now, "sess-123") + decoded_t, decoded_id = decode_cursor(cursor) + + assert decoded_t == now + assert decoded_id == "sess-123" + + +def test_cursor_encode_decode_with_null_timestamp() -> None: + cursor = encode_cursor(None, "sess-null") + decoded_t, decoded_id = decode_cursor(cursor) + + assert decoded_t is None + assert decoded_id == "sess-null" + + +def test_invalid_cursor_raises_value_error() -> None: + with pytest.raises(ValueError, match="Invalid pagination cursor"): + decode_cursor("not-a-valid-cursor!") + + +@pytest.mark.asyncio +async def test_sql_repository_pagination_and_ordering(tmp_path: Path) -> None: + db_url = f"sqlite+aiosqlite:///{tmp_path / 'test_pag.db'}" + engine = create_db_engine(db_url) + async with engine.begin() as conn: + await conn.run_sync(SQLModel.metadata.create_all) + sessions = session_factory(engine) + + repo = SqlRepository(sessions) + user_id = "user-pag-1" + await repo.upsert_user(user_id) + + base_time = datetime(2026, 8, 20, 12, 0, 0, tzinfo=UTC) + + await repo.create_session("sess-1", user_id=user_id, title="Session 1") + await repo.create_session("sess-2", user_id=user_id, title="Session 2") + await repo.create_session("sess-3", user_id=user_id, title="Session 3") + await repo.create_session("sess-4", user_id=user_id, title="Session 4") + await repo.create_session("sess-5", user_id=user_id, title="Session 5") + + async with sessions() as session: + from agent_manager.infrastructure.persistence.tables import ConversationSessionRow + + r1 = await session.get(ConversationSessionRow, "sess-1") + assert r1 is not None + r1.last_message_at = base_time + timedelta(hours=2) + + r2 = await session.get(ConversationSessionRow, "sess-2") + assert r2 is not None + r2.last_message_at = base_time + timedelta(hours=1) + + r3 = await session.get(ConversationSessionRow, "sess-3") + assert r3 is not None + r3.last_message_at = base_time + timedelta(hours=1) + + await session.commit() + + # Page 1: limit 2 + p1 = await repo.list_sessions(user_id, limit=2) + assert [s.session_id for s in p1.sessions] == ["sess-1", "sess-3"] + assert p1.next_cursor is not None + + # Page 2: limit 2 + p2 = await repo.list_sessions(user_id, limit=2, cursor=p1.next_cursor) + assert [s.session_id for s in p2.sessions] == ["sess-2", "sess-5"] + assert p2.next_cursor is not None + + # Page 3: limit 2 + p3 = await repo.list_sessions(user_id, limit=2, cursor=p2.next_cursor) + assert [s.session_id for s in p3.sessions] == ["sess-4"] + assert p3.next_cursor is None + + await engine.dispose() + + +def test_api_conversations_pagination_endpoint(client: TestClient) -> None: + u1 = bearer("user-api-pag") + c1 = client.post("/conversations", headers=u1).json()["conversation_id"] + c2 = client.post("/conversations", headers=u1).json()["conversation_id"] + c3 = client.post("/conversations", headers=u1).json()["conversation_id"] + + res1 = client.get("/conversations?limit=2", headers=u1).json() + assert len(res1["items"]) == 2 + assert res1["next_cursor"] is not None + + cursor_q = quote(res1["next_cursor"]) + res2 = client.get(f"/conversations?limit=2&cursor={cursor_q}", headers=u1).json() + assert len(res2["items"]) == 1 + assert res2["next_cursor"] is None + + fetched_ids = [item["conversation_id"] for item in res1["items"] + res2["items"]] + assert set(fetched_ids) == {c1, c2, c3} diff --git a/tests/agent_manager/test_repository_contract.py b/tests/agent_manager/test_repository_contract.py index d8ba24e1..909cad4c 100644 --- a/tests/agent_manager/test_repository_contract.py +++ b/tests/agent_manager/test_repository_contract.py @@ -56,7 +56,7 @@ async def test_create_session_never_reassigns_an_existing_owner(repo: Repository stored = await repo.get_session("shared-id") assert stored is not None assert stored.user_id == "alice" - assert await repo.list_sessions("bob") == [] + assert (await repo.list_sessions("bob")).sessions == [] async def test_create_session_writes_nothing_when_the_id_is_taken(repo: Repository) -> None: @@ -107,7 +107,7 @@ async def test_appending_a_message_never_claims_the_conversation(repo: Repositor unowned = await repo.get_session("unowned") assert owned is not None and owned.user_id == "alice" assert unowned is not None and unowned.user_id is None - assert await repo.list_sessions("bob") == [] + assert (await repo.list_sessions("bob")).sessions == [] async def test_messages_in_insertion_order(repo: Repository) -> None: @@ -334,8 +334,8 @@ async def test_linking_a_visitor_moves_their_sessions_once(repo: Repository) -> assert await repo.link_anonymous_user("anon:v1", "ext:alice") == 1 moved = await repo.get_session("pre-login") assert moved is not None and moved.user_id == "ext:alice" - assert [s.session_id for s in await repo.list_sessions("ext:alice")] == ["pre-login"] - assert await repo.list_sessions("anon:v1") == [] + assert [s.session_id for s in (await repo.list_sessions("ext:alice")).sessions] == ["pre-login"] + assert (await repo.list_sessions("anon:v1")).sessions == [] visitor = await repo.get_user("anon:v1") assert visitor is not None and visitor.linked_to_user_id == "ext:alice" @@ -343,7 +343,7 @@ async def test_linking_a_visitor_moves_their_sessions_once(repo: Repository) -> # Spent: a replayed pass moves nothing, whoever presents it. await repo.upsert_user("ext:bob") assert await repo.link_anonymous_user("anon:v1", "ext:bob") == 0 - assert await repo.list_sessions("ext:bob") == [] + assert (await repo.list_sessions("ext:bob")).sessions == [] async def test_linking_an_unknown_visitor_is_a_no_op(repo: Repository) -> None: diff --git a/tests/agent_manager/test_service.py b/tests/agent_manager/test_service.py index 676f0ab5..71f4c99c 100644 --- a/tests/agent_manager/test_service.py +++ b/tests/agent_manager/test_service.py @@ -169,8 +169,8 @@ async def test_reads_of_an_owned_conversation_refuse_other_callers() -> None: with pytest.raises(ConversationAccessDenied): await service.usage(cid, caller) - assert await service.list_conversations(BOB) == [] - assert await service.list_conversations(VISITOR) == [] + assert (await service.list_conversations(BOB)).sessions == [] + assert (await service.list_conversations(VISITOR)).sessions == [] async def test_create_refuses_a_session_id_owned_by_someone_else() -> None: @@ -257,8 +257,10 @@ async def test_signing_in_moves_a_visitors_conversations_onto_their_account() -> moved = await service.link_anonymous(VISITOR, ALICE) assert moved == 1 - assert [s.session_id for s in await service.list_conversations(ALICE)] == ["pre-login"] - assert await service.list_conversations(VISITOR) == [] + assert [s.session_id for s in (await service.list_conversations(ALICE)).sessions] == [ + "pre-login" + ] + assert (await service.list_conversations(VISITOR)).sessions == [] assert [m.content for m in await service.history(before_login, ALICE)] == [ "how much does it cost?", "answer:how much does it cost?", @@ -273,8 +275,10 @@ async def test_a_visitor_pass_can_only_be_adopted_once() -> None: assert await service.link_anonymous(VISITOR, ALICE) == 1 assert await service.link_anonymous(VISITOR, BOB) == 0 - assert [s.session_id for s in await service.list_conversations(ALICE)] == ["pre-login"] - assert await service.list_conversations(BOB) == [] + assert [s.session_id for s in (await service.list_conversations(ALICE)).sessions] == [ + "pre-login" + ] + assert (await service.list_conversations(BOB)).sessions == [] async def test_a_visitor_cannot_adopt_another_visitor() -> None: @@ -292,7 +296,7 @@ async def test_adopting_merges_into_conversations_the_account_already_had() -> N await service.link_anonymous(VISITOR, ALICE) - assert {s.session_id for s in await service.list_conversations(ALICE)} == { + assert {s.session_id for s in (await service.list_conversations(ALICE)).sessions} == { "signed-in", "pre-login", } From 032ee04174d2bbb0f24d59f74f8b062172687852 Mon Sep 17 00:00:00 2001 From: rishu685 Date: Sat, 22 Aug 2026 11:06:23 +0530 Subject: [PATCH 02/10] test(agent_manager): update test_api for paginated conversations response --- tests/agent_manager/test_api.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/tests/agent_manager/test_api.py b/tests/agent_manager/test_api.py index 4aac80b3..89ff9bee 100644 --- a/tests/agent_manager/test_api.py +++ b/tests/agent_manager/test_api.py @@ -324,9 +324,9 @@ def test_a_visitor_pass_does_not_shadow_the_host_session_cookie() -> None: # The conversation belongs to Dana, not to the visitor she used to be. as_dana = TestClient(app, cookies=session_cookie(id="u_8412")) - assert [t["conversation_id"] for t in as_dana.get("/conversations").json()] == [cid] + assert [t["conversation_id"] for t in as_dana.get("/conversations").json()["items"]] == [cid] still_a_visitor = {"Authorization": f"Bearer {visitor_pass}"} - assert TestClient(app).get("/conversations", headers=still_a_visitor).json() == [] + assert TestClient(app).get("/conversations", headers=still_a_visitor).json()["items"] == [] def test_a_host_bearer_token_outranks_the_session_cookie() -> None: @@ -348,9 +348,13 @@ def test_a_host_bearer_token_outranks_the_session_cookie() -> None: cid = caller.post("/conversations").json()["conversation_id"] assert [ - t["conversation_id"] for t in TestClient(app).get("/conversations", headers=as_noam).json() + t["conversation_id"] + for t in TestClient(app).get("/conversations", headers=as_noam).json()["items"] ] == [cid] - assert TestClient(app, cookies=session_cookie(id="u_asaf")).get("/conversations").json() == [] + assert ( + TestClient(app, cookies=session_cookie(id="u_asaf")).get("/conversations").json()["items"] + == [] + ) def test_a_visitor_pass_is_an_identity_of_its_own(unauthenticated: TestClient) -> None: From 31475043de7fa7987f7881fcd9ab5dd8a45934ab Mon Sep 17 00:00:00 2001 From: rishu685 Date: Sat, 22 Aug 2026 11:10:23 +0530 Subject: [PATCH 03/10] build(widget): rebuild widget bundle for chat history pagination --- src/agent_manager/api/static/widget.js | 63 ++++++++++++++++++++++---- 1 file changed, 53 insertions(+), 10 deletions(-) diff --git a/src/agent_manager/api/static/widget.js b/src/agent_manager/api/static/widget.js index c3d3cf23..b43547c6 100644 --- a/src/agent_manager/api/static/widget.js +++ b/src/agent_manager/api/static/widget.js @@ -52163,15 +52163,21 @@ var AgentChatClient = class { const data = await response.json(); return String(data.conversation_id); } - async listConversations() { - const response = await this.request("/conversations"); + async listConversations(limit = 20, cursor) { + const params = new URLSearchParams({ limit: String(limit) }); + if (cursor) params.set("cursor", cursor); + const response = await this.request(`/conversations?${params.toString()}`); const data = await response.json(); - if (!Array.isArray(data)) return []; - return data.map((thread) => ({ + const rawItems = Array.isArray(data.items) ? data.items : []; + const items = rawItems.map((thread) => ({ conversation_id: String(thread.conversation_id), title: thread.title ?? null, last_message_at: thread.last_message_at ?? null })); + return { + items, + next_cursor: data.next_cursor ? String(data.next_cursor) : null + }; } async getMessages(conversationId) { const response = await this.request(`/conversations/${conversationId}/messages`); @@ -53392,7 +53398,10 @@ function useConversation(client, endpoint, onReplaced) { }, [client] ); - const listThreads = (0, import_react9.useCallback)(() => client.listConversations().catch(() => []), [client]); + const listThreads = (0, import_react9.useCallback)( + (limit, cursor) => client.listConversations(limit, cursor).catch(() => ({ items: [], next_cursor: null })), + [client] + ); const switchTo = (0, import_react9.useCallback)( (conversationId) => setStoredConversationId(endpoint, conversationId), [endpoint] @@ -53480,6 +53489,9 @@ function AgentChatApp({ const canSubmit = !isExecutionActive && !budgetExceeded && !approvalBlocksComposer; const canStop = isExecutionActive; const [threads, setThreads] = (0, import_react10.useState)([]); + const [nextCursor, setNextCursor] = (0, import_react10.useState)(null); + const [loadingMoreThreads, setLoadingMoreThreads] = (0, import_react10.useState)(false); + const [hasMoreThreads, setHasMoreThreads] = (0, import_react10.useState)(false); const [threadsOpen, setThreadsOpen] = (0, import_react10.useState)(false); const launcherRef = (0, import_react10.useRef)(null); const inputRef = (0, import_react10.useRef)(null); @@ -53519,8 +53531,8 @@ function AgentChatApp({ }, []); const refreshUsage = (0, import_react10.useCallback)( async (cid) => { - const next2 = await conversation.loadUsage(cid); - setUsageById((prev) => ({ ...prev, [cid]: next2 })); + const u4 = await conversation.loadUsage(cid); + setUsageById((prev) => ({ ...prev, [cid]: u4 })); }, [conversation] ); @@ -53561,9 +53573,27 @@ function AgentChatApp({ launcherRef.current?.focus({ preventScroll: true }); }, [inline]); const openThreads = (0, import_react10.useCallback)(async () => { - setThreads(await conversation.listThreads()); setThreadsOpen(true); + setLoadingMoreThreads(true); + const res = await conversation.listThreads(20, null); + setThreads(res.items); + setNextCursor(res.next_cursor); + setHasMoreThreads(res.next_cursor !== null); + setLoadingMoreThreads(false); }, [conversation]); + const loadMoreThreads = (0, import_react10.useCallback)(async () => { + if (loadingMoreThreads || !hasMoreThreads || !nextCursor) return; + setLoadingMoreThreads(true); + const res = await conversation.listThreads(20, nextCursor); + setThreads((prev) => { + const existingIds = new Set(prev.map((t) => t.conversation_id)); + const newItems = res.items.filter((t) => !existingIds.has(t.conversation_id)); + return [...prev, ...newItems]; + }); + setNextCursor(res.next_cursor); + setHasMoreThreads(res.next_cursor !== null); + setLoadingMoreThreads(false); + }, [conversation, hasMoreThreads, loadingMoreThreads, nextCursor]); const openThread = (0, import_react10.useCallback)( async (conversationId) => { conversation.switchTo(conversationId); @@ -53911,6 +53941,9 @@ function AgentChatApp({ open: threadsOpen, threads, activeId: getStoredConversationId(config.endpoint), + loadingMore: loadingMoreThreads, + hasMore: hasMoreThreads, + onLoadMore: () => void loadMoreThreads(), onSelect: openThread, onNew: startNewThread, onClose: () => setThreadsOpen(false) @@ -54160,10 +54193,19 @@ function ThreadDrawer({ open, threads, activeId, + loadingMore, + hasMore, + onLoadMore, onSelect, onNew, onClose }) { + const handleScroll = (e) => { + const { scrollTop, clientHeight, scrollHeight } = e.currentTarget; + if (scrollHeight - scrollTop - clientHeight < 40 && hasMore && !loadingMore) { + onLoadMore(); + } + }; return /* @__PURE__ */ (0, import_jsx_runtime4.jsxs)("div", { className: `thread-drawer${open ? " open" : ""}`, inert: !open, children: [ /* @__PURE__ */ (0, import_jsx_runtime4.jsxs)("div", { className: "thread-drawer-head", children: [ /* @__PURE__ */ (0, import_jsx_runtime4.jsx)("span", { children: "Chats" }), @@ -54173,7 +54215,7 @@ function ThreadDrawer({ /* @__PURE__ */ (0, import_jsx_runtime4.jsx)(SquarePen, { "aria-hidden": true }), "New chat" ] }), - /* @__PURE__ */ (0, import_jsx_runtime4.jsxs)("div", { className: "thread-list", children: [ + /* @__PURE__ */ (0, import_jsx_runtime4.jsxs)("div", { className: "thread-list", onScroll: handleScroll, children: [ threads.map((thread) => /* @__PURE__ */ (0, import_jsx_runtime4.jsx)( "button", { @@ -54185,7 +54227,8 @@ function ThreadDrawer({ }, thread.conversation_id )), - threads.length === 0 ? /* @__PURE__ */ (0, import_jsx_runtime4.jsx)("p", { className: "thread-empty", children: "No conversations yet" }) : null + threads.length === 0 && !loadingMore ? /* @__PURE__ */ (0, import_jsx_runtime4.jsx)("p", { className: "thread-empty", children: "No conversations yet" }) : null, + loadingMore ? /* @__PURE__ */ (0, import_jsx_runtime4.jsx)("p", { className: "thread-empty", children: "Loading..." }) : null ] }) ] }); } From d25be318b948758399b40e54f02e6edd4602dab6 Mon Sep 17 00:00:00 2001 From: rishu685 Date: Sat, 22 Aug 2026 11:45:37 +0530 Subject: [PATCH 04/10] test(e2e): update Playwright route mocks for paginated conversations response envelope --- tests/e2e/playground.spec.ts | 8 ++++++-- tests/e2e/widget.spec.ts | 6 +++--- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/tests/e2e/playground.spec.ts b/tests/e2e/playground.spec.ts index 810da2f3..44a401a3 100644 --- a/tests/e2e/playground.spec.ts +++ b/tests/e2e/playground.spec.ts @@ -15,14 +15,18 @@ async function mockConversationApi(page: Page) { contentType: "application/json", body: method === "GET" - ? JSON.stringify([]) + ? JSON.stringify({ items: [], next_cursor: null }) : JSON.stringify({ conversation_id: "conv-playground", session_id: "conv-playground" }), }); }); await page.route(/\/conversations\?/, async (route) => { calls.push(`GET ${new URL(route.request().url()).pathname}`); - await route.fulfill({ status: 200, contentType: "application/json", body: "[]" }); + await route.fulfill({ + status: 200, + contentType: "application/json", + body: JSON.stringify({ items: [], next_cursor: null }), + }); }); await page.route("**/conversations/*/messages", async (route: Route) => { diff --git a/tests/e2e/widget.spec.ts b/tests/e2e/widget.spec.ts index 316c9719..74821410 100644 --- a/tests/e2e/widget.spec.ts +++ b/tests/e2e/widget.spec.ts @@ -31,7 +31,7 @@ async function mockConversationApi( await route.fulfill({ status: 200, contentType: "application/json", - body: JSON.stringify(options.threads ?? []), + body: JSON.stringify({ items: options.threads ?? [], next_cursor: null }), }); }); @@ -44,7 +44,7 @@ async function mockConversationApi( contentType: "application/json", body: method === "GET" - ? JSON.stringify(options.threads ?? []) + ? JSON.stringify({ items: options.threads ?? [], next_cursor: null }) : JSON.stringify({ conversation_id: "conv-smoke", session_id: "conv-smoke" }), }); }); @@ -147,7 +147,7 @@ async function mockApprovalApi( await page.route("**/conversations", async (route) => { const body = route.request().method() === "GET" - ? [] + ? { items: [], next_cursor: null } : { conversation_id: "conv-approval", session_id: "conv-approval" }; await route.fulfill({ status: 200, contentType: "application/json", body: JSON.stringify(body) }); }); From 3b2c7b0355e43d92f4b0350d1393c2ccf53abd0a Mon Sep 17 00:00:00 2001 From: rishu685 Date: Sun, 23 Aug 2026 22:57:22 +0530 Subject: [PATCH 05/10] fix(agent_manager): address review feedback on pagination for PR #124 (#116) - Align database expression index on COALESCE(last_message_at, created_at) - Fix cursor codec and keyset predicate to encode coalesced active timestamps - Enforce PageRequest limits at domain/port level - Handle empty string cursors as absent, map InvalidCursorError to 400 Bad Request - Add request generation counter in React widget to eliminate stale async race conditions - Remove temporary Page.sessions property shim - Expand page-walk test suite for exact sequence ordering over mixed NULL/non-NULL data --- docs/api.mdx | 30 +++++ src/agent_manager/api/errors.py | 6 + src/agent_manager/api/routes/conversations.py | 9 +- src/agent_manager/api/static/widget.js | 65 +++++++---- .../api/static/widget/api/AgentChatClient.ts | 7 +- .../api/static/widget/react/AgentChatApp.tsx | 65 +++++++---- .../static/widget/react/useConversation.ts | 3 +- src/agent_manager/application/__init__.py | 2 + .../application/conversation_service.py | 9 +- src/agent_manager/application/errors.py | 13 +++ src/agent_manager/domain/__init__.py | 10 ++ src/agent_manager/domain/models.py | 27 ++++- src/agent_manager/domain/repository.py | 7 +- .../persistence/memory_repository.py | 106 +++++++++++------- .../0005_add_session_pagination_index.py | 31 +++++ .../infrastructure/persistence/pagination.py | 59 ++++++++++ .../persistence/sql_repository.py | 83 +++++--------- .../infrastructure/persistence/tables.py | 13 ++- tests/agent_manager/test_pagination.py | 63 ++++++++--- .../agent_manager/test_repository_contract.py | 49 +++++++- tests/agent_manager/test_service.py | 18 ++- tests/e2e/widget.spec.ts | 52 +++++++++ 22 files changed, 538 insertions(+), 189 deletions(-) create mode 100644 src/agent_manager/infrastructure/persistence/migrations/versions/0005_add_session_pagination_index.py create mode 100644 src/agent_manager/infrastructure/persistence/pagination.py diff --git a/docs/api.mdx b/docs/api.mdx index eb5bd5b3..0d177d2a 100644 --- a/docs/api.mdx +++ b/docs/api.mdx @@ -166,6 +166,36 @@ Response: Errors: `409` the supplied `session_id` belongs to another user. +### `GET /conversations` + +List the calling user's conversations, ordered by most recently active first, using keyset pagination. + +**Query Parameters:** +- `limit` *(integer, optional, default: 20, min: 1, max: 100)* — Maximum number of conversations to return per page. +- `cursor` *(string, optional)* — Opaque pagination cursor token obtained from `next_cursor` of the previous page. + +```bash +curl "http://localhost:8100/conversations?limit=20" \ + -H "Authorization: Bearer $TOKEN" +``` + +Response: + +```json +{ + "items": [ + { + "conversation_id": "0d5a…", + "title": "Order status inquiry", + "last_message_at": "2026-08-20T14:00:00+00:00" + } + ], + "next_cursor": "eyJ0IjoiMjAyNi0wOC0yMFQxNDowMDowMCswMDowMCIsImlkIjoic2Vzcy0xMiJ9" +} +``` + +When `next_cursor` is `null`, no further pages remain. Cursors are opaque server tokens and must not be constructed manually by clients. + ### `POST /conversations/{id}/messages` Send a message. Prior history is assembled automatically. diff --git a/src/agent_manager/api/errors.py b/src/agent_manager/api/errors.py index 5513e36f..b0984d29 100644 --- a/src/agent_manager/api/errors.py +++ b/src/agent_manager/api/errors.py @@ -22,12 +22,17 @@ ConversationMessageNotFound, ConversationNotFound, ConversationTokenBudgetExceeded, + InvalidCursorError, ) BUDGET_EXCEEDED_DETAIL = { "error_type": "context_limit_exceeded", "message": "This conversation has reached its context limit. Start a new chat to continue.", } +INVALID_CURSOR_DETAIL = { + "error_type": "invalid_cursor", + "message": "invalid pagination cursor", +} INTERNAL_ERROR_MESSAGE = "Internal server error" _HTTP_ERRORS: dict[type[Exception], tuple[int, Any]] = { @@ -38,6 +43,7 @@ ConversationMessageNotFound: (404, "message not found on active conversation branch"), ConversationBranchConflict: (409, "conversation branch changed; reload and try again"), ConversationLinkRefused: (403, "a visitor cannot adopt another visitor"), + InvalidCursorError: (400, INVALID_CURSOR_DETAIL), } diff --git a/src/agent_manager/api/routes/conversations.py b/src/agent_manager/api/routes/conversations.py index b7b7dd5c..441ebaf7 100644 --- a/src/agent_manager/api/routes/conversations.py +++ b/src/agent_manager/api/routes/conversations.py @@ -29,6 +29,7 @@ StreamEventOut, TokenBudgetResponse, ) +from agent_manager.domain import DEFAULT_PAGE_LIMIT, MAX_PAGE_LIMIT, PageRequest router = APIRouter() logger = logging.getLogger(__name__) @@ -50,17 +51,19 @@ async def create_conversation( async def list_conversations( service: Service, caller: Caller, - limit: int = Query(default=20, ge=1, le=100), + limit: int = Query(default=DEFAULT_PAGE_LIMIT, ge=1, le=MAX_PAGE_LIMIT), cursor: str | None = Query(default=None), ) -> PaginatedConversationsResponse: - paginated = await service.list_conversations(caller, limit=limit, cursor=cursor) + with as_http_error(): + page = PageRequest(limit=limit, cursor=cursor) + paginated = await service.list_conversations(caller, page=page) items = [ ConversationSummary( conversation_id=session.session_id, title=session.title, last_message_at=session.last_message_at, ) - for session in paginated.sessions + for session in paginated.items ] return PaginatedConversationsResponse(items=items, next_cursor=paginated.next_cursor) diff --git a/src/agent_manager/api/static/widget.js b/src/agent_manager/api/static/widget.js index b43547c6..d575d6c9 100644 --- a/src/agent_manager/api/static/widget.js +++ b/src/agent_manager/api/static/widget.js @@ -52176,7 +52176,7 @@ var AgentChatClient = class { })); return { items, - next_cursor: data.next_cursor ? String(data.next_cursor) : null + next_cursor: data.next_cursor ?? null }; } async getMessages(conversationId) { @@ -53399,7 +53399,7 @@ function useConversation(client, endpoint, onReplaced) { [client] ); const listThreads = (0, import_react9.useCallback)( - (limit, cursor) => client.listConversations(limit, cursor).catch(() => ({ items: [], next_cursor: null })), + (limit, cursor) => client.listConversations(limit, cursor), [client] ); const switchTo = (0, import_react9.useCallback)( @@ -53444,6 +53444,8 @@ function useConversation(client, endpoint, onReplaced) { // src/agent_manager/api/static/widget/react/AgentChatApp.tsx var import_jsx_runtime4 = __toESM(require_jsx_runtime(), 1); +var THREADS_PAGE_SIZE = 20; +var SCROLL_THRESHOLD_PX = 40; var DEFAULT_GREETING = "How can I help you today?"; var GENERIC_ERROR = "Something went wrong. Please try again."; var COPIED_RESET_MS = 2e3; @@ -53491,8 +53493,10 @@ function AgentChatApp({ const [threads, setThreads] = (0, import_react10.useState)([]); const [nextCursor, setNextCursor] = (0, import_react10.useState)(null); const [loadingMoreThreads, setLoadingMoreThreads] = (0, import_react10.useState)(false); - const [hasMoreThreads, setHasMoreThreads] = (0, import_react10.useState)(false); const [threadsOpen, setThreadsOpen] = (0, import_react10.useState)(false); + const hasMoreThreads = nextCursor !== null; + const isLoadingMoreRef = (0, import_react10.useRef)(false); + const threadsGenerationRef = (0, import_react10.useRef)(0); const launcherRef = (0, import_react10.useRef)(null); const inputRef = (0, import_react10.useRef)(null); const approvalRequestsRef = (0, import_react10.useRef)(/* @__PURE__ */ new Set()); @@ -53531,8 +53535,8 @@ function AgentChatApp({ }, []); const refreshUsage = (0, import_react10.useCallback)( async (cid) => { - const u4 = await conversation.loadUsage(cid); - setUsageById((prev) => ({ ...prev, [cid]: u4 })); + const next2 = await conversation.loadUsage(cid); + setUsageById((prev) => ({ ...prev, [cid]: next2 })); }, [conversation] ); @@ -53573,27 +53577,44 @@ function AgentChatApp({ launcherRef.current?.focus({ preventScroll: true }); }, [inline]); const openThreads = (0, import_react10.useCallback)(async () => { + threadsGenerationRef.current += 1; + const currentGen = threadsGenerationRef.current; setThreadsOpen(true); setLoadingMoreThreads(true); - const res = await conversation.listThreads(20, null); - setThreads(res.items); - setNextCursor(res.next_cursor); - setHasMoreThreads(res.next_cursor !== null); - setLoadingMoreThreads(false); + isLoadingMoreRef.current = true; + try { + const res = await conversation.listThreads(THREADS_PAGE_SIZE, null); + if (threadsGenerationRef.current !== currentGen) return; + setThreads(res.items); + setNextCursor(res.next_cursor); + } finally { + if (threadsGenerationRef.current === currentGen) { + setLoadingMoreThreads(false); + isLoadingMoreRef.current = false; + } + } }, [conversation]); const loadMoreThreads = (0, import_react10.useCallback)(async () => { - if (loadingMoreThreads || !hasMoreThreads || !nextCursor) return; + if (isLoadingMoreRef.current || !nextCursor) return; + const currentGen = threadsGenerationRef.current; + isLoadingMoreRef.current = true; setLoadingMoreThreads(true); - const res = await conversation.listThreads(20, nextCursor); - setThreads((prev) => { - const existingIds = new Set(prev.map((t) => t.conversation_id)); - const newItems = res.items.filter((t) => !existingIds.has(t.conversation_id)); - return [...prev, ...newItems]; - }); - setNextCursor(res.next_cursor); - setHasMoreThreads(res.next_cursor !== null); - setLoadingMoreThreads(false); - }, [conversation, hasMoreThreads, loadingMoreThreads, nextCursor]); + try { + const res = await conversation.listThreads(THREADS_PAGE_SIZE, nextCursor); + if (threadsGenerationRef.current !== currentGen) return; + setThreads((prev) => { + const existingIds = new Set(prev.map((t) => t.conversation_id)); + const newItems = res.items.filter((t) => !existingIds.has(t.conversation_id)); + return [...prev, ...newItems]; + }); + setNextCursor(res.next_cursor); + } finally { + if (threadsGenerationRef.current === currentGen) { + setLoadingMoreThreads(false); + isLoadingMoreRef.current = false; + } + } + }, [conversation, nextCursor]); const openThread = (0, import_react10.useCallback)( async (conversationId) => { conversation.switchTo(conversationId); @@ -54202,7 +54223,7 @@ function ThreadDrawer({ }) { const handleScroll = (e) => { const { scrollTop, clientHeight, scrollHeight } = e.currentTarget; - if (scrollHeight - scrollTop - clientHeight < 40 && hasMore && !loadingMore) { + if (scrollHeight - scrollTop - clientHeight < SCROLL_THRESHOLD_PX && hasMore && !loadingMore) { onLoadMore(); } }; diff --git a/src/agent_manager/api/static/widget/api/AgentChatClient.ts b/src/agent_manager/api/static/widget/api/AgentChatClient.ts index 364795ea..7c5348d4 100644 --- a/src/agent_manager/api/static/widget/api/AgentChatClient.ts +++ b/src/agent_manager/api/static/widget/api/AgentChatClient.ts @@ -74,15 +74,16 @@ export class AgentChatClient { const response = await this.request(`/conversations?${params.toString()}`); const data = await response.json(); - const rawItems = Array.isArray(data.items) ? data.items : []; - const items: ThreadSummary[] = rawItems.map((thread: any) => ({ + const rawItems: Array<{ conversation_id: string; title?: string | null; last_message_at?: string | null }> = + Array.isArray(data.items) ? data.items : []; + const items: ThreadSummary[] = rawItems.map((thread) => ({ conversation_id: String(thread.conversation_id), title: thread.title ?? null, last_message_at: thread.last_message_at ?? null, })); return { items, - next_cursor: data.next_cursor ? String(data.next_cursor) : null, + next_cursor: (data.next_cursor as string | null) ?? null, }; } diff --git a/src/agent_manager/api/static/widget/react/AgentChatApp.tsx b/src/agent_manager/api/static/widget/react/AgentChatApp.tsx index 26497f3b..594520ad 100644 --- a/src/agent_manager/api/static/widget/react/AgentChatApp.tsx +++ b/src/agent_manager/api/static/widget/react/AgentChatApp.tsx @@ -43,6 +43,9 @@ import { import { reduceStreamEvent } from "./streamReducer"; import { useConversation } from "./useConversation"; +const THREADS_PAGE_SIZE = 20; +const SCROLL_THRESHOLD_PX = 40; + const DEFAULT_GREETING = "How can I help you today?"; const GENERIC_ERROR = "Something went wrong. Please try again."; const COPIED_RESET_MS = 2000; @@ -116,8 +119,10 @@ export function AgentChatApp({ const [threads, setThreads] = useState([]); const [nextCursor, setNextCursor] = useState(null); const [loadingMoreThreads, setLoadingMoreThreads] = useState(false); - const [hasMoreThreads, setHasMoreThreads] = useState(false); const [threadsOpen, setThreadsOpen] = useState(false); + const hasMoreThreads = nextCursor !== null; + const isLoadingMoreRef = useRef(false); + const threadsGenerationRef = useRef(0); const launcherRef = useRef(null); const inputRef = useRef(null); const approvalRequestsRef = useRef(new Set()); @@ -165,8 +170,8 @@ export function AgentChatApp({ const refreshUsage = useCallback( async (cid: string) => { - const u = await conversation.loadUsage(cid); - setUsageById((prev) => ({ ...prev, [cid]: u })); + const next = await conversation.loadUsage(cid); + setUsageById((prev) => ({ ...prev, [cid]: next })); }, [conversation], ); @@ -216,28 +221,48 @@ export function AgentChatApp({ }, [inline]); const openThreads = useCallback(async () => { + threadsGenerationRef.current += 1; + const currentGen = threadsGenerationRef.current; setThreadsOpen(true); setLoadingMoreThreads(true); - const res = await conversation.listThreads(20, null); - setThreads(res.items); - setNextCursor(res.next_cursor); - setHasMoreThreads(res.next_cursor !== null); - setLoadingMoreThreads(false); + isLoadingMoreRef.current = true; + try { + const res = await conversation.listThreads(THREADS_PAGE_SIZE, null); + if (threadsGenerationRef.current !== currentGen) return; + setThreads(res.items); + setNextCursor(res.next_cursor); + } finally { + if (threadsGenerationRef.current === currentGen) { + setLoadingMoreThreads(false); + isLoadingMoreRef.current = false; + } + } }, [conversation]); const loadMoreThreads = useCallback(async () => { - if (loadingMoreThreads || !hasMoreThreads || !nextCursor) return; + if (isLoadingMoreRef.current || !nextCursor) return; + const currentGen = threadsGenerationRef.current; + isLoadingMoreRef.current = true; setLoadingMoreThreads(true); - const res = await conversation.listThreads(20, nextCursor); - setThreads((prev) => { - const existingIds = new Set(prev.map((t) => t.conversation_id)); - const newItems = res.items.filter((t) => !existingIds.has(t.conversation_id)); - return [...prev, ...newItems]; - }); - setNextCursor(res.next_cursor); - setHasMoreThreads(res.next_cursor !== null); - setLoadingMoreThreads(false); - }, [conversation, hasMoreThreads, loadingMoreThreads, nextCursor]); + try { + const res = await conversation.listThreads(THREADS_PAGE_SIZE, nextCursor); + if (threadsGenerationRef.current !== currentGen) return; + setThreads((prev) => { + // Keyset pagination sorts by (last_message_at, session_id). Since last_message_at + // is mutable, newly active threads can jump across pages. The drawer presents a snapshot + // taken when opened; deduplication prevents duplicate items if order mutates mid-scroll. + const existingIds = new Set(prev.map((t) => t.conversation_id)); + const newItems = res.items.filter((t) => !existingIds.has(t.conversation_id)); + return [...prev, ...newItems]; + }); + setNextCursor(res.next_cursor); + } finally { + if (threadsGenerationRef.current === currentGen) { + setLoadingMoreThreads(false); + isLoadingMoreRef.current = false; + } + } + }, [conversation, nextCursor]); const openThread = useCallback( async (conversationId: string) => { @@ -972,7 +997,7 @@ function ThreadDrawer({ }) { const handleScroll = (e: React.UIEvent) => { const { scrollTop, clientHeight, scrollHeight } = e.currentTarget; - if (scrollHeight - scrollTop - clientHeight < 40 && hasMore && !loadingMore) { + if (scrollHeight - scrollTop - clientHeight < SCROLL_THRESHOLD_PX && hasMore && !loadingMore) { onLoadMore(); } }; diff --git a/src/agent_manager/api/static/widget/react/useConversation.ts b/src/agent_manager/api/static/widget/react/useConversation.ts index f8c974c1..02849442 100644 --- a/src/agent_manager/api/static/widget/react/useConversation.ts +++ b/src/agent_manager/api/static/widget/react/useConversation.ts @@ -185,8 +185,7 @@ export function useConversation( ); const listThreads = useCallback( - (limit?: number, cursor?: string | null) => - client.listConversations(limit, cursor).catch(() => ({ items: [], next_cursor: null })), + (limit?: number, cursor?: string | null) => client.listConversations(limit, cursor), [client], ); diff --git a/src/agent_manager/application/__init__.py b/src/agent_manager/application/__init__.py index fee71641..a9cf7f11 100644 --- a/src/agent_manager/application/__init__.py +++ b/src/agent_manager/application/__init__.py @@ -9,6 +9,7 @@ ConversationMessageNotFound, ConversationNotFound, ConversationTokenBudgetExceeded, + InvalidCursorError, ) from agent_manager.application.prepared_conversation_turn import PreparedConversationTurn @@ -21,5 +22,6 @@ "ConversationNotFound", "ConversationService", "ConversationTokenBudgetExceeded", + "InvalidCursorError", "PreparedConversationTurn", ] diff --git a/src/agent_manager/application/conversation_service.py b/src/agent_manager/application/conversation_service.py index 6e50290a..24b71126 100644 --- a/src/agent_manager/application/conversation_service.py +++ b/src/agent_manager/application/conversation_service.py @@ -42,7 +42,8 @@ from agent_manager.domain import ( ConversationMessage, ConversationSession, - PaginatedSessions, + Page, + PageRequest, Principal, Repository, Role, @@ -137,9 +138,9 @@ async def usage(self, conversation_id: str, principal: Principal) -> TokenBudget return TokenBudgetUsage.from_totals(used, self._max_tokens) async def list_conversations( - self, principal: Principal, *, limit: int = 50, cursor: str | None = None - ) -> PaginatedSessions: - return await self._repository.list_sessions(principal.user_id, limit=limit, cursor=cursor) + self, principal: Principal, page: PageRequest | None = None + ) -> Page[ConversationSession]: + return await self._repository.list_sessions(principal.user_id, page=page) async def send( self, diff --git a/src/agent_manager/application/errors.py b/src/agent_manager/application/errors.py index 2b6294ff..a12a792d 100644 --- a/src/agent_manager/application/errors.py +++ b/src/agent_manager/application/errors.py @@ -1,5 +1,18 @@ """Application-layer failures exposed by conversation use cases.""" +from agent_manager.infrastructure.persistence.pagination import InvalidCursorError + +__all__ = [ + "ConversationAccessDenied", + "ConversationAlreadyExists", + "ConversationBranchConflict", + "ConversationLinkRefused", + "ConversationMessageNotFound", + "ConversationNotFound", + "ConversationTokenBudgetExceeded", + "InvalidCursorError", +] + class ConversationNotFound(Exception): """An operation targeted a conversation id that does not exist.""" diff --git a/src/agent_manager/domain/__init__.py b/src/agent_manager/domain/__init__.py index 1fe8c61b..0c4ada46 100644 --- a/src/agent_manager/domain/__init__.py +++ b/src/agent_manager/domain/__init__.py @@ -2,12 +2,16 @@ from agent_manager.domain.identity import IdentityNamespace, Principal from agent_manager.domain.models import ( + DEFAULT_PAGE_LIMIT, + MAX_PAGE_LIMIT, BudgetSeverity, ConversationContext, ConversationMessage, ConversationSession, ConversationSnapshot, Message, + Page, + PageRequest, PaginatedSessions, Role, TokenBudgetUsage, @@ -15,15 +19,21 @@ thread_title, ) from agent_manager.domain.repository import Repository +from agent_manager.infrastructure.persistence.pagination import InvalidCursorError __all__ = [ + "DEFAULT_PAGE_LIMIT", + "MAX_PAGE_LIMIT", "BudgetSeverity", "ConversationContext", "ConversationMessage", "ConversationSession", "ConversationSnapshot", "IdentityNamespace", + "InvalidCursorError", "Message", + "Page", + "PageRequest", "PaginatedSessions", "Principal", "Repository", diff --git a/src/agent_manager/domain/models.py b/src/agent_manager/domain/models.py index bc42f560..db89f837 100644 --- a/src/agent_manager/domain/models.py +++ b/src/agent_manager/domain/models.py @@ -5,7 +5,9 @@ from dataclasses import dataclass, field from datetime import datetime from enum import StrEnum -from typing import Any +from typing import Any, Generic, TypeVar + +T = TypeVar("T") BUDGET_WARNING_PERCENT = 65.0 BUDGET_CRITICAL_PERCENT = 85.0 @@ -56,12 +58,31 @@ class ConversationSession: expires_at: datetime | None = None +DEFAULT_PAGE_LIMIT = 20 +MAX_PAGE_LIMIT = 100 + + @dataclass(frozen=True) -class PaginatedSessions: - sessions: list[ConversationSession] +class PageRequest: + limit: int = DEFAULT_PAGE_LIMIT + cursor: str | None = None + + def __post_init__(self) -> None: + bounded_limit = max(1, min(self.limit, MAX_PAGE_LIMIT)) + object.__setattr__(self, "limit", bounded_limit) + if self.cursor is not None and not self.cursor.strip(): + object.__setattr__(self, "cursor", None) + + +@dataclass(frozen=True) +class Page(Generic[T]): + items: list[T] next_cursor: str | None = None +PaginatedSessions = Page[ConversationSession] + + @dataclass(frozen=True) class ConversationMessage: message_id: str diff --git a/src/agent_manager/domain/repository.py b/src/agent_manager/domain/repository.py index d56fdaec..02e5816a 100644 --- a/src/agent_manager/domain/repository.py +++ b/src/agent_manager/domain/repository.py @@ -12,7 +12,8 @@ ConversationSession, ConversationSnapshot, Message, - PaginatedSessions, + Page, + PageRequest, Role, User, ) @@ -68,8 +69,8 @@ async def get_session(self, session_id: str) -> ConversationSession | None: ... @abstractmethod async def list_sessions( - self, user_id: str, *, limit: int = 50, cursor: str | None = None - ) -> PaginatedSessions: + self, user_id: str, page: PageRequest | None = None + ) -> Page[ConversationSession]: """A user's sessions, most-recently-active first, with cursor pagination.""" @abstractmethod diff --git a/src/agent_manager/infrastructure/persistence/memory_repository.py b/src/agent_manager/infrastructure/persistence/memory_repository.py index f7b721bd..da0ebd96 100644 --- a/src/agent_manager/infrastructure/persistence/memory_repository.py +++ b/src/agent_manager/infrastructure/persistence/memory_repository.py @@ -7,6 +7,7 @@ from dataclasses import replace from datetime import UTC, datetime from typing import Any +from uuid import uuid4 from agent_manager.domain import ( ConversationContext, @@ -14,12 +15,14 @@ ConversationSession, ConversationSnapshot, Message, - PaginatedSessions, + Page, + PageRequest, Repository, Role, User, ) -from agent_manager.infrastructure.persistence.sql_repository import ( +from agent_manager.infrastructure.persistence.pagination import ( + _utc, decode_cursor, encode_cursor, ) @@ -27,15 +30,20 @@ _EPOCH = datetime(1970, 1, 1, tzinfo=UTC) -def _is_after_cursor(s: ConversationSession, cursor_t: datetime | None, cursor_id: str) -> bool: - sid = s.session_id or "" - if cursor_t is not None: - if s.last_message_at is None: - return True - if s.last_message_at < cursor_t: - return True - return s.last_message_at == cursor_t and sid < cursor_id - return s.last_message_at is None and sid < cursor_id +def _effective_t(s: ConversationSession) -> datetime: + dt = s.last_message_at or s.created_at or _EPOCH + res = _utc(dt) + return res if res is not None else _EPOCH + + +def _is_after_cursor(s: ConversationSession, cursor_t: datetime, cursor_id: str) -> bool: + target_t = _utc(cursor_t) or _EPOCH + eff_t = _effective_t(s) + if eff_t < target_t: + return True + if eff_t == target_t: + return (s.session_id or "") < cursor_id + return False class MemoryRepository(Repository): @@ -54,23 +62,23 @@ async def upsert_user( display_name: str | None = None, metadata: dict[str, Any] | None = None, ) -> User: - now = datetime.now(UTC) existing = self._users.get(user_id) + now = datetime.now(UTC) user = User( user_id=user_id, external_user_id=external_user_id if external_user_id is not None - else existing.external_user_id - if existing - else None, - username=username if username is not None else existing.username if existing else None, + else (existing.external_user_id if existing else None), + username=username + if username is not None + else (existing.username if existing else None), display_name=display_name if display_name is not None - else existing.display_name - if existing - else None, + else (existing.display_name if existing else None), linked_to_user_id=existing.linked_to_user_id if existing else None, - metadata=dict(metadata or (existing.metadata if existing else {})), + metadata=dict( + metadata if metadata is not None else (existing.metadata if existing else {}) + ), created_at=existing.created_at if existing else now, updated_at=now, ) @@ -81,14 +89,30 @@ async def get_user(self, user_id: str) -> User | None: return self._users.get(user_id) async def link_anonymous_user(self, anonymous_user_id: str, user_id: str) -> int: - visitor = self._users.get(anonymous_user_id) - if visitor is None or visitor.linked_to_user_id is not None: + if anonymous_user_id == user_id: + return 0 + + target = self._users.get(user_id) + if target and target.linked_to_user_id: return 0 - self._users[anonymous_user_id] = replace(visitor, linked_to_user_id=user_id) - moved = [s for s in self._sessions.values() if s.user_id == anonymous_user_id] - for session in moved: - self._sessions[session.session_id] = replace(session, user_id=user_id) - return len(moved) + + anon = self._users.get(anonymous_user_id) + if anon and anon.linked_to_user_id: + return 0 + + moved = 0 + now = datetime.now(UTC) + for sid, session in list(self._sessions.items()): + if session.user_id == anonymous_user_id: + self._sessions[sid] = replace(session, user_id=user_id, updated_at=now) + moved += 1 + + if anon: + self._users[anonymous_user_id] = replace( + anon, linked_to_user_id=user_id, updated_at=now + ) + + return moved async def create_session( self, @@ -101,11 +125,11 @@ async def create_session( metadata: dict[str, Any] | None = None, expires_at: datetime | None = None, ) -> ConversationSession: - sid = session_id or uuid.uuid4().hex + sid = session_id or uuid4().hex + if sid in self._sessions: + return self._sessions[sid] + now = datetime.now(UTC) - existing = self._sessions.get(sid) - if existing is not None: - return existing session = ConversationSession( session_id=sid, user_id=user_id, @@ -126,15 +150,14 @@ async def get_session(self, session_id: str) -> ConversationSession | None: return self._sessions.get(session_id) async def list_sessions( - self, user_id: str, *, limit: int = 50, cursor: str | None = None - ) -> PaginatedSessions: + self, user_id: str, page: PageRequest | None = None + ) -> Page[ConversationSession]: + page = page or PageRequest() + limit = page.limit + cursor = page.cursor sessions = [s for s in self._sessions.values() if s.user_id == user_id] sessions.sort( - key=lambda s: ( - s.last_message_at is not None, - s.last_message_at or _EPOCH, - s.session_id, - ), + key=lambda s: (_effective_t(s), s.session_id or ""), reverse=True, ) @@ -145,11 +168,14 @@ async def list_sessions( has_more = len(sessions) > limit result_sessions = sessions[:limit] if has_more else sessions next_cursor = ( - encode_cursor(result_sessions[-1].last_message_at, result_sessions[-1].session_id) + encode_cursor( + _effective_t(result_sessions[-1]), + result_sessions[-1].session_id, + ) if has_more and result_sessions else None ) - return PaginatedSessions(sessions=result_sessions, next_cursor=next_cursor) + return Page(items=result_sessions, next_cursor=next_cursor) async def rename_session(self, session_id: str, title: str) -> None: session = self._sessions.get(session_id) diff --git a/src/agent_manager/infrastructure/persistence/migrations/versions/0005_add_session_pagination_index.py b/src/agent_manager/infrastructure/persistence/migrations/versions/0005_add_session_pagination_index.py new file mode 100644 index 00000000..16200355 --- /dev/null +++ b/src/agent_manager/infrastructure/persistence/migrations/versions/0005_add_session_pagination_index.py @@ -0,0 +1,31 @@ +"""add session pagination composite index + +Revision ID: 0005 +Revises: 0004 +Create Date: 2026-08-22 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision: str = "0005" +down_revision: str | None = "0004" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_index( + "idx_conversation_sessions_user_active_session", + "conversation_sessions", + ["user_id", sa.text("COALESCE(last_message_at, created_at)"), "session_id"], + ) + + +def downgrade() -> None: + op.drop_index( + "idx_conversation_sessions_user_active_session", + table_name="conversation_sessions", + ) diff --git a/src/agent_manager/infrastructure/persistence/pagination.py b/src/agent_manager/infrastructure/persistence/pagination.py new file mode 100644 index 00000000..735f2c77 --- /dev/null +++ b/src/agent_manager/infrastructure/persistence/pagination.py @@ -0,0 +1,59 @@ +"""Pagination token encoding, decoding, and error types.""" + +from __future__ import annotations + +import base64 +import binascii +import json +from datetime import UTC, datetime + + +class InvalidCursorError(Exception): + """Raised when a pagination cursor token is malformed or unparseable.""" + + +def _utc(value: datetime | None) -> datetime | None: + if value is None: + return None + if value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + + +def encode_cursor(active_at: datetime, session_id: str) -> str: + """Encode an active timestamp and session ID into an opaque base64 cursor token.""" + utc_dt = _utc(active_at) + if utc_dt is None: + raise ValueError("active_at timestamp cannot be None") + payload = { + "t": utc_dt.isoformat(), + "id": session_id, + } + raw_bytes = json.dumps(payload).encode("utf-8") + return base64.urlsafe_b64encode(raw_bytes).decode("ascii") + + +def decode_cursor(cursor_str: str) -> tuple[datetime, str]: + """Decode a cursor token into (active_at_utc, session_id).""" + if not cursor_str or not cursor_str.strip(): + raise InvalidCursorError("Pagination cursor token cannot be empty") + try: + raw_bytes = base64.urlsafe_b64decode(cursor_str.encode("ascii")) + data = json.loads(raw_bytes.decode("utf-8")) + if not isinstance(data, dict) or "id" not in data or "t" not in data or data["t"] is None: + raise ValueError("Invalid cursor payload structure") + parsed_dt = datetime.fromisoformat(data["t"]) + active_at = _utc(parsed_dt) + if active_at is None: + raise ValueError("Failed to parse active_at timestamp") + session_id = str(data["id"]) + return active_at, session_id + except ( + binascii.Error, + UnicodeDecodeError, + json.JSONDecodeError, + ValueError, + KeyError, + TypeError, + ) as exc: + raise InvalidCursorError("Invalid pagination cursor token") from exc diff --git a/src/agent_manager/infrastructure/persistence/sql_repository.py b/src/agent_manager/infrastructure/persistence/sql_repository.py index 70206065..19333876 100644 --- a/src/agent_manager/infrastructure/persistence/sql_repository.py +++ b/src/agent_manager/infrastructure/persistence/sql_repository.py @@ -6,14 +6,12 @@ from __future__ import annotations -import base64 -import json import uuid from dataclasses import replace from datetime import UTC, datetime, timedelta from typing import Any -from sqlalchemy import and_, delete, literal, or_, update +from sqlalchemy import and_, delete, func, literal, or_, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import async_sessionmaker from sqlalchemy.orm import aliased @@ -26,11 +24,17 @@ ConversationSession, ConversationSnapshot, Message, - PaginatedSessions, + Page, + PageRequest, Repository, Role, User, ) +from agent_manager.infrastructure.persistence.pagination import ( + _utc, + decode_cursor, + encode_cursor, +) from agent_manager.infrastructure.persistence.tables import ( ConversationMessageRow, ConversationSessionRow, @@ -39,27 +43,6 @@ ) -def encode_cursor(last_message_at: datetime | None, session_id: str) -> str: - payload = { - "t": last_message_at.isoformat() if last_message_at is not None else None, - "id": session_id, - } - raw_bytes = json.dumps(payload).encode("utf-8") - return base64.urlsafe_b64encode(raw_bytes).decode("ascii") - - -def decode_cursor(cursor_str: str) -> tuple[datetime | None, str]: - try: - raw_bytes = base64.urlsafe_b64decode(cursor_str.encode("ascii")) - data = json.loads(raw_bytes.decode("utf-8")) - t_str = data.get("t") - last_message_at = datetime.fromisoformat(t_str) if t_str is not None else None - session_id = str(data["id"]) - return last_message_at, session_id - except Exception as exc: - raise ValueError(f"Invalid pagination cursor: {cursor_str}") from exc - - class SqlRepository(Repository): def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None: self._sessions = sessions @@ -176,36 +159,33 @@ async def get_session(self, session_id: str) -> ConversationSession | None: return _session(row) if row else None async def list_sessions( - self, user_id: str, *, limit: int = 50, cursor: str | None = None - ) -> PaginatedSessions: + self, user_id: str, page: PageRequest | None = None + ) -> Page[ConversationSession]: + page = page or PageRequest() + limit = page.limit + cursor = page.cursor + sort_key = func.coalesce( + ConversationSessionRow.last_message_at, ConversationSessionRow.created_at + ) conditions: list[Any] = [ConversationSessionRow.user_id == user_id] if cursor is not None: cursor_t, cursor_id = decode_cursor(cursor) - if cursor_t is not None: - conditions.append( - or_( - col(ConversationSessionRow.last_message_at) < cursor_t, - and_( - col(ConversationSessionRow.last_message_at) == cursor_t, - col(ConversationSessionRow.session_id) < cursor_id, - ), - col(ConversationSessionRow.last_message_at).is_(None), - ) - ) - else: - conditions.append( + conditions.append( + or_( + sort_key < cursor_t, and_( - col(ConversationSessionRow.last_message_at).is_(None), + sort_key == cursor_t, col(ConversationSessionRow.session_id) < cursor_id, - ) + ), ) + ) stmt = ( select(ConversationSessionRow) .where(*conditions) .order_by( - col(ConversationSessionRow.last_message_at).desc().nullslast(), + sort_key.desc(), col(ConversationSessionRow.session_id).desc(), ) .limit(limit + 1) @@ -218,13 +198,16 @@ async def list_sessions( rows = rows[:limit] next_cursor = ( - encode_cursor(rows[-1].last_message_at, rows[-1].session_id) + encode_cursor( + rows[-1].last_message_at or rows[-1].created_at, + rows[-1].session_id, + ) if has_more and rows else None ) - return PaginatedSessions( - sessions=[_session(row) for row in rows], + return Page( + items=[_session(row) for row in rows], next_cursor=next_cursor, ) @@ -717,11 +700,3 @@ def _bound_messages( if total >= max_chars: break return list(reversed(kept)) - - -def _utc(value: datetime | None) -> datetime | None: - if value is None: - return None - if value.tzinfo is None: - return value.replace(tzinfo=UTC) - return value.astimezone(UTC) diff --git a/src/agent_manager/infrastructure/persistence/tables.py b/src/agent_manager/infrastructure/persistence/tables.py index 258735da..05612905 100644 --- a/src/agent_manager/infrastructure/persistence/tables.py +++ b/src/agent_manager/infrastructure/persistence/tables.py @@ -14,7 +14,7 @@ from datetime import datetime -from sqlalchemy import JSON, Column, DateTime, Float, Index, Integer, Text +from sqlalchemy import JSON, Column, DateTime, Float, Index, Integer, Text, func from sqlmodel import Field, SQLModel @@ -33,6 +33,17 @@ class ConversationUserRow(SQLModel, table=True): class ConversationSessionRow(SQLModel, table=True): __tablename__ = "conversation_sessions" + __table_args__ = ( + Index( + "idx_conversation_sessions_user_active_session", + "user_id", + func.coalesce( + Column("last_message_at", DateTime(timezone=True)), + Column("created_at", DateTime(timezone=True)), + ), + "session_id", + ), + ) session_id: str = Field(primary_key=True, max_length=64) user_id: str | None = Field(default=None, foreign_key="conversation_users.user_id", index=True) diff --git a/tests/agent_manager/test_pagination.py b/tests/agent_manager/test_pagination.py index cd8d7a6f..b60fe848 100644 --- a/tests/agent_manager/test_pagination.py +++ b/tests/agent_manager/test_pagination.py @@ -10,13 +10,14 @@ import agent_manager.infrastructure.persistence.tables # noqa: F401 from agent_manager.application import ConversationService +from agent_manager.domain import InvalidCursorError, PageRequest from agent_manager.infrastructure.persistence.database import create_db_engine, session_factory from agent_manager.infrastructure.persistence.memory_repository import MemoryRepository -from agent_manager.infrastructure.persistence.sql_repository import ( - SqlRepository, +from agent_manager.infrastructure.persistence.pagination import ( decode_cursor, encode_cursor, ) +from agent_manager.infrastructure.persistence.sql_repository import SqlRepository from tests.agent_manager.conftest import RecordingEngine, bearer, build_test_app @@ -35,17 +36,40 @@ def test_cursor_encode_decode_round_trip() -> None: assert decoded_id == "sess-123" -def test_cursor_encode_decode_with_null_timestamp() -> None: - cursor = encode_cursor(None, "sess-null") - decoded_t, decoded_id = decode_cursor(cursor) +def test_invalid_cursor_raises_invalid_cursor_error() -> None: + with pytest.raises(InvalidCursorError): + decode_cursor("not-a-valid-cursor!") - assert decoded_t is None - assert decoded_id == "sess-null" +def test_empty_cursor_in_page_request_normalizes_to_none() -> None: + req = PageRequest(limit=20, cursor="") + assert req.cursor is None -def test_invalid_cursor_raises_value_error() -> None: - with pytest.raises(ValueError, match="Invalid pagination cursor"): - decode_cursor("not-a-valid-cursor!") + req_spaces = PageRequest(limit=20, cursor=" ") + assert req_spaces.cursor is None + + +def test_limit_bounding_in_page_request() -> None: + req_high = PageRequest(limit=10_000_000) + assert req_high.limit == 100 + + req_low = PageRequest(limit=-5) + assert req_low.limit == 1 + + +def test_a_malformed_cursor_is_rejected_as_client_error(client: TestClient) -> None: + response = client.get("/conversations?cursor=garbage", headers=bearer("u1")) + assert response.status_code == 400 + detail = response.json().get("detail", {}) + assert detail.get("error_type") == "invalid_cursor" + + +def test_empty_cursor_query_param_returns_first_page(client: TestClient) -> None: + u1 = bearer("user-empty-cursor") + client.post("/conversations", headers=u1) + response = client.get("/conversations?cursor=", headers=u1) + assert response.status_code == 200 + assert len(response.json()["items"]) == 1 @pytest.mark.asyncio @@ -71,6 +95,11 @@ async def test_sql_repository_pagination_and_ordering(tmp_path: Path) -> None: async with sessions() as session: from agent_manager.infrastructure.persistence.tables import ConversationSessionRow + for i in range(1, 6): + r = await session.get(ConversationSessionRow, f"sess-{i}") + if r: + r.created_at = base_time + r1 = await session.get(ConversationSessionRow, "sess-1") assert r1 is not None r1.last_message_at = base_time + timedelta(hours=2) @@ -86,18 +115,18 @@ async def test_sql_repository_pagination_and_ordering(tmp_path: Path) -> None: await session.commit() # Page 1: limit 2 - p1 = await repo.list_sessions(user_id, limit=2) - assert [s.session_id for s in p1.sessions] == ["sess-1", "sess-3"] + p1 = await repo.list_sessions(user_id, page=PageRequest(limit=2)) + assert [s.session_id for s in p1.items] == ["sess-1", "sess-3"] assert p1.next_cursor is not None # Page 2: limit 2 - p2 = await repo.list_sessions(user_id, limit=2, cursor=p1.next_cursor) - assert [s.session_id for s in p2.sessions] == ["sess-2", "sess-5"] + p2 = await repo.list_sessions(user_id, page=PageRequest(limit=2, cursor=p1.next_cursor)) + assert [s.session_id for s in p2.items] == ["sess-2", "sess-5"] assert p2.next_cursor is not None # Page 3: limit 2 - p3 = await repo.list_sessions(user_id, limit=2, cursor=p2.next_cursor) - assert [s.session_id for s in p3.sessions] == ["sess-4"] + p3 = await repo.list_sessions(user_id, page=PageRequest(limit=2, cursor=p2.next_cursor)) + assert [s.session_id for s in p3.items] == ["sess-4"] assert p3.next_cursor is None await engine.dispose() @@ -119,4 +148,4 @@ def test_api_conversations_pagination_endpoint(client: TestClient) -> None: assert res2["next_cursor"] is None fetched_ids = [item["conversation_id"] for item in res1["items"] + res2["items"]] - assert set(fetched_ids) == {c1, c2, c3} + assert fetched_ids == [c3, c2, c1] diff --git a/tests/agent_manager/test_repository_contract.py b/tests/agent_manager/test_repository_contract.py index ccc8cb74..71ea51d9 100644 --- a/tests/agent_manager/test_repository_contract.py +++ b/tests/agent_manager/test_repository_contract.py @@ -13,7 +13,13 @@ from sqlmodel import SQLModel, select import agent_manager.infrastructure.persistence.tables # noqa: F401 (register tables) -from agent_manager.domain import ConversationMessage, Repository, Role +from agent_manager.domain import ( + ConversationMessage, + InvalidCursorError, + PageRequest, + Repository, + Role, +) from agent_manager.infrastructure.persistence.database import create_db_engine, session_factory from agent_manager.infrastructure.persistence.memory_repository import MemoryRepository from agent_manager.infrastructure.persistence.sql_repository import SqlRepository @@ -56,7 +62,7 @@ async def test_create_session_never_reassigns_an_existing_owner(repo: Repository stored = await repo.get_session("shared-id") assert stored is not None assert stored.user_id == "alice" - assert (await repo.list_sessions("bob")).sessions == [] + assert (await repo.list_sessions("bob")).items == [] async def test_create_session_writes_nothing_when_the_id_is_taken(repo: Repository) -> None: @@ -107,7 +113,7 @@ async def test_appending_a_message_never_claims_the_conversation(repo: Repositor unowned = await repo.get_session("unowned") assert owned is not None and owned.user_id == "alice" assert unowned is not None and unowned.user_id is None - assert (await repo.list_sessions("bob")).sessions == [] + assert (await repo.list_sessions("bob")).items == [] async def test_messages_in_insertion_order(repo: Repository) -> None: @@ -468,8 +474,8 @@ async def test_linking_a_visitor_moves_their_sessions_once(repo: Repository) -> assert await repo.link_anonymous_user("anon:v1", "ext:alice") == 1 moved = await repo.get_session("pre-login") assert moved is not None and moved.user_id == "ext:alice" - assert [s.session_id for s in (await repo.list_sessions("ext:alice")).sessions] == ["pre-login"] - assert (await repo.list_sessions("anon:v1")).sessions == [] + assert [s.session_id for s in (await repo.list_sessions("ext:alice")).items] == ["pre-login"] + assert (await repo.list_sessions("anon:v1")).items == [] visitor = await repo.get_user("anon:v1") assert visitor is not None and visitor.linked_to_user_id == "ext:alice" @@ -477,9 +483,40 @@ async def test_linking_a_visitor_moves_their_sessions_once(repo: Repository) -> # Spent: a replayed pass moves nothing, whoever presents it. await repo.upsert_user("ext:bob") assert await repo.link_anonymous_user("anon:v1", "ext:bob") == 0 - assert (await repo.list_sessions("ext:bob")).sessions == [] + assert (await repo.list_sessions("ext:bob")).items == [] async def test_linking_an_unknown_visitor_is_a_no_op(repo: Repository) -> None: await repo.upsert_user("ext:alice") assert await repo.link_anonymous_user("anon:never-seen", "ext:alice") == 0 + + +async def test_pagination_contract(repo: Repository) -> None: + """Comprehensive contract tests for repository pagination.""" + user_id = "paginated_user" + await repo.upsert_user(user_id) + + # 1. Multi-page iteration returning each session exactly once + await repo.create_session("s1", user_id=user_id) + await repo.create_session("s2", user_id=user_id) + await repo.create_session("s3", user_id=user_id) + + page1 = await repo.list_sessions(user_id, page=PageRequest(limit=2)) + assert len(page1.items) == 2 + assert page1.next_cursor is not None + + page2 = await repo.list_sessions(user_id, page=PageRequest(limit=2, cursor=page1.next_cursor)) + assert len(page2.items) == 1 + assert page2.next_cursor is None + + all_ids = [s.session_id for s in page1.items + page2.items] + assert all_ids == ["s3", "s2", "s1"] + + # 2. Page boundary landing exactly on limit + page_exact = await repo.list_sessions(user_id, page=PageRequest(limit=3)) + assert len(page_exact.items) == 3 + assert page_exact.next_cursor is None + + # 3. Malformed cursor raises InvalidCursorError + with pytest.raises(InvalidCursorError): + await repo.list_sessions(user_id, page=PageRequest(cursor="invalid_garbage_token")) diff --git a/tests/agent_manager/test_service.py b/tests/agent_manager/test_service.py index 89f827cb..073b6aa4 100644 --- a/tests/agent_manager/test_service.py +++ b/tests/agent_manager/test_service.py @@ -352,8 +352,8 @@ async def test_reads_of_an_owned_conversation_refuse_other_callers() -> None: with pytest.raises(ConversationAccessDenied): await service.usage(cid, caller) - assert (await service.list_conversations(BOB)).sessions == [] - assert (await service.list_conversations(VISITOR)).sessions == [] + assert (await service.list_conversations(BOB)).items == [] + assert (await service.list_conversations(VISITOR)).items == [] async def test_create_refuses_a_session_id_owned_by_someone_else() -> None: @@ -440,10 +440,8 @@ async def test_signing_in_moves_a_visitors_conversations_onto_their_account() -> moved = await service.link_anonymous(VISITOR, ALICE) assert moved == 1 - assert [s.session_id for s in (await service.list_conversations(ALICE)).sessions] == [ - "pre-login" - ] - assert (await service.list_conversations(VISITOR)).sessions == [] + assert [s.session_id for s in (await service.list_conversations(ALICE)).items] == ["pre-login"] + assert (await service.list_conversations(VISITOR)).items == [] assert [m.content for m in await service.history(before_login, ALICE)] == [ "how much does it cost?", "answer:how much does it cost?", @@ -458,10 +456,8 @@ async def test_a_visitor_pass_can_only_be_adopted_once() -> None: assert await service.link_anonymous(VISITOR, ALICE) == 1 assert await service.link_anonymous(VISITOR, BOB) == 0 - assert [s.session_id for s in (await service.list_conversations(ALICE)).sessions] == [ - "pre-login" - ] - assert (await service.list_conversations(BOB)).sessions == [] + assert [s.session_id for s in (await service.list_conversations(ALICE)).items] == ["pre-login"] + assert (await service.list_conversations(BOB)).items == [] async def test_a_visitor_cannot_adopt_another_visitor() -> None: @@ -479,7 +475,7 @@ async def test_adopting_merges_into_conversations_the_account_already_had() -> N await service.link_anonymous(VISITOR, ALICE) - assert {s.session_id for s in (await service.list_conversations(ALICE)).sessions} == { + assert {s.session_id for s in (await service.list_conversations(ALICE)).items} == { "signed-in", "pre-login", } diff --git a/tests/e2e/widget.spec.ts b/tests/e2e/widget.spec.ts index 74821410..55e2cf37 100644 --- a/tests/e2e/widget.spec.ts +++ b/tests/e2e/widget.spec.ts @@ -1409,3 +1409,55 @@ test("auto-mount does not duplicate an authored element and attributes control t await expect.poll(() => shadowText(page, ".messages")).toContain("The attribute greeting wins."); await expect.poll(() => shadowClassContains(page, ".panel", "open")).toBe(true); }); + +test("thread drawer paginates and appends next pages on scroll", async ({ page }) => { + let callCount = 0; + await pinVisitorPass(page); + + await page.route(/\/conversations\?/, async (route) => { + callCount += 1; + const url = new URL(route.request().url()); + const cursor = url.searchParams.get("cursor"); + if (!cursor) { + await route.fulfill({ + status: 200, + contentType: "application/json", + body: JSON.stringify({ + items: [ + { conversation_id: "thread-1", title: "Thread One", last_message_at: "2026-06-28T00:00:00Z" }, + { conversation_id: "thread-2", title: "Thread Two", last_message_at: "2026-06-27T00:00:00Z" }, + ], + next_cursor: "page-2-token", + }), + }); + } else { + await route.fulfill({ + status: 200, + contentType: "application/json", + body: JSON.stringify({ + items: [ + { conversation_id: "thread-3", title: "Thread Three", last_message_at: "2026-06-26T00:00:00Z" }, + ], + next_cursor: null, + }), + }); + } + }); + + await page.goto("/widget-demo.html"); + await shadowClick(page, ".launcher"); + await shadowClick(page, '[aria-label="Conversations"]'); + + await expect.poll(() => shadowText(page, ".thread-drawer")).toContain("Thread One"); + await expect.poll(() => shadowText(page, ".thread-drawer")).toContain("Thread Two"); + + await (await widget(page)).evaluate((el) => { + const drawer = el.shadowRoot?.querySelector(".thread-list"); + if (drawer) { + drawer.scrollTop = drawer.scrollHeight; + drawer.dispatchEvent(new Event("scroll")); + } + }); + + await expect.poll(() => shadowText(page, ".thread-drawer")).toContain("Thread Three"); +}); From 1bc0a4fb95df06488b66afdf0d184604d1bf0031 Mon Sep 17 00:00:00 2001 From: rishu685 Date: Sun, 23 Aug 2026 23:16:37 +0530 Subject: [PATCH 06/10] refactor(agent_manager): fix architecture layering and pagination review findings - Move pagination codec and InvalidCursorError to domain/pagination.py (eliminates domain -> infra dependency) - Export public ensure_utc helper function across repositories - Explicitly normalize tz-naive SQL row timestamps before cursor encoding - Align table index definition in tables.py with Alembic migration 0005 using text(...) - Add explicit message timestamps in contract tests for deterministic pagination order - Update loadMoreThreads comment in AgentChatApp.tsx --- .../api/static/widget/react/AgentChatApp.tsx | 4 +- src/agent_manager/application/errors.py | 2 +- src/agent_manager/domain/__init__.py | 10 ++- src/agent_manager/domain/pagination.py | 60 +++++++++++++++ .../persistence/memory_repository.py | 8 +- .../infrastructure/persistence/pagination.py | 77 +++++-------------- .../persistence/sql_repository.py | 28 +++---- .../infrastructure/persistence/tables.py | 7 +- .../agent_manager/test_repository_contract.py | 29 ++++++- 9 files changed, 138 insertions(+), 87 deletions(-) create mode 100644 src/agent_manager/domain/pagination.py diff --git a/src/agent_manager/api/static/widget/react/AgentChatApp.tsx b/src/agent_manager/api/static/widget/react/AgentChatApp.tsx index 594520ad..5cca742d 100644 --- a/src/agent_manager/api/static/widget/react/AgentChatApp.tsx +++ b/src/agent_manager/api/static/widget/react/AgentChatApp.tsx @@ -249,8 +249,8 @@ export function AgentChatApp({ if (threadsGenerationRef.current !== currentGen) return; setThreads((prev) => { // Keyset pagination sorts by (last_message_at, session_id). Since last_message_at - // is mutable, newly active threads can jump across pages. The drawer presents a snapshot - // taken when opened; deduplication prevents duplicate items if order mutates mid-scroll. + // is mutable, a thread updated while scrolling could appear across page boundaries; + // deduplication prevents duplicate items if order mutates mid-scroll. const existingIds = new Set(prev.map((t) => t.conversation_id)); const newItems = res.items.filter((t) => !existingIds.has(t.conversation_id)); return [...prev, ...newItems]; diff --git a/src/agent_manager/application/errors.py b/src/agent_manager/application/errors.py index a12a792d..eaccf47a 100644 --- a/src/agent_manager/application/errors.py +++ b/src/agent_manager/application/errors.py @@ -1,6 +1,6 @@ """Application-layer failures exposed by conversation use cases.""" -from agent_manager.infrastructure.persistence.pagination import InvalidCursorError +from agent_manager.domain.pagination import InvalidCursorError __all__ = [ "ConversationAccessDenied", diff --git a/src/agent_manager/domain/__init__.py b/src/agent_manager/domain/__init__.py index 0c4ada46..8a42eca1 100644 --- a/src/agent_manager/domain/__init__.py +++ b/src/agent_manager/domain/__init__.py @@ -18,8 +18,13 @@ User, thread_title, ) +from agent_manager.domain.pagination import ( + InvalidCursorError, + decode_cursor, + encode_cursor, + ensure_utc, +) from agent_manager.domain.repository import Repository -from agent_manager.infrastructure.persistence.pagination import InvalidCursorError __all__ = [ "DEFAULT_PAGE_LIMIT", @@ -40,5 +45,8 @@ "Role", "TokenBudgetUsage", "User", + "decode_cursor", + "encode_cursor", + "ensure_utc", "thread_title", ] diff --git a/src/agent_manager/domain/pagination.py b/src/agent_manager/domain/pagination.py new file mode 100644 index 00000000..ea4ff976 --- /dev/null +++ b/src/agent_manager/domain/pagination.py @@ -0,0 +1,60 @@ +"""Pagination models, token codec, and error types.""" + +from __future__ import annotations + +import base64 +import binascii +import json +from datetime import UTC, datetime + + +class InvalidCursorError(Exception): + """Raised when a pagination cursor token is malformed or unparseable.""" + + +def ensure_utc(value: datetime | None) -> datetime | None: + """Ensure a datetime is timezone-aware UTC.""" + if value is None: + return None + if value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + + +def encode_cursor(active_at: datetime, session_id: str) -> str: + """Encode an active timestamp and session ID into an opaque base64 cursor token.""" + utc_dt = ensure_utc(active_at) + if utc_dt is None: + raise ValueError("active_at timestamp cannot be None") + payload = { + "t": utc_dt.isoformat(), + "id": session_id, + } + raw_bytes = json.dumps(payload).encode("utf-8") + return base64.urlsafe_b64encode(raw_bytes).decode("ascii") + + +def decode_cursor(cursor_str: str) -> tuple[datetime, str]: + """Decode a cursor token into (active_at_utc, session_id).""" + if not cursor_str or not cursor_str.strip(): + raise InvalidCursorError("Pagination cursor token cannot be empty") + try: + raw_bytes = base64.urlsafe_b64decode(cursor_str.encode("ascii")) + data = json.loads(raw_bytes.decode("utf-8")) + if not isinstance(data, dict) or "id" not in data or "t" not in data or data["t"] is None: + raise ValueError("Invalid cursor payload structure") + parsed_dt = datetime.fromisoformat(data["t"]) + active_at = ensure_utc(parsed_dt) + if active_at is None: + raise ValueError("Failed to parse active_at timestamp") + session_id = str(data["id"]) + return active_at, session_id + except ( + binascii.Error, + UnicodeDecodeError, + json.JSONDecodeError, + ValueError, + KeyError, + TypeError, + ) as exc: + raise InvalidCursorError("Invalid pagination cursor token") from exc diff --git a/src/agent_manager/infrastructure/persistence/memory_repository.py b/src/agent_manager/infrastructure/persistence/memory_repository.py index da0ebd96..fcf08ad2 100644 --- a/src/agent_manager/infrastructure/persistence/memory_repository.py +++ b/src/agent_manager/infrastructure/persistence/memory_repository.py @@ -21,10 +21,10 @@ Role, User, ) -from agent_manager.infrastructure.persistence.pagination import ( - _utc, +from agent_manager.domain.pagination import ( decode_cursor, encode_cursor, + ensure_utc, ) _EPOCH = datetime(1970, 1, 1, tzinfo=UTC) @@ -32,12 +32,12 @@ def _effective_t(s: ConversationSession) -> datetime: dt = s.last_message_at or s.created_at or _EPOCH - res = _utc(dt) + res = ensure_utc(dt) return res if res is not None else _EPOCH def _is_after_cursor(s: ConversationSession, cursor_t: datetime, cursor_id: str) -> bool: - target_t = _utc(cursor_t) or _EPOCH + target_t = ensure_utc(cursor_t) or _EPOCH eff_t = _effective_t(s) if eff_t < target_t: return True diff --git a/src/agent_manager/infrastructure/persistence/pagination.py b/src/agent_manager/infrastructure/persistence/pagination.py index 735f2c77..87228b35 100644 --- a/src/agent_manager/infrastructure/persistence/pagination.py +++ b/src/agent_manager/infrastructure/persistence/pagination.py @@ -1,59 +1,18 @@ -"""Pagination token encoding, decoding, and error types.""" - -from __future__ import annotations - -import base64 -import binascii -import json -from datetime import UTC, datetime - - -class InvalidCursorError(Exception): - """Raised when a pagination cursor token is malformed or unparseable.""" - - -def _utc(value: datetime | None) -> datetime | None: - if value is None: - return None - if value.tzinfo is None: - return value.replace(tzinfo=UTC) - return value.astimezone(UTC) - - -def encode_cursor(active_at: datetime, session_id: str) -> str: - """Encode an active timestamp and session ID into an opaque base64 cursor token.""" - utc_dt = _utc(active_at) - if utc_dt is None: - raise ValueError("active_at timestamp cannot be None") - payload = { - "t": utc_dt.isoformat(), - "id": session_id, - } - raw_bytes = json.dumps(payload).encode("utf-8") - return base64.urlsafe_b64encode(raw_bytes).decode("ascii") - - -def decode_cursor(cursor_str: str) -> tuple[datetime, str]: - """Decode a cursor token into (active_at_utc, session_id).""" - if not cursor_str or not cursor_str.strip(): - raise InvalidCursorError("Pagination cursor token cannot be empty") - try: - raw_bytes = base64.urlsafe_b64decode(cursor_str.encode("ascii")) - data = json.loads(raw_bytes.decode("utf-8")) - if not isinstance(data, dict) or "id" not in data or "t" not in data or data["t"] is None: - raise ValueError("Invalid cursor payload structure") - parsed_dt = datetime.fromisoformat(data["t"]) - active_at = _utc(parsed_dt) - if active_at is None: - raise ValueError("Failed to parse active_at timestamp") - session_id = str(data["id"]) - return active_at, session_id - except ( - binascii.Error, - UnicodeDecodeError, - json.JSONDecodeError, - ValueError, - KeyError, - TypeError, - ) as exc: - raise InvalidCursorError("Invalid pagination cursor token") from exc +"""Infrastructure alias re-exporting pagination utilities from domain.""" + +from agent_manager.domain.pagination import ( + InvalidCursorError, + decode_cursor, + encode_cursor, + ensure_utc, +) + +# Backward-compatibility private alias +_utc = ensure_utc + +__all__ = [ + "InvalidCursorError", + "decode_cursor", + "encode_cursor", + "ensure_utc", +] diff --git a/src/agent_manager/infrastructure/persistence/sql_repository.py b/src/agent_manager/infrastructure/persistence/sql_repository.py index 19333876..30b3560e 100644 --- a/src/agent_manager/infrastructure/persistence/sql_repository.py +++ b/src/agent_manager/infrastructure/persistence/sql_repository.py @@ -30,10 +30,10 @@ Role, User, ) -from agent_manager.infrastructure.persistence.pagination import ( - _utc, +from agent_manager.domain.pagination import ( decode_cursor, encode_cursor, + ensure_utc, ) from agent_manager.infrastructure.persistence.tables import ( ConversationMessageRow, @@ -199,7 +199,7 @@ async def list_sessions( next_cursor = ( encode_cursor( - rows[-1].last_message_at or rows[-1].created_at, + ensure_utc(rows[-1].last_message_at or rows[-1].created_at), # type: ignore[arg-type] rows[-1].session_id, ) if has_more and rows @@ -579,8 +579,8 @@ def _user(row: ConversationUserRow) -> User: display_name=row.display_name, linked_to_user_id=row.linked_to_user_id, metadata=dict(row.metadata_json or {}), - created_at=_utc(row.created_at), - updated_at=_utc(row.updated_at), + created_at=ensure_utc(row.created_at), + updated_at=ensure_utc(row.updated_at), ) @@ -593,10 +593,10 @@ def _session(row: ConversationSessionRow) -> ConversationSession: title=row.title, head_message_id=row.head_message_id, metadata=dict(row.metadata_json or {}), - created_at=_utc(row.created_at), - updated_at=_utc(row.updated_at), - last_message_at=_utc(row.last_message_at), - expires_at=_utc(row.expires_at), + created_at=ensure_utc(row.created_at), + updated_at=ensure_utc(row.updated_at), + last_message_at=ensure_utc(row.last_message_at), + expires_at=ensure_utc(row.expires_at), ) @@ -648,7 +648,7 @@ def _message(row: ConversationMessageRow) -> ConversationMessage: status=row.status, error_type=row.error_type, metadata=dict(row.metadata_json or {}), - created_at=_utc(row.created_at) or row.created_at, + created_at=ensure_utc(row.created_at) or row.created_at, ) @@ -664,7 +664,7 @@ def _message_json(row: ConversationMessageRow) -> dict[str, Any]: "tool_name": row.tool_name, "provider": row.provider, "status": row.status, - "created_at": (_utc(row.created_at) or row.created_at).isoformat(), + "created_at": (ensure_utc(row.created_at) or row.created_at).isoformat(), "metadata": dict(row.metadata_json or {}), } @@ -676,10 +676,10 @@ def _snapshot(row: ConversationSnapshotRow) -> ConversationSnapshot: conversation_json=dict(row.conversation_json or {}), message_count=row.message_count, last_message_id=row.last_message_id, - last_message_at=_utc(row.last_message_at), + last_message_at=ensure_utc(row.last_message_at), model_context_tokens=row.model_context_tokens, - updated_at=_utc(row.updated_at) or row.updated_at, - expires_at=_utc(row.expires_at), + updated_at=ensure_utc(row.updated_at) or row.updated_at, + expires_at=ensure_utc(row.expires_at), ) diff --git a/src/agent_manager/infrastructure/persistence/tables.py b/src/agent_manager/infrastructure/persistence/tables.py index 05612905..a6aa8d9b 100644 --- a/src/agent_manager/infrastructure/persistence/tables.py +++ b/src/agent_manager/infrastructure/persistence/tables.py @@ -14,7 +14,7 @@ from datetime import datetime -from sqlalchemy import JSON, Column, DateTime, Float, Index, Integer, Text, func +from sqlalchemy import JSON, Column, DateTime, Float, Index, Integer, Text, text from sqlmodel import Field, SQLModel @@ -37,10 +37,7 @@ class ConversationSessionRow(SQLModel, table=True): Index( "idx_conversation_sessions_user_active_session", "user_id", - func.coalesce( - Column("last_message_at", DateTime(timezone=True)), - Column("created_at", DateTime(timezone=True)), - ), + text("COALESCE(last_message_at, created_at)"), "session_id", ), ) diff --git a/tests/agent_manager/test_repository_contract.py b/tests/agent_manager/test_repository_contract.py index 71ea51d9..e789229d 100644 --- a/tests/agent_manager/test_repository_contract.py +++ b/tests/agent_manager/test_repository_contract.py @@ -496,10 +496,37 @@ async def test_pagination_contract(repo: Repository) -> None: user_id = "paginated_user" await repo.upsert_user(user_id) - # 1. Multi-page iteration returning each session exactly once + base_time = datetime(2026, 8, 20, 12, 0, 0, tzinfo=UTC) await repo.create_session("s1", user_id=user_id) await repo.create_session("s2", user_id=user_id) await repo.create_session("s3", user_id=user_id) + await repo.append_message( + ConversationMessage( + message_id="m1", + session_id="s1", + role=Role.USER, + content="m1", + created_at=base_time, + ) + ) + await repo.append_message( + ConversationMessage( + message_id="m2", + session_id="s2", + role=Role.USER, + content="m2", + created_at=base_time + timedelta(hours=1), + ) + ) + await repo.append_message( + ConversationMessage( + message_id="m3", + session_id="s3", + role=Role.USER, + content="m3", + created_at=base_time + timedelta(hours=2), + ) + ) page1 = await repo.list_sessions(user_id, page=PageRequest(limit=2)) assert len(page1.items) == 2 From ace375d21d67e1e54b5b5b28fec54c291ffcfb04 Mon Sep 17 00:00:00 2001 From: rishu685 Date: Sun, 23 Aug 2026 23:21:31 +0530 Subject: [PATCH 07/10] docs(adr): add ADR 0003 for chat history keyset pagination BREAKING CHANGE: GET /conversations response shape changed from flat list [ConversationSummary] to paginated envelope { items: [ConversationSummary], next_cursor: str | null } --- ...at-history-pagination-and-keyset-cursor.md | 39 +++++++++++++++++++ 1 file changed, 39 insertions(+) create mode 100644 docs/adr/0003-chat-history-pagination-and-keyset-cursor.md diff --git a/docs/adr/0003-chat-history-pagination-and-keyset-cursor.md b/docs/adr/0003-chat-history-pagination-and-keyset-cursor.md new file mode 100644 index 00000000..d96058dc --- /dev/null +++ b/docs/adr/0003-chat-history-pagination-and-keyset-cursor.md @@ -0,0 +1,39 @@ +# ADR 0003 — Chat history keyset pagination and cursor format + +- **Status:** Accepted +- **Date:** 2026-08-23 + +## Context + +`GET /conversations` previously returned a flat, unbounded list of all conversations for a user. As conversation counts grow per user, loading the complete list in one request causes high latency, database scan overhead, and excessive payload sizes for clients. + +Additionally, simple offset pagination (`OFFSET N`) suffers from severe database performance degradation on large offsets and produces inconsistent results (skipped or duplicated rows) if conversations receive messages or are created while a user is paginating. + +## Decision + +1. **Keyset Cursor Pagination**: + - `GET /conversations` uses keyset pagination based on the composite ordering key `(COALESCE(last_message_at, created_at) DESC, session_id DESC)`. + - The active timestamp `COALESCE(last_message_at, created_at)` ensures empty conversations without messages sort predictably by their creation time alongside active threads. + +2. **Opaque Base64 Cursor Token**: + - Cursors are opaque server tokens containing base64-encoded JSON `{"t": "", "id": ""}`. + - The pagination token codec (`encode_cursor`, `decode_cursor`) and exception `InvalidCursorError` live in `agent_manager.domain.pagination` (pure Python domain value objects and utilities with zero framework dependencies). + - Malformed or invalid cursor tokens raise `InvalidCursorError`, which is mapped by `as_http_error()` to `HTTP 400 Bad Request` with payload `{ "error_type": "invalid_cursor", "message": "invalid pagination cursor" }`. + +3. **Domain Layer Bounds**: + - `PageRequest` value object encapsulates pagination parameters (`limit`, `cursor`). + - Default page size is 20 (`DEFAULT_PAGE_LIMIT`) and maximum page limit is 100 (`MAX_PAGE_LIMIT`), enforced at domain instantiation time in `PageRequest.__post_init__`. + +4. **Database Indexing**: + - Migration `0005_add_session_pagination_index.py` and `tables.py` add an expression index `idx_conversation_sessions_user_active_session` on `(user_id, COALESCE(last_message_at, created_at), session_id)` to enable fast index range seeks for keyset pagination. + +## Contract changes + +- **BREAKING CHANGE**: `GET /conversations` response shape changed from a flat list `[ConversationSummary, ...]` to a paginated envelope object `{ "items": [ConversationSummary, ...], "next_cursor": "..." | null }`. +- `GET /conversations` accepts optional query parameters `limit` (integer, 1..100) and `cursor` (opaque string). + +## Consequences + +- Clients fetch subsequent pages using `next_cursor` until `next_cursor` is `null`. +- Keyset range seeks eliminate `OFFSET` database performance degradation and prevent skipped/duplicated sessions when thread activity changes mid-page. +- Frontends deduplicate threads by `conversation_id` to handle live thread updates gracefully. From b5e757837e7fe28de1dd0f2b10fb2fa1a2bb3a06 Mon Sep 17 00:00:00 2001 From: rishu685 Date: Sun, 23 Aug 2026 23:24:36 +0530 Subject: [PATCH 08/10] fix(widget): add drawer error notice and retry UI, and expand contract tests - Add error handling and retry UI in thread drawer - Add Load More button for non-overflowing list viewports - Annotate Page[T].items as Sequence[T] - Expand contract test to 5 sessions with mixed last_message_at and created_at timestamps --- src/agent_manager/api/static/widget.js | 60 +++++++++----- .../api/static/widget/react/AgentChatApp.tsx | 38 ++++++++- src/agent_manager/domain/models.py | 3 +- .../agent_manager/test_repository_contract.py | 82 ++++++++++++++----- 4 files changed, 137 insertions(+), 46 deletions(-) diff --git a/src/agent_manager/api/static/widget.js b/src/agent_manager/api/static/widget.js index d575d6c9..e9614007 100644 --- a/src/agent_manager/api/static/widget.js +++ b/src/agent_manager/api/static/widget.js @@ -52641,20 +52641,6 @@ function randomId() { return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`; } -// src/agent_manager/api/static/widget/storage/conversationStorage.ts -function conversationStorageKey(endpoint) { - return `agent-chat:${endpoint}`; -} -function getStoredConversationId(endpoint, storage = localStorage) { - return storage.getItem(conversationStorageKey(endpoint)); -} -function setStoredConversationId(endpoint, conversationId, storage = localStorage) { - storage.setItem(conversationStorageKey(endpoint), conversationId); -} -function removeStoredConversationId(endpoint, storage = localStorage) { - storage.removeItem(conversationStorageKey(endpoint)); -} - // src/agent_manager/api/static/widget/react/shadcnAiElements.tsx var import_react8 = __toESM(require_react(), 1); @@ -53320,6 +53306,22 @@ function upsertTool(tools, next2) { // src/agent_manager/api/static/widget/react/useConversation.ts var import_react9 = __toESM(require_react(), 1); + +// src/agent_manager/api/static/widget/storage/conversationStorage.ts +function conversationStorageKey(endpoint) { + return `agent-chat:${endpoint}`; +} +function getStoredConversationId(endpoint, storage = localStorage) { + return storage.getItem(conversationStorageKey(endpoint)); +} +function setStoredConversationId(endpoint, conversationId, storage = localStorage) { + storage.setItem(conversationStorageKey(endpoint), conversationId); +} +function removeStoredConversationId(endpoint, storage = localStorage) { + storage.removeItem(conversationStorageKey(endpoint)); +} + +// src/agent_manager/api/static/widget/react/useConversation.ts var isUnusableConversation = (error) => error instanceof AgentChatHttpError && (error.status === 404 || error.status === 403); function useConversation(client, endpoint, onReplaced) { const startConversation = (0, import_react9.useCallback)(async () => { @@ -53493,6 +53495,7 @@ function AgentChatApp({ const [threads, setThreads] = (0, import_react10.useState)([]); const [nextCursor, setNextCursor] = (0, import_react10.useState)(null); const [loadingMoreThreads, setLoadingMoreThreads] = (0, import_react10.useState)(false); + const [threadsError, setThreadsError] = (0, import_react10.useState)(null); const [threadsOpen, setThreadsOpen] = (0, import_react10.useState)(false); const hasMoreThreads = nextCursor !== null; const isLoadingMoreRef = (0, import_react10.useRef)(false); @@ -53581,12 +53584,17 @@ function AgentChatApp({ const currentGen = threadsGenerationRef.current; setThreadsOpen(true); setLoadingMoreThreads(true); + setThreadsError(null); isLoadingMoreRef.current = true; try { const res = await conversation.listThreads(THREADS_PAGE_SIZE, null); if (threadsGenerationRef.current !== currentGen) return; setThreads(res.items); setNextCursor(res.next_cursor); + } catch (err) { + if (threadsGenerationRef.current !== currentGen) return; + const msg = err instanceof AgentChatHttpError ? err.message : GENERIC_ERROR; + setThreadsError(msg); } finally { if (threadsGenerationRef.current === currentGen) { setLoadingMoreThreads(false); @@ -53599,6 +53607,7 @@ function AgentChatApp({ const currentGen = threadsGenerationRef.current; isLoadingMoreRef.current = true; setLoadingMoreThreads(true); + setThreadsError(null); try { const res = await conversation.listThreads(THREADS_PAGE_SIZE, nextCursor); if (threadsGenerationRef.current !== currentGen) return; @@ -53608,6 +53617,10 @@ function AgentChatApp({ return [...prev, ...newItems]; }); setNextCursor(res.next_cursor); + } catch (err) { + if (threadsGenerationRef.current !== currentGen) return; + const msg = err instanceof AgentChatHttpError ? err.message : GENERIC_ERROR; + setThreadsError(msg); } finally { if (threadsGenerationRef.current === currentGen) { setLoadingMoreThreads(false); @@ -53961,11 +53974,13 @@ function AgentChatApp({ { open: threadsOpen, threads, - activeId: getStoredConversationId(config.endpoint), + activeId, loadingMore: loadingMoreThreads, + error: threadsError, hasMore: hasMoreThreads, onLoadMore: () => void loadMoreThreads(), - onSelect: openThread, + onRetry: () => threads.length === 0 ? void openThreads() : void loadMoreThreads(), + onSelect: (cid) => void openThread(cid), onNew: startNewThread, onClose: () => setThreadsOpen(false) } @@ -54215,15 +54230,17 @@ function ThreadDrawer({ threads, activeId, loadingMore, + error, hasMore, onLoadMore, + onRetry, onSelect, onNew, onClose }) { const handleScroll = (e) => { const { scrollTop, clientHeight, scrollHeight } = e.currentTarget; - if (scrollHeight - scrollTop - clientHeight < SCROLL_THRESHOLD_PX && hasMore && !loadingMore) { + if (scrollHeight - scrollTop - clientHeight < SCROLL_THRESHOLD_PX && hasMore && !loadingMore && !error) { onLoadMore(); } }; @@ -54248,8 +54265,13 @@ function ThreadDrawer({ }, thread.conversation_id )), - threads.length === 0 && !loadingMore ? /* @__PURE__ */ (0, import_jsx_runtime4.jsx)("p", { className: "thread-empty", children: "No conversations yet" }) : null, - loadingMore ? /* @__PURE__ */ (0, import_jsx_runtime4.jsx)("p", { className: "thread-empty", children: "Loading..." }) : null + threads.length === 0 && !loadingMore && !error ? /* @__PURE__ */ (0, import_jsx_runtime4.jsx)("p", { className: "thread-empty", children: "No conversations yet" }) : null, + error ? /* @__PURE__ */ (0, import_jsx_runtime4.jsxs)("div", { className: "thread-empty thread-error", children: [ + /* @__PURE__ */ (0, import_jsx_runtime4.jsx)("p", { children: error }), + /* @__PURE__ */ (0, import_jsx_runtime4.jsx)("button", { className: "thread-retry-btn", onClick: onRetry, type: "button", children: "Retry" }) + ] }) : null, + loadingMore ? /* @__PURE__ */ (0, import_jsx_runtime4.jsx)("p", { className: "thread-empty", children: "Loading..." }) : null, + hasMore && !loadingMore && !error ? /* @__PURE__ */ (0, import_jsx_runtime4.jsx)("button", { className: "thread-load-more-btn", onClick: onLoadMore, type: "button", children: "Load more" }) : null ] }) ] }); } diff --git a/src/agent_manager/api/static/widget/react/AgentChatApp.tsx b/src/agent_manager/api/static/widget/react/AgentChatApp.tsx index 5cca742d..b3fc2166 100644 --- a/src/agent_manager/api/static/widget/react/AgentChatApp.tsx +++ b/src/agent_manager/api/static/widget/react/AgentChatApp.tsx @@ -119,6 +119,7 @@ export function AgentChatApp({ const [threads, setThreads] = useState([]); const [nextCursor, setNextCursor] = useState(null); const [loadingMoreThreads, setLoadingMoreThreads] = useState(false); + const [threadsError, setThreadsError] = useState(null); const [threadsOpen, setThreadsOpen] = useState(false); const hasMoreThreads = nextCursor !== null; const isLoadingMoreRef = useRef(false); @@ -225,12 +226,17 @@ export function AgentChatApp({ const currentGen = threadsGenerationRef.current; setThreadsOpen(true); setLoadingMoreThreads(true); + setThreadsError(null); isLoadingMoreRef.current = true; try { const res = await conversation.listThreads(THREADS_PAGE_SIZE, null); if (threadsGenerationRef.current !== currentGen) return; setThreads(res.items); setNextCursor(res.next_cursor); + } catch (err) { + if (threadsGenerationRef.current !== currentGen) return; + const msg = err instanceof AgentChatHttpError ? err.message : GENERIC_ERROR; + setThreadsError(msg); } finally { if (threadsGenerationRef.current === currentGen) { setLoadingMoreThreads(false); @@ -244,6 +250,7 @@ export function AgentChatApp({ const currentGen = threadsGenerationRef.current; isLoadingMoreRef.current = true; setLoadingMoreThreads(true); + setThreadsError(null); try { const res = await conversation.listThreads(THREADS_PAGE_SIZE, nextCursor); if (threadsGenerationRef.current !== currentGen) return; @@ -256,6 +263,10 @@ export function AgentChatApp({ return [...prev, ...newItems]; }); setNextCursor(res.next_cursor); + } catch (err) { + if (threadsGenerationRef.current !== currentGen) return; + const msg = err instanceof AgentChatHttpError ? err.message : GENERIC_ERROR; + setThreadsError(msg); } finally { if (threadsGenerationRef.current === currentGen) { setLoadingMoreThreads(false); @@ -643,11 +654,13 @@ export function AgentChatApp({ void loadMoreThreads()} - onSelect={openThread} + onRetry={() => (threads.length === 0 ? void openThreads() : void loadMoreThreads())} + onSelect={(cid) => void openThread(cid)} onNew={startNewThread} onClose={() => setThreadsOpen(false)} /> @@ -979,8 +992,10 @@ function ThreadDrawer({ threads, activeId, loadingMore, + error, hasMore, onLoadMore, + onRetry, onSelect, onNew, onClose, @@ -989,15 +1004,17 @@ function ThreadDrawer({ threads: ThreadSummary[]; activeId: string | null; loadingMore: boolean; + error: string | null; hasMore: boolean; onLoadMore: () => void; + onRetry: () => void; onSelect: (conversationId: string) => void; onNew: () => void; onClose: () => void; }) { const handleScroll = (e: React.UIEvent) => { const { scrollTop, clientHeight, scrollHeight } = e.currentTarget; - if (scrollHeight - scrollTop - clientHeight < SCROLL_THRESHOLD_PX && hasMore && !loadingMore) { + if (scrollHeight - scrollTop - clientHeight < SCROLL_THRESHOLD_PX && hasMore && !loadingMore && !error) { onLoadMore(); } }; @@ -1026,8 +1043,21 @@ function ThreadDrawer({ {thread.title || "New chat"} ))} - {threads.length === 0 && !loadingMore ?

No conversations yet

: null} + {threads.length === 0 && !loadingMore && !error ?

No conversations yet

: null} + {error ? ( +
+

{error}

+ +
+ ) : null} {loadingMore ?

Loading...

: null} + {hasMore && !loadingMore && !error ? ( + + ) : null}
); diff --git a/src/agent_manager/domain/models.py b/src/agent_manager/domain/models.py index db89f837..490262c9 100644 --- a/src/agent_manager/domain/models.py +++ b/src/agent_manager/domain/models.py @@ -2,6 +2,7 @@ from __future__ import annotations +from collections.abc import Sequence from dataclasses import dataclass, field from datetime import datetime from enum import StrEnum @@ -76,7 +77,7 @@ def __post_init__(self) -> None: @dataclass(frozen=True) class Page(Generic[T]): - items: list[T] + items: Sequence[T] next_cursor: str | None = None diff --git a/tests/agent_manager/test_repository_contract.py b/tests/agent_manager/test_repository_contract.py index e789229d..0461f1ed 100644 --- a/tests/agent_manager/test_repository_contract.py +++ b/tests/agent_manager/test_repository_contract.py @@ -497,53 +497,91 @@ async def test_pagination_contract(repo: Repository) -> None: await repo.upsert_user(user_id) base_time = datetime(2026, 8, 20, 12, 0, 0, tzinfo=UTC) - await repo.create_session("s1", user_id=user_id) - await repo.create_session("s2", user_id=user_id) + s1 = await repo.create_session("s1", user_id=user_id) + s2 = await repo.create_session("s2", user_id=user_id) await repo.create_session("s3", user_id=user_id) + await repo.create_session("s4", user_id=user_id) + await repo.create_session("s5", user_id=user_id) + + # If repo is SqlRepository, set created_at explicitly so created_at ordering is deterministic + if isinstance(repo, SqlRepository): + async with repo._sessions() as session: + from agent_manager.infrastructure.persistence.tables import ConversationSessionRow + + r1 = await session.get(ConversationSessionRow, "s1") + if r1: + r1.created_at = base_time + r2 = await session.get(ConversationSessionRow, "s2") + if r2: + r2.created_at = base_time + timedelta(minutes=10) + r3 = await session.get(ConversationSessionRow, "s3") + if r3: + r3.created_at = base_time + r4 = await session.get(ConversationSessionRow, "s4") + if r4: + r4.created_at = base_time + r5 = await session.get(ConversationSessionRow, "s5") + if r5: + r5.created_at = base_time + await session.commit() + else: + from dataclasses import replace + + repo._sessions["s1"] = replace(s1, created_at=base_time) + repo._sessions["s2"] = replace(s2, created_at=base_time + timedelta(minutes=10)) + + # Append messages to s3, s4, s5 (s4 and s5 share the same last_message_at timestamp) await repo.append_message( ConversationMessage( - message_id="m1", - session_id="s1", + message_id="m3", + session_id="s3", role=Role.USER, - content="m1", - created_at=base_time, + content="m3", + created_at=base_time + timedelta(hours=1), ) ) await repo.append_message( ConversationMessage( - message_id="m2", - session_id="s2", + message_id="m4", + session_id="s4", role=Role.USER, - content="m2", - created_at=base_time + timedelta(hours=1), + content="m4", + created_at=base_time + timedelta(hours=2), ) ) await repo.append_message( ConversationMessage( - message_id="m3", - session_id="s3", + message_id="m5", + session_id="s5", role=Role.USER, - content="m3", + content="m5", created_at=base_time + timedelta(hours=2), ) ) + # Page 1: limit 2 -> ["s5", "s4"] page1 = await repo.list_sessions(user_id, page=PageRequest(limit=2)) - assert len(page1.items) == 2 + assert [s.session_id for s in page1.items] == ["s5", "s4"] assert page1.next_cursor is not None + # Page 2: limit 2 -> ["s3", "s2"] page2 = await repo.list_sessions(user_id, page=PageRequest(limit=2, cursor=page1.next_cursor)) - assert len(page2.items) == 1 - assert page2.next_cursor is None + assert [s.session_id for s in page2.items] == ["s3", "s2"] + assert page2.next_cursor is not None + + # Page 3: limit 2 -> ["s1"] + page3 = await repo.list_sessions(user_id, page=PageRequest(limit=2, cursor=page2.next_cursor)) + assert [s.session_id for s in page3.items] == ["s1"] + assert page3.next_cursor is None - all_ids = [s.session_id for s in page1.items + page2.items] - assert all_ids == ["s3", "s2", "s1"] + all_ids = [s.session_id for s in page1.items + page2.items + page3.items] + assert all_ids == ["s5", "s4", "s3", "s2", "s1"] - # 2. Page boundary landing exactly on limit - page_exact = await repo.list_sessions(user_id, page=PageRequest(limit=3)) - assert len(page_exact.items) == 3 + # Page boundary landing exactly on limit + page_exact = await repo.list_sessions(user_id, page=PageRequest(limit=5)) + assert len(page_exact.items) == 5 assert page_exact.next_cursor is None - # 3. Malformed cursor raises InvalidCursorError + # Malformed cursor raises InvalidCursorError with pytest.raises(InvalidCursorError): await repo.list_sessions(user_id, page=PageRequest(cursor="invalid_garbage_token")) From a8f83522d79688268abb6de4b753503ed0c354db Mon Sep 17 00:00:00 2001 From: rishu685 Date: Sun, 23 Aug 2026 23:27:26 +0530 Subject: [PATCH 09/10] fix(test): resolve mypy typecheck errors in test_repository_contract.py --- tests/agent_manager/test_repository_contract.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/agent_manager/test_repository_contract.py b/tests/agent_manager/test_repository_contract.py index 0461f1ed..011a6139 100644 --- a/tests/agent_manager/test_repository_contract.py +++ b/tests/agent_manager/test_repository_contract.py @@ -524,7 +524,7 @@ async def test_pagination_contract(repo: Repository) -> None: if r5: r5.created_at = base_time await session.commit() - else: + elif isinstance(repo, MemoryRepository): from dataclasses import replace repo._sessions["s1"] = replace(s1, created_at=base_time) @@ -574,7 +574,7 @@ async def test_pagination_contract(repo: Repository) -> None: assert [s.session_id for s in page3.items] == ["s1"] assert page3.next_cursor is None - all_ids = [s.session_id for s in page1.items + page2.items + page3.items] + all_ids = [s.session_id for s in [*page1.items, *page2.items, *page3.items]] assert all_ids == ["s5", "s4", "s3", "s2", "s1"] # Page boundary landing exactly on limit From df2880072a58e4f1f9c02d2899f352e7ba184d83 Mon Sep 17 00:00:00 2001 From: rishu685 Date: Sun, 23 Aug 2026 23:29:02 +0530 Subject: [PATCH 10/10] docs(adr): rename ADR 0003 to 0003-cursor-pagination-and-list-response-envelope.md --- ....md => 0003-cursor-pagination-and-list-response-envelope.md} | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) rename docs/adr/{0003-chat-history-pagination-and-keyset-cursor.md => 0003-cursor-pagination-and-list-response-envelope.md} (97%) diff --git a/docs/adr/0003-chat-history-pagination-and-keyset-cursor.md b/docs/adr/0003-cursor-pagination-and-list-response-envelope.md similarity index 97% rename from docs/adr/0003-chat-history-pagination-and-keyset-cursor.md rename to docs/adr/0003-cursor-pagination-and-list-response-envelope.md index d96058dc..6386cd31 100644 --- a/docs/adr/0003-chat-history-pagination-and-keyset-cursor.md +++ b/docs/adr/0003-cursor-pagination-and-list-response-envelope.md @@ -1,4 +1,4 @@ -# ADR 0003 — Chat history keyset pagination and cursor format +# ADR 0003 — Cursor pagination and list response envelope - **Status:** Accepted - **Date:** 2026-08-23