From c1c9cb39f0cc1f61eca1635efebb36e9b6a4b33f Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Mon, 13 Jul 2026 10:30:40 -0400 Subject: [PATCH] feat: interactive `record --url` + secret-typed parameters (never persisted) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the #1 adoption gap: the README promised "record a GUI workflow once" but the only recorder (`demo-record`) ran the hard-coded MockMed script. There was no way to record your OWN app. record --url: - New `openadapt-flow record --url ` opens a headed browser on the user's own app and watches real clicks/typing/keys/scrolls via in-page capture-phase DOM listeners (installed with add_init_script so they survive navigations), writing the EXACT recording format `compile` already consumes. Stop with Ctrl-C or by closing the window. record -> compile -> replay now closes the self-serve loop for any app, not just the bundled demo. - Architecture: the expose_binding callback only appends raw events to a Python list (calling any page method inside a sync-API binding callback deadlocks the driver); the main loop drains it and does all screenshotting. Each step's before-frame is the previous step's settled frame (no post-navigation race); type/scroll runs capture their after-frame+structural state at the moment they happen so a following navigating click can't corrupt them. Structured DOM identity is captured in-page at click time, arming the identity ladder on interactively-recorded bundles. Reuses the existing Recorder via a new `record_observed` seam — the recording format is not forked. Secret-typed parameters: - input[type=password] is auto-detected as secret; any field can be marked with `--secret `. A secret's literal value is NEVER read into Python, never written to meta.json / events.jsonl / the compiled bundle, and its field region is redacted (solid black) from the persisted before/after frames. - At replay the value is injected from OPENADAPT_FLOW_SECRET_; a missing secret fails fast with an actionable message naming the env var. - Schema: ir.Step.secret + Workflow.secret_params; compiler carries the secret through with text=None; replayer resolves it from the environment. Tests: tests/test_secret_params.py (fast unit: recorder redaction/non-persist, compiler carry-through, replayer env injection + missing-secret error) and tests/test_interactive_recorder.py (headless scripted record -> compile -> replay proving the loop, no secret leak in any artifact, frame redaction, and env injection). Full suite: 962 passed, 9 skipped. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CKrVJJy5jWVCkXAqgUqtqZ --- README.md | 31 +- openadapt_flow/__main__.py | 62 +++ openadapt_flow/compiler/compile.py | 15 +- openadapt_flow/interactive_recorder.py | 533 +++++++++++++++++++++++++ openadapt_flow/ir.py | 20 + openadapt_flow/recorder.py | 121 +++++- openadapt_flow/runtime/replayer.py | 31 +- tests/test_interactive_recorder.py | 213 ++++++++++ tests/test_secret_params.py | 222 ++++++++++ 9 files changed, 1238 insertions(+), 10 deletions(-) create mode 100644 openadapt_flow/interactive_recorder.py create mode 100644 tests/test_interactive_recorder.py create mode 100644 tests/test_secret_params.py diff --git a/README.md b/README.md index 78b131c..e392dd5 100644 --- a/README.md +++ b/README.md @@ -33,8 +33,35 @@ openadapt-flow replay bundle --drift theme # drift the UI, watch i ``` The last two commands serve the bundled MockMed demo app and write an -illustrated `REPORT.md` per run. Pass `--url` to replay against your own app; -recorded parameter values are the defaults and `--param` overrides them. +illustrated `REPORT.md` per run. + +### Record your own app + +`record --url` opens a headed browser on YOUR app and watches what you do — +real clicks, typing, key presses and scrolls — writing the same recording +format `compile` consumes. Perform the workflow, then press Ctrl-C (or close +the window) to finish: + +```bash +openadapt-flow record --url https://your.app --out rec # do the task, Ctrl-C +openadapt-flow compile rec --out bundle --name my-task +openadapt-flow replay bundle --url https://your.app # replay it +``` + +Pass `--url` to `replay` to run against your own app; recorded parameter values +are the defaults and `--param` overrides them. + +**Secrets never get recorded.** A `input[type=password]` field (or any field +named with `--secret `) is a secret parameter: its value is never written +to the recording, the events log, the compiled bundle, or the saved frames (its +region is redacted). At replay it is injected from the environment and a missing +one fails fast: + +```bash +openadapt-flow record --url https://your.app --out rec --secret password +export OPENADAPT_FLOW_SECRET_PASSWORD='…' # supplied at replay +openadapt-flow replay bundle --url https://your.app +``` ## How it works diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index 3b49e89..3086c14 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -3,6 +3,8 @@ Subcommands (thin wrappers over the module APIs; sibling modules are imported lazily inside each handler so ``--help`` always works): +- ``record`` — open a headed browser on your OWN app (``--url``) and + record what you do into the format ``compile`` consumes. - ``demo-record`` — serve MockMed locally and record the canonical demo. - ``compile`` — compile a recording directory into a workflow bundle. - ``replay`` — replay a bundle; serves the bundled MockMed demo app when no @@ -40,6 +42,28 @@ def _with_drift(url: str, drift: str | None) -> str: return f"{url.rstrip('/')}/?drift={drift}" +def _cmd_record(args: argparse.Namespace) -> int: + from openadapt_flow.interactive_recorder import record_interactive + + out = record_interactive( + args.url, + Path(args.out), + secret_fields=tuple(args.secret or ()), + param_fields=tuple(args.param or ()), + headless=args.headless, + ) + print(f"Recording written to {out}") + secrets = sorted(args.secret or ()) + if secrets: + print( + "Secret field(s) recorded (values NOT stored): " + + ", ".join(secrets) + + ". At replay, export " + + ", ".join(f"OPENADAPT_FLOW_SECRET_{name.upper()}" for name in secrets) + ) + return 0 + + def _cmd_demo_record(args: argparse.Namespace) -> int: from openadapt_flow.demo_driver import record_triage_demo from openadapt_flow.mockmed.server import serve @@ -274,6 +298,44 @@ def build_parser() -> argparse.ArgumentParser: ) sub = parser.add_subparsers(dest="command", required=True) + p = sub.add_parser( + "record", + help="Record YOUR app interactively in a headed browser (--url)", + ) + p.add_argument( + "--url", required=True, help="URL of the app to record against" + ) + p.add_argument("--out", required=True, help="Recording output directory") + p.add_argument( + "--secret", + action="append", + default=[], + metavar="FIELD", + help=( + "Mark a typed field (by name or id) as a SECRET; its value is " + "never persisted and is injected at replay from " + "OPENADAPT_FLOW_SECRET_. input[type=password] is always " + "treated as secret. Repeatable." + ), + ) + p.add_argument( + "--param", + action="append", + default=[], + metavar="FIELD", + help=( + "Record a typed field (by name or id) as a PARAMETER; its " + "demonstrated value becomes the default, overridable at replay " + "with --param. Repeatable." + ), + ) + p.add_argument( + "--headless", + action="store_true", + help="Run the browser headless (scripted/CI recording)", + ) + p.set_defaults(func=_cmd_record) + p = sub.add_parser( "demo-record", help="Serve MockMed and record the canonical triage demo", diff --git a/openadapt_flow/compiler/compile.py b/openadapt_flow/compiler/compile.py index ef59ae5..89f14fe 100644 --- a/openadapt_flow/compiler/compile.py +++ b/openadapt_flow/compiler/compile.py @@ -923,7 +923,15 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]: elif kind == "type": param = event.get("param") text = event.get("text") - if param: + secret = bool(event.get("secret")) + if secret: + # A secret's literal value is never in the recording, so it + # is never in the bundle either: the step carries only the + # param name, and the value is injected from the environment + # at replay (see ir.Step.secret / runtime.Replayer). + text = None + intent = f"type <{param}> (secret)" + elif param: intent = f"type <{param}>" else: intent = f"type '{_text_preview(text or '')}'" @@ -935,6 +943,7 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]: action=ActionKind.TYPE, text=text, param=param, + secret=secret, ), before_png, after_png, @@ -1017,7 +1026,8 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]: # A parameterized TYPE step's changed region is the typed # value's own pixels — never assert it (it varies per run). include_region_stable=not ( - step.action is ActionKind.TYPE and step.param is not None + step.action is ActionKind.TYPE + and (step.param is not None or step.secret) ), before_lines=( cached_lines(i, "before", step_before) @@ -1063,6 +1073,7 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]: recording_id=meta.get("id"), viewport=tuple(viewport) if viewport else None, params=params, + secret_params=list(meta.get("secret_params") or []), steps=steps, ) diff --git a/openadapt_flow/interactive_recorder.py b/openadapt_flow/interactive_recorder.py new file mode 100644 index 0000000..60697bd --- /dev/null +++ b/openadapt_flow/interactive_recorder.py @@ -0,0 +1,533 @@ +"""Interactive recorder: capture a demonstration the USER drives live. + +``openadapt-flow record --url `` opens a real (headed) Playwright browser +pointed at the user's OWN app and simply watches: it listens to the user's +real clicks, typing, key presses and scrolls via in-page capture-phase DOM +listeners (the same technique ``playwright codegen`` uses) and writes the +EXACT recording format the compiler already consumes (``meta.json`` + +``events.jsonl`` + ``frames/{i:04d}_before.png`` / ``_after.png``). + + record --url … → compile → replay + +closes the self-serve loop for the user's own app, not just the bundled demo. + +Design (why it looks the way it does): + +* **Append-only binding, work in the loop.** Calling any Playwright page + method from inside an ``expose_binding`` callback deadlocks the sync driver, + so the binding callback does the ONE cheap thing it safely can — append the + raw event to a Python list — and the main loop drains that list and does all + the screenshotting/settling. Listeners are installed with ``add_init_script`` + so they survive navigations, and a navigating click's event is delivered + over the pipe before the new document loads. +* **Frames chain like a driven demo.** A demonstration's screen is static + between actions, so each step's BEFORE frame is simply the previous step's + settled frame (captured before the current action happened — no post- + navigation race), and its AFTER frame is captured once the screen settles. +* **Structured identity is captured in-page** at click time (pre-navigation), + mirroring ``PlaywrightBackend.structured_text_at`` exactly, so the compiler's + DOM-identity tier arms on interactively-recorded bundles too. + +Secrets never touch Python: a field is secret when it is ``input[type= +password]`` or its name/id is passed via ``--secret``. For a secret field the +in-page listener emits NO value at all (only that a secret was typed, plus the +field rectangle for redaction); the literal is never read, never sent over the +pipe, never written to meta/events/frames/bundle. See ``ir.Step.secret`` and +``docs`` for the full contract. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Callable, Optional + +from openadapt_flow.backends.playwright_backend import PlaywrightBackend +from openadapt_flow.recorder import Recorder + +# Named keys worth recording as their own KEY step (navigation/submit intent). +# Editing keys (Backspace/Delete) are intentionally omitted: their effect is +# already reflected in the field's value, read via the ``input`` event. +_SPECIAL_KEYS = ( + "Enter", + "Tab", + "Escape", + "ArrowUp", + "ArrowDown", + "ArrowLeft", + "ArrowRight", + "PageUp", + "PageDown", + "Home", + "End", +) + +# In-page recorder script. Installed via add_init_script so it re-arms on every +# document (navigations). Emits raw events to the Python side via the +# __oaflow_emit binding. __SECRET_NAMES__ / __SPECIAL_KEYS__ are substituted in. +_INIT_JS = r""" +(() => { + if (window.__oaflowInstalled) return; + window.__oaflowInstalled = true; + const SECRET_NAMES = __SECRET_NAMES__; + const SPECIAL = __SPECIAL_KEYS__; + + function structuredIdentity(px, py) { + // Mirrors PlaywrightBackend.structured_text_at: the REAL characters of the + // clicked row (MRN/name/DOB), excluding the clicked target's own cell. + try { + const el = document.elementFromPoint(px, py); + if (!el) return null; + const row = el.closest('tr, [role="row"], li, [role="listitem"]'); + if (!row) return null; + const own = el.closest('td, th, [role="cell"], [role="gridcell"]') || el; + own.setAttribute('data-oaflow-own', '1'); + let body = ''; + try { + const clone = row.cloneNode(true); + const marked = clone.querySelector('[data-oaflow-own="1"]'); + if (marked) marked.remove(); + body = clone.textContent || ''; + } finally { + own.removeAttribute('data-oaflow-own'); + } + const parts = []; + const aria = row.getAttribute ? row.getAttribute('aria-label') : null; + if (aria) parts.push(aria); + if (body) parts.push(body); + const joined = parts.join(' ').replace(/\s+/g, ' ').trim(); + return joined || null; + } catch (e) { return null; } + } + + function isSecretEl(el) { + if (!el) return false; + if ((el.type || '').toLowerCase() === 'password') return true; + const n = el.name || '', i = el.id || ''; + return SECRET_NAMES.indexOf(n) >= 0 || SECRET_NAMES.indexOf(i) >= 0; + } + + function emit(o) { try { window.__oaflow_emit(o); } catch (e) {} } + + document.addEventListener('click', (e) => { + emit({ + kind: 'click', + x: Math.round(e.clientX), y: Math.round(e.clientY), + sid: structuredIdentity(e.clientX, e.clientY), + url: location.href, title: document.title, + }); + }, true); + + document.addEventListener('input', (e) => { + const el = e.target; + const secret = isSecretEl(el); + const r = (el.getBoundingClientRect && el.getBoundingClientRect()) + || { left: 0, top: 0, width: 0, height: 0 }; + const o = { + kind: 'input', + field: el.name || el.id || null, + secret: secret, + rect: [Math.round(r.left), Math.round(r.top), + Math.round(r.width), Math.round(r.height)], + url: location.href, title: document.title, + }; + // The literal value of a SECRET field is never read or transmitted. + if (!secret) o.value = (el.value != null ? String(el.value) : ''); + emit(o); + }, true); + + document.addEventListener('keydown', (e) => { + if (SPECIAL.indexOf(e.key) < 0) return; + emit({ kind: 'key', key: e.key, url: location.href, title: document.title }); + }, true); + + document.addEventListener('wheel', (e) => { + emit({ + kind: 'scroll', + dx: Math.round(e.deltaX), dy: Math.round(e.deltaY), + url: location.href, title: document.title, + }); + }, true); +})(); +""" + + +class InteractiveRecorder: + """Drives a live headed browser and records what the user does. + + Use :func:`record_interactive` for the common case; this class is exposed + for tests, which drive synthetic input via :attr:`page` and pump the loop + deterministically. + """ + + def __init__( + self, + url: str, + out_dir: Path | str, + *, + secret_fields: tuple[str, ...] = (), + param_fields: tuple[str, ...] = (), + headless: bool = False, + poll_ms: int = 60, + settle_timeout_s: float = 5.0, + settle_stable_frames: int = 2, + settle_interval_s: float = 0.15, + viewport: tuple[int, int] = (1280, 800), + ) -> None: + self._url = url + self._out_dir = Path(out_dir) + self._secret_fields = set(secret_fields) + self._param_fields = set(param_fields) + self._headless = headless + self._poll_ms = poll_ms + self._viewport = viewport + self._settle = dict( + settle_timeout_s=settle_timeout_s, + settle_stable_frames=settle_stable_frames, + settle_interval_s=settle_interval_s, + ) + self._pyq: list[dict[str, Any]] = [] + self._pending_type: Optional[dict[str, Any]] = None + self._pending_scroll: Optional[dict[str, Any]] = None + self.done = False + + # Set on start(). + self._pw = None + self._browser = None + self.page = None + self.backend: Optional[PlaywrightBackend] = None + self.recorder: Optional[Recorder] = None + self._last_frame: bytes = b"" + self._last_structural: dict[str, Any] = {} + + # -- lifecycle ----------------------------------------------------------- + + def start(self) -> None: + """Launch the browser, install the in-page listeners, capture the + initial settled frame.""" + from playwright.sync_api import sync_playwright + + self._pw = sync_playwright().start() + try: + self._browser = self._pw.chromium.launch(headless=self._headless) + except Exception: + self._pw.stop() + raise + self.page = self._browser.new_page( + viewport={"width": self._viewport[0], "height": self._viewport[1]}, + device_scale_factor=1, + ) + self.page.on("close", lambda _=None: setattr(self, "done", True)) + self.page.expose_binding( + "__oaflow_emit", + lambda source, detail: self._pyq.append(detail), + ) + init_js = _INIT_JS.replace( + "__SECRET_NAMES__", json.dumps(sorted(self._secret_fields)) + ).replace("__SPECIAL_KEYS__", json.dumps(list(_SPECIAL_KEYS))) + self.page.add_init_script(init_js) + self.page.goto(self._url) + try: + self.page.wait_for_load_state("load") + except Exception: + pass + self.backend = PlaywrightBackend(self.page) + self.recorder = Recorder( + self.backend, self._out_dir, app_url=self._url, **self._settle + ) + self._last_frame = self.recorder._wait_settled() + self._last_structural = self._structural_state() + + def run(self) -> Path: + """Human loop: pump until the user stops (Ctrl-C / closes the window), + then flush and finish.""" + print( + f"Recording {self._url}\n" + " Perform your workflow in the browser window.\n" + " Press Ctrl-C here (or close the browser window) to finish." + ) + try: + while not self.done: + if not self._pump(): + break + except KeyboardInterrupt: + print("\n[record] stopping…") + return self.finish() + + def run_script(self, script: Callable[[Any, Callable[[], None]], None]) -> Path: + """Scripted loop (tests): run ``script(page, pump)`` — which performs + synthetic input and calls ``pump()`` to let the recorder drain — then + flush and finish.""" + script(self.page, self.pump) + return self.finish() + + def finish(self) -> Path: + """Flush trailing input, write meta.json, tear the browser down.""" + try: + self._flush_type() + self._flush_scroll() + finally: + assert self.recorder is not None + out = self.recorder.finish() + try: + if self._browser is not None: + self._browser.close() + finally: + if self._pw is not None: + self._pw.stop() + return out + + # -- event pump ---------------------------------------------------------- + + def pump(self) -> bool: + """One public pump tick (used by scripted tests). Returns False when + the page/browser is gone.""" + return self._pump() + + def _pump(self) -> bool: + try: + self.page.wait_for_timeout(self._poll_ms) + except Exception: + self.done = True + return False + batch = self._pyq[:] + del self._pyq[:] + if not batch: + # Distinct scroll gestures are separated by pauses; flush a + # completed scroll on idle so each becomes its own step. A type run + # is NOT idle-flushed (a mid-word pause must not split it) — it + # flushes on the next boundary event or at finish(). + self._flush_scroll() + return True + for ev in batch: + self._process(ev) + return True + + def _process(self, ev: dict[str, Any]) -> None: + kind = ev.get("kind") + if kind == "input": + self._flush_scroll() + self._accumulate_input(ev) + elif kind == "scroll": + self._flush_type() + self._accumulate_scroll(ev) + elif kind == "click": + self._flush_type() + self._flush_scroll() + self._record_click(ev) + elif kind == "key": + self._flush_type() + self._flush_scroll() + self._record_key(ev) + + # -- accumulation / flush ------------------------------------------------ + + def _accumulate_input(self, ev: dict[str, Any]) -> None: + field = ev.get("field") + if ( + self._pending_type is not None + and self._pending_type.get("field") != field + ): + self._flush_type() # focus moved to a different field + if self._pending_type is None: + self._pending_type = { + "field": field, + "secret": bool(ev.get("secret")), + "value": "", + "rect": ev.get("rect"), + } + pt = self._pending_type + pt["secret"] = pt["secret"] or bool(ev.get("secret")) + if ev.get("rect"): + pt["rect"] = ev["rect"] + if not pt["secret"]: + pt["value"] = ev.get("value", pt["value"]) + # The structural context for the whole run is its FIRST input's frame. + pt.setdefault("structural_before", dict(self._last_structural)) + # Capture the field-with-text after-frame NOW, while the typed value is + # on screen and BEFORE any following navigating action executes. In a + # human recording the pump cadence reaches here between the last + # keystroke and the next click, so this frame is the settled field — + # not a screen the next click has already navigated to. + assert self.backend is not None + pt["after_frame"] = self.backend.screenshot() + pt["structural_after"] = self._structural_state() + + def _flush_type(self) -> None: + pt = self._pending_type + self._pending_type = None + if pt is None: + return + field = pt.get("field") + structural_before = pt.get("structural_before", self._last_structural) + assert self.recorder is not None + after_png = pt.get("after_frame") + structural_after = pt.get("structural_after") + if pt["secret"]: + rect = pt.get("rect") or None + redact = tuple(rect) if rect and rect[2] and rect[3] else None + self.recorder.record_observed( + {"kind": "type"}, + before_png=self._last_frame, + structural_before=structural_before, + param=field or "secret", + secret=True, + redact_region=redact, + after_png=after_png, + structural_after=structural_after, + ) + elif field and field in self._param_fields: + self.recorder.record_observed( + {"kind": "type", "text": pt["value"]}, + before_png=self._last_frame, + structural_before=structural_before, + param=field, + after_png=after_png, + structural_after=structural_after, + ) + else: + # Non-secret, unparameterized: recorded as a literal (replayed + # verbatim), matching the demo driver's username/note handling. + self.recorder.record_observed( + {"kind": "type", "text": pt["value"]}, + before_png=self._last_frame, + structural_before=structural_before, + after_png=after_png, + structural_after=structural_after, + ) + self._set_last(after_png, structural_after) + + def _accumulate_scroll(self, ev: dict[str, Any]) -> None: + if self._pending_scroll is None: + self._pending_scroll = { + "dx": 0, + "dy": 0, + "structural_before": dict(self._last_structural), + } + ps = self._pending_scroll + ps["dx"] += int(ev.get("dx", 0)) + ps["dy"] += int(ev.get("dy", 0)) + # Post-scroll after-state, captured now (before any following action). + assert self.backend is not None + ps["after_frame"] = self.backend.screenshot() + ps["structural_after"] = self._structural_state() + + def _flush_scroll(self) -> None: + ps = self._pending_scroll + self._pending_scroll = None + if ps is None or (ps["dx"] == 0 and ps["dy"] == 0): + return + assert self.recorder is not None + after_png = ps.get("after_frame") + structural_after = ps.get("structural_after") + self.recorder.record_observed( + {"kind": "scroll", "dx": ps["dx"], "dy": ps["dy"]}, + before_png=self._last_frame, + structural_before=ps.get("structural_before", self._last_structural), + after_png=after_png, + structural_after=structural_after, + ) + self._set_last(after_png, structural_after) + + def _record_click(self, ev: dict[str, Any]) -> None: + assert self.recorder is not None + self.recorder.record_observed( + {"kind": "click", "x": int(ev["x"]), "y": int(ev["y"])}, + before_png=self._last_frame, + structural_before=self._last_structural, + structured_identity=ev.get("sid"), + ) + self._advance() + + def _record_key(self, ev: dict[str, Any]) -> None: + assert self.recorder is not None + self.recorder.record_observed( + {"kind": "key", "key": ev["key"]}, + before_png=self._last_frame, + structural_before=self._last_structural, + ) + self._advance() + + # -- internals ----------------------------------------------------------- + + def _advance(self) -> None: + """After an IMMEDIATE step (click/key), the current settled frame + becomes the next step's BEFORE frame.""" + assert self.backend is not None + self._last_frame = self.backend.screenshot() + self._last_structural = self._structural_state() + + def _set_last( + self, after_png: Optional[bytes], structural_after: Optional[dict] + ) -> None: + """After a DEFERRED/coalesced step (type/scroll), the next step's + BEFORE frame is the after-state captured when the step actually + happened — NOT a live screenshot, which a later navigating action may + already have moved on from.""" + if after_png is not None: + self._last_frame = after_png + else: + self._advance() + return + if structural_after is not None: + self._last_structural = structural_after + + def _structural_state(self) -> dict[str, Any]: + state: dict[str, Any] = {} + for attr, key in ( + ("url", "url"), + ("page_title", "title"), + ("page_count", "pages"), + ): + try: + value = getattr(self.backend, attr, None) + except Exception: + value = None + if value is not None: + state[key] = value + return state + + +def record_interactive( + url: str, + out_dir: Path | str, + *, + secret_fields: tuple[str, ...] = (), + param_fields: tuple[str, ...] = (), + headless: bool = False, + script: Optional[Callable[[Any, Callable[[], None]], None]] = None, + **kwargs: Any, +) -> Path: + """Record a live demonstration the user drives against ``url``. + + Args: + url: The app to record against (the user's own app). + out_dir: Recording output directory (meta.json + events.jsonl + + frames/), the exact format ``compile`` consumes. + secret_fields: Field ``name``/``id`` values to treat as secrets, in + addition to any ``input[type=password]`` (auto-detected). A + secret's literal value is never persisted (see module docstring). + param_fields: Field ``name``/``id`` values recorded as PARAMETERS + (their demonstrated value becomes the default, overridable at + replay with ``--param``); all other non-secret typed fields are + recorded as literals. + headless: Run the browser headless (used by scripted/CI recording; + a human recording is headed). + script: Test hook — ``script(page, pump)`` drives synthetic input and + pumps the loop; when given, the human wait loop is skipped. + + Returns: + The recording directory. + """ + session = InteractiveRecorder( + url, + out_dir, + secret_fields=secret_fields, + param_fields=param_fields, + headless=headless, + **kwargs, + ) + session.start() + if script is not None: + return session.run_script(script) + return session.run() diff --git a/openadapt_flow/ir.py b/openadapt_flow/ir.py index cf6318a..c1f20d7 100644 --- a/openadapt_flow/ir.py +++ b/openadapt_flow/ir.py @@ -171,6 +171,18 @@ class Step(BaseModel): anchor: Optional[Anchor] = None # None for pure keyboard/wait steps text: Optional[str] = None # literal text for TYPE param: Optional[str] = None # if set, TYPE text comes from params[param] + secret: bool = Field( + default=False, + description=( + "TYPE steps only: the parameter is a SECRET (e.g. a password)." + " Its literal value is NEVER stored in the recording, the events" + " log, or this bundle; at replay it is injected from the" + " environment variable OPENADAPT_FLOW_SECRET_ (the param" + " name upper-cased). ``text`` is always None for a secret step," + " and ``param`` names the required secret. A missing secret at" + " replay is a clear, fail-fast error (see runtime.Replayer)." + ), + ) key: Optional[str] = None # for KEY, e.g. "Enter" scroll_dx: Optional[int] = None # for SCROLL: wheel delta, px right scroll_dy: Optional[int] = None # for SCROLL: wheel delta, px down @@ -214,6 +226,14 @@ class Workflow(BaseModel): params: dict[str, str] = Field( default_factory=dict, description="param name -> example/default value" ) + secret_params: list[str] = Field( + default_factory=list, + description=( + "Names of SECRET parameters (e.g. passwords). Their values are" + " NEVER stored here or in ``params``; each is injected at replay" + " from OPENADAPT_FLOW_SECRET_ (see Step.secret)." + ), + ) steps: list[Step] = Field(default_factory=list) # -- bundle I/O --------------------------------------------------------- diff --git a/openadapt_flow/recorder.py b/openadapt_flow/recorder.py index a6fbeb4..fef1d17 100644 --- a/openadapt_flow/recorder.py +++ b/openadapt_flow/recorder.py @@ -30,7 +30,7 @@ from typing import Any, Callable, Optional import imagehash -from PIL import Image +from PIL import Image, ImageDraw from openadapt_flow.backend import Backend @@ -80,6 +80,7 @@ def __init__( self._settle_stable_frames = settle_stable_frames self._settle_timeout_s = settle_timeout_s self._params: dict[str, str] = {} + self._secret_params: set[str] = set() self._i = 0 self._t0 = time.monotonic() @@ -135,6 +136,7 @@ def finish(self) -> Path: "viewport": [int(viewport[0]), int(viewport[1])], "app_url": self._app_url, "params": dict(self._params), + "secret_params": sorted(self._secret_params), } (self._dir / "meta.json").write_text(json.dumps(meta, indent=2)) return self._dir @@ -143,9 +145,7 @@ def finish(self) -> Path: def _record(self, event: dict[str, Any], act: Callable[[], None]) -> None: """Capture before frame, act, wait settle, capture after, log event.""" - i = self._i before = self._backend.screenshot() - (self._frames_dir / f"{i:04d}_before.png").write_bytes(before) structural_before = self._structural_state() # Structured identity of the clicked target (DOM / a11y text), when # the backend exposes it: captured on the BEFORE frame, before the @@ -160,17 +160,128 @@ def _record(self, event: dict[str, Any], act: Callable[[], None]) -> None: event = {**event, "structured_identity": structured} act() after = self._wait_settled() - (self._frames_dir / f"{i:04d}_after.png").write_bytes(after) + self._commit(event, before, after, structural_before) + + def record_observed( + self, + event: dict[str, Any], + *, + before_png: bytes, + structural_before: dict[str, Any], + structured_identity: Optional[str] = None, + param: Optional[str] = None, + secret: bool = False, + redact_region: Optional[tuple[int, int, int, int]] = None, + after_png: Optional[bytes] = None, + structural_after: Optional[dict[str, Any]] = None, + ) -> None: + """Persist an event the USER already performed (no backend action). + + The driving methods (``click`` / ``type_text`` / ...) perform the + action and screenshot around it. ``record_observed`` instead records + an action the caller OBSERVED in a live session the user is driving: + the action already happened, so nothing is performed on the backend. + + Frame chaining mirrors a driven demonstration: ``before_png`` is the + pre-action frame the caller supplies (the previous step's settled + frame — the screen the user saw before acting), and the after frame is + captured now, once the screen settles. + + Args: + event: The event dict (``{"kind": ..., ...}``) without ``i``/``t``. + before_png: Pre-action frame (the previous settled frame). + structural_before: URL/title/page-count observed before the action + (captured in-page at action time, pre-navigation). + structured_identity: DOM/a11y identity of the clicked row, captured + in-page at click time; stored on click/double_click events. + param: Parameter name for a TYPE event, if any. + secret: TYPE only. When True the value is a SECRET: the event + carries NO ``text`` (never persisted), ``param`` is registered + as a secret parameter, and the value is injected from the + environment at replay (see ir.Step.secret). + redact_region: (x, y, w, h) blacked out in BOTH the before and + after frames before they are written — a secret field's pixels + must never persist to disk. + after_png: Pre-captured settled after-frame. When None, the screen + is settled and captured now. + """ + event = dict(event) + if event.get("kind") in ("click", "double_click") and structured_identity: + event["structured_identity"] = structured_identity + if param is not None: + event["param"] = param + if secret: + event["secret"] = True + event.pop("text", None) + self._secret_params.add(param) + else: + self._params[param] = str(event.get("text", "")) + # A caller that already captured the settled after-frame at the right + # moment (e.g. a typed field's value BEFORE a following navigating + # click) passes it in; otherwise we settle and capture now. + if after_png is None: + after_png = self._wait_settled() + self._commit( + event, + before_png, + after_png, + structural_before, + redact_region=redact_region, + structural_after=structural_after, + ) + + def _commit( + self, + event: dict[str, Any], + before_png: bytes, + after_png: bytes, + structural_before: dict[str, Any], + *, + redact_region: Optional[tuple[int, int, int, int]] = None, + structural_after: Optional[dict[str, Any]] = None, + ) -> None: + """Write this step's before/after frames and its events.jsonl line. + + ``redact_region`` (when given) is blacked out in both frames before + they hit disk — the single choke point that keeps a secret field's + pixels out of every persisted frame. ``structural_after`` lets a + caller supply the post-action URL/title/page-count captured at the + right moment (a deferred type-run flush must not read a URL a LATER + navigating click produced); when None it is read now. + """ + i = self._i + if redact_region is not None: + before_png = self._redact(before_png, redact_region) + after_png = self._redact(after_png, redact_region) + (self._frames_dir / f"{i:04d}_before.png").write_bytes(before_png) + (self._frames_dir / f"{i:04d}_after.png").write_bytes(after_png) line: dict[str, Any] = {"i": i, **event} for key, value in structural_before.items(): line[f"{key}_before"] = value - for key, value in self._structural_state().items(): + if structural_after is None: + structural_after = self._structural_state() + for key, value in structural_after.items(): line[f"{key}_after"] = value line["t"] = round(time.monotonic() - self._t0, 3) with self._events_path.open("a") as f: f.write(json.dumps(line) + "\n") self._i += 1 + @staticmethod + def _redact( + png: bytes, region: tuple[int, int, int, int] + ) -> bytes: + """Return ``png`` with ``region`` (x, y, w, h) filled solid black.""" + x, y, w, h = (int(v) for v in region) + with Image.open(io.BytesIO(png)) as img: + frame = img.convert("RGB") + ImageDraw.Draw(frame).rectangle( + [max(0, x), max(0, y), x + w, y + h], fill=(0, 0, 0) + ) + out = io.BytesIO() + frame.save(out, format="PNG") + return out.getvalue() + def _structured_identity_at(self, x: int, y: int) -> Optional[str]: """Structured (DOM / a11y) identity text under (x, y), if available. diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 1a97565..3ac78e3 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -19,6 +19,7 @@ from __future__ import annotations import math +import os import time from datetime import date, datetime, timezone from pathlib import Path @@ -83,6 +84,17 @@ # budget of ~2.5x the total recorded distance. SCROLL_BUDGET_FACTOR = 2.5 +# A secret TYPE step's value is never stored in the bundle; it is read at +# replay from this environment variable (the param name upper-cased, with +# non-alphanumeric characters mapped to '_'). See ir.Step.secret. +SECRET_ENV_PREFIX = "OPENADAPT_FLOW_SECRET_" + + +def secret_env_var(param: str) -> str: + """Environment variable a secret parameter's value is read from.""" + key = "".join(ch if ch.isalnum() else "_" for ch in param).upper() + return f"{SECRET_ENV_PREFIX}{key}" + class Replayer: """Replays a Workflow against a Backend using injected vision. @@ -515,7 +527,24 @@ def _act( return None if step.action is ActionKind.TYPE: - if step.param is not None: + if step.secret: + # Secret value is never in the bundle/params: inject it from + # the environment, failing fast with an actionable message. + env_var = secret_env_var(step.param or "") + text = os.environ.get(env_var, "") + if not step.param: + return ( + f"Step '{step.id}' ({step.intent}) is marked secret " + "but names no parameter" + ) + if not text: + return ( + f"Step '{step.id}' ({step.intent}) requires secret " + f"parameter '{step.param}', but the environment " + f"variable {env_var} is not set — export it with the " + "secret value and re-run" + ) + elif step.param is not None: if step.param not in params: return ( f"Step '{step.id}' ({step.intent}) requires parameter " diff --git a/tests/test_interactive_recorder.py b/tests/test_interactive_recorder.py new file mode 100644 index 0000000..e00640d --- /dev/null +++ b/tests/test_interactive_recorder.py @@ -0,0 +1,213 @@ +"""End-to-end: `record --url` interactively records the user's OWN app. + +A scripted, headless driver plays the canonical MockMed triage flow as if a +user were clicking and typing in the headed browser; the InteractiveRecorder +watches real DOM events and writes the exact recording format the compiler +consumes. We then compile it and replay it — closing the self-serve loop +(record -> compile -> replay) for an arbitrary app, not just the bundled demo. + +The password field (``input[type=password]``) is auto-detected as a secret: +its value is never persisted, its region is redacted from the frames, and at +replay it is injected from the environment. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Iterator + +import numpy as np +import pytest +from PIL import Image + +from openadapt_flow.compiler import compile_recording +from openadapt_flow.interactive_recorder import record_interactive +from openadapt_flow.ir import Workflow +from openadapt_flow.mockmed.server import serve + +pytestmark = pytest.mark.timeout(600) + +NOTE = "Follow-up in two weeks; recheck blood pressure at that visit." +SECRET = "s3cr3t-PASSWORD-never-persist" +REPLAY_NOTE = "A DIFFERENT note value supplied only at replay time." + + +def _drive_triage(page, pump) -> None: + """Play the MockMed triage flow, pumping the recorder between actions.""" + + def click(selector: str) -> None: + page.wait_for_selector(selector, state="visible", timeout=20000) + page.click(selector) + pump() + pump() + + click("#username") + page.keyboard.type("nurse.demo") + pump() + click("#password") + page.keyboard.type(SECRET) # input[type=password] -> auto secret + pump() + click("#signin") + click(".open-btn") + click("#new-encounter") + click("#type-triage") + click("#note") + page.keyboard.type(NOTE) + pump() + pump() + click("#save-encounter") + for _ in range(4): + pump() + + +@pytest.fixture(scope="module") +def server_url() -> Iterator[str]: + url, stop = serve(port=0) + yield url + stop() + + +@pytest.fixture(scope="module") +def recording( + server_url: str, tmp_path_factory: pytest.TempPathFactory +) -> Path: + out = tmp_path_factory.mktemp("interactive_rec") / "rec" + return record_interactive( + server_url, + out, + param_fields=("note",), + headless=True, + script=_drive_triage, + ) + + +@pytest.fixture(scope="module") +def bundle( + recording: Path, tmp_path_factory: pytest.TempPathFactory +) -> Path: + out = tmp_path_factory.mktemp("interactive_bundle") / "bundle" + compile_recording(recording, out, name="my-app-triage") + return out + + +def _events(rec_dir: Path) -> list[dict]: + return [ + json.loads(line) + for line in (rec_dir / "events.jsonl").read_text().splitlines() + if line.strip() + ] + + +def test_recording_shape_matches_the_demonstration(recording: Path) -> None: + meta = json.loads((recording / "meta.json").read_text()) + assert meta["viewport"] == [1280, 800] + assert meta["app_url"] + + events = _events(recording) + # The same 11-action triage flow the canonical demo produces. + assert [e["kind"] for e in events] == [ + "click", # username field + "type", # username (literal) + "click", # password field + "type", # password (SECRET) + "click", # Sign In + "click", # Open referral + "click", # New Encounter + "click", # Triage + "click", # Note field + "type", # note (parameter) + "click", # Save Encounter + ] + # every event has before/after frames on disk + for i in range(len(events)): + for suffix in ("before", "after"): + assert (recording / "frames" / f"{i:04d}_{suffix}.png").is_file() + + # structured identity was captured in-page for at least one row click + assert any(e.get("structured_identity") for e in events if e["kind"] == "click") + + +def test_secret_auto_detected_and_never_persisted(recording: Path) -> None: + meta = json.loads((recording / "meta.json").read_text()) + assert meta["secret_params"] == ["password"] + + type_events = [e for e in _events(recording) if e["kind"] == "type"] + secret_events = [e for e in type_events if e.get("secret")] + assert len(secret_events) == 1 + assert secret_events[0]["param"] == "password" + assert "text" not in secret_events[0] + + # The literal secret appears in NO persisted text artifact. + blob = (recording / "meta.json").read_text() + ( + recording / "events.jsonl" + ).read_text() + assert SECRET not in blob + + +def test_secret_field_region_redacted_in_frames(recording: Path) -> None: + events = _events(recording) + secret_i = next(e["i"] for e in events if e.get("secret")) + for suffix in ("before", "after"): + arr = np.asarray( + Image.open( + recording / "frames" / f"{secret_i:04d}_{suffix}.png" + ).convert("RGB") + ) + # Redaction fills the field rect solid black — a sizeable pure-black + # block that an unredacted login screen would never contain. + assert int((arr.sum(axis=2) == 0).sum()) > 500 + + +def test_secret_absent_from_compiled_bundle(bundle: Path) -> None: + workflow = Workflow.load(bundle) + assert workflow.secret_params == ["password"] + secret_steps = [s for s in workflow.steps if s.secret] + assert len(secret_steps) == 1 and secret_steps[0].text is None + # No persisted text file in the bundle carries the literal. + for path in bundle.rglob("*"): + if path.suffix in (".json", ".py", ".txt", ".md"): + assert SECRET not in path.read_text() + + +def _replay(bundle: Path, url: str, run_dir: Path): + from playwright.sync_api import sync_playwright + + from openadapt_flow.backends.playwright_backend import PlaywrightBackend + from openadapt_flow.runtime import Replayer + + with sync_playwright() as p: + browser = p.chromium.launch(headless=True) + page = browser.new_page( + viewport={"width": 1280, "height": 800}, device_scale_factor=1 + ) + try: + page.goto(url) + report = Replayer(PlaywrightBackend(page)).run( + Workflow.load(bundle), + params={"note": REPLAY_NOTE}, + bundle_dir=bundle, + run_dir=run_dir, + ) + finally: + browser.close() + return report + + +def test_replay_requires_secret_from_env( + bundle: Path, server_url: str, tmp_path: Path, monkeypatch +) -> None: + monkeypatch.delenv("OPENADAPT_FLOW_SECRET_PASSWORD", raising=False) + report = _replay(bundle, server_url, tmp_path / "run_missing") + assert not report.success + failed = [r for r in report.results if not r.ok] + assert failed and "OPENADAPT_FLOW_SECRET_PASSWORD" in (failed[0].error or "") + + +def test_replay_succeeds_with_secret_and_new_param( + bundle: Path, server_url: str, tmp_path: Path, monkeypatch +) -> None: + monkeypatch.setenv("OPENADAPT_FLOW_SECRET_PASSWORD", SECRET) + report = _replay(bundle, server_url, tmp_path / "run_ok") + assert report.success, [r.error for r in report.results if not r.ok] + assert len(report.results) == 11 diff --git a/tests/test_secret_params.py b/tests/test_secret_params.py new file mode 100644 index 0000000..3b9b5b9 --- /dev/null +++ b/tests/test_secret_params.py @@ -0,0 +1,222 @@ +"""Secret-typed parameters: never persisted, injected from the environment. + +Fast unit tests (no browser) covering the secret contract end to end: + +* the Recorder never writes a secret's literal (meta/events) and redacts its + field region from the persisted frames; +* the compiler carries the secret through to ``Step.secret`` / + ``Workflow.secret_params`` with ``text=None`` and no leak; +* the Replayer injects the value from ``OPENADAPT_FLOW_SECRET_`` and + fails fast with an actionable message when it is missing. +""" + +from __future__ import annotations + +import io +import json +from pathlib import Path + +import numpy as np +import pytest +from PIL import Image + +from openadapt_flow.compiler import compile_recording +from openadapt_flow.ir import ActionKind, Step, Workflow +from openadapt_flow.recorder import Recorder +from openadapt_flow.runtime.replayer import Replayer, secret_env_var + +SECRET = "hunter2-SUPER-secret-value" + + +def _png(size=(1280, 800), color=(250, 250, 250)) -> bytes: + buf = io.BytesIO() + Image.new("RGB", size, color).save(buf, format="PNG") + return buf.getvalue() + + +class FakeBackend: + def __init__(self) -> None: + self._png = _png() + self.actions: list = [] + + @property + def viewport(self): + return (1280, 800) + + def screenshot(self): + return self._png + + def click(self, x, y, *, double=False): + self.actions.append(("click", x, y, double)) + + def type_text(self, text): + self.actions.append(("type", text)) + + def press(self, key): + self.actions.append(("press", key)) + + def scroll(self, dx, dy): + self.actions.append(("scroll", dx, dy)) + + +def _events(rec_dir: Path) -> list[dict]: + return [ + json.loads(line) + for line in (rec_dir / "events.jsonl").read_text().splitlines() + if line.strip() + ] + + +# -- Recorder --------------------------------------------------------------- + + +def test_recorder_secret_not_persisted_and_region_redacted(tmp_path: Path) -> None: + rec = Recorder(FakeBackend(), tmp_path / "rec", app_url="http://app/") + before = _png() + rec.record_observed( + {"kind": "type"}, + before_png=before, + structural_before={}, + param="password", + secret=True, + redact_region=(100, 200, 300, 40), + ) + rec_dir = rec.finish() + + # meta: the param is listed as secret, its value is NOWHERE. + meta = json.loads((rec_dir / "meta.json").read_text()) + assert meta["secret_params"] == ["password"] + assert "password" not in meta["params"] + blob = (rec_dir / "meta.json").read_text() + ( + rec_dir / "events.jsonl" + ).read_text() + assert SECRET not in blob + + # events: the type event carries the param + secret flag but NO text. + (event,) = _events(rec_dir) + assert event["kind"] == "type" + assert event["param"] == "password" + assert event["secret"] is True + assert "text" not in event + + # frames: the redacted region is solid black in both frames. + for suffix in ("before", "after"): + arr = np.asarray( + Image.open(rec_dir / "frames" / f"0000_{suffix}.png").convert("RGB") + ) + region = arr[200:240, 100:400] + assert region.sum() == 0, f"{suffix} region not redacted" + + +def test_recorder_non_secret_param_keeps_value(tmp_path: Path) -> None: + rec = Recorder(FakeBackend(), tmp_path / "rec") + rec.record_observed( + {"kind": "type", "text": "visible note"}, + before_png=_png(), + structural_before={}, + param="note", + ) + rec_dir = rec.finish() + meta = json.loads((rec_dir / "meta.json").read_text()) + assert meta["params"] == {"note": "visible note"} + assert meta["secret_params"] == [] + (event,) = _events(rec_dir) + assert event["text"] == "visible note" and event["param"] == "note" + + +# -- compiler --------------------------------------------------------------- + + +def _write_secret_recording(rec_dir: Path) -> None: + (rec_dir / "frames").mkdir(parents=True) + (rec_dir / "frames" / "0000_before.png").write_bytes(_png()) + (rec_dir / "frames" / "0000_after.png").write_bytes(_png()) + (rec_dir / "meta.json").write_text( + json.dumps( + { + "id": "abc", + "created_at": "2026-07-13T00:00:00+00:00", + "viewport": [1280, 800], + "app_url": "http://app/", + "params": {}, + "secret_params": ["password"], + } + ) + ) + (rec_dir / "events.jsonl").write_text( + json.dumps({"i": 0, "kind": "type", "param": "password", "secret": True}) + + "\n" + ) + + +def test_compiler_carries_secret_without_value(tmp_path: Path) -> None: + rec_dir = tmp_path / "rec" + _write_secret_recording(rec_dir) + bundle = tmp_path / "bundle" + workflow = compile_recording(rec_dir, bundle, name="secret-wf") + + assert workflow.secret_params == ["password"] + (step,) = workflow.steps + assert step.action is ActionKind.TYPE + assert step.secret is True + assert step.param == "password" + assert step.text is None + # Nothing in the persisted bundle carries a value for the secret. + assert "password" not in workflow.params + assert SECRET not in (bundle / "workflow.json").read_text() + + +# -- replayer --------------------------------------------------------------- + + +def test_secret_env_var_mapping() -> None: + assert secret_env_var("password") == "OPENADAPT_FLOW_SECRET_PASSWORD" + assert secret_env_var("api-key") == "OPENADAPT_FLOW_SECRET_API_KEY" + + +def _secret_workflow() -> Workflow: + return Workflow( + name="secret-only", + params={}, + secret_params=["password"], + steps=[ + Step( + id="step_000", + intent="type (secret)", + action=ActionKind.TYPE, + param="password", + secret=True, + ) + ], + ) + + +def test_replayer_injects_secret_from_env(tmp_path, monkeypatch) -> None: + from tests.test_replayer import FakeBackend as RBackend, FakeVision + + monkeypatch.setenv("OPENADAPT_FLOW_SECRET_PASSWORD", SECRET) + backend = RBackend() + bundle = tmp_path / "bundle" + (bundle / "templates").mkdir(parents=True) + report = Replayer(backend, vision=FakeVision()).run( + _secret_workflow(), params={}, bundle_dir=bundle, run_dir=tmp_path / "run" + ) + assert report.success, [r.error for r in report.results] + assert ("type", SECRET) in backend.actions + + +def test_replayer_missing_secret_errors_clearly(tmp_path, monkeypatch) -> None: + from tests.test_replayer import FakeBackend as RBackend, FakeVision + + monkeypatch.delenv("OPENADAPT_FLOW_SECRET_PASSWORD", raising=False) + backend = RBackend() + bundle = tmp_path / "bundle" + (bundle / "templates").mkdir(parents=True) + report = Replayer(backend, vision=FakeVision()).run( + _secret_workflow(), params={}, bundle_dir=bundle, run_dir=tmp_path / "run" + ) + assert not report.success + (result,) = report.results + assert "OPENADAPT_FLOW_SECRET_PASSWORD" in (result.error or "") + # Nothing was typed: the secret never reached the backend. + assert not any(a[0] == "type" for a in backend.actions)