feat: query device state with CEL discovery predicates

docs/discovery.md

@@ -107,7 +107,7 @@ function(estop)                                               fleet emergency-st
 
 ### Discovery
 
-#### `discover(selector, offset=0, limit=200)`
+#### `discover(selector, offset=0, limit=200, *, where=None)`
 
 Resolves a selector to matched entities. Returns devices, function tuples,
 or event tuples depending on the selector scope. The response includes a
@@ -119,6 +119,65 @@ and switches to a name-and-labels summary above
 `DEVICE_CONNECT_FUNCTION_THRESHOLD` (default 20). The threshold is
 configurable via environment variable.
 
+##### State predicates
+
+Pass a CEL `where` expression to query stored device state without invoking
+a function. This also works for reporting-only sensors and instruments:
+
+```python
+discover(
+    "device(model_id:ophyd_async:SimStage)",
+    where="status.x_readback < 51.99 && status.x_setpoint == 2.0",
+)
+# {"scope": "device_only", "matched": 1, "returned": 1, "offset": 0,
+#  "next_offset": None,
+#  "results": [{"device_id": "stage-0051",
+#               "device_type": "ophyd_async:SimStage", "location": "lab-A"}],
+#  "label_histogram": {...}}
+```
+
+The registry uses the same CEL evaluator as `broadcast`. Discovery binds:
+
+| Variable | Source |
+| --- | --- |
+| `status` | Latest stored heartbeat status, including driver-specific fields |
+| `identity` | Registered identity plus `device_id` |
+| `labels` | Device capability labels, with legacy `status.location` and `identity.device_type` defaults for `location` and `type`; declared labels take precedence |
+
+Discovery does not provide broadcast's `bindings` payload. Missing fields
+and evaluation type errors make that device a non-match, as in broadcast.
+Malformed, empty or non-string expressions produce JSON-RPC `-32602`
+(invalid params), exposed by the tool as `error.code = "invalid_predicate"`.
+
+`discovery/listDevices` accepts optional `where` alongside `device_type`,
+`location`, `offset` and `limit`. It evaluates the predicate before paging;
+`total_matched` and `next_offset` refer to state matches. A successful
+predicate query includes `where_applied: true`. For predicate queries,
+visibility ACLs filter the matches before pagination and counting, so
+neither the total nor the cursor exposes hidden devices' state.
+The agent tool resolves its selector over
+visible state matches, then calculates `matched`, the label histogram and
+the requested page. For function/event selectors, `where` filters the
+owning devices before resolving their functions/events.
+
+With `where`, device-only tool results contain only `device_id`,
+`device_type` and `location`; full records remain inside the registry and
+tool process. A fleet of 5,000 devices with one state match therefore
+requires one registry page and returns one compact tool row. The registry
+still scans its tenant's stored snapshot; CPU cost scales with fleet size.
+
+State queries bypass the SDK's local fleet cache. The registry's existing
+snapshot cache still applies (`DC_FLEET_CACHE_TTL`, default 2 seconds), and
+state freshness also depends on device heartbeat frequency. Pages are not
+a transactionally frozen view if devices change state between requests.
+
+The server installs CEL support by default; agent tools need no predicate
+extra for registry discovery. Upgrade the server and agent's edge SDK
+together: clients reject replies without `where_applied: true` rather than
+returning an unfiltered fleet. There is no client fallback for older
+servers, and `where` requires registry mode (D2D is unsupported).
+Calls without `where` retain their existing behavior and response shape.
+
 #### `discover_labels(key=None, offset=0, limit=50)`
 
 Returns the fleet label vocabulary. Use this first when you do not know

packages/device-connect-agent-tools/device_connect_agent_tools/adapters/claude.py

@@ -91,15 +91,27 @@ async def discover_labels(args: dict[str, Any]) -> dict[str, Any]:
     "'device(category:camera, location:zone-A/*)', "
     "'device(*).function(direction:write)', 'event(modality:motion)'. "
     "Response includes a label_histogram (per-key vocabulary across the "
-    "matched set) so the agent can narrow next.",
-    {"selector": str, "offset": int, "limit": int},
+    "matched set) so the agent can narrow next. Optional CEL where filters "
+    "registry-stored status, identity and labels; device-only results are "
+    "compact ids, types and locations, including devices with no functions.",
+    {
+        "type": "object",
+        "properties": {
+            "selector": {"type": "string"},
+            "offset": {"type": "integer"},
+            "limit": {"type": "integer"},
+            "where": {"type": "string", "description": "CEL predicate over status, identity and labels"},
+        },
+        "required": ["selector", "offset", "limit"],
+    },
 )
 async def discover(args: dict[str, Any]) -> dict[str, Any]:
     return _text(
         _discover(
             selector=args["selector"],
             offset=int(args.get("offset", 0)),
             limit=int(args.get("limit", 200)),
+            where=args.get("where"),
         )
     )
 

packages/device-connect-agent-tools/device_connect_agent_tools/connection.py

@@ -432,20 +432,31 @@ def list_devices(
         self,
         device_type: Optional[str] = None,
         location: Optional[str] = None,
+        *,
+        where: Optional[str] = None,
     ) -> List[Dict[str, Any]]:
         """List devices via the discovery provider (D2D or registry)."""
-        return self._run(self._async_list_devices(device_type, location))
+        return self._run(self._async_list_devices(device_type, location, where=where))
 
     async def _async_list_devices(
         self,
         device_type: Optional[str] = None,
         location: Optional[str] = None,
+        *,
+        where: Optional[str] = None,
     ) -> List[Dict[str, Any]]:
         if self._provider is None:
             raise RuntimeError("Not connected — call connect() first")
-        devices = await self._provider.list_devices(
-            device_type=device_type, location=location,
-        )
+        if where is not None:
+            if not isinstance(self._provider, _SDKRegistryClient):
+                raise RuntimeError("Discovery where predicates require a registry connection; D2D is unsupported")
+            devices = await self._provider.list_devices(
+                device_type=device_type, location=location, where=where,
+            )
+        else:
+            devices = await self._provider.list_devices(
+                device_type=device_type, location=location,
+            )
         return [flatten_device(d) for d in devices]
 
     def invalidate_cache(self) -> None:

packages/device-connect-agent-tools/device_connect_agent_tools/tools.py

@@ -218,6 +218,8 @@ def discover(
     selector: str,
     offset: int = 0,
     limit: int = DEFAULT_DISCOVER_LIMIT,
+    *,
+    where: str | None = None,
 ) -> dict[str, Any]:
     """Resolve a selector to matched devices, functions, or events.
 
@@ -241,6 +243,11 @@ def discover(
         selector: A selector expression string.
         offset: Pagination offset (rows skipped).
         limit: Max rows per page (capped at DISCOVER_HARD_LIMIT).
+        where: Optional CEL predicate over registry-stored ``status``,
+            ``identity`` and ``labels``. Filters devices before resolving
+            functions/events. Device-only results contain device_id,
+            device_type and location. No callable function is required.
+            Requires a registry with state discovery support.
 
     Returns:
         A response envelope:
@@ -279,7 +286,15 @@ def discover(
 
     try:
         conn = get_connection()
-        devices = conn.list_devices()
+        if where is None:
+            devices = conn.list_devices()
+        else:
+            from device_connect_edge.predicate import PredicateCompileError
+
+            try:
+                devices = conn.list_devices(where=where)
+            except PredicateCompileError as e:
+                return _empty_envelope(scope=sel.scope.value, error=_error("invalid_predicate", str(e)))
     except Exception as e:
         logger.error("discover(%r) failed loading fleet: %s", selector, e)
         return _empty_envelope(
@@ -298,7 +313,13 @@ def discover(
         total = len(matched_devices)
         page_devices, next_offset = _paginate(matched_devices, safe_offset, safe_limit)
         expand = SMALL_FLEET_THRESHOLD > 0 and total <= SMALL_FLEET_THRESHOLD
-        results = [_device_summary_for_discover(d, expand) for d in page_devices]
+        if where is not None:
+            results = [
+                {key: d.get(key) for key in ("device_id", "device_type", "location")}
+                for d in page_devices
+            ]
+        else:
+            results = [_device_summary_for_discover(d, expand) for d in page_devices]
         histogram, multivalued, unique = label_histogram(matched_devices, count_unique=True)
         formatted_histogram = _format_label_histogram(histogram, multivalued, unique)
         return {

packages/device-connect-agent-tools/tests/test_claude_adapter.py

@@ -78,6 +78,18 @@ def _mock_sdk_and_connection():
 
 
 class TestClaudeAdapterExports:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("where_args", [{}, {"where": "status.online"}])
+    async def test_discover_exposes_and_forwards_optional_where(self, where_args):
+        from device_connect_agent_tools.adapters import claude as adapter
+
+        schema = adapter.discover._tool_schema
+        assert schema["properties"]["where"]["type"] == "string"
+        assert "where" not in schema["required"]
+        with patch.object(adapter, "_discover", return_value={"matched": 0}) as discover:
+            await adapter.discover.__wrapped__({"selector": "device(*)", **where_args})
+        discover.assert_called_once_with(selector="device(*)", offset=0, limit=200, where=where_args.get("where"))
+
     def test_module_exports_all_tools(self):
         from device_connect_agent_tools.adapters import claude as adapter
 

packages/device-connect-agent-tools/tests/test_discover.py

@@ -13,6 +13,54 @@
 import pytest
 
 from device_connect_agent_tools import tools as tools_mod
+from device_connect_edge.predicate import PredicateCompileError
+
+
+def test_discover_where_returns_compact_rows_and_paginates_matching_devices():
+    conn = MagicMock()
+    conn.list_devices.return_value = [
+        {"device_id": f"stage-{i}", "device_type": "stage", "location": "lab-A",
+         "labels": {"model_id": "ophyd_async:SimStage"},
+         "functions": [{"name": "huge_schema", "parameters": {"description": "x" * 10000}}]}
+        for i in range(3)
+    ] + [{"device_id": "sensor", "labels": {"model_id": "sensor"}}]
+    with patch.object(tools_mod, "get_connection", return_value=conn):
+        result = tools_mod.discover(
+            "device(model_id:ophyd_async:SimStage)", offset=1, limit=1, where="status.x_readback < 51.99",
+        )
+    conn.list_devices.assert_called_once_with(where="status.x_readback < 51.99")
+    assert (result["matched"], result["returned"], result["next_offset"]) == (3, 1, 2)
+    assert result["results"] == [{"device_id": "stage-1", "device_type": "stage", "location": "lab-A"}]
+    assert result["label_histogram"]["model_id"]["values"] == {"ophyd_async:SimStage": 3}
+
+
+def test_discover_invalid_where_is_a_structured_error():
+    conn = MagicMock()
+    conn.list_devices.side_effect = PredicateCompileError("failed to compile where")
+    with patch.object(tools_mod, "get_connection", return_value=conn):
+        result = tools_mod.discover("device(*)", where="status.x > > 1")
+    assert result["error"]["code"] == "invalid_predicate"
+    assert result["matched"] == 0
+    assert result["results"] == []
+
+
+def test_discover_where_does_not_mislabel_connection_value_errors():
+    conn = MagicMock()
+    conn.list_devices.side_effect = ValueError("malformed registry response")
+    with patch.object(tools_mod, "get_connection", return_value=conn):
+        result = tools_mod.discover("device(*)", where="status.online")
+    assert result["error"]["code"] == "connection_error"
+
+
+def test_discover_where_also_filters_function_and_event_scopes():
+    conn = MagicMock()
+    conn.list_devices.return_value = [SAMPLE_DEVICES[0]]
+    with patch.object(tools_mod, "get_connection", return_value=conn):
+        functions = tools_mod.discover("function(*)", where="status.online")
+        events = tools_mod.discover("event(*)", where="status.online")
+    assert functions["matched"] == 1
+    assert events["matched"] == 2
+    assert all(row["device_id"] == "cam-001" for row in functions["results"] + events["results"])
 
 
 # -- Fixture: labeled fleet ---------------------------------------

packages/device-connect-edge/device_connect_edge/device.py

@@ -1463,7 +1463,14 @@ def _run() -> bool:
         deadline = loop.time() + timeout_s
 
         try:
-            await asyncio.wait_for(self._where_eval_semaphore.acquire(), timeout_s)
+            if self._where_eval_semaphore.locked():
+                await asyncio.wait_for(self._where_eval_semaphore.acquire(), timeout_s)
+            else:
+                # acquire() completes immediately when capacity is available.
+                # wait_for() would schedule another task and yield even then,
+                # letting unrelated device work consume the whole deadline
+                # before this predicate is submitted to its idle executor.
+                await self._where_eval_semaphore.acquire()
         except asyncio.TimeoutError:
             self._logger.warning(
                 "Broadcast %s: where predicate timed out after %.3fs (skipping)",