remove debug info

.github/workflows/tests-full.yml

@@ -47,6 +47,7 @@ jobs:
       - uses: actions/checkout@v4
         with:
           ref: ${{ github.event_name == 'repository_dispatch' && github.event.client_payload.pr_ref || (github.event.pull_request.number && format('refs/pull/{0}/merge', github.event.pull_request.number)) || github.ref }}
+
       - uses: astral-sh/setup-uv@v7
         with:
           enable-cache: true
@@ -55,10 +56,10 @@ jobs:
         run: uv lock --upgrade
         if: matrix.setup-script == 'latest'
       - name: Sync dependencies (latest)
-        run: uv sync --frozen --no-default-groups --extra apo --group dev --group agents --group torch-gpu-stable
+        run: uv sync --frozen --no-default-groups --extra apo --extra mongo --group dev --group agents --group torch-gpu-stable
         if: matrix.setup-script == 'latest'
       - name: Sync dependencies (stable & legacy)
-        run: uv sync --frozen --no-default-groups --extra apo --group dev --group agents --group torch-gpu-${{ matrix.setup-script }}
+        run: uv sync --frozen --no-default-groups --extra apo --extra mongo --group dev --group agents --group torch-gpu-${{ matrix.setup-script }}
         if: matrix.setup-script != 'latest'
       - name: Freeze dependencies
         run: |
@@ -81,6 +82,36 @@ jobs:
       - name: Build dashboard
         run: cd dashboard && npm run build
 
+      - name: Start MongoDB container
+        run: |
+          set -euo pipefail
+          cat /etc/security/limits.conf
+          docker run -d \
+            --name mongodb-test \
+            --ulimit nofile=65535:65535 \
+            -p 27017:27017 \
+            mongo:8.2 \
+            --replSet test-rs
+
+          # Wait for mongod to come up
+          for i in $(seq 1 30); do
+            if docker exec mongodb-test mongosh --quiet --eval 'db.runCommand({ ping: 1 })' >/dev/null 2>&1; then
+              echo "Mongo is up"
+              break
+            fi
+            echo "Waiting for Mongo..."
+            sleep 2
+          done
+
+          # Init replica set (simple single-node)
+          docker exec mongodb-test mongosh --quiet --eval '
+            rs.initiate({
+              _id: "test-rs",
+              members: [{ _id: 0, host: "localhost:27017" }]
+            })
+          '
+        shell: bash
+
       - name: Launch LiteLLM Proxy
         run: |
           ./scripts/litellm_run.sh
@@ -95,6 +126,8 @@ jobs:
           PYTEST_ADDOPTS: "--color=yes"
           OPENAI_BASE_URL: http://localhost:12306/
           OPENAI_API_KEY: dummy
+          AGL_TEST_MONGO_URI: mongodb://localhost:27017/?replicaSet=test-rs
+
 
   minimal-examples:
     if: >
@@ -160,6 +193,7 @@ jobs:
           source .venv/bin/activate
           cd examples/minimal
           python write_traces.py otel
+          sleep 5
 
       - name: Write Traces via AgentOps Tracer
         env:
@@ -170,6 +204,7 @@ jobs:
           source .venv/bin/activate
           cd examples/minimal
           python write_traces.py agentops
+          sleep 5
 
       - name: Write Traces via Otel Tracer with Client
         run: |

.github/workflows/tests.yml

@@ -37,6 +37,7 @@ jobs:
           uv sync --frozen \
             --extra apo \
             --extra verl \
+            --extra mongo \
             --group dev \
             --group torch-cpu \
             --group torch-stable \
@@ -166,7 +167,7 @@ jobs:
 
       - name: Run tests
         run: |
-          uv run pytest -v --durations=0 tests
+          uv run pytest -v --durations=0 tests -m "not mongo"
         env:
           PYTEST_ADDOPTS: "--color=yes"
 

agentlightning/llm_proxy.py

@@ -853,7 +853,6 @@ class StreamConversionMiddleware(BaseHTTPMiddleware):
         )  # e.g., "stop", "length", "tool_calls", "content_filter"
 
         def sse_chunk(obj: Dict[str, Any]) -> str:
-            print("sse_chunk: ", obj)
             return f"data: {json.dumps(obj, ensure_ascii=False)}\n\n"
 
         # 1) initial chunk with the role

agentlightning/store/client_server.py

@@ -799,8 +799,26 @@ class LightningStoreServer(LightningStore):
                 # wait_for_rollouts can block for a long time; avoid holding the lock
                 # so other requests can make progress while we wait.
                 return await getattr(self.store, method_name)(*args, **kwargs)
-            with self._lock:
+
+            # If it's already thread-safe, we can just call the method directly.
+            # Acquiring the threading lock directly would block the event loop if it's
+            # already held by another thread (for example, the HTTP server thread).
+            # Potential fix here are needed to make it work. For example:
+            # ```
+            # acquired = self._lock.acquire(blocking=False)
+            # if not acquired:
+            #     await asyncio.to_thread(self._lock.acquire)
+            # try:
+            #     return await getattr(self.store, method_name)(*args, **kwargs)
+            # finally:
+            #     self._lock.release()
+            # ```
+            # Or we can just bypass the lock for thread-safe stores.
+            if self.store is not None and self.store.capabilities.get("thread_safe", False):
                 return await getattr(self.store, method_name)(*args, **kwargs)
+            else:
+                with self._lock:
+                    return await getattr(self.store, method_name)(*args, **kwargs)
         if self._client is None:
             self._client = LightningStoreClient(self.endpoint)
         return await getattr(self._client, method_name)(*args, **kwargs)
@@ -1605,6 +1623,7 @@ class LightningStoreClient(LightningStore):
             attempt_id=attempt_id,
             sequence_id=sequence_id,
         )
+        print("created span", span)
         await self.add_span(span)
         return span
 

agentlightning/store/collection/base.py

@@ -3,17 +3,31 @@
 from __future__ import annotations
 
 from typing import (
+    TYPE_CHECKING,
     Any,
     AsyncContextManager,
+    Awaitable,
+    Callable,
+    Dict,
     Generic,
+    List,
+    Literal,
+    Mapping,
+    MutableMapping,
     Optional,
     Sequence,
+    Tuple,
     Type,
     TypeVar,
+    cast,
 )
 
+if TYPE_CHECKING:
+    from typing import Self
+
 from agentlightning.types import (
     Attempt,
+    FilterField,
     FilterOptions,
     PaginatedResult,
     ResourcesUpdate,
@@ -36,13 +50,13 @@ class Collection(Generic[T]):
         raise NotImplementedError()
 
     def __repr__(self) -> str:
-        return f"<{self.__class__.__name__}[{self.item_type().__name__}] ({self.size()})>"
+        return f"<{self.__class__.__name__}[{self.item_type().__name__}]>"
 
     def item_type(self) -> Type[T]:
         """Get the type of the items in the collection."""
         raise NotImplementedError()
 
-    def size(self) -> int:
+    async def size(self) -> int:
         """Get the number of items in the collection."""
         raise NotImplementedError()
 
@@ -132,7 +146,7 @@ class Queue(Generic[T]):
     """Behaves like a deque. Supporting appending items to the end and popping items from the front."""
 
     def __repr__(self) -> str:
-        return f"<{self.__class__.__name__}[{self.item_type().__name__}] ({self.size()})>"
+        return f"<{self.__class__.__name__}[{self.item_type().__name__}]>"
 
     def item_type(self) -> Type[T]:
         """Get the type of the items in the queue."""
@@ -177,7 +191,7 @@ class Queue(Generic[T]):
         """
         raise NotImplementedError()
 
-    def size(self) -> int:
+    async def size(self) -> int:
         """Get the number of items in the queue."""
         raise NotImplementedError()
 
@@ -186,7 +200,7 @@ class KeyValue(Generic[K, V]):
     """Behaves like a dictionary. Supporting addition, updating, and deletion of items."""
 
     def __repr__(self) -> str:
-        return f"<{self.__class__.__name__} ({self.size()})>"
+        return f"<{self.__class__.__name__}>"
 
     async def has(self, key: K) -> bool:
         """Check if the given key is in the dictionary."""
@@ -204,7 +218,7 @@ class KeyValue(Generic[K, V]):
         """Pop the value for the given key, or the default value if the key is not found."""
         raise NotImplementedError()
 
-    def size(self) -> int:
+    async def size(self) -> int:
         """Get the number of items in the dictionary."""
         raise NotImplementedError()
 
@@ -251,7 +265,7 @@ class LightningCollections:
         """Dictionary (counter) of span sequence IDs."""
         raise NotImplementedError()
 
-    def atomic(self, *args: Any, **kwargs: Any) -> AsyncContextManager[None]:
+    def atomic(self, *args: Any, **kwargs: Any) -> AsyncContextManager[Self]:
         """Perform a atomic operation on the collections.
 
         Subclass may use args and kwargs to support multiple levels of atomicity.
@@ -261,3 +275,82 @@ class LightningCollections:
             **kwargs: Keyword arguments to pass to the operation.
         """
         raise NotImplementedError()
+
+    async def execute(self, callback: Callable[[Self], Awaitable[T]]) -> T:
+        """Execute the given callback within an atomic operation."""
+        async with self.atomic() as collections:
+            return await callback(collections)
+
+
+FilterMap = Mapping[str, FilterField]
+
+
+def merge_must_filters(target: MutableMapping[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

agentlightning/store/collection/memory.py

@@ -9,7 +9,6 @@ from collections import deque
 from contextlib import asynccontextmanager
 from typing import (
     Any,
-    AsyncGenerator,
     Deque,
     Dict,
     Iterable,
@@ -23,7 +22,6 @@ from typing import (
     Type,
     TypeVar,
     Union,
-    cast,
 )
 
 from agentlightning.types import (
@@ -40,9 +38,12 @@ from agentlightning.types import (
 
 from .base import (
     Collection,
+    FilterMap,
     KeyValue,
     LightningCollections,
     Queue,
+    normalize_filter_options,
+    resolve_sort_options,
 )
 
 T = TypeVar("T")  # Recommended to be a BaseModel, not a dict
@@ -58,81 +59,9 @@ ListBasedCollectionItemType = Union[
     Dict[Any, T],  # leaf node dictionary
 ]
 
-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],
@@ -280,12 +209,12 @@ class ListBasedCollection(Collection[T]):
         """Return the Pydantic model type of items stored in this collection."""
         return self._item_type
 
-    def size(self) -> int:
+    async def size(self) -> int:
         """Return the number of items stored in the collection."""
         return self._size
 
     def __repr__(self) -> str:
-        return f"<{self.__class__.__name__}[{self.item_type().__name__}] ({self.size()})>"
+        return f"<{self.__class__.__name__}[{self.item_type().__name__}] ({self._size})>"
 
     # -------------------------------------------------------------------------
     # Internal helpers
@@ -520,8 +449,8 @@ 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, must_filters, filter_logic = _normalize_filter_options(filter)
-        sort_by, sort_order = _resolve_sort_options(sort)
+        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.
@@ -574,8 +503,8 @@ 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, must_filters, filter_logic = _normalize_filter_options(filter)
-        sort_by, sort_order = _resolve_sort_options(sort)
+        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:
@@ -655,6 +584,9 @@ class DequeBasedQueue(Queue[T]):
     def item_type(self) -> Type[T]:
         return self._item_type
 
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__}[{self.item_type().__name__}] ({len(self._items)})>"
+
     async def has(self, item: T) -> bool:
         if not isinstance(item, self._item_type):
             raise TypeError(f"Expected item of type {self._item_type.__name__}, got {type(item).__name__}")
@@ -686,7 +618,7 @@ class DequeBasedQueue(Queue[T]):
             result.append(item)
         return result
 
-    def size(self) -> int:
+    async def size(self) -> int:
         return len(self._items)
 
 
@@ -708,7 +640,7 @@ class DictBasedKeyValue(KeyValue[K, V]):
     async def pop(self, key: K, default: V | None = None) -> V | None:
         return self._values.pop(key, default)
 
-    def size(self) -> int:
+    async def size(self) -> int:
         return len(self._values)
 
 
@@ -759,9 +691,10 @@ class InMemoryLightningCollections(LightningCollections):
         return self._span_sequence_ids
 
     @asynccontextmanager
-    async def atomic(self, *args: Any, **kwargs: Any) -> AsyncGenerator[None, None]:
+    async def atomic(self, *args: Any, **kwargs: Any):
+        """In-memory collections apply a lock outside. It doesn't need to manipulate the collections inside."""
         async with self._lock:
-            yield
+            yield self
 
     async def evict_spans_for_rollout(self, rollout_id: str) -> None:
         """Evict all spans for a given rollout ID.

agentlightning/store/memory.py

@@ -19,6 +19,7 @@ from typing import (
     Optional,
     Set,
     TypeVar,
+    Union,
     cast,
 )
 
@@ -26,7 +27,7 @@ from pydantic import BaseModel
 
 from agentlightning.types import AttemptedRollout, PaginatedResult, Rollout, Span
 
-from .base import LightningStoreCapabilities, is_finished, is_running
+from .base import UNSET, LightningStoreCapabilities, Unset, is_finished, is_running
 from .collection import InMemoryLightningCollections
 from .collection_based import CollectionBasedLightningStore
 
@@ -68,9 +69,6 @@ class InMemoryLightningStore(CollectionBasedLightningStore[InMemoryLightningColl
     In-memory implementation of LightningStore using Python data structures.
     Thread-safe and async-compatible but data is not persistent.
 
-    The methods in this class should generally not call each other,
-    especially those that are locked.
-
     Args:
         eviction_memory_threshold: The threshold for evicting spans in bytes.
             By default, it's 70% of the total VRAM available.
@@ -127,6 +125,9 @@ class InMemoryLightningStore(CollectionBasedLightningStore[InMemoryLightningColl
         # Running rollouts cache, including preparing and running rollouts
         self._running_rollout_ids: Set[str] = set()
 
+        # Caches the latest resources ID.
+        self._latest_resources_id: Union[str, None, Unset] = UNSET
+
     @property
     def capabilities(self) -> LightningStoreCapabilities:
         """Return the capabilities of the store."""
@@ -139,8 +140,8 @@ class InMemoryLightningStore(CollectionBasedLightningStore[InMemoryLightningColl
 
     async def wait_for_rollout(self, rollout_id: str, timeout: Optional[float] = None) -> Optional[Rollout]:
         """Wait for a specific rollout to complete with a timeout."""
-        async with self.collections.atomic():
-            rollout = await self.collections.rollouts.get({"rollout_id": {"exact": rollout_id}})
+        async with self.collections.atomic() as collections:
+            rollout = await collections.rollouts.get({"rollout_id": {"exact": rollout_id}})
             if rollout and is_finished(rollout):
                 return rollout
 
@@ -167,8 +168,8 @@ class InMemoryLightningStore(CollectionBasedLightningStore[InMemoryLightningColl
 
             # If event was set (not timeout), check if rollout is finished
             if result:
-                async with self.collections.atomic():
-                    rollout = await self.collections.rollouts.get({"rollout_id": {"exact": rollout_id}})
+                async with self.collections.atomic() as collections:
+                    rollout = await collections.rollouts.get({"rollout_id": {"exact": rollout_id}})
                     if rollout and is_finished(rollout):
                         return rollout
 
@@ -192,14 +193,12 @@ class InMemoryLightningStore(CollectionBasedLightningStore[InMemoryLightningColl
         if rollout.rollout_id not in self._start_time_by_rollout:
             self._start_time_by_rollout[rollout.rollout_id] = rollout.start_time
 
-    async def get_running_rollouts(self) -> List[AttemptedRollout]:
+    async def get_running_rollouts(self, collections: InMemoryLightningCollections) -> 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": list(self._running_rollout_ids)}}
-        )
+        rollouts = await 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(
+            latest_attempt = await collections.attempts.get(
                 filter={"rollout_id": {"exact": rollout.rollout_id}},
                 sort={"name": "sequence_id", "order": "desc"},
             )
@@ -220,15 +219,24 @@ class InMemoryLightningStore(CollectionBasedLightningStore[InMemoryLightningColl
             raise RuntimeError(f"Spans for rollout {rollout_id} have been evicted")
         return await super().query_spans(rollout_id, attempt_id, **kwargs)
 
-    async def _add_span_unlocked(self, span: Span) -> Span:
+    async def _add_span_unlocked(self, collections: InMemoryLightningCollections, span: Span) -> Span:
         """In-memory store needs to maintain the span data in memory, and evict spans when memory is low."""
 
-        await super()._add_span_unlocked(span)
+        await super()._add_span_unlocked(collections, span)
         self._account_span_size(span)
-        await self._maybe_evict_spans()
+        await self._maybe_evict_spans(collections)
 
         return span
 
+    async def _get_latest_resources_id(self, collections: InMemoryLightningCollections) -> Optional[str]:
+        if isinstance(self._latest_resources_id, Unset):
+            latest_resources = await collections.resources.get(sort={"name": "update_time", "order": "desc"})
+            if latest_resources:
+                self._latest_resources_id = latest_resources.resources_id
+            else:
+                self._latest_resources_id = None
+        return self._latest_resources_id
+
     @staticmethod
     def _resolve_memory_threshold(
         value: float | int | None,
@@ -269,7 +277,7 @@ class InMemoryLightningStore(CollectionBasedLightningStore[InMemoryLightningColl
         self._total_span_bytes += size
         return size
 
-    async def _maybe_evict_spans(self) -> None:
+    async def _maybe_evict_spans(self, collections: InMemoryLightningCollections) -> None:
         if self._total_span_bytes <= self._eviction_threshold_bytes:
             return
 
@@ -288,11 +296,11 @@ class InMemoryLightningStore(CollectionBasedLightningStore[InMemoryLightningColl
             if self._total_span_bytes <= self._safe_threshold_bytes:
                 break
             logger.debug(f"Evicting spans for rollout {rollout_id} to free up memory...")
-            await self._evict_spans_for_rollout(rollout_id)
+            await self._evict_spans_for_rollout(collections, rollout_id)
         logger.info(f"Freed up {memory_consumed_before - self._total_span_bytes} bytes of memory")
 
-    async def _evict_spans_for_rollout(self, rollout_id: str) -> None:
-        await self.collections.evict_spans_for_rollout(rollout_id)
+    async def _evict_spans_for_rollout(self, collections: InMemoryLightningCollections, rollout_id: str) -> None:
+        await collections.evict_spans_for_rollout(rollout_id)
         removed_bytes = self._span_bytes_by_rollout.pop(rollout_id, 0)
         if removed_bytes > 0:
             # There is something removed for real

agentlightning/store/mongo.py

@@ -0,0 +1,82 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+import hashlib
+import logging
+import uuid
+from typing import (
+    Any,
+    Callable,
+    Mapping,
+    TypeVar,
+)
+
+from pymongo import AsyncMongoClient
+
+from .base import LightningStoreCapabilities
+from .collection.mongo import MongoClientPool, MongoLightningCollections
+from .collection_based import CollectionBasedLightningStore
+
+T_callable = TypeVar("T_callable", bound=Callable[..., Any])
+
+logger = logging.getLogger(__name__)
+
+
+def _generate_partition_id() -> str:
+    return "pt-" + hashlib.sha1(uuid.uuid4().bytes).hexdigest()[:12]
+
+
+class MongoLightningStore(CollectionBasedLightningStore[MongoLightningCollections]):
+    """
+    MongoDB implementation of LightningStore using MongoDB collections.
+    Data is persistent and can be shared between multiple processes.
+
+    Args:
+        client: The MongoDB client. Could be a string URI or an instance of AsyncMongoClient.
+        database: The MongoDB database. Could be a string name or an instance of AsyncDatabase.
+            You must provide at least one of client or database.
+        partition_id: The partition id. Useful when sharing the database among multiple Agent-lightning trainers.
+    """
+
+    def __init__(
+        self,
+        *,
+        client: AsyncMongoClient[Mapping[str, Any]] | str,
+        database_name: str | None = None,
+        partition_id: str | None = None,
+    ) -> None:
+        self._auto_created_client = False
+        if isinstance(client, str):
+            self._client = AsyncMongoClient[Mapping[str, Any]](client)
+            self._auto_created_client = True
+        else:
+            self._client = client
+        if database_name is None:
+            database_name = "agentlightning"
+            logger.info("No database name provided, using default 'agentlightning'")
+
+        if partition_id is None:
+            partition_id = _generate_partition_id()
+            logger.info("No partition id provided, generated a new one: %s", partition_id)
+
+        self._client_pool = MongoClientPool(self._client)
+
+        super().__init__(collections=MongoLightningCollections(self._client_pool, database_name, partition_id))
+
+    @property
+    def capabilities(self) -> LightningStoreCapabilities:
+        """Return the capabilities of the store."""
+        return LightningStoreCapabilities(
+            thread_safe=True,
+            async_safe=True,
+            zero_copy=True,
+            otlp_traces=False,
+        )
+
+    async def close(self) -> None:
+        """Close the store by closing the client pool."""
+        await self._client_pool.close()
+        # If I created the client, I should close it too.
+        if self._auto_created_client:
+            await self._client.close()

pyproject.toml

@@ -38,6 +38,11 @@ verl = [
   "vllm>=0.8.4", # Due to interface change of ExternalZeroMQDistributedExecutor
 ]
 
+# Store-related dependencies.
+mongo = [
+  "pymongo",
+]
+
 [project.scripts]
 agl = "agentlightning.cli:main"
 
@@ -285,6 +290,13 @@ exclude = [
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
+markers = [
+  "openai: tests that require OpenAI API",
+  "gpu: tests that require GPU",
+  "agentops: tests that require AgentOps",
+  "llmproxy: tests that require LiteLLM",
+  "mongo: tests that require MongoDB",
+]
 
 [tool.black]
 line-length = 120

pyrightconfig.fast.json

@@ -7,7 +7,9 @@
     "agentlightning/instrumentation",
     "agentlightning/algorithm/apo",
     "agentlightning/algorithm/verl",
-    "agentlightning/cli/vllm.py"
+    "agentlightning/cli/vllm.py",
+    "agentlightning/store/collection/mongo.py",
+    "agentlightning/store/mongo.py"
   ],
 
   "pythonVersion": "3.12",

scripts/mongo/README.md

@@ -0,0 +1,9 @@
+# MongoDB Development Setup
+
+This script is used to setup MongoDB for development.
+
+## Usage
+
+```bash
+docker compose up -d
+```

scripts/mongo/docker-compose.yml

@@ -0,0 +1,10 @@
+services:
+  mongo:
+    image: mongo:latest
+    container_name: mongo-dev
+    ports:
+      - "27017:27017"
+    command: ["mongod", "--bind_ip_all", "--replSet", "rs0"]
+    volumes:
+      - ./data:/data/db
+      - ./init-rs.js:/docker-entrypoint-initdb.d/init-rs.js:ro

scripts/mongo/init-rs.js

@@ -0,0 +1,6 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+rs.initiate({
+  _id: "rs0",
+  members: [{ _id: 0, host: "localhost:27017" }],
+});

tests/store/conftest.py

@@ -1,41 +1,72 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+from __future__ import annotations
+
+import os
 import time
 from itertools import count
+from typing import TYPE_CHECKING, Any, AsyncGenerator, Dict, List, Sequence
 from unittest.mock import Mock
+from uuid import uuid4
 
 import pytest
+import pytest_asyncio
 from opentelemetry.sdk.trace import ReadableSpan
+from pydantic import BaseModel, Field
 from pytest import FixtureRequest
 
 from agentlightning.store.base import LightningStore
+from agentlightning.store.collection import DequeBasedQueue, DictBasedKeyValue, KeyValue, ListBasedCollection, Queue
+from agentlightning.store.collection.base import Collection
 from agentlightning.store.memory import InMemoryLightningStore
 
+if TYPE_CHECKING:
+    from pymongo import AsyncMongoClient
+    from pymongo.asynchronous.database import AsyncDatabase
+
 __all__ = [
     "inmemory_store",
     "mock_readable_span",
+    "sample_items",
+    "sample_collection",
+    "SampleItem",
+    "QueueItem",
+    "deque_queue",
+    "dict_key_value",
+    "dict_key_value_data",
+    "temporary_mongo_database",
 ]
 
 
+mongo_uri = os.getenv("AGL_TEST_MONGO_URI", "mongodb://localhost:27017/?replicaSet=rs0")
+
+
 @pytest.fixture
 def inmemory_store() -> InMemoryLightningStore:
     """Create a fresh InMemoryLightningStore instance."""
     return InMemoryLightningStore()
 
 
-@pytest.fixture
-def sql_store():
-    """Placeholder fixture for SQL store implementation. Returns None until SQL store is ready."""
-    return None
+@pytest_asyncio.fixture
+async def mongo_store(temporary_mongo_database: AsyncDatabase[Any]):
+    """Fixture for MongoDB store implementation."""
+    from agentlightning.store.mongo import MongoLightningStore
+
+    db = MongoLightningStore(client=temporary_mongo_database.client, database_name=temporary_mongo_database.name)
+    try:
+        yield db
+    finally:
+        await db.close()
 
 
-# Uncomment this when sql store is ready
-# @pytest.fixture(params=["inmemory_store", "sql_store"])
-@pytest.fixture(params=["inmemory_store"])
-def store_fixture(request: FixtureRequest) -> LightningStore:
-    """Parameterized fixture that provides different store implementations for testing.
-    Currently supports InMemoryLightningStore, with SQL store support planned.
-    """
+@pytest.fixture(
+    params=[
+        "inmemory_store",
+        pytest.param("mongo_store", marks=pytest.mark.mongo),
+    ]
+)
+def store_fixture(request: FixtureRequest) -> AsyncGenerator[LightningStore, None]:
+    """Parameterized fixture that provides different store implementations for testing."""
     return request.getfixturevalue(request.param)
 
 
@@ -73,3 +104,269 @@ def mock_readable_span() -> ReadableSpan:
     span.resource = Mock(attributes={}, schema_url="")
 
     return span
+
+
+class SampleItem(BaseModel):
+    partition: str
+    index: int
+    name: str
+    status: str
+    tags: List[str] = Field(default_factory=list)
+    score: float | None = None
+    rank: int | None = None
+    updated_time: float | None = None
+    payload: Dict[str, int] = Field(default_factory=dict)
+    metadata: str | None = None
+
+
+class QueueItem(BaseModel):
+    idx: int
+
+
+@pytest_asyncio.fixture
+async def mongo_client():
+    from pymongo import AsyncMongoClient
+
+    client = AsyncMongoClient[Any](mongo_uri, serverSelectionTimeoutMS=5000)
+    try:
+        await client.admin.command("ping")
+    except Exception as exc:  # depends on external service
+        await client.close()
+        raise RuntimeError(f"MongoDB not available: {exc}")
+
+    try:
+        yield client
+    finally:
+        await client.close()
+
+
+@pytest_asyncio.fixture
+async def temporary_mongo_database(mongo_client: AsyncMongoClient[Any]):
+    """Yield a temporary MongoDB database for integration tests."""
+    db_name = f"agentlightning-test-{uuid4().hex}"
+    db = mongo_client[db_name]  # type: ignore
+    try:
+        yield db
+    finally:
+        await mongo_client.drop_database(db_name)
+
+
+### Collection fixtures ###
+
+
+@pytest.fixture()
+def sample_items() -> List[SampleItem]:
+    return [
+        SampleItem(
+            partition="alpha",
+            index=1,
+            name="urgent-phase-one",
+            status="new",
+            tags=["core", "urgent"],
+            score=10.5,
+            rank=3,
+            updated_time=12.0,
+            payload={"priority": 10},
+            metadata="alpha-start",
+        ),
+        SampleItem(
+            partition="alpha",
+            index=2,
+            name="phase-two",
+            status="running",
+            tags=["core"],
+            score=5.0,
+            rank=2,
+            updated_time=None,
+            payload={"priority": 5},
+            metadata=None,
+        ),
+        SampleItem(
+            partition="alpha",
+            index=3,
+            name="delayed-phase",
+            status="blocked",
+            tags=["delayed"],
+            score=None,
+            rank=5,
+            updated_time=15.1,
+            payload={"priority": 8},
+            metadata="delayed-phase",
+        ),
+        SampleItem(
+            partition="beta",
+            index=1,
+            name="beta-critical",
+            status="new",
+            tags=["beta", "urgent"],
+            score=8.0,
+            rank=1,
+            updated_time=7.0,
+            payload={"priority": 7},
+            metadata="beta critical",
+        ),
+        SampleItem(
+            partition="beta",
+            index=2,
+            name="beta optional",
+            status="done",
+            tags=["beta"],
+            score=3.0,
+            rank=None,
+            updated_time=2.0,
+            payload={"priority": 1},
+            metadata="optional path",
+        ),
+        SampleItem(
+            partition="gamma",
+            index=1,
+            name="gamma-phase",
+            status="running",
+            tags=[],
+            score=9.5,
+            rank=4,
+            updated_time=None,
+            payload={"priority": 9},
+            metadata="gamma-phase data",
+        ),
+        SampleItem(
+            partition="gamma",
+            index=2,
+            name="gamma-late",
+            status="done",
+            tags=["late", "core"],
+            score=1.0,
+            rank=6,
+            updated_time=20.0,
+            payload={"priority": 2},
+            metadata="gamma late entry",
+        ),
+        SampleItem(
+            partition="delta",
+            index=1,
+            name="delta misc",
+            status="archived",
+            tags=["misc"],
+            score=4.2,
+            rank=7,
+            updated_time=11.0,
+            payload={"priority": 3},
+            metadata="delta misc block",
+        ),
+    ]
+
+
+### Generic collection fixtures ###
+
+
+@pytest.fixture()
+def sample_collection_memory(sample_items: Sequence[SampleItem]) -> ListBasedCollection[SampleItem]:
+    collection: Collection[SampleItem] = ListBasedCollection(list(sample_items), SampleItem, ("partition", "index"))
+    setattr(collection, "_test_backend", "memory")
+    return collection
+
+
+@pytest_asyncio.fixture
+async def sample_collection_mongo(temporary_mongo_database: AsyncDatabase[Any], sample_items: Sequence[SampleItem]):
+    from agentlightning.store.collection.mongo import MongoBasedCollection, MongoClientPool
+
+    async with MongoClientPool(temporary_mongo_database.client) as client_pool:
+        collection = MongoBasedCollection(
+            client_pool,
+            temporary_mongo_database.name,
+            "sample-items",
+            "partition-123",
+            ["partition", "index"],
+            SampleItem,
+        )
+        await collection.insert(sample_items)
+        setattr(collection, "_test_backend", "mongo")
+        yield collection
+
+
+@pytest.fixture(
+    params=[
+        "memory",
+        pytest.param("mongo", marks=pytest.mark.mongo),
+    ]
+)
+def sample_collection(request: pytest.FixtureRequest):
+    backend = request.param
+    return request.getfixturevalue("sample_collection_" + backend)
+
+
+### Generic queue fixtures ###
+
+
+@pytest.fixture
+def deque_queue_memory() -> DequeBasedQueue[QueueItem]:
+    return DequeBasedQueue(QueueItem, [QueueItem(idx=i) for i in range(3)])
+
+
+@pytest_asyncio.fixture
+async def deque_queue_mongo(temporary_mongo_database: AsyncDatabase[Any]):
+    from agentlightning.store.collection.mongo import MongoBasedQueue, MongoClientPool
+
+    async with MongoClientPool(temporary_mongo_database.client) as client_pool:
+        queue = MongoBasedQueue[QueueItem](
+            client_pool,
+            temporary_mongo_database.name,
+            "queue-items",
+            "partition-1",
+            QueueItem,
+        )
+        await queue.enqueue([QueueItem(idx=i) for i in range(3)])
+        yield queue
+
+
+@pytest.fixture(
+    params=[
+        "memory",
+        pytest.param("mongo", marks=pytest.mark.mongo),
+    ]
+)
+def deque_queue(request: pytest.FixtureRequest) -> AsyncGenerator[Queue[QueueItem], None]:
+    backend = request.param
+    return request.getfixturevalue("deque_queue_" + backend)
+
+
+### Generic key-value fixtures ###
+
+
+@pytest.fixture()
+def dict_key_value_data() -> Dict[str, int]:
+    return {"alpha": 1, "beta": 2}
+
+
+@pytest.fixture()
+def dict_key_value_memory(dict_key_value_data: Dict[str, int]) -> DictBasedKeyValue[str, int]:
+    return DictBasedKeyValue(dict_key_value_data)
+
+
+@pytest_asyncio.fixture
+async def dict_key_value_mongo(temporary_mongo_database: AsyncDatabase[Any], dict_key_value_data: Dict[str, int]):
+    from agentlightning.store.collection.mongo import MongoBasedKeyValue, MongoClientPool
+
+    async with MongoClientPool(temporary_mongo_database.client) as client_pool:
+        key_value = MongoBasedKeyValue[str, int](
+            client_pool,
+            temporary_mongo_database.name,
+            "key-value-items",
+            "partition-1",
+            str,
+            int,
+        )
+        for key, value in dict_key_value_data.items():
+            await key_value.set(key, value)
+        yield key_value
+
+
+@pytest.fixture(
+    params=[
+        "memory",
+        pytest.param("mongo", marks=pytest.mark.mongo),
+    ]
+)
+def dict_key_value(request: pytest.FixtureRequest) -> AsyncGenerator[KeyValue[str, int], None]:
+    backend = request.param
+    return request.getfixturevalue("dict_key_value_" + backend)

tests/store/test_collection_memory.py

@@ -2,135 +2,29 @@
 
 from __future__ import annotations
 
-from typing import Dict, Iterable, List, Literal, Mapping, Sequence, Tuple
+import asyncio
+import time
+from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Literal, Mapping, Sequence, Tuple, Union
+from uuid import uuid4
 
 import pytest
-from pydantic import BaseModel, Field
+from pydantic import BaseModel
 
 import agentlightning.store.collection.memory as memory_module
 from agentlightning.store.collection import DequeBasedQueue, DictBasedKeyValue, ListBasedCollection
+from agentlightning.store.collection.base import Collection
 from agentlightning.store.collection.memory import _item_matches_filters  # pyright: ignore[reportPrivateUsage]
+from agentlightning.types import Rollout
+from tests.store.conftest import QueueItem, SampleItem
 
-
-class SampleItem(BaseModel):
-    partition: str
-    index: int
-    name: str
-    status: str
-    tags: List[str] = Field(default_factory=list)
-    score: float | None = None
-    rank: int | None = None
-    updated_time: float | None = None
-    payload: Dict[str, int] = Field(default_factory=dict)
-    metadata: str | None = None
+if TYPE_CHECKING:
+    from pymongo.asynchronous.database import AsyncDatabase
 
 
 def _build_collection(items: Iterable[SampleItem] = ()) -> ListBasedCollection[SampleItem]:
     return ListBasedCollection(list(items), SampleItem, ("partition", "index"))
 
 
-@pytest.fixture()
-def sample_items() -> List[SampleItem]:
-    return [
-        SampleItem(
-            partition="alpha",
-            index=1,
-            name="urgent-phase-one",
-            status="new",
-            tags=["core", "urgent"],
-            score=10.5,
-            rank=3,
-            updated_time=12.0,
-            payload={"priority": 10},
-            metadata="alpha-start",
-        ),
-        SampleItem(
-            partition="alpha",
-            index=2,
-            name="phase-two",
-            status="running",
-            tags=["core"],
-            score=5.0,
-            rank=2,
-            updated_time=None,
-            payload={"priority": 5},
-            metadata=None,
-        ),
-        SampleItem(
-            partition="alpha",
-            index=3,
-            name="delayed-phase",
-            status="blocked",
-            tags=["delayed"],
-            score=None,
-            rank=5,
-            updated_time=15.1,
-            payload={"priority": 8},
-            metadata="delayed-phase",
-        ),
-        SampleItem(
-            partition="beta",
-            index=1,
-            name="beta-critical",
-            status="new",
-            tags=["beta", "urgent"],
-            score=8.0,
-            rank=1,
-            updated_time=7.0,
-            payload={"priority": 7},
-            metadata="beta critical",
-        ),
-        SampleItem(
-            partition="beta",
-            index=2,
-            name="beta optional",
-            status="done",
-            tags=["beta"],
-            score=3.0,
-            rank=None,
-            updated_time=2.0,
-            payload={"priority": 1},
-            metadata="optional path",
-        ),
-        SampleItem(
-            partition="gamma",
-            index=1,
-            name="gamma-phase",
-            status="running",
-            tags=[],
-            score=9.5,
-            rank=4,
-            updated_time=None,
-            payload={"priority": 9},
-            metadata="gamma-phase data",
-        ),
-        SampleItem(
-            partition="gamma",
-            index=2,
-            name="gamma-late",
-            status="done",
-            tags=["late", "core"],
-            score=1.0,
-            rank=6,
-            updated_time=20.0,
-            payload={"priority": 2},
-            metadata="gamma late entry",
-        ),
-        SampleItem(
-            partition="delta",
-            index=1,
-            name="delta misc",
-            status="archived",
-            tags=["misc"],
-            score=4.2,
-            rank=7,
-            updated_time=11.0,
-            payload={"priority": 3},
-            metadata="delta misc block",
-        ),
-    ]
-
-
 BASE_KEY_ORDER: List[Tuple[str, int]] = [
     ("alpha", 1),
     ("alpha", 2),
@@ -143,11 +37,6 @@ BASE_KEY_ORDER: List[Tuple[str, int]] = [
 ]
 
 
-@pytest.fixture()
-def sample_collection(sample_items: Sequence[SampleItem]) -> ListBasedCollection[SampleItem]:
-    return _build_collection(sample_items)
-
-
 def _key_pairs(items: Sequence[SampleItem]) -> List[Tuple[str, int]]:
     return [(item.partition, item.index) for item in items]
 
@@ -161,43 +50,50 @@ def test_list_collection_requires_primary_keys(sample_items: Sequence[SampleItem
         ListBasedCollection(list(sample_items), SampleItem, ())
 
 
-def test_list_collection_primary_keys(sample_collection: ListBasedCollection[SampleItem]) -> None:
+@pytest.mark.asyncio()
+async def test_list_collection_primary_keys(sample_collection: Collection[SampleItem]) -> None:
     assert tuple(sample_collection.primary_keys()) == ("partition", "index")
 
 
-def test_list_collection_item_type(sample_collection: ListBasedCollection[SampleItem]) -> None:
+@pytest.mark.asyncio()
+async def test_list_collection_item_type(sample_collection: Collection[SampleItem]) -> None:
     assert sample_collection.item_type() is SampleItem
 
 
-def test_list_collection_initial_size(
-    sample_collection: ListBasedCollection[SampleItem], sample_items: Sequence[SampleItem]
+@pytest.mark.asyncio()
+async def test_list_collection_initial_size(
+    sample_collection: Collection[SampleItem], sample_items: Sequence[SampleItem]
 ) -> None:
-    assert sample_collection.size() == len(sample_items)
-
-
-def test_list_collection_repr_contains_model_info(sample_collection: ListBasedCollection[SampleItem]) -> None:
-    result = repr(sample_collection)
-    assert "ListBasedCollection" in result and "SampleItem" in result and str(sample_collection.size()) in result
+    assert (await sample_collection.size()) == len(sample_items)
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_insert_adds_item(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_repr_contains_model_info(sample_collection: Collection[SampleItem]) -> None:
+    result = repr(sample_collection)
+    assert sample_collection.__class__.__name__ in result
+    assert "SampleItem" in result
+    if isinstance(sample_collection, ListBasedCollection):
+        assert str(await sample_collection.size()) in result
+
+
+@pytest.mark.asyncio()
+async def test_list_collection_insert_adds_item(sample_collection: Collection[SampleItem]) -> None:
     new_item = SampleItem(partition="omega", index=1, name="omega", status="new")
     await sample_collection.insert([new_item])
-    assert sample_collection.size() == 9
+    assert (await sample_collection.size()) == 9
     result = await sample_collection.get({"partition": {"exact": "omega"}, "index": {"exact": 1}})
     assert result == new_item
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_insert_duplicate_raises(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_insert_duplicate_raises(sample_collection: Collection[SampleItem]) -> None:
     duplicate = SampleItem(partition="alpha", index=1, name="dup", status="new")
     with pytest.raises(ValueError):
         await sample_collection.insert([duplicate])
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_insert_wrong_type(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_insert_wrong_type(sample_collection: Collection[SampleItem]) -> None:
     class Another(BaseModel):
         partition: str
         index: int
@@ -208,7 +104,7 @@ async def test_list_collection_insert_wrong_type(sample_collection: ListBasedCol
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_update_existing(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_update_existing(sample_collection: Collection[SampleItem]) -> None:
     updated = SampleItem(partition="alpha", index=1, name="updated", status="new")
     await sample_collection.update([updated])
     result = await sample_collection.get({"partition": {"exact": "alpha"}, "index": {"exact": 1}})
@@ -216,74 +112,74 @@ async def test_list_collection_update_existing(sample_collection: ListBasedColle
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_update_missing_raises(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_update_missing_raises(sample_collection: Collection[SampleItem]) -> None:
     missing = SampleItem(partition="omega", index=99, name="missing", status="lost")
     with pytest.raises(ValueError):
         await sample_collection.update([missing])
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_delete_existing(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_delete_existing(sample_collection: Collection[SampleItem]) -> None:
     target = SampleItem(partition="alpha", index=1, name="ignored", status="new")
     await sample_collection.delete([target])
-    assert sample_collection.size() == 7
+    assert (await sample_collection.size()) == 7
     result = await sample_collection.get({"partition": {"exact": "alpha"}, "index": {"exact": 1}})
     assert result is None
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_delete_missing_raises(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_delete_missing_raises(sample_collection: Collection[SampleItem]) -> None:
     missing = SampleItem(partition="omega", index=3, name="x", status="y")
     with pytest.raises(ValueError):
         await sample_collection.delete([missing])
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_upsert_inserts_when_missing(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_upsert_inserts_when_missing(sample_collection: Collection[SampleItem]) -> None:
     created = SampleItem(partition="omega", index=4, name="new", status="queued")
     await sample_collection.upsert([created])
-    assert sample_collection.size() == 9
+    assert (await sample_collection.size()) == 9
     fetched = await sample_collection.get({"partition": {"exact": "omega"}, "index": {"exact": 4}})
     assert fetched == created
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_upsert_updates_when_existing(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_upsert_updates_when_existing(sample_collection: Collection[SampleItem]) -> None:
     replacement = SampleItem(partition="beta", index=2, name="replacement", status="done")
     await sample_collection.upsert([replacement])
-    assert sample_collection.size() == 8
+    assert (await sample_collection.size()) == 8
     fetched = await sample_collection.get({"partition": {"exact": "beta"}, "index": {"exact": 2}})
     assert fetched == replacement
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_delete_multiple_items(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_delete_multiple_items(sample_collection: Collection[SampleItem]) -> None:
     await sample_collection.delete(
         [
             SampleItem(partition="alpha", index=1, name="", status=""),
             SampleItem(partition="beta", index=1, name="", status=""),
         ]
     )
-    assert sample_collection.size() == 6
+    assert (await sample_collection.size()) == 6
 
 
 @pytest.mark.asyncio()
 async def test_list_collection_insert_accepts_tuple_sequence(
-    sample_collection: ListBasedCollection[SampleItem],
+    sample_collection: Collection[SampleItem],
 ) -> None:
     extra = (
         SampleItem(partition="tuple", index=1, name="a", status="pending"),
         SampleItem(partition="tuple", index=2, name="b", status="pending"),
     )
     await sample_collection.insert(extra)
-    assert sample_collection.size() == 10
+    assert (await sample_collection.size()) == 10
     fetched = await sample_collection.query(filter={"partition": {"exact": "tuple"}})
     assert _sorted_pairs(fetched.items) == [("tuple", 1), ("tuple", 2)]
 
 
 @pytest.mark.asyncio()
 async def test_list_collection_query_without_filters_returns_all(
-    sample_collection: ListBasedCollection[SampleItem],
+    sample_collection: Collection[SampleItem],
 ) -> None:
     result = await sample_collection.query()
     assert result.total == 8
@@ -331,10 +227,17 @@ async def test_list_collection_query_without_filters_returns_all(
     ],
 )
 async def test_list_collection_query_filters(
-    sample_collection: ListBasedCollection[SampleItem],
+    sample_collection: Collection[SampleItem],
     filters: Dict[str, Dict[str, object]],
     expected: Sequence[Tuple[str, int]],
+    request: pytest.FixtureRequest,
 ) -> None:
+    # Mongo implementation raises ValueError for non-iterable values in within filter
+    if request.node.callspec.id == "mongo-within-non-iterable":  # type: ignore
+        with pytest.raises(ValueError):
+            await sample_collection.query(filter=filters)  # type: ignore[arg-type]
+        return
+
     result = await sample_collection.query(filter=filters)  # type: ignore[arg-type]
     assert _sorted_pairs(result.items) == sorted(expected)
     assert result.total == len(expected)
@@ -367,7 +270,7 @@ async def test_list_collection_query_filters(
     ],
 )
 async def test_list_collection_filter_logic(
-    sample_collection: ListBasedCollection[SampleItem],
+    sample_collection: Collection[SampleItem],
     filters: Dict[str, Dict[str, object]],
     filter_logic: Literal["and", "or"],
     expected: Sequence[Tuple[str, int]],
@@ -380,7 +283,7 @@ async def test_list_collection_filter_logic(
 
 @pytest.mark.asyncio()
 async def test_list_collection_must_filters_respected_with_or(
-    sample_collection: ListBasedCollection[SampleItem],
+    sample_collection: Collection[SampleItem],
 ) -> None:
     filters = {
         "_aggregate": "or",
@@ -394,7 +297,7 @@ async def test_list_collection_must_filters_respected_with_or(
 
 @pytest.mark.asyncio()
 async def test_list_collection_must_filters_accept_sequence(
-    sample_collection: ListBasedCollection[SampleItem],
+    sample_collection: Collection[SampleItem],
 ) -> None:
     filters = {
         "_aggregate": "or",
@@ -411,9 +314,11 @@ async def test_list_collection_must_filters_accept_sequence(
 
 @pytest.mark.asyncio()
 async def test_list_collection_must_filters_limit_tree_scan_even_with_or(
-    sample_collection: ListBasedCollection[SampleItem],
+    sample_collection: Collection[SampleItem],
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
+    if not isinstance(sample_collection, ListBasedCollection):
+        pytest.skip("This test is only valid for pure-memory collections")
     seen: List[Tuple[str, int]] = []
     original = _item_matches_filters
 
@@ -467,9 +372,11 @@ async def test_list_collection_primary_key_prefix_limits_filter_checks(
 
 @pytest.mark.asyncio()
 async def test_list_collection_full_primary_key_avoids_tree_scan(
-    sample_collection: ListBasedCollection[SampleItem],
+    sample_collection: Collection[SampleItem],
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
+    if not isinstance(sample_collection, ListBasedCollection):
+        pytest.skip("This test is only valid for pure-memory collections")
     call_count = 0
     original_iter_items = (  # pyright: ignore[reportPrivateUsage,reportUnknownMemberType,reportUnknownVariableType]
         ListBasedCollection._iter_items  # pyright: ignore[reportPrivateUsage,reportUnknownMemberType]
@@ -504,23 +411,42 @@ async def test_list_collection_full_primary_key_avoids_tree_scan(
         ("rank", "desc", 4, [("delta", 1), ("gamma", 2), ("alpha", 3), ("gamma", 1)]),
         ("score", "asc", 4, [("alpha", 3), ("gamma", 2), ("beta", 2), ("delta", 1)]),
         ("score", "desc", 4, [("alpha", 1), ("gamma", 1), ("beta", 1), ("alpha", 2)]),
-        ("updated_time", "asc", 4, [("beta", 2), ("beta", 1), ("delta", 1), ("alpha", 1)]),
-        ("updated_time", "desc", 4, [("gamma", 1), ("alpha", 2), ("gamma", 2), ("alpha", 3)]),
+        (
+            "updated_time",
+            "asc",
+            4,
+            (
+                [("alpha", 2), ("gamma", 1), ("beta", 2), ("beta", 1)],
+                [("beta", 2), ("beta", 1), ("delta", 1), ("alpha", 1)],
+            ),
+        ),
+        (
+            "updated_time",
+            "desc",
+            4,
+            (
+                [("gamma", 2), ("alpha", 3), ("alpha", 1), ("delta", 1)],
+                [("gamma", 1), ("alpha", 2), ("gamma", 2), ("alpha", 3)],
+            ),
+        ),
     ],
 )
 async def test_list_collection_sorting(
-    sample_collection: ListBasedCollection[SampleItem],
+    sample_collection: Collection[SampleItem],
     sort_by: str,
     sort_order: str,
     limit: int,
-    expected: Sequence[Tuple[str, int]],
+    expected: Union[Sequence[Tuple[str, int]], Tuple[Sequence[Tuple[str, int]], ...]],
 ) -> None:
     result = await sample_collection.query(sort={"name": sort_by, "order": sort_order}, limit=limit)  # type: ignore[arg-type]
-    assert _key_pairs(result.items) == list(expected)
+    if isinstance(expected, tuple):
+        assert any(_key_pairs(result.items) == list(expected) for expected in expected)
+    else:
+        assert _key_pairs(result.items) == list(expected)
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_sort_by_missing_field_raises(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_sort_by_missing_field_raises(sample_collection: Collection[SampleItem]) -> None:
     with pytest.raises(ValueError):
         await sample_collection.query(sort={"name": "does_not_exist", "order": "asc"})
 
@@ -538,7 +464,7 @@ async def test_list_collection_sort_by_missing_field_raises(sample_collection: L
     ],
 )
 async def test_list_collection_pagination_without_sort(
-    sample_collection: ListBasedCollection[SampleItem],
+    sample_collection: Collection[SampleItem],
     limit: int,
     offset: int,
     expected: Sequence[Tuple[str, int]],
@@ -550,21 +476,21 @@ async def test_list_collection_pagination_without_sort(
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_pagination_with_sort(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_pagination_with_sort(sample_collection: Collection[SampleItem]) -> None:
     result = await sample_collection.query(sort={"name": "name", "order": "asc"}, limit=2, offset=3)
     assert _key_pairs(result.items) == [("delta", 1), ("gamma", 2)]
     assert result.total == 8
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_limit_unbounded_with_sort(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_limit_unbounded_with_sort(sample_collection: Collection[SampleItem]) -> None:
     result = await sample_collection.query(sort={"name": "name", "order": "asc"}, limit=-1, offset=6)
     assert _key_pairs(result.items) == [("alpha", 2), ("alpha", 1)]
     assert result.total == 8
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_limit_zero_reports_total(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_limit_zero_reports_total(sample_collection: Collection[SampleItem]) -> None:
     result = await sample_collection.query(filter={"status": {"exact": "done"}}, limit=0)
     assert result.items == []
     assert result.total == 2
@@ -572,7 +498,7 @@ async def test_list_collection_limit_zero_reports_total(sample_collection: ListB
 
 @pytest.mark.asyncio()
 async def test_list_collection_offset_beyond_total_returns_empty(
-    sample_collection: ListBasedCollection[SampleItem],
+    sample_collection: Collection[SampleItem],
 ) -> None:
     result = await sample_collection.query(filter={"status": {"exact": "done"}}, offset=10)
     assert result.items == []
@@ -581,7 +507,7 @@ async def test_list_collection_offset_beyond_total_returns_empty(
 
 @pytest.mark.asyncio()
 async def test_list_collection_query_reports_total_with_limit(
-    sample_collection: ListBasedCollection[SampleItem],
+    sample_collection: Collection[SampleItem],
 ) -> None:
     result = await sample_collection.query(filter={"partition": {"exact": "alpha"}}, limit=1)
     assert result.total == 3
@@ -589,28 +515,28 @@ async def test_list_collection_query_reports_total_with_limit(
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_get_returns_first_match(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_get_returns_first_match(sample_collection: Collection[SampleItem]) -> None:
     item = await sample_collection.get({"status": {"exact": "new"}})
     assert item is not None
-    assert (item.partition, item.index) == ("beta", 1)
+    assert (item.partition, item.index) in [("beta", 1), ("alpha", 1)]
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_get_returns_none(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_get_returns_none(sample_collection: Collection[SampleItem]) -> None:
     result = await sample_collection.get({"partition": {"exact": "missing"}})
     assert result is None
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_get_respects_filter_logic(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_get_respects_filter_logic(sample_collection: Collection[SampleItem]) -> None:
     filters = {"status": {"exact": "done"}, "tags": {"contains": "urgent"}, "_aggregate": "or"}
     item = await sample_collection.get(filters)  # type: ignore[arg-type]
     assert item is not None
-    assert (item.partition, item.index) == ("gamma", 2)
+    assert (item.partition, item.index) in [("gamma", 2), ("alpha", 1)]
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_get_honors_sort_by(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_get_honors_sort_by(sample_collection: Collection[SampleItem]) -> None:
     filters = {"partition": {"exact": "alpha"}}
     item = await sample_collection.get(filters, sort={"name": "rank", "order": "asc"})  # type: ignore[arg-type]
     assert item is not None
@@ -618,7 +544,7 @@ async def test_list_collection_get_honors_sort_by(sample_collection: ListBasedCo
 
 
 @pytest.mark.asyncio()
-async def test_list_collection_get_honors_sort_order(sample_collection: ListBasedCollection[SampleItem]) -> None:
+async def test_list_collection_get_honors_sort_order(sample_collection: Collection[SampleItem]) -> None:
     filters = {"partition": {"exact": "alpha"}}
     item = await sample_collection.get(filters, sort={"name": "rank", "order": "desc"})  # type: ignore[arg-type]
     assert item is not None
@@ -652,14 +578,14 @@ async def test_list_collection_bulk_delete_and_size() -> None:
     items = [SampleItem(partition="bulk", index=i, name=f"item-{i}", status="bulk") for i in range(40)]
     collection = _build_collection(items)
     await collection.delete(items[:20])
-    assert collection.size() == 20
+    assert (await collection.size()) == 20
     await collection.delete(items[20:])
-    assert collection.size() == 0
+    assert (await collection.size()) == 0
 
 
 @pytest.mark.asyncio()
 async def test_list_collection_query_rejects_unknown_operator(
-    sample_collection: ListBasedCollection[SampleItem],
+    sample_collection: Collection[SampleItem],
 ) -> None:
     with pytest.raises(ValueError):
         await sample_collection.query(filter={"status": {"invalid": "x"}})  # type: ignore[arg-type]
@@ -673,17 +599,9 @@ async def test_list_collection_query_result_type() -> None:
     assert result.offset == 0
 
 
-class QueueItem(BaseModel):
-    idx: int
-
-
-@pytest.fixture()
-def deque_queue() -> DequeBasedQueue[QueueItem]:
-    return DequeBasedQueue(QueueItem, [QueueItem(idx=i) for i in range(3)])
-
-
-def test_deque_queue_initial_size(deque_queue: DequeBasedQueue[QueueItem]) -> None:
-    assert deque_queue.size() == 3
+@pytest.mark.asyncio()
+async def test_deque_queue_initial_size(deque_queue: DequeBasedQueue[QueueItem]) -> None:
+    assert (await deque_queue.size()) == 3
 
 
 def test_deque_queue_item_type(deque_queue: DequeBasedQueue[QueueItem]) -> None:
@@ -701,7 +619,7 @@ async def test_deque_queue_enqueue_appends_items(deque_queue: DequeBasedQueue[Qu
     items = [QueueItem(idx=3), QueueItem(idx=4)]
     returned = await deque_queue.enqueue(items)
     assert returned == items
-    assert deque_queue.size() == 5
+    assert (await deque_queue.size()) == 5
 
 
 @pytest.mark.asyncio()
@@ -718,7 +636,7 @@ async def test_deque_queue_enqueue_rejects_wrong_type(deque_queue: DequeBasedQue
 async def test_deque_queue_dequeue_respects_limit(deque_queue: DequeBasedQueue[QueueItem], limit: int) -> None:
     result = await deque_queue.dequeue(limit)
     assert len(result) == min(limit, 3)
-    assert deque_queue.size() == 3 - min(limit, 3)
+    assert (await deque_queue.size()) == 3 - min(limit, 3)
 
 
 @pytest.mark.asyncio()
@@ -730,14 +648,14 @@ async def test_deque_queue_dequeue_zero_returns_empty(deque_queue: DequeBasedQue
 async def test_deque_queue_dequeue_more_than_available(deque_queue: DequeBasedQueue[QueueItem]) -> None:
     result = await deque_queue.dequeue(10)
     assert len(result) == 3
-    assert deque_queue.size() == 0
+    assert (await deque_queue.size()) == 0
 
 
 @pytest.mark.asyncio()
 async def test_deque_queue_peek_preserves_items(deque_queue: DequeBasedQueue[QueueItem]) -> None:
     snapshot = await deque_queue.peek(2)
     assert [item.idx for item in snapshot] == [0, 1]
-    assert deque_queue.size() == 3
+    assert (await deque_queue.size()) == 3
 
 
 @pytest.mark.asyncio()
@@ -757,25 +675,15 @@ async def test_deque_queue_handles_large_volume() -> None:
     queue = DequeBasedQueue(QueueItem)
     items = [QueueItem(idx=i) for i in range(2000)]
     await queue.enqueue(items)
-    assert queue.size() == 2000
+    assert (await queue.size()) == 2000
     drained = await queue.dequeue(1500)
     assert len(drained) == 1500
-    assert queue.size() == 500
-
-
-@pytest.fixture()
-def dict_key_value_data() -> Dict[str, int]:
-    return {"alpha": 1, "beta": 2}
-
-
-@pytest.fixture()
-def dict_key_value(dict_key_value_data: Dict[str, int]) -> DictBasedKeyValue[str, int]:
-    return DictBasedKeyValue(dict_key_value_data)
+    assert (await queue.size()) == 500
 
 
 @pytest.mark.asyncio()
 async def test_dict_key_value_initial_state(dict_key_value: DictBasedKeyValue[str, int]) -> None:
-    assert dict_key_value.size() == 2
+    assert await dict_key_value.size() == 2
     assert await dict_key_value.get("alpha") == 1
     assert await dict_key_value.get("missing") is None
 
@@ -789,20 +697,20 @@ async def test_dict_key_value_has_handles_presence(dict_key_value: DictBasedKeyV
 @pytest.mark.asyncio()
 async def test_dict_key_value_set_updates_and_expands(dict_key_value: DictBasedKeyValue[str, int]) -> None:
     await dict_key_value.set("gamma", 3)
-    assert dict_key_value.size() == 3
+    assert await dict_key_value.size() == 3
     await dict_key_value.set("alpha", 99)
     assert await dict_key_value.get("alpha") == 99
-    assert dict_key_value.size() == 3
+    assert await dict_key_value.size() == 3
 
 
 @pytest.mark.asyncio()
 async def test_dict_key_value_pop_returns_default(dict_key_value: DictBasedKeyValue[str, int]) -> None:
     result = await dict_key_value.pop("beta")
     assert result == 2
-    assert dict_key_value.size() == 1
+    assert await dict_key_value.size() == 1
     result = await dict_key_value.pop("missing", 42)
     assert result == 42
-    assert dict_key_value.size() == 1
+    assert await dict_key_value.size() == 1
 
 
 @pytest.mark.asyncio()
@@ -811,3 +719,136 @@ async def test_dict_key_value_does_not_mutate_input_mapping(dict_key_value_data:
     await key_value.set("gamma", 3)  # type: ignore[arg-type]
     await key_value.pop("alpha")  # type: ignore[arg-type]
     assert dict_key_value_data == {"alpha": 1, "beta": 2}
+
+
+@pytest.mark.mongo
+@pytest.mark.asyncio()
+async def test_mongo_based_sanity_check(temporary_mongo_database: AsyncDatabase[Any]) -> None:
+    from agentlightning.store.collection.mongo import (
+        MongoBasedCollection,
+        MongoBasedKeyValue,
+        MongoBasedQueue,
+        MongoClientPool,
+    )
+
+    async with MongoClientPool(temporary_mongo_database.client) as client_pool:
+        collection = MongoBasedCollection[Any](
+            client_pool, temporary_mongo_database.name, "test", "test-123", ["rollout_id"], Rollout
+        )
+        await collection.ensure_collection()
+
+        start_time = time.time()
+        await collection.insert(
+            [Rollout(rollout_id="test-123", input="test-123", start_time=start_time, status="running")]
+        )
+
+        result = await collection.query(filter={"status": {"exact": "running"}})
+        assert result.items == [
+            Rollout(rollout_id="test-123", input="test-123", start_time=start_time, status="running")
+        ]
+
+        rollout_queue = MongoBasedQueue[str](
+            client_pool, temporary_mongo_database.name, "rollout_queue", "partition-1", str
+        )
+        await rollout_queue.ensure_collection()
+
+        await rollout_queue.enqueue(["r1", "r2", "r3"])
+        assert await rollout_queue.size() == 3
+        assert await rollout_queue.peek(2) == ["r1", "r2"]
+        assert await rollout_queue.dequeue(2) == ["r1", "r2"]
+        assert await rollout_queue.size() == 1
+
+        span_kv = MongoBasedKeyValue[str, int](
+            client_pool, temporary_mongo_database.name, "span_sequence_ids", "partition-1", str, int
+        )
+        await span_kv.ensure_collection()
+
+        await span_kv.set("span-123", 1)
+        assert await span_kv.has("span-123")
+        assert await span_kv.get("span-123") == 1
+        assert await span_kv.pop("span-123") == 1
+        assert not await span_kv.has("span-123")
+
+
+@pytest.mark.mongo
+@pytest.mark.asyncio()
+async def test_mongo_ensure_collection_creates_partition_scoped_index(
+    temporary_mongo_database: AsyncDatabase[Any],
+) -> None:
+    from agentlightning.store.collection.mongo import MongoBasedCollection, MongoClientPool
+
+    collection_name = f"ensure-{uuid4().hex}"
+    async with MongoClientPool(temporary_mongo_database.client) as client_pool:
+        collection = MongoBasedCollection[Any](
+            client_pool,
+            temporary_mongo_database.name,
+            collection_name,
+            "partition-ensure",
+            ["name", "index"],
+            SampleItem,
+        )
+        await collection.ensure_collection()
+
+        unique_index = None
+        async for index in await temporary_mongo_database[collection_name].list_indexes():  # type: ignore
+            if index["name"] == "uniq_partition_name_index" and index.get("unique"):  # type: ignore
+                unique_index = index  # type: ignore
+                break
+
+        assert unique_index is not None, "expected unique partition/index key"
+        key_pairs = list(unique_index["key"].items())  # type: ignore
+        assert key_pairs == [("partition_id", 1), ("name", 1), ("index", 1)]
+
+
+@pytest.mark.mongo
+@pytest.mark.asyncio()
+async def test_mongo_ensure_collection_survives_concurrent_calls(temporary_mongo_database: AsyncDatabase[Any]) -> None:
+    from agentlightning.store.collection.mongo import MongoBasedCollection, MongoClientPool
+
+    collection_name = f"ensure-{uuid4().hex}"
+
+    async def ensure_once() -> None:
+        async with MongoClientPool(temporary_mongo_database.client) as client_pool:
+            collection = MongoBasedCollection(
+                client_pool,
+                temporary_mongo_database.name,
+                collection_name,
+                "partition-concurrent",
+                ["index"],
+                SampleItem,
+            )
+            await collection.ensure_collection()
+
+    await asyncio.gather(*(ensure_once() for _ in range(20)))
+
+    names = await temporary_mongo_database.list_collection_names()
+    assert names.count(collection_name) == 1
+
+    unique_indexes = []
+    async for index in await temporary_mongo_database[collection_name].list_indexes():  # type: ignore
+        if index["name"].startswith("uniq_partition_"):  # type: ignore
+            unique_indexes.append(index["name"])  # type: ignore
+    assert unique_indexes == ["uniq_partition_index"]
+
+
+@pytest.mark.mongo
+@pytest.mark.asyncio()
+async def test_mongo_ensure_collection_repeats_without_altering_indexes(
+    temporary_mongo_database: AsyncDatabase[Any],
+) -> None:
+    from agentlightning.store.collection.mongo import MongoBasedCollection, MongoClientPool
+
+    collection_name = f"ensure-{uuid4().hex}"
+    async with MongoClientPool(temporary_mongo_database.client) as client_pool:
+        collection = MongoBasedCollection(
+            client_pool, temporary_mongo_database.name, collection_name, "partition-repeat", ["index"], SampleItem
+        )
+        await collection.ensure_collection()
+        await collection.ensure_collection()
+
+        unique_indexes = []
+        async for index in await temporary_mongo_database[collection_name].list_indexes():  # type: ignore
+            if index["name"].startswith("uniq_partition_"):  # type: ignore
+                unique_indexes.append((index["name"], list(index["key"].items())))  # type: ignore
+
+        assert unique_indexes == [("uniq_partition_index", [("partition_id", 1), ("index", 1)])]

tests/store/test_implementation.py

@@ -882,7 +882,7 @@ async def test_query_resources_returns_history(store_fixture: LightningStore) ->
     )
 
     history = await store_fixture.query_resources()
-    assert [item.resources_id for item in history] == [first.resources_id, second.resources_id]
+    assert set([item.resources_id for item in history]) == {first.resources_id, second.resources_id}
     assert isinstance(history[0], ResourcesUpdate)
     assert isinstance(history[1], ResourcesUpdate)
 
@@ -1197,7 +1197,7 @@ async def test_duplicate_span_id_error(
     await store_fixture.add_otel_span(rollout.rollout_id, attempt_id, mock_readable_span)
 
     await store_fixture.add_otel_span(rollout.rollout_id, attempt_id, mock_readable_span)
-    assert "Duplicate span added" in caplog.text
+    assert "Duplicated span added" in caplog.text
 
 
 @pytest.mark.asyncio
@@ -1859,7 +1859,8 @@ async def test_wait_with_timeout_none_polling(store_fixture: LightningStore) ->
     await store_fixture.update_rollout(rollout_id=rollout.rollout_id, status="succeeded")
 
     # The wait should complete now
-    completed = await asyncio.wait_for(wait_task, timeout=1.0)
+    timeout = 1.0 if isinstance(store_fixture, InMemoryLightningStore) else 11.0
+    completed = await asyncio.wait_for(wait_task, timeout=timeout)
     assert len(completed) == 1
     assert completed[0].rollout_id == rollout.rollout_id
     assert completed[0].status == "succeeded"
@@ -2014,7 +2015,7 @@ async def test_wait_nonexistent_rollout_with_finite_timeout(store_fixture: Light
     elapsed = time.time() - start
 
     # Should timeout quickly (not wait indefinitely)
-    assert elapsed < 0.2
+    assert elapsed < 1.0
     assert len(completed) == 0
 
 
@@ -2080,8 +2081,12 @@ async def test_wait_polling_interval_with_timeout_none(store_fixture: LightningS
     completed = await wait_and_complete()
     elapsed = time.time() - start
 
-    # Should complete after ~0.5s (when we set the event)
-    assert 0.4 < elapsed < 0.7
+    if isinstance(store_fixture, InMemoryLightningStore):
+        # Should complete after ~0.5s (when we set the event)
+        assert 0.4 < elapsed < 0.7
+    else:
+        # Should be more than 5 seconds
+        assert 5 < elapsed < 15
     assert len(completed) == 1
     assert completed[0].status == "succeeded"
 

tests/store/test_restful.py

@@ -19,6 +19,7 @@ import pytest
 import pytest_asyncio
 from portpicker import pick_unused_port
 
+from agentlightning.store import LightningStore
 from agentlightning.store.client_server import LightningStoreClient, LightningStoreServer
 from agentlightning.store.memory import InMemoryLightningStore
 from agentlightning.types import (
@@ -69,12 +70,11 @@ async def _run_server_with_cors(cors_origins: List[str] | str | None = None):
 
 
 @pytest_asyncio.fixture
-async def server_client() -> (
-    AsyncGenerator[Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str], None]
-):
-    store = InMemoryLightningStore()
+async def server_client(
+    store_fixture: LightningStore,
+) -> AsyncGenerator[Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str], None]:
     port = pick_unused_port()
-    server = LightningStoreServer(store, "127.0.0.1", port)
+    server = LightningStoreServer(store_fixture, "127.0.0.1", port)
     await server.start()
     client = LightningStoreClient(server.endpoint)
     session = aiohttp.ClientSession()

uv.lock

@@ -152,6 +152,9 @@ dependencies = [
 apo = [
     { name = "poml", marker = "sys_platform == 'linux' or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-core-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-tinker') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-trl') or (extra == 'group-14-agentlightning-core-stable' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-core-stable' and extra == 'group-14-agentlightning-torch-legacy') or (extra == 'group-14-agentlightning-tinker' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-tinker' and extra == 'group-14-agentlightning-torch-legacy') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-cu128') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-gpu-stable') or (extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable') or (extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-trl') or (extra == 'group-14-agentlightning-torch-gpu-stable' and extra == 'group-14-agentlightning-torch-legacy') or (extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-trl')" },
 ]
+mongo = [
+    { name = "pymongo", marker = "sys_platform == 'linux' or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-core-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-tinker') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-trl') or (extra == 'group-14-agentlightning-core-stable' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-core-stable' and extra == 'group-14-agentlightning-torch-legacy') or (extra == 'group-14-agentlightning-tinker' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-tinker' and extra == 'group-14-agentlightning-torch-legacy') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-cu128') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-gpu-stable') or (extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable') or (extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-trl') or (extra == 'group-14-agentlightning-torch-gpu-stable' and extra == 'group-14-agentlightning-torch-legacy') or (extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-trl')" },
+]
 verl = [
     { name = "verl", version = "0.5.0", source = { registry = "https://pypi.org/simple" }, marker = "(sys_platform == 'linux' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (sys_platform != 'linux' and extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable') or (sys_platform != 'linux' and extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-stable') or (sys_platform != 'linux' and extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-trl') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-core-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-tinker') or (extra == 'group-14-agentlightning-core-stable' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-core-stable' and extra == 'group-14-agentlightning-torch-legacy') or (extra == 'group-14-agentlightning-tinker' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-tinker' and extra == 'group-14-agentlightning-torch-legacy') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-cu128') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-gpu-stable') or (extra != 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable') or (extra != 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra != 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-trl') or (extra != 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable' and extra == 'group-14-agentlightning-torch-legacy') or (extra != 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra != 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-trl') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-trl') or (extra == 'group-14-agentlightning-core-legacy' and extra != 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra != 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra != 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-trl') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-trl') or (extra != 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable' and extra == 'group-14-agentlightning-torch-legacy') or (extra != 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra != 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-trl')" },
     { name = "verl", version = "0.6.1", source = { registry = "https://pypi.org/simple" }, marker = "(sys_platform == 'linux' and extra == 'group-14-agentlightning-core-stable') or (sys_platform == 'linux' and extra == 'group-14-agentlightning-tinker') or (sys_platform == 'linux' and extra == 'group-14-agentlightning-torch-cpu') or (sys_platform == 'linux' and extra != 'group-14-agentlightning-torch-gpu-legacy') or (sys_platform != 'linux' and extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable') or (sys_platform != 'linux' and extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-stable') or (sys_platform != 'linux' and extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-trl') or (sys_platform == 'linux' and extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable') or (sys_platform == 'linux' and extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-stable') or (sys_platform == 'linux' and extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-trl') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-core-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-tinker') or (extra == 'group-14-agentlightning-core-stable' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-core-stable' and extra == 'group-14-agentlightning-torch-legacy') or (extra == 'group-14-agentlightning-tinker' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-tinker' and extra == 'group-14-agentlightning-torch-legacy') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-cu128') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-gpu-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-trl') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-trl') or (extra != 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable') or (extra != 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra != 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-trl') or (extra != 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable' and extra == 'group-14-agentlightning-torch-legacy') or (extra != 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra != 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-trl')" },
@@ -359,6 +362,7 @@ requires-dist = [
     { name = "portpicker" },
     { name = "psutil" },
     { name = "pydantic", specifier = ">=2.11" },
+    { name = "pymongo", marker = "extra == 'mongo'" },
     { name = "rich" },
     { name = "setproctitle" },
     { name = "uvicorn" },
@@ -366,7 +370,7 @@ requires-dist = [
     { name = "verl", marker = "extra == 'verl'", specifier = ">=0.5.0" },
     { name = "vllm", marker = "extra == 'verl'", specifier = ">=0.8.4" },
 ]
-provides-extras = ["apo", "verl"]
+provides-extras = ["apo", "verl", "mongo"]
 
 [package.metadata.requires-dev]
 agents = [
@@ -8472,6 +8476,77 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/81/40/b2d7b9fdccc63e48ae4dbd363b6b89eb7ac346ea49ed667bb71f92af3021/pymdown_extensions-10.17.1-py3-none-any.whl", hash = "sha256:1f160209c82eecbb5d8a0d8f89a4d9bd6bdcbde9a8537761844cfc57ad5cd8a6", size = 266310, upload-time = "2025-11-11T21:44:56.809Z" },
 ]
 
+[[package]]
+name = "pymongo"
+version = "4.15.4"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "dnspython", marker = "sys_platform == 'linux' or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-core-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-tinker') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra == 'group-14-agentlightning-core-legacy' and extra == 'group-14-agentlightning-trl') or (extra == 'group-14-agentlightning-core-stable' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-core-stable' and extra == 'group-14-agentlightning-torch-legacy') or (extra == 'group-14-agentlightning-tinker' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-tinker' and extra == 'group-14-agentlightning-torch-legacy') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-cu128') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-gpu-legacy') or (extra == 'group-14-agentlightning-torch-cpu' and extra == 'group-14-agentlightning-torch-gpu-stable') or (extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-gpu-stable') or (extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra == 'group-14-agentlightning-torch-gpu-legacy' and extra == 'group-14-agentlightning-trl') or (extra == 'group-14-agentlightning-torch-gpu-stable' and extra == 'group-14-agentlightning-torch-legacy') or (extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-torch-stable') or (extra == 'group-14-agentlightning-torch-legacy' and extra == 'group-14-agentlightning-trl')" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/74/81/6d66e62a5d1c5323dca79e9fb34ac8211df76f6c16625f9499a37b796314/pymongo-4.15.4.tar.gz", hash = "sha256:6ba7cdf46f03f406f77969a8081cfb659af16c0eee26b79a0a14e25f6c00827b", size = 2471218, upload-time = "2025-11-11T20:52:37.31Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/5c/9e/5d6bd240a8d32e088078b37dcfa1579028c51d91168c0e992827ec1e87e6/pymongo-4.15.4-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:84c7c7624a1298295487d0dfd8dbec75d14db44c017b5087c7fe7d6996a96e3d", size = 811325, upload-time = "2025-11-11T20:50:22.286Z" },
+    { url = "https://files.pythonhosted.org/packages/1f/aa/1d707a836c436af60faa2db6f2706f9e74b5056d0bd77deb243c023b5e7b/pymongo-4.15.4-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:71a5ab372ebe4e05453bae86a008f6db98b5702df551219fb2f137c394d71c3a", size = 811668, upload-time = "2025-11-11T20:50:24.171Z" },
+    { url = "https://files.pythonhosted.org/packages/e8/17/8c50a695a7029d582da50875085465f01bf83e5146fb7dc3f671168aedb7/pymongo-4.15.4-cp310-cp310-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:eee407bf1058a8f0d5b203028997b42ea6fc80a996537cc2886f89573bc0770f", size = 1185476, upload-time = "2025-11-11T20:50:25.635Z" },
+    { url = "https://files.pythonhosted.org/packages/3e/e8/5449663ec341fb83c3e4d51011f65b61de8427620679510cca57386c9446/pymongo-4.15.4-cp310-cp310-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:22e286f5b9c13963bcaf9b9241846d388ac5022225a9e11c5364393a8cc3eb49", size = 1203857, upload-time = "2025-11-11T20:50:27.162Z" },
+    { url = "https://files.pythonhosted.org/packages/2f/ff/98768d4294f271175aedbad209d748ac769a3f35bee35f8c82b57b03ea4a/pymongo-4.15.4-cp310-cp310-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:67c3b84a2a0e1794b2fbfe22dc36711a03c6bc147d9d2e0f8072fabed7a65092", size = 1242538, upload-time = "2025-11-11T20:50:29.322Z" },
+    { url = "https://files.pythonhosted.org/packages/18/91/57b4a08b81686e0148a93ecd0149d747a31be82aafa0708143d642662893/pymongo-4.15.4-cp310-cp310-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:94e50149fb9d982c234d0efa9c0eec4a04db7e82a412d3dae2c4f03a9926360e", size = 1232831, upload-time = "2025-11-11T20:50:31.326Z" },
+    { url = "https://files.pythonhosted.org/packages/ff/17/52434425cde25e6d0743a6d8af8a8b88ffbd05cce595993facf09a5d0559/pymongo-4.15.4-cp310-cp310-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:c1903c0966969cf3e7b30922956bd82eb09e6a3f3d7431a727d12f20104f66d3", size = 1200182, upload-time = "2025-11-11T20:50:33.339Z" },
+    { url = "https://files.pythonhosted.org/packages/9d/7f/d5c975dcbfd339f3cd3eae2055fe6d96fb546508e1954fe263c0304e0317/pymongo-4.15.4-cp310-cp310-win32.whl", hash = "sha256:20ffcd883b6e187ef878558d0ebf9f09cc46807b6520022592522d3cdd21022d", size = 798325, upload-time = "2025-11-11T20:50:35.214Z" },
+    { url = "https://files.pythonhosted.org/packages/b2/1f/78b2d3d7b35284c5da80342ce2b7e4087901ff8fb030eccaa654b5d3d061/pymongo-4.15.4-cp310-cp310-win_amd64.whl", hash = "sha256:68ea93e7d19d3aa3182a6e41ba68288b9b234a3b0a70b368feb95fff3f94413f", size = 808144, upload-time = "2025-11-11T20:50:36.621Z" },
+    { url = "https://files.pythonhosted.org/packages/30/d2/95505fb5a699180a215553f622702464bc47000e5e782cc846098dcdfc37/pymongo-4.15.4-cp310-cp310-win_arm64.whl", hash = "sha256:abfe72630190c0dc8f2222b02af7c4e5f72809d06b2ccb3f3ca83f6a7b60e302", size = 800933, upload-time = "2025-11-11T20:50:38.573Z" },
+    { url = "https://files.pythonhosted.org/packages/71/a4/b1a724352ab47a8925f30931a6aa6f905dcf473d8404156ef608ec325fbd/pymongo-4.15.4-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:b2967bda6ccac75aefad26c4ef295f5054181d69928bb9d1159227d6771e8887", size = 865881, upload-time = "2025-11-11T20:50:40.275Z" },
+    { url = "https://files.pythonhosted.org/packages/09/d4/6f4db5b64b0b71f0cbe608a80aea8b2580b5e1db4da1f9a70ae5531e9f1d/pymongo-4.15.4-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:7df1fad859c61bdbe0e2a0dec8f5893729d99b4407b88568e0e542d25f383f57", size = 866225, upload-time = "2025-11-11T20:50:41.842Z" },
+    { url = "https://files.pythonhosted.org/packages/0f/44/9d96fa635b838348109f904f558aa6675fdfb0a9265060050d7a92afbf97/pymongo-4.15.4-cp311-cp311-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:990c4898787e706d0ab59141cf5085c981d89c3f86443cd6597939d9f25dd71d", size = 1429778, upload-time = "2025-11-11T20:50:43.801Z" },
+    { url = "https://files.pythonhosted.org/packages/9f/e6/eac0b3ca4ea1cd437983f1409cb6260e606cce11ea3cb6f5ccd8629fa5c2/pymongo-4.15.4-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ad7ff0347e8306fc62f146bdad0635d9eec1d26e246c97c14dd1a189d3480e3f", size = 1456739, upload-time = "2025-11-11T20:50:45.479Z" },
+    { url = "https://files.pythonhosted.org/packages/73/7e/b7adba0c8dfc2dced7632c61425a70048bddf953b07bf6232a4ea7f0fb7e/pymongo-4.15.4-cp311-cp311-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:dd8c78c59fd7308239ef9bcafb7cd82f08cbc9466d1cfda22f9025c83468bf6d", size = 1514659, upload-time = "2025-11-11T20:50:47.517Z" },
+    { url = "https://files.pythonhosted.org/packages/20/8b/cdc129f1bee5595018c52ff81baaec818301e705ee39cf00d9d5f68a3d0d/pymongo-4.15.4-cp311-cp311-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:44d95677aa23fe479bb531b393a4fad0210f808af52e4ab2b79c0b540c828957", size = 1500700, upload-time = "2025-11-11T20:50:49.183Z" },
+    { url = "https://files.pythonhosted.org/packages/1f/02/e706a63f00542531a4c723258ae3da3439925de02215710a18813fbe1db4/pymongo-4.15.4-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:4ab985e61376ae5a04f162fb6bdddaffc7beec883ffbd9d84ea86a71be794d74", size = 1452011, upload-time = "2025-11-11T20:50:51.568Z" },
+    { url = "https://files.pythonhosted.org/packages/37/36/6b78b105e8e1174ebda592ad31f02cb98ee9bd8bb2eeb621f54e2c714d03/pymongo-4.15.4-cp311-cp311-win32.whl", hash = "sha256:2f811e93dbcba0c488518ceae7873a40a64b6ad273622a18923ef2442eaab55c", size = 844471, upload-time = "2025-11-11T20:50:53.362Z" },
+    { url = "https://files.pythonhosted.org/packages/a5/0d/3d009eed6ae045ee4f62877878070a07405af5e368d60a4a35efd177c25b/pymongo-4.15.4-cp311-cp311-win_amd64.whl", hash = "sha256:53bfcd8c11086a2457777cb4b1a6588d9dd6af77aeab47e04f2af02e3a077e59", size = 859189, upload-time = "2025-11-11T20:50:55.198Z" },
+    { url = "https://files.pythonhosted.org/packages/d5/40/d5713b1d5e0b10402446632bab6a88918cd13e5fe1fa26beac177eb37dac/pymongo-4.15.4-cp311-cp311-win_arm64.whl", hash = "sha256:2096964b2b93607ed80a62ac6664396a826b7fe34e2b1eed3f20784681a17827", size = 848369, upload-time = "2025-11-11T20:50:57.164Z" },
+    { url = "https://files.pythonhosted.org/packages/75/bb/09176c965d994352efd1407c9139799218f3fe1d18382dff34ef64e0bd22/pymongo-4.15.4-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:4ab4eef031e722a8027c338c3d71704a8c85c17c64625d61c6effdf8a893b971", size = 920943, upload-time = "2025-11-11T20:50:59.056Z" },
+    { url = "https://files.pythonhosted.org/packages/94/97/d212bd8d9106acecf6948cc0a0ed640f58d8afaed427481b9e79db08f45c/pymongo-4.15.4-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:0e12551e28007a341d15ebca5a024ef487edf304d612fba5efa1fd6b4d9a95a9", size = 920687, upload-time = "2025-11-11T20:51:00.683Z" },
+    { url = "https://files.pythonhosted.org/packages/ff/81/7be727d6172fd80d8dd1c6fedb78675936396d2f2067fab270e443e04621/pymongo-4.15.4-cp312-cp312-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:1d21998fb9ccb3ea6d59a9f9971591b9efbcfbbe46350f7f8badef9b107707f3", size = 1690340, upload-time = "2025-11-11T20:51:02.392Z" },
+    { url = "https://files.pythonhosted.org/packages/42/5a/91bf00e9d30d18b3e8ef3fa222964ba1e073d82c5f38dae027e63d36bcfd/pymongo-4.15.4-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9f83e8895d42eb51d259694affa9607c4d56e1c784928ccbbac568dc20df86a8", size = 1726082, upload-time = "2025-11-11T20:51:04.353Z" },
+    { url = "https://files.pythonhosted.org/packages/ff/08/b7d8e765efa64cddf1844e8b889454542c765f8d119c87a4904f45addc07/pymongo-4.15.4-cp312-cp312-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:0bd8126a507afa8ce4b96976c8e28402d091c40b7d98e3b5987a371af059d9e7", size = 1800624, upload-time = "2025-11-11T20:51:06.222Z" },
+    { url = "https://files.pythonhosted.org/packages/35/b0/40ec073ccc2cf95e8743315e6c92a81f37698d2e618c83ec7d9c3b647bd0/pymongo-4.15.4-cp312-cp312-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:e799e2cba7fcad5ab29f678784f90b1792fcb6393d571ecbe4c47d2888af30f3", size = 1785469, upload-time = "2025-11-11T20:51:07.893Z" },
+    { url = "https://files.pythonhosted.org/packages/82/da/b1a27064404d5081f5391c3c81e4a6904acccb4766598e3aa14399d36feb/pymongo-4.15.4-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:563e793ad87633e50ad43a8cd2c740fbb17fca4a4637185996575ddbe99960b8", size = 1718540, upload-time = "2025-11-11T20:51:09.574Z" },
+    { url = "https://files.pythonhosted.org/packages/e7/8c/bee6159b4e434dc0413b399af2bd3795ef7427b2c2fe1b304df250c0a3d8/pymongo-4.15.4-cp312-cp312-win32.whl", hash = "sha256:39bb3c12c772241778f4d7bf74885782c8d68b309d3c69891fe39c729334adbd", size = 891308, upload-time = "2025-11-11T20:51:11.67Z" },
+    { url = "https://files.pythonhosted.org/packages/cf/cb/cb70455fe2eadf4f6ccd27fe215e342b242e8b53780aeafb96cd1c3bf506/pymongo-4.15.4-cp312-cp312-win_amd64.whl", hash = "sha256:6f43326f36bc540b04f5a7f1aa8be40b112d7fc9f6e785ae3797cd72a804ffdd", size = 910911, upload-time = "2025-11-11T20:51:13.283Z" },
+    { url = "https://files.pythonhosted.org/packages/41/81/20486a697474b7de25faee91d9c478eb410ae78cb4e50b15000184944a48/pymongo-4.15.4-cp312-cp312-win_arm64.whl", hash = "sha256:263cfa2731a4bbafdce2cf06cd511eba8957bd601b3cad9b4723f2543d42c730", size = 896347, upload-time = "2025-11-11T20:51:15.981Z" },
+    { url = "https://files.pythonhosted.org/packages/51/10/09551492e484f7055194d91c071c827fc65261156e4daced35e67e97b893/pymongo-4.15.4-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:6ff080f23a12c943346e2bba76cf19c3d14fb3625956792aa22b69767bfb36de", size = 975326, upload-time = "2025-11-11T20:51:17.693Z" },
+    { url = "https://files.pythonhosted.org/packages/aa/6e/8f153a6d7eaec9b334975000e16bfd11ec4050e8729d3e2ee67d7022f526/pymongo-4.15.4-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:c4690e01d03773f7af21b1a8428029bd534c9fe467c6b594c591d8b992c0a975", size = 975132, upload-time = "2025-11-11T20:51:19.58Z" },
+    { url = "https://files.pythonhosted.org/packages/7c/7d/037498c1354fae1ce2fc7738c981a7447a5fee021c22e76083540cc1f9d6/pymongo-4.15.4-cp313-cp313-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:78bfe3917d0606b30a91b02ad954c588007f82e2abb2575ac2665259b051a753", size = 1950964, upload-time = "2025-11-11T20:51:21.262Z" },
+    { url = "https://files.pythonhosted.org/packages/ef/96/7c6b14956ef2ab99600d93b43429387394df6a99f5293cd0371c59a77a02/pymongo-4.15.4-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:f53c83c3fd80fdb412ce4177d4f59b70b9bb1add6106877da044cf21e996316b", size = 1995249, upload-time = "2025-11-11T20:51:23.248Z" },
+    { url = "https://files.pythonhosted.org/packages/2a/16/0e0495b38dd64efbfd6f2eb47535895c8df4a78e384aee78190fe2ecfa84/pymongo-4.15.4-cp313-cp313-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:2e41d6650c1cd77a8e7556ad65133455f819f8c8cdce3e9cf4bbf14252b7d805", size = 2086580, upload-time = "2025-11-11T20:51:25.294Z" },
+    { url = "https://files.pythonhosted.org/packages/7d/c0/692545232a17d5772d15c7e50d54415bdd9b88018e2228607c96766af961/pymongo-4.15.4-cp313-cp313-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:b60fd8125f52efffd697490b6ccebc6e09d44069ad9c8795df0a684a9a8f4b3c", size = 2070189, upload-time = "2025-11-11T20:51:27.162Z" },
+    { url = "https://files.pythonhosted.org/packages/6f/9f/aae8eb4650d9a62f26baca4f4da2a0f5cd1aabcd4229dabc43cd71e09ea2/pymongo-4.15.4-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:4d1a1a0406acd000377f34ae91cdb501fa73601a2d071e4a661e0c862e1b166e", size = 1985254, upload-time = "2025-11-11T20:51:29.136Z" },
+    { url = "https://files.pythonhosted.org/packages/b1/cd/50f49788caa317c7b00ccf0869805cb2b3046c2510f960cb07e8d3a74f73/pymongo-4.15.4-cp313-cp313-win32.whl", hash = "sha256:9c5710ed5f2af95315db0ee8ae02e9ff1e85e7b068c507d980bc24fe9d025257", size = 938134, upload-time = "2025-11-11T20:51:31.254Z" },
+    { url = "https://files.pythonhosted.org/packages/10/ad/6e96ccb3b7ab8be2e22b1c50b98aed0cae19253174bca6807fc8fd1ce34c/pymongo-4.15.4-cp313-cp313-win_amd64.whl", hash = "sha256:61b0863c7f9b460314db79b7f8541d3b490b453ece49afd56b611b214fc4b3b1", size = 962595, upload-time = "2025-11-11T20:51:33.118Z" },
+    { url = "https://files.pythonhosted.org/packages/22/23/9b9255e432df4bc276ecb9bb6e81c3376d8ee2b19de02d3751bb5c4a6fb1/pymongo-4.15.4-cp313-cp313-win_arm64.whl", hash = "sha256:0255af7d5c23c5e8cb4d9bb12906b142acebab0472117e1d5e3a8e6e689781cb", size = 944298, upload-time = "2025-11-11T20:51:35.13Z" },
+    { url = "https://files.pythonhosted.org/packages/9f/e6/f315ea84656adcd18d5b5e8b362b47c36bf606843098688cc0809b28c8a8/pymongo-4.15.4-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:539f9fa5bb04a09fc2965cdcae3fc91d1c6a1f4f1965b34df377bc7119e3d7cd", size = 1029994, upload-time = "2025-11-11T20:51:36.808Z" },
+    { url = "https://files.pythonhosted.org/packages/bb/0c/0c364db72cd80a503829885643478dd144a8bf05e1e853c89648a06ad34b/pymongo-4.15.4-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:68354a77cf78424d27216b1cb7c9b0f67da16aae855045279ba8d73bb61f5ad0", size = 1029615, upload-time = "2025-11-11T20:51:38.551Z" },
+    { url = "https://files.pythonhosted.org/packages/50/71/6f37eea22ffa5b136c1ca0a21ba390c273b582d800bc979961fbd46c9bcc/pymongo-4.15.4-cp314-cp314-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:a9a90d556c2ef1572d2aef525ef19477a82d659d117eb3a51fa99e617d07dc44", size = 2211805, upload-time = "2025-11-11T20:51:40.657Z" },
+    { url = "https://files.pythonhosted.org/packages/24/09/3a538cb82766ce89559c4ca0d5694f782485080db6a8f628784dc7debba8/pymongo-4.15.4-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:a1aac57614fb86a3fa707af3537c30eda5e7fd1be712c1f723296292ac057afe", size = 2264618, upload-time = "2025-11-11T20:51:42.651Z" },
+    { url = "https://files.pythonhosted.org/packages/51/6b/66b4fe2d3c566ed655d95b1d8947dfea05642b05a285a3081d6cebc4f5da/pymongo-4.15.4-cp314-cp314-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:c6c21b49c5e021d9ce02cac33525c722d4c6887f7cde19a5a9154f66cb845e84", size = 2371810, upload-time = "2025-11-11T20:51:44.372Z" },
+    { url = "https://files.pythonhosted.org/packages/92/2b/3989960c7de983c5cc05b2d43b26fa560fe9de433ee60b83259d6ee2cde3/pymongo-4.15.4-cp314-cp314-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:e93828768470026099119295c68ed0dbc0a50022558be5e334f6dbda054f1d32", size = 2351848, upload-time = "2025-11-11T20:51:46.548Z" },
+    { url = "https://files.pythonhosted.org/packages/31/93/ee9f8a42eed6ecb8dda52e586a470bf88007a298b0f1a2c4ea1ff352af8e/pymongo-4.15.4-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:11840e9eb5a650ac190f2a3473631073daddbabdbb2779b6709dfddd3ba3b872", size = 2251338, upload-time = "2025-11-11T20:51:48.335Z" },
+    { url = "https://files.pythonhosted.org/packages/a0/36/c6609f632bcaffcdf9f7e67cb888402a1df049a7c3ff2f56067a0b451a59/pymongo-4.15.4-cp314-cp314-win32.whl", hash = "sha256:f0907b46df97b01911bf2e10ddbb23c2303629e482d81372031fd7f4313b9013", size = 992893, upload-time = "2025-11-11T20:51:50.775Z" },
+    { url = "https://files.pythonhosted.org/packages/f0/23/4ec0f7c9bf3397b6cafaf714f5bfe0a9944e7af088daa01d258eec031118/pymongo-4.15.4-cp314-cp314-win_amd64.whl", hash = "sha256:111d7f65ccbde908546cb36d14e22f12a73a4de236fd056f41ed515d1365f134", size = 1021204, upload-time = "2025-11-11T20:51:52.691Z" },
+    { url = "https://files.pythonhosted.org/packages/2b/71/3813d15fa5ce6fb5fb40775bedc95a1970790f5aba968d92b014a796aab6/pymongo-4.15.4-cp314-cp314-win_arm64.whl", hash = "sha256:c689a5d057ef013612b5aa58e6bf52f7fdb186e22039f1a3719985b5d0399932", size = 1000608, upload-time = "2025-11-11T20:51:54.442Z" },
+    { url = "https://files.pythonhosted.org/packages/12/e7/10f3bc034fcec374dc46462b369205527478199a803169cb10e9e4b48c68/pymongo-4.15.4-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:cdfa57760745387cde93615a48f622bf1eeae8ae28103a8a5100b9389eec22f9", size = 1086725, upload-time = "2025-11-11T20:51:57.266Z" },
+    { url = "https://files.pythonhosted.org/packages/40/ee/b59cad7d46598d48708bd2a6559ea8b9cbb6fb9665d617b5a52b58de81b3/pymongo-4.15.4-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:4fd6ba610e5a54090c4055a15f38d19ad8bf11e6bbc5a173e945c755a16db455", size = 1086660, upload-time = "2025-11-11T20:51:59.114Z" },
+    { url = "https://files.pythonhosted.org/packages/0a/84/58efbde2b52a577f9162bb9b97605b6669354bb171bc241a0dc2639536d7/pymongo-4.15.4-cp314-cp314t-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:bd3c7945b8a5563aa3951db26ba534372fba4c781473f5d55ce6340b7523cb0f", size = 2531617, upload-time = "2025-11-11T20:52:01.006Z" },
+    { url = "https://files.pythonhosted.org/packages/f8/cd/7bd739d04b67c99f00c942465b8ab7659dc2c1ad80108b5f4f74eecdf9f3/pymongo-4.15.4-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:41e98a31e79d74e9d78bc1638b71c3a10a910eae7d3318e2ae8587c760931451", size = 2603756, upload-time = "2025-11-11T20:52:03.029Z" },
+    { url = "https://files.pythonhosted.org/packages/4a/39/5a3b01f7e5fd464656421246516723c02067e85bbfb52d30da7d79b8336f/pymongo-4.15.4-cp314-cp314t-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:d18d89073b5e752391c237d2ee86ceec1e02a4ad764b3029f24419eedd12723e", size = 2725205, upload-time = "2025-11-11T20:52:04.968Z" },
+    { url = "https://files.pythonhosted.org/packages/c7/a8/b06231d5ea48d0fcc47bf6c2cebfd8dbea3eda1a1d7bf786443cb9ef5b94/pymongo-4.15.4-cp314-cp314t-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:edbff27a56a80b8fe5c0319200c44e63b1349bf20db27d9734ddcf23c0d72b35", size = 2704793, upload-time = "2025-11-11T20:52:07.164Z" },
+    { url = "https://files.pythonhosted.org/packages/d0/a3/c0ea0da1185d3be4e73923ab3b74f14f424b40f787c710690c83004f147a/pymongo-4.15.4-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:f1d75f5b51304176631c12e5bf47eed021446669e5f99379b76fd2bd3929c1b4", size = 2582263, upload-time = "2025-11-11T20:52:09.016Z" },
+    { url = "https://files.pythonhosted.org/packages/c8/f7/29ce41f9e55b1dd912bed39b76e9326e23ff6c097c4a8de88b2c5bcd54e5/pymongo-4.15.4-cp314-cp314t-win32.whl", hash = "sha256:e1bf4e0689cc48e0cfa6aef17f107c298d8898de0c6e782ea5c98450ae93a62f", size = 1044009, upload-time = "2025-11-11T20:52:11.138Z" },
+    { url = "https://files.pythonhosted.org/packages/01/71/3fade727cc4c7ac77fe19c4e3a6bbfb66d7f46796108ba106f236c64492f/pymongo-4.15.4-cp314-cp314t-win_amd64.whl", hash = "sha256:3fc347ea5eda6c3a7177c3a9e4e9b4e570a444a351effda4a898c2d352a1ccd1", size = 1078479, upload-time = "2025-11-11T20:52:13.324Z" },
+    { url = "https://files.pythonhosted.org/packages/60/0f/d450350f103db4bb856cb1ee60c8b1fa68d5ac50c846896d74deba3e9950/pymongo-4.15.4-cp314-cp314t-win_arm64.whl", hash = "sha256:2d921b84c681c5385a6f7ba2b5740cb583544205a00877aad04b5b12ab86ad26", size = 1051155, upload-time = "2025-11-11T20:52:15.185Z" },
+]
+
 [[package]]
 name = "pynacl"
 version = "1.6.1"