Store RESTful API updates (#259)

agentlightning/client.py

@@ -332,7 +332,9 @@ class DevTaskLoader(AgentLightningClient):
         if isinstance(resources, ResourcesUpdate):
             self._resources_update = resources
         else:
-            self._resources_update = ResourcesUpdate(resources_id="local", resources=resources)
+            self._resources_update = ResourcesUpdate(
+                resources_id="local", resources=resources, create_time=time.time(), update_time=time.time(), version=1
+            )
 
         # Store rollouts posted back to the loader for easy debugging of local runs
         self._rollouts: List[RolloutLegacy] = []

agentlightning/server.py

@@ -142,7 +142,13 @@ class ServerDataStore:
         async with self._resources_lock:
             resources = self._resource_versions.get(resources_id)
             if resources:
-                return ResourcesUpdate(resources_id=resources_id, resources=resources)
+                return ResourcesUpdate(
+                    resources_id=resources_id,
+                    resources=resources,
+                    create_time=time.time(),
+                    update_time=time.time(),
+                    version=1,
+                )
             return None
 
     async def get_latest_resources(self) -> Optional[ResourcesUpdate]:
@@ -357,7 +363,9 @@ class AgentLightningServer:
         if not self._store:
             raise RuntimeError("Store not initialized. The server may not be running.")
         resources_id = f"res-{uuid.uuid4()}"
-        update = ResourcesUpdate(resources_id=resources_id, resources=resources)
+        update = ResourcesUpdate(
+            resources_id=resources_id, resources=resources, create_time=time.time(), update_time=time.time(), version=1
+        )
         await self._store.update_resources(update)
         return resources_id
 

agentlightning/store/base.py

@@ -307,6 +307,17 @@ class LightningStore:
         """
         raise NotImplementedError()
 
+    async def query_resources(self) -> List[ResourcesUpdate]:
+        """List every stored resource snapshot in insertion order.
+
+        Returns:
+            A chronological list of [`ResourcesUpdate`][agentlightning.ResourcesUpdate] objects.
+
+        Raises:
+            NotImplementedError: Subclasses must implement retrieval.
+        """
+        raise NotImplementedError()
+
     async def get_resources_by_id(self, resources_id: str) -> Optional[ResourcesUpdate]:
         """Return a specific named resource snapshot by identifier.
 

agentlightning/store/client_server.py

@@ -9,14 +9,16 @@ import threading
 import time
 import traceback
 from contextlib import suppress
-from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Sequence
+from typing import Any, Awaitable, Callable, Dict, Generic, List, Literal, Optional, Sequence, TypeVar
 
 import aiohttp
 import uvicorn
-from fastapi import Body, FastAPI, HTTPException, Request, Response
+from fastapi import Body, Depends, FastAPI, HTTPException
+from fastapi import Query as FastAPIQuery
+from fastapi import Request, Response
 from fastapi.responses import JSONResponse
 from opentelemetry.sdk.trace import ReadableSpan
-from pydantic import BaseModel, TypeAdapter
+from pydantic import BaseModel, Field, TypeAdapter
 
 from agentlightning.types import (
     Attempt,
@@ -37,6 +39,15 @@ logger = logging.getLogger(__name__)
 
 AGL_API_V1_PREFIX = "/agl/v1"
 
+T = TypeVar("T")
+
+
+class PaginatedResponse(BaseModel, Generic[T]):
+    items: List[T]
+    limit: int
+    offset: int
+    total: int
+
 
 class RolloutRequest(BaseModel):
     input: TaskInput
@@ -47,8 +58,17 @@ class RolloutRequest(BaseModel):
 
 
 class QueryRolloutsRequest(BaseModel):
-    status: Optional[List[RolloutStatus]] = None
-    rollout_ids: Optional[List[str]] = None
+    status_in: Optional[List[RolloutStatus]] = Field(FastAPIQuery(default=None))
+    rollout_id_in: Optional[List[str]] = Field(FastAPIQuery(default=None))
+    rollout_id_contains: Optional[str] = None
+    # Pagination
+    limit: int = -1
+    offset: int = 0
+    # Sorting
+    sort_by: Optional[str] = None
+    sort_order: Literal["asc", "desc"] = "asc"
+    # Filtering logic
+    filter_logic: Literal["and", "or"] = "and"
 
 
 class WaitForRolloutsRequest(BaseModel):
@@ -81,6 +101,137 @@ class UpdateAttemptRequest(BaseModel):
     metadata: Optional[Dict[str, Any]] = None
 
 
+class QueryAttemptsRequest(BaseModel):
+    # Pagination
+    limit: int = -1
+    offset: int = 0
+    # Sorting
+    sort_by: Optional[str] = None
+    sort_order: Literal["asc", "desc"] = "asc"
+
+
+class QueryResourcesRequest(BaseModel):
+    # Pagination
+    limit: int = -1
+    offset: int = 0
+    # Sorting
+    sort_by: Optional[str] = None
+    sort_order: Literal["asc", "desc"] = "asc"
+
+
+class QuerySpansRequest(BaseModel):
+    rollout_id: str
+    attempt_id: Optional[str] = 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] = None
+    sort_order: Literal["asc", "desc"] = "asc"
+
+
+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 LightningStoreServer(LightningStore):
     """
     Server wrapper that exposes a LightningStore via HTTP API.
@@ -383,13 +534,29 @@ class LightningStoreServer(LightningStore):
                 metadata=request.metadata,
             )
 
-        @self.app.get(AGL_API_V1_PREFIX + "/rollouts", response_model=List[Rollout])
-        async def query_rollouts():  # pyright: ignore[reportUnusedFunction]
-            return await self.query_rollouts()
+        @self.app.get(AGL_API_V1_PREFIX + "/rollouts", response_model=PaginatedResponse[Rollout])
+        async def query_rollouts(params: QueryRolloutsRequest = Depends()):  # pyright: ignore[reportUnusedFunction]
+            # Get all rollouts from the underlying store
+            all_rollouts = await self.query_rollouts()
 
-        @self.app.post(AGL_API_V1_PREFIX + "/rollouts/search", response_model=List[Rollout])
-        async def search_rollouts(request: QueryRolloutsRequest):  # pyright: ignore[reportUnusedFunction]
-            return await self.query_rollouts(status=request.status, rollout_ids=request.rollout_ids)
+            # 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,
+            )
 
         @self.app.get(AGL_API_V1_PREFIX + "/rollouts/{rollout_id}", response_model=Rollout)
         async def get_rollout_by_id(rollout_id: str):  # pyright: ignore[reportUnusedFunction]
@@ -439,14 +606,42 @@ class LightningStoreServer(LightningStore):
                 metadata=_get_mandatory_field_or_unset(request, "metadata"),
             )
 
-        @self.app.get(AGL_API_V1_PREFIX + "/rollouts/{rollout_id}/attempts", response_model=List[Attempt])
-        async def query_attempts(rollout_id: str):  # pyright: ignore[reportUnusedFunction]
-            return await self.query_attempts(rollout_id)
+        @self.app.get(AGL_API_V1_PREFIX + "/rollouts/{rollout_id}/attempts", response_model=PaginatedResponse[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,
+            )
 
         @self.app.get(AGL_API_V1_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)
 
+        @self.app.get(AGL_API_V1_PREFIX + "/resources", response_model=PaginatedResponse[ResourcesUpdate])
+        async def query_resources(params: QueryResourcesRequest = Depends()):  # pyright: ignore[reportUnusedFunction]
+            # Get all resources
+            all_resources = await self.query_resources()
+
+            return _apply_filters_sort_paginate(
+                all_resources,
+                {},  # No filters for resources
+                "and",
+                params.sort_by,
+                params.sort_order,
+                params.limit,
+                params.offset,
+            )
+
         @self.app.post(AGL_API_V1_PREFIX + "/resources", status_code=201, response_model=ResourcesUpdate)
         async def add_resources(resources: NamedResources):  # pyright: ignore[reportUnusedFunction]
             return await self.add_resources(resources)
@@ -469,12 +664,33 @@ class LightningStoreServer(LightningStore):
         async def add_span(span: Span):  # pyright: ignore[reportUnusedFunction]
             return await self.add_span(span)
 
-        @self.app.get(AGL_API_V1_PREFIX + "/spans", response_model=List[Span])
-        async def query_spans(  # pyright: ignore[reportUnusedFunction]
-            rollout_id: str,
-            attempt_id: Optional[str] = None,
-        ):
-            return await self.query_spans(rollout_id, attempt_id)
+        @self.app.get(AGL_API_V1_PREFIX + "/spans", response_model=PaginatedResponse[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
+            )
 
         @self.app.post(AGL_API_V1_PREFIX + "/spans/next", response_model=NextSequenceIdResponse)
         async def get_next_span_sequence_id(request: NextSequenceIdRequest):  # pyright: ignore[reportUnusedFunction]
@@ -549,6 +765,9 @@ class LightningStoreServer(LightningStore):
     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 get_rollout_by_id(self, rollout_id: str) -> Optional[Rollout]:
         return await self._call_store_method("get_rollout_by_id", rollout_id)
 
@@ -908,19 +1127,20 @@ class LightningStoreClient(LightningStore):
     async def query_rollouts(
         self, *, status: Optional[Sequence[RolloutStatus]] = None, rollout_ids: Optional[Sequence[str]] = None
     ) -> List[Rollout]:
-        if status or rollout_ids:
-            payload = QueryRolloutsRequest(
-                status=list(status) if status else None,
-                rollout_ids=list(rollout_ids) if rollout_ids else None,
-            ).model_dump(exclude_none=True)
-            data = await self._request_json("post", "/rollouts/search", json=payload)
-        else:
-            data = await self._request_json("get", "/rollouts")
-        return [Rollout.model_validate(item) for item in data]
+        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)
+
+        data = await self._request_json("get", "/rollouts", params=params if params else None)
+        # Extract items from PaginatedResponse
+        return [Rollout.model_validate(item) for item in data["items"]]
 
     async def query_attempts(self, rollout_id: str) -> List[Attempt]:
         data = await self._request_json("get", f"/rollouts/{rollout_id}/attempts")
-        return [Attempt.model_validate(item) for item in data]
+        # Extract items from PaginatedResponse
+        return [Attempt.model_validate(item) for item in data["items"]]
 
     async def get_latest_attempt(self, rollout_id: str) -> Optional[Attempt]:
         """
@@ -964,6 +1184,16 @@ class LightningStoreClient(LightningStore):
             logger.error(f"get_rollout_by_id failed after all retries for rollout_id={rollout_id}: {e}", exc_info=True)
             return None
 
+    async def query_resources(self) -> List[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"]]
+
     async def add_resources(self, resources: NamedResources) -> ResourcesUpdate:
         data = await self._request_json("post", "/resources", json=TypeAdapter(NamedResources).dump_python(resources))
         return ResourcesUpdate.model_validate(data)
@@ -1077,7 +1307,8 @@ class LightningStoreClient(LightningStore):
         if attempt_id is not None:
             params["attempt_id"] = attempt_id
         data = await self._request_json("get", "/spans", params=params)
-        return [Span.model_validate(item) for item in data]
+        # Extract items from PaginatedResponse
+        return [Span.model_validate(item) for item in data["items"]]
 
     async def update_rollout(
         self,

agentlightning/store/memory.py

@@ -26,6 +26,7 @@ from typing import (
     Sequence,
     Set,
     TypeVar,
+    Union,
     cast,
 )
 
@@ -439,16 +440,40 @@ class InMemoryLightningStore(LightningStore):
                 status_set = set(status)
                 rollouts = [rollout for rollout in rollouts if rollout.status in status_set]
 
+            # Attach the latest attempt to the rollout objects
+            rollouts = [self._rollout_to_attempted_rollout_unlocked(rollout) for rollout in rollouts]
+
             return rollouts
 
     @_healthcheck_wrapper
-    async def get_rollout_by_id(self, rollout_id: str) -> Optional[Rollout]:
+    async def get_rollout_by_id(self, rollout_id: str) -> Optional[Union[Rollout, AttemptedRollout]]:
         """Retrieves a specific rollout by its ID.
 
         See [`LightningStore.get_rollout_by_id()`][agentlightning.LightningStore.get_rollout_by_id] for semantics.
+
+        If the rollout has been attempted, the latest attempt will also be returned.
         """
         async with self._lock:
-            return self._rollouts.get(rollout_id)
+            rollout = self._rollouts.get(rollout_id)
+            if rollout is None:
+                return None
+            return self._rollout_to_attempted_rollout_unlocked(rollout)
+
+    def _rollout_to_attempted_rollout_unlocked(self, rollout: Rollout) -> Union[Rollout, AttemptedRollout]:
+        """Query the latest attempt for the rollout, and attach it to the rollout object.
+
+        If the rollout has no attempts, return the rollout object itself.
+        """
+        latest_attempt = self._get_latest_attempt_unlocked(rollout.rollout_id)
+        if latest_attempt is None:
+            return rollout
+        else:
+            return AttemptedRollout(**rollout.model_dump(), attempt=latest_attempt)
+
+    def _get_latest_attempt_unlocked(self, rollout_id: str) -> Optional[Attempt]:
+        """The unlocked version of `get_latest_attempt`."""
+        attempts = self._attempts.get(rollout_id, [])
+        return max(attempts, key=lambda a: a.sequence_id) if attempts else None
 
     @_healthcheck_wrapper
     async def query_attempts(self, rollout_id: str) -> List[Attempt]:
@@ -467,10 +492,13 @@ class InMemoryLightningStore(LightningStore):
         See [`LightningStore.get_latest_attempt()`][agentlightning.LightningStore.get_latest_attempt] for semantics.
         """
         async with self._lock:
-            attempts = self._attempts.get(rollout_id, [])
-            if not attempts:
-                return None
-            return max(attempts, key=lambda a: a.sequence_id)
+            return self._get_latest_attempt_unlocked(rollout_id)
+
+    @_healthcheck_wrapper
+    async def query_resources(self) -> List[ResourcesUpdate]:
+        """Return every stored resource snapshot in insertion order."""
+        async with self._lock:
+            return list(self._resources.values())
 
     @_healthcheck_wrapper
     async def add_resources(self, resources: NamedResources) -> ResourcesUpdate:
@@ -480,7 +508,14 @@ class InMemoryLightningStore(LightningStore):
         """
         resources_id = _generate_resources_id()
         async with self._lock:
-            update = ResourcesUpdate(resources_id=resources_id, resources=resources)
+            current_time = time.time()
+            update = ResourcesUpdate(
+                resources_id=resources_id,
+                resources=resources,
+                create_time=current_time,
+                update_time=current_time,
+                version=1,
+            )
             self._resources[resources_id] = update
             self._latest_resources_id = resources_id
             return update
@@ -493,7 +528,23 @@ class InMemoryLightningStore(LightningStore):
         See [`LightningStore.update_resources()`][agentlightning.LightningStore.update_resources] for semantics.
         """
         async with self._lock:
-            update = ResourcesUpdate(resources_id=resources_id, resources=resources)
+            current_time = time.time()
+            if resources_id not in self._resources:
+                update = ResourcesUpdate(
+                    resources_id=resources_id,
+                    resources=resources,
+                    create_time=current_time,
+                    update_time=current_time,
+                    version=1,
+                )
+            else:
+                update = self._resources[resources_id].model_copy(
+                    update={
+                        "resources": resources,
+                        "update_time": current_time,
+                        "version": self._resources[resources_id].version + 1,
+                    }
+                )
             self._resources[resources_id] = update
             self._latest_resources_id = resources_id
             return update

agentlightning/types/resources.py

@@ -194,5 +194,11 @@ class ResourcesUpdate(BaseModel):
 
     resources_id: str
     """Identifier used to version the resources."""
+    create_time: float
+    """Timestamp of the creation time of the resources."""
+    update_time: float
+    """Timestamp of the last update time of the resources."""
+    version: int
+    """Version of the resources."""
     resources: NamedResources
     """Mapping of resource names to their definitions."""

docs/reference/restful.md

@@ -0,0 +1,21 @@
+# RESTful API References
+
+!!! warning
+
+    The following contents are still under construction.
+
+## Store RESTful API
+
+<div id="swagger-ui"></div>
+<link rel="stylesheet" href="https://unpkg.com/swagger-ui-dist/swagger-ui.css" />
+<script src="https://unpkg.com/swagger-ui-dist/swagger-ui-bundle.js"></script>
+<script>
+  window.onload = () => {
+    window.ui = SwaggerUIBundle({
+      url: '../../assets/store-openapi.json',
+      dom_id: '#swagger-ui',
+      presets: [SwaggerUIBundle.presets.apis],
+      layout: "BaseLayout"
+    });
+  }
+</script>

mkdocs.yml

@@ -122,6 +122,7 @@ nav:
     - Store: reference/store.md
     - Trainer: reference/trainer.md
     - Types: reference/types.md
+    - RESTful: reference/restful.md
     - Internal: reference/internal.md
   - Miscellaneous:
     - Contributing Guide: community/contributing.md

scripts/export_openapi.py

@@ -0,0 +1,26 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Generate OpenAPI specification for the LightningStore server.
+
+Run this every time when you make changes to the LightningStore server.
+"""
+
+import asyncio
+import json
+
+from agentlightning.store.client_server import LightningStoreServer
+from agentlightning.store.memory import InMemoryLightningStore
+
+
+async def main():
+    store = InMemoryLightningStore()
+    server = LightningStoreServer(store, host="0.0.0.0", port=23333)
+    await server.start()
+
+    with open("docs/assets/store-openapi.json", "w") as f:
+        json.dump(server.app.openapi(), f)  # type: ignore
+    await server.stop()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

tests/store/test_client_server.py

@@ -180,6 +180,43 @@ async def test_add_resources_via_client(server_client: Tuple[LightningStoreServe
     assert latest.resources_id == resources_update.resources_id
 
 
+@pytest.mark.asyncio
+async def test_query_resources_history(server_client: Tuple[LightningStoreServer, LightningStoreClient]) -> None:
+    """Server and client should return identical resource history ordering."""
+    server, client = server_client
+
+    assert await server.query_resources() == []
+    assert await client.query_resources() == []
+
+    first = await server.add_resources(
+        cast(
+            Any,
+            {
+                "llm": LLM(
+                    resource_type="llm",
+                    endpoint="http://localhost:8000",
+                    model="hist-model-1",
+                    sampling_parameters={},
+                )
+            },
+        )
+    )
+    second = await server.update_resources(
+        "manual-id",
+        cast(
+            Any,
+            {"prompt": PromptTemplate(resource_type="prompt_template", template="Hi {user}", engine="f-string")},
+        ),
+    )
+
+    server_history = await server.query_resources()
+    client_history = await client.query_resources()
+
+    expected_ids = [first.resources_id, second.resources_id]
+    assert sorted([item.resources_id for item in server_history]) == sorted(expected_ids)
+    assert sorted([item.resources_id for item in client_history]) == sorted(expected_ids)
+
+
 @pytest.mark.asyncio
 async def test_client_server_end_to_end(
     server_client: Tuple[LightningStoreServer, LightningStoreClient], mock_readable_span: ReadableSpan

tests/store/test_memory.py

@@ -157,6 +157,47 @@ async def test_query_rollouts_by_status(inmemory_store: InMemoryLightningStore)
     assert len(none) == 0
 
 
+@pytest.mark.asyncio
+async def test_query_rollouts_returns_latest_attempt(inmemory_store: InMemoryLightningStore) -> None:
+    """Querying rollouts should attach the most recent attempt when present."""
+    attempted = await inmemory_store.start_rollout(input={"sample": "latest"})
+    latest_attempt = await inmemory_store.start_attempt(attempted.rollout_id)
+
+    results = await inmemory_store.query_rollouts(rollout_ids=[attempted.rollout_id])
+    assert len(results) == 1
+
+    retrieved = results[0]
+    assert type(retrieved) is AttemptedRollout
+    assert retrieved.attempt.attempt_id == latest_attempt.attempt.attempt_id
+    assert retrieved.attempt.sequence_id == latest_attempt.attempt.sequence_id
+
+
+@pytest.mark.asyncio
+async def test_get_rollout_by_id_returns_latest_attempt(inmemory_store: InMemoryLightningStore) -> None:
+    """Fetching a rollout by ID should include the latest attempt when available."""
+    attempted = await inmemory_store.start_rollout(input={"foo": "bar"})
+    second_attempt = await inmemory_store.start_attempt(attempted.rollout_id)
+
+    retrieved = await inmemory_store.get_rollout_by_id(attempted.rollout_id)
+    assert retrieved is not None
+    assert type(retrieved) is AttemptedRollout
+    assert retrieved.attempt.attempt_id == second_attempt.attempt.attempt_id
+    assert retrieved.attempt.sequence_id == second_attempt.attempt.sequence_id
+
+
+@pytest.mark.asyncio
+async def test_get_rollout_by_id_without_attempt_returns_rollout(
+    inmemory_store: InMemoryLightningStore,
+) -> None:
+    """Rollouts with no attempts should be returned without the Attempt wrapper."""
+    queued = await inmemory_store.enqueue_rollout(input={"foo": "bar"})
+
+    retrieved = await inmemory_store.get_rollout_by_id(queued.rollout_id)
+    assert retrieved is not None
+    assert type(retrieved) is Rollout
+    assert not hasattr(retrieved, "attempt")
+
+
 @pytest.mark.asyncio
 async def test_get_rollout_by_id(inmemory_store: InMemoryLightningStore) -> None:
     """Test retrieving rollouts by their ID."""
@@ -474,6 +515,29 @@ async def test_add_resources_multiple_times_generates_unique_ids(inmemory_store:
     assert latest.resources_id == update2.resources_id
 
 
+@pytest.mark.asyncio
+async def test_query_resources_returns_history(inmemory_store: InMemoryLightningStore) -> None:
+    """query_resources should list snapshots in the order they were stored."""
+    assert await inmemory_store.query_resources() == []
+
+    first = await inmemory_store.add_resources(
+        {
+            "llm": LLM(resource_type="llm", endpoint="http://localhost:8080", model="model-v1"),
+        }
+    )
+    second = await inmemory_store.update_resources(
+        "custom-snapshot",
+        {
+            "prompt": PromptTemplate(resource_type="prompt_template", template="Hi {name}", engine="f-string"),
+        },
+    )
+
+    history = await inmemory_store.query_resources()
+    assert [item.resources_id for item in history] == [first.resources_id, second.resources_id]
+    assert isinstance(history[0], ResourcesUpdate)
+    assert isinstance(history[1], ResourcesUpdate)
+
+
 @pytest.mark.asyncio
 async def test_resource_lifecycle(inmemory_store: InMemoryLightningStore) -> None:
     """Test adding, updating, and retrieving resources."""
@@ -526,7 +590,13 @@ async def test_task_inherits_latest_resources(inmemory_store: InMemoryLightningS
     """Test that new tasks inherit latest resources_id if not specified."""
     # Set up resources with proper PromptTemplate
     prompt = PromptTemplate(resource_type="prompt_template", template="Hello {name}!", engine="f-string")
-    update = ResourcesUpdate(resources_id="current", resources={"greeting": prompt})
+    update = ResourcesUpdate(
+        resources_id="current",
+        resources={"greeting": prompt},
+        create_time=time.time(),
+        update_time=time.time(),
+        version=1,
+    )
     await inmemory_store.update_resources(update.resources_id, update.resources)
 
     # Task without explicit resources_id
@@ -539,7 +609,13 @@ async def test_task_inherits_latest_resources(inmemory_store: InMemoryLightningS
 
     # Update resources
     new_prompt = PromptTemplate(resource_type="prompt_template", template="Hi {name}!", engine="f-string")
-    update2 = ResourcesUpdate(resources_id="new", resources={"greeting": new_prompt})
+    update2 = ResourcesUpdate(
+        resources_id="new",
+        resources={"greeting": new_prompt},
+        create_time=time.time(),
+        update_time=time.time(),
+        version=1,
+    )
     await inmemory_store.update_resources(update2.resources_id, update2.resources)
 
     # New task gets new resources
@@ -1261,7 +1337,9 @@ async def test_concurrent_resource_updates(inmemory_store: InMemoryLightningStor
             model=f"model-v{ver}",
             sampling_parameters={"temperature": 0.5 + ver * 0.01},
         )
-        update = ResourcesUpdate(resources_id=f"v{ver}", resources={"llm": llm})
+        update = ResourcesUpdate(
+            resources_id=f"v{ver}", resources={"llm": llm}, create_time=time.time(), update_time=time.time(), version=1
+        )
         await inmemory_store.update_resources(update.resources_id, update.resources)
 
     # Update concurrently
@@ -2004,3 +2082,139 @@ async def test_requeued_attempt_recovers_after_retry_started(
     assert rollout.status == "preparing"
 
     assert await inmemory_store.dequeue_rollout() is None
+
+
+@pytest.mark.asyncio
+async def test_resources_update_tracks_create_and_update_times(inmemory_store: InMemoryLightningStore) -> None:
+    """Test that ResourcesUpdate tracks create_time and update_time correctly."""
+    llm = LLM(
+        resource_type="llm",
+        endpoint="http://localhost:8080",
+        model="test-model",
+        sampling_parameters={"temperature": 0.7},
+    )
+
+    # Add initial resource
+    start_time = time.time()
+    update1 = await inmemory_store.add_resources({"main_llm": llm})
+
+    # Verify create_time is set and reasonable
+    assert update1.create_time >= start_time
+    assert update1.create_time <= time.time()
+
+    # Initially, update_time should equal create_time
+    assert update1.update_time == update1.create_time
+    assert update1.version == 1
+
+    # Wait a bit and update the same resource
+    await asyncio.sleep(0.01)
+    llm_v2 = LLM(
+        resource_type="llm",
+        endpoint="http://localhost:8080",
+        model="test-model-v2",
+        sampling_parameters={"temperature": 0.8},
+    )
+    update2 = await inmemory_store.update_resources(update1.resources_id, {"main_llm": llm_v2})
+
+    # Verify update_time changed but create_time stayed the same
+    assert update2.resources_id == update1.resources_id
+    assert update2.create_time == update1.create_time  # create_time should not change
+    assert update2.update_time > update1.update_time  # update_time should be newer
+    assert update2.version == 2  # version should increment
+
+
+@pytest.mark.asyncio
+async def test_resources_update_version_increments(inmemory_store: InMemoryLightningStore) -> None:
+    """Test that ResourcesUpdate version increments correctly with each update."""
+    llm = LLM(
+        resource_type="llm",
+        endpoint="http://localhost:8080",
+        model="test-model",
+        sampling_parameters={"temperature": 0.7},
+    )
+
+    # Add initial resource
+    update1 = await inmemory_store.add_resources({"main_llm": llm})
+    assert update1.version == 1
+
+    # Update it multiple times
+    for i in range(2, 6):
+        llm_updated = LLM(
+            resource_type="llm",
+            endpoint="http://localhost:8080",
+            model=f"test-model-v{i}",
+            sampling_parameters={"temperature": 0.7},
+        )
+        update = await inmemory_store.update_resources(update1.resources_id, {"main_llm": llm_updated})
+        assert update.version == i
+        assert update.resources_id == update1.resources_id
+        assert update.create_time == update1.create_time
+
+
+@pytest.mark.asyncio
+async def test_resources_different_ids_have_independent_versions(inmemory_store: InMemoryLightningStore) -> None:
+    """Test that different resources_ids have independent version counters."""
+    llm1 = LLM(
+        resource_type="llm",
+        endpoint="http://localhost:8080",
+        model="model-1",
+        sampling_parameters={"temperature": 0.7},
+    )
+    llm2 = LLM(
+        resource_type="llm",
+        endpoint="http://localhost:8080",
+        model="model-2",
+        sampling_parameters={"temperature": 0.8},
+    )
+
+    # Add two different resources
+    res1 = await inmemory_store.add_resources({"llm": llm1})
+    res2 = await inmemory_store.add_resources({"llm": llm2})
+
+    # Both should start at version 1
+    assert res1.version == 1
+    assert res2.version == 1
+    assert res1.resources_id != res2.resources_id
+
+    # Update res1 twice
+    for i in range(2):
+        llm_updated = LLM(
+            resource_type="llm",
+            endpoint="http://localhost:8080",
+            model=f"model-1-v{i+2}",
+            sampling_parameters={"temperature": 0.7},
+        )
+        res1 = await inmemory_store.update_resources(res1.resources_id, {"llm": llm_updated})
+
+    # res1 should be at version 3, res2 should still be at version 1
+    assert res1.version == 3
+    retrieved_res2 = await inmemory_store.get_resources_by_id(res2.resources_id)
+    assert retrieved_res2 is not None
+    assert retrieved_res2.version == 1
+
+
+@pytest.mark.asyncio
+async def test_query_resources_returns_all_fields(inmemory_store: InMemoryLightningStore) -> None:
+    """Test that query_resources returns all ResourcesUpdate fields."""
+    llm = LLM(
+        resource_type="llm",
+        endpoint="http://localhost:8080",
+        model="test-model",
+        sampling_parameters={"temperature": 0.7},
+    )
+
+    # Add multiple resources
+    await inmemory_store.add_resources({"llm": llm})
+    await asyncio.sleep(0.01)
+    await inmemory_store.add_resources({"llm": llm})
+
+    # Query all resources
+    all_resources = await inmemory_store.query_resources()
+
+    assert len(all_resources) == 2
+    for res in all_resources:
+        assert res.resources_id is not None
+        assert res.create_time > 0
+        assert res.update_time > 0
+        assert res.version >= 1
+        assert res.resources is not None

tests/store/test_restful.py

@@ -0,0 +1,872 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""
+Tests for RESTful API pagination, sorting, and filtering functionality.
+
+Test categories:
+- Rollouts pagination, sorting, and filtering
+- Attempts pagination and sorting
+- Resources pagination and sorting
+- Spans pagination, sorting, and filtering
+"""
+
+import asyncio
+import contextlib
+import socket
+from typing import AsyncGenerator, List, Tuple
+
+import aiohttp
+import pytest
+import pytest_asyncio
+
+from agentlightning.store.client_server import LightningStoreClient, LightningStoreServer
+from agentlightning.store.memory import InMemoryLightningStore
+from agentlightning.types import LLM, AttemptedRollout, OtelResource, Rollout, Span, TraceStatus
+
+
+def _get_free_port() -> int:
+    with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+def _make_span(rollout_id: str, attempt_id: str, sequence_id: int, name: str) -> Span:
+    return Span(
+        rollout_id=rollout_id,
+        attempt_id=attempt_id,
+        sequence_id=sequence_id,
+        trace_id=f"{sequence_id:032x}",
+        span_id=f"{sequence_id:016x}",
+        parent_id=None,
+        name=name,
+        status=TraceStatus(status_code="OK"),
+        attributes={},
+        events=[],
+        links=[],
+        start_time=1.0,
+        end_time=2.0,
+        context=None,
+        parent=None,
+        resource=OtelResource(attributes={}, schema_url=""),
+    )
+
+
+@pytest_asyncio.fixture
+async def server_client() -> (
+    AsyncGenerator[Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str], None]
+):
+    store = InMemoryLightningStore()
+    port = _get_free_port()
+    server = LightningStoreServer(store, "127.0.0.1", port)
+    await server.start()
+    client = LightningStoreClient(server.endpoint)
+    session = aiohttp.ClientSession()
+    # Get the full API endpoint with /agl/v1 prefix
+    api_endpoint = client.server_address
+    try:
+        yield server, client, session, api_endpoint
+    finally:
+        await session.close()
+        await client.close()
+        await server.stop()
+
+
+# Rollouts Pagination, Sorting, and Filtering Tests
+
+
+@pytest.mark.asyncio
+async def test_rollouts_pagination_basic(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test basic pagination for rollouts endpoint."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create 10 rollouts
+    for i in range(10):
+        await server.enqueue_rollout(input={"index": i})
+
+    # Get first page with limit=3
+    async with session.get(f"{api_endpoint}/rollouts", params={"limit": 3, "offset": 0}) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] == 10
+        assert data["limit"] == 3
+        assert data["offset"] == 0
+        assert len(data["items"]) == 3
+
+    # Get second page
+    async with session.get(f"{api_endpoint}/rollouts", params={"limit": 3, "offset": 3}) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] == 10
+        assert data["limit"] == 3
+        assert data["offset"] == 3
+        assert len(data["items"]) == 3
+
+
+@pytest.mark.asyncio
+async def test_rollouts_pagination_disabled(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test pagination can be disabled with limit=-1."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create 15 rollouts
+    for i in range(15):
+        await server.enqueue_rollout(input={"index": i})
+
+    # Get all rollouts with limit=-1
+    async with session.get(f"{api_endpoint}/rollouts", params={"limit": -1}) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] == 15
+        assert data["limit"] == -1
+        assert len(data["items"]) == 15
+
+
+@pytest.mark.asyncio
+async def test_rollouts_sorting_by_start_time(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test sorting rollouts by start_time."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollouts with small delays
+    rollouts: List[Rollout] = []
+    for i in range(5):
+        r = await server.enqueue_rollout(input={"index": i})
+        rollouts.append(r)
+        await asyncio.sleep(0.01)
+
+    # Sort ascending by start_time
+    async with session.get(
+        f"{api_endpoint}/rollouts", params={"sort_by": "start_time", "sort_order": "asc", "limit": -1}
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        items = data["items"]
+        assert len(items) == 5
+        # Should be in ascending order
+        for i in range(len(items) - 1):
+            assert items[i]["start_time"] <= items[i + 1]["start_time"]
+
+    # Sort descending by start_time (default)
+    async with session.get(
+        f"{api_endpoint}/rollouts", params={"sort_by": "start_time", "sort_order": "desc", "limit": -1}
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        items = data["items"]
+        # Should be in descending order
+        for i in range(len(items) - 1):
+            assert items[i]["start_time"] >= items[i + 1]["start_time"]
+
+
+@pytest.mark.asyncio
+async def test_rollouts_filter_by_status(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test filtering rollouts by status."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollouts with different statuses
+    r1 = await server.enqueue_rollout(input={"id": 1})
+    r2 = await server.enqueue_rollout(input={"id": 2})
+    _r3 = await server.enqueue_rollout(input={"id": 3})
+
+    await server.update_rollout(rollout_id=r1.rollout_id, status="succeeded")
+    await server.update_rollout(rollout_id=r2.rollout_id, status="failed")
+    # r3 remains queuing
+
+    # Filter by single status
+    async with session.get(f"{api_endpoint}/rollouts", params={"status_in": ["succeeded"], "limit": -1}) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] == 1
+        assert data["items"][0]["rollout_id"] == r1.rollout_id
+
+    # Filter by multiple statuses
+    async with session.get(
+        f"{api_endpoint}/rollouts", params={"status_in": ["succeeded", "failed"], "limit": -1}
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] == 2
+        rollout_ids = {item["rollout_id"] for item in data["items"]}
+        assert rollout_ids == {r1.rollout_id, r2.rollout_id}
+
+
+@pytest.mark.asyncio
+async def test_rollouts_filter_by_rollout_id_in(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test filtering rollouts by rollout_id_in."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create multiple rollouts
+    r1 = await server.enqueue_rollout(input={"id": 1})
+    _r2 = await server.enqueue_rollout(input={"id": 2})
+    r3 = await server.enqueue_rollout(input={"id": 3})
+
+    # Filter by specific rollout IDs
+    async with session.get(
+        f"{api_endpoint}/rollouts", params={"rollout_id_in": [r1.rollout_id, r3.rollout_id], "limit": -1}
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] == 2
+        rollout_ids = {item["rollout_id"] for item in data["items"]}
+        assert rollout_ids == {r1.rollout_id, r3.rollout_id}
+
+
+@pytest.mark.asyncio
+async def test_rollouts_filter_by_rollout_id_contains(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test filtering rollouts by rollout_id_contains."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollouts
+    r1 = await server.enqueue_rollout(input={"id": 1})
+    _r2 = await server.enqueue_rollout(input={"id": 2})
+    _r3 = await server.enqueue_rollout(input={"id": 3})
+
+    # Extract a substring from r1's ID
+    substring = r1.rollout_id[3:8]  # Get middle part of ID
+
+    # Filter by substring
+    async with session.get(f"{api_endpoint}/rollouts", params={"rollout_id_contains": substring, "limit": -1}) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] >= 1
+        # Verify all results contain the substring
+        for item in data["items"]:
+            assert substring in item["rollout_id"]
+
+
+@pytest.mark.asyncio
+async def test_rollouts_filter_logic_and(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test rollouts filtering with AND logic."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollouts
+    r1 = await server.enqueue_rollout(input={"id": 1})
+    r2 = await server.enqueue_rollout(input={"id": 2})
+    r3 = await server.enqueue_rollout(input={"id": 3})
+
+    await server.update_rollout(rollout_id=r1.rollout_id, status="succeeded")
+    await server.update_rollout(rollout_id=r2.rollout_id, status="succeeded")
+    await server.update_rollout(rollout_id=r3.rollout_id, status="failed")
+
+    # Filter with AND logic: status=succeeded AND rollout_id in list
+    async with session.get(
+        f"{api_endpoint}/rollouts",
+        params={
+            "status_in": ["succeeded"],
+            "rollout_id_in": [r1.rollout_id, r3.rollout_id],
+            "filter_logic": "and",
+            "limit": -1,
+        },
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        # Only r1 matches both conditions (r3 is failed, r2 is not in the ID list)
+        assert data["total"] == 1
+        assert data["items"][0]["rollout_id"] == r1.rollout_id
+
+
+@pytest.mark.asyncio
+async def test_rollouts_filter_logic_or(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test rollouts filtering with OR logic."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollouts
+    r1 = await server.enqueue_rollout(input={"id": 1})
+    _r2 = await server.enqueue_rollout(input={"id": 2})
+    r3 = await server.enqueue_rollout(input={"id": 3})
+
+    await server.update_rollout(rollout_id=r1.rollout_id, status="succeeded")
+    # r2 and r3 remain queuing
+
+    # Filter with OR logic: status=succeeded OR rollout_id=r3
+    async with session.get(
+        f"{api_endpoint}/rollouts",
+        params={"status_in": ["succeeded"], "rollout_id_in": [r3.rollout_id], "filter_logic": "or", "limit": -1},
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        # Both r1 (succeeded) and r3 (in ID list) should match
+        assert data["total"] == 2
+        rollout_ids = {item["rollout_id"] for item in data["items"]}
+        assert rollout_ids == {r1.rollout_id, r3.rollout_id}
+
+
+@pytest.mark.asyncio
+async def test_rollouts_sorting_with_none_values(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test sorting rollouts by fields that may have None values."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollouts - mode is optional and can be None
+    _r1 = await server.enqueue_rollout(input={"id": 1}, mode="train")
+    r2 = await server.enqueue_rollout(input={"id": 2})  # mode=None
+    _r3 = await server.enqueue_rollout(input={"id": 3}, mode="test")
+
+    # Sort by mode ascending (None values should be treated as empty string/0)
+    async with session.get(
+        f"{api_endpoint}/rollouts", params={"sort_by": "mode", "sort_order": "asc", "limit": -1}
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        items = data["items"]
+        assert len(items) == 3
+        # Items with None mode should come first (treated as 0)
+        assert items[0]["rollout_id"] == r2.rollout_id
+        assert items[0]["mode"] is None
+
+    # Sort by mode descending
+    async with session.get(
+        f"{api_endpoint}/rollouts", params={"sort_by": "mode", "sort_order": "desc", "limit": -1}
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        items = data["items"]
+        assert len(items) == 3
+        # Items with actual mode should come first
+        assert items[0]["mode"] is not None
+        # Item with None should be last
+        assert items[2]["rollout_id"] == r2.rollout_id
+        assert items[2]["mode"] is None
+
+
+@pytest.mark.asyncio
+async def test_rollouts_sorting_by_unsupported_field(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test sorting by a field that doesn't exist on the model."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollouts
+    for i in range(3):
+        await server.enqueue_rollout(input={"id": i})
+
+    # Try to sort by a non-existent field - should return 400 error
+    async with session.get(
+        f"{api_endpoint}/rollouts", params={"sort_by": "nonexistent_field", "sort_order": "asc", "limit": -1}
+    ) 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"
+        )
+
+
+# Attempts Pagination and Sorting Tests
+
+
+@pytest.mark.asyncio
+async def test_attempts_pagination_basic(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test basic pagination for attempts endpoint."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollout and multiple attempts
+    rollout = await server.enqueue_rollout(input={"test": "data"})
+    for _ in range(5):
+        await server.start_attempt(rollout.rollout_id)
+
+    # Get first page
+    async with session.get(
+        f"{api_endpoint}/rollouts/{rollout.rollout_id}/attempts", params={"limit": 2, "offset": 0}
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] == 5
+        assert data["limit"] == 2
+        assert data["offset"] == 0
+        assert len(data["items"]) == 2
+
+    # Get all attempts
+    async with session.get(f"{api_endpoint}/rollouts/{rollout.rollout_id}/attempts", params={"limit": -1}) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] == 5
+        assert len(data["items"]) == 5
+
+
+@pytest.mark.asyncio
+async def test_attempts_sorting(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test sorting attempts by start_time."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollout and attempts
+    rollout = await server.enqueue_rollout(input={"test": "data"})
+    attempts: List[AttemptedRollout] = []
+    for i in range(3):
+        attempt = await server.start_attempt(rollout.rollout_id)
+        attempts.append(attempt)
+        await asyncio.sleep(0.01)
+
+    # Sort by start_time descending (default)
+    async with session.get(
+        f"{api_endpoint}/rollouts/{rollout.rollout_id}/attempts",
+        params={"sort_by": "start_time", "sort_order": "desc", "limit": -1},
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        items = data["items"]
+        for i in range(len(items) - 1):
+            assert items[i]["start_time"] >= items[i + 1]["start_time"]
+
+    # Sort ascending
+    async with session.get(
+        f"{api_endpoint}/rollouts/{rollout.rollout_id}/attempts",
+        params={"sort_by": "start_time", "sort_order": "asc", "limit": -1},
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        items = data["items"]
+        for i in range(len(items) - 1):
+            assert items[i]["start_time"] <= items[i + 1]["start_time"]
+
+
+@pytest.mark.asyncio
+async def test_attempts_sorting_with_none_values(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test sorting attempts by fields that may have None values."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollout and attempts
+    rollout = await server.enqueue_rollout(input={"test": "data"})
+    a1 = await server.start_attempt(rollout.rollout_id)
+    a2 = await server.start_attempt(rollout.rollout_id)
+    a3 = await server.start_attempt(rollout.rollout_id)
+
+    # Set worker_id for some attempts, leave others with None
+    await server.update_attempt(rollout_id=rollout.rollout_id, attempt_id=a1.attempt.attempt_id, worker_id="worker-1")
+    await server.update_attempt(rollout_id=rollout.rollout_id, attempt_id=a3.attempt.attempt_id, worker_id="worker-2")
+    # a2 remains with worker_id=None
+
+    # Sort by worker_id ascending (None values should be treated as empty/0)
+    async with session.get(
+        f"{api_endpoint}/rollouts/{rollout.rollout_id}/attempts",
+        params={"sort_by": "worker_id", "sort_order": "asc", "limit": -1},
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        items = data["items"]
+        assert len(items) == 3
+        # Items with None worker_id should come first
+        assert items[0]["attempt_id"] == a2.attempt.attempt_id
+        assert items[0]["worker_id"] is None
+
+
+# Resources Pagination and Sorting Tests
+
+
+@pytest.mark.asyncio
+async def test_resources_pagination_basic(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test basic pagination for resources endpoint."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create multiple resources
+    for i in range(7):
+        llm = LLM(
+            resource_type="llm",
+            endpoint="http://localhost:8080",
+            model=f"model-v{i}",
+            sampling_parameters={"temperature": 0.7},
+        )
+        await server.add_resources({"llm": llm})
+
+    # Get first page
+    async with session.get(f"{api_endpoint}/resources", params={"limit": 3, "offset": 0}) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] == 7
+        assert data["limit"] == 3
+        assert len(data["items"]) == 3
+
+    # Get all resources
+    async with session.get(f"{api_endpoint}/resources", params={"limit": -1}) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] == 7
+        assert len(data["items"]) == 7
+
+
+@pytest.mark.asyncio
+async def test_resources_sorting_by_resources_id(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test sorting resources by resources_id."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create resources
+    for i in range(5):
+        llm = LLM(
+            resource_type="llm",
+            endpoint="http://localhost:8080",
+            model=f"model-{i}",
+            sampling_parameters={"temperature": 0.7},
+        )
+        await server.add_resources({"llm": llm})
+
+    # Sort by resources_id ascending (default)
+    async with session.get(
+        f"{api_endpoint}/resources", params={"sort_by": "resources_id", "sort_order": "asc", "limit": -1}
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        items = data["items"]
+        for i in range(len(items) - 1):
+            assert items[i]["resources_id"] <= items[i + 1]["resources_id"]
+
+    # Sort descending
+    async with session.get(
+        f"{api_endpoint}/resources", params={"sort_by": "resources_id", "sort_order": "desc", "limit": -1}
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        items = data["items"]
+        for i in range(len(items) - 1):
+            assert items[i]["resources_id"] >= items[i + 1]["resources_id"]
+
+
+# Spans Pagination, Sorting, and Filtering Tests
+
+
+@pytest.mark.asyncio
+async def test_spans_pagination_basic(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test basic pagination for spans endpoint."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollout and attempt
+    rollout = await server.start_rollout(input={"test": "data"})
+
+    # Add multiple spans
+    for i in range(10):
+        span = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, i + 1, f"span-{i}")
+        await server.add_span(span)
+
+    # Get first page
+    async with session.get(
+        f"{api_endpoint}/spans", params={"rollout_id": rollout.rollout_id, "limit": 3, "offset": 0}
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] == 10
+        assert data["limit"] == 3
+        assert len(data["items"]) == 3
+
+    # Get all spans
+    async with session.get(f"{api_endpoint}/spans", params={"rollout_id": rollout.rollout_id, "limit": -1}) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] == 10
+        assert len(data["items"]) == 10
+
+
+@pytest.mark.asyncio
+async def test_spans_sorting_by_start_time(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test sorting spans by start_time."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollout and spans
+    rollout = await server.start_rollout(input={"test": "data"})
+    for i in range(5):
+        span = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, i + 1, f"span-{i}")
+        await server.add_span(span)
+
+    # Sort descending (default)
+    async with session.get(
+        f"{api_endpoint}/spans",
+        params={"rollout_id": rollout.rollout_id, "sort_by": "start_time", "sort_order": "desc", "limit": -1},
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        items = data["items"]
+        # All our test spans have the same start_time, so just verify the structure
+        assert len(items) == 5
+
+
+@pytest.mark.asyncio
+async def test_spans_filter_by_trace_id(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test filtering spans by trace_id."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollout and spans with different trace IDs
+    rollout = await server.start_rollout(input={"test": "data"})
+
+    span1 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 1, "span1")
+    span1.trace_id = "trace-123"
+    await server.add_span(span1)
+
+    span2 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 2, "span2")
+    span2.trace_id = "trace-456"
+    await server.add_span(span2)
+
+    span3 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 3, "span3")
+    span3.trace_id = "trace-123"
+    await server.add_span(span3)
+
+    # Filter by exact trace_id
+    async with session.get(
+        f"{api_endpoint}/spans", params={"rollout_id": rollout.rollout_id, "trace_id": "trace-123", "limit": -1}
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] == 2
+        for item in data["items"]:
+            assert item["trace_id"] == "trace-123"
+
+
+@pytest.mark.asyncio
+async def test_spans_filter_by_name_contains(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test filtering spans by name_contains."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollout and spans with different names
+    rollout = await server.start_rollout(input={"test": "data"})
+
+    span1 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 1, "api_call_fetch")
+    await server.add_span(span1)
+
+    span2 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 2, "database_query")
+    await server.add_span(span2)
+
+    span3 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 3, "api_call_update")
+    await server.add_span(span3)
+
+    # Filter by name_contains
+    async with session.get(
+        f"{api_endpoint}/spans", params={"rollout_id": rollout.rollout_id, "name_contains": "api_call", "limit": -1}
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["total"] == 2
+        for item in data["items"]:
+            assert "api_call" in item["name"]
+
+
+@pytest.mark.asyncio
+async def test_spans_filter_logic_and(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test spans filtering with AND logic."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollout and spans
+    rollout = await server.start_rollout(input={"test": "data"})
+
+    span1 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 1, "api_call")
+    span1.trace_id = "trace-123"
+    await server.add_span(span1)
+
+    span2 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 2, "api_call")
+    span2.trace_id = "trace-456"
+    await server.add_span(span2)
+
+    span3 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 3, "database")
+    span3.trace_id = "trace-123"
+    await server.add_span(span3)
+
+    # Filter with AND logic: trace_id=trace-123 AND name contains "api"
+    async with session.get(
+        f"{api_endpoint}/spans",
+        params={
+            "rollout_id": rollout.rollout_id,
+            "trace_id": "trace-123",
+            "name_contains": "api",
+            "filter_logic": "and",
+            "limit": -1,
+        },
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        # Only span1 matches both conditions
+        assert data["total"] == 1
+        assert data["items"][0]["name"] == "api_call"
+        assert data["items"][0]["trace_id"] == "trace-123"
+
+
+@pytest.mark.asyncio
+async def test_spans_filter_logic_or(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test spans filtering with OR logic."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollout and spans
+    rollout = await server.start_rollout(input={"test": "data"})
+
+    span1 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 1, "api_call")
+    span1.trace_id = "trace-123"
+    await server.add_span(span1)
+
+    span2 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 2, "api_call")
+    span2.trace_id = "trace-456"
+    await server.add_span(span2)
+
+    span3 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 3, "database")
+    span3.trace_id = "trace-123"
+    await server.add_span(span3)
+
+    span4 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 4, "other")
+    span4.trace_id = "trace-789"
+    await server.add_span(span4)
+
+    # Filter with OR logic: trace_id=trace-123 OR name contains "api"
+    async with session.get(
+        f"{api_endpoint}/spans",
+        params={
+            "rollout_id": rollout.rollout_id,
+            "trace_id": "trace-123",
+            "name_contains": "api",
+            "filter_logic": "or",
+            "limit": -1,
+        },
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        # span1, span2, and span3 should match
+        assert data["total"] == 3
+
+
+@pytest.mark.asyncio
+async def test_spans_sorting_with_none_values(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test sorting spans by fields that may have None values."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollout and spans
+    rollout = await server.start_rollout(input={"test": "data"})
+
+    # Create spans with different parent_id values (parent_id can be None)
+    span1 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 1, "span1")
+    span1.parent_id = None  # Root span
+    await server.add_span(span1)
+
+    span2 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 2, "span2")
+    span2.parent_id = "parent-a"  # Child span
+    await server.add_span(span2)
+
+    span3 = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, 3, "span3")
+    span3.parent_id = "parent-b"  # Child span
+    await server.add_span(span3)
+
+    # Sort by parent_id ascending (None values should be treated as empty/0)
+    async with session.get(
+        f"{api_endpoint}/spans",
+        params={"rollout_id": rollout.rollout_id, "sort_by": "parent_id", "sort_order": "asc", "limit": -1},
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        items = data["items"]
+        assert len(items) == 3
+        # Items with None parent_id should come first (treated as 0)
+        assert items[0]["sequence_id"] == 1
+        assert items[0]["parent_id"] is None
+
+    # Sort by parent_id descending
+    async with session.get(
+        f"{api_endpoint}/spans",
+        params={"rollout_id": rollout.rollout_id, "sort_by": "parent_id", "sort_order": "desc", "limit": -1},
+    ) as resp:
+        assert resp.status == 200
+        data = await resp.json()
+        items = data["items"]
+        assert len(items) == 3
+        # Items with actual parent_id should come first
+        assert items[0]["parent_id"] is not None
+        # Item with None should be last
+        assert items[2]["sequence_id"] == 1
+        assert items[2]["parent_id"] is None
+
+
+@pytest.mark.asyncio
+async def test_spans_sorting_by_unsupported_field(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test sorting spans by a field that doesn't exist on the model."""
+    server, _client, session, api_endpoint = server_client
+
+    # Create rollout and spans
+    rollout = await server.start_rollout(input={"test": "data"})
+    for i in range(3):
+        span = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, i + 1, f"span-{i}")
+        await server.add_span(span)
+
+    # Try to sort by a non-existent field
+    async with session.get(
+        f"{api_endpoint}/spans",
+        params={"rollout_id": rollout.rollout_id, "sort_by": "invalid_field", "sort_order": "asc", "limit": -1},
+    ) 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"
+
+
+# Client Compatibility Tests
+
+
+@pytest.mark.asyncio
+async def test_client_query_rollouts_extracts_items(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test that client correctly extracts items from PaginatedResponse."""
+    server, client, _session, _api_endpoint = server_client
+
+    # Create rollouts
+    for i in range(5):
+        await server.enqueue_rollout(input={"index": i})
+
+    # Query via client (should extract items and return List[Rollout])
+    rollouts = await client.query_rollouts()
+    assert isinstance(rollouts, list)
+    assert len(rollouts) == 5
+    for rollout in rollouts:
+        assert isinstance(rollout, Rollout)
+
+
+@pytest.mark.asyncio
+async def test_client_query_with_filters(
+    server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
+) -> None:
+    """Test that client correctly passes filters to server."""
+    server, client, _session, _api_endpoint = server_client
+
+    # Create rollouts with different statuses
+    r1 = await server.enqueue_rollout(input={"id": 1})
+    r2 = await server.enqueue_rollout(input={"id": 2})
+    await server.update_rollout(rollout_id=r1.rollout_id, status="succeeded")
+
+    # Query via client with status filter
+    rollouts = await client.query_rollouts(status=["succeeded"])
+    assert len(rollouts) == 1
+    assert rollouts[0].rollout_id == r1.rollout_id
+
+    # Query via client with rollout_ids filter
+    rollouts = await client.query_rollouts(rollout_ids=[r2.rollout_id])
+    assert len(rollouts) == 1
+    assert rollouts[0].rollout_id == r2.rollout_id

tests/store/test_threading.py

@@ -1,6 +1,7 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import asyncio
+import time
 from concurrent.futures import ThreadPoolExecutor
 from typing import Any, Dict, cast
 from unittest.mock import MagicMock
@@ -77,7 +78,13 @@ class IncrementingResourceStore(LightningStore):
         snapshot = self.counter
         await asyncio.sleep(0.01)
         self.counter = snapshot + 1
-        return ResourcesUpdate(resources_id=f"res-{self.counter}", resources=resources)
+        return ResourcesUpdate(
+            resources_id=f"res-{self.counter}",
+            resources=resources,
+            create_time=time.time(),
+            update_time=time.time(),
+            version=1,
+        )
 
 
 def make_span(rollout_id: str, attempt_id: str, sequence_id: int = 1) -> Span:
@@ -128,7 +135,9 @@ async def test_threaded_store_delegates_all_methods() -> None:
         metadata={},
         attempt=base_attempt,
     )
-    resources_update = ResourcesUpdate(resources_id="resources-1", resources={})
+    resources_update = ResourcesUpdate(
+        resources_id="resources-1", resources={}, create_time=time.time(), update_time=time.time(), version=1
+    )
     span = make_span(rollout_id, attempt_id)
     readable_span = MagicMock(spec=ReadableSpan)
 
@@ -298,7 +307,9 @@ async def test_threaded_store_add_resources_delegates() -> None:
         sampling_parameters={"temperature": 0.7},
     )
     resources: NamedResources = cast(NamedResources, {"main_llm": llm})
-    resources_update = ResourcesUpdate(resources_id="resources-1", resources=resources)
+    resources_update = ResourcesUpdate(
+        resources_id="resources-1", resources=resources, create_time=time.time(), update_time=time.time(), version=1
+    )
 
     return_values = {
         "add_resources": resources_update,

tests/test_client.py

@@ -296,7 +296,13 @@ def test_local_client_core_functionality(sample_resources: NamedResources):
     assert client.task_count == 2
 
     # Test initialization with Task objects and ResourcesUpdate
-    resources_update = ResourcesUpdate(resources_id="version123", resources=sample_resources)
+    resources_update = ResourcesUpdate(
+        resources_id="version123",
+        resources=sample_resources,
+        create_time=time.time(),
+        update_time=time.time(),
+        version=1,
+    )
     tasks = [Task(rollout_id="existing_task", input="existing_input", resources_id="version123")]
     client2 = DevTaskLoader(tasks=tasks, resources=resources_update)
 
@@ -341,7 +347,13 @@ def test_local_client_error_handling(sample_resources: NamedResources):
         DevTaskLoader(tasks=mixed_tasks, resources=sample_resources)
 
     # Wrong resource ID should raise error
-    resources_update = ResourcesUpdate(resources_id="version123", resources=sample_resources)
+    resources_update = ResourcesUpdate(
+        resources_id="version123",
+        resources=sample_resources,
+        create_time=time.time(),
+        update_time=time.time(),
+        version=1,
+    )
     client = DevTaskLoader(tasks=["input1"], resources=resources_update)
     with pytest.raises(ValueError, match="Resource ID 'wrong_id' not found"):
         client.get_resources_by_id("wrong_id")