fix(flow): bind API step fields by which step typed them, not by equal values

packages/flow/docs/API_TIER.md

@@ -50,12 +50,21 @@ every field it sent:
 
 | Source | Meaning | At run time |
 | --- | --- | --- |
-| `param` | The value of a recorded parameter | The run's value |
-| `select_label` | A select whose option the person picked by its label | The option with that label in the live form; refuse if it isn't exactly one |
+| `param` | A field a recorded step typed a parameter into | The run's value |
+| `select_label` | A select a recorded step typed an option's label into | The option with that label in the live form; refuse if it isn't exactly one |
 | `session` | A CSRF or session value the server issued | Read from the live form in this session |
 | `form_default` | The form's own default | Must still equal the recorded default |
 
-If a field can't be explained, nothing is proposed. The compiler also
+A field is bound by provenance: which recorded step typed into it (the
+field the click before the typing targeted), never by its value matching a
+parameter's. So a parameter whose demonstration value was blank still binds
+its field, and two parameters with equal values bind their own fields.
+
+If a field can't be explained, nothing is proposed. That includes a field
+that carries a parameter's value although no recorded step typed it there, a
+typed field the save sent with another value, a typed field the save doesn't
+send, a typing step whose field isn't a control of the form, and a field
+typed twice with different values. The compiler also
 doesn't propose an API step in these cases:
 
 - the save sends a secret field;

packages/flow/openadapt_flow/compiler/api_proposal.py

@@ -95,6 +95,78 @@ def _templated(value: str, by_value: dict[str, list[str]]) -> Optional[str]:
     return "{" + names[0] + "}" if len(names) == 1 else None
 
 
+_ID_SELECTOR = re.compile(r"^#([A-Za-z_][\w:.-]*)$")
+_NAME_SELECTOR = re.compile(r"""^(?:[a-z]+)?\[name=["']?([^"'\]]+)["']?\]$""")
+
+
+def _control_for(target: Any, controls: dict[str, Any]) -> Optional[str]:
+    """The form control a recorded click target is, by id, name or label."""
+
+    if not isinstance(target, dict):
+        return None
+    selector = str(target.get("selector") or "")
+    by_id = _ID_SELECTOR.match(selector)
+    if by_id:
+        hits = [n for n, c in controls.items() if c.get("id") == by_id.group(1)]
+        hits = hits or ([by_id.group(1)] if by_id.group(1) in controls else [])
+        return hits[0] if len(hits) == 1 else None
+    by_name = _NAME_SELECTOR.match(selector)
+    if by_name:
+        return by_name.group(1) if by_name.group(1) in controls else None
+    name = target.get("name")
+    return name if isinstance(name, str) and name in controls else None
+
+
+def _typed_fields(
+    steps: list[Step], events: list[dict[str, Any]], controls: dict[str, Any]
+) -> tuple[dict[str, Optional[str]], list[str]]:
+    """Which form field each folded step typed into, and with which parameter.
+
+    Provenance, not values: a TYPE step types into the field the click before
+    it targeted (the compiled anchor, else the recorded event's structural
+    target). Returns ``{field: param}`` (``None`` for literal text) and the
+    problems that make the binding ambiguous.
+    """
+
+    typed: dict[str, Optional[str]] = {}
+    problems: list[str] = []
+    current: Optional[str] = None
+    for step in steps:
+        if step.action in (ActionKind.CLICK, ActionKind.DOUBLE_CLICK):
+            target = None
+            if step.anchor is not None and step.anchor.structural is not None:
+                target = step.anchor.structural.model_dump()
+            else:
+                i = _event_index(step)
+                if i is not None and i < len(events):
+                    target = events[i].get("structural")
+            current = _control_for(target, controls)
+            continue
+        if step.action is not ActionKind.TYPE:
+            continue
+        param = step.param or _placeholder(step.text)
+        if current is None:
+            if param is not None:
+                problems.append(
+                    f"{step.id} types {{{param}}} but the field it types into "
+                    "isn't a control of the form"
+                )
+            continue
+        if current in typed and typed[current] != param:
+            problems.append(
+                f"{current!r} is typed twice with different values "
+                f"({step.id}); the binding is ambiguous"
+            )
+            continue
+        typed[current] = param
+    return typed, problems
+
+
+def _placeholder(text: Optional[str]) -> Optional[str]:
+    match = re.fullmatch(r"\{([A-Za-z_][A-Za-z0-9_]*)\}", text or "")
+    return match.group(1) if match else None
+
+
 def _success_pattern(line: dict[str, Any]) -> Optional[str]:
     """Where the accepted save sent the browser, with record numbers generalized."""
 
@@ -214,13 +286,74 @@ def carried(line: dict[str, Any]) -> int:
         )
     form_t = float(form_line["t"])
 
+    # Steps the API step stands in for: those recorded on the form page.
+    folded: list[Step] = []
+    for step in steps:
+        i = _event_index(step)
+        if i is not None and i < len(events) and float(events[i]["t"]) > form_t:
+            folded.append(step)
+    if not folded or folded[-1] is not save:
+        return (
+            workflow,
+            None,
+            ["the save is not part of the steps recorded on the form page"],
+        )
+
+    # Bind each field from what the steps typed into it, never from a value
+    # that merely matches: a blank or repeated demonstration value would bind
+    # the wrong field, or none (a later run's value silently dropped).
+    typed, problems = _typed_fields(folded, events, controls)
+    if problems:
+        return workflow, None, problems
+    sent = {name for name, _ in post["form"]}
+    for name, param in typed.items():
+        if param is not None and name not in sent:
+            return (
+                workflow,
+                None,
+                [f"{{{param}}} is typed into {name!r} but the save doesn't send it"],
+            )
+
     fields: list[WebFormField] = []
     for name, value in post["form"]:
         ctl = controls.get(name)
-        template = _templated(value, by_value) if value else None
-        if template is not None:
-            fields.append(WebFormField(name=name, source="param", param=template[1:-1]))
-            continue
+        if name in typed and typed[name] is not None:
+            param = typed[name]
+            demo = params.get(param, "")
+            if value == demo:
+                fields.append(WebFormField(name=name, source="param", param=param))
+                continue
+            label = next(
+                (
+                    o["label"]
+                    for o in ((ctl or {}).get("options") or [])
+                    if o["value"] == value
+                ),
+                None,
+            )
+            if label is not None and label == demo:
+                fields.append(
+                    WebFormField(name=name, source="select_label", param=param)
+                )
+                continue
+            return (
+                workflow,
+                None,
+                [
+                    f"field {name!r} is typed with {{{param}}} but the save sent "
+                    "another value"
+                ],
+            )
+        if value and value in by_value:
+            return (
+                workflow,
+                None,
+                [
+                    f"field {name!r} carries the value of "
+                    f"{', '.join('{' + n + '}' for n in by_value[value])} but no "
+                    "recorded step typed it there"
+                ],
+            )
         if (
             value.startswith("<session:")
             and ctl is not None
@@ -232,12 +365,16 @@ def carried(line: dict[str, Any]) -> int:
             label = next(
                 (o["label"] for o in ctl["options"] if o["value"] == value), None
             )
-            names = by_value.get(label or "") or []
-            if len(names) == 1:
-                fields.append(
-                    WebFormField(name=name, source="select_label", param=names[0])
+            if label is not None and label in by_value:
+                return (
+                    workflow,
+                    None,
+                    [
+                        f"field {name!r} shows the value of "
+                        f"{', '.join('{' + n + '}' for n in by_value[label])} but "
+                        "no recorded step chose it there"
+                    ],
                 )
-                continue
         if ctl is not None and (ctl.get("default") or "") == value:
             fields.append(WebFormField(name=name, source="form_default", value=value))
             continue
@@ -249,18 +386,6 @@ def carried(line: dict[str, Any]) -> int:
             ],
         )
 
-    # Steps the API step stands in for: those recorded on the form page.
-    folded: list[Step] = []
-    for step in steps:
-        i = _event_index(step)
-        if i is not None and i < len(events) and float(events[i]["t"]) > form_t:
-            folded.append(step)
-    if not folded or folded[-1] is not save:
-        return (
-            workflow,
-            None,
-            ["the save is not part of the steps recorded on the form page"],
-        )
     for step in folded:
         if step.guard is not None or step.wait_until is not None:
             return (

packages/flow/openadapt_flow/runtime/actuators/web_form.py

@@ -88,10 +88,12 @@ def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> N
             else:
                 value = a.get("value", "on" if kind in ("checkbox", "radio") else "")
             self._form["controls"].setdefault(
-                name, {"kind": kind, "default": value, "options": None}
+                name,
+                {"kind": kind, "default": value, "options": None, "id": a.get("id")},
             )
         elif tag == "select" and name:
             self._select = {
+                "id": a.get("id"),
                 "kind": "select",
                 "default": None,
                 "options": [],
@@ -106,7 +108,12 @@ def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> N
             }
             self._select["options"].append(self._option)
         elif tag == "textarea" and name:
-            self._textarea = {"kind": "textarea", "default": "", "options": None}
+            self._textarea = {
+                "kind": "textarea",
+                "default": "",
+                "options": None,
+                "id": a.get("id"),
+            }
             self._form["controls"].setdefault(name, self._textarea)
 
     def handle_data(self, data: str) -> None:

packages/flow/tests/test_api_proposal_provenance.py

@@ -0,0 +1,106 @@
+"""API step fields are bound by which step typed them, not by equal values.
+
+The reviewer's cases for #1269: a blank demonstration value bound as the
+form's default, so a later run's value was silently dropped; and equal
+demonstration values made the binding ambiguous. Synthetic values only.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from openadapt_flow.compiler.api_proposal import propose_api_step
+from openadapt_flow.runtime.actuators.web_form import WebFormActuator, form_controls
+from tests.test_api_tier_web_form import FORM_HTML, _recording, _workflow
+
+
+def _save(**fields):
+    form = {
+        "csrf_token_form": "<session:csrf_token_form>",
+        "form_lname": "Marsh",
+        "form_state": "MA",
+        "form_note": "Referred",
+        "form_extra": "",
+        "form_status": "active",
+    }
+    form.update(fields)
+    return [[k, v] for k, v in form.items()]
+
+
+def _with_params(rec, **params):
+    meta = json.loads((rec / "meta.json").read_text())
+    meta["params"].update(params)
+    (rec / "meta.json").write_text(json.dumps(meta))
+    return rec
+
+
+def _fields(step):
+    return {f.name: f for f in step.api_binding.web_form.fields}
+
+
+def test_a_blank_demonstration_value_is_still_the_parameter(tmp_path):
+    rec = _with_params(_recording(tmp_path, save_form=_save(form_note="")), note="")
+    wf = _workflow()
+    wf.params["note"] = ""
+    _wf, step, reasons = propose_api_step(wf, rec)
+    assert step is not None, reasons
+    note = _fields(step)["form_note"]
+    assert (note.source, note.param) == ("param", "note")
+    # A later run's note is sent, not dropped for the form's empty default.
+    controls = form_controls(
+        FORM_HTML.replace("<session:csrf_token_form>", "live"), "/app/save.php"
+    )
+    body, problem = WebFormActuator.build_body(
+        step.api_binding.web_form,
+        {"lname": "Marsh", "state_label": "Massachusetts", "note": "New note"},
+        controls,
+    )
+    assert problem is None and ("form_note", "New note") in body
+
+
+def test_equal_demonstration_values_bind_each_field_to_its_own_step(tmp_path):
+    rec = _with_params(
+        _recording(tmp_path, save_form=_save(form_note="Marsh")), note="Marsh"
+    )
+    wf = _workflow()
+    wf.params["note"] = "Marsh"
+    _wf, step, reasons = propose_api_step(wf, rec)
+    assert step is not None, reasons
+    fields = _fields(step)
+    assert fields["form_lname"].param == "lname"
+    assert fields["form_note"].param == "note"
+
+
+def test_a_field_carrying_a_value_no_step_typed_there_is_refused(tmp_path):
+    rec = _recording(tmp_path, save_form=_save() + [["form_copy", "Marsh"]])
+    _wf, step, reasons = propose_api_step(_workflow(), rec)
+    assert step is None
+    assert "no recorded step typed it there" in " ".join(reasons)
+
+
+def test_a_typed_field_sent_with_another_value_is_refused(tmp_path):
+    rec = _recording(tmp_path, save_form=_save(form_note="REFERRED"))
+    _wf, step, reasons = propose_api_step(_workflow(), rec)
+    assert step is None
+    assert "sent another value" in " ".join(reasons)
+
+
+def test_a_typed_field_the_save_does_not_send_is_refused(tmp_path):
+    form = [row for row in _save() if row[0] != "form_note"]
+    rec = _recording(tmp_path, save_form=form)
+    _wf, step, reasons = propose_api_step(_workflow(), rec)
+    assert step is None
+    assert "doesn't send it" in " ".join(reasons)
+
+
+@pytest.mark.parametrize("selector", ["#nowhere", None])
+def test_a_typed_step_whose_field_is_unknown_is_refused(tmp_path, selector):
+    rec = _recording(tmp_path)
+    events = [json.loads(x) for x in (rec / "events.jsonl").read_text().splitlines()]
+    events[4]["structural"] = {"selector": selector, "role": "textbox", "name": "?"}
+    (rec / "events.jsonl").write_text("".join(json.dumps(e) + "\n" for e in events))
+    _wf, step, reasons = propose_api_step(_workflow(), rec)
+    assert step is None
+    assert "isn't a control of the form" in " ".join(reasons)

packages/flow/tests/test_api_tier_web_form.py

@@ -201,8 +201,14 @@ def _recording(
     rec = tmp_path / "rec"
     rec.mkdir()
     (rec / "meta.json").write_text(json.dumps({"params": PARAMS}))
+    targets = {
+        0: {"selector": '[name="form_lname"]', "role": "textbox", "name": "Last"},
+        2: {"selector": '[name="form_state"]', "role": "combobox", "name": "State"},
+        4: {"selector": '[name="form_note"]', "role": "textbox", "name": "Note"},
+    }
     events = [
         {"i": i, "kind": k, "t": float(i + 1)}
+        | ({"structural": targets[i]} if i in targets else {})
         for i, k in enumerate(
             ["click", "type", "click", "type", "click", "type", "click", "click"]
         )
@@ -321,6 +327,7 @@ def test_proposal_explains_every_field_and_carries_the_duplicate_check(tmp_path)
             {
                 "save_form": [
                     ["form_lname", "Marsh"],
+                    ["form_state", "MA"],
                     ["form_note", "Referred"],
                     ["mystery", "x9"],
                 ]
@@ -331,6 +338,7 @@ def test_proposal_explains_every_field_and_carries_the_duplicate_check(tmp_path)
             {
                 "save_form": [
                     ["form_lname", "Marsh"],
+                    ["form_state", "MA"],
                     ["form_note", "Referred"],
                     ["pin", nc.SECRET],
                 ]