fix(pagination): fix reverse paging and reject stale-sort cursors

packages/tracecat-ee/tracecat_ee/inbox/providers/agent_runs.py

@@ -23,7 +23,13 @@
 from tracecat.inbox.schemas import InboxItemRead, UserSummary, WorkflowSummary
 from tracecat.inbox.types import InboxGroup, InboxItemStatus, InboxItemType
 from tracecat.logger import logger
-from tracecat.pagination import BaseCursorPaginator, CursorPaginatedResponse
+from tracecat.pagination import (
+    BaseCursorPaginator,
+    CursorPaginatedResponse,
+    build_cursor_page,
+    take_cursor_page,
+    validate_cursor_sort_column,
+)
 from tracecat_ee.agent.types import AgentWorkflowID
 
 # The error signal is fully persisted (AgentSession.last_error), so Temporal is
@@ -291,8 +297,10 @@ async def list_items(
             updated_after=updated_after,
         )
 
-        # Determine sort column and direction
-        sort_col = order_by or "created_at"
+        # Determine sort column and direction. Normalize to the two columns the
+        # query can actually order by, so an unrecognized `order_by` cannot mint
+        # a cursor labelled with a column the scan never sorted on.
+        sort_col = "updated_at" if order_by == "updated_at" else "created_at"
         sort_desc = sort != "asc"
         # Scan in the direction that walks toward the rows adjacent to the
         # cursor. Reverse pagination flips the scan so the LIMIT keeps the rows
@@ -312,7 +320,12 @@ async def list_items(
         if cursor:
             try:
                 cursor_data = self.decode_cursor(cursor)
-                cursor_value = cursor_data.sort_value
+                # A cursor from a different sort cannot filter this query:
+                # applying a created_at anchor to updated_at skips or duplicates
+                # rows. Reject it instead of silently mis-filtering.
+                cursor_value = validate_cursor_sort_column(
+                    cursor_data, sort_column=sort_col, expected_type=datetime
+                )
                 cursor_id = uuid.UUID(cursor_data.id)
 
                 # Select the correct column based on sort_col
@@ -379,69 +392,39 @@ async def list_items(
         stmt = base_stmt.order_by(order_clause, id_order).limit(limit + 1)
 
         result = await self.session.execute(stmt)
-        sessions = list(result.scalars().all())
-
-        # Check if there are more items
-        has_more = len(sessions) > limit
-        if has_more:
-            sessions = sessions[:limit]
+        scanned_sessions, has_more = take_cursor_page(
+            list(result.scalars().all()), limit=limit
+        )
 
-        # The scan ran in scan_desc order; reverse back into display order so
-        # the page reads in the requested sort regardless of pagination
-        # direction.
-        if reverse:
-            sessions.reverse()
+        # The scan ran in scan_desc order; build_cursor_page reverses the page
+        # back into display order so it reads in the requested sort regardless
+        # of pagination direction, and swaps the cursors and flags accordingly.
+        page = build_cursor_page(
+            scanned_sessions,
+            cursor=cursor,
+            reverse=reverse,
+            has_more=has_more,
+            encode_cursor=lambda session: self.encode_cursor(
+                id=session.id,
+                sort_column=sort_col,
+                # Use the correct column value based on sort_col
+                sort_value=(
+                    session.updated_at
+                    if sort_col == "updated_at"
+                    else session.created_at
+                ),
+            ),
+        )
 
         # Enrich sessions
-        items = await self._enrich_sessions(sessions)
-
-        # Generate cursors
-        next_cursor = None
-        prev_cursor = None
-
-        if items:
-            if has_more:
-                last_item = items[-1]
-                # Get the session for this item to access the sort column value
-                last_session = next(
-                    (s for s in sessions if s.id == last_item.source_id), None
-                )
-                if last_session:
-                    # Use the correct column value based on sort_col
-                    sort_value = (
-                        last_session.updated_at
-                        if sort_col == "updated_at"
-                        else last_session.created_at
-                    )
-                    next_cursor = self.encode_cursor(
-                        id=last_item.id,
-                        sort_column=sort_col,
-                        sort_value=sort_value,
-                    )
-            if cursor:
-                first_item = items[0]
-                first_session = next(
-                    (s for s in sessions if s.id == first_item.source_id), None
-                )
-                if first_session:
-                    # Use the correct column value based on sort_col
-                    sort_value = (
-                        first_session.updated_at
-                        if sort_col == "updated_at"
-                        else first_session.created_at
-                    )
-                    prev_cursor = self.encode_cursor(
-                        id=first_item.id,
-                        sort_column=sort_col,
-                        sort_value=sort_value,
-                    )
+        items = await self._enrich_sessions(page.items)
 
         return CursorPaginatedResponse(
             items=items,
-            next_cursor=next_cursor,
-            prev_cursor=prev_cursor,
-            has_more=has_more,
-            has_previous=cursor is not None,
+            next_cursor=page.next_cursor,
+            prev_cursor=page.prev_cursor,
+            has_more=page.has_more,
+            has_previous=page.has_previous,
             total_estimate=None,
         )
 
@@ -616,7 +599,13 @@ async def _list_items_grouped(
         if cursor:
             try:
                 cursor_data = self.decode_cursor(cursor)
-                last_key = (cursor_data.sort_value, uuid.UUID(cursor_data.id))
+                # Same cross-sort guard as the ungrouped path: a created_at
+                # anchor applied to an updated_at scan resumes at the wrong
+                # position and skips or repeats sessions.
+                scan_value = validate_cursor_sort_column(
+                    cursor_data, sort_column=sort_col, expected_type=datetime
+                )
+                last_key = (scan_value, uuid.UUID(cursor_data.id))
             except (ValueError, KeyError) as e:
                 # Surface malformed cursors as a client error so the router can
                 # return 400 instead of silently restarting the scan.

tests/unit/api/test_api_cases.py

@@ -36,7 +36,7 @@
     TracecatConflictError,
     TracecatValidationError,
 )
-from tracecat.pagination import CursorPaginatedResponse
+from tracecat.pagination import CursorPaginatedResponse, InvalidCursorError
 
 
 @pytest.fixture
@@ -263,6 +263,33 @@ async def test_list_cases_validates_field_ids_even_when_page_is_empty(
     )
 
 
+@pytest.mark.anyio
+async def test_list_cases_returns_400_for_a_stale_sort_cursor(
+    client: TestClient,
+    test_admin_role: Role,
+) -> None:
+    """A cursor from another sort is a client error, not a silent first page."""
+    with patch.object(cases_router, "CasesService") as MockService:
+        mock_svc = AsyncMock()
+        mock_svc.list_cases.side_effect = InvalidCursorError(
+            "Cursor was created for sort column 'created_at', "
+            "but this request sorts by 'priority'."
+        )
+        MockService.return_value = mock_svc
+
+        response = client.get(
+            "/cases",
+            params={
+                "workspace_id": str(test_admin_role.workspace_id),
+                "cursor": "stale-cursor",
+                "order_by": "priority",
+            },
+        )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert "sorts by 'priority'" in response.json()["detail"]
+
+
 @pytest.mark.anyio
 async def test_list_case_events_includes_comment_activity(
     client: TestClient,

tests/unit/test_cases_service.py

@@ -53,7 +53,7 @@
     TracecatConflictError,
     TracecatNotFoundError,
 )
-from tracecat.pagination import CursorPaginationParams
+from tracecat.pagination import CursorPaginationParams, InvalidCursorError
 from tracecat.tables.enums import SqlType
 
 pytestmark = pytest.mark.usefixtures("db")
@@ -1746,6 +1746,180 @@ async def test_search_cases_aliases_list_cases(
 
         assert search_response.model_dump() == list_response.model_dump()
 
+    async def test_search_cases_reverse_pagination_returns_preceding_page(
+        self, cases_service: CasesService
+    ) -> None:
+        """Paging backward should land on the page immediately before the cursor."""
+        for i in range(6):
+            await cases_service.create_case(
+                CaseCreate(
+                    summary=f"Case {i}",
+                    description=f"Case {i}",
+                    status=CaseStatus.NEW,
+                    priority=CasePriority.MEDIUM,
+                    severity=CaseSeverity.LOW,
+                )
+            )
+            await asyncio.sleep(0.01)
+
+        async def page(cursor: str | None, reverse: bool = False):
+            return await cases_service.search_cases(
+                params=CursorPaginationParams(limit=2, cursor=cursor, reverse=reverse)
+            )
+
+        page_1 = await page(None)
+        page_2 = await page(page_1.next_cursor)
+        page_3 = await page(page_2.next_cursor)
+
+        # Sanity check the forward walk before asserting on the backward one.
+        forward_ids = [
+            item.id for item in (*page_1.items, *page_2.items, *page_3.items)
+        ]
+        assert len(set(forward_ids)) == 6
+
+        # Stepping back from page 3 must return page 2, not the first page.
+        assert page_3.prev_cursor is not None
+        back_to_2 = await page(page_3.prev_cursor, reverse=True)
+        assert [item.id for item in back_to_2.items] == [
+            item.id for item in page_2.items
+        ]
+        assert back_to_2.has_more is True
+        assert back_to_2.has_previous is True
+        assert back_to_2.prev_cursor is not None
+
+        # Reverse responses must keep advertising a usable prev_cursor.
+        back_to_1 = await page(back_to_2.prev_cursor, reverse=True)
+        assert [item.id for item in back_to_1.items] == [
+            item.id for item in page_1.items
+        ]
+        assert back_to_1.has_previous is False
+        assert back_to_1.prev_cursor is None
+        assert back_to_1.has_more is True
+
+        # And the reverse page's next_cursor must walk forward again.
+        forward_to_3 = await page(back_to_2.next_cursor)
+        assert [item.id for item in forward_to_3.items] == [
+            item.id for item in page_3.items
+        ]
+
+    async def test_search_cases_reverse_pagination_ascending_sort(
+        self, cases_service: CasesService
+    ) -> None:
+        """Backward paging should also invert an ascending sort correctly."""
+        for i in range(6):
+            await cases_service.create_case(
+                CaseCreate(
+                    summary=f"Ascending {i}",
+                    description=f"Ascending {i}",
+                    status=CaseStatus.NEW,
+                    priority=CasePriority.MEDIUM,
+                    severity=CaseSeverity.LOW,
+                )
+            )
+            await asyncio.sleep(0.01)
+
+        async def page(cursor: str | None, reverse: bool = False):
+            return await cases_service.search_cases(
+                params=CursorPaginationParams(limit=2, cursor=cursor, reverse=reverse),
+                order_by="created_at",
+                sort="asc",
+            )
+
+        page_1 = await page(None)
+        page_2 = await page(page_1.next_cursor)
+        page_3 = await page(page_2.next_cursor)
+
+        back_to_2 = await page(page_3.prev_cursor, reverse=True)
+
+        assert [item.id for item in back_to_2.items] == [
+            item.id for item in page_2.items
+        ]
+        assert back_to_2.has_more is True
+        assert back_to_2.has_previous is True
+
+    async def test_search_cases_reverse_pagination_enum_sort(
+        self, cases_service: CasesService
+    ) -> None:
+        """Backward paging should work for enum sorts, which rank rather than compare."""
+        priorities = [
+            CasePriority.CRITICAL,
+            CasePriority.HIGH,
+            CasePriority.MEDIUM,
+            CasePriority.LOW,
+        ]
+        for i, priority in enumerate(priorities):
+            await cases_service.create_case(
+                CaseCreate(
+                    summary=f"Priority {i}",
+                    description=f"Priority {i}",
+                    status=CaseStatus.NEW,
+                    priority=priority,
+                    severity=CaseSeverity.LOW,
+                )
+            )
+            await asyncio.sleep(0.01)
+
+        async def page(cursor: str | None, reverse: bool = False):
+            return await cases_service.search_cases(
+                params=CursorPaginationParams(limit=1, cursor=cursor, reverse=reverse),
+                order_by="priority",
+                sort="desc",
+            )
+
+        page_1 = await page(None)
+        page_2 = await page(page_1.next_cursor)
+        page_3 = await page(page_2.next_cursor)
+
+        back_to_2 = await page(page_3.prev_cursor, reverse=True)
+
+        assert [item.id for item in back_to_2.items] == [
+            item.id for item in page_2.items
+        ]
+        assert back_to_2.items[0].priority == page_2.items[0].priority
+        assert back_to_2.has_more is True
+        assert back_to_2.has_previous is True
+
+    async def test_search_cases_rejects_a_cursor_from_a_different_sort(
+        self, cases_service: CasesService
+    ) -> None:
+        """A stale-sort cursor must fail loudly instead of rewinding to page 1."""
+        for i in range(3):
+            await cases_service.create_case(
+                CaseCreate(
+                    summary=f"Stale cursor {i}",
+                    description=f"Stale cursor {i}",
+                    status=CaseStatus.NEW,
+                    priority=CasePriority.MEDIUM,
+                    severity=CaseSeverity.LOW,
+                )
+            )
+            await asyncio.sleep(0.01)
+
+        page_1 = await cases_service.search_cases(
+            params=CursorPaginationParams(limit=1, cursor=None, reverse=False),
+            order_by="created_at",
+        )
+        assert page_1.next_cursor is not None
+
+        # Same cursor, different sort column: the cursor's sort value cannot be
+        # compared against the new ORDER BY.
+        with pytest.raises(InvalidCursorError):
+            await cases_service.search_cases(
+                params=CursorPaginationParams(
+                    limit=1, cursor=page_1.next_cursor, reverse=False
+                ),
+                order_by="priority",
+            )
+
+        # The cursor still works for the sort it was issued under.
+        page_2 = await cases_service.search_cases(
+            params=CursorPaginationParams(
+                limit=1, cursor=page_1.next_cursor, reverse=False
+            ),
+            order_by="created_at",
+        )
+        assert page_2.items[0].id != page_1.items[0].id
+
     async def test_search_cases_gates_duration_selectinload(
         self, cases_service: CasesService
     ) -> None: