Migrate WES pagination to unified keyset-based approach

Replace offset-based and security.encode_id() pagination with generalized
keyset abstraction supporting both single-ID and composite keysets.

New abstraction (lib/galaxy/model/keyset_token_pagination.py):
- KeysetToken Protocol: polymorphic via to_values()/from_values()
- SingleKeysetToken: single ID keyset for runs pagination
- KeysetPagination: base64+JSON encoder/decoder
- TaskKeysetToken (wes.py): composite (step_order, job_index) keyset

Changes to WES pagination:
- list_runs: SingleKeysetToken with WHERE id < last_id (unchanged query)
- get_run_tasks: TaskKeysetToken with tuple comparison WHERE (step_order, job_index) > (last_step, last_idx)
- Removed offset-based _encode_page_token/_decode_page_token
- Removed _encode_keyset_token/_decode_keyset_token

Benefits:
- Cursor-stable pagination under concurrent changes
- Efficient composite keyset filtering on UNION queries
- Consistent token encoding across all WES endpoints
- Protocol-based design - no type checking needed

Tests: 9 unit tests for keyset pagination abstraction (all passing)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: jmchilton

lib/galaxy/model/keyset_token_pagination.py

@@ -0,0 +1,116 @@
+"""Keyset-based pagination support for cursor-stable pagination."""
+
+import base64
+import json
+from dataclasses import dataclass
+from typing import (
+    Optional,
+    Protocol,
+    Type,
+    TypeVar,
+)
+
+from galaxy import exceptions
+
+
+class KeysetToken(Protocol):
+    """Protocol for keyset tokens that can be encoded/decoded.
+
+    Implementations must provide:
+    - to_values(): Convert token to normalized list of values for encoding
+    - from_values(): Reconstruct token from decoded values (classmethod)
+    """
+
+    def to_values(self) -> list:
+        """Convert token to normalized list of values for encoding.
+
+        Returns:
+            List of values to be JSON-encoded
+        """
+        ...
+
+    @classmethod
+    def from_values(cls, values: list) -> "KeysetToken":
+        """Reconstruct token from decoded values.
+
+        Args:
+            values: List of values from JSON decoding
+
+        Returns:
+            Token instance
+        """
+        ...
+
+
+@dataclass
+class SingleKeysetToken:
+    """Single ID column keyset token.
+
+    Used for pagination on a single numeric ID column (e.g., database IDs).
+    """
+
+    last_id: int
+
+    def to_values(self) -> list:
+        """Convert to normalized values."""
+        return [self.last_id]
+
+    @classmethod
+    def from_values(cls, values: list) -> "SingleKeysetToken":
+        """Reconstruct from decoded values."""
+        if len(values) < 1:
+            raise ValueError("SingleKeysetToken requires at least 1 value")
+        return cls(last_id=values[0])
+
+
+T = TypeVar("T", bound=KeysetToken)
+
+
+class KeysetPagination:
+    """Keyset pagination encoder/decoder using Protocol.
+
+    Encodes tokens to opaque base64 strings, works with any KeysetToken
+    implementation via Protocol duck typing.
+    """
+
+    def encode_token(self, token: KeysetToken) -> str:
+        """Encode keyset token to opaque base64 string.
+
+        Works with any KeysetToken implementation via Protocol.
+
+        Args:
+            token: Token implementing KeysetToken protocol
+
+        Returns:
+            Base64-encoded token string
+        """
+        values = token.to_values()
+        payload = json.dumps(values)
+        return base64.b64encode(payload.encode()).decode()
+
+    def decode_token(
+        self,
+        encoded: Optional[str],
+        token_class: Type[T],
+    ) -> Optional[T]:
+        """Decode token using provided token class.
+
+        Args:
+            encoded: Base64-encoded token string
+            token_class: Token class with from_values() classmethod
+
+        Returns:
+            Decoded token instance or None if encoded is None
+
+        Raises:
+            MessageException: If token is invalid
+        """
+        if not encoded:
+            return None
+
+        try:
+            payload = base64.b64decode(encoded.encode()).decode()
+            values = json.loads(payload)
+            return token_class.from_values(values)
+        except (ValueError, TypeError, json.JSONDecodeError) as e:
+            raise exceptions.MessageException(f"Invalid page_token: {str(e)}")

lib/galaxy/webapps/galaxy/services/wes.py

@@ -3,6 +3,7 @@
 import base64
 import json
 import logging
+from dataclasses import dataclass
 from typing import (
     Any,
     List,
@@ -17,6 +18,7 @@
 from sqlalchemy import (
     literal,
     select,
+    tuple_,
     union_all,
 )
 from sqlalchemy.orm import joinedload
@@ -38,6 +40,10 @@
     WorkflowInvocation,
     WorkflowInvocationStep,
 )
+from galaxy.model.keyset_token_pagination import (
+    KeysetPagination,
+    SingleKeysetToken,
+)
 from galaxy.schema.wes import (
     DefaultWorkflowEngineParameter,
     RunId,
@@ -73,41 +79,28 @@
 }
 
 WES_TO_GALAXY_STATE = {v: k for k, v in GALAXY_TO_WES_STATE.items()}
-PAGINATION_KEYSET_TOKEN_ENCODE_KEY = "pag_tok"
-
 
-def _encode_page_token(offset: int) -> str:
-    """Encode an offset as a base64 page token.
 
-    Args:
-        offset: The offset (row number) for pagination
+@dataclass
+class TaskKeysetToken:
+    """Composite keyset token for task pagination (step_order, job_index).
 
-    Returns:
-        Base64-encoded page token
+    Used to identify position in task list for cursor-based pagination.
     """
-    return base64.b64encode(str(offset).encode()).decode()
 
+    step_order: int
+    job_index: int
 
-def _encode_keyset_token(security: IdEncodingHelper, last_id: int) -> str:
-    """Encode last seen ID as keyset page token."""
-    return security.encode_id(last_id, kind=PAGINATION_KEYSET_TOKEN_ENCODE_KEY)
+    def to_values(self) -> list:
+        """Convert token to normalized list of values for encoding."""
+        return [self.step_order, self.job_index]
 
-
-def _decode_keyset_token(security: IdEncodingHelper, page_token: Optional[str]) -> Optional[int]:
-    """Decode keyset page token to last seen ID.
-
-    Returns None if no token, raises on invalid token.
-    """
-    if not page_token:
-        return None
-
-    try:
-        last_id = security.decode_id(page_token, kind=PAGINATION_KEYSET_TOKEN_ENCODE_KEY)
-        if last_id < 0:
-            raise ValueError("ID cannot be negative")
-        return last_id
-    except (ValueError, TypeError) as e:
-        raise exceptions.MessageException(f"Invalid page_token: {str(e)}")
+    @classmethod
+    def from_values(cls, values: list) -> "TaskKeysetToken":
+        """Reconstruct token from decoded values."""
+        if len(values) < 2:
+            raise ValueError("TaskKeysetToken requires at least 2 values")
+        return cls(step_order=values[0], job_index=values[1])
 
 
 def _parse_gxworkflow_uri(workflow_url: str) -> tuple[str, bool]:
@@ -159,31 +152,6 @@ def _parse_gxworkflow_uri(workflow_url: str) -> tuple[str, bool]:
         raise exceptions.MessageException(f"Error parsing gxworkflow:// URI: {str(e)}")
 
 
-def _decode_page_token(page_token: Optional[str]) -> int:
-    """Decode a base64 page token to an offset.
-
-    Args:
-        page_token: The base64-encoded page token
-
-    Returns:
-        The offset (row number) for pagination
-
-    Raises:
-        exceptions.MessageException: If token is invalid
-    """
-    if not page_token:
-        return 0
-
-    try:
-        offset_str = base64.b64decode(page_token.encode()).decode()
-        offset = int(offset_str)
-        if offset < 0:
-            raise ValueError("Offset cannot be negative")
-        return offset
-    except (ValueError, TypeError) as e:
-        raise exceptions.MessageException(f"Invalid page_token: {str(e)}")
-
-
 def _load_workflow_content(
     trans: ProvidesUserContext,
     workflow_attachment: Optional[UploadFile],
@@ -384,6 +352,7 @@ def __init__(
         self._workflows_service = workflows_service
         self._config = config
         self._security = security
+        self._keyset_pagination = KeysetPagination()
 
     def service_info(self, trans: ProvidesUserContext, request_url: str) -> ServiceInfo:
         """Return WES service information.
@@ -628,7 +597,8 @@ def list_runs(
             RunListResponse with paginated list of runs and next_page_token if more results exist
         """
         # Decode keyset token to get last seen ID
-        last_id = _decode_keyset_token(self._security, page_token)
+        token = self._keyset_pagination.decode_token(page_token, token_class=SingleKeysetToken)
+        last_id = token.last_id if token else None
 
         # Build query with keyset filtering
         query = trans.sa_session.query(WorkflowInvocation).join(History).where(History.user_id == trans.user.id)
@@ -650,7 +620,8 @@ def list_runs(
         next_page_token = None
         if has_more and invocations:
             last_invocation = invocations[page_size - 1]
-            next_page_token = _encode_keyset_token(self._security, last_invocation.id)
+            token = SingleKeysetToken(last_id=last_invocation.id)
+            next_page_token = self._keyset_pagination.encode_token(token)
 
         return RunListResponse(runs=runs, next_page_token=next_page_token)
 
@@ -777,33 +748,42 @@ def _get_paginated_task_rows(
         self,
         trans: ProvidesUserContext,
         invocation_id: int,
-        offset: int,
+        last_token: Optional[TaskKeysetToken],
         limit: int,
     ) -> List[dict]:
-        """Fetch paginated task rows from database.
+        """Fetch paginated task rows using composite keyset pagination.
+
+        Uses (step_order, job_index) as composite keyset for cursor-based pagination.
 
         Returns list of dicts with keys: step_id, step_order, task_type, job_id, job_index
         """
         # Build UNION subquery
         task_rows_subquery = self._build_task_rows_query(invocation_id).subquery()
 
         # Apply ordering and pagination
-        stmt = (
-            select(
-                task_rows_subquery.c.step_id,
-                task_rows_subquery.c.step_order,
-                task_rows_subquery.c.task_type,
-                task_rows_subquery.c.job_id,
-                task_rows_subquery.c.job_index,
-            )
-            .order_by(
-                task_rows_subquery.c.step_order,
-                task_rows_subquery.c.job_index,
-            )
-            .offset(offset)
-            .limit(limit + 1)  # Fetch one extra to detect more results
+        stmt = select(
+            task_rows_subquery.c.step_id,
+            task_rows_subquery.c.step_order,
+            task_rows_subquery.c.task_type,
+            task_rows_subquery.c.job_id,
+            task_rows_subquery.c.job_index,
+        ).order_by(
+            task_rows_subquery.c.step_order,
+            task_rows_subquery.c.job_index,
         )
 
+        # Apply composite keyset filter if we have a cursor
+        if last_token is not None:
+            stmt = stmt.where(
+                tuple_(
+                    task_rows_subquery.c.step_order,
+                    task_rows_subquery.c.job_index,
+                )
+                > tuple_(last_token.step_order, last_token.job_index)
+            )
+
+        stmt = stmt.limit(limit + 1)  # Fetch one extra to detect more results
+
         result = trans.sa_session.execute(stmt)
         return [dict(row._mapping) for row in result]
 
@@ -900,28 +880,28 @@ def get_run_tasks(
     ) -> TaskListResponse:
         """Get paginated list of tasks for a workflow run.
 
-        Uses database-level pagination via UNION query to avoid loading
-        all steps/jobs into memory.
+        Uses composite keyset pagination via UNION query to avoid loading
+        all steps/jobs into memory and for cursor-based stability.
 
         Args:
             trans: Galaxy transaction/context
             run_id: The WES run ID (Galaxy invocation ID)
             page_size: Number of tasks per page (default 10, max 100)
-            page_token: Token for pagination (base64-encoded offset)
+            page_token: Token for pagination (composite keyset: step_order, job_index)
 
         Returns:
             TaskListResponse with paginated tasks
         """
         invocation = self._get_invocation(trans, run_id)
 
-        # Decode page token to offset
-        offset = _decode_page_token(page_token)
+        # Decode composite keyset token
+        token = self._keyset_pagination.decode_token(page_token, token_class=TaskKeysetToken)
 
         # Fetch paginated task rows (+1 to detect more results)
         task_rows = self._get_paginated_task_rows(
             trans,
             invocation.id,
-            offset,
+            token,
             page_size,
         )
 
@@ -938,8 +918,10 @@ def get_run_tasks(
 
         # Generate next page token
         next_page_token = None
-        if has_more:
-            next_page_token = _encode_page_token(offset + page_size)
+        if has_more and task_rows:
+            last_row = task_rows[-1]
+            token = TaskKeysetToken(step_order=last_row["step_order"], job_index=last_row["job_index"])
+            next_page_token = self._keyset_pagination.encode_token(token)
 
         return TaskListResponse(
             task_logs=task_logs if task_logs else None,