query: add fieldFilters/sort/offset to list_runs

.github/workflows/contract-test-prod.yml

@@ -0,0 +1,26 @@
+name: contract-test (prod)
+
+# Runs the OpenAPI schema contract test against production.
+
+on:
+  push:
+    branches:
+      - main
+      - 'releases/**'
+  pull_request:
+    branches:
+      - main
+      - 'releases/**'
+  workflow_dispatch:
+
+# Must grant at least what the called workflow's job requests (contents: read);
+# a called workflow cannot receive more token permission than its caller.
+permissions:
+  contents: read
+
+jobs:
+  prod:
+    uses: ./.github/workflows/contract-test.yml
+    with:
+      url_api: https://pluto-api.trainy.ai
+    secrets: inherit

.github/workflows/contract-test-staging.yml

@@ -0,0 +1,28 @@
+name: contract-test (staging)
+
+# Runs the OpenAPI schema contract test against staging/dev, where the server
+# side typically deploys first — so client/server enum drift is caught before
+# it reaches production.
+
+on:
+  push:
+    branches:
+      - main
+      - 'releases/**'
+  pull_request:
+    branches:
+      - main
+      - 'releases/**'
+  workflow_dispatch:
+
+# Must grant at least what the called workflow's job requests (contents: read);
+# a called workflow cannot receive more token permission than its caller.
+permissions:
+  contents: read
+
+jobs:
+  staging:
+    uses: ./.github/workflows/contract-test.yml
+    with:
+      url_api: https://pluto-api-dev.trainy.ai
+    secrets: inherit

.github/workflows/contract-test.yml

@@ -0,0 +1,57 @@
+name: contract-test
+
+# Reusable workflow: run the OpenAPI schema contract test (tests/test_contract.py)
+# against a given Pluto API host. Callers (contract-test-prod.yml,
+# contract-test-staging.yml) supply the base URL so the two environments share
+# one identical job definition and can never drift.
+
+on:
+  workflow_call:
+    inputs:
+      url_api:
+        description: Pluto API base URL (e.g. https://pluto-api.trainy.ai)
+        required: true
+        type: string
+
+permissions: {}
+
+jobs:
+  contract:
+    # Reuse the "integration" environment so secrets.MLOP_API_TOKEN resolves.
+    environment: integration
+    permissions:
+      contents: read
+    runs-on: ubicloud-standard-4
+    strategy:
+      matrix:
+        python-version: ["3.11"]
+        poetry-version: ["2.1.1"]
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+
+      - name: Set up Python
+        uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405  # v6.2.0
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install Poetry
+        run: pip install poetry==${{ matrix.poetry-version }}
+
+      - name: Install deps
+        run: |
+          set -euox pipefail
+          # Clear stale cached venv that may be missing dev dependencies.
+          rm -rf .venv
+          # `--extras full` is required even for this light job: importing
+          # pluto.query pulls in pluto/__init__.py -> pluto/file.py -> soundfile.
+          poetry install --with dev --extras full
+          pip install -e ".[dev,full]"
+
+      - name: Run contract test
+        env:
+          PLUTO_URL_API: ${{ inputs.url_api }}
+          PLUTO_API_KEY: ${{ secrets.MLOP_API_TOKEN }}
+        run: |
+          set -euox pipefail
+          poetry run pytest tests/test_contract.py -rs -v

docs-api/query.mdx

@@ -44,6 +44,9 @@ list_runs(
     search: Optional[str] = None,
     tags: Optional[List[str]] = None,
     limit: int = 50,
+    sort: Optional[str] = None,
+    offset: int = 0,
+    filters: Optional[Dict[str, Any]] = None,
 ) -> List[Dict[str, Any]]
 ```
 
@@ -55,9 +58,16 @@ List runs in a project.
 | `search` | `Optional[str]` | Full-text search on run name. |
 | `tags` | `Optional[List[str]]` | Filter by tags (AND logic). Only runs matching *all* specified tags are returned. |
 | `limit` | `int` | Maximum number of runs to return (max 200). |
+| `sort` | `Optional[str]` | Server-side ordering as `[+\|-]<field>` (`-` = descending). Accepts built-in columns (`createdAt`, `updatedAt`, `name`, `status` and their snake_case aliases), `config.<key>.value` / `systemMetadata.<key>.value`, `summary_metrics.<name>` (LAST aggregation), and `heartbeat_at` (last logged data point). Defaults to `createdAt desc` server-side. |
+| `offset` | `int` | Skip-based pagination offset (0–100,000). Combine with `limit` to page; advance `offset` until a short page is returned. |
+| `filters` | `Optional[Dict[str, Any]]` | A wandb-compatible MongoDB-style query (the OR-capable filter surface). Boolean `$and`/`$or`/`$not` combine leaf terms; each leaf is `{field: value}` (equality) or `{field: {$op: value}}` with ops `$eq`, `$ne`, `$gt`, `$gte`, `$lt`, `$lte`, `$in`, `$nin`, `$regex`. Fields: `state`/`status`, `heartbeat_at`, `created_at`/`updated_at`, `name`, `tags`, `config.<key>`, `summaryMetrics.<key>`. `$or`/`$not` and `heartbeat_at`/`summaryMetrics.*` leaves require `project`. |
 
 **Returns:** `List[Dict[str, Any]]` — List of run dicts with keys: `id`, `name`, `displayId`, `status`, `tags`, `config`, `createdAt`, `updatedAt`, `url`.
 
+**Raises:**
+
+- `ValueError` — If `filters` uses an unrecognised operator or field.
+
 #### `Client.get_run`
 
 ```python
@@ -275,6 +285,9 @@ list_runs(
     search: Optional[str] = None,
     tags: Optional[List[str]] = None,
     limit: int = 50,
+    sort: Optional[str] = None,
+    offset: int = 0,
+    filters: Optional[Dict[str, Any]] = None,
 ) -> List[Dict[str, Any]]
 ```
 

pluto/query.py

@@ -18,6 +18,7 @@
     runs = client.list_runs("my-project")
 """
 
+import json
 import logging
 import os
 import time
@@ -45,6 +46,94 @@ def __init__(self, message: str, status_code: Optional[int] = None):
         super().__init__(message)
 
 
+# wandb-compatible vocabulary for the ``filters`` query (the ``/api/runs/list``
+# ``filter`` param). The server defines this grammar canonically in
+# ``lib/queries/run-filter-grammar.ts`` and publishes it as the
+# ``RunFilterGrammar`` OpenAPI component; these constants are the client mirror,
+# kept honest by ``tests/test_contract.py`` (drift fails CI). The server is the
+# authority on detailed semantics; the client does a light structural check so
+# obvious mistakes fail fast with a clear ``ValueError`` instead of an HTTP 400.
+_FILTER_BOOL_OPS = {'$and', '$or', '$not'}
+_FILTER_LEAF_OPS = {
+    '$eq',
+    '$ne',
+    '$gt',
+    '$gte',
+    '$lt',
+    '$lte',
+    '$in',
+    '$nin',
+    '$regex',
+}
+# Recognised leaf field names (exact) plus dotted-prefix families.
+_FILTER_FIELDS = {
+    'state',
+    'status',
+    'heartbeat_at',
+    'heartbeatAt',
+    'created_at',
+    'createdAt',
+    'updated_at',
+    'updatedAt',
+    'name',
+    'displayName',
+    'display_name',
+    'tags',
+}
+_FILTER_FIELD_PREFIXES = (
+    'config.',
+    'systemMetadata.',
+    'summaryMetrics.',
+    'summary_metrics.',
+)
+
+# Server clamps offset to this range (MAX_JSON_SORT_OFFSET).
+_MAX_OFFSET = 100_000
+
+
+def _validate_filters(node: Any, _depth: int = 0) -> None:
+    """Light structural validation of a wandb-style ``filters`` dict.
+
+    Recursively checks boolean operators (``$and``/``$or``/``$not``), leaf
+    operators, and field names against the supported vocabulary, raising
+    ``ValueError`` on anything unrecognised. The server enforces full semantics.
+    """
+    if _depth > 50:
+        raise ValueError('filters nested too deeply')
+    if not isinstance(node, dict):
+        raise ValueError(f'filters node must be a dict, got {type(node).__name__}')
+    for key, value in node.items():
+        if key in ('$and', '$or'):
+            if not isinstance(value, list):
+                raise ValueError(f'{key} expects a list')
+            for child in value:
+                _validate_filters(child, _depth + 1)
+        elif key == '$not':
+            _validate_filters(value, _depth + 1)
+        elif key.startswith('$'):
+            raise ValueError(f'unknown boolean operator: {key!r}')
+        else:
+            _validate_filter_field(key)
+            if isinstance(value, dict) and any(k.startswith('$') for k in value):
+                for op in value:
+                    if op not in _FILTER_LEAF_OPS:
+                        raise ValueError(
+                            f'unknown leaf operator {op!r} on field {key!r}; '
+                            f'expected one of {sorted(_FILTER_LEAF_OPS)}'
+                        )
+
+
+def _validate_filter_field(field_name: str) -> None:
+    if field_name in _FILTER_FIELDS:
+        return
+    if any(
+        field_name.startswith(p) and len(field_name) > len(p)
+        for p in _FILTER_FIELD_PREFIXES
+    ):
+        return
+    raise ValueError(f'unknown filter field: {field_name!r}')
+
+
 class Client:
     """HTTP client for reading data from the Pluto server.
 
@@ -115,6 +204,9 @@ def list_runs(
         search: Optional[str] = None,
         tags: Optional[List[str]] = None,
         limit: int = 50,
+        sort: Optional[str] = None,
+        offset: int = 0,
+        filters: Optional[Dict[str, Any]] = None,
     ) -> List[Dict[str, Any]]:
         """List runs in a project.
 
@@ -124,17 +216,74 @@ def list_runs(
             tags: Filter by tags (AND logic). Only runs matching *all*
                 specified tags are returned.
             limit: Maximum number of runs to return (max 200).
+            sort: Server-side ordering as ``[+|-]<field>`` (``-`` = descending).
+                Accepts built-in columns (``createdAt``, ``updatedAt``,
+                ``name``, ``status`` and their snake_case aliases),
+                ``config.<key>.value`` / ``systemMetadata.<key>.value``,
+                ``summary_metrics.<name>`` (LAST aggregation), and
+                ``heartbeat_at`` (last logged data point). Defaults to
+                ``createdAt desc`` server-side.
+            offset: Skip-based pagination offset (0–100,000). Combine with
+                ``limit`` to page; advance ``offset`` until a short page is
+                returned.
+            filters: A wandb-compatible MongoDB-style query (the OR-capable
+                filter surface). Boolean ``$and``/``$or``/``$not`` combine leaf
+                terms; each leaf is ``{field: value}`` (equality) or
+                ``{field: {$op: value}}`` with ops ``$eq``, ``$ne``, ``$gt``,
+                ``$gte``, ``$lt``, ``$lte``, ``$in``, ``$nin``, ``$regex``.
+                Fields: ``state``/``status``, ``heartbeat_at``,
+                ``created_at``/``updated_at``, ``name``, ``tags``,
+                ``config.<key>``, ``summaryMetrics.<key>``.
+                ``$or``/``$not`` and ``heartbeat_at``/``summaryMetrics.*`` leaves
+                require ``project``.
+
+        ``filters`` AND-combines with ``search``/``tags``. To find interrupted
+        spot jobs to retry, find the *alive* set and exclude it — or directly
+        select non-completed runs that have gone stale::
+
+            # wandb-style "alive" set (running OR reported data in the last hour):
+            list_runs(project, filters={"$or": [
+                {"state": "running"},
+                {"heartbeat_at": {"$gte": cutoff_iso}},
+            ]})
+
+            # retry candidates directly (not completed AND stale):
+            list_runs(project, filters={"$and": [
+                {"status": {"$ne": "COMPLETED"}},
+                {"heartbeat_at": {"$lt": cutoff_iso}},
+            ]})
 
         Returns:
             List of run dicts with keys: ``id``, ``name``, ``displayId``,
             ``status``, ``tags``, ``config``, ``createdAt``, ``updatedAt``,
             ``url``.
+
+        Raises:
+            ValueError: If ``filters`` uses an unrecognised operator or field.
         """
         params: Dict[str, Any] = {'projectName': project, 'limit': min(limit, 200)}
         if search is not None:
             params['search'] = search
         if tags is not None:
             params['tags'] = ','.join(tags)
+        if filters is not None:
+            _validate_filters(filters)
+            params['filter'] = json.dumps(filters)
+        if sort is not None:
+            params['sort'] = sort
+        if offset:
+            clamped = max(0, min(offset, _MAX_OFFSET))
+            if clamped != offset:
+                logger.debug(
+                    '%s: offset %d clamped to %d (max %d)',
+                    tag,
+                    offset,
+                    clamped,
+                    _MAX_OFFSET,
+                )
+            # A negative offset clamps to 0; don't send a redundant offset=0.
+            if clamped:
+                params['offset'] = clamped
         return self._get('/api/runs/list', params=params)['runs']
 
     def get_run(
@@ -549,9 +698,20 @@ def list_runs(
     search: Optional[str] = None,
     tags: Optional[List[str]] = None,
     limit: int = 50,
+    sort: Optional[str] = None,
+    offset: int = 0,
+    filters: Optional[Dict[str, Any]] = None,
 ) -> List[Dict[str, Any]]:
     """List runs in a project. See :meth:`Client.list_runs`."""
-    return _get_client().list_runs(project, search=search, tags=tags, limit=limit)
+    return _get_client().list_runs(
+        project,
+        search=search,
+        tags=tags,
+        limit=limit,
+        sort=sort,
+        offset=offset,
+        filters=filters,
+    )
 
 
 def get_run(project: str, run_id: Union[int, str]) -> Dict[str, Any]: