diff --git a/packages/tracecat-ee/tracecat_ee/inbox/providers/agent_runs.py b/packages/tracecat-ee/tracecat_ee/inbox/providers/agent_runs.py index 8eaa14a824..59ae936b60 100644 --- a/packages/tracecat-ee/tracecat_ee/inbox/providers/agent_runs.py +++ b/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. diff --git a/tests/unit/api/test_api_cases.py b/tests/unit/api/test_api_cases.py index cc1089b908..b752b4c4e7 100644 --- a/tests/unit/api/test_api_cases.py +++ b/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, diff --git a/tests/unit/test_cases_service.py b/tests/unit/test_cases_service.py index 25475d67bd..299c3d0293 100644 --- a/tests/unit/test_cases_service.py +++ b/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: diff --git a/tests/unit/test_cursor_page.py b/tests/unit/test_cursor_page.py new file mode 100644 index 0000000000..b758db9fab --- /dev/null +++ b/tests/unit/test_cursor_page.py @@ -0,0 +1,408 @@ +"""Tests for the shared keyset pagination page builder. + +These exercise the contract that every cursor-paginated service relies on: +a reverse scan runs in the inverted sort order, and the page it produces must +be handed back to the client in forward (display) semantics. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime + +import pytest +import sqlalchemy as sa + +from tracecat.pagination import ( + BaseCursorPaginator, + CursorData, + InvalidCursorError, + build_cursor_page, + keyset_filter, + take_cursor_page, + validate_cursor_sort_column, +) + + +@dataclass(frozen=True) +class Row: + """A row in a list sorted newest-first by ``rank``.""" + + rank: int + + @property + def cursor(self) -> str: + return f"cursor-{self.rank}" + + +# Display order is descending by rank: rank 0 is the first row of page 1. +ROWS = [Row(rank=i) for i in range(100)] + + +def scan(*, limit: int, cursor: str | None, reverse: bool) -> list[Row]: + """Mimic a keyset scan: filter past the cursor, order, over-fetch by one. + + Forward scans walk down the display order; reverse scans walk back up it in + the inverted order, which is what puts the rows adjacent to the cursor + inside the LIMIT. + """ + anchor = None if cursor is None else int(cursor.removeprefix("cursor-")) + if reverse: + rows = [row for row in ROWS if anchor is not None and row.rank < anchor] + rows.sort(key=lambda row: row.rank, reverse=True) + else: + rows = [row for row in ROWS if anchor is None or row.rank > anchor] + rows.sort(key=lambda row: row.rank) + return rows[: limit + 1] + + +def page(*, limit: int, cursor: str | None = None, reverse: bool = False): + rows, has_more = take_cursor_page( + scan(limit=limit, cursor=cursor, reverse=reverse), limit=limit + ) + return build_cursor_page( + rows, + cursor=cursor, + reverse=reverse, + has_more=has_more, + encode_cursor=lambda row: row.cursor, + ) + + +def ranks(rows: list[Row]) -> list[int]: + return [row.rank for row in rows] + + +def test_take_cursor_page_trims_the_extra_row() -> None: + rows, has_more = take_cursor_page(ROWS[:11], limit=10) + + assert ranks(rows) == list(range(10)) + assert has_more is True + + +def test_take_cursor_page_reports_no_more_when_short() -> None: + rows, has_more = take_cursor_page(ROWS[:3], limit=10) + + assert ranks(rows) == [0, 1, 2] + assert has_more is False + + +def test_first_page_has_no_previous() -> None: + first = page(limit=20) + + assert ranks(first.items) == list(range(20)) + assert first.has_more is True + assert first.has_previous is False + assert first.prev_cursor is None + assert first.next_cursor == "cursor-19" + + +def test_forward_page_exposes_both_cursors() -> None: + second = page(limit=20, cursor="cursor-19") + + assert ranks(second.items) == list(range(20, 40)) + assert second.has_more is True + assert second.has_previous is True + assert second.next_cursor == "cursor-39" + assert second.prev_cursor == "cursor-20" + + +def test_reverse_page_returns_the_preceding_page_in_display_order() -> None: + # Regression: the reverse branch used to return page 1 from any page, since + # it selected everything before the cursor and then kept the top `limit`. + third = page(limit=20, cursor="cursor-39") + assert ranks(third.items) == list(range(40, 60)) + + back = page(limit=20, cursor=third.prev_cursor, reverse=True) + + assert ranks(back.items) == list(range(20, 40)) + assert back.has_more is True + assert back.has_previous is True + + +def test_reverse_page_cursors_round_trip() -> None: + third = page(limit=20, cursor="cursor-39") + back = page(limit=20, cursor=third.prev_cursor, reverse=True) + + # Stepping back again must reach page 1, not repeat page 2. + assert back.prev_cursor == "cursor-20" + further_back = page(limit=20, cursor=back.prev_cursor, reverse=True) + assert ranks(further_back.items) == list(range(20)) + + # And stepping forward from the reverse page must return page 3. + assert back.next_cursor == "cursor-39" + forward = page(limit=20, cursor=back.next_cursor) + assert ranks(forward.items) == ranks(third.items) + + +def test_reverse_page_onto_the_first_page_reports_no_previous() -> None: + second = page(limit=20, cursor="cursor-19") + back = page(limit=20, cursor=second.prev_cursor, reverse=True) + + assert ranks(back.items) == list(range(20)) + # Nothing precedes page 1, but page 2 is still ahead of it. + assert back.has_previous is False + assert back.prev_cursor is None + assert back.has_more is True + assert back.next_cursor == "cursor-19" + + +def test_empty_page_has_no_cursors() -> None: + empty = build_cursor_page( + [], + cursor="cursor-99", + reverse=True, + has_more=False, + encode_cursor=lambda row: "unreachable", + ) + + assert empty.items == [] + assert empty.next_cursor is None + assert empty.prev_cursor is None + assert empty.has_more is False + assert empty.has_previous is False + + +def test_validate_cursor_sort_column_returns_the_sort_value() -> None: + cursor = CursorData(id="abc", sort_column="created_at", sort_value="2026-01-01") + + assert validate_cursor_sort_column(cursor, sort_column="created_at") == ( + datetime(2026, 1, 1) + ) + + +def test_validate_cursor_sort_column_rejects_a_different_sort() -> None: + """A cursor from another sort must 400, not silently rewind to page 1.""" + cursor = CursorData(id="abc", sort_column="created_at", sort_value="2026-01-01") + + with pytest.raises(InvalidCursorError, match="sorts by 'priority'"): + validate_cursor_sort_column(cursor, sort_column="priority") + + +def test_validate_cursor_sort_column_rejects_a_sortless_cursor() -> None: + """Cursors predating sort-aware pagination carry no sort column.""" + cursor = CursorData(id="abc") + + with pytest.raises(InvalidCursorError): + validate_cursor_sort_column(cursor, sort_column="created_at") + + +def test_validate_cursor_sort_column_rejects_a_missing_sort_value() -> None: + cursor = CursorData(id="abc", sort_column="created_at") + + with pytest.raises(InvalidCursorError, match="missing a sort value"): + validate_cursor_sort_column(cursor, sort_column="created_at") + + +def test_validate_cursor_sort_column_rejects_a_wrongly_typed_sort_value() -> None: + """Enum and task-count sorts order by a rank, so their cursors carry ints.""" + cursor = CursorData(id="abc", sort_column="priority", sort_value="high") + + with pytest.raises(InvalidCursorError, match="wrong type"): + validate_cursor_sort_column(cursor, sort_column="priority", expected_type=int) + + +def test_validate_cursor_sort_column_accepts_an_explicit_null_when_allowed() -> None: + """A nullable sort column can legitimately anchor a page on a NULL value.""" + cursor = BaseCursorPaginator.decode_cursor( + BaseCursorPaginator.encode_cursor("abc", sort_column="score", sort_value=None) + ) + + assert cursor.has_sort_value is True + assert ( + validate_cursor_sort_column(cursor, sort_column="score", allow_null=True) + is None + ) + + +def test_validate_cursor_sort_column_rejects_a_null_by_default() -> None: + """Sorts on NOT NULL columns keep treating a NULL anchor as unusable.""" + cursor = BaseCursorPaginator.decode_cursor( + BaseCursorPaginator.encode_cursor( + "abc", sort_column="created_at", sort_value=None + ) + ) + + with pytest.raises(InvalidCursorError, match="missing a sort value"): + validate_cursor_sort_column(cursor, sort_column="created_at") + + +def test_validate_cursor_sort_column_rejects_an_omitted_sort_value_when_null_allowed() -> ( + None +): + """An absent field is a legacy cursor, not a row whose sort value is NULL.""" + cursor = CursorData.model_validate({"id": "abc", "sort_column": "score"}) + + assert cursor.has_sort_value is False + with pytest.raises(InvalidCursorError, match="missing a sort value"): + validate_cursor_sort_column(cursor, sort_column="score", allow_null=True) + + +def test_invalid_cursor_error_is_a_value_error() -> None: + """Routers map ValueError to 400; the subclass keeps that mapping.""" + assert issubclass(InvalidCursorError, ValueError) + + +# Keyset filtering over a nullable sort column, exercised on in-memory SQLite. +# NULL placement is pinned with explicit NULLS FIRST/LAST, so the ordering +# matches PostgreSQL's defaults and the predicate is checked against real rows +# rather than a rendered SQL string. +_KEYSET_TABLE = sa.Table( + "keyset_rows", + sa.MetaData(), + sa.Column("id", sa.Integer, primary_key=True), + sa.Column("score", sa.Integer, nullable=True), +) +# Ids paired with a nullable sort value, deliberately interleaving NULLs with +# ties so both the tie-breaker and the NULL/non-NULL transition are crossed. +_KEYSET_ROWS = [ + (1, None), + (2, 10), + (3, None), + (4, 5), + (5, 10), + (6, None), + (7, 1), + (8, 5), +] +# (score ASC NULLS LAST, id ASC). The descending display order is its reverse. +_ASCENDING_IDS = [7, 4, 8, 2, 5, 1, 3, 6] + + +@pytest.fixture +def keyset_conn(): + engine = sa.create_engine("sqlite://") + with engine.begin() as conn: + _KEYSET_TABLE.create(conn) + conn.execute( + _KEYSET_TABLE.insert(), + [{"id": id_, "score": score} for id_, score in _KEYSET_ROWS], + ) + with engine.connect() as conn: + yield conn + engine.dispose() + + +def _keyset_scan(conn, *, cursor: str | None, ascending: bool, limit: int): + """Run one over-fetched keyset scan, mirroring what the services build.""" + stmt = sa.select(_KEYSET_TABLE.c.id, _KEYSET_TABLE.c.score) + if cursor is not None: + cursor_data = BaseCursorPaginator.decode_cursor(cursor) + stmt = stmt.where( + keyset_filter( + _KEYSET_TABLE.c.score, + _KEYSET_TABLE.c.id, + sort_value=validate_cursor_sort_column( + cursor_data, sort_column="score", allow_null=True + ), + id_value=int(cursor_data.id), + ascending=ascending, + ) + ) + sort_order = ( + _KEYSET_TABLE.c.score.asc().nulls_last() + if ascending + else _KEYSET_TABLE.c.score.desc().nulls_first() + ) + id_order = _KEYSET_TABLE.c.id.asc() if ascending else _KEYSET_TABLE.c.id.desc() + return list(conn.execute(stmt.order_by(sort_order, id_order).limit(limit + 1))) + + +def _keyset_page( + conn, *, cursor: str | None, ascending: bool, limit: int, reverse: bool = False +): + """One page in display order, following the cursor contract end to end.""" + scanned = _keyset_scan( + conn, cursor=cursor, ascending=ascending != reverse, limit=limit + ) + rows, has_more = take_cursor_page(scanned, limit=limit) + return build_cursor_page( + rows, + cursor=cursor, + reverse=reverse, + has_more=has_more, + encode_cursor=lambda row: BaseCursorPaginator.encode_cursor( + row.id, sort_column="score", sort_value=row.score + ), + ) + + +@pytest.mark.parametrize("ascending", [True, False]) +def test_keyset_filter_walks_every_row_across_the_null_boundary( + keyset_conn, ascending: bool +) -> None: + """Regression: a page anchored on a NULL sort value used to be unreachable. + + Nullable columns let a page boundary land inside the NULL block, so the + server issues a cursor whose sort value is NULL. Following it must resume + inside that block rather than reject the cursor or restart the scan. + """ + expected = _ASCENDING_IDS if ascending else list(reversed(_ASCENDING_IDS)) + + seen: list[int] = [] + cursor: str | None = None + while True: + page = _keyset_page(keyset_conn, cursor=cursor, ascending=ascending, limit=2) + seen.extend(row.id for row in page.items) + if not page.has_more: + break + cursor = page.next_cursor + + assert seen == expected + + +@pytest.mark.parametrize("ascending", [True, False]) +def test_keyset_filter_reverses_across_the_null_boundary( + keyset_conn, ascending: bool +) -> None: + """Paging back from a NULL-adjacent page must land on the page before it.""" + expected = _ASCENDING_IDS if ascending else list(reversed(_ASCENDING_IDS)) + + page_1 = _keyset_page(keyset_conn, cursor=None, ascending=ascending, limit=3) + page_2 = _keyset_page( + keyset_conn, cursor=page_1.next_cursor, ascending=ascending, limit=3 + ) + assert [row.id for row in page_2.items] == expected[3:6] + + back = _keyset_page( + keyset_conn, + cursor=page_2.prev_cursor, + ascending=ascending, + limit=3, + reverse=True, + ) + + assert [row.id for row in back.items] == expected[:3] + + +@pytest.mark.parametrize("ascending", [True, False]) +def test_keyset_filter_omits_null_branches_for_not_null_columns( + ascending: bool, +) -> None: + """NOT NULL sorts keep the tight two-branch predicate they had before.""" + predicate = keyset_filter( + _KEYSET_TABLE.c.score, + _KEYSET_TABLE.c.id, + sort_value=5, + id_value=4, + ascending=ascending, + nullable=False, + ) + + assert "NULL" not in str(predicate.compile(compile_kwargs={"literal_binds": True})) + + +@pytest.mark.parametrize("reverse", [False, True]) +def test_build_cursor_page_does_not_mutate_the_input(reverse: bool) -> None: + rows = list(ROWS[:5]) + original = list(rows) + + build_cursor_page( + rows, + cursor="cursor-9", + reverse=reverse, + has_more=True, + encode_cursor=lambda row: row.cursor, + ) + + assert rows == original diff --git a/tests/unit/test_inbox_agent_runs_cursor.py b/tests/unit/test_inbox_agent_runs_cursor.py new file mode 100644 index 0000000000..e4d6e7b225 --- /dev/null +++ b/tests/unit/test_inbox_agent_runs_cursor.py @@ -0,0 +1,83 @@ +"""Cursor-contract tests for the agent runs inbox provider. + +The provider paginates on either ``created_at`` or ``updated_at``. Applying a +cursor minted for one column to a scan ordered by the other resumes at the +wrong keyset position, which silently skips or repeats sessions, so both the +ungrouped and grouped decode paths must reject a cross-sort cursor. + +These exercise cursor validation only, which happens before any query runs, so +the session is a stub. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from unittest.mock import MagicMock +from uuid import uuid4 + +import pytest +from tracecat_ee.inbox.providers.agent_runs import AgentRunsInboxProvider + +from tracecat.auth.types import Role +from tracecat.inbox.types import InboxGroup +from tracecat.pagination import BaseCursorPaginator + +pytestmark = pytest.mark.anyio + + +@pytest.fixture +def provider() -> AgentRunsInboxProvider: + role = Role( + type="user", workspace_id=uuid4(), user_id=uuid4(), service_id="tracecat-api" + ) + return AgentRunsInboxProvider(session=MagicMock(), role=role) + + +def _cursor(sort_column: str) -> str: + return BaseCursorPaginator.encode_cursor( + uuid4(), + sort_column=sort_column, + sort_value=datetime(2026, 1, 1, tzinfo=UTC), + ) + + +async def test_list_items_rejects_a_cursor_from_a_different_sort( + provider: AgentRunsInboxProvider, +) -> None: + with pytest.raises(ValueError, match="sorts by 'updated_at'"): + await provider.list_items(cursor=_cursor("created_at"), order_by="updated_at") + + +async def test_grouped_list_items_rejects_a_cursor_from_a_different_sort( + provider: AgentRunsInboxProvider, +) -> None: + with pytest.raises(ValueError, match="sorts by 'updated_at'"): + await provider.list_items( + cursor=_cursor("created_at"), + order_by="updated_at", + group=InboxGroup.RUNNING, + ) + + +@pytest.mark.parametrize("group", [None, InboxGroup.RUNNING]) +async def test_list_items_rejects_a_sortless_cursor( + provider: AgentRunsInboxProvider, group: InboxGroup | None +) -> None: + """Cursors predating sort-aware pagination carry no sort column.""" + cursor = BaseCursorPaginator.encode_cursor(uuid4()) + + with pytest.raises(ValueError, match="Cursor was created for sort column None"): + await provider.list_items(cursor=cursor, order_by="created_at", group=group) + + +@pytest.mark.parametrize("group", [None, InboxGroup.RUNNING]) +async def test_list_items_rejects_a_non_datetime_sort_value( + provider: AgentRunsInboxProvider, group: InboxGroup | None +) -> None: + """Both columns are timestamps; anything else cannot filter the keyset.""" + cursor = BaseCursorPaginator.encode_cursor( + uuid4(), sort_column="created_at", sort_value=7 + ) + + with pytest.raises(ValueError, match="wrong type"): + await provider.list_items(cursor=cursor, order_by="created_at", group=group) diff --git a/tests/unit/test_tables_service.py b/tests/unit/test_tables_service.py index b7a90c871d..14cbf144c5 100644 --- a/tests/unit/test_tables_service.py +++ b/tests/unit/test_tables_service.py @@ -1,7 +1,7 @@ from collections.abc import Iterator from datetime import UTC, date, datetime, timedelta, timezone from decimal import Decimal -from typing import Any +from typing import Any, Literal from uuid import UUID, uuid4 import pytest @@ -17,7 +17,11 @@ from tracecat.db.models import Table, TableColumn, Workspace from tracecat.exceptions import TracecatAuthorizationError, TracecatNotFoundError from tracecat.logger import logger -from tracecat.pagination import CursorPaginationParams +from tracecat.pagination import ( + BaseCursorPaginator, + CursorPaginationParams, + InvalidCursorError, +) from tracecat.tables.common import ( ColumnHasDuplicateValuesError, handle_default_value, @@ -1437,6 +1441,132 @@ async def test_list_rows_reverse_pagination_flags( assert reverse_page.has_more is True assert reverse_page.has_previous is False + async def test_list_rows_reverse_pagination_returns_preceding_page( + self, tables_service: TablesService, table: Table + ) -> None: + """Paging backward should land on the page immediately before the cursor.""" + for i in range(6): + await tables_service.insert_row( + table, TableRowInsert(data={"name": f"Row{i}", "age": i}) + ) + + page_1 = await _list_rows_page(tables_service, table, limit=2) + page_2 = await _list_rows_page( + tables_service, table, limit=2, cursor=page_1.next_cursor + ) + page_3 = await _list_rows_page( + tables_service, table, limit=2, cursor=page_2.next_cursor + ) + + assert page_3.prev_cursor is not None + back_to_2 = await _list_rows_page( + tables_service, table, limit=2, cursor=page_3.prev_cursor, reverse=True + ) + + assert [row["id"] for row in back_to_2.items] == [ + row["id"] for row in page_2.items + ] + assert back_to_2.has_more is True + assert back_to_2.has_previous is True + + forward_to_3 = await _list_rows_page( + tables_service, table, limit=2, cursor=back_to_2.next_cursor + ) + assert [row["id"] for row in forward_to_3.items] == [ + row["id"] for row in page_3.items + ] + + async def test_list_rows_rejects_a_cursor_from_a_different_sort( + self, tables_service: TablesService, table: Table + ) -> None: + """A stale-sort cursor must fail loudly instead of rewinding to page 1.""" + for i in range(3): + await tables_service.insert_row( + table, TableRowInsert(data={"name": f"Stale{i}", "age": i}) + ) + + page_1 = await _list_rows_page(tables_service, table, limit=1) + assert page_1.next_cursor is not None + + with pytest.raises(InvalidCursorError): + await tables_service.list_rows( + table, + params=CursorPaginationParams( + limit=1, cursor=page_1.next_cursor, reverse=False + ), + order_by="age", + sort="asc", + ) + + @pytest.mark.parametrize("sort", ["asc", "desc"]) + async def test_list_rows_follows_a_null_sort_value_cursor( + self, + tables_service: TablesService, + table: Table, + sort: Literal["asc", "desc"], + ) -> None: + """A page boundary inside the NULL block must stay followable. + + Custom columns are nullable by default, so ``list_rows`` legitimately + encodes ``sort_value=None`` whenever a page ends on an unset value. + Rejecting that server-issued cursor made every row after the first page + unreachable. + """ + unset_ids = set() + for i in range(3): + row = await tables_service.insert_row( + table, TableRowInsert(data={"name": f"Unset{i}"}) + ) + unset_ids.add(str(row["id"])) + # age 0 also guards the NULL check against a falsy-value regression. + set_ids = set() + for i in range(2): + row = await tables_service.insert_row( + table, TableRowInsert(data={"name": f"Set{i}", "age": i}) + ) + set_ids.add(str(row["id"])) + + async def page(cursor: str | None, *, reverse: bool = False): + return await tables_service.list_rows( + table, + params=CursorPaginationParams(limit=2, cursor=cursor, reverse=reverse), + order_by="age", + sort=sort, + ) + + seen: list[str] = [] + pages: list[list[str]] = [] + null_anchored = False + cursor: str | None = None + while True: + result = await page(cursor) + ids = [str(row["id"]) for row in result.items] + pages.append(ids) + seen.extend(ids) + if not result.has_more: + break + assert result.next_cursor is not None + anchor = BaseCursorPaginator.decode_cursor(result.next_cursor) + assert anchor.sort_column == "age" + null_anchored |= anchor.has_sort_value and anchor.sort_value is None + cursor = result.next_cursor + assert len(pages) <= 5, "pagination did not advance" + + assert null_anchored, "expected a page boundary anchored on a NULL sort value" + assert seen == list(dict.fromkeys(seen)) + assert set(seen) == unset_ids | set_ids + + # PostgreSQL sorts NULLs last ascending and first descending, so the + # unset rows form a contiguous block at one end of the walk. + assert set(seen[-3:] if sort == "asc" else seen[:3]) == unset_ids + + # Paging back must land on the preceding page rather than restarting. + page_1 = await page(None) + page_2 = await page(page_1.next_cursor) + assert page_2.prev_cursor is not None + back = await page(page_2.prev_cursor, reverse=True) + assert [str(row["id"]) for row in back.items] == pages[0] + async def test_table_editor_list_rows_reverse_pagination_flags( self, tables_service: TablesService, table: Table ) -> None: diff --git a/tracecat/agent/preset/service.py b/tracecat/agent/preset/service.py index dc7f11e009..026416a44d 100644 --- a/tracecat/agent/preset/service.py +++ b/tracecat/agent/preset/service.py @@ -81,6 +81,8 @@ BaseCursorPaginator, CursorPaginatedResponse, CursorPaginationParams, + build_cursor_page, + take_cursor_page, ) from tracecat.registry.actions.service import RegistryActionsService from tracecat.secrets import secrets_manager @@ -1361,33 +1363,26 @@ async def list_versions( updated_at, ) in result.tuples().all() ] - has_more = len(versions) > params.limit - items = versions[: params.limit] - - next_cursor = None - if has_more and items: - last_version = items[-1] - next_cursor = paginator.encode_cursor( - last_version.id, - sort_column="version", - sort_value=last_version.version, - ) + scanned, has_more = take_cursor_page(versions, limit=params.limit) - prev_cursor = None - if params.cursor and items: - first_version = items[0] - prev_cursor = paginator.encode_cursor( - first_version.id, + page = build_cursor_page( + scanned, + cursor=params.cursor, + reverse=params.reverse, + has_more=has_more, + encode_cursor=lambda version: paginator.encode_cursor( + version.id, sort_column="version", - sort_value=first_version.version, - ) + sort_value=version.version, + ), + ) return CursorPaginatedResponse( - items=list(items), - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=params.cursor is not None, + items=page.items, + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, ) @requires_entitlement(Entitlement.AGENT_ADDONS) diff --git a/tracecat/agent/skill/service.py b/tracecat/agent/skill/service.py index b3152659fb..af3d75d687 100644 --- a/tracecat/agent/skill/service.py +++ b/tracecat/agent/skill/service.py @@ -71,6 +71,8 @@ BaseCursorPaginator, CursorPaginatedResponse, CursorPaginationParams, + build_cursor_page, + take_cursor_page, ) from tracecat.service import BaseWorkspaceService, requires_entitlement from tracecat.storage import blob @@ -1628,33 +1630,26 @@ async def list_skills( stmt = stmt.order_by(Skill.updated_at.desc(), Skill.id.desc()) stmt = stmt.limit(params.limit + 1) skills = (await self.session.execute(stmt)).scalars().all() - has_more = len(skills) > params.limit - items = skills[: params.limit] - - next_cursor = None - if has_more and items: - last = items[-1] - next_cursor = paginator.encode_cursor( - last.id, - sort_column="updated_at", - sort_value=last.updated_at, - ) + scanned, has_more = take_cursor_page(skills, limit=params.limit) - prev_cursor = None - if params.cursor and items: - first = items[0] - prev_cursor = paginator.encode_cursor( - first.id, + page = build_cursor_page( + scanned, + cursor=params.cursor, + reverse=params.reverse, + has_more=has_more, + encode_cursor=lambda skill: paginator.encode_cursor( + skill.id, sort_column="updated_at", - sort_value=first.updated_at, - ) + sort_value=skill.updated_at, + ), + ) return CursorPaginatedResponse( - items=[self._build_skill_read_minimal(skill) for skill in items], - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=params.cursor is not None, + items=[self._build_skill_read_minimal(skill) for skill in page.items], + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, ) @requires_entitlement(Entitlement.AGENT_ADDONS) @@ -2202,26 +2197,19 @@ async def list_versions( stmt = stmt.order_by(SkillVersion.version.desc(), SkillVersion.id.desc()) stmt = stmt.limit(params.limit + 1) versions = (await self.session.execute(stmt)).scalars().all() - has_more = len(versions) > params.limit - items = versions[: params.limit] - - next_cursor = None - if has_more and items: - last = items[-1] - next_cursor = paginator.encode_cursor( - last.id, - sort_column="version", - sort_value=last.version, - ) + scanned, has_more = take_cursor_page(versions, limit=params.limit) - prev_cursor = None - if params.cursor and items: - first = items[0] - prev_cursor = paginator.encode_cursor( - first.id, + page = build_cursor_page( + scanned, + cursor=params.cursor, + reverse=params.reverse, + has_more=has_more, + encode_cursor=lambda version: paginator.encode_cursor( + version.id, sort_column="version", - sort_value=first.version, - ) + sort_value=version.version, + ), + ) return CursorPaginatedResponse( items=[ @@ -2238,12 +2226,12 @@ async def list_versions( created_at=version.created_at, updated_at=version.updated_at, ) - for version in items + for version in page.items ], - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=params.cursor is not None, + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, ) @requires_entitlement(Entitlement.AGENT_ADDONS) diff --git a/tracecat/cases/rows/service.py b/tracecat/cases/rows/service.py index f7dc8c1c39..eea757c38e 100644 --- a/tracecat/cases/rows/service.py +++ b/tracecat/cases/rows/service.py @@ -24,7 +24,12 @@ from tracecat.cases.service import CaseEventsService from tracecat.db.models import Case, CaseTableRow, Table from tracecat.exceptions import TracecatNotFoundError -from tracecat.pagination import BaseCursorPaginator, CursorPaginatedResponse +from tracecat.pagination import ( + BaseCursorPaginator, + CursorPaginatedResponse, + build_cursor_page, + take_cursor_page, +) from tracecat.service import BaseWorkspaceService from tracecat.tables.service import TablesService @@ -67,7 +72,6 @@ async def list_rows( CaseTableRow.case_id == case_id, ) .options(selectinload(CaseTableRow.case)) - .order_by(CaseTableRow.created_at.desc(), CaseTableRow.id.desc()) ) if cursor: @@ -98,9 +102,6 @@ async def list_rows( ), ) ) - stmt = stmt.order_by( - CaseTableRow.created_at.asc(), CaseTableRow.id.asc() - ) else: stmt = stmt.where( or_( @@ -112,44 +113,42 @@ async def list_rows( ) ) + # Reverse pagination scans away from the cursor in ascending order so + # LIMIT keeps the rows nearest the cursor; build_cursor_page puts them + # back into display order. + if reverse: + stmt = stmt.order_by(CaseTableRow.created_at.asc(), CaseTableRow.id.asc()) + else: + stmt = stmt.order_by(CaseTableRow.created_at.desc(), CaseTableRow.id.desc()) + stmt = stmt.limit(limit + 1) result = await self.session.execute(stmt) links = result.scalars().all() - has_more = len(links) > limit - items = links[:limit] if has_more else links - has_previous = cursor is not None + scanned_links, has_more = take_cursor_page(links, limit=limit) - if reverse: - items = list(reversed(items)) - - hydrated = await self._hydrate_links(items, include_row_data=include_row_data) - - next_cursor = None - prev_cursor = None - if items and has_more: - next_cursor = paginator.encode_cursor( - items[-1].id, - sort_column="created_at", - sort_value=items[-1].created_at, - ) - if items and cursor: - prev_cursor = paginator.encode_cursor( - items[0].id, + page = build_cursor_page( + scanned_links, + cursor=cursor, + reverse=reverse, + has_more=has_more, + encode_cursor=lambda link: paginator.encode_cursor( + link.id, sort_column="created_at", - sort_value=items[0].created_at, - ) + sort_value=link.created_at, + ), + ) - if reverse: - next_cursor, prev_cursor = prev_cursor, next_cursor - has_more, has_previous = has_previous, has_more + hydrated = await self._hydrate_links( + page.items, include_row_data=include_row_data + ) return CursorPaginatedResponse( items=hydrated, - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=has_previous, + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, ) async def link_row( diff --git a/tracecat/cases/service.py b/tracecat/cases/service.py index 304eaf1f3b..93d753b736 100644 --- a/tracecat/cases/service.py +++ b/tracecat/cases/service.py @@ -127,6 +127,9 @@ BaseCursorPaginator, CursorPaginatedResponse, CursorPaginationParams, + build_cursor_page, + take_cursor_page, + validate_cursor_sort_column, ) from tracecat.service import BaseWorkspaceService, requires_entitlement from tracecat.tables.common import ( @@ -493,81 +496,84 @@ async def search_cases( cursor_data = paginator.decode_cursor(params.cursor) cursor_id = uuid.UUID(cursor_data.id) - # Check if cursor was created with the same sort column (for proper pagination) - cursor_sort_value = cursor_data.sort_value - cursor_has_sort_value = ( - cursor_data.sort_column == sort_column and cursor_sort_value is not None + # A cursor from a different sort cannot filter this query. Reject it + # instead of dropping the filter, which would silently return the + # first page while still reporting has_previous. + # Enum and task-count sorts order by a computed rank, so their + # cursors carry that rank rather than the column value. + sort_cursor_value = validate_cursor_sort_column( + cursor_data, + sort_column=sort_column, + expected_type=int + if sort_column == "tasks" or enum_sort_values is not None + else None, ) - if cursor_has_sort_value and sort_column == "tasks": - cursor_has_sort_value = isinstance(cursor_sort_value, int) - elif cursor_has_sort_value and enum_sort_values is not None: - cursor_has_sort_value = isinstance(cursor_sort_value, int) - - if cursor_has_sort_value: - sort_filter_col = sort_attr - sort_cursor_value = cursor_sort_value - - # Composite filtering: (sort_col, id) matches ORDER BY - # Use id as tie-breaker since it's always unique - if sort_direction == "asc": - if params.reverse: - # Going backward: get records before cursor in sort order - stmt = stmt.where( - or_( - sort_filter_col < sort_cursor_value, - and_( - sort_filter_col == sort_cursor_value, - Case.id < cursor_id, - ), - ) - ) - else: - # Going forward: get records after cursor in sort order - stmt = stmt.where( - or_( - sort_filter_col > sort_cursor_value, - and_( - sort_filter_col == sort_cursor_value, - Case.id > cursor_id, - ), - ) + sort_filter_col = sort_attr + + # Composite filtering: (sort_col, id) matches ORDER BY + # Use id as tie-breaker since it's always unique + if sort_direction == "asc": + if params.reverse: + # Going backward: get records before cursor in sort order + stmt = stmt.where( + or_( + sort_filter_col < sort_cursor_value, + and_( + sort_filter_col == sort_cursor_value, + Case.id < cursor_id, + ), ) + ) else: - # Descending order - if params.reverse: - # Going backward: get records after cursor in sort order - stmt = stmt.where( - or_( - sort_filter_col > sort_cursor_value, - and_( - sort_filter_col == sort_cursor_value, - Case.id > cursor_id, - ), - ) + # Going forward: get records after cursor in sort order + stmt = stmt.where( + or_( + sort_filter_col > sort_cursor_value, + and_( + sort_filter_col == sort_cursor_value, + Case.id > cursor_id, + ), ) - else: - # Going forward: get records before cursor in sort order - stmt = stmt.where( - or_( - sort_filter_col < sort_cursor_value, - and_( - sort_filter_col == sort_cursor_value, - Case.id < cursor_id, - ), - ) + ) + else: + # Descending order + if params.reverse: + # Going backward: get records after cursor in sort order + stmt = stmt.where( + or_( + sort_filter_col > sort_cursor_value, + and_( + sort_filter_col == sort_cursor_value, + Case.id > cursor_id, + ), + ) + ) + else: + # Going forward: get records before cursor in sort order + stmt = stmt.where( + or_( + sort_filter_col < sort_cursor_value, + and_( + sort_filter_col == sort_cursor_value, + Case.id < cursor_id, + ), ) + ) # Apply sorting: (sort_col, id) for stable pagination - # Use id as tie-breaker unless we're already sorting by id + # Reverse pagination scans away from the cursor in the inverted sort + # order, so the rows nearest the cursor are the ones kept by LIMIT. + # build_cursor_page puts them back into display order. + scan_ascending = (sort_direction == "asc") != params.reverse if sort_column == "id": # No tie-breaker needed when sorting by id (already unique) - if sort_direction == "asc": + if scan_ascending: stmt = stmt.order_by(sort_attr.asc()) else: stmt = stmt.order_by(sort_attr.desc()) else: # Add id as tie-breaker for non-unique columns - if sort_direction == "asc": + if scan_ascending: stmt = stmt.order_by(sort_attr.asc(), Case.id.asc()) else: stmt = stmt.order_by(sort_attr.desc(), Case.id.desc()) @@ -577,17 +583,10 @@ async def search_cases( result = await self.session.execute(stmt) all_cases = result.scalars().all() - # Check if there are more items - has_more = len(all_cases) > params.limit - cases = all_cases[: params.limit] if has_more else all_cases + scanned_cases, has_more = take_cursor_page(all_cases, limit=params.limit) # Fetch task counts for all cases in one query (needed for cursor generation if sorting by tasks) - task_counts = await self.get_task_counts([case.id for case in cases]) - - # Generate cursors with sort column info for proper pagination - next_cursor = None - prev_cursor = None - has_previous = params.cursor is not None + task_counts = await self.get_task_counts([case.id for case in scanned_cases]) def get_cursor_sort_value(case: Case) -> datetime | str | int | float | None: """Encode cursor sort values using the same semantics as ORDER BY.""" @@ -599,31 +598,19 @@ def get_cursor_sort_value(case: Case) -> datetime | str | int | float | None: ) return getattr(case, sort_column, None) - if has_more and cases: - last_case = cases[-1] - sort_value = get_cursor_sort_value(last_case) - next_cursor = paginator.encode_cursor( - last_case.id, + # Generate cursors with sort column info for proper pagination + page = build_cursor_page( + scanned_cases, + cursor=params.cursor, + reverse=params.reverse, + has_more=has_more, + encode_cursor=lambda case: paginator.encode_cursor( + case.id, sort_column=sort_column, - sort_value=sort_value, - ) - - if params.cursor and cases: - first_case = cases[0] - sort_value = get_cursor_sort_value(first_case) - # For reverse pagination, swap the cursor meaning - if params.reverse: - next_cursor = paginator.encode_cursor( - first_case.id, - sort_column=sort_column, - sort_value=sort_value, - ) - else: - prev_cursor = paginator.encode_cursor( - first_case.id, - sort_column=sort_column, - sort_value=sort_value, - ) + sort_value=get_cursor_sort_value(case), + ), + ) + cases = page.items # Convert to CaseReadMinimal objects with tags and dropdown values case_items = [] @@ -685,10 +672,10 @@ def get_cursor_sort_value(case: Case) -> datetime | str | int | float | None: return CursorPaginatedResponse( items=case_items, - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=has_previous, + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, total_estimate=total_estimate, ) diff --git a/tracecat/pagination.py b/tracecat/pagination.py index 6e5f8d3aa7..a4bc292740 100644 --- a/tracecat/pagination.py +++ b/tracecat/pagination.py @@ -2,8 +2,10 @@ import base64 import json +from collections.abc import Callable, Sequence +from dataclasses import dataclass from datetime import datetime -from typing import TypeVar +from typing import Any, Literal, TypeVar, overload from uuid import UUID import sqlalchemy as sa @@ -45,6 +47,94 @@ class CursorPaginatedResponse[T](BaseModel): ) +class InvalidCursorError(ValueError): + """A cursor that cannot be applied to the query it was submitted with. + + Raised when the cursor's sort column or sort value does not line up with + the current sort, which happens when a client changes the sort without + restarting pagination. Dropping the cursor filter instead would silently + return the first page while still reporting ``has_previous``. + + Subclasses ``ValueError`` so routers surface it as 400, the same way + ``decode_cursor`` already reports a structurally malformed cursor. + """ + + +@dataclass(slots=True) +class CursorPage[T]: + """A single page of keyset-paginated rows in display order.""" + + items: list[T] + next_cursor: str | None + prev_cursor: str | None + has_more: bool + has_previous: bool + + +def take_cursor_page[T](rows: Sequence[T], *, limit: int) -> tuple[list[T], bool]: + """Trim an over-fetched keyset scan to a page. + + Args: + rows: Rows returned by a scan that requested ``limit + 1`` rows. + limit: Page size requested by the caller. + + Returns: + The page rows (still in scan order) and whether the scan found more + rows beyond the page in the direction it was scanning. + """ + has_more = len(rows) > limit + return list(rows[:limit]), has_more + + +def build_cursor_page[T]( + rows: Sequence[T], + *, + cursor: str | None, + reverse: bool, + has_more: bool, + encode_cursor: Callable[[T], str], +) -> CursorPage[T]: + """Derive page items, cursors, and flags from a trimmed keyset scan. + + Reverse pagination scans away from the cursor in the inverted sort order, + so the caller must invert its ``ORDER BY`` (and the cursor predicate) when + ``reverse`` is set. This function then puts the rows back into display + order and swaps the cursors and flags, since the scan's "more rows ahead" + is the page's "more rows behind". + + Args: + rows: Page rows in scan order, already trimmed by ``take_cursor_page``. + cursor: The cursor the scan started from, if any. + reverse: Whether the scan ran backwards from ``cursor``. + has_more: Whether the scan found rows beyond the page. + encode_cursor: Encodes the cursor anchored at a given row. + + Returns: + The page in display order with cursors and flags in forward semantics. + """ + items = list(rows) + next_cursor = encode_cursor(items[-1]) if has_more and items else None + prev_cursor = encode_cursor(items[0]) if cursor is not None and items else None + has_previous = cursor is not None + + if reverse: + items.reverse() + next_cursor, prev_cursor = prev_cursor, next_cursor + # In reverse mode "next" walks back toward the page we came from, which + # is only reachable when this page produced an anchor cursor. Tying it + # to a bare `cursor is not None` would advertise has_more=True with + # next_cursor=None on an empty page, enabling a dead pagination control. + has_more, has_previous = next_cursor is not None, has_more + + return CursorPage( + items=items, + next_cursor=next_cursor, + prev_cursor=prev_cursor, + has_more=has_more, + has_previous=has_previous, + ) + + class CursorData(BaseModel): """Internal structure for cursor data.""" @@ -72,6 +162,141 @@ def parse_datetime_string( return v return v + @property + def has_sort_value(self) -> bool: + """Whether the cursor carries an explicit sort value. + + ``encode_cursor`` always serializes ``sort_value``, so a cursor issued + by this server has the key even when the anchor row's sort column is + NULL. An absent key means a cursor from before sort-aware pagination + (or a hand-built one), which cannot filter anything. Keeping the two + apart is what lets a nullable sort follow its own ``sort_value: null`` + cursor instead of rejecting it as malformed. + """ + return "sort_value" in self.model_fields_set + + +@overload +def validate_cursor_sort_column( + cursor: CursorData, + *, + sort_column: str, + expected_type: type | tuple[type, ...] | None = ..., + allow_null: Literal[False] = ..., +) -> str | int | float | datetime: ... + + +@overload +def validate_cursor_sort_column( + cursor: CursorData, + *, + sort_column: str, + expected_type: type | tuple[type, ...] | None = ..., + allow_null: Literal[True], +) -> str | int | float | datetime | None: ... + + +def validate_cursor_sort_column( + cursor: CursorData, + *, + sort_column: str, + expected_type: type | tuple[type, ...] | None = None, + allow_null: bool = False, +) -> str | int | float | datetime | None: + """Return the cursor's sort value, or raise if it cannot filter this query. + + Args: + cursor: The decoded cursor. + sort_column: The sort column of the current request. + expected_type: Type the sort value must have, when the column's cursor + representation is narrower than the query's column type (enum sorts + encode a rank, not the enum value). + allow_null: Whether the sort column admits NULLs. When set, a cursor + anchored on a NULL sort value returns ``None`` instead of raising, + and the caller must pair it with a NULL-aware keyset predicate (see + ``keyset_filter``). Legacy cursors that omit the field entirely are + still rejected. + + Raises: + InvalidCursorError: The cursor belongs to a different sort, carries no + usable sort value, or carries one of the wrong type. + """ + if cursor.sort_column != sort_column: + raise InvalidCursorError( + f"Cursor was created for sort column {cursor.sort_column!r}, " + f"but this request sorts by {sort_column!r}. " + "Restart pagination without a cursor after changing the sort." + ) + if cursor.sort_value is None: + if allow_null and cursor.has_sort_value: + return None + raise InvalidCursorError( + f"Cursor is missing a sort value for column {sort_column!r}. " + "Restart pagination without a cursor." + ) + if expected_type is not None and not isinstance(cursor.sort_value, expected_type): + raise InvalidCursorError( + f"Cursor sort value for column {sort_column!r} has the wrong type. " + "Restart pagination without a cursor." + ) + return cursor.sort_value + + +def keyset_filter( + sort_col: sa.ColumnElement[Any], + id_col: sa.ColumnElement[Any], + *, + sort_value: str | int | float | datetime | None, + id_value: Any, + ascending: bool, + nullable: bool = True, +) -> sa.ColumnElement[bool]: + """Build the predicate selecting rows strictly after a cursor anchor. + + Args: + sort_col: The column (or expression) the query sorts by. + id_col: The unique tie-breaker column, sorted alongside ``sort_col``. + sort_value: The anchor row's sort value; ``None`` anchors inside the + NULL block. + id_value: The anchor row's tie-breaker value. + ascending: The direction of the *scan*, not of the requested sort. + Reverse pagination inverts the scan so ``LIMIT`` keeps the rows + nearest the cursor, and this predicate must follow that inversion. + nullable: Whether ``sort_col`` can hold NULLs. Pass ``False`` for + NOT NULL columns to keep the predicate free of NULL branches; such + a column can never produce a ``sort_value`` of ``None``. + + NULL placement follows PostgreSQL's defaults, which invert consistently: + ``ASC`` puts NULLs last and ``DESC`` puts them first, so ``ASC NULLS LAST`` + reversed is exactly ``DESC NULLS FIRST``. An ascending scan therefore + treats NULL as greater than every value, a descending scan treats it as + less than every value, and inside the NULL block only ``id_col`` orders + rows. Callers must order by the matching ``nulls_last()``/``nulls_first()``. + """ + if ascending: + if sort_value is None: + # NULLs sort last, so nothing outside the NULL block follows. + return sa.and_(sort_col.is_(None), id_col > id_value) + after = [sort_col > sort_value] + if nullable: + # NULLs sort after every value, and `>` is unknown against them. + after.append(sort_col.is_(None)) + after.append(sa.and_(sort_col == sort_value, id_col > id_value)) + return sa.or_(*after) + + if sort_value is None: + # NULLs sort first, so every non-NULL row follows the NULL block. + return sa.or_( + sort_col.is_not(None), + sa.and_(sort_col.is_(None), id_col < id_value), + ) + # NULLs sort before every value, so a non-NULL anchor already excludes + # them: `<` is unknown against NULL and drops those rows. + return sa.or_( + sort_col < sort_value, + sa.and_(sort_col == sort_value, id_col < id_value), + ) + class BaseCursorPaginator: """Base class for cursor-based pagination.""" diff --git a/tracecat/tables/service.py b/tracecat/tables/service.py index 9ff7037c99..b3937dec97 100644 --- a/tracecat/tables/service.py +++ b/tracecat/tables/service.py @@ -41,6 +41,11 @@ BaseCursorPaginator, CursorPaginatedResponse, CursorPaginationParams, + InvalidCursorError, + build_cursor_page, + keyset_filter, + take_cursor_page, + validate_cursor_sort_column, ) from tracecat.service import BaseWorkspaceService from tracecat.tables.common import ( @@ -1444,6 +1449,21 @@ async def list_rows( sort_col = sa.column(self._sanitize_identifier(sort_column)) + # Custom columns default to nullable, so a page boundary can legitimately + # land on a NULL sort value. System columns are NOT NULL by construction. + if sort_column in _TABLE_SYSTEM_COLUMNS: + sort_column_nullable = False + else: + column_meta = next( + (col for col in table.columns if col.name == sort_column), None + ) + sort_column_nullable = column_meta is None or column_meta.nullable + + # Reverse pagination scans away from the cursor in the inverted sort + # order so LIMIT keeps the rows nearest the cursor; build_cursor_page + # puts them back into display order. + scan_ascending = (sort_direction == "asc") != params.reverse + # Apply cursor-based pagination with sort-column-aware filtering if params.cursor: try: @@ -1453,79 +1473,49 @@ async def list_rows( cursor_id = UUID(cursor_data.id) - # Check if cursor was created with the same sort column - cursor_sort_value = cursor_data.sort_value - cursor_has_sort_value = ( - cursor_data.sort_column == sort_column and cursor_sort_value is not None + # A cursor from a different sort cannot filter this query. Reject it + # instead of dropping the filter, which would silently return the + # first page while still reporting has_previous. A NULL sort value + # is accepted for nullable columns: this service issues such cursors + # whenever a page boundary falls inside the NULL block. + sort_cursor_value = validate_cursor_sort_column( + cursor_data, sort_column=sort_column, allow_null=True ) + if sort_cursor_value is None and not sort_column_nullable: + raise InvalidCursorError( + f"Cursor carries a NULL sort value for NOT NULL column " + f"{sort_column!r}. Restart pagination without a cursor." + ) - if cursor_has_sort_value: - # Use sort column value for cursor filtering - sort_cursor_value = cursor_sort_value - - # Composite filtering: (sort_col, id) matches ORDER BY - if sort_direction == "asc": - if params.reverse: - # Going backward: get records before cursor in sort order - stmt = stmt.where( - sa.or_( - sort_col < sort_cursor_value, - sa.and_( - sort_col == sort_cursor_value, - sa.column("id") < cursor_id, - ), - ) - ) - else: - # Going forward: get records after cursor in sort order - stmt = stmt.where( - sa.or_( - sort_col > sort_cursor_value, - sa.and_( - sort_col == sort_cursor_value, - sa.column("id") > cursor_id, - ), - ) - ) - else: - # Descending order - if params.reverse: - # Going backward: get records after cursor in sort order - stmt = stmt.where( - sa.or_( - sort_col > sort_cursor_value, - sa.and_( - sort_col == sort_cursor_value, - sa.column("id") > cursor_id, - ), - ) - ) - else: - # Going forward: get records before cursor in sort order - stmt = stmt.where( - sa.or_( - sort_col < sort_cursor_value, - sa.and_( - sort_col == sort_cursor_value, - sa.column("id") < cursor_id, - ), - ) - ) + # Composite filtering: (sort_col, id) matches ORDER BY + stmt = stmt.where( + keyset_filter( + sort_col, + sa.column("id"), + sort_value=sort_cursor_value, + id_value=cursor_id, + ascending=scan_ascending, + nullable=sort_column_nullable, + ) + ) - # Apply sorting: (sort_col, id) for stable pagination - # Use id as tie-breaker unless we're already sorting by id + # Apply sorting: (sort_col, id) for stable pagination. NULL placement is + # spelled out so it matches the keyset predicate above rather than + # relying on the server default. + order_clause = ( + sort_col.asc().nulls_last() + if scan_ascending + else sort_col.desc().nulls_first() + ) if sort_column == "id": # No tie-breaker needed when sorting by id (already unique) - if sort_direction == "asc": - stmt = stmt.order_by(sort_col.asc()) - else: - stmt = stmt.order_by(sort_col.desc()) + stmt = stmt.order_by(order_clause) else: # Add id as tie-breaker for non-unique columns - if sort_direction == "asc": - stmt = stmt.order_by(sort_col.asc(), sa.column("id").asc()) - else: - stmt = stmt.order_by(sort_col.desc(), sa.column("id").desc()) + id_order = ( + sa.column("id").asc() if scan_ascending else sa.column("id").desc() + ) + stmt = stmt.order_by(order_clause, id_order) # Fetch limit + 1 to determine if there are more items stmt = stmt.limit(params.limit + 1) @@ -1552,46 +1542,27 @@ async def list_rows( raise # Check if there are more items - has_more = len(rows) > params.limit - if has_more: - rows = rows[: params.limit] - has_previous = params.cursor is not None + scanned_rows, has_more = take_cursor_page(rows, limit=params.limit) # Generate cursors with sort column info for proper pagination - next_cursor = None - prev_cursor = None - - if rows: - if has_more: - # Generate next cursor from the last item - last_item = rows[-1] - next_cursor = BaseCursorPaginator.encode_cursor( - last_item["id"], - sort_column=sort_column, - sort_value=last_item.get(sort_column), - ) - - if params.cursor: - # If we used a cursor to get here, we can go back - first_item = rows[0] - prev_cursor = BaseCursorPaginator.encode_cursor( - first_item["id"], - sort_column=sort_column, - sort_value=first_item.get(sort_column), - ) - - # If we were doing reverse pagination, swap the cursors and reverse items - if params.reverse: - rows = list(reversed(rows)) - next_cursor, prev_cursor = prev_cursor, next_cursor - has_more, has_previous = has_previous, has_more + page = build_cursor_page( + scanned_rows, + cursor=params.cursor, + reverse=params.reverse, + has_more=has_more, + encode_cursor=lambda row: BaseCursorPaginator.encode_cursor( + row["id"], + sort_column=sort_column, + sort_value=row.get(sort_column), + ), + ) return CursorPaginatedResponse( - items=rows, - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=has_previous, + items=page.items, + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, ) async def batch_insert_rows( diff --git a/tracecat/workflow/management/management.py b/tracecat/workflow/management/management.py index 821b9b3463..c1ca7c42fb 100644 --- a/tracecat/workflow/management/management.py +++ b/tracecat/workflow/management/management.py @@ -56,6 +56,8 @@ BaseCursorPaginator, CursorPaginatedResponse, CursorPaginationParams, + build_cursor_page, + take_cursor_page, ) from tracecat.registry.lock.service import RegistryLockService from tracecat.registry.lock.types import RegistryLock @@ -522,12 +524,7 @@ async def list_workflows( stmt = stmt.options(selectinload(Workflow.tags)) results = await self.session.execute(stmt) - raw_items = list(results.all()) - - # Check if there are more items - has_more = len(raw_items) > params.limit - if has_more: - raw_items = raw_items[: params.limit] + raw_items, has_more = take_cursor_page(list(results.all()), limit=params.limit) # Process results into the expected format items = [] @@ -560,37 +557,25 @@ async def list_workflows( items.append((workflow, latest_defn, trigger_summary)) # Generate cursors - next_cursor = None - prev_cursor = None - - if items: - if has_more: - last_workflow = items[-1][0] # Get the workflow from the tuple - next_cursor = paginator.encode_cursor( - last_workflow.id, - sort_column="created_at", - sort_value=last_workflow.created_at, - ) - - if params.cursor: - first_workflow = items[0][0] # Get the workflow from the tuple - prev_cursor = paginator.encode_cursor( - first_workflow.id, - sort_column="created_at", - sort_value=first_workflow.created_at, - ) - - # If we were doing reverse pagination, swap the cursors and reverse items - if params.reverse: - items = list(reversed(items)) - next_cursor, prev_cursor = prev_cursor, next_cursor + page = build_cursor_page( + items, + cursor=params.cursor, + reverse=params.reverse, + has_more=has_more, + # Get the workflow from the tuple + encode_cursor=lambda item: paginator.encode_cursor( + item[0].id, + sort_column="created_at", + sort_value=item[0].created_at, + ), + ) return CursorPaginatedResponse( - items=items, - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=params.cursor is not None, + items=page.items, + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, ) async def get_workflow(