bug fixes

agentlightning/adapter/base.py

@@ -1,6 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-from typing import Generic, List, TypeVar
+from typing import Generic, Sequence, TypeVar
 
 from opentelemetry.sdk.trace import ReadableSpan
 
@@ -66,7 +66,7 @@ class Adapter(Generic[T_from, T_to]):
         raise NotImplementedError("Adapter.adapt() is not implemented")
 
 
-class OtelTraceAdapter(Adapter[List[ReadableSpan], T_to], Generic[T_to]):
+class OtelTraceAdapter(Adapter[Sequence[ReadableSpan], T_to], Generic[T_to]):
     """Base class for adapters that convert OpenTelemetry trace spans into other formats.
 
     This specialization of [`Adapter`][agentlightning.Adapter] expects a list of
@@ -84,7 +84,7 @@ class OtelTraceAdapter(Adapter[List[ReadableSpan], T_to], Generic[T_to]):
     """
 
 
-class TraceAdapter(Adapter[List[Span], T_to], Generic[T_to]):
+class TraceAdapter(Adapter[Sequence[Span], T_to], Generic[T_to]):
     """Base class for adapters that convert trace spans into other formats.
 
     This class specializes [`Adapter`][agentlightning.Adapter] for working with

agentlightning/adapter/messages.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import json
 from collections import defaultdict
-from typing import TYPE_CHECKING, Any, Dict, Generator, Iterable, List, Optional, TypedDict, Union, cast
+from typing import TYPE_CHECKING, Any, Dict, Generator, Iterable, List, Optional, Sequence, TypedDict, Union, cast
 
 from pydantic import TypeAdapter
 
@@ -208,7 +208,7 @@ class TraceToMessages(TraceAdapter[List[OpenAIMessages]]):
         children of the associated completion span.
     """
 
-    def get_tool_calls(self, completion: Span, all_spans: List[Span], /) -> Iterable[Dict[str, Any]]:
+    def get_tool_calls(self, completion: Span, all_spans: Sequence[Span], /) -> Iterable[Dict[str, Any]]:
         """Yield tool call payloads for a completion span.
 
         Args:
@@ -231,7 +231,7 @@ class TraceToMessages(TraceAdapter[List[OpenAIMessages]]):
             if tool_call:
                 yield tool_call
 
-    def adapt(self, source: List[Span], /) -> List[OpenAIMessages]:
+    def adapt(self, source: Sequence[Span], /) -> List[OpenAIMessages]:
         """Transform trace spans into OpenAI chat payloads.
 
         Args:

agentlightning/adapter/triplet.py

@@ -6,7 +6,7 @@ import json
 import logging
 import re
 from enum import Enum
-from typing import Any, Dict, List, Optional, Tuple, Union, cast
+from typing import Any, Dict, List, Optional, Sequence, Tuple, Union, cast
 
 from opentelemetry.sdk.trace import ReadableSpan
 from pydantic import BaseModel
@@ -670,7 +670,7 @@ class TracerTraceToTriplet(TraceToTripletBase):
         trace_tree.visualize(filename, interested_span_match=interested_span_match)
         return trace_tree
 
-    def adapt(self, source: Union[List[Span], List[ReadableSpan]], /) -> List[Triplet]:  # type: ignore
+    def adapt(self, source: Union[Sequence[Span], Sequence[ReadableSpan]], /) -> List[Triplet]:  # type: ignore
         """Convert tracer spans into [`Triplet`][agentlightning.Triplet] trajectories.
 
         Args:
@@ -800,7 +800,7 @@ class LlmProxyTraceToTriplet(TraceToTripletBase):
         rid = attrs.get("gen_ai.response.id") or attrs.get("llm.hosted_vllm.id")
         return str(rid) if isinstance(rid, str) and rid else None
 
-    def adapt(self, source: List[Span], /) -> List[Triplet]:  # type: ignore
+    def adapt(self, source: Sequence[Span], /) -> List[Triplet]:  # type: ignore
         """Convert LLM Proxy spans into [`Triplet`][agentlightning.Triplet] trajectories.
 
         Args:

agentlightning/algorithm/fast.py

@@ -143,7 +143,7 @@ class Baseline(FastAlgorithm):
         store = self.get_store()
 
         for index in train_indices + val_indices:
-            queuing_rollouts = await store.query_rollouts(status=["queuing", "requeuing"])
+            queuing_rollouts = await store.query_rollouts(status_in=["queuing", "requeuing"])
             if len(queuing_rollouts) <= 1:
                 # Only enqueue a new rollout when there is at most 1 rollout in the queue.
                 sample = dataset[index]
@@ -222,7 +222,7 @@ class Baseline(FastAlgorithm):
                     f"Processing index {index}. {len(train_indices)} train indices and {len(val_indices)} val indices in total."
                 )
                 while True:
-                    queuing_rollouts = await store.query_rollouts(status=["queuing", "requeuing"])
+                    queuing_rollouts = await store.query_rollouts(status_in=["queuing", "requeuing"])
                     if len(queuing_rollouts) <= self.max_queue_length:
                         # Only enqueue a new rollout when there is at most "max_queue_length" rollout in the queue.
                         sample = concatenated_dataset[index]

agentlightning/store/base.py

@@ -18,6 +18,7 @@ from agentlightning.types import (
     Span,
     TaskInput,
     Worker,
+    WorkerStatus,
 )
 
 
@@ -292,30 +293,77 @@ class LightningStore:
         raise NotImplementedError()
 
     async def query_rollouts(
-        self, *, status: Optional[Sequence[RolloutStatus]] = None, rollout_ids: Optional[Sequence[str]] = None
-    ) -> List[Rollout]:
+        self,
+        *,
+        status_in: Optional[Sequence[RolloutStatus]] = None,
+        rollout_id_in: Optional[Sequence[str]] = None,
+        rollout_id_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+        # Deprecated fields
+        status: Optional[Sequence[RolloutStatus]] = None,
+        rollout_ids: Optional[Sequence[str]] = None,
+    ) -> Sequence[Rollout]:
         """Retrieve rollouts filtered by status and/or explicit identifiers.
 
+        This interface supports structured filtering, sorting, and pagination so
+        callers can build simple dashboards without copying data out of the
+        store. The legacy parameters `status` and `rollout_ids` remain valid and
+        are treated as aliases for `status_in` and `rollout_id_in`
+        respectively—when both the new and deprecated parameters are supplied
+        the new parameters take precedence.
+
         Args:
-            status: Optional whitelist of [`RolloutStatus`][agentlightning.RolloutStatus] values.
-            rollout_ids: Optional whitelist of rollout identifiers to include.
+            status_in: Optional whitelist of [`RolloutStatus`][agentlightning.RolloutStatus] values.
+            rollout_id_in: Optional whitelist of rollout identifiers to include.
+            rollout_id_contains: Optional substring match for rollout identifiers.
+            filter_logic: Logical operator to combine filters.
+            sort_by: Optional field to sort by. Must reference a numeric or string
+                field on [`Rollout`][agentlightning.Rollout].
+            sort_order: Direction to sort when `sort_by` is provided.
+            limit: Maximum number of rows to return. Use `-1` for "no limit".
+            offset: Number of rows to skip before returning results.
+            status: Deprecated field. Use `status_in` instead.
+            rollout_ids: Deprecated field. Use `rollout_id_in` instead.
 
         Returns:
-            A list of matching rollouts. Ordering is backend-defined but must be deterministic.
+            A sequence of matching rollouts (or [`AttemptedRollout`][agentlightning.AttemptedRollout]
+            when attempts exist). Ordering is deterministic when `sort_by` is set.
+            The return value is not guaranteed to be a list.
 
         Raises:
             NotImplementedError: Subclasses must implement the query.
         """
         raise NotImplementedError()
 
-    async def query_attempts(self, rollout_id: str) -> List[Attempt]:
+    async def query_attempts(
+        self,
+        rollout_id: str,
+        *,
+        sort_by: Optional[str] = "sequence_id",
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> Sequence[Attempt]:
         """Return every attempt ever created for `rollout_id` in ascending sequence order.
 
+        The parameters allow callers to re-order or paginate the attempts so that
+        large retry histories can be streamed lazily.
+
         Args:
             rollout_id: Identifier of the rollout being inspected.
+            sort_by: Field to sort by. Must be a numeric or string field of
+                [`Attempt`][agentlightning.Attempt]. Defaults to `sequence_id` (oldest first).
+            sort_order: Order to sort by.
+            limit: Limit on the number of results. `-1` for unlimited.
+            offset: Offset into the results.
 
         Returns:
-            Attempts sorted by `sequence_id` (oldest first). Returns an empty list when none exist.
+            Sequence of Attempts. Returns an empty sequence when none exist.
+            The return value is not guaranteed to be a list.
 
         Raises:
             NotImplementedError: Subclasses must implement the query.
@@ -352,11 +400,35 @@ class LightningStore:
         """
         raise NotImplementedError()
 
-    async def query_resources(self) -> List[ResourcesUpdate]:
+    async def query_resources(
+        self,
+        *,
+        resources_id: Optional[str] = None,
+        resources_id_contains: Optional[str] = None,
+        # Filter logic is not supported here because I can't see why it's needed.
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> Sequence[ResourcesUpdate]:
         """List every stored resource snapshot in insertion order.
 
+        Supports lightweight filtering, sorting, and pagination for embedding in
+        dashboards.
+
+        Args:
+            resources_id: Optional identifier of the resources to include.
+            resources_id_contains: Optional substring match for resources identifiers.
+            sort_by: Optional field to sort by (must be numeric or string on
+                [`ResourcesUpdate`][agentlightning.ResourcesUpdate]).
+            sort_order: Order to sort by.
+            limit: Limit on the number of results. `-1` for unlimited.
+            offset: Offset into the results.
+
         Returns:
-            A chronological list of [`ResourcesUpdate`][agentlightning.ResourcesUpdate] objects.
+            [`ResourcesUpdate`][agentlightning.ResourcesUpdate] objects.
+            By default, resources are sorted in a deterministic but undefined order.
+            The return value is not guaranteed to be a list.
 
         Raises:
             NotImplementedError: Subclasses must implement retrieval.
@@ -439,19 +511,61 @@ class LightningStore:
         """
         raise NotImplementedError()
 
-    async def query_spans(self, rollout_id: str, attempt_id: str | Literal["latest"] | None = None) -> List[Span]:
+    async def query_spans(
+        self,
+        rollout_id: str,
+        attempt_id: str | Literal["latest"] | None = None,
+        *,
+        # Filtering
+        trace_id: Optional[str] = None,
+        trace_id_contains: Optional[str] = None,
+        span_id: Optional[str] = None,
+        span_id_contains: Optional[str] = None,
+        parent_id: Optional[str] = None,
+        parent_id_contains: Optional[str] = None,
+        name: Optional[str] = None,
+        name_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        # Pagination
+        limit: int = -1,
+        offset: int = 0,
+        # Sorting
+        sort_by: Optional[str] = "sequence_id",
+        sort_order: Literal["asc", "desc"] = "asc",
+    ) -> Sequence[Span]:
         """Return the stored spans for a rollout, optionally scoped to one attempt.
 
-        Spans must be returned in ascending `sequence_id` order. Implementations may raise
-        a `RuntimeError` when spans were evicted or expired.
+        Supports a handful of filters that cover the most common debugging
+        scenarios (matching `trace_id`/`span_id`/`parent_id` or substring
+        matches on the span name). `attempt_id="latest"` acts as a convenience
+        that resolves the most recent attempt before evaluating filters. When
+        `attempt_id=None`, spans across every attempt are eligible. By default
+        results are sorted by `sequence_id` (oldest first). Implementations may
+        raise a `RuntimeError` when spans were evicted or expired.
 
         Args:
             rollout_id: Identifier of the rollout being inspected.
             attempt_id: Attempt identifier to filter by. Pass `"latest"` to retrieve only the
                 most recent attempt, or `None` to return all spans across attempts.
+            trace_id: Optional trace ID to filter by.
+            trace_id_contains: Optional substring match for trace IDs.
+            span_id: Optional span ID to filter by.
+            span_id_contains: Optional substring match for span IDs.
+            parent_id: Optional parent span ID to filter by.
+            parent_id_contains: Optional substring match for parent span IDs.
+            name: Optional span name to filter by.
+            name_contains: Optional substring match for span names.
+            filter_logic: Logical operator to combine the optional filters above.
+                The `rollout_id` argument is always applied with AND semantics.
+            limit: Limit on the number of results. `-1` for unlimited.
+            offset: Offset into the results.
+            sort_by: Field to sort by. Must be a numeric or string field of
+                [`Span`][agentlightning.Span].
+            sort_order: Order to sort by.
 
         Returns:
             An ordered list of spans (possibly empty).
+            The return value is not guaranteed to be a list.
 
         Raises:
             NotImplementedError: Subclasses must implement the query.
@@ -578,11 +692,29 @@ class LightningStore:
 
     async def query_workers(
         self,
-    ) -> List[Worker]:
+        *,
+        status_in: Optional[Sequence[WorkerStatus]] = None,
+        worker_id_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> Sequence[Worker]:
         """Query all workers in the system.
 
+        Args:
+            status_in: Optional whitelist of [`WorkerStatus`][agentlightning.WorkerStatus] values.
+            worker_id_contains: Optional substring match for worker identifiers.
+            filter_logic: Logical operator to combine the optional filters above.
+            sort_by: Field to sort by. Must be a numeric or string field of [`Worker`][agentlightning.Worker].
+            sort_order: Order to sort by.
+            limit: Limit on the number of results. `-1` for unlimited.
+            offset: Offset into the results.
+
         Returns:
-            A list of all workers.
+            Sequence of Workers. Returns an empty sequence when none exist.
+            The return value is not guaranteed to be a list.
         """
         raise NotImplementedError()
 

agentlightning/store/client_server.py

@@ -9,7 +9,21 @@ import threading
 import time
 import traceback
 from pathlib import Path
-from typing import Any, Awaitable, Callable, Dict, Generic, List, Literal, Optional, Sequence, TypeVar, Union
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    Dict,
+    List,
+    Literal,
+    Mapping,
+    Optional,
+    Sequence,
+    Tuple,
+    Type,
+    TypeVar,
+    Union,
+)
 
 import aiohttp
 from fastapi import APIRouter, Body, Depends, FastAPI, HTTPException
@@ -32,6 +46,7 @@ from agentlightning.types import (
     AttemptedRollout,
     AttemptStatus,
     NamedResources,
+    PaginatedResult,
     ResourcesUpdate,
     Rollout,
     RolloutConfig,
@@ -54,13 +69,7 @@ API_AGL_PREFIX = "/agl"
 API_V1_AGL_PREFIX = API_V1_PREFIX + API_AGL_PREFIX
 
 T = TypeVar("T")
-
-
-class PaginatedResponse(BaseModel, Generic[T]):
-    items: List[T]
-    limit: int
-    offset: int
-    total: int
+T_model = TypeVar("T_model", bound=BaseModel)
 
 
 class RolloutRequest(BaseModel):
@@ -128,7 +137,7 @@ class QueryAttemptsRequest(BaseModel):
     limit: int = -1
     offset: int = 0
     # Sorting
-    sort_by: Optional[str] = None
+    sort_by: Optional[str] = "sequence_id"
     sort_order: Literal["asc", "desc"] = "asc"
 
 
@@ -161,7 +170,7 @@ class QuerySpansRequest(BaseModel):
     limit: int = -1
     offset: int = 0
     # Sorting
-    sort_by: Optional[str] = None
+    sort_by: Optional[str] = "sequence_id"
     sort_order: Literal["asc", "desc"] = "asc"
 
 
@@ -178,98 +187,6 @@ class QueryWorkersRequest(BaseModel):
     filter_logic: Literal["and", "or"] = "and"
 
 
-def _apply_filters_sort_paginate(
-    items: List[T],
-    filters: Dict[str, Any],
-    filter_logic: Literal["and", "or"],
-    sort_by: Optional[str],
-    sort_order: Literal["asc", "desc"],
-    limit: int,
-    offset: int,
-) -> PaginatedResponse[T]:
-    """Apply filtering, sorting, and pagination to a list of items."""
-    # Apply filters
-    filtered_items: List[T] = []
-    if not filters:
-        filtered_items = items
-    else:
-        for item in items:
-            matches: List[bool] = []
-            for key, value in filters.items():
-                if value is None:
-                    continue
-
-                # Handle _in suffix (list membership)
-                if key.endswith("_in"):
-                    field = key[:-3]
-                    item_value = getattr(item, field, None)
-                    matches.append(item_value in value if isinstance(value, list) else False)
-                # Handle _contains suffix (substring match)
-                elif key.endswith("_contains"):
-                    field = key[:-9]
-                    item_value = getattr(item, field, None)
-                    if item_value is not None and isinstance(item_value, str) and isinstance(value, str):
-                        matches.append(value in item_value)
-                    else:
-                        matches.append(False)
-                # Exact match
-                else:
-                    item_value = getattr(item, key, None)
-                    matches.append(item_value == value)
-
-            if matches:
-                if filter_logic == "and":
-                    if all(matches):
-                        filtered_items.append(item)
-                else:  # "or"
-                    if any(matches):
-                        filtered_items.append(item)
-
-    # Apply sorting
-
-    def _get_sort_value(item: T, sort_by: str) -> Any:
-        if sort_by.endswith("_time"):
-            value = getattr(item, sort_by, None)
-            if value is None:
-                value = float("inf")
-            return value
-        else:
-            # Other than _time, we assume the value must be a string
-            value = getattr(item, sort_by, None)
-            if value is None:
-                if sort_by not in item.__class__.model_fields:  # type: ignore
-                    raise HTTPException(
-                        status_code=400,
-                        detail=f"Failed to sort items by {sort_by}: {sort_by} is not a field of {item.__class__.__name__}",
-                    )
-                field_type = str(item.__class__.model_fields[sort_by].annotation)  # type: ignore
-                if "str" in field_type or "Literal" in field_type:
-                    return ""
-                if "int" in field_type:
-                    return 0
-                if "float" in field_type:
-                    return 0.0
-                raise HTTPException(
-                    status_code=400, detail=f"Failed to sort items by {sort_by}: {value} is not a string or number"
-                )
-            return value
-
-    if sort_by:
-        reverse = sort_order == "desc"
-        filtered_items.sort(key=lambda x: _get_sort_value(x, sort_by), reverse=reverse)
-
-    # Get total count before pagination
-    total = len(filtered_items)
-
-    # Apply pagination
-    if limit == -1:
-        paginated_items = filtered_items[offset:]
-    else:
-        paginated_items = filtered_items[offset : offset + limit]
-
-    return PaginatedResponse(items=paginated_items, limit=limit, offset=offset, total=total)
-
-
 class CachedStaticFiles(StaticFiles):
     def file_response(self, *args: Any, **kwargs: Any) -> Response:
         resp = super().file_response(*args, **kwargs)
@@ -546,6 +463,48 @@ class LightningStoreServer(LightningStore):
 
         api = APIRouter(prefix=API_V1_PREFIX)
 
+        def _validate_paginated_request(
+            request: Union[
+                QueryRolloutsRequest,
+                QueryAttemptsRequest,
+                QueryResourcesRequest,
+                QueryWorkersRequest,
+                QuerySpansRequest,
+            ],
+            target_type: Type[T_model],
+        ) -> None:
+            """Raise an error early if the request is not a valid paginated request."""
+            if request.sort_by is not None and request.sort_by not in target_type.model_fields:
+                raise HTTPException(
+                    status_code=400,
+                    detail=f"Invalid sort_by: {request.sort_by}, allowed fields are: {', '.join(target_type.model_fields.keys())}",
+                )
+            if request.sort_order not in ["asc", "desc"]:
+                raise HTTPException(
+                    status_code=400, detail=f"Invalid sort_order: {request.sort_order}, allowed values are: asc, desc"
+                )
+            if request.limit == 0 or (request.limit < 0 and request.limit != -1):
+                raise HTTPException(status_code=400, detail="Limit must be greater than 0 or -1 for no limit")
+            if not request.offset >= 0:
+                raise HTTPException(status_code=400, detail="Offset must be greater than or equal to 0")
+            if hasattr(request, "filter_logic") and request.filter_logic not in ["and", "or"]:  # type: ignore
+                raise HTTPException(
+                    status_code=400, detail=f"Invalid filter_logic: {request.filter_logic}, allowed values are: and, or"  # type: ignore
+                )
+
+        def _build_paginated_response(items: Sequence[Any], *, limit: int, offset: int) -> PaginatedResult[Any]:
+            """FastAPI routes expect PaginatedResult payloads; wrap plain lists accordingly."""
+            if isinstance(items, PaginatedResult):
+                return items
+
+            # Assuming it's a list.
+            server_logger.warning(
+                "PaginatedResult expected; got a plain list. Converting to PaginatedResult. "
+                "Total items count will be inaccurate: %d",
+                len(items),
+            )
+            return PaginatedResult(items=items, limit=limit, offset=offset, total=len(items))
+
         @api.get(API_AGL_PREFIX + "/health")
         async def health():  # pyright: ignore[reportUnusedFunction]
             return {"status": "ok"}
@@ -577,29 +536,21 @@ class LightningStoreServer(LightningStore):
                 metadata=request.metadata,
             )
 
-        @api.get(API_AGL_PREFIX + "/rollouts", response_model=PaginatedResponse[Union[AttemptedRollout, Rollout]])
+        @api.get(API_AGL_PREFIX + "/rollouts", response_model=PaginatedResult[Union[AttemptedRollout, Rollout]])
         async def query_rollouts(params: QueryRolloutsRequest = Depends()):  # pyright: ignore[reportUnusedFunction]
+            _validate_paginated_request(params, Rollout)
             # Get all rollouts from the underlying store
-            all_rollouts = await self.query_rollouts()
-
-            # Build filter dict
-            filters: Dict[str, Any] = {}
-            if params.status_in:
-                filters["status_in"] = params.status_in
-            if params.rollout_id_in:
-                filters["rollout_id_in"] = params.rollout_id_in
-            if params.rollout_id_contains is not None:
-                filters["rollout_id_contains"] = params.rollout_id_contains
-
-            return _apply_filters_sort_paginate(
-                all_rollouts,
-                filters,
-                params.filter_logic,
-                params.sort_by,
-                params.sort_order,
-                params.limit,
-                params.offset,
+            results = await self.query_rollouts(
+                status_in=params.status_in,
+                rollout_id_in=params.rollout_id_in,
+                rollout_id_contains=params.rollout_id_contains,
+                filter_logic=params.filter_logic,
+                sort_by=params.sort_by,
+                sort_order=params.sort_order,
+                limit=params.limit,
+                offset=params.offset,
             )
+            return _build_paginated_response(results, limit=params.limit, offset=params.offset)
 
         @api.get(API_AGL_PREFIX + "/rollouts/{rollout_id}", response_model=Union[AttemptedRollout, Rollout])
         async def get_rollout_by_id(rollout_id: str):  # pyright: ignore[reportUnusedFunction]
@@ -649,25 +600,19 @@ class LightningStoreServer(LightningStore):
                 metadata=_get_mandatory_field_or_unset(request, "metadata"),
             )
 
-        @api.get(API_AGL_PREFIX + "/workers", response_model=PaginatedResponse[Worker])
+        @api.get(API_AGL_PREFIX + "/workers", response_model=PaginatedResult[Worker])
         async def query_workers(params: QueryWorkersRequest = Depends()):  # pyright: ignore[reportUnusedFunction]
-            all_workers = await self.query_workers()
-
-            filters: Dict[str, Any] = {}
-            if params.status_in:
-                filters["status_in"] = params.status_in
-            if params.worker_id_contains is not None:
-                filters["worker_id_contains"] = params.worker_id_contains
-
-            return _apply_filters_sort_paginate(
-                all_workers,
-                filters,
-                params.filter_logic,
-                params.sort_by,
-                params.sort_order,
-                params.limit,
-                params.offset,
+            _validate_paginated_request(params, Worker)
+            workers = await self.query_workers(
+                status_in=params.status_in,
+                worker_id_contains=params.worker_id_contains,
+                filter_logic=params.filter_logic,
+                sort_by=params.sort_by,
+                sort_order=params.sort_order,
+                limit=params.limit,
+                offset=params.offset,
             )
+            return _build_paginated_response(workers, limit=params.limit, offset=params.offset)
 
         @api.get(API_AGL_PREFIX + "/workers/{worker_id}", response_model=Optional[Worker])
         async def get_worker(worker_id: str):  # pyright: ignore[reportUnusedFunction]
@@ -682,48 +627,36 @@ class LightningStoreServer(LightningStore):
                 heartbeat_stats=_get_mandatory_field_or_unset(request, "heartbeat_stats"),
             )
 
-        @api.get(API_AGL_PREFIX + "/rollouts/{rollout_id}/attempts", response_model=PaginatedResponse[Attempt])
+        @api.get(API_AGL_PREFIX + "/rollouts/{rollout_id}/attempts", response_model=PaginatedResult[Attempt])
         async def query_attempts(  # pyright: ignore[reportUnusedFunction]
             rollout_id: str, params: QueryAttemptsRequest = Depends()
         ):
-            # Get all attempts for the rollout
-            all_attempts = await self.query_attempts(rollout_id)
-
-            return _apply_filters_sort_paginate(
-                all_attempts,
-                {},  # No filters for attempts
-                "and",
-                params.sort_by,
-                params.sort_order,
-                params.limit,
-                params.offset,
+            _validate_paginated_request(params, Attempt)
+            attempts = await self.query_attempts(
+                rollout_id,
+                sort_by=params.sort_by,
+                sort_order=params.sort_order,
+                limit=params.limit,
+                offset=params.offset,
             )
+            return _build_paginated_response(attempts, limit=params.limit, offset=params.offset)
 
         @api.get(API_AGL_PREFIX + "/rollouts/{rollout_id}/attempts/latest", response_model=Optional[Attempt])
         async def get_latest_attempt(rollout_id: str):  # pyright: ignore[reportUnusedFunction]
             return await self.get_latest_attempt(rollout_id)
 
-        @api.get(API_AGL_PREFIX + "/resources", response_model=PaginatedResponse[ResourcesUpdate])
+        @api.get(API_AGL_PREFIX + "/resources", response_model=PaginatedResult[ResourcesUpdate])
         async def query_resources(params: QueryResourcesRequest = Depends()):  # pyright: ignore[reportUnusedFunction]
-            # Get all resources
-            all_resources = await self.query_resources()
-
-            # Build filter dict
-            filters: Dict[str, Any] = {}
-            if params.resources_id is not None:
-                filters["resources_id"] = params.resources_id
-            if params.resources_id_contains is not None:
-                filters["resources_id_contains"] = params.resources_id_contains
-
-            return _apply_filters_sort_paginate(
-                all_resources,
-                filters,
-                "and",
-                params.sort_by,
-                params.sort_order,
-                params.limit,
-                params.offset,
+            _validate_paginated_request(params, ResourcesUpdate)
+            resources = await self.query_resources(
+                resources_id=params.resources_id,
+                resources_id_contains=params.resources_id_contains,
+                sort_by=params.sort_by,
+                sort_order=params.sort_order,
+                limit=params.limit,
+                offset=params.offset,
             )
+            return _build_paginated_response(resources, limit=params.limit, offset=params.offset)
 
         @api.post(API_AGL_PREFIX + "/resources", status_code=201, response_model=ResourcesUpdate)
         async def add_resources(resources: NamedResources):  # pyright: ignore[reportUnusedFunction]
@@ -747,33 +680,27 @@ class LightningStoreServer(LightningStore):
         async def add_span(span: Span):  # pyright: ignore[reportUnusedFunction]
             return await self.add_span(span)
 
-        @api.get(API_AGL_PREFIX + "/spans", response_model=PaginatedResponse[Span])
+        @api.get(API_AGL_PREFIX + "/spans", response_model=PaginatedResult[Span])
         async def query_spans(params: QuerySpansRequest = Depends()):  # pyright: ignore[reportUnusedFunction]
-            # Get all spans for the rollout/attempt
-            all_spans = await self.query_spans(params.rollout_id, params.attempt_id)
-
-            # Build filter dict
-            filters: Dict[str, Any] = {}
-            if params.trace_id is not None:
-                filters["trace_id"] = params.trace_id
-            if params.trace_id_contains is not None:
-                filters["trace_id_contains"] = params.trace_id_contains
-            if params.span_id is not None:
-                filters["span_id"] = params.span_id
-            if params.span_id_contains is not None:
-                filters["span_id_contains"] = params.span_id_contains
-            if params.parent_id is not None:
-                filters["parent_id"] = params.parent_id
-            if params.parent_id_contains is not None:
-                filters["parent_id_contains"] = params.parent_id_contains
-            if params.name is not None:
-                filters["name"] = params.name
-            if params.name_contains is not None:
-                filters["name_contains"] = params.name_contains
-
-            return _apply_filters_sort_paginate(
-                all_spans, filters, params.filter_logic, params.sort_by, params.sort_order, params.limit, params.offset
+            _validate_paginated_request(params, Span)
+            spans = await self.query_spans(
+                params.rollout_id,
+                params.attempt_id,
+                trace_id=params.trace_id,
+                trace_id_contains=params.trace_id_contains,
+                span_id=params.span_id,
+                span_id_contains=params.span_id_contains,
+                parent_id=params.parent_id,
+                parent_id_contains=params.parent_id_contains,
+                name=params.name,
+                name_contains=params.name_contains,
+                filter_logic=params.filter_logic,
+                sort_by=params.sort_by,
+                sort_order=params.sort_order,
+                limit=params.limit,
+                offset=params.offset,
             )
+            return _build_paginated_response(spans, limit=params.limit, offset=params.offset)
 
         @api.post(API_AGL_PREFIX + "/spans/next", response_model=NextSequenceIdResponse)
         async def get_next_span_sequence_id(request: NextSequenceIdRequest):  # pyright: ignore[reportUnusedFunction]
@@ -919,18 +846,73 @@ class LightningStoreServer(LightningStore):
         return await self._call_store_method("start_attempt", rollout_id)
 
     async def query_rollouts(
-        self, *, status: Optional[Sequence[RolloutStatus]] = None, rollout_ids: Optional[Sequence[str]] = None
-    ) -> List[Rollout]:
-        return await self._call_store_method("query_rollouts", status=status, rollout_ids=rollout_ids)
+        self,
+        *,
+        status_in: Optional[Sequence[RolloutStatus]] = None,
+        rollout_id_in: Optional[Sequence[str]] = None,
+        rollout_id_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+        status: Optional[Sequence[RolloutStatus]] = None,
+        rollout_ids: Optional[Sequence[str]] = None,
+    ) -> PaginatedResult[Union[AttemptedRollout, Rollout]]:
+        return await self._call_store_method(
+            "query_rollouts",
+            status_in=status_in,
+            rollout_id_in=rollout_id_in,
+            rollout_id_contains=rollout_id_contains,
+            filter_logic=filter_logic,
+            sort_by=sort_by,
+            sort_order=sort_order,
+            limit=limit,
+            offset=offset,
+            status=status,
+            rollout_ids=rollout_ids,
+        )
 
-    async def query_attempts(self, rollout_id: str) -> List[Attempt]:
-        return await self._call_store_method("query_attempts", rollout_id)
+    async def query_attempts(
+        self,
+        rollout_id: str,
+        *,
+        sort_by: Optional[str] = "sequence_id",
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> PaginatedResult[Attempt]:
+        return await self._call_store_method(
+            "query_attempts",
+            rollout_id,
+            sort_by=sort_by,
+            sort_order=sort_order,
+            limit=limit,
+            offset=offset,
+        )
 
     async def get_latest_attempt(self, rollout_id: str) -> Optional[Attempt]:
         return await self._call_store_method("get_latest_attempt", rollout_id)
 
-    async def query_resources(self) -> List[ResourcesUpdate]:
-        return await self._call_store_method("query_resources")
+    async def query_resources(
+        self,
+        *,
+        resources_id: Optional[str] = None,
+        resources_id_contains: Optional[str] = None,
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> PaginatedResult[ResourcesUpdate]:
+        return await self._call_store_method(
+            "query_resources",
+            resources_id=resources_id,
+            resources_id_contains=resources_id_contains,
+            sort_by=sort_by,
+            sort_order=sort_order,
+            limit=limit,
+            offset=offset,
+        )
 
     async def get_rollout_by_id(self, rollout_id: str) -> Optional[Rollout]:
         return await self._call_store_method("get_rollout_by_id", rollout_id)
@@ -975,8 +957,39 @@ class LightningStoreServer(LightningStore):
         self,
         rollout_id: str,
         attempt_id: str | Literal["latest"] | None = None,
-    ) -> List[Span]:
-        return await self._call_store_method("query_spans", rollout_id, attempt_id)
+        *,
+        trace_id: Optional[str] = None,
+        trace_id_contains: Optional[str] = None,
+        span_id: Optional[str] = None,
+        span_id_contains: Optional[str] = None,
+        parent_id: Optional[str] = None,
+        parent_id_contains: Optional[str] = None,
+        name: Optional[str] = None,
+        name_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        limit: int = -1,
+        offset: int = 0,
+        sort_by: Optional[str] = "sequence_id",
+        sort_order: Literal["asc", "desc"] = "asc",
+    ) -> PaginatedResult[Span]:
+        return await self._call_store_method(
+            "query_spans",
+            rollout_id,
+            attempt_id,
+            trace_id=trace_id,
+            trace_id_contains=trace_id_contains,
+            span_id=span_id,
+            span_id_contains=span_id_contains,
+            parent_id=parent_id,
+            parent_id_contains=parent_id_contains,
+            name=name,
+            name_contains=name_contains,
+            filter_logic=filter_logic,
+            limit=limit,
+            offset=offset,
+            sort_by=sort_by,
+            sort_order=sort_order,
+        )
 
     async def update_rollout(
         self,
@@ -1018,8 +1031,27 @@ class LightningStoreServer(LightningStore):
             metadata,
         )
 
-    async def query_workers(self) -> List[Worker]:
-        return await self._call_store_method("query_workers")
+    async def query_workers(
+        self,
+        *,
+        status_in: Optional[Sequence[WorkerStatus]] = None,
+        worker_id_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> PaginatedResult[Worker]:
+        return await self._call_store_method(
+            "query_workers",
+            status_in=status_in,
+            worker_id_contains=worker_id_contains,
+            filter_logic=filter_logic,
+            sort_by=sort_by,
+            sort_order=sort_order,
+            limit=limit,
+            offset=offset,
+        )
 
     async def get_worker_by_id(self, worker_id: str) -> Optional[Worker]:
         return await self._call_store_method("get_worker_by_id", worker_id)
@@ -1187,7 +1219,7 @@ class LightningStoreClient(LightningStore):
         path: str,
         *,
         json: Any | None = None,
-        params: Dict[str, Any] | None = None,
+        params: Mapping[str, Any] | Sequence[Tuple[str, Any]] | None = None,
     ) -> Any:
         """
         Make an HTTP request with:
@@ -1347,17 +1379,43 @@ class LightningStoreClient(LightningStore):
         return AttemptedRollout.model_validate(data)
 
     async def query_rollouts(
-        self, *, status: Optional[Sequence[RolloutStatus]] = None, rollout_ids: Optional[Sequence[str]] = None
-    ) -> List[Rollout]:
-        params: Dict[str, Any] = {}
-        if status is not None:
-            params["status_in"] = status
-        if rollout_ids is not None:
-            params["rollout_id_in"] = list(rollout_ids)
+        self,
+        *,
+        status_in: Optional[Sequence[RolloutStatus]] = None,
+        rollout_id_in: Optional[Sequence[str]] = None,
+        rollout_id_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+        status: Optional[Sequence[RolloutStatus]] = None,
+        rollout_ids: Optional[Sequence[str]] = None,
+    ) -> PaginatedResult[Union[AttemptedRollout, Rollout]]:
+        params_list: List[Tuple[str, Any]] = []
 
-        data = await self._request_json("get", "/rollouts", params=params if params else None)
-        # Extract items from PaginatedResponse
-        return [
+        def _extend(key: str, values: Sequence[Any]) -> None:
+            for value in values:
+                params_list.append((key, value))
+
+        resolved_status = status_in if status_in is not None else status
+        resolved_rollout_ids = rollout_id_in if rollout_id_in is not None else rollout_ids
+
+        if resolved_status is not None:
+            _extend("status_in", resolved_status)
+        if resolved_rollout_ids is not None:
+            _extend("rollout_id_in", resolved_rollout_ids)
+        if rollout_id_contains is not None:
+            params_list.append(("rollout_id_contains", rollout_id_contains))
+        params_list.append(("filter_logic", filter_logic))
+        if sort_by is not None:
+            params_list.append(("sort_by", sort_by))
+            params_list.append(("sort_order", sort_order))
+        params_list.append(("limit", limit))
+        params_list.append(("offset", offset))
+
+        data = await self._request_json("get", "/rollouts", params=params_list or None)
+        items = [
             (
                 AttemptedRollout.model_validate(item)
                 if isinstance(item, dict) and "attempt" in item
@@ -1365,11 +1423,27 @@ class LightningStoreClient(LightningStore):
             )
             for item in data["items"]
         ]
+        return PaginatedResult(items=items, limit=data["limit"], offset=data["offset"], total=data["total"])
 
-    async def query_attempts(self, rollout_id: str) -> List[Attempt]:
-        data = await self._request_json("get", f"/rollouts/{rollout_id}/attempts")
-        # Extract items from PaginatedResponse
-        return [Attempt.model_validate(item) for item in data["items"]]
+    async def query_attempts(
+        self,
+        rollout_id: str,
+        *,
+        sort_by: Optional[str] = "sequence_id",
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> PaginatedResult[Attempt]:
+        params: List[Tuple[str, Any]] = [
+            ("limit", limit),
+            ("offset", offset),
+        ]
+        if sort_by is not None:
+            params.append(("sort_by", sort_by))
+            params.append(("sort_order", sort_order))
+        data = await self._request_json("get", f"/rollouts/{rollout_id}/attempts", params=params)
+        items = [Attempt.model_validate(item) for item in data["items"]]
+        return PaginatedResult(items=items, limit=data["limit"], offset=data["offset"], total=data["total"])
 
     async def get_latest_attempt(self, rollout_id: str) -> Optional[Attempt]:
         """
@@ -1420,15 +1494,34 @@ class LightningStoreClient(LightningStore):
             )
             return None
 
-    async def query_resources(self) -> List[ResourcesUpdate]:
+    async def query_resources(
+        self,
+        *,
+        resources_id: Optional[str] = None,
+        resources_id_contains: Optional[str] = None,
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> PaginatedResult[ResourcesUpdate]:
         """
         List all resource snapshots stored on the server.
         """
-        data = await self._request_json("get", "/resources")
-        if not data:
-            return []
-        # Extract items from PaginatedResponse
-        return [ResourcesUpdate.model_validate(item) for item in data["items"]]
+        params: List[Tuple[str, Any]] = [
+            ("limit", limit),
+            ("offset", offset),
+        ]
+        if sort_by is not None:
+            params.append(("sort_by", sort_by))
+            params.append(("sort_order", sort_order))
+        if resources_id is not None:
+            params.append(("resources_id", resources_id))
+        if resources_id_contains is not None:
+            params.append(("resources_id_contains", resources_id_contains))
+
+        data = await self._request_json("get", "/resources", params=params)
+        items = [ResourcesUpdate.model_validate(item) for item in data["items"]]
+        return PaginatedResult(items=items, limit=data["limit"], offset=data["offset"], total=data["total"])
 
     async def add_resources(self, resources: NamedResources) -> ResourcesUpdate:
         data = await self._request_json("post", "/resources", json=TypeAdapter(NamedResources).dump_python(resources))
@@ -1538,13 +1631,49 @@ class LightningStoreClient(LightningStore):
         self,
         rollout_id: str,
         attempt_id: str | Literal["latest"] | None = None,
-    ) -> List[Span]:
-        params: Dict[str, str] = {"rollout_id": rollout_id}
+        *,
+        trace_id: Optional[str] = None,
+        trace_id_contains: Optional[str] = None,
+        span_id: Optional[str] = None,
+        span_id_contains: Optional[str] = None,
+        parent_id: Optional[str] = None,
+        parent_id_contains: Optional[str] = None,
+        name: Optional[str] = None,
+        name_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        limit: int = -1,
+        offset: int = 0,
+        sort_by: Optional[str] = "sequence_id",
+        sort_order: Literal["asc", "desc"] = "asc",
+    ) -> PaginatedResult[Span]:
+        params: List[Tuple[str, Any]] = [("rollout_id", rollout_id)]
         if attempt_id is not None:
-            params["attempt_id"] = attempt_id
+            params.append(("attempt_id", attempt_id))
+        if trace_id is not None:
+            params.append(("trace_id", trace_id))
+        if trace_id_contains is not None:
+            params.append(("trace_id_contains", trace_id_contains))
+        if span_id is not None:
+            params.append(("span_id", span_id))
+        if span_id_contains is not None:
+            params.append(("span_id_contains", span_id_contains))
+        if parent_id is not None:
+            params.append(("parent_id", parent_id))
+        if parent_id_contains is not None:
+            params.append(("parent_id_contains", parent_id_contains))
+        if name is not None:
+            params.append(("name", name))
+        if name_contains is not None:
+            params.append(("name_contains", name_contains))
+        params.append(("filter_logic", filter_logic))
+        if sort_by is not None:
+            params.append(("sort_by", sort_by))
+            params.append(("sort_order", sort_order))
+        params.append(("limit", limit))
+        params.append(("offset", offset))
         data = await self._request_json("get", "/spans", params=params)
-        # Extract items from PaginatedResponse
-        return [Span.model_validate(item) for item in data["items"]]
+        items = [Span.model_validate(item) for item in data["items"]]
+        return PaginatedResult(items=items, limit=data["limit"], offset=data["offset"], total=data["total"])
 
     async def update_rollout(
         self,
@@ -1599,10 +1728,34 @@ class LightningStoreClient(LightningStore):
         )
         return Attempt.model_validate(data)
 
-    async def query_workers(self) -> List[Worker]:
-        data = await self._request_json("get", "/workers")
-        items = data.get("items", [])
-        return [Worker.model_validate(item) for item in items]
+    async def query_workers(
+        self,
+        *,
+        status_in: Optional[Sequence[WorkerStatus]] = None,
+        worker_id_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> PaginatedResult[Worker]:
+        params: List[Tuple[str, Any]] = [
+            ("limit", limit),
+            ("offset", offset),
+        ]
+        if status_in is not None:
+            for value in status_in:
+                params.append(("status_in", value))
+        if worker_id_contains is not None:
+            params.append(("worker_id_contains", worker_id_contains))
+        params.append(("filter_logic", filter_logic))
+        if sort_by is not None:
+            params.append(("sort_by", sort_by))
+            params.append(("sort_order", sort_order))
+
+        data = await self._request_json("get", "/workers", params=params)
+        items = [Worker.model_validate(item) for item in data.get("items", [])]
+        return PaginatedResult(items=items, limit=data["limit"], offset=data["offset"], total=data["total"])
 
     async def get_worker_by_id(self, worker_id: str) -> Optional[Worker]:
         data = await self._request_json("get", f"/workers/{worker_id}")

agentlightning/store/collection/base.py

@@ -6,23 +6,19 @@ from typing import (
     Any,
     AsyncContextManager,
     Generic,
-    Iterable,
-    Literal,
-    Mapping,
     Optional,
     Sequence,
     Type,
-    TypedDict,
     TypeVar,
-    Union,
 )
 
-from pydantic import BaseModel
-
 from agentlightning.types import (
     Attempt,
+    FilterOptions,
+    PaginatedResult,
     ResourcesUpdate,
     Rollout,
+    SortOptions,
     Span,
     Worker,
 )
@@ -32,68 +28,6 @@ K = TypeVar("K")
 V = TypeVar("V")
 
 
-class FilterField(TypedDict, total=False):
-    """An operator dict for a single field."""
-
-    exact: Any
-    within: Iterable[Any]
-    contains: str
-
-
-FilterOptions = Mapping[Union[str, Literal["_aggregate"]], Union[FilterField, Literal["and", "or"]]]
-
-"""A mapping of field name -> operator dict.
-
-Each operator dict can contain:
-
-- "exact": value for exact equality.
-- "within": iterable of allowed values.
-- "contains": substring to search for in string fields.
-
-Example:
-
-```json
-{
-    "_aggregate": "or",
-    "status": {"exact": "active"},
-    "id": {"within": [1, 2, 3]},
-    "name": {"contains": "foo"},
-}
-```
-
-The filter can also have a special field called "_aggregate" that can be used to specify the logic
-to combine the results of the filters:
-
-- "and": all conditions must match. This is the default value if not specified.
-- "or": at least one condition must match.
-
-All conditions within a field and between different fields are
-stored in a unified pool and combined using `_aggregate`.
-"""
-
-
-class SortOptions(TypedDict):
-    """Options for sorting the collection."""
-
-    name: str
-    """The name of the field to sort by."""
-    order: Literal["asc", "desc"]
-    """The order to sort by."""
-
-
-class PaginatedResult(BaseModel, Generic[T]):
-    """Result of a paginated query."""
-
-    items: Sequence[T]
-    """Items in the result."""
-    limit: int
-    """Limit of the result."""
-    offset: int
-    """Offset of the result."""
-    total: int
-    """Total number of items in the collection."""
-
-
 class Collection(Generic[T]):
     """Behaves like a list of items. Supporting addition, updating, and deletion of items."""
 
@@ -123,10 +57,10 @@ class Collection(Generic[T]):
 
         Args:
             filter:
-                The filters to apply to the collection. See [`FilterOptions`][agentlightning.store.collection.FilterOptions].
+                The filters to apply to the collection. See [`FilterOptions`][agentlightning.FilterOptions].
 
             sort:
-                The options for sorting the collection. See [`SortOptions`][agentlightning.store.collection.SortOptions].
+                The options for sorting the collection. See [`SortOptions`][agentlightning.SortOptions].
                 The field must exist in the model. If field might contain null values, in which case the behavior is undefined
                 (i.e., depending on the implementation).
 

agentlightning/store/collection/memory.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 import weakref
 from collections import deque
 from contextlib import asynccontextmanager
@@ -27,27 +28,29 @@ from typing import (
 
 from agentlightning.types import (
     Attempt,
+    FilterField,
+    FilterOptions,
+    PaginatedResult,
     ResourcesUpdate,
     Rollout,
+    SortOptions,
     Span,
     Worker,
 )
 
 from .base import (
     Collection,
-    FilterField,
-    FilterOptions,
     KeyValue,
     LightningCollections,
-    PaginatedResult,
     Queue,
-    SortOptions,
 )
 
 T = TypeVar("T")  # Recommended to be a BaseModel, not a dict
 K = TypeVar("K")
 V = TypeVar("V")
 
+logger = logging.getLogger(__name__)
+
 # Nested structure type:
 # dict[pk1] -> dict[pk2] -> ... -> item
 ListBasedCollectionItemType = Union[
@@ -59,6 +62,176 @@ FilterMap = Mapping[str, FilterField]
 MutationMode = Literal["insert", "update", "upsert", "delete"]
 
 
+def _merge_must_filters(target: Dict[str, FilterField], definition: Any) -> None:
+    """Normalize a `_must` filter group into the provided mapping.
+
+    Mainly for validation purposes.
+    """
+    if definition is None:
+        return
+
+    entries: List[Mapping[str, FilterField]] = []
+    if isinstance(definition, Mapping):
+        entries.append(cast(Mapping[str, FilterField], definition))
+    elif isinstance(definition, Sequence) and not isinstance(definition, (str, bytes)):
+        for entry in definition:  # type: ignore
+            if not isinstance(entry, Mapping):
+                raise TypeError("Each `_must` entry must be a mapping of field names to operators")
+            entries.append(cast(Mapping[str, FilterField], entry))
+    else:
+        raise TypeError("`_must` filters must be provided as a mapping or sequence of mappings")
+
+    for entry in entries:
+        for field_name, ops in entry.items():
+            existing = target.get(field_name, {})
+            merged_ops: Dict[str, Any] = dict(existing)
+            for op_name, expected in ops.items():
+                if op_name in merged_ops:
+                    raise ValueError(f"Duplicate operator '{op_name}' for field '{field_name}' in must filters")
+                merged_ops[op_name] = expected
+            target[field_name] = cast(FilterField, merged_ops)
+
+
+def _normalize_filter_options(
+    filter_options: Optional[FilterOptions],
+) -> Tuple[Optional[FilterMap], Optional[FilterMap], Literal["and", "or"]]:
+    """Convert FilterOptions to the internal structure and resolve aggregate logic."""
+    if not filter_options:
+        return None, None, "and"
+
+    aggregate = cast(Literal["and", "or"], filter_options.get("_aggregate", "and"))
+    if aggregate not in ("and", "or"):
+        raise ValueError(f"Unsupported filter aggregate '{aggregate}'")
+
+    # Extract normalized filters and must filters from the filter options.
+    normalized: Dict[str, FilterField] = {}
+    must_filters: Dict[str, FilterField] = {}
+    for field_name, ops in filter_options.items():
+        if field_name == "_aggregate":
+            continue
+        if field_name == "_must":
+            _merge_must_filters(must_filters, ops)
+            continue
+        normalized[field_name] = cast(FilterField, dict(ops))  # type: ignore
+
+    return (normalized or None, must_filters or None, aggregate)
+
+
+def _resolve_sort_options(sort: Optional[SortOptions]) -> Tuple[Optional[str], Literal["asc", "desc"]]:
+    """Extract sort field/order from the caller-provided SortOptions."""
+    if not sort:
+        return None, "asc"
+
+    sort_name = sort.get("name")
+    if not sort_name:
+        raise ValueError("Sort options must include a 'name' field")
+
+    sort_order = sort.get("order", "asc")
+    if sort_order not in ("asc", "desc"):
+        raise ValueError(f"Unsupported sort order '{sort_order}'")
+
+    return sort_name, sort_order
+
+
+def _item_matches_filters(
+    item: object,
+    filters: Optional[FilterMap],
+    filter_logic: Literal["and", "or"],
+    must_filters: Optional[FilterMap] = None,
+) -> bool:
+    """Check whether an item matches the provided filter definition.
+
+    Filter format:
+
+    ```json
+    {
+        "_aggregate": "or",
+        "field_name": {
+            "exact": <value>,
+            "within": <iterable_of_allowed_values>,
+            "contains": <substring_or_element>,
+        },
+        ...
+    }
+    ```
+
+    Operators within the same field are stored in a unified pool and combined using
+    a universal logical operator.
+    """
+    if must_filters and not _item_matches_filters(item, must_filters, "and"):
+        return False
+
+    if not filters:
+        return True
+
+    all_conditions_match: List[bool] = []
+
+    for field_name, ops in filters.items():
+        item_value = getattr(item, field_name, None)
+
+        for op_name, expected in ops.items():
+            # Ignore no-op filters
+            if expected is None:
+                continue
+
+            if op_name == "exact":
+                all_conditions_match.append(item_value == expected)
+
+            elif op_name == "within":
+                try:
+                    all_conditions_match.append(item_value in expected)  # type: ignore[arg-type]
+                except TypeError:
+                    all_conditions_match.append(False)
+
+            elif op_name == "contains":
+                if item_value is None:
+                    all_conditions_match.append(False)
+                elif isinstance(item_value, str) and isinstance(expected, str):
+                    all_conditions_match.append(expected in item_value)
+                else:
+                    # Fallback: treat as generic iterable containment.
+                    try:
+                        all_conditions_match.append(expected in item_value)  # type: ignore[arg-type]
+                    except TypeError:
+                        all_conditions_match.append(False)
+            else:
+                raise ValueError(f"Unsupported filter operator '{op_name}' for field '{field_name}'")
+
+    return all(all_conditions_match) if filter_logic == "and" else any(all_conditions_match)
+
+
+def _get_sort_value(item: object, sort_by: str) -> Any:
+    """Get a sort key for the given item/field.
+
+    - If the field name ends with '_time', values are treated as comparable timestamps.
+    - For other fields we try to infer a safe default from the Pydantic model annotation.
+    """
+    value = getattr(item, sort_by, None)
+
+    if sort_by.endswith("_time"):
+        # For *_time fields, push missing values to the end.
+        return float("inf") if value is None else value
+
+    if value is None:
+        # Introspect model field type to choose a reasonable default for None.
+        model_fields = getattr(item.__class__, "model_fields", {})
+        if sort_by not in model_fields:
+            raise ValueError(
+                f"Failed to sort items by '{sort_by}': field does not exist " f"on {item.__class__.__name__}"
+            )
+
+        field_type_str = str(model_fields[sort_by].annotation)
+        if "str" in field_type_str or "Literal" in field_type_str:
+            return ""
+        if "int" in field_type_str:
+            return 0
+        if "float" in field_type_str:
+            return 0.0
+        raise ValueError(f"Failed to sort items by '{sort_by}': unsupported field type {field_type_str!r}")
+
+    return value
+
+
 class ListBasedCollection(Collection[T]):
     """In-memory implementation of Collection using a nested dict for O(1) primary-key lookup.
 
@@ -219,46 +392,11 @@ class ListBasedCollection(Collection[T]):
         else:
             raise ValueError(f"Unknown mutation mode: {mode}")
 
-    @staticmethod
-    def _normalize_filter_options(
-        filter_options: Optional[FilterOptions],
-    ) -> Tuple[Optional[FilterMap], Literal["and", "or"]]:
-        """Convert FilterOptions to the internal structure and resolve aggregate logic."""
-        if not filter_options:
-            return None, "and"
-
-        aggregate = cast(Literal["and", "or"], filter_options.get("_aggregate", "and"))
-        if aggregate not in ("and", "or"):
-            raise ValueError(f"Unsupported filter aggregate '{aggregate}'")
-
-        normalized: Dict[str, FilterField] = {}
-        for field_name, ops in filter_options.items():
-            if field_name == "_aggregate":
-                continue
-            normalized[field_name] = cast(FilterField, ops)
-
-        return (normalized or None, aggregate)
-
-    @staticmethod
-    def _resolve_sort_options(sort: Optional[SortOptions]) -> Tuple[Optional[str], Literal["asc", "desc"]]:
-        """Extract sort field/order from the caller-provided SortOptions."""
-        if not sort:
-            return None, "asc"
-
-        sort_name = sort.get("name")
-        if not sort_name:
-            raise ValueError("Sort options must include a 'name' field")
-
-        sort_order = sort.get("order", "asc")
-        if sort_order not in ("asc", "desc"):
-            raise ValueError(f"Unsupported sort order '{sort_order}'")
-
-        return sort_name, sort_order
-
     def _iter_items(
         self,
         root: Optional[Mapping[Any, Any]] = None,
         filters: Optional[FilterMap] = None,
+        must_filters: Optional[FilterMap] = None,
         filter_logic: Literal["and", "or"] = "and",
     ) -> Iterable[T]:
         """Iterate over all items in the nested dictionary structure, optionally applying filters."""
@@ -272,7 +410,7 @@ class ListBasedCollection(Collection[T]):
             for value in node.values():
                 # Leaf nodes contain items; intermediate nodes are dicts.
                 if isinstance(value, self._item_type):
-                    if self._item_matches_filters(value, filters, filter_logic):
+                    if _item_matches_filters(value, filters, filter_logic, must_filters):
                         yield value
                 elif isinstance(value, dict):
                     stack.append(value)  # type: ignore
@@ -285,37 +423,69 @@ class ListBasedCollection(Collection[T]):
     def _iter_matching_items(
         self,
         filters: Optional[FilterMap],
+        must_filters: Optional[FilterMap],
         filter_logic: Literal["and", "or"],
     ) -> Iterable[T]:
         """Efficiently iterate over items matching filters, using primary-key prefix when possible."""
-        # Fast path: no filters or non-AND logic -> just delegate to _iter_items.
-        if not filters or filter_logic != "and":
-            return self._iter_items(filters=filters, filter_logic=filter_logic)
+        # Fast path: when optional filters can't form a prefix, fall back to scanning.
+        if filter_logic != "and" and must_filters is None:
+            return self._iter_items(filters=filters, must_filters=must_filters, filter_logic=filter_logic)
 
         # Try to derive a primary-key prefix from exact filters.
         pk_values_prefix: List[Any] = []
+        prefix_sources: List[FilterMap] = []
+        if must_filters:
+            prefix_sources.append(must_filters)
+        if filter_logic == "and" and filters:
+            prefix_sources.append(filters)
+
         for pk in self._primary_keys:
-            field_ops = filters.get(pk)  # type: ignore[union-attr]
-            if not field_ops:
+            # combined_ops are: [{"exact": value}, {"within": [...]}, ...]
+            combined_ops: List[FilterField] = []
+            for source in prefix_sources:
+                field_ops = source.get(pk)  # type: ignore[union-attr]
+                if field_ops:
+                    combined_ops.append(field_ops)
+            if not combined_ops:
                 break
             # Only allow a pure {"exact": value} constraint.
-            if set(field_ops.keys()) != {"exact"}:
+            exact_value: Any | None = None
+            allow_prefix = True
+            for ops in combined_ops:
+                if set(ops.keys()) != {"exact"}:
+                    allow_prefix = False
+                    break
+                candidate = ops.get("exact")
+                if candidate is None:
+                    allow_prefix = False
+                    break
+                if exact_value is not None and candidate != exact_value:
+                    # Contradictory exact filters mean no items can match.
+                    logger.warning(f"Contradictory exact filters for field '{pk}': {exact_value} != {candidate}")
+                    return ()
+                exact_value = candidate
+
+            if not allow_prefix:
                 break
-            value = field_ops.get("exact")
+
+            value = exact_value
             if value is None:
                 break
             pk_values_prefix.append(value)
 
         if not pk_values_prefix:
-            return self._iter_items(filters=filters, filter_logic=filter_logic)
+            return self._iter_items(filters=filters, must_filters=must_filters, filter_logic=filter_logic)
 
         try:
             if len(pk_values_prefix) == len(self._primary_keys):
                 # All primary keys specified -> at most a single item.
                 parent, final_key = self._locate_node(pk_values_prefix, create_missing=False)
                 single_item = parent.get(final_key)
-                if isinstance(single_item, self._item_type) and self._item_matches_filters(
-                    single_item, filters, filter_logic
+                if isinstance(single_item, self._item_type) and _item_matches_filters(
+                    single_item,
+                    filters,
+                    filter_logic,
+                    must_filters,
                 ):
                     return (single_item,)
                 return ()
@@ -324,107 +494,17 @@ class ListBasedCollection(Collection[T]):
                 parent, final_key = self._locate_node(pk_values_prefix, create_missing=False)
                 subtree = parent.get(final_key)
                 if isinstance(subtree, dict):
-                    return self._iter_items(subtree, filters=filters, filter_logic=filter_logic)  # type: ignore
+                    return self._iter_items(
+                        subtree,  # type: ignore
+                        filters=filters,
+                        must_filters=must_filters,
+                        filter_logic=filter_logic,
+                    )
                 return ()
         except KeyError:
             # No items exist for this primary-key prefix.
             return ()
 
-    @staticmethod
-    def _item_matches_filters(
-        item: T,
-        filters: Optional[FilterMap],
-        filter_logic: Literal["and", "or"],
-    ) -> bool:
-        """Check whether an item matches the provided filter definition.
-
-        Filter format:
-
-        ```json
-        {
-            "_aggregate": "or",
-            "field_name": {
-                "exact": <value>,
-                "within": <iterable_of_allowed_values>,
-                "contains": <substring_or_element>,
-            },
-            ...
-        }
-        ```
-
-        Operators within the same field are stored in a unified pool and combined using
-        a universal logical operator.
-        """
-        if not filters:
-            return True
-
-        all_conditions_match: List[bool] = []
-
-        for field_name, ops in filters.items():
-            item_value = getattr(item, field_name, None)
-
-            for op_name, expected in ops.items():
-                # Ignore no-op filters
-                if expected is None:
-                    continue
-
-                if op_name == "exact":
-                    all_conditions_match.append(item_value == expected)
-
-                elif op_name == "within":
-                    try:
-                        all_conditions_match.append(item_value in expected)  # type: ignore[arg-type]
-                    except TypeError:
-                        all_conditions_match.append(False)
-
-                elif op_name == "contains":
-                    if item_value is None:
-                        all_conditions_match.append(False)
-                    elif isinstance(item_value, str) and isinstance(expected, str):
-                        all_conditions_match.append(expected in item_value)
-                    else:
-                        # Fallback: treat as generic iterable containment.
-                        try:
-                            all_conditions_match.append(expected in item_value)  # type: ignore[arg-type]
-                        except TypeError:
-                            all_conditions_match.append(False)
-                else:
-                    raise ValueError(f"Unsupported filter operator '{op_name}' for field '{field_name}'")
-
-        return all(all_conditions_match) if filter_logic == "and" else any(all_conditions_match)
-
-    @staticmethod
-    def _get_sort_value(item: T, sort_by: str) -> Any:
-        """Get a sort key for the given item/field.
-
-        - If the field name ends with '_time', values are treated as comparable timestamps.
-        - For other fields we try to infer a safe default from the Pydantic model annotation.
-        """
-        value = getattr(item, sort_by, None)
-
-        if sort_by.endswith("_time"):
-            # For *_time fields, push missing values to the end.
-            return float("inf") if value is None else value
-
-        if value is None:
-            # Introspect model field type to choose a reasonable default for None.
-            model_fields = getattr(item.__class__, "model_fields", {})
-            if sort_by not in model_fields:
-                raise ValueError(
-                    f"Failed to sort items by '{sort_by}': field does not exist " f"on {item.__class__.__name__}"
-                )
-
-            field_type_str = str(model_fields[sort_by].annotation)
-            if "str" in field_type_str or "Literal" in field_type_str:
-                return ""
-            if "int" in field_type_str:
-                return 0
-            if "float" in field_type_str:
-                return 0.0
-            raise ValueError(f"Failed to sort items by '{sort_by}': unsupported field type {field_type_str!r}")
-
-        return value
-
     async def query(
         self,
         filter: Optional[FilterOptions] = None,
@@ -440,9 +520,9 @@ class ListBasedCollection(Collection[T]):
             limit: Max number of items to return. Use -1 for "no limit".
             offset: Number of items to skip from the start of the *matching* items.
         """
-        filters, filter_logic = self._normalize_filter_options(filter)
-        sort_by, sort_order = self._resolve_sort_options(sort)
-        items_iter: Iterable[T] = self._iter_matching_items(filters, filter_logic)
+        filters, must_filters, filter_logic = _normalize_filter_options(filter)
+        sort_by, sort_order = _resolve_sort_options(sort)
+        items_iter: Iterable[T] = self._iter_matching_items(filters, must_filters, filter_logic)
 
         # No sorting: stream through items and apply pagination on the fly.
         if not sort_by:
@@ -474,7 +554,7 @@ class ListBasedCollection(Collection[T]):
 
         total_matched = len(all_matches)
         reverse = sort_order == "desc"
-        all_matches.sort(key=lambda x: self._get_sort_value(x, sort_by), reverse=reverse)
+        all_matches.sort(key=lambda x: _get_sort_value(x, sort_by), reverse=reverse)
 
         if limit == -1:
             paginated_items = all_matches[offset:]
@@ -494,9 +574,9 @@ class ListBasedCollection(Collection[T]):
         sort: Optional[SortOptions] = None,
     ) -> Optional[T]:
         """Return the first (or best-sorted) item that matches the given filters, or None."""
-        filters, filter_logic = self._normalize_filter_options(filter)
-        sort_by, sort_order = self._resolve_sort_options(sort)
-        items_iter: Iterable[T] = self._iter_matching_items(filters, filter_logic)
+        filters, must_filters, filter_logic = _normalize_filter_options(filter)
+        sort_by, sort_order = _resolve_sort_options(sort)
+        items_iter: Iterable[T] = self._iter_matching_items(filters, must_filters, filter_logic)
 
         if not sort_by:
             # Just return the first matching item, if any.
@@ -509,7 +589,7 @@ class ListBasedCollection(Collection[T]):
         best_key: Any = None
 
         for item in items_iter:
-            key = self._get_sort_value(item, sort_by)
+            key = _get_sort_value(item, sort_by)
             if best_item is None:
                 best_item = item
                 best_key = key

agentlightning/store/collection_based.py

@@ -8,6 +8,7 @@ import hashlib
 import logging
 import time
 import uuid
+import warnings
 from typing import (
     Any,
     Callable,
@@ -29,14 +30,18 @@ from agentlightning.types import (
     Attempt,
     AttemptedRollout,
     AttemptStatus,
+    FilterField,
     NamedResources,
+    PaginatedResult,
     ResourcesUpdate,
     Rollout,
     RolloutConfig,
     RolloutStatus,
+    SortOptions,
     Span,
     TaskInput,
     Worker,
+    WorkerStatus,
 )
 
 from .base import UNSET, LightningStore, LightningStoreCapabilities, Unset, is_finished, is_queuing
@@ -380,22 +385,58 @@ class CollectionBasedLightningStore(LightningStore, Generic[T_collections]):
 
     @_healthcheck_wrapper
     async def query_rollouts(
-        self, *, status: Optional[Sequence[RolloutStatus]] = None, rollout_ids: Optional[Sequence[str]] = None
-    ) -> List[Rollout]:
-        """Retrieves rollouts filtered by their status and rollout ids.
-        If no status is provided, returns all rollouts.
+        self,
+        *,
+        status_in: Optional[Sequence[RolloutStatus]] = None,
+        rollout_id_in: Optional[Sequence[str]] = None,
+        rollout_id_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+        status: Optional[Sequence[RolloutStatus]] = None,
+        rollout_ids: Optional[Sequence[str]] = None,
+    ) -> PaginatedResult[Union[Rollout, AttemptedRollout]]:
+        """Retrieve rollouts with filtering and pagination.
 
         See [`LightningStore.query_rollouts()`][agentlightning.LightningStore.query_rollouts] for semantics.
         """
         # Construct filters condition
+        if status_in is not None:
+            resolved_status = status_in
+        elif status is not None:
+            warnings.warn("status is deprecated, use status_in instead", DeprecationWarning, stacklevel=3)
+            resolved_status = status
+        else:
+            resolved_status = None
+
+        if rollout_id_in is not None:
+            resolved_rollout_ids = rollout_id_in
+        elif rollout_ids is not None:
+            warnings.warn("rollout_ids is deprecated, use rollout_id_in instead", DeprecationWarning, stacklevel=3)
+            resolved_rollout_ids = rollout_ids
+        else:
+            resolved_rollout_ids = None
+
         filters: FilterOptions = {}
-        if rollout_ids is not None:
-            filters["rollout_id"] = {"within": list(rollout_ids)}
-        if status is not None:
-            filters["status"] = {"within": list(status)}
+        filters["_aggregate"] = filter_logic
+        if resolved_status is not None:
+            filters["status"] = {"within": list(resolved_status)}
+        if resolved_rollout_ids is not None:
+            rollout_id_field = cast(FilterField, filters.setdefault("rollout_id", {}))
+            rollout_id_field["within"] = list(resolved_rollout_ids)
+        if rollout_id_contains is not None:
+            rollout_id_field = cast(FilterField, filters.setdefault("rollout_id", {}))
+            rollout_id_field["contains"] = rollout_id_contains
 
         async with self.collections.atomic():
-            rollouts = await self.collections.rollouts.query(filter=filters or None)
+            rollouts = await self.collections.rollouts.query(
+                filter=filters if list(filters.keys()) != ["_aggregate"] else None,
+                sort=SortOptions(name=sort_by, order=sort_order) if sort_by else None,
+                limit=limit,
+                offset=offset,
+            )
 
             # Attach the latest attempt to the rollout objects
             # TODO: Maybe we can use asyncio.gather here to speed up the process?
@@ -403,7 +444,9 @@ class CollectionBasedLightningStore(LightningStore, Generic[T_collections]):
                 await self._rollout_to_attempted_rollout_unlocked(rollout) for rollout in rollouts.items
             ]
 
-            return attempted_rollouts
+            return PaginatedResult(
+                items=attempted_rollouts, limit=rollouts.limit, offset=rollouts.offset, total=rollouts.total
+            )
 
     async def _query_attempts_for_rollout_unlocked(self, rollout_id: str) -> List[Attempt]:
         """The unlocked version of `query_attempts_for_rollout`."""
@@ -446,18 +489,23 @@ class CollectionBasedLightningStore(LightningStore, Generic[T_collections]):
         )
 
     @_healthcheck_wrapper
-    async def query_attempts(self, rollout_id: str) -> List[Attempt]:
-        """Retrieves all attempts associated with a specific rollout ID.
-        Returns an empty list if no attempts are found.
-
-        See [`LightningStore.query_attempts()`][agentlightning.LightningStore.query_attempts] for semantics.
-        """
+    async def query_attempts(
+        self,
+        rollout_id: str,
+        *,
+        sort_by: Optional[str] = "sequence_id",
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> PaginatedResult[Attempt]:
+        """Retrieve attempts for a rollout with optional ordering/pagination."""
         async with self.collections.atomic():
-            result = await self.collections.attempts.query(
+            return await self.collections.attempts.query(
                 filter={"rollout_id": {"exact": rollout_id}},
-                sort={"name": "sequence_id", "order": "asc"},
+                sort={"name": sort_by, "order": sort_order} if sort_by else None,
+                limit=limit,
+                offset=offset,
             )
-            return list(result.items)
 
     @_healthcheck_wrapper
     async def get_latest_attempt(self, rollout_id: str) -> Optional[Attempt]:
@@ -469,12 +517,32 @@ class CollectionBasedLightningStore(LightningStore, Generic[T_collections]):
             return await self._get_latest_attempt_unlocked(rollout_id)
 
     @_healthcheck_wrapper
-    async def query_resources(self) -> List[ResourcesUpdate]:
+    async def query_resources(
+        self,
+        *,
+        resources_id: Optional[str] = None,
+        resources_id_contains: Optional[str] = None,
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> PaginatedResult[ResourcesUpdate]:
         """Return every stored resource snapshot in insertion order."""
+        filters: FilterOptions = {}
+        if resources_id is not None:
+            resources_id_field = cast(FilterField, filters.setdefault("resources_id", {}))
+            resources_id_field["exact"] = resources_id
+        if resources_id_contains is not None:
+            resources_id_field = cast(FilterField, filters.setdefault("resources_id", {}))
+            resources_id_field["contains"] = resources_id_contains
+
         async with self.collections.atomic():
-            # No sorting and no pagination by default
-            result = await self.collections.resources.query()
-            return list(result.items)
+            return await self.collections.resources.query(
+                filter=filters or None,
+                sort={"name": sort_by, "order": sort_order} if sort_by else None,
+                limit=limit,
+                offset=offset,
+            )
 
     @_healthcheck_wrapper
     async def add_resources(self, resources: NamedResources) -> ResourcesUpdate:
@@ -709,16 +777,36 @@ class CollectionBasedLightningStore(LightningStore, Generic[T_collections]):
         return None
 
     @_healthcheck_wrapper
-    async def query_spans(self, rollout_id: str, attempt_id: str | Literal["latest"] | None = None) -> List[Span]:
+    async def query_spans(
+        self,
+        rollout_id: str,
+        attempt_id: str | Literal["latest"] | None = None,
+        *,
+        trace_id: Optional[str] = None,
+        trace_id_contains: Optional[str] = None,
+        span_id: Optional[str] = None,
+        span_id_contains: Optional[str] = None,
+        parent_id: Optional[str] = None,
+        parent_id_contains: Optional[str] = None,
+        name: Optional[str] = None,
+        name_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        limit: int = -1,
+        offset: int = 0,
+        sort_by: Optional[str] = "sequence_id",
+        sort_order: Literal["asc", "desc"] = "asc",
+    ) -> PaginatedResult[Span]:
         """
-        Query and retrieve all spans associated with a specific rollout ID.
+        Query and retrieve spans associated with a specific rollout ID.
         Returns an empty list if no spans are found.
 
         See [`LightningStore.query_spans()`][agentlightning.LightningStore.query_spans] for semantics.
         """
+
         async with self.collections.atomic():
+            resolved_attempt_id: Optional[str]
             if attempt_id is None:
-                spans = await self.collections.spans.query(filter={"rollout_id": {"exact": rollout_id}})
+                resolved_attempt_id = None
             elif attempt_id == "latest":
                 latest_attempt = await self.collections.attempts.get(
                     filter={"rollout_id": {"exact": rollout_id}},
@@ -726,18 +814,39 @@ class CollectionBasedLightningStore(LightningStore, Generic[T_collections]):
                 )
                 if not latest_attempt:
                     logger.debug(f"No attempts found for rollout {rollout_id} when querying latest spans")
-                    return []
-                spans = await self.collections.spans.query(
-                    filter={
-                        "rollout_id": {"exact": rollout_id},
-                        "attempt_id": {"exact": latest_attempt.attempt_id},
-                    }
-                )
+                    return PaginatedResult(items=[], limit=limit, offset=offset, total=0)
+                resolved_attempt_id = latest_attempt.attempt_id
             else:
-                spans = await self.collections.spans.query(
-                    filter={"rollout_id": {"exact": rollout_id}, "attempt_id": {"exact": attempt_id}}
-                )
-            return list(spans.items)
+                resolved_attempt_id = attempt_id
+
+            must_filter: Dict[str, FilterField] = {"rollout_id": {"exact": rollout_id}}
+            if resolved_attempt_id is not None:
+                must_filter["attempt_id"] = {"exact": resolved_attempt_id}
+            filter_options: FilterOptions = {
+                "_aggregate": filter_logic,  # this can be and/or
+                "_must": must_filter,  # Must satisfy all the filters in the must list
+            }
+
+            def _resolve_filter_field(
+                field_name: str, filter_exact: Optional[str] | None, filter_contains: Optional[str] | None
+            ) -> None:
+                field = cast(FilterField, filter_options.setdefault(field_name, {}))
+                if filter_exact is not None:
+                    field["exact"] = filter_exact
+                if filter_contains is not None:
+                    field["contains"] = filter_contains
+
+            _resolve_filter_field("trace_id", trace_id, trace_id_contains)
+            _resolve_filter_field("span_id", span_id, span_id_contains)
+            _resolve_filter_field("parent_id", parent_id, parent_id_contains)
+            _resolve_filter_field("name", name, name_contains)
+
+            return await self.collections.spans.query(
+                filter=filter_options,
+                sort={"name": sort_by, "order": sort_order} if sort_by else None,
+                limit=limit,
+                offset=offset,
+            )
 
     @_healthcheck_wrapper
     async def update_rollout(
@@ -911,11 +1020,32 @@ class CollectionBasedLightningStore(LightningStore, Generic[T_collections]):
         return attempt
 
     @_healthcheck_wrapper
-    async def query_workers(self) -> List[Worker]:
+    async def query_workers(
+        self,
+        *,
+        status_in: Optional[Sequence[WorkerStatus]] = None,
+        worker_id_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> PaginatedResult[Worker]:
         """Return the current snapshot of all workers."""
+        filters: FilterOptions = {}
+        if status_in is not None:
+            filters["status"] = {"within": list(status_in)}
+        if worker_id_contains is not None:
+            filters["worker_id"] = {"contains": worker_id_contains}
+        filters["_aggregate"] = filter_logic
+
         async with self.collections.atomic():
-            result = await self.collections.workers.query()
-            return list(result.items)
+            return await self.collections.workers.query(
+                filter=filters if list(filters.keys()) != ["_aggregate"] else None,
+                sort={"name": sort_by, "order": sort_order} if sort_by else None,
+                limit=limit,
+                offset=offset,
+            )
 
     @_healthcheck_wrapper
     async def get_worker_by_id(self, worker_id: str) -> Optional[Worker]:

agentlightning/store/memory.py

@@ -24,11 +24,7 @@ from typing import (
 
 from pydantic import BaseModel
 
-from agentlightning.types import (
-    AttemptedRollout,
-    Rollout,
-    Span,
-)
+from agentlightning.types import AttemptedRollout, PaginatedResult, Rollout, Span
 
 from .base import LightningStoreCapabilities, is_finished, is_running
 from .collection import InMemoryLightningCollections
@@ -198,7 +194,9 @@ class InMemoryLightningStore(CollectionBasedLightningStore[InMemoryLightningColl
 
     async def get_running_rollouts(self) -> List[AttemptedRollout]:
         """Accelerated version of `get_running_rollouts` for in-memory store. Used for healthcheck."""
-        rollouts = await self.collections.rollouts.query(filter={"rollout_id": {"within": self._running_rollout_ids}})
+        rollouts = await self.collections.rollouts.query(
+            filter={"rollout_id": {"within": list(self._running_rollout_ids)}}
+        )
         running_rollouts: List[AttemptedRollout] = []
         for rollout in rollouts.items:
             latest_attempt = await self.collections.attempts.get(
@@ -212,10 +210,15 @@ class InMemoryLightningStore(CollectionBasedLightningStore[InMemoryLightningColl
             running_rollouts.append(AttemptedRollout(**rollout.model_dump(), attempt=latest_attempt))
         return running_rollouts
 
-    async def query_spans(self, rollout_id: str, attempt_id: str | Literal["latest"] | None = None) -> List[Span]:
+    async def query_spans(
+        self,
+        rollout_id: str,
+        attempt_id: str | Literal["latest"] | None = None,
+        **kwargs: Any,
+    ) -> PaginatedResult[Span]:
         if rollout_id in self._evicted_rollout_span_sets:
             raise RuntimeError(f"Spans for rollout {rollout_id} have been evicted")
-        return await super().query_spans(rollout_id, attempt_id)
+        return await super().query_spans(rollout_id, attempt_id, **kwargs)
 
     async def _add_span_unlocked(self, span: Span) -> Span:
         """In-memory store needs to maintain the span data in memory, and evict spans when memory is low."""

agentlightning/store/threading.py

@@ -19,6 +19,7 @@ from agentlightning.types import (
     Span,
     TaskInput,
     Worker,
+    WorkerStatus,
 )
 
 from .base import UNSET, LightningStore, LightningStoreCapabilities, Unset
@@ -79,15 +80,48 @@ class LightningStoreThreaded(LightningStore):
     async def query_rollouts(
         self,
         *,
+        status_in: Optional[Sequence[RolloutStatus]] = None,
+        rollout_id_in: Optional[Sequence[str]] = None,
+        rollout_id_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
         status: Optional[Sequence[RolloutStatus]] = None,
         rollout_ids: Optional[Sequence[str]] = None,
-    ) -> List[Rollout]:
+    ) -> Sequence[Rollout]:
         with self._lock:
-            return await self.store.query_rollouts(status=status, rollout_ids=rollout_ids)
+            return await self.store.query_rollouts(
+                status_in=status_in,
+                rollout_id_in=rollout_id_in,
+                rollout_id_contains=rollout_id_contains,
+                filter_logic=filter_logic,
+                sort_by=sort_by,
+                sort_order=sort_order,
+                limit=limit,
+                offset=offset,
+                status=status,
+                rollout_ids=rollout_ids,
+            )
 
-    async def query_attempts(self, rollout_id: str) -> List[Attempt]:
+    async def query_attempts(
+        self,
+        rollout_id: str,
+        *,
+        sort_by: Optional[str] = "sequence_id",
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> Sequence[Attempt]:
         with self._lock:
-            return await self.store.query_attempts(rollout_id)
+            return await self.store.query_attempts(
+                rollout_id,
+                sort_by=sort_by,
+                sort_order=sort_order,
+                limit=limit,
+                offset=offset,
+            )
 
     async def get_rollout_by_id(self, rollout_id: str) -> Optional[Rollout]:
         with self._lock:
@@ -97,6 +131,26 @@ class LightningStoreThreaded(LightningStore):
         with self._lock:
             return await self.store.get_latest_attempt(rollout_id)
 
+    async def query_resources(
+        self,
+        *,
+        resources_id: Optional[str] = None,
+        resources_id_contains: Optional[str] = None,
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> Sequence[ResourcesUpdate]:
+        with self._lock:
+            return await self.store.query_resources(
+                resources_id=resources_id,
+                resources_id_contains=resources_id_contains,
+                sort_by=sort_by,
+                sort_order=sort_order,
+                limit=limit,
+                offset=offset,
+            )
+
     async def add_resources(self, resources: NamedResources) -> ResourcesUpdate:
         with self._lock:
             return await self.store.add_resources(resources)
@@ -139,9 +193,39 @@ class LightningStoreThreaded(LightningStore):
         self,
         rollout_id: str,
         attempt_id: str | Literal["latest"] | None = None,
-    ) -> List[Span]:
+        *,
+        trace_id: Optional[str] = None,
+        trace_id_contains: Optional[str] = None,
+        span_id: Optional[str] = None,
+        span_id_contains: Optional[str] = None,
+        parent_id: Optional[str] = None,
+        parent_id_contains: Optional[str] = None,
+        name: Optional[str] = None,
+        name_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        limit: int = -1,
+        offset: int = 0,
+        sort_by: Optional[str] = "sequence_id",
+        sort_order: Literal["asc", "desc"] = "asc",
+    ) -> Sequence[Span]:
         with self._lock:
-            return await self.store.query_spans(rollout_id, attempt_id)
+            return await self.store.query_spans(
+                rollout_id,
+                attempt_id,
+                trace_id=trace_id,
+                trace_id_contains=trace_id_contains,
+                span_id=span_id,
+                span_id_contains=span_id_contains,
+                parent_id=parent_id,
+                parent_id_contains=parent_id_contains,
+                name=name,
+                name_contains=name_contains,
+                filter_logic=filter_logic,
+                limit=limit,
+                offset=offset,
+                sort_by=sort_by,
+                sort_order=sort_order,
+            )
 
     async def update_rollout(
         self,
@@ -183,9 +267,26 @@ class LightningStoreThreaded(LightningStore):
                 metadata=metadata,
             )
 
-    async def query_workers(self) -> List[Worker]:
+    async def query_workers(
+        self,
+        *,
+        status_in: Optional[Sequence[WorkerStatus]] = None,
+        worker_id_contains: Optional[str] = None,
+        filter_logic: Literal["and", "or"] = "and",
+        sort_by: Optional[str] = None,
+        sort_order: Literal["asc", "desc"] = "asc",
+        limit: int = -1,
+        offset: int = 0,
+    ) -> Sequence[Worker]:
         with self._lock:
-            return await self.store.query_workers()
+            return await self.store.query_workers(
+                status_in=status_in,
+                worker_id_contains=worker_id_contains,
+                sort_by=sort_by,
+                sort_order=sort_order,
+                limit=limit,
+                offset=offset,
+            )
 
     async def get_worker_by_id(self, worker_id: str) -> Optional[Worker]:
         with self._lock:

agentlightning/types/core.py

@@ -10,14 +10,19 @@ from typing import (
     Callable,
     Dict,
     Generic,
+    Iterator,
     List,
     Literal,
+    Mapping,
     Optional,
     Protocol,
+    Sequence,
     SupportsIndex,
+    TypedDict,
     TypeVar,
     Union,
     cast,
+    overload,
 )
 
 from opentelemetry.sdk.trace import ReadableSpan
@@ -51,6 +56,10 @@ __all__ = [
     "Hook",
     "Worker",
     "WorkerStatus",
+    "PaginatedResult",
+    "FilterOptions",
+    "SortOptions",
+    "FilterField",
 ]
 
 T_co = TypeVar("T_co", covariant=True)
@@ -421,3 +430,104 @@ class Hook(ParallelWorkerBase):
         Subclasses can override this method for cleanup or additional
         logging. By default, this is a no-op.
         """
+
+
+class FilterField(TypedDict, total=False):
+    """An operator dict for a single field."""
+
+    exact: Any
+    within: Sequence[Any]
+    contains: str
+
+
+FilterOptions = Mapping[
+    Union[str, Literal["_aggregate", "_must"]],
+    Union[FilterField, Literal["and", "or"], Mapping[str, FilterField]],
+]
+"""A mapping of field name -> operator dict.
+
+Each operator dict can contain:
+
+- "exact": value for exact equality.
+- "within": iterable of allowed values.
+- "contains": substring to search for in string fields.
+
+The filter can also have a special field called "_aggregate" that can be used to specify the logic
+to combine the results of the filters:
+
+- "and": all conditions must match. This is the default value if not specified.
+- "or": at least one condition must match.
+
+All conditions within a field and between different fields are
+stored in a unified pool and combined using `_aggregate`.
+
+The filter can also have a special group called "_must", which is a mapping of filters that must all match,
+no matter whether the aggregate logic is "and" or "or".
+
+Example:
+
+```json
+{
+    "_aggregate": "or",
+    "_must": {
+        "city": {"exact": "New York"},
+        "timezone": {"within": ["America/New_York", "America/Los_Angeles"]},
+    },
+    "status": {"exact": "active"},
+    "id": {"within": [1, 2, 3]},
+    "name": {"contains": "foo"},
+}
+```
+"""
+
+
+class SortOptions(TypedDict):
+    """Options for sorting the collection."""
+
+    name: str
+    """The name of the field to sort by."""
+    order: Literal["asc", "desc"]
+    """The order to sort by."""
+
+
+T_item = TypeVar("T_item")
+
+
+class PaginatedResult(BaseModel, Sequence[T_item]):
+    """Result of a paginated query.
+
+    Behaves like a sequence, but also carries pagination metadata (limit, offset, total).
+    """
+
+    items: Sequence[T_item]
+    """Items in the result."""
+    limit: int
+    """Limit of the result."""
+    offset: int
+    """Offset of the result."""
+    total: int
+    """Total number of items in the collection."""
+
+    def __len__(self) -> int:
+        return len(self.items)
+
+    @overload
+    def __getitem__(self, index: int) -> T_item: ...
+
+    @overload
+    def __getitem__(self, index: slice) -> Sequence[T_item]: ...
+
+    def __getitem__(self, index: Union[int, slice]) -> Union[T_item, Sequence[T_item]]:
+        return self.items[index]
+
+    # Overriding __iter__ enables list(paginated_result) to work as expected,
+    # but changes Pydantic's default dict iteration behavior (which would otherwise
+    # iterate over field names).
+    def __iter__(self) -> Iterator[T_item]:  # type: ignore
+        return iter(self.items)
+
+    def __repr__(self) -> str:
+        first_item_repr = repr(self.items[0]) if self.items else "empty"
+        items_repr = f"[{first_item_repr}, ...]" if len(self.items) > 1 else first_item_repr
+        slice_repr = f"{self.offset}:" if self.limit == -1 else f"{self.offset}:{self.offset + self.limit}"
+        return f"<PaginatedResult ({slice_repr} of {self.total}) {items_repr}>"

docs/deep-dive/store.md

@@ -205,7 +205,7 @@ await server.start()      # starts uvicorn in a daemon thread and waits for /hea
 # Client (same or different process)
 client = agl.LightningStoreClient("http://localhost:4747")
 
-print(await client.query_rollouts(status=["queuing"]))
+print(await client.query_rollouts(status_in=["queuing"]))
 
 await client.close()
 await server.stop()

docs/how-to/write-first-algorithm.md

@@ -99,7 +99,7 @@ async def find_best_prompt(store, prompts_to_test, task_input):
         await store.wait_for_rollouts([rollout.rollout_id])
 
         # 4. Query the completed rollout and its spans
-        completed_rollout = (await store.query_rollouts([rollout.rollout_id]))[0]
+        completed_rollout = await store.get_rollout_by_id(rollout.rollout_id)
         print(f"[Algo] Received Result: {completed_rollout.model_dump_json(indent=None)}")
 
         spans = await store.query_spans(rollout.rollout_id)

docs/reference/store.md

@@ -26,12 +26,6 @@
 
 ::: agentlightning.store.collection.KeyValue
 
-::: agentlightning.store.collection.FilterOptions
-
-::: agentlightning.store.collection.SortOptions
-
-::: agentlightning.store.collection.PaginatedResult
-
 ::: agentlightning.store.collection.LightningCollections
 
 ::: agentlightning.store.collection.ListBasedCollection

docs/reference/types.md

@@ -32,6 +32,14 @@
 
 ::: agentlightning.Hook
 
+::: agentlightning.PaginatedResult
+
+::: agentlightning.FilterOptions
+
+::: agentlightning.SortOptions
+
+::: agentlightning.FilterField
+
 ## Resources
 
 ::: agentlightning.Resource

examples/apo/apo_custom_algorithm.py

@@ -29,7 +29,7 @@ python apo_custom_algorithm_trainer.py
 
 import argparse
 import asyncio
-from typing import List, Optional
+from typing import Optional, Sequence
 
 from openai import AsyncOpenAI
 from rich.console import Console
@@ -117,7 +117,7 @@ async def apo_rollout(task: str, prompt_template: agl.PromptTemplate) -> float:
     return await llm_judge(task, text)
 
 
-async def log_llm_span(spans: List[agl.Span]) -> None:
+async def log_llm_span(spans: Sequence[agl.Span]) -> None:
     """Logs the LLM related spans that records prompts and responses."""
     for span in spans:
         if "chat.completion" in span.name:

examples/minimal/llm_proxy.py

@@ -24,7 +24,7 @@ dotenv run python llm_proxy.py openai gpt-4.1-mini
 import argparse
 import asyncio
 import os
-from typing import List, no_type_check
+from typing import Sequence, no_type_check
 
 import aiohttp
 from portpicker import pick_unused_port
@@ -189,7 +189,7 @@ def _verify_response_body(response_body: dict, model_name: str):
         assert "chatgpt" in response_body["choices"][0]["message"]["content"].lower()
 
 
-def _verify_span(spans: List[agl.Span]):
+def _verify_span(spans: Sequence[agl.Span]):
     """Only a few spans are checked here.
 
     `raw_gen_ai_request` span:

examples/minimal/write_traces.py

@@ -15,7 +15,7 @@ agl store --port 45993 --log-level DEBUG
 import argparse
 import asyncio
 import time
-from typing import List
+from typing import Sequence
 
 from openai import AsyncOpenAI
 from rich.console import Console
@@ -112,7 +112,7 @@ async def send_traces_via_agentops(use_client: bool = False):
         await store.close()
 
 
-async def _verify_agentops_traces(spans: List[Span], use_client: bool = False):
+async def _verify_agentops_traces(spans: Sequence[Span], use_client: bool = False):
     """Expected traces to something like:
 
     ```python

examples/tinker/agl_tinker/rollout.py

@@ -40,7 +40,7 @@ T_task = TypeVar("T_task")
 WAIT_FOR_ROLLOUTS_INTERVAL = 5.0
 
 
-def reconstruct_transitions(spans: List[Span], adapter: TraceToTripletBase, rollout_id: str) -> Trajectory:
+def reconstruct_transitions(spans: Sequence[Span], adapter: TraceToTripletBase, rollout_id: str) -> Trajectory:
     """Convert Agent-lightning spans into a Tinker `Trajectory`.
 
     This function infers observations, actions, and rewards from the trace triplets emitted by Agent-lightning's

tests/algorithm/test_apo.py

@@ -2,7 +2,7 @@
 
 # pyright: reportPrivateUsage=false
 
-from typing import Any, Dict, Iterator, List, Optional, Sequence, Tuple, cast
+from typing import Any, Dict, Iterator, List, Literal, Optional, Sequence, Tuple, cast
 from unittest.mock import AsyncMock, Mock
 
 import pytest
@@ -32,13 +32,13 @@ class DummyTraceMessagesAdapter(TraceToMessages):
         super().__init__()
         self.seen_spans: Sequence[Span] | None = None
 
-    def adapt(self, source: List[Span], /) -> List[Dict[str, Any]]:  # type: ignore[override]
+    def adapt(self, source: Sequence[Span], /) -> List[Dict[str, Any]]:  # type: ignore[override]
         self.seen_spans = list(source)
         return [dict(payload="converted")]
 
 
 class WrongAdapter(TraceAdapter[List[int]]):
-    def adapt(self, source: List[Span], /) -> List[int]:
+    def adapt(self, source: Sequence[Span], /) -> List[int]:
         return [len(source)]
 
 
@@ -79,7 +79,12 @@ class DummyStore:
             return self.wait_results_queue.pop(0)
         return []
 
-    async def query_spans(self, rollout_id: str) -> List[Span]:
+    async def query_spans(
+        self,
+        rollout_id: str,
+        attempt_id: str | Literal["latest"] | None = None,
+        **_: Any,
+    ) -> List[Span]:
         return list(self.query_spans_map.get(rollout_id, []))
 
 

tests/algorithm/test_baseline.py

@@ -4,7 +4,7 @@ import asyncio
 import logging
 from contextlib import suppress
 from dataclasses import dataclass
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Sequence
 
 import pytest
 
@@ -24,7 +24,7 @@ LOGGER_NAME = "agentlightning.algorithm.fast"
 
 
 class _AdapterStub(TraceAdapter[Dict[str, Any]]):
-    def adapt(self, source: List[Span], /) -> Dict[str, Any]:
+    def adapt(self, source: Sequence[Span], /) -> Dict[str, Any]:
         return {
             "count": len(source),
             "attempt_ids": sorted({span.attempt_id for span in source}),

tests/llm_proxy/test_llm_proxy_gpu.py

@@ -15,7 +15,7 @@ There are some specific TODOs for each test function.
 import ast
 import asyncio
 import json
-from typing import Any, Dict, List, Type, Union, cast
+from typing import Any, Dict, List, Sequence, Type, Union, cast
 
 import anthropic
 import openai
@@ -239,7 +239,7 @@ def _get_async_client_for_resource(resource: LLM):
     return openai.AsyncOpenAI(base_url=resource.endpoint, api_key="token-abc123", timeout=120, max_retries=0)
 
 
-def _find_span(spans: list[Span], name: str):
+def _find_span(spans: Sequence[Span], name: str):
     return [s for s in spans if s.name == name]
 
 

tests/store/dummy_store.py

@@ -1,6 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-from typing import Any, Dict, List, Literal, Optional, Sequence
+from typing import Any, Dict, List, Literal, Optional
 
 from opentelemetry.sdk.trace import ReadableSpan
 
@@ -65,14 +65,12 @@ class DummyLightningStore(LightningStore):
         self.calls.append(("start_attempt", (rollout_id,), {}))
         return self.return_values["start_attempt"]
 
-    async def query_rollouts(
-        self, *, status: Optional[Sequence[RolloutStatus]] = None, rollout_ids: Optional[Sequence[str]] = None
-    ) -> List[Rollout]:
-        self.calls.append(("query_rollouts", (), {"status": status, "rollout_ids": rollout_ids}))
+    async def query_rollouts(self, *args: Any, **kwargs: Any) -> List[Rollout]:
+        self.calls.append(("query_rollouts", args, kwargs))
         return self.return_values["query_rollouts"]
 
-    async def query_attempts(self, rollout_id: str) -> List[Attempt]:
-        self.calls.append(("query_attempts", (rollout_id,), {}))
+    async def query_attempts(self, *args: Any, **kwargs: Any) -> List[Attempt]:
+        self.calls.append(("query_attempts", args, kwargs))
         return self.return_values["query_attempts"]
 
     async def get_rollout_by_id(self, rollout_id: str) -> Optional[Rollout]:
@@ -99,6 +97,10 @@ class DummyLightningStore(LightningStore):
         self.calls.append(("get_latest_resources", (), {}))
         return self.return_values["get_latest_resources"]
 
+    async def query_resources(self, *args: Any, **kwargs: Any) -> List[ResourcesUpdate]:
+        self.calls.append(("query_resources", args, kwargs))
+        return self.return_values["query_resources"]
+
     async def add_span(self, span: Span) -> Span:
         self.calls.append(("add_span", (span,), {}))
         return self.return_values["add_span"]
@@ -121,12 +123,8 @@ class DummyLightningStore(LightningStore):
         self.calls.append(("get_next_span_sequence_id", (rollout_id, attempt_id), {}))
         return self.return_values["get_next_span_sequence_id"]
 
-    async def query_spans(
-        self,
-        rollout_id: str,
-        attempt_id: str | Literal["latest"] | None = None,
-    ) -> List[Span]:
-        self.calls.append(("query_spans", (rollout_id, attempt_id), {}))
+    async def query_spans(self, *args: Any, **kwargs: Any) -> List[Span]:
+        self.calls.append(("query_spans", args, kwargs))
         return self.return_values["query_spans"]
 
     async def update_rollout(
@@ -166,8 +164,8 @@ class DummyLightningStore(LightningStore):
         )
         return self.return_values["update_attempt"]
 
-    async def query_workers(self) -> List[Worker]:
-        self.calls.append(("query_workers", (), {}))
+    async def query_workers(self, *args: Any, **kwargs: Any) -> List[Worker]:
+        self.calls.append(("query_workers", args, kwargs))
         return self.return_values["query_workers"]
 
     async def get_worker_by_id(self, worker_id: str) -> Optional[Worker]:

tests/store/test_client_server.py

@@ -18,7 +18,7 @@ from yarl import URL
 from agentlightning.store.base import UNSET, LightningStore
 from agentlightning.store.client_server import LightningStoreClient, LightningStoreServer
 from agentlightning.store.memory import InMemoryLightningStore
-from agentlightning.types import LLM, OtelResource, PromptTemplate, RolloutConfig, Span, TraceStatus
+from agentlightning.types import LLM, OtelResource, PaginatedResult, PromptTemplate, RolloutConfig, Span, TraceStatus
 from agentlightning.utils.server_launcher import LaunchMode, PythonServerLauncherArgs
 
 
@@ -216,8 +216,13 @@ async def test_query_resources_history(server_client: Tuple[LightningStoreServer
     """Server and client should return identical resource history ordering."""
     server, client = server_client
 
-    assert await server.query_resources() == []
-    assert await client.query_resources() == []
+    server_history_empty = await server.query_resources()
+    assert isinstance(server_history_empty, PaginatedResult)
+    assert len(server_history_empty) == 0
+
+    client_history_empty = await client.query_resources()
+    assert isinstance(client_history_empty, PaginatedResult)
+    assert len(client_history_empty) == 0
 
     first = await server.add_resources(
         cast(
@@ -248,6 +253,26 @@ async def test_query_resources_history(server_client: Tuple[LightningStoreServer
     assert sorted([item.resources_id for item in client_history]) == sorted(expected_ids)
 
 
+@pytest.mark.asyncio
+async def test_client_query_resources_filters_and_pagination(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient],
+) -> None:
+    _, client = server_client
+
+    alpha = PromptTemplate(resource_type="prompt_template", template="alpha", engine="jinja")
+    beta = PromptTemplate(resource_type="prompt_template", template="beta", engine="jinja")
+
+    await client.update_resources("manual-alpha", cast(Any, {"prompt": alpha}))
+    await client.update_resources("manual-beta", cast(Any, {"prompt": beta}))
+
+    contains_beta = await client.query_resources(resources_id_contains="beta")
+    assert [item.resources_id for item in contains_beta] == ["manual-beta"]
+
+    sorted_ids = sorted(["manual-alpha", "manual-beta"], reverse=True)
+    paged = await client.query_resources(sort_by="resources_id", sort_order="desc", limit=1, offset=1)
+    assert [item.resources_id for item in paged] == sorted_ids[1:2]
+
+
 @pytest.mark.asyncio
 async def test_client_server_end_to_end(
     server_client: Tuple[LightningStoreServer, LightningStoreClient], mock_readable_span: ReadableSpan
@@ -413,6 +438,27 @@ async def test_client_server_end_to_end(
     assert wait_result and wait_result[0].status == "succeeded"
 
 
+@pytest.mark.asyncio
+async def test_client_query_rollouts_filters_and_pagination(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient],
+) -> None:
+    _, client = server_client
+
+    rollouts = [await client.enqueue_rollout(input={"idx": idx}) for idx in range(3)]
+    await client.update_rollout(rollout_id=rollouts[0].rollout_id, status="failed")
+
+    failed = await client.query_rollouts(status_in=["failed"])
+    assert [rollout.rollout_id for rollout in failed] == [rollouts[0].rollout_id]
+
+    substring = rollouts[2].rollout_id[-4:]
+    contains = await client.query_rollouts(rollout_id_contains=substring)
+    assert any(rollout.rollout_id == rollouts[2].rollout_id for rollout in contains)
+
+    sorted_ids = sorted([rollout.rollout_id for rollout in rollouts], reverse=True)
+    paged = await client.query_rollouts(sort_by="rollout_id", sort_order="desc", limit=1, offset=1)
+    assert [rollout.rollout_id for rollout in paged] == sorted_ids[1:2]
+
+
 @pytest.mark.asyncio
 async def test_update_rollout_none_vs_unset(server_client: Tuple[LightningStoreServer, LightningStoreClient]) -> None:
     _, client = server_client
@@ -548,6 +594,37 @@ async def test_worker_status_transitions_via_attempts(
     assert idle.current_attempt_id is None
 
 
+@pytest.mark.asyncio
+async def test_client_query_workers_filters(server_client: Tuple[LightningStoreServer, LightningStoreClient]) -> None:
+    _, client = server_client
+
+    await client.update_worker("alpha-worker", heartbeat_stats={"cpu": 0.2})
+    await client.update_worker("beta-worker", heartbeat_stats={"cpu": 0.8})
+
+    busy_rollout = await client.start_rollout(input={"worker": "alpha"})
+    await client.update_attempt(
+        busy_rollout.rollout_id,
+        busy_rollout.attempt.attempt_id,
+        worker_id="alpha-worker",
+        status="running",
+    )
+
+    busy_workers = await client.query_workers(status_in=["busy"])
+    assert [worker.worker_id for worker in busy_workers] == ["alpha-worker"]
+
+    contains_beta = await client.query_workers(worker_id_contains="beta")
+    assert [worker.worker_id for worker in contains_beta] == ["beta-worker"]
+
+    or_filtered = await client.query_workers(
+        status_in=["busy"],
+        worker_id_contains="beta",
+        filter_logic="or",
+        sort_by="worker_id",
+        sort_order="asc",
+    )
+    assert [worker.worker_id for worker in or_filtered] == ["alpha-worker", "beta-worker"]
+
+
 @pytest.mark.asyncio
 async def test_get_worker_by_id(server_client: Tuple[LightningStoreServer, LightningStoreClient]) -> None:
     server, client = server_client
@@ -586,6 +663,46 @@ async def test_update_attempt_rejects_none_values(
     assert exc_info.value.status == 400
 
 
+@pytest.mark.asyncio
+async def test_client_query_spans_filters_and_pagination(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient],
+) -> None:
+    server, client = server_client
+
+    attempted = await server.start_rollout(input={"span": "filters"})
+    attempt_id = attempted.attempt.attempt_id
+
+    spans = [
+        _make_span(attempted.rollout_id, attempt_id, 1, "planner"),
+        _make_span(attempted.rollout_id, attempt_id, 2, "reward"),
+        _make_span(attempted.rollout_id, attempt_id, 3, "tool-call"),
+    ]
+    for span in spans:
+        await server.add_span(span)
+
+    planner = await client.query_spans(attempted.rollout_id, attempt_id=attempt_id, name_contains="plan")
+    assert [span.name for span in planner] == ["planner"]
+
+    or_filtered = await client.query_spans(
+        attempted.rollout_id,
+        attempt_id=attempt_id,
+        span_id=spans[0].span_id,
+        trace_id_contains=spans[2].trace_id[-4:],
+        filter_logic="or",
+    )
+    assert {span.span_id for span in or_filtered} == {spans[0].span_id, spans[2].span_id}
+
+    paged = await client.query_spans(
+        attempted.rollout_id,
+        attempt_id=attempt_id,
+        sort_by="sequence_id",
+        sort_order="desc",
+        limit=1,
+        offset=1,
+    )
+    assert [span.span_id for span in paged] == [spans[1].span_id]
+
+
 @pytest.mark.asyncio
 async def test_concurrent_add_otel_span_sequence_ids_unique(
     server_client: Tuple[LightningStoreServer, LightningStoreClient], mock_readable_span: ReadableSpan

tests/store/test_collection_memory.py

@@ -7,7 +7,9 @@ from typing import Dict, Iterable, List, Literal, Mapping, Sequence, Tuple
 import pytest
 from pydantic import BaseModel, Field
 
+import agentlightning.store.collection.memory as memory_module
 from agentlightning.store.collection import DequeBasedQueue, DictBasedKeyValue, ListBasedCollection
+from agentlightning.store.collection.memory import _item_matches_filters  # pyright: ignore[reportPrivateUsage]
 
 
 class SampleItem(BaseModel):
@@ -376,6 +378,67 @@ async def test_list_collection_filter_logic(
     assert _sorted_pairs(result.items) == sorted(expected)
 
 
+@pytest.mark.asyncio()
+async def test_list_collection_must_filters_respected_with_or(
+    sample_collection: ListBasedCollection[SampleItem],
+) -> None:
+    filters = {
+        "_aggregate": "or",
+        "_must": {"partition": {"exact": "alpha"}},
+        "status": {"exact": "done"},
+        "tags": {"contains": "urgent"},
+    }
+    result = await sample_collection.query(filter=filters)  # type: ignore[arg-type]
+    assert _sorted_pairs(result.items) == [("alpha", 1)]
+
+
+@pytest.mark.asyncio()
+async def test_list_collection_must_filters_accept_sequence(
+    sample_collection: ListBasedCollection[SampleItem],
+) -> None:
+    filters = {
+        "_aggregate": "or",
+        "_must": [
+            {"partition": {"exact": "beta"}},
+            {"index": {"exact": 2}},
+        ],
+        "status": {"exact": "new"},
+        "tags": {"contains": "beta"},
+    }
+    result = await sample_collection.query(filter=filters)  # type: ignore[arg-type]
+    assert _sorted_pairs(result.items) == [("beta", 2)]
+
+
+@pytest.mark.asyncio()
+async def test_list_collection_must_filters_limit_tree_scan_even_with_or(
+    sample_collection: ListBasedCollection[SampleItem],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    seen: List[Tuple[str, int]] = []
+    original = _item_matches_filters
+
+    def tracking(
+        item: SampleItem,
+        filters: object,
+        filter_logic: str,
+        must_filters: object | None = None,
+    ) -> bool:
+        seen.append((item.partition, item.index))
+        return original(item, filters, filter_logic, must_filters)  # type: ignore[arg-type]
+
+    monkeypatch.setattr(memory_module, "_item_matches_filters", tracking)
+
+    filters = {
+        "_aggregate": "or",
+        "_must": {"partition": {"exact": "gamma"}},
+        "status": {"exact": "done"},
+        "tags": {"contains": "urgent"},
+    }
+    result = await sample_collection.query(filter=filters)  # type: ignore[arg-type]
+    assert _sorted_pairs(result.items) == [("gamma", 2)]
+    assert set(seen) == {("gamma", 1), ("gamma", 2)}
+
+
 @pytest.mark.asyncio()
 async def test_list_collection_primary_key_prefix_limits_filter_checks(
     sample_items: Sequence[SampleItem],
@@ -383,15 +446,18 @@ async def test_list_collection_primary_key_prefix_limits_filter_checks(
 ) -> None:
     collection = _build_collection(sample_items)
     seen: List[Tuple[str, int]] = []
-    original = (  # pyright: ignore[reportPrivateUsage,reportUnknownMemberType,reportUnknownVariableType]
-        ListBasedCollection._item_matches_filters  # pyright: ignore[reportPrivateUsage,reportUnknownMemberType]
-    )
+    original = _item_matches_filters
 
-    def tracking(item: SampleItem, filters: object, filter_logic: str) -> bool:
+    def tracking(
+        item: SampleItem,
+        filters: object,
+        filter_logic: str,
+        must_filters: object | None = None,
+    ) -> bool:
         seen.append((item.partition, item.index))
-        return original(item, filters, filter_logic)  # type: ignore[arg-type]
+        return original(item, filters, filter_logic, must_filters)  # type: ignore[arg-type]
 
-    monkeypatch.setattr(ListBasedCollection, "_item_matches_filters", staticmethod(tracking))  # type: ignore[arg-type]
+    monkeypatch.setattr(memory_module, "_item_matches_filters", tracking)
 
     filters = {"partition": {"exact": "alpha"}, "index": {"within": {1, 2}}}
     result = await collection.query(filter=filters)  # type: ignore[arg-type]
@@ -413,11 +479,12 @@ async def test_list_collection_full_primary_key_avoids_tree_scan(
         self: ListBasedCollection[SampleItem],
         root: Mapping[str, object] | None = None,
         filters: object | None = None,
+        must_filters: object | None = None,
         filter_logic: str = "and",
     ) -> Iterable[SampleItem]:
         nonlocal call_count
         call_count += 1
-        return original_iter_items(self, root, filters, filter_logic)  # type: ignore[arg-type]
+        return original_iter_items(self, root, filters, must_filters, filter_logic)  # type: ignore[arg-type]
 
     monkeypatch.setattr(ListBasedCollection, "_iter_items", tracking)
 

tests/store/test_implementation.py

@@ -19,7 +19,7 @@ import asyncio
 import logging
 import sys
 import time
-from typing import List, Optional, cast
+from typing import List, Optional, Sequence, cast
 from unittest.mock import Mock
 
 import pytest
@@ -33,6 +33,7 @@ from agentlightning.types import (
     Event,
     Link,
     OtelResource,
+    PaginatedResult,
     PromptTemplate,
     ResourcesUpdate,
     Rollout,
@@ -42,6 +43,24 @@ from agentlightning.types import (
     TraceStatus,
 )
 
+# Typing tests
+
+
+def test_paginated_result_behaves_like_sequence() -> None:
+    result = PaginatedResult(items=["a", "b", "c"], limit=2, offset=1, total=5)
+
+    assert isinstance(result, Sequence)
+    assert len(result) == 3
+    assert result[0] == "a"
+    assert result[1:] == ["b", "c"]
+    assert list(result) == ["a", "b", "c"]
+
+    assert repr(result) == "<PaginatedResult (1:3 of 5) ['a', ...]>"
+
+    result2 = PaginatedResult(items=["a", "b", "c"], limit=-1, offset=1, total=5)
+    assert repr(result2) == "<PaginatedResult (1: of 5) ['a', ...]>"
+
+
 # Core CRUD Operations Tests
 
 
@@ -174,6 +193,186 @@ async def test_query_rollouts_returns_latest_attempt(store_fixture: LightningSto
     assert retrieved.attempt.sequence_id == latest_attempt.attempt.sequence_id
 
 
+@pytest.mark.asyncio
+async def test_query_rollouts_supports_new_filters(store_fixture: LightningStore) -> None:
+    """The expanded query interface should honor filtering, sorting, and pagination."""
+    rollouts = [await store_fixture.enqueue_rollout(input={"idx": idx}) for idx in range(3)]
+    await store_fixture.update_rollout(rollout_id=rollouts[2].rollout_id, status="failed")
+
+    failed = await store_fixture.query_rollouts(status_in=["failed"])
+    assert [r.rollout_id for r in failed] == [rollouts[2].rollout_id]
+
+    sorted_desc = sorted([r.rollout_id for r in rollouts], reverse=True)
+    paged = await store_fixture.query_rollouts(sort_by="rollout_id", sort_order="desc", limit=2)
+    assert [r.rollout_id for r in paged] == sorted_desc[:2]
+
+    offset_item = await store_fixture.query_rollouts(sort_by="rollout_id", sort_order="desc", limit=1, offset=1)
+    assert [r.rollout_id for r in offset_item] == sorted_desc[1:2]
+
+    contains = await store_fixture.query_rollouts(
+        rollout_id_contains=rollouts[0].rollout_id[-4:],
+        status_in=["queuing"],
+    )
+    assert any(r.rollout_id == rollouts[0].rollout_id for r in contains)
+
+    or_filtered = await store_fixture.query_rollouts(
+        status_in=["succeeded"],
+        rollout_id_in=[rollouts[1].rollout_id],
+        filter_logic="or",
+    )
+    assert [r.rollout_id for r in or_filtered] == [rollouts[1].rollout_id]
+
+
+@pytest.mark.asyncio
+async def test_query_rollouts_status_in_takes_precedence(store_fixture: LightningStore) -> None:
+    """status_in should override the legacy status parameter when both are provided."""
+    failed = await store_fixture.enqueue_rollout(input={"kind": "failed"})
+    pending = await store_fixture.enqueue_rollout(input={"kind": "pending"})
+
+    await store_fixture.update_rollout(rollout_id=failed.rollout_id, status="failed")
+
+    results = await store_fixture.query_rollouts(status=["queuing"], status_in=["failed"])
+    assert [rollout.rollout_id for rollout in results] == [failed.rollout_id]
+
+    legacy = await store_fixture.query_rollouts(status=["queuing"])
+    assert any(rollout.rollout_id == pending.rollout_id for rollout in legacy)
+
+
+@pytest.mark.asyncio
+async def test_query_rollouts_rollout_id_in_takes_precedence(store_fixture: LightningStore) -> None:
+    """rollout_id_in should override the legacy rollout_ids parameter."""
+    keep = await store_fixture.enqueue_rollout(input={"kind": "keep"})
+    ignored = await store_fixture.enqueue_rollout(input={"kind": "ignored"})
+
+    results = await store_fixture.query_rollouts(rollout_ids=[ignored.rollout_id], rollout_id_in=[keep.rollout_id])
+    assert [rollout.rollout_id for rollout in results] == [keep.rollout_id]
+
+
+@pytest.mark.asyncio
+async def test_query_rollouts_filter_logic_controls_contains_behavior(store_fixture: LightningStore) -> None:
+    """Changing filter_logic should alter how status and substring filters are combined."""
+    failed = await store_fixture.enqueue_rollout(input={"kind": "failed"})
+    still_queueing = await store_fixture.enqueue_rollout(input={"kind": "queue"})
+    await store_fixture.update_rollout(rollout_id=failed.rollout_id, status="failed")
+
+    substring = still_queueing.rollout_id[-6:]
+    and_results = await store_fixture.query_rollouts(
+        status_in=["failed"],
+        rollout_id_contains=substring,
+    )
+    assert len(and_results) == 0
+
+    or_results = await store_fixture.query_rollouts(
+        status_in=["failed"],
+        rollout_id_contains=substring,
+        filter_logic="or",
+        sort_by="rollout_id",
+    )
+    returned_ids = {rollout.rollout_id for rollout in or_results}
+    assert returned_ids == {failed.rollout_id, still_queueing.rollout_id}
+
+
+@pytest.mark.asyncio
+async def test_query_rollouts_combined_filters_with_sort_and_pagination(store_fixture: LightningStore) -> None:
+    """Complex combinations of filters should work with explicit sorting and pagination."""
+    rollouts = [await store_fixture.enqueue_rollout(input={"idx": idx}) for idx in range(4)]
+    await store_fixture.update_rollout(rollout_id=rollouts[0].rollout_id, status="failed")
+    await store_fixture.update_rollout(rollout_id=rollouts[3].rollout_id, status="failed")
+
+    substring = rollouts[2].rollout_id[-5:]
+    filtered = await store_fixture.query_rollouts(
+        status_in=["failed"],
+        rollout_id_contains=substring,
+        filter_logic="or",
+        sort_by="rollout_id",
+        sort_order="asc",
+        limit=2,
+        offset=1,
+    )
+
+    expected_ids = sorted({rollouts[0].rollout_id, rollouts[2].rollout_id, rollouts[3].rollout_id})
+    assert [item.rollout_id for item in filtered] == expected_ids[1:3]
+
+
+@pytest.mark.asyncio
+async def test_query_rollouts_reports_pagination_metadata(store_fixture: LightningStore) -> None:
+    """Paginated rollouts should expose limit/offset/total values."""
+    rollouts = [await store_fixture.enqueue_rollout(input={"idx": idx}) for idx in range(3)]
+
+    paginated = await store_fixture.query_rollouts(sort_by="rollout_id", sort_order="asc", limit=1, offset=1)
+    assert isinstance(paginated, PaginatedResult)
+    assert paginated.limit == 1
+    assert paginated.offset == 1
+    assert paginated.total == len(rollouts)
+
+
+@pytest.mark.asyncio
+async def test_query_attempts_supports_sort_and_limit(store_fixture: LightningStore) -> None:
+    """Attempt queries should respect sorting and pagination controls."""
+    attempted = await store_fixture.start_rollout(input={"payload": "attempt-filters"})
+    await store_fixture.start_attempt(attempted.rollout_id)
+    await store_fixture.start_attempt(attempted.rollout_id)
+
+    attempts_desc = await store_fixture.query_attempts(attempted.rollout_id, sort_by="sequence_id", sort_order="desc")
+    assert [attempt.sequence_id for attempt in attempts_desc] == [3, 2, 1]
+
+    middle_attempt = await store_fixture.query_attempts(
+        attempted.rollout_id,
+        sort_by="sequence_id",
+        sort_order="desc",
+        limit=1,
+        offset=1,
+    )
+    assert [attempt.sequence_id for attempt in middle_attempt] == [2]
+
+
+@pytest.mark.asyncio
+async def test_query_attempts_offset_past_end_returns_empty(store_fixture: LightningStore) -> None:
+    """Offsets beyond the result size should return an empty list."""
+    attempted = await store_fixture.start_rollout(input={"payload": "attempt-offset"})
+    await store_fixture.start_attempt(attempted.rollout_id)
+    await store_fixture.start_attempt(attempted.rollout_id)
+
+    results = await store_fixture.query_attempts(
+        attempted.rollout_id,
+        sort_by="sequence_id",
+        sort_order="asc",
+        limit=1,
+        offset=10,
+    )
+    assert len(results) == 0
+
+
+@pytest.mark.asyncio
+async def test_query_attempts_zero_limit_returns_no_items(store_fixture: LightningStore) -> None:
+    """A zero limit should be treated as 'return nothing' even when attempts exist."""
+    attempted = await store_fixture.start_rollout(input={"payload": "attempt-limit"})
+    await store_fixture.start_attempt(attempted.rollout_id)
+
+    results = await store_fixture.query_attempts(attempted.rollout_id, limit=0)
+    assert len(results) == 0
+
+
+@pytest.mark.asyncio
+async def test_query_attempts_reports_pagination_metadata(store_fixture: LightningStore) -> None:
+    """Attempt pagination should retain metadata."""
+    attempted = await store_fixture.start_rollout(input={"payload": "attempt-pagination"})
+    await store_fixture.start_attempt(attempted.rollout_id)
+    await store_fixture.start_attempt(attempted.rollout_id)
+
+    paginated = await store_fixture.query_attempts(
+        attempted.rollout_id,
+        sort_by="sequence_id",
+        sort_order="asc",
+        limit=1,
+        offset=1,
+    )
+    assert isinstance(paginated, PaginatedResult)
+    assert paginated.limit == 1
+    assert paginated.offset == 1
+    assert paginated.total == 3
+
+
 @pytest.mark.asyncio
 async def test_get_rollout_by_id_returns_latest_attempt(store_fixture: LightningStore) -> None:
     """Fetching a rollout by ID should include the latest attempt when available."""
@@ -490,6 +689,111 @@ async def test_update_and_query_workers(store_fixture: LightningStore) -> None:
         await store_fixture.update_worker("worker-1", heartbeat_stats=None)  # type: ignore[arg-type]
 
 
+@pytest.mark.asyncio
+async def test_query_workers_supports_filters(store_fixture: LightningStore) -> None:
+    """Worker queries should support filtering, sorting, and pagination."""
+    await store_fixture.update_worker("alpha-worker", heartbeat_stats={"cpu": 0.2})
+    await store_fixture.update_worker("beta-worker", heartbeat_stats={"cpu": 0.8})
+
+    busy_rollout = await store_fixture.start_rollout(input={"worker": "alpha"})
+    await store_fixture.update_attempt(
+        busy_rollout.rollout_id,
+        busy_rollout.attempt.attempt_id,
+        worker_id="alpha-worker",
+        status="running",
+    )
+
+    idle_rollout = await store_fixture.start_rollout(input={"worker": "beta"})
+    await store_fixture.update_attempt(
+        idle_rollout.rollout_id,
+        idle_rollout.attempt.attempt_id,
+        worker_id="beta-worker",
+        status="succeeded",
+    )
+
+    busy_workers = await store_fixture.query_workers(status_in=["busy"])
+    assert [worker.worker_id for worker in busy_workers] == ["alpha-worker"]
+
+    contains_beta = await store_fixture.query_workers(worker_id_contains="beta")
+    assert [worker.worker_id for worker in contains_beta] == ["beta-worker"]
+
+    sorted_workers = await store_fixture.query_workers(sort_by="worker_id", sort_order="desc")
+    assert [worker.worker_id for worker in sorted_workers] == ["beta-worker", "alpha-worker"]
+
+    paged = await store_fixture.query_workers(sort_by="worker_id", sort_order="desc", limit=1, offset=1)
+    assert [worker.worker_id for worker in paged] == ["alpha-worker"]
+
+
+@pytest.mark.asyncio
+async def test_query_workers_filter_logic_or_combines_conditions(store_fixture: LightningStore) -> None:
+    """filter_logic should dictate whether worker filters act as AND or OR."""
+    await store_fixture.update_worker("cpu-worker", heartbeat_stats={"cpu": 0.1})
+
+    busy_rollout = await store_fixture.start_rollout(input={"task": "busy"})
+    await store_fixture.update_attempt(
+        busy_rollout.rollout_id,
+        busy_rollout.attempt.attempt_id,
+        worker_id="busy-worker",
+        status="running",
+    )
+
+    and_results = await store_fixture.query_workers(status_in=["busy"], worker_id_contains="cpu")
+    assert len(and_results) == 0
+
+    or_results = await store_fixture.query_workers(
+        status_in=["busy"],
+        worker_id_contains="cpu",
+        filter_logic="or",
+        sort_by="worker_id",
+    )
+    assert [worker.worker_id for worker in or_results] == ["busy-worker", "cpu-worker"]
+
+
+@pytest.mark.asyncio
+async def test_query_workers_filter_logic_and_with_sort(store_fixture: LightningStore) -> None:
+    """Explicit AND logic should combine substring filters with status filters and respect sorting."""
+    await store_fixture.update_worker("alpha-worker", heartbeat_stats={"cpu": 0.1})
+    await store_fixture.update_worker("beta-worker", heartbeat_stats={"cpu": 0.1})
+
+    busy_rollout = await store_fixture.start_rollout(input={"task": "alpha"})
+    await store_fixture.update_attempt(
+        busy_rollout.rollout_id,
+        busy_rollout.attempt.attempt_id,
+        worker_id="alpha-worker",
+        status="running",
+    )
+    idle_rollout = await store_fixture.start_rollout(input={"task": "beta"})
+    await store_fixture.update_attempt(
+        idle_rollout.rollout_id,
+        idle_rollout.attempt.attempt_id,
+        worker_id="beta-worker",
+        status="succeeded",
+    )
+
+    filtered = await store_fixture.query_workers(
+        status_in=["busy"],
+        worker_id_contains="alpha",
+        filter_logic="and",
+        sort_by="worker_id",
+        sort_order="asc",
+        limit=1,
+    )
+    assert [worker.worker_id for worker in filtered] == ["alpha-worker"]
+
+
+@pytest.mark.asyncio
+async def test_query_workers_reports_pagination_metadata(store_fixture: LightningStore) -> None:
+    """Worker pagination should expose metadata for callers."""
+    for worker_id in ["worker-a", "worker-b", "worker-c"]:
+        await store_fixture.update_worker(worker_id, heartbeat_stats={"cpu": 0.1})
+
+    paginated = await store_fixture.query_workers(sort_by="worker_id", sort_order="asc", limit=1, offset=1)
+    assert isinstance(paginated, PaginatedResult)
+    assert paginated.limit == 1
+    assert paginated.offset == 1
+    assert paginated.total == 3
+
+
 # Resource Management Tests
 
 
@@ -563,7 +867,7 @@ async def test_add_resources_multiple_times_generates_unique_ids(store_fixture:
 @pytest.mark.asyncio
 async def test_query_resources_returns_history(store_fixture: LightningStore) -> None:
     """query_resources should list snapshots in the order they were stored."""
-    assert await store_fixture.query_resources() == []
+    assert len(await store_fixture.query_resources()) == 0
 
     first = await store_fixture.add_resources(
         {
@@ -583,6 +887,95 @@ async def test_query_resources_returns_history(store_fixture: LightningStore) ->
     assert isinstance(history[1], ResourcesUpdate)
 
 
+@pytest.mark.asyncio
+async def test_query_resources_supports_filters(store_fixture: LightningStore) -> None:
+    """Resource queries should support substring filters and pagination."""
+    alpha = PromptTemplate(resource_type="prompt_template", template="alpha", engine="jinja")
+    beta = PromptTemplate(resource_type="prompt_template", template="beta", engine="jinja")
+
+    await store_fixture.update_resources("manual-alpha", {"prompt": alpha})
+    await store_fixture.update_resources("manual-beta", {"prompt": beta})
+
+    contains_beta = await store_fixture.query_resources(resources_id_contains="beta")
+    assert [item.resources_id for item in contains_beta] == ["manual-beta"]
+
+    sorted_ids = sorted(["manual-alpha", "manual-beta"], reverse=True)
+    paged = await store_fixture.query_resources(sort_by="resources_id", sort_order="desc", limit=1)
+    assert [item.resources_id for item in paged] == sorted_ids[:1]
+
+    offset_item = await store_fixture.query_resources(sort_by="resources_id", sort_order="desc", limit=1, offset=1)
+    assert [item.resources_id for item in offset_item] == sorted_ids[1:2]
+
+
+@pytest.mark.asyncio
+async def test_query_resources_combines_exact_and_contains_filters(store_fixture: LightningStore) -> None:
+    """Exact and substring filters should be usable together."""
+    alpha = PromptTemplate(resource_type="prompt_template", template="alpha", engine="jinja")
+    beta = PromptTemplate(resource_type="prompt_template", template="beta", engine="jinja")
+
+    await store_fixture.update_resources("manual-alpha", {"prompt": alpha})
+    await store_fixture.update_resources("manual-beta", {"prompt": beta})
+
+    results = await store_fixture.query_resources(resources_id="manual-alpha", resources_id_contains="manual")
+    assert [item.resources_id for item in results] == ["manual-alpha"]
+
+
+@pytest.mark.asyncio
+async def test_query_resources_offset_beyond_range_returns_empty(store_fixture: LightningStore) -> None:
+    """Large offsets should simply return an empty result."""
+    await store_fixture.update_resources(
+        "snapshot-a",
+        {"prompt": PromptTemplate(resource_type="prompt_template", template="a", engine="jinja")},
+    )
+    await store_fixture.update_resources(
+        "snapshot-b",
+        {"prompt": PromptTemplate(resource_type="prompt_template", template="b", engine="jinja")},
+    )
+
+    results = await store_fixture.query_resources(sort_by="resources_id", limit=1, offset=5)
+    assert len(results) == 0
+
+
+@pytest.mark.asyncio
+async def test_query_resources_contains_with_sort_and_pagination(store_fixture: LightningStore) -> None:
+    """Substring filters should combine with sort order, limit, and offset."""
+    for suffix in ["alpha", "beta", "gamma"]:
+        await store_fixture.update_resources(
+            f"manual-{suffix}",
+            {"prompt": PromptTemplate(resource_type="prompt_template", template=suffix, engine="jinja")},
+        )
+
+    filtered = await store_fixture.query_resources(
+        resources_id_contains="manual-",
+        sort_by="resources_id",
+        sort_order="asc",
+        limit=2,
+        offset=1,
+    )
+
+    expected_ids = sorted([f"manual-{suffix}" for suffix in ["alpha", "beta", "gamma"]])
+    assert [item.resources_id for item in filtered] == expected_ids[1:3]
+
+
+@pytest.mark.asyncio
+async def test_query_resources_reports_pagination_metadata(store_fixture: LightningStore) -> None:
+    """Resource pagination should expose metadata fields."""
+    await store_fixture.update_resources(
+        "snapshot-a",
+        {"prompt": PromptTemplate(resource_type="prompt_template", template="a", engine="jinja")},
+    )
+    await store_fixture.update_resources(
+        "snapshot-b",
+        {"prompt": PromptTemplate(resource_type="prompt_template", template="b", engine="jinja")},
+    )
+
+    paginated = await store_fixture.query_resources(sort_by="resources_id", limit=1, offset=1)
+    assert isinstance(paginated, PaginatedResult)
+    assert paginated.limit == 1
+    assert paginated.offset == 1
+    assert paginated.total == 2
+
+
 @pytest.mark.asyncio
 async def test_resource_lifecycle(store_fixture: LightningStore) -> None:
     """Test adding, updating, and retrieving resources."""
@@ -866,6 +1259,296 @@ async def test_query_spans_by_attempt(store_fixture: LightningStore, mock_readab
     assert len(no_spans) == 0
 
 
+@pytest.mark.asyncio
+async def test_query_spans_supports_filters(store_fixture: LightningStore) -> None:
+    """Span queries should honor filtering logic and pagination."""
+    attempted = await store_fixture.start_rollout(input={"payload": "span-filters"})
+    attempt_id = attempted.attempt.attempt_id
+
+    def build_span(idx: int, *, name: str, parent: Optional[str]) -> Span:
+        trace_hex = f"{idx:032x}"
+        span_hex = f"{idx:016x}"
+        return Span(
+            rollout_id=attempted.rollout_id,
+            attempt_id=attempt_id,
+            sequence_id=idx,
+            trace_id=trace_hex,
+            span_id=span_hex,
+            parent_id=parent,
+            name=name,
+            status=TraceStatus(status_code="OK"),
+            attributes={},
+            events=[Event(name=f"event-{idx}", attributes={})],
+            links=[],
+            start_time=None,
+            end_time=None,
+            context=SpanContext(trace_id=trace_hex, span_id=span_hex, is_remote=False, trace_state={}),
+            parent=None,
+            resource=OtelResource(attributes={}, schema_url=""),
+        )
+
+    created_spans = [
+        build_span(1, name="reward", parent=None),
+        build_span(2, name="planner", parent=f"{1:016x}"),
+        build_span(3, name="tool-call", parent=f"{2:016x}"),
+    ]
+    for span in created_spans:
+        await store_fixture.add_span(span)
+
+    trace_filtered = await store_fixture.query_spans(attempted.rollout_id, trace_id=created_spans[1].trace_id)
+    assert [s.span_id for s in trace_filtered] == [created_spans[1].span_id]
+
+    or_filtered = await store_fixture.query_spans(
+        attempted.rollout_id,
+        span_id=created_spans[0].span_id,
+        trace_id_contains=created_spans[2].trace_id[-4:],
+        filter_logic="or",
+    )
+    assert {s.span_id for s in or_filtered} == {created_spans[0].span_id, created_spans[2].span_id}
+
+    parent_filtered = await store_fixture.query_spans(
+        attempted.rollout_id,
+        parent_id_contains=created_spans[1].span_id[-4:],
+    )
+    assert [s.span_id for s in parent_filtered] == [created_spans[2].span_id]
+
+    sorted_ids = sorted([span.span_id for span in created_spans], reverse=True)
+    paged = await store_fixture.query_spans(
+        attempted.rollout_id,
+        sort_by="span_id",
+        sort_order="desc",
+        limit=1,
+        offset=1,
+    )
+    assert [span.span_id for span in paged] == sorted_ids[1:2]
+
+
+@pytest.mark.asyncio
+async def test_query_spans_filter_logic_respects_rollout_scope(store_fixture: LightningStore) -> None:
+    """Even with OR logic, query_spans should not leak spans from other rollouts."""
+    first = await store_fixture.start_rollout(input={"payload": "first-span"})
+    second = await store_fixture.start_rollout(input={"payload": "second-span"})
+
+    def make_span(rollout_id: str, attempt_id: str, idx: int, name: str) -> Span:
+        span_hex = f"{idx:016x}"
+        trace_hex = f"{idx:032x}"
+        return Span(
+            rollout_id=rollout_id,
+            attempt_id=attempt_id,
+            sequence_id=idx,
+            trace_id=trace_hex,
+            span_id=span_hex,
+            parent_id=None,
+            name=name,
+            status=TraceStatus(status_code="OK"),
+            attributes={},
+            events=[],
+            links=[],
+            start_time=None,
+            end_time=None,
+            context=SpanContext(trace_id=trace_hex, span_id=span_hex, is_remote=False, trace_state={}),
+            parent=None,
+            resource=OtelResource(attributes={}, schema_url=""),
+        )
+
+    span_first = make_span(first.rollout_id, first.attempt.attempt_id, 1, "alpha")
+    span_second = make_span(second.rollout_id, second.attempt.attempt_id, 2, "beta")
+    await store_fixture.add_span(span_first)
+    await store_fixture.add_span(span_second)
+
+    results = await store_fixture.query_spans(
+        first.rollout_id,
+        attempt_id=first.attempt.attempt_id,
+        name_contains="alpha",
+        trace_id=span_second.trace_id,
+        filter_logic="or",
+    )
+    assert [span.span_id for span in results] == [span_first.span_id]
+
+
+@pytest.mark.asyncio
+async def test_query_spans_supports_name_contains_filter(store_fixture: LightningStore) -> None:
+    """name_contains filtering should happen before pagination."""
+    attempted = await store_fixture.start_rollout(input={"payload": "span-names"})
+    attempt_id = attempted.attempt.attempt_id
+
+    def build_span(idx: int, name: str) -> Span:
+        span_hex = f"{idx:016x}"
+        trace_hex = f"{idx:032x}"
+        return Span(
+            rollout_id=attempted.rollout_id,
+            attempt_id=attempt_id,
+            sequence_id=idx,
+            trace_id=trace_hex,
+            span_id=span_hex,
+            parent_id=None,
+            name=name,
+            status=TraceStatus(status_code="OK"),
+            attributes={},
+            events=[],
+            links=[],
+            start_time=None,
+            end_time=None,
+            context=SpanContext(trace_id=trace_hex, span_id=span_hex, is_remote=False, trace_state={}),
+            parent=None,
+            resource=OtelResource(attributes={}, schema_url=""),
+        )
+
+    matching = build_span(1, "planner-step")
+    non_matching = build_span(2, "tool-call")
+    await store_fixture.add_span(matching)
+    await store_fixture.add_span(non_matching)
+
+    results = await store_fixture.query_spans(
+        attempted.rollout_id,
+        name_contains="plan",
+        sort_by="sequence_id",
+        limit=1,
+    )
+    assert [span.span_id for span in results] == [matching.span_id]
+
+
+@pytest.mark.asyncio
+async def test_query_spans_multiple_filters_require_all(store_fixture: LightningStore) -> None:
+    """Using multiple exact/substring filters together should narrow down to a single span."""
+    attempted = await store_fixture.start_rollout(input={"payload": "span-multi"})
+    attempt_id = attempted.attempt.attempt_id
+
+    def build_span(seq: int, parent: Optional[str], name: str) -> Span:
+        span_hex = f"{seq:016x}"
+        trace_hex = f"{(seq * 10):032x}"
+        return Span(
+            rollout_id=attempted.rollout_id,
+            attempt_id=attempt_id,
+            sequence_id=seq,
+            trace_id=trace_hex,
+            span_id=span_hex,
+            parent_id=parent,
+            name=name,
+            status=TraceStatus(status_code="OK"),
+            attributes={},
+            events=[],
+            links=[],
+            start_time=None,
+            end_time=None,
+            context=SpanContext(trace_id=trace_hex, span_id=span_hex, is_remote=False, trace_state={}),
+            parent=None,
+            resource=OtelResource(attributes={}, schema_url=""),
+        )
+
+    spans = [
+        build_span(1, None, "phase-plan"),
+        build_span(2, f"{1:016x}", "phase-run"),
+        build_span(3, f"{2:016x}", "tool-call"),
+    ]
+    for span in spans:
+        await store_fixture.add_span(span)
+
+    target = spans[2]
+    filtered = await store_fixture.query_spans(
+        attempted.rollout_id,
+        trace_id_contains=target.trace_id[-4:],
+        span_id_contains=target.span_id[-4:],
+        parent_id=target.parent_id,
+        name_contains="tool",
+    )
+    assert [span.span_id for span in filtered] == [target.span_id]
+
+
+@pytest.mark.asyncio
+async def test_query_spans_reports_pagination_metadata(store_fixture: LightningStore) -> None:
+    """Span pagination should return limit/offset/total values."""
+    attempted = await store_fixture.start_rollout(input={"payload": "span-pagination"})
+    attempt_id = attempted.attempt.attempt_id
+
+    def build_span(idx: int) -> Span:
+        span_hex = f"{idx:016x}"
+        trace_hex = f"{idx:032x}"
+        return Span(
+            rollout_id=attempted.rollout_id,
+            attempt_id=attempt_id,
+            sequence_id=idx,
+            trace_id=trace_hex,
+            span_id=span_hex,
+            parent_id=None,
+            name=f"span-{idx}",
+            status=TraceStatus(status_code="OK"),
+            attributes={},
+            events=[],
+            links=[],
+            start_time=None,
+            end_time=None,
+            context=SpanContext(trace_id=trace_hex, span_id=span_hex, is_remote=False, trace_state={}),
+            parent=None,
+            resource=OtelResource(attributes={}, schema_url=""),
+        )
+
+    for idx in range(1, 4):
+        await store_fixture.add_span(build_span(idx))
+
+    paginated = await store_fixture.query_spans(
+        attempted.rollout_id,
+        attempt_id=attempt_id,
+        sort_by="sequence_id",
+        sort_order="asc",
+        limit=1,
+        offset=1,
+    )
+    assert isinstance(paginated, PaginatedResult)
+    assert paginated.limit == 1
+    assert paginated.offset == 1
+    assert paginated.total == 3
+
+
+@pytest.mark.asyncio
+async def test_query_spans_or_filters_with_sort_and_offset(store_fixture: LightningStore) -> None:
+    """OR logic combined with sort + pagination should return deterministic slices."""
+    attempted = await store_fixture.start_rollout(input={"payload": "span-or"})
+    attempt_id = attempted.attempt.attempt_id
+
+    def build_span(seq: int, name: str, parent: Optional[str]) -> Span:
+        span_hex = f"{seq:016x}"
+        trace_hex = f"{(seq * 3):032x}"
+        return Span(
+            rollout_id=attempted.rollout_id,
+            attempt_id=attempt_id,
+            sequence_id=seq,
+            trace_id=trace_hex,
+            span_id=span_hex,
+            parent_id=parent,
+            name=name,
+            status=TraceStatus(status_code="OK"),
+            attributes={},
+            events=[],
+            links=[],
+            start_time=None,
+            end_time=None,
+            context=SpanContext(trace_id=trace_hex, span_id=span_hex, is_remote=False, trace_state={}),
+            parent=None,
+            resource=OtelResource(attributes={}, schema_url=""),
+        )
+
+    span_plan = build_span(10, "phase-plan", None)
+    span_run = build_span(20, "phase-run", span_plan.span_id)
+    span_tool = build_span(30, "tool-call", span_run.span_id)
+    for span in [span_plan, span_run, span_tool]:
+        await store_fixture.add_span(span)
+
+    filtered = await store_fixture.query_spans(
+        attempted.rollout_id,
+        name_contains="phase",
+        parent_id_contains=span_run.span_id[-4:],
+        span_id=span_tool.span_id,
+        filter_logic="or",
+        sort_by="sequence_id",
+        sort_order="desc",
+        limit=1,
+        offset=1,
+    )
+
+    assert [span.span_id for span in filtered] == [span_run.span_id]
+
+
 @pytest.mark.asyncio
 async def test_span_eviction_removes_oldest_rollouts(mock_readable_span: Mock, monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr("agentlightning.store.memory._detect_total_memory_bytes", lambda: 100)
@@ -1545,15 +2228,15 @@ async def test_add_span_with_missing_attempt(store_fixture: LightningStore, mock
 async def test_query_empty_spans(store_fixture: LightningStore) -> None:
     """Test querying spans for non-existent rollout returns empty."""
     spans = await store_fixture.query_spans("nonexistent")
-    assert spans == []
+    assert len(spans) == 0
 
     # With attempt_id
     spans = await store_fixture.query_spans("nonexistent", attempt_id="attempt-1")
-    assert spans == []
+    assert len(spans) == 0
 
     # With latest
     spans = await store_fixture.query_spans("nonexistent", attempt_id="latest")
-    assert spans == []
+    assert len(spans) == 0
 
 
 @pytest.mark.asyncio
@@ -1562,7 +2245,7 @@ async def test_query_latest_with_no_spans(store_fixture: LightningStore) -> None
     rollout = await store_fixture.enqueue_rollout(input={"test": "data"})
 
     spans = await store_fixture.query_spans(rollout.rollout_id, attempt_id="latest")
-    assert spans == []
+    assert len(spans) == 0
 
 
 @pytest.mark.asyncio

tests/store/test_restful.py

@@ -21,7 +21,16 @@ from portpicker import pick_unused_port
 
 from agentlightning.store.client_server import LightningStoreClient, LightningStoreServer
 from agentlightning.store.memory import InMemoryLightningStore
-from agentlightning.types import LLM, AttemptedRollout, OtelResource, Rollout, Span, TraceStatus
+from agentlightning.types import (
+    LLM,
+    AttemptedRollout,
+    OtelResource,
+    PaginatedResult,
+    PromptTemplate,
+    Rollout,
+    Span,
+    TraceStatus,
+)
 
 
 def _make_span(rollout_id: str, attempt_id: str, sequence_id: int, name: str) -> Span:
@@ -418,9 +427,7 @@ async def test_rollouts_sorting_by_unsupported_field(
     ) as resp:
         assert resp.status == 400
         data = await resp.json()
-        assert (
-            data["detail"] == "Failed to sort items by nonexistent_field: nonexistent_field is not a field of Rollout"
-        )
+        assert "Invalid sort_by: nonexistent_field, allowed fields are: " in data["detail"]
 
 
 # Attempts Pagination and Sorting Tests
@@ -966,7 +973,130 @@ async def test_spans_sorting_by_unsupported_field(
     ) as resp:
         assert resp.status == 400
         data = await resp.json()
-        assert data["detail"] == "Failed to sort items by invalid_field: invalid_field is not a field of Span"
+        assert "Invalid sort_by: invalid_field, allowed fields are: " in data["detail"]
+
+
+# LightningStoreClient._request_json pagination metadata tests
+
+
+@pytest.mark.asyncio
+async def test_request_json_rollouts_returns_pagination_metadata(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    server, client, _session, _api_endpoint = server_client
+
+    for idx in range(3):
+        await server.enqueue_rollout(input={"idx": idx})
+
+    params = [
+        ("sort_by", "rollout_id"),
+        ("sort_order", "asc"),
+        ("limit", 1),
+        ("offset", 1),
+    ]
+    data = await client._request_json("get", "/rollouts", params=params)  # pyright: ignore[reportPrivateUsage]
+    assert data["limit"] == 1
+    assert data["offset"] == 1
+    assert data["total"] == 3
+    assert len(data["items"]) == 1
+
+
+@pytest.mark.asyncio
+async def test_request_json_attempts_returns_pagination_metadata(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    _server, client, _session, _api_endpoint = server_client
+
+    attempted = await client.start_rollout(input={"payload": "attempts"})
+    await client.start_attempt(attempted.rollout_id)
+    await client.start_attempt(attempted.rollout_id)
+
+    params = [
+        ("sort_by", "sequence_id"),
+        ("sort_order", "asc"),
+        ("limit", 1),
+        ("offset", 1),
+    ]
+    data = await client._request_json(  # pyright: ignore[reportPrivateUsage]
+        "get", f"/rollouts/{attempted.rollout_id}/attempts", params=params
+    )
+    assert data["limit"] == 1
+    assert data["offset"] == 1
+    assert data["total"] == 3
+    assert len(data["items"]) == 1
+
+
+@pytest.mark.asyncio
+async def test_request_json_resources_returns_pagination_metadata(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    _server, client, _session, _api_endpoint = server_client
+
+    alpha = PromptTemplate(resource_type="prompt_template", template="alpha", engine="jinja")
+    beta = PromptTemplate(resource_type="prompt_template", template="beta", engine="jinja")
+
+    await client.update_resources("manual-alpha", {"prompt": alpha})
+    await client.update_resources("manual-beta", {"prompt": beta})
+
+    params = [
+        ("sort_by", "resources_id"),
+        ("sort_order", "asc"),
+        ("limit", 1),
+        ("offset", 1),
+    ]
+    data = await client._request_json("get", "/resources", params=params)  # pyright: ignore[reportPrivateUsage]
+    assert data["limit"] == 1
+    assert data["offset"] == 1
+    assert data["total"] == 2
+    assert len(data["items"]) == 1
+
+
+@pytest.mark.asyncio
+async def test_request_json_workers_returns_pagination_metadata(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    _server, client, _session, _api_endpoint = server_client
+
+    for worker_id in ["worker-a", "worker-b", "worker-c"]:
+        await client.update_worker(worker_id, heartbeat_stats={"cpu": 0.1})
+
+    params = [
+        ("sort_by", "worker_id"),
+        ("sort_order", "asc"),
+        ("limit", 1),
+        ("offset", 1),
+    ]
+    data = await client._request_json("get", "/workers", params=params)  # pyright: ignore[reportPrivateUsage]
+    assert data["limit"] == 1
+    assert data["offset"] == 1
+    assert data["total"] == 3
+    assert len(data["items"]) == 1
+
+
+@pytest.mark.asyncio
+async def test_request_json_spans_returns_pagination_metadata(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    server, client, _session, _api_endpoint = server_client
+
+    attempted = await server.start_rollout(input={"span": "meta"})
+    attempt_id = attempted.attempt.attempt_id
+    for idx in range(1, 4):
+        await server.add_span(_make_span(attempted.rollout_id, attempt_id, idx, f"span-{idx}"))
+
+    params = [
+        ("rollout_id", attempted.rollout_id),
+        ("attempt_id", attempt_id),
+        ("sort_by", "sequence_id"),
+        ("sort_order", "asc"),
+        ("limit", 1),
+        ("offset", 1),
+    ]
+    data = await client._request_json("get", "/spans", params=params)  # pyright: ignore[reportPrivateUsage]
+    assert data["limit"] == 1
+    assert data["offset"] == 1
+    assert data["total"] == 3
+    assert len(data["items"]) == 1
 
 
 # Client Compatibility Tests
@@ -983,9 +1113,10 @@ async def test_client_query_rollouts_extracts_items(
     for i in range(5):
         await server.enqueue_rollout(input={"index": i})
 
-    # Query via client (should extract items and return List[Rollout])
+    # Query via client (should return PaginatedResult that behaves like a sequence)
     rollouts = await client.query_rollouts()
-    assert isinstance(rollouts, list)
+    assert isinstance(rollouts, PaginatedResult)
+    assert rollouts.total == 5
     assert len(rollouts) == 5
     for rollout in rollouts:
         assert isinstance(rollout, Rollout)