abrichr · Oct 19, 2026 · Oct 19, 2026
@@ -60,6 +60,14 @@ 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.
 
+The save's URL and the form page's URL keep their query. A path segment or
+query value that equals exactly one input's demonstration value (a record
+identifier such as `record_id=17`) becomes that parameter; the rest is kept
+as recorded. A hidden form field that echoes such an identifier (an update
+form's `pid`) is the same parameter. The form page, the pre-checks and the
+save must share one origin, and the API step only ever sends to the live
+form's own origin.
+
 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
@@ -71,6 +79,8 @@ doesn't propose an API step in these cases:
 - the form page wasn't captured;
 - the save isn't the program's last step;
 - the program has branches;
+- a URL value equals two inputs' values (ambiguous), or the save or a
+  pre-check is on another origin than the form;
 - a covered step has a guard, a wait rule, or a record-identifying region the
   person marked with `record --identifier`;
 - a covered step made a request whose response must show text the GUI checks

@@ -95,6 +95,54 @@ def _templated(value: str, by_value: dict[str, list[str]]) -> Optional[str]:
     return "{" + names[0] + "}" if len(names) == 1 else None
 
 
+def _origin(url: str) -> str:
+    parts = up.urlsplit(url)
+    return f"{parts.scheme.lower()}://{parts.netloc.lower()}"
+
+
+def _url_template(url: str, by_value: dict[str, list[str]]) -> tuple[str, list[str]]:
+    """``url``'s path and query, with input values as ``{param}``.
+
+    A path segment or query value that equals exactly one input's
+    demonstration value becomes that parameter (a record identifier such as
+    ``record_id=17``); the rest is kept as recorded. A value equal to two
+    inputs is ambiguous and reported.
+    """
+
+    parts = up.urlsplit(url)
+    problems: list[str] = []
+
+    def bind(value: str, where: str) -> str:
+        names = by_value.get(value) or []
+        if len(names) > 1:
+            problems.append(
+                f"{where} {value!r} equals {', '.join('{' + n + '}' for n in names)}; "
+                "the binding is ambiguous"
+            )
+            return value
+        return "{" + names[0] + "}" if names else value
+
+    segments = [
+        bind(up.unquote(seg), "the URL path segment") if seg else seg
+        for seg in parts.path.split("/")
+    ]
+    path = "/".join(
+        seg if seg.startswith("{") else up.quote(seg, safe=":@!$&'()*+,;=-._~")
+        for seg in segments
+    )
+    query = [
+        (k, bind(v, f"the URL query value {k}=") if v else v)
+        for k, v in up.parse_qsl(parts.query, keep_blank_values=True)
+    ]
+    if query:
+        # Placeholders stay readable; fill_url encodes the run's values.
+        path += "?" + "&".join(
+            up.quote(k) + "=" + (v if v.startswith("{") else up.quote(v))
+            for k, v in query
+        )
+    return path, problems
+
+
 _ID_SELECTOR = re.compile(r"^#([A-Za-z_][\w:.-]*)$")
 _NAME_SELECTOR = re.compile(r"""^(?:[a-z]+)?\[name=["']?([^"'\]]+)["']?\]$""")
 
@@ -259,7 +307,9 @@ def carried(line: dict[str, Any]) -> int:
     post = max(posts, key=carried)
     if carried(post) < max(1, len(by_value) // 2):
         return workflow, None, ["the save's request does not carry the recorded values"]
-    action = _path(post["url"])
+    action, url_problems = _url_template(post["url"], by_value)
+    if url_problems:
+        return workflow, None, url_problems
     if any(v == nc.SECRET for _, v in post["form"]):
         return workflow, None, ["the save sends a secret field; it stays on the GUI"]
 
@@ -273,7 +323,7 @@ def carried(line: dict[str, Any]) -> int:
             and ln.get("status") == 200
             and "<form" in (ln.get("response_text") or "")
         ):
-            found = form_controls(ln["response_text"], action)
+            found = form_controls(ln["response_text"], _path(post["url"]))
             if found is not None:
                 form_line, controls = ln, found
     if form_line is None or controls is None:
@@ -285,6 +335,18 @@ def carried(line: dict[str, Any]) -> int:
             ],
         )
     form_t = float(form_line["t"])
+    if _origin(form_line["url"]) != _origin(post["url"]):
+        return (
+            workflow,
+            None,
+            [
+                "the save posts to another origin than the page that served the "
+                "form; the API step sends only to the form's own origin"
+            ],
+        )
+    form_path, url_problems = _url_template(form_line["url"], by_value)
+    if url_problems:
+        return workflow, None, url_problems
 
     # Steps the API step stands in for: those recorded on the form page.
     folded: list[Step] = []
@@ -344,6 +406,19 @@ def carried(line: dict[str, Any]) -> int:
                     "another value"
                 ],
             )
+        url_bound = [
+            n for n in by_value.get(value, []) if "{" + n + "}" in action + form_path
+        ]
+        if (
+            value
+            and len(by_value.get(value, [])) == 1
+            and url_bound
+            and (ctl or {}).get("kind") == "hidden"
+        ):
+            # The record's identifier, echoed by the form for the record the
+            # URL names (an update form's hidden pid): the same parameter.
+            fields.append(WebFormField(name=name, source="param", param=url_bound[0]))
+            continue
         if value and value in by_value:
             return (
                 workflow,
@@ -456,6 +531,12 @@ def carried(line: dict[str, Any]) -> int:
                     f"{step.id} checks {texts!r} on screen and no response it caused carries it"
                 ],
             )
+        if _origin(hit["url"]) != _origin(form_line["url"]):
+            return (
+                workflow,
+                None,
+                [f"the check at {step.id} reads another origin than the form's"],
+            )
         query = []
         for k, v in up.parse_qsl(up.urlsplit(hit["url"]).query, keep_blank_values=True):
             if v.startswith("<session:"):
@@ -484,12 +565,7 @@ def carried(line: dict[str, Any]) -> int:
         }
     )
     spec = WebFormSpec(
-        form_path=_path(form_line["url"])
-        + (
-            f"?{up.urlsplit(form_line['url']).query}"
-            if up.urlsplit(form_line["url"]).query
-            else ""
-        ),
+        form_path=form_path,
         fields=fields,
         controls=sorted(controls),
         client_rules=client_rules(form_line["response_text"]),

@@ -425,13 +425,16 @@ def unavailable(reason: str) -> ApiActuationResult:
         if spec is None:
             return unavailable("the binding has no recorded form")
         try:
-            page = self.transport.get(spec.form_path)
+            # Record identifiers in the URLs are run parameters (fill_url).
+            form_path = fill_url(spec.form_path, params)
+            save_path = fill_url(binding.url_template, params)
+            page = self.transport.get(form_path)
             self.log.append({"stage": "read_form", "status": page.status})
             if page.status != 200:
                 return unavailable(
                     f"the form page returned HTTP {page.status} (signed out?)"
                 )
-            controls = form_controls(page.text, binding.url_template)
+            controls = form_controls(page.text, up.urlsplit(save_path).path)
             if controls is None:
                 return unavailable("the form page no longer has the recorded form")
             if spec.controls and sorted(controls) != sorted(spec.controls):
@@ -473,9 +476,9 @@ def unavailable(reason: str) -> ApiActuationResult:
             return unavailable(f"a read before the save failed ({type(exc).__name__})")
         return PreparedSave(
             method=binding.method,
-            path=binding.url_template,
+            path=save_path,
             body=tuple(body),
-            referer=self.transport.base_url + spec.form_path,
+            referer=self.transport.base_url + form_path,
             session_fields=frozenset(
                 f.name for f in spec.fields if f.source == "session"
             ),
@@ -584,6 +587,34 @@ class _Missing(KeyError):
     pass
 
 
+def fill_url(template: str, params: dict[str, str]) -> str:
+    """Fill a URL template's ``{param}`` placeholders, URL-encoded.
+
+    A path segment is quoted whole (``/`` included); a query value is
+    form-encoded. Anything not a placeholder is kept as recorded. Raises
+    ``KeyError`` for a parameter the run doesn't have.
+    """
+
+    path, sep, query = template.partition("?")
+    path = re.sub(
+        r"\{([A-Za-z_][A-Za-z0-9_]*)\}",
+        lambda m: up.quote(str(_value(params, m.group(1))), safe=""),
+        path,
+    )
+    if not sep:
+        return path
+    pairs = [
+        (k, _fill(v, params)) for k, v in up.parse_qsl(query, keep_blank_values=True)
+    ]
+    return path + "?" + up.urlencode(pairs)
+
+
+def _value(params: dict[str, str], key: str) -> str:
+    if key not in params:
+        raise _Missing(key)
+    return str(params[key])
+
+
 def _fill(template: str, params: dict[str, str]) -> str:
     def sub(m: re.Match[str]) -> str:
         key = m.group(1)

@@ -0,0 +1,139 @@
+"""Record identifiers in an API step's URLs are run parameters.
+
+The reviewer's case for #1269: a save to /save.php?record_id=17 compiled to
+/save.php (the query lost), while /new.php?record_id=17 stayed literal in the
+form-fetch URL, so a later run read and wrote record 17. Synthetic values only.
+"""
+
+from __future__ import annotations
+
+import json
+
+from openadapt_flow import network_capture as nc
+from openadapt_flow.compiler.api_proposal import propose_api_step
+from openadapt_flow.runtime.actuators import ActuationStatus
+from openadapt_flow.runtime.actuators.web_form import WebFormActuator, fill_url
+from tests.test_api_tier_web_form import (
+    FORM_HTML,
+    FakeTransport,
+    _recording,
+    _workflow,
+)
+
+
+def _with_record_id(rec, *, save_url, form_url, record_id="17", extra=None):
+    meta = json.loads((rec / "meta.json").read_text())
+    meta["params"].update({"record_id": record_id, **(extra or {})})
+    (rec / "meta.json").write_text(json.dumps(meta))
+    rows = [json.loads(x) for x in (rec / nc.NETWORK_FILE).read_text().splitlines()]
+    for row in rows[1:]:
+        if row.get("method") == "POST":
+            row["url"] = save_url
+        elif "<form" in (row.get("response_text") or ""):
+            row["url"] = form_url
+    (rec / nc.NETWORK_FILE).write_text("".join(json.dumps(r) + "\n" for r in rows))
+    return rec
+
+
+def _propose(tmp_path, **kw):
+    rec = _with_record_id(_recording(tmp_path), **kw)
+    wf = _workflow()
+    wf.params["record_id"] = kw.get("record_id", "17")
+    return propose_api_step(wf, rec)
+
+
+def test_a_record_id_in_the_save_and_form_urls_becomes_a_parameter(tmp_path):
+    _wf, step, reasons = _propose(
+        tmp_path,
+        save_url="http://app/app/save.php?record_id=17&mode=edit",
+        form_url="http://app/app/new.php?record_id=17",
+    )
+    assert step is not None, reasons
+    binding = step.api_binding
+    assert binding.url_template == "/app/save.php?record_id={record_id}&mode=edit"
+    assert binding.web_form.form_path == "/app/new.php?record_id={record_id}"
+
+
+class _RecordTransport(FakeTransport):
+    def get(self, path):
+        self.gets.append(path)
+        page = self.form if path.startswith("/app/new.php") else self.dup
+        return type("R", (), {"status": 200, "text": page, "headers": {}})()
+
+
+def test_a_later_run_reads_and_writes_its_own_record(tmp_path):
+    _wf, step, reasons = _propose(
+        tmp_path,
+        save_url="http://app/app/save.php?record_id=17",
+        form_url="http://app/app/new.php?record_id=17",
+    )
+    assert step is not None, reasons
+    transport = _RecordTransport(
+        form=FORM_HTML.replace("<session:csrf_token_form>", "live")
+    )
+    out = WebFormActuator(transport).actuate(
+        step.api_binding,
+        {
+            "lname": "Marsh",
+            "state_label": "Massachusetts",
+            "note": "Referred",
+            "record_id": "42",
+        },
+    )
+    assert out.status is ActuationStatus.ACTUATED, out.reason
+    assert transport.gets[0] == "/app/new.php?record_id=42"
+    ((path, _body),) = transport.posts
+    assert path == "/app/save.php?record_id=42"
+
+
+def test_an_ambiguous_url_value_is_refused(tmp_path):
+    _wf, step, reasons = _propose(
+        tmp_path,
+        save_url="http://app/app/save.php?record_id=17",
+        form_url="http://app/app/new.php?record_id=17",
+        extra={"visit_id": "17"},
+    )
+    assert step is None
+    assert "ambiguous" in " ".join(reasons)
+
+
+def test_a_save_to_another_origin_is_refused(tmp_path):
+    _wf, step, reasons = _propose(
+        tmp_path,
+        save_url="https://elsewhere.example/app/save.php",
+        form_url="http://app/app/new.php",
+    )
+    assert step is None
+    assert "another origin" in " ".join(reasons)
+
+
+def test_fill_url_encodes_values_and_keeps_the_rest():
+    assert fill_url("/p/{id}/save?x={q}&mode=edit", {"id": "a/b", "q": "x y&z"}) == (
+        "/p/a%2Fb/save?x=x+y%26z&mode=edit"
+    )
+    assert fill_url("/plain.php", {}) == "/plain.php"
+
+
+def test_a_hidden_field_echoing_the_url_record_id_is_the_same_parameter(tmp_path):
+    rec = _recording(tmp_path)
+    rows = [json.loads(x) for x in (rec / nc.NETWORK_FILE).read_text().splitlines()]
+    for row in rows[1:]:
+        if "<form" in (row.get("response_text") or ""):
+            row["response_text"] = row["response_text"].replace(
+                '<input name="form_note"',
+                '<input type="hidden" name="pid" value="17"><input name="form_note"',
+            )
+        if row.get("method") == "POST":
+            row["form"].append(["pid", "17"])
+    (rec / nc.NETWORK_FILE).write_text("".join(json.dumps(r) + "\n" for r in rows))
+    rec = _with_record_id(
+        rec,
+        save_url="http://app/app/save.php?record_id=17",
+        form_url="http://app/app/new.php?record_id=17",
+    )
+    wf = _workflow()
+    wf.params["record_id"] = "17"
+    _wf, step, reasons = propose_api_step(wf, rec)
+    assert step is not None, reasons
+    pid = {f.name: f for f in step.api_binding.web_form.fields}["pid"]
+    assert (pid.source, pid.param) == ("param", "record_id")