feat(policy): schema-driven condition builder for write/destructive tools (#966)

The previous "Add predicate" UX required users to type both the dotted
arg path (e.g. `args.domain`) and the value as JSON. Two problems:
  1. They need to know what fields each tool takes.
  2. They need to know what values are legal (which HA domains exist,
     which entities, etc.).

Replace the free-text path input with a dropdown sourced from the tool's
JSON schema, and replace the free-text value input with a (multi-)select
sourced from HA when the path has a known value source (domain, service,
entity_id today; trivially extensible). Free-text is still available via
an "(other — type a path)" escape hatch and as the automatic fallback
for ops that don't pair with a registry (regex / contains / gt / lt).

Server:
- New `/api/policy/tool-schema?name=...` returns
  `{paths: [...], value_sources: {path: source_key}}`. Read-only tools
  return empty paths so the UI falls back to free-text (gating those is
  low-value but still permitted manually).
- New `/api/policy/value-source?source=...` resolves a source key to a
  live list of choices. In-process 30s TTL cache avoids hammering HA
  when the user explores paths.
- value_sources.py registry maps (tool_name, arg_path) → source_key for
  the common write/destructive surface (call_service, set_entity,
  set_integration_enabled, get_history, etc.). New mappings are one
  dict entry plus, if a new source key, one fetcher.
- Both endpoints mount in addon + secret-prefix routes. Sidecar serves
  503 stubs (no FastMCP registry / HA client in that process).

UI: rename user-facing "predicate" → "condition" (CS jargon → SQL/JIRA
terminology users actually recognise; internal Pydantic class stays
`Predicate` so the wire format is unchanged). Form fetches the schema
lazily on first open, caches it on the card, refetches value choices
when path/op changes.

Includes test_schema_handlers.py covering: missing-name 400, sidecar
503, unknown-tool 404, read-only empty-paths, write-tool paths +
registry, JSON-schema enum passthrough, value-source 400 paths, both
HA-services payload shapes, domain filtering for entities/services,
and upstream-fetch 502 mapping.

Author: kingpanther13

src/ha_mcp/policy/handlers.py

@@ -13,12 +13,58 @@
 from .approval_queue import ApprovalQueue
 from .model import Policy
 from .persistence import load_policy, save_policy
+from .value_sources import (
+    all_value_sources_for,
+    fetch_value_source,
+)
+
+
+def _is_write_or_destructive(tool: Any) -> bool:
+    """True iff the tool can mutate state (not read-only) — the only
+    surface gating-UX polish is worth investing in. Read-only tools
+    still gate correctly if a user adds a rule, they just get the
+    free-text predicate fallback in the UI."""
+    ann = getattr(tool, "annotations", None)
+    # No annotations = treat as potentially write (safe default for the
+    # UI, matches the runtime gate which doesn't skip read-only).
+    return ann is None or getattr(ann, "readOnlyHint", None) is not True
+
+
+def _extract_arg_paths(parameters: dict[str, Any]) -> list[dict[str, Any]]:
+    """Turn a tool's JSON-schema ``parameters`` into a list of
+    ``{path, type, enum, description, required}`` entries the UI can
+    render as a dropdown.
+
+    Only top-level properties are surfaced — nested objects exist in a
+    few tools but the predicate language already supports dotted paths,
+    so users can fall back to free-text for those if needed.
+    """
+    if not isinstance(parameters, dict):
+        return []
+    props = parameters.get("properties") or {}
+    required = set(parameters.get("required") or [])
+    out: list[dict[str, Any]] = []
+    for name, schema in props.items():
+        if not isinstance(schema, dict):
+            continue
+        out.append(
+            {
+                "path": f"args.{name}",
+                "label": name,
+                "type": schema.get("type"),
+                "enum": schema.get("enum"),
+                "description": (schema.get("description") or "")[:200],
+                "required": name in required,
+            }
+        )
+    return out
 
 
 def build_policy_handlers(
     *,
     data_dir: Path,
     queue: ApprovalQueue,
+    server: Any | None = None,
 ) -> dict[str, Callable[[Request], Any]]:
 
     async def get_config(_: Request) -> JSONResponse:
@@ -121,10 +167,79 @@ async def post_deny(request: Request) -> JSONResponse:
             )
         return JSONResponse({"denied": True})
 
+    async def get_tool_schema(request: Request) -> JSONResponse:
+        """Return the predicate-builder hints for one tool.
+
+        Powers the schema-driven path/value pickers in the Tool Security
+        Policies tab. Returns ``paths: []`` for read-only tools so the
+        UI knows to hide the pickers (free-text fallback still works).
+        Returns 503 when the sidecar/stub backend is in use — the
+        sidecar has no FastMCP registry to introspect.
+        """
+        name = request.query_params.get("name") or ""
+        if not name:
+            return JSONResponse({"error": "missing 'name' query param"}, 400)
+        if server is None:
+            return JSONResponse(
+                {"error": "tool schema introspection unavailable in this mode"},
+                503,
+            )
+        try:
+            tools = await server.mcp.local_provider._list_tools()
+        except Exception as e:
+            return JSONResponse({"error": f"tool list failed: {e}"}, 500)
+        tool = next((t for t in tools if getattr(t, "name", None) == name), None)
+        if tool is None:
+            return JSONResponse({"error": f"tool not found: {name}"}, 404)
+        if not _is_write_or_destructive(tool):
+            return JSONResponse(
+                {
+                    "tool_name": name,
+                    "is_write_or_destructive": False,
+                    "paths": [],
+                    "value_sources": {},
+                }
+            )
+        return JSONResponse(
+            {
+                "tool_name": name,
+                "is_write_or_destructive": True,
+                "paths": _extract_arg_paths(getattr(tool, "parameters", {}) or {}),
+                "value_sources": all_value_sources_for(name),
+            }
+        )
+
+    async def get_value_source(request: Request) -> JSONResponse:
+        """Return live legal values for a named value source.
+
+        Sources are defined in ``policy/value_sources.py``. Extra query
+        params (e.g. ``domain=light``) are passed through to the
+        fetcher to support cascading selects.
+        """
+        source = request.query_params.get("source") or ""
+        if not source:
+            return JSONResponse({"error": "missing 'source' query param"}, 400)
+        if server is None:
+            return JSONResponse(
+                {"error": "value-source fetch unavailable in this mode"}, 503
+            )
+        params = {k: v for k, v in request.query_params.items() if k != "source"}
+        try:
+            values = await fetch_value_source(
+                source, client=server.client, params=params
+            )
+        except ValueError as e:
+            return JSONResponse({"error": str(e)}, 400)
+        except Exception as e:
+            return JSONResponse({"error": f"value-source fetch failed: {e}"}, 502)
+        return JSONResponse({"source": source, "values": values})
+
     return {
         "policy_get_config": get_config,
         "policy_put_config": put_config,
         "policy_get_pending": get_pending,
         "policy_post_approve": post_approve,
         "policy_post_deny": post_deny,
+        "policy_get_tool_schema": get_tool_schema,
+        "policy_get_value_source": get_value_source,
     }

src/ha_mcp/policy/value_sources.py

@@ -0,0 +1,159 @@
+"""Value-source registry for the predicate builder UI (#966).
+
+When a user picks a tool + arg-path in the Tool Security Policies tab,
+the UI needs to know whether to render a free-text input or a dropdown
+of legal values. This module maps `(tool_name, arg_path)` pairs to a
+named value source, plus implements the fetchers that read live values
+out of Home Assistant.
+
+Read-only tools are explicitly out of scope for gating UX polish, so the
+registry covers write/destructive tools only. Anything not in the
+registry falls back to free-text JSON entry in the UI.
+"""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+# (tool_name, arg_path) → value_source key. arg_path is the same dotted
+# string the user picks in the predicate "path" dropdown — i.e. with the
+# "args." prefix the evaluator uses.
+VALUE_SOURCE_REGISTRY: dict[tuple[str, str], str] = {
+    # Service-call gating — by far the most common case the user wants
+    # to author (gate ha_call_service when domain in [lock, alarm_*]).
+    ("ha_call_service", "args.domain"): "ha_domains",
+    ("ha_call_service", "args.service"): "ha_services",
+    ("ha_call_service", "args.entity_id"): "ha_entities",
+    # Bulk-control mirrors call_service's arg shape per item, but the
+    # outer wrapper takes a list — the predicate language can't reach
+    # into list items today, so no registry entries here. Free-text
+    # fallback still works.
+    ("ha_set_entity", "args.entity_id"): "ha_entities",
+    ("ha_get_state", "args.entity_id"): "ha_entities",
+    ("ha_get_entity", "args.entity_id"): "ha_entities",
+    ("ha_get_history", "args.entity_ids"): "ha_entities",
+    ("ha_remove_entity", "args.entity_id"): "ha_entities",
+    ("ha_update_device", "args.entity_id"): "ha_entities",
+    ("ha_get_entity_exposure", "args.entity_id"): "ha_entities",
+    ("ha_set_integration_enabled", "args.entity_id"): "ha_entities",
+}
+
+# Fetched-value TTL cache so the UI can click through path options
+# without hammering HA. 30s is long enough to cover normal exploration
+# but short enough that newly-added domains/entities appear quickly.
+_CACHE_TTL_SECONDS = 30.0
+_cache: dict[str, tuple[float, list[str]]] = {}
+
+
+def value_source_for(tool_name: str, arg_path: str) -> str | None:
+    return VALUE_SOURCE_REGISTRY.get((tool_name, arg_path))
+
+
+def all_value_sources_for(tool_name: str) -> dict[str, str]:
+    """Return {arg_path: value_source} for one tool — used by the UI to
+    decide which paths render as dropdowns vs free-text."""
+    return {
+        arg_path: source
+        for (tn, arg_path), source in VALUE_SOURCE_REGISTRY.items()
+        if tn == tool_name
+    }
+
+
+def _cache_get(key: str) -> list[str] | None:
+    entry = _cache.get(key)
+    if entry is None:
+        return None
+    ts, value = entry
+    if time.monotonic() - ts > _CACHE_TTL_SECONDS:
+        return None
+    return value
+
+
+def _cache_set(key: str, value: list[str]) -> None:
+    _cache[key] = (time.monotonic(), value)
+
+
+async def fetch_value_source(
+    source: str,
+    *,
+    client: Any,
+    params: dict[str, str] | None = None,
+) -> list[str]:
+    """Fetch live choices for a known value source.
+
+    Raises ValueError if ``source`` is unknown so the handler can return
+    a 400 instead of an empty list (which would look like "no choices
+    available" to the user).
+    """
+    params = params or {}
+    fetcher = _FETCHERS.get(source)
+    if fetcher is None:
+        raise ValueError(f"Unknown value source: {source!r}")
+    cache_key = source + "|" + "&".join(f"{k}={v}" for k, v in sorted(params.items()))
+    cached = _cache_get(cache_key)
+    if cached is not None:
+        return cached
+    values = sorted(await fetcher(client, params))
+    _cache_set(cache_key, values)
+    return values
+
+
+async def _fetch_ha_domains(client: Any, _params: dict[str, str]) -> list[str]:
+    services = await client.get_services()
+    # /services returns either {domain: {service: ...}} or
+    # [{domain, services}, ...] depending on HA version; handle both.
+    if isinstance(services, dict):
+        return list(services.keys())
+    if isinstance(services, list):
+        return [s["domain"] for s in services if isinstance(s, dict) and "domain" in s]
+    return []
+
+
+async def _fetch_ha_services(client: Any, params: dict[str, str]) -> list[str]:
+    services = await client.get_services()
+    domain_filter = params.get("domain")
+    out: set[str] = set()
+    if isinstance(services, dict):
+        if domain_filter:
+            out.update((services.get(domain_filter) or {}).keys())
+        else:
+            for svcs in services.values():
+                if isinstance(svcs, dict):
+                    out.update(svcs.keys())
+        return list(out)
+    if isinstance(services, list):
+        for entry in services:
+            if not isinstance(entry, dict):
+                continue
+            if domain_filter and entry.get("domain") != domain_filter:
+                continue
+            svcs = entry.get("services") or {}
+            if isinstance(svcs, dict):
+                out.update(svcs.keys())
+        return list(out)
+    return []
+
+
+async def _fetch_ha_entities(client: Any, params: dict[str, str]) -> list[str]:
+    states = await client.get_states()
+    domain_filter = params.get("domain")
+    out: list[str] = []
+    for s in states:
+        if not isinstance(s, dict):
+            continue
+        eid = s.get("entity_id")
+        if not isinstance(eid, str):
+            continue
+        if domain_filter and not eid.startswith(domain_filter + "."):
+            continue
+        out.append(eid)
+    return out
+
+
+_FETCHERS: dict[str, Callable[[Any, dict[str, str]], Awaitable[list[str]]]] = {
+    "ha_domains": _fetch_ha_domains,
+    "ha_services": _fetch_ha_services,
+    "ha_entities": _fetch_ha_entities,
+}