Add WhatsApp trajectory parameter preview in Gradio.

Substitute runtime intent parameters into recorded steps using skill schema bindings before showing trajectory JSON in the demo.

Co-authored-by: Cursor <cursoragent@cursor.com>

app.py

@@ -17,6 +17,7 @@ from pathlib import Path
 import gradio as gr
 import requests
 
+from src.parameter_binder import apply_parameters
 from src.skill_router import SKILL_TO_TRAJECTORY, load_trajectory
 from src.skill_utils import resolve_skill
 
@@ -216,11 +217,13 @@ def _build_skills_catalog_card() -> str:
 
 def _trajectory_summary(skill: str, parameters: dict | None = None) -> tuple[str, str, str]:
     data = load_trajectory(skill)
+    params = parameters or {}
+    if params:
+        data = apply_parameters(data, skill, params)
     steps = data.get("steps", [])
     task = data.get("task", "—")
     app_pkg = data.get("app", "—")
     trajectory_file = Path(SKILL_TO_TRAJECTORY[skill]).name
-    params = parameters or {}
 
     summary = (
         f"### {SKILL_LABELS.get(skill, skill)}\n\n"
@@ -233,10 +236,12 @@ def _trajectory_summary(skill: str, parameters: dict | None = None) -> tuple[str
     if params:
         param_lines = "  \n".join(f"**{k}** · {v}" for k, v in params.items())
         summary += f"\n\n**Extracted parameters**  \n{param_lines}"
+        summary += "\n\n**Trajectory preview** · runtime parameters substituted into recorded steps"
 
     preview = {
         "skill": skill,
         "parameters": params,
+        "parameterized": bool(params),
         "task": task,
         "app": app_pkg,
         "steps": len(steps),

data/skill_schemas.json

@@ -55,7 +55,27 @@
         "required": true,
         "description": "Message body text"
       }
-    }
+    },
+    "bindings": [
+      {
+        "parameter": "contact",
+        "action": "set_text",
+        "resource_id_suffix": "search_input",
+        "package_contains": "whatsapp"
+      },
+      {
+        "parameter": "contact",
+        "action": "click",
+        "after_search": true,
+        "set_content_description": false
+      },
+      {
+        "parameter": "message",
+        "action": "set_text",
+        "resource_id_suffix": "entry",
+        "package_contains": "whatsapp"
+      }
+    ]
   },
   "camera_take_photo": {
     "description": "Open the camera and take a photo",

src/parameter_binder.py

@@ -0,0 +1,195 @@
+"""
+Substitute runtime intent parameters into recorded trajectory steps.
+
+Preview-only for the Gradio demo today; Pocket Automator will mirror this at replay time.
+"""
+
+from __future__ import annotations
+
+import copy
+import json
+from typing import Any
+
+from src.paths import DATA_DIR
+
+SKILL_SCHEMAS_PATH = DATA_DIR / "skill_schemas.json"
+
+_SEARCH_INPUT_SUFFIXES = (
+    "search_input",
+    "search_src_text",
+    "search_container",
+    "search_box",
+    "search_bar",
+    "search",
+)
+
+_MESSAGE_BODY_SUFFIXES = (
+    "entry",
+    "message",
+    "message_text",
+)
+
+_COMPOSE_BODY_SUFFIXES = (
+    "editor",
+    "body",
+)
+
+
+def load_skill_schemas() -> dict[str, Any]:
+    return json.loads(SKILL_SCHEMAS_PATH.read_text(encoding="utf-8"))
+
+
+def load_skill_schema(skill: str) -> dict[str, Any]:
+    schemas = load_skill_schemas()
+    if skill not in schemas:
+        raise KeyError(f"Unknown skill schema: {skill!r}")
+    return schemas[skill]
+
+
+def _resource_id_suffix(resource_id: str | None) -> str:
+    if not resource_id:
+        return ""
+    return resource_id.rsplit("/", 1)[-1].lower()
+
+
+def _suffix_matches(resource_id: str | None, suffix: str) -> bool:
+    if not suffix:
+        return False
+    needle = suffix.lower()
+    rid = _resource_id_suffix(resource_id)
+    return rid == needle or rid.endswith(needle)
+
+
+def is_search_input_step(step: dict[str, Any]) -> bool:
+    action = step.get("action") or {}
+    if action.get("type") != "set_text":
+        return False
+    resource_id = action.get("resourceId") or ""
+    if any(_suffix_matches(resource_id, suffix) for suffix in _SEARCH_INPUT_SUFFIXES):
+        return True
+    package_name = (step.get("packageName") or "").lower()
+    class_name = (action.get("className") or "").lower()
+    return "spotify" in package_name and "edittext" in class_name
+
+
+def find_preceding_search_step(steps: list[dict[str, Any]], before_index: int) -> dict[str, Any] | None:
+    for index in range(before_index - 1, -1, -1):
+        step = steps[index]
+        action = step.get("action") or {}
+        if action.get("type") != "set_text":
+            continue
+        value = action.get("value")
+        if not value or not str(value).strip():
+            continue
+        if is_search_input_step(step):
+            return step
+    return None
+
+
+def _binding_matches(
+    binding: dict[str, Any],
+    step: dict[str, Any],
+    steps: list[dict[str, Any]],
+    index: int,
+) -> bool:
+    action = step.get("action") or {}
+    if action.get("type") != binding.get("action"):
+        return False
+
+    resource_id_suffix = binding.get("resource_id_suffix")
+    if resource_id_suffix and not _suffix_matches(action.get("resourceId"), resource_id_suffix):
+        return False
+
+    if binding.get("after_search"):
+        return find_preceding_search_step(steps, index) is not None
+
+    package_contains = binding.get("package_contains")
+    if package_contains and package_contains.lower() not in (step.get("packageName") or "").lower():
+        return False
+
+    return True
+
+
+def _apply_binding(action: dict[str, Any], binding: dict[str, Any], value: str) -> dict[str, Any]:
+    updated = dict(action)
+    if binding.get("action") == "set_text":
+        updated["value"] = value
+    elif binding.get("action") == "click":
+        updated["text"] = value
+        if binding.get("set_content_description"):
+            updated["contentDescription"] = value
+    return updated
+
+
+def apply_parameters(
+    trajectory: dict[str, Any],
+    skill: str,
+    parameters: dict[str, str] | None,
+) -> dict[str, Any]:
+    """Return a copy of the trajectory with bound parameter values substituted."""
+    if not parameters:
+        return trajectory
+
+    schema = load_skill_schema(skill)
+    bindings = schema.get("bindings") or []
+    if not bindings:
+        return trajectory
+
+    result = copy.deepcopy(trajectory)
+    steps = result.get("steps") or []
+
+    for index, step in enumerate(steps):
+        action = step.get("action")
+        if not isinstance(action, dict):
+            continue
+
+        for binding in bindings:
+            param_name = binding.get("parameter")
+            if not param_name or param_name not in parameters:
+                continue
+            if not _binding_matches(binding, step, steps, index):
+                continue
+            step["action"] = _apply_binding(action, binding, str(parameters[param_name]))
+            action = step["action"]
+
+    return result
+
+
+def _self_test() -> None:
+    from src.skill_router import load_trajectory
+
+    trajectory = load_trajectory("whatsapp_send_message")
+    bound = apply_parameters(
+        trajectory,
+        "whatsapp_send_message",
+        {"contact": "mom", "message": "i'm on my way"},
+    )
+
+    search_values = [
+        step["action"]["value"]
+        for step in bound["steps"]
+        if step["action"]["type"] == "set_text"
+        and _suffix_matches(step["action"].get("resourceId"), "search_input")
+    ]
+    message_values = [
+        step["action"]["value"]
+        for step in bound["steps"]
+        if step["action"]["type"] == "set_text"
+        and _suffix_matches(step["action"].get("resourceId"), "entry")
+    ]
+
+    assert search_values and all(value == "mom" for value in search_values), search_values
+    assert message_values and all(value == "i'm on my way" for value in message_values), message_values
+
+    post_search_clicks = [
+        step["action"]
+        for index, step in enumerate(bound["steps"])
+        if step["action"]["type"] == "click"
+        and find_preceding_search_step(bound["steps"], index) is not None
+    ]
+    assert post_search_clicks and all(click.get("text") == "mom" for click in post_search_clicks)
+
+
+if __name__ == "__main__":
+    _self_test()
+    print("parameter_binder: ok")