diff --git a/packages/flow/docs/JIT.md b/packages/flow/docs/JIT.md index 82ec8fc72..5af26e557 100644 --- a/packages/flow/docs/JIT.md +++ b/packages/flow/docs/JIT.md @@ -50,6 +50,46 @@ Claude computer-use trajectory against the bundled MockMed app: python examples/jit/run_mockmed.py ``` +## Desktop and remote apps + +`WebApp(url=...)` is a browser app. `SurfaceApp` is any other app, named the +way a deployment file names it: the `backend` section of a `deploy.yaml`. + +```python +from openadapt_flow.jit import SurfaceApp + +app = SurfaceApp.from_deployment("deploy.yaml", record_reader=read_records) +# or +app = SurfaceApp({"kind": "windows", "agent_url": "http://127.0.0.1:5001"}) +``` + +| `backend.kind` | The section names | The agent sees | +| --- | --- | --- | +| `windows` | `agent_url` (the in-guest agent) | screenshots; clicks keep the UI Automation locator | +| `macos` | `macos_app`, optionally `macos_window_title` | screenshots; clicks keep the Accessibility locator | +| `linux` | `linux_app`, `linux_window_title`, and `linux_allow_physical_input: true` (see below) | screenshots; clicks keep the AT-SPI locator | +| `rdp` | `rdp_host`, or `rdp_window` for a local client window | pixels only | +| `citrix` | nothing (the Citrix Workspace window) or `rdp_window`; runs need `rdp_readiness_text` | pixels only | + +An agent clicks at window coordinates. The Linux backend refuses +coordinate input unless the section sets `linux_allow_physical_input: true`, +which allows window-bound X11 input; without it the capture ends +`capture_failed` with "coordinate input is disabled for native Linux". + +The agent drives the app through the same `AuthoringSession` and +`CaptureComputer` the MCP recording tools use. Every canary and run opens a +fresh backend from the same section. The app's origin is `://` +in lower case, for example `macos://textedit` or `rdp://10.0.0.5`. Window +titles and credentials are left out: a title can carry a patient or document +name. A program admitted on one origin is refused on any other, by the JIT +wrapper and by `openadapt-flow run --config` alike. An admission policy names +a desktop or remote app in `allowed_hosts` with its kind +(`rdp://10.0.0.5`); a bare host there names a web app only, so it never +covers the remote desktop on the same address. + +The Windows origin is the agent's host and port. Two VMs reached through the +same tunnel port share it, so use a separate cache directory per environment. + ## What each run does | State of the task | `run(**params)` does | `status` | @@ -241,8 +281,13 @@ explores again. ## Limits -- Web apps only. `AuthoringSession` can drive desktop, RDP and Citrix, but the - wrapper launches a Playwright browser for capture and for runs. +- On RDP and Citrix the agent sees only pixels, and a pixel click carries no + verified record identity. The standard run gate refuses the canary + ("Identity coverage"), exactly as `run --profile standard` refuses any + pixel-only bundle that was not qualified. Nothing is sent to the window. +- Citrix canaries and runs refuse without `backend.rdp_readiness_text`. +- The `playwright_mcp` action format needs a browser page; a `SurfaceApp` + refuses it. - Without a `record_reader`, the candidate has no effect contract, admission refuses it, and no run can reach `VERIFIED`. - Secret inputs are not lifted. A secret is typed by a person at a pause diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index 7b61d75b7..9303e9528 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -798,6 +798,7 @@ def _build_and_run_replayer( surface_override: bool = False, execution_origin: Optional[str] = None, execution_entry_url: Optional[str] = None, + execution_target_origin: Optional[str] = None, run_id: Optional[str] = None, approval_gate=None, qualification_fault_driver=None, @@ -836,6 +837,7 @@ def _build_and_run_replayer( surface_override=surface_override, execution_origin=execution_origin, execution_entry_url=execution_entry_url, + execution_target_origin=execution_target_origin, run_id=( remote_delivery_run_id if delivery_authority_kind == "cloud_runner" @@ -1086,6 +1088,7 @@ def _replay_desktop( qualification_campaign_guard=qualification_campaign_guard, runtime_config=runtime_config, execution_target_kind=_report_backend_kind(backend_cfg.kind), + execution_target_origin=_surface_target_origin_or_none(backend_cfg), surface_override=bool(getattr(args, "_surface_override", False)), run_id=run_id, approval_gate=approval_gate, @@ -1103,6 +1106,18 @@ def _replay_desktop( ) +def _surface_target_origin_or_none(backend_cfg) -> Optional[str]: + """The non-web app's ``://``, for an admitted JIT program's + canary match. A section that can't name one gives None, which that match + refuses; every other bundle ignores it.""" + from openadapt_flow.surface_selection import surface_target_origin + + try: + return surface_target_origin(backend_cfg) + except ValueError: + return None + + def _record_until_stop(args: argparse.Namespace): """Poll ``--until-cmd`` until it exits 0. Ctrl-C remains the fallback.""" diff --git a/packages/flow/openadapt_flow/jit/__init__.py b/packages/flow/openadapt_flow/jit/__init__.py index d2aab4e12..12aa2e920 100644 --- a/packages/flow/openadapt_flow/jit/__init__.py +++ b/packages/flow/openadapt_flow/jit/__init__.py @@ -25,10 +25,13 @@ ) from openadapt_flow.jit.compile_agent import ( RECORDING_WARNING, + SURFACE_KINDS, + App, CompiledAgent, ComputerUseAgent, JitResult, Recording, + SurfaceApp, TaskRequest, WebApp, compile_agent, @@ -42,6 +45,7 @@ from openadapt_flow.jit.signature import ParamError, TaskSignature __all__ = [ + "App", "AdmissionPolicy", "AuthorityError", "Principal", @@ -50,6 +54,7 @@ "principal_from_key", "DIALECTS", "RECORDING_WARNING", + "SURFACE_KINDS", "AdmissionError", "BundleState", "CaptureComputer", @@ -61,6 +66,7 @@ "Observation", "ParamError", "Recording", + "SurfaceApp", "TaskRequest", "TaskSignature", "UnsupportedAction", diff --git a/packages/flow/openadapt_flow/jit/authority.py b/packages/flow/openadapt_flow/jit/authority.py index dc03def83..24783ae13 100644 --- a/packages/flow/openadapt_flow/jit/authority.py +++ b/packages/flow/openadapt_flow/jit/authority.py @@ -316,7 +316,8 @@ class AdmissionPolicy(BaseModel): signer: str = Field(pattern=_ID) signed_at: str = Field(min_length=10, max_length=64) expires_at: str = Field(min_length=10, max_length=64) - #: ``host`` or ``host:port`` of the target app's origin. + #: ``host`` or ``host:port`` of a web app's origin, or a desktop or + #: remote app's ``://`` (``macos://textedit``). allowed_hosts: tuple[str, ...] = Field(min_length=1) #: Step actions (``openadapt_flow.ir.ActionKind`` values). allowed_actions: tuple[str, ...] = Field(min_length=1) @@ -460,8 +461,17 @@ def scope_problems( problems: list[str] = [] if target_origin is not None: - host = (urlsplit(target_origin).netloc or "").lower() - if host not in policy.allowed_hosts: + parts = urlsplit(target_origin) + if parts.scheme in ("http", "https"): + # A bare host names a web app only. + host = (parts.netloc or "").lower() + allowed = {h.lower() for h in policy.allowed_hosts if "://" not in h} + else: + # A desktop or remote app is named with its kind + # (``macos://textedit``), so a web host entry never covers it. + host = target_origin.lower() + allowed = {h.lower() for h in policy.allowed_hosts if "://" in h} + if host not in allowed: problems.append(f"the app at {host or target_origin!r} is not allowed") for step in workflow.steps: if step.action.value not in policy.allowed_actions: diff --git a/packages/flow/openadapt_flow/jit/compile_agent.py b/packages/flow/openadapt_flow/jit/compile_agent.py index 1956ffb9e..e72884736 100644 --- a/packages/flow/openadapt_flow/jit/compile_agent.py +++ b/packages/flow/openadapt_flow/jit/compile_agent.py @@ -50,8 +50,11 @@ "CompiledAgent", "ComputerUseAgent", "JitResult", + "App", "RECORDING_WARNING", "Recording", + "SURFACE_KINDS", + "SurfaceApp", "TaskRequest", "WebApp", "compile_agent", @@ -61,7 +64,7 @@ ADMIT_FILE = lifecycle.ADMISSION_FILE NEEDS_HUMAN_ADMIT = "needs_human_admit" -#: Printed to stderr before every recording starts, before the browser opens. +#: Printed to stderr before every recording starts, before the app opens. #: An ``on_warning`` callback gets it too; it can't replace the print. RECORDING_WARNING = ( "Before you record: the recording keeps every screen the agent sees and " @@ -100,6 +103,193 @@ def origin(self) -> str: parts = urlsplit(self.url) return f"{parts.scheme}://{parts.netloc}" + @property + def kind(self) -> str: + return "web" + + @property + def entry_url(self) -> Optional[str]: + return self.url + + @property + def backend_hints(self) -> Optional[dict[str, Any]]: + return None + + @property + def deployment_surface(self) -> Optional[str]: + return None + + def open(self) -> tuple[Any, Callable[[], None]]: + """Launch a browser at :attr:`url`. Returns the backend and its close.""" + + from openadapt_flow.backends.playwright_backend import PlaywrightBackend + + return PlaywrightBackend.launch(self.url, headless=self.headless) + + +#: Backend kinds :class:`SurfaceApp` drives. ``web`` is :class:`WebApp`. +SURFACE_KINDS = ("windows", "macos", "linux", "rdp", "citrix") + + +def _surface_kind(kind: Optional[str]) -> str: + k = (kind or "").strip().lower().replace("_", "-") + return "rdp" if k == "remote-display" else k + + +@dataclass(frozen=True) +class SurfaceApp: + """A native or remote-display app the task runs in. + + The app is named the way a deployment file names it: the ``backend`` + section of a ``deploy.yaml`` (``kind`` plus that kind's target). The agent + drives it through the same :class:`~openadapt_flow.authoring.AuthoringSession` + and :class:`CaptureComputer` the MCP recording tools use, and every canary + and run opens a fresh backend from the same section. + + ============ ======================================================= + ``kind`` Target the section must name + ============ ======================================================= + ``windows`` ``agent_url`` (the in-guest agent; UI Automation) + ``macos`` ``macos_app`` (Accessibility; ``macos_window_title``) + ``linux`` ``linux_app`` and ``linux_window_title`` (AT-SPI); + agent clicks need ``linux_allow_physical_input`` + ``rdp`` ``rdp_host`` (network RDP) or ``rdp_window`` (local client) + ``citrix`` nothing (the host's Citrix Workspace window), or + ``rdp_window``; runs need ``rdp_readiness_text`` + ============ ======================================================= + + RDP and Citrix are screen-only: the agent works from pixels and keys. + + Args: + backend: The deployment ``backend`` section, as a + :class:`~openadapt_flow.deployment.BackendConfig` or a mapping. + record_reader: As for :class:`WebApp`. + reset: As for :class:`WebApp`. + open_backend: Builds the live backend from ``backend``. Defaults to + :func:`openadapt_flow.backends.factory.build_backend`. Pass one to + inject a client (for example a test fixture). + """ + + backend: Any + record_reader: Optional[Callable[[], Optional[list[dict[str, Any]]]]] = None + reset: Optional[Callable[[], None]] = None + open_backend: Optional[Callable[[Any], Any]] = field(default=None, repr=False) + + def __post_init__(self) -> None: + from openadapt_flow.deployment import BackendConfig + + cfg = self.backend + if not isinstance(cfg, BackendConfig): + cfg = BackendConfig.model_validate(dict(cfg)) + object.__setattr__(self, "backend", cfg) + kind = _surface_kind(cfg.kind) + if kind == "web": + raise ValueError("backend.kind 'web' is a browser app: use WebApp(url=...)") + if kind not in SURFACE_KINDS: + raise ValueError( + f"unknown backend.kind {cfg.kind!r} (expected one of " + f"{', '.join(SURFACE_KINDS)})" + ) + missing = { + "windows": None if cfg.agent_url else "backend.agent_url", + "macos": None if cfg.macos_app else "backend.macos_app", + "linux": ( + None + if cfg.linux_app and cfg.linux_window_title + else "backend.linux_app and backend.linux_window_title" + ), + "rdp": ( + None + if bool(cfg.rdp_host) != bool(cfg.rdp_window) + else "exactly one of backend.rdp_host or backend.rdp_window" + ), + "citrix": ( + "no backend.rdp_host (Citrix drives a local Workspace window)" + if cfg.rdp_host + else None + ), + }[kind] + if missing: + raise ValueError(f"backend.kind {kind!r} needs {missing}") + self.origin # noqa: B018 - refuses a credential in agent_url now + + @classmethod + def from_deployment(cls, path: Path | str, **kwargs: Any) -> "SurfaceApp": + """The app named by the ``backend`` section of a deployment file.""" + + from openadapt_flow.deployment import load_deployment + + return cls(load_deployment(path).backend, **kwargs) + + @property + def kind(self) -> str: + return _surface_kind(self.backend.kind) + + @property + def url(self) -> Optional[str]: + return None + + @property + def entry_url(self) -> Optional[str]: + return None + + @property + def origin(self) -> str: + """``://``, from + :func:`~openadapt_flow.surface_selection.surface_target_origin`. A + canary admits the program on this origin only. An admission policy + names it in ``allowed_hosts`` the same way (``macos://textedit``); a + bare host there names a web app only.""" + + from openadapt_flow.surface_selection import surface_target_origin + + origin = surface_target_origin(self.backend) + assert origin is not None + return origin + + @property + def backend_hints(self) -> Optional[dict[str, Any]]: + if self.kind not in ("rdp", "citrix"): + return None + return { + "rdp_window": self.backend.rdp_window, + "rdp_window_title": self.backend.rdp_window_title, + "rdp_readiness_text": self.backend.rdp_readiness_text, + } + + @property + def deployment_surface(self) -> Optional[str]: + return self.backend.deployment_surface + + def run_refusal(self) -> Optional[str]: + """Why a governed canary or run can't start here, or None.""" + + if ( + self.kind == "citrix" + and not str(self.backend.rdp_readiness_text or "").strip() + ): + return ( + "governed Citrix execution needs a stable current-frame " + "readiness marker before any action: set backend.rdp_readiness_text" + ) + return None + + def open(self) -> tuple[Any, Callable[[], None]]: + """Build a fresh backend from the section. Returns it and its close.""" + + if self.open_backend is not None: + backend = self.open_backend(self.backend) + else: + from openadapt_flow.backends.factory import build_backend + + backend = build_backend(self.backend) + close = getattr(backend, "close", None) + return backend, (close if callable(close) else (lambda: None)) + + +#: What :func:`compile_agent` runs a task in. +App = WebApp | SurfaceApp + @dataclass(frozen=True) class TaskRequest: @@ -188,7 +378,6 @@ def __init__( def __enter__(self) -> "Recording": from openadapt_flow.authoring import AuthoringSession - from openadapt_flow.backends.playwright_backend import PlaywrightBackend owner = self._owner if owner.admitted or owner.candidate() is not None: @@ -212,21 +401,22 @@ def __enter__(self) -> "Recording": ) if owner.app.reset is not None: owner.app.reset() - backend, self._close = PlaywrightBackend.launch( - owner.app.url, headless=owner.app.headless - ) + backend, self._close = owner.app.open() try: self._session = AuthoringSession( backend, program / "recording", - backend_kind="web", + backend_kind=owner.app.kind, app_url=owner.app.url, + backend_hints=owner.app.backend_hints, system_of_record_reader=owner.app.record_reader, ) self.computer = CaptureComputer( self._session, dialect=self._dialect, - page=backend.page, + page=getattr(backend, "page", None) + if owner.app.kind == "web" + else None, sleep=owner._sleep, ) self._session.start_record() @@ -286,7 +476,7 @@ class CompiledAgent: agent: Optional[ComputerUseAgent] task: TaskSignature - app: WebApp + app: App verifier: Any policy: str cache_dir: Path @@ -764,7 +954,6 @@ def _run_governed( the person attending its canary. """ - from openadapt_flow.backends.playwright_backend import PlaywrightBackend from openadapt_flow.deployment import DeploymentConfig, PolicySection from openadapt_flow.execution_profiles import ( ExecutionProfile, @@ -782,6 +971,9 @@ def _run_governed( lifecycle.verify_admission(self.bundle_dir, key_dir=self.key_dir) except AdmissionError as exc: return f"a DRAFT never runs unattended: {exc}", None + refusal = getattr(self.app, "run_refusal", lambda: None)() + if refusal: + return f"refused before any action: {refusal}", None gate = evaluate_run_gate( workflow, bundle_dir=self.bundle_dir, @@ -806,9 +998,7 @@ def _run_governed( ) if self.app.reset is not None: self.app.reset() - backend, close = PlaywrightBackend.launch( - self.app.url, headless=self.app.headless - ) + backend, close = self.app.open() try: report = Replayer( backend, @@ -818,14 +1008,18 @@ def _run_governed( require_settled=True, attended_draft_by=attended_by, draft_key_dir=self.key_dir, + deployment_surface=self.app.deployment_surface, ).run( workflow.model_copy(deep=True), params=values, bundle_dir=self.bundle_dir, run_dir=run_dir, - execution_target_kind="web", - execution_origin=self.app.origin, - execution_entry_url=self.app.url, + execution_target_kind=self.app.kind, + execution_origin=self.app.origin if self.app.kind == "web" else None, + execution_entry_url=self.app.entry_url, + execution_target_origin=( + None if self.app.kind == "web" else self.app.origin + ), ) finally: close() @@ -884,7 +1078,7 @@ def compile_agent( agent: Optional[ComputerUseAgent] = None, *, task: TaskSignature, - app: WebApp, + app: App, verifier: Any, policy: str, cache_dir: Path | str = ".openadapt-jit", @@ -901,8 +1095,11 @@ def compile_agent( agent: The computer-use agent, used on a cache miss only. Leave it out to drive your own loop with :meth:`CompiledAgent.recording`. task: The task id, the instruction, and the typed parameter schema. - app: The web app, with the read-only system-of-record snapshot used - while the agent's run is recorded. + app: The app: :class:`WebApp` for a browser app, or + :class:`SurfaceApp` for a Windows, macOS, Linux, RDP or Citrix + app named by a deployment ``backend`` section. Either carries the + read-only system-of-record snapshot used while the agent's run is + recorded. verifier: The effect verifier a run reads the system of record with, such as :class:`~openadapt_flow.runtime.effects.RestRecordVerifier`. policy: The policy the program must pass to be admitted and run, such @@ -931,6 +1128,12 @@ def compile_agent( dialect = dialect or "generic" if dialect not in dialects(): raise ValueError(f"unknown dialect {dialect!r} (expected one of {dialects()})") + if dialect == "playwright_mcp" and app.kind != "web": + raise ValueError( + "the playwright_mcp dialect targets CSS selectors in a browser page; " + f"a {app.kind} app has none. Use anthropic, openai or generic " + "(coordinates and keys)" + ) return CompiledAgent( agent=agent, task=task, diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 8760c06e3..930d5b806 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -1344,6 +1344,7 @@ def run( surface_override: bool = False, execution_origin: Optional[str] = None, execution_entry_url: Optional[str] = None, + execution_target_origin: Optional[str] = None, prior_screenshots_may_leave_box: bool = False, prior_model_calls: int = 0, prior_external_network_calls: Literal[ @@ -1416,6 +1417,12 @@ def run( execution_entry_url: Browser URL requested before replay. Hosted validation binds this to the runner's entry URL so workflows that start below `/` retain their required path. + execution_target_origin: For a non-web run, the app this run + drives, as ``://`` (see + :func:`openadapt_flow.surface_selection.surface_target_origin`). + Only an admitted JIT program's canary match reads it; it is + not written to the report, which keeps ``execution_origin`` + for browser origins. prior_screenshots_may_leave_box: Durable-resume audit carry. True when an earlier leg of the same logical run was configured with an egress-capable screenshot consumer. @@ -1453,7 +1460,7 @@ def run( workflow=workflow, attended_by=self._attended_draft_by, key_dir=self._draft_key_dir, - execution_origin=execution_origin, + execution_origin=execution_origin or execution_target_origin, effect_verifier=self.effect_verifier, at_run=True, ) diff --git a/packages/flow/openadapt_flow/surface_selection.py b/packages/flow/openadapt_flow/surface_selection.py index 189ca6ff1..ea2afcb38 100644 --- a/packages/flow/openadapt_flow/surface_selection.py +++ b/packages/flow/openadapt_flow/surface_selection.py @@ -205,3 +205,41 @@ def store_last_surface(surface: ExecutionTargetKind) -> None: path.write_text(json.dumps(existing, indent=2) + "\n") except OSError: return + + +def surface_target_origin(backend: object) -> Optional[str]: + """The app a non-web backend section names, as ``://``. + + ``backend`` is a :class:`~openadapt_flow.deployment.BackendConfig`. The + target is the Windows agent's host and port, the macOS or Linux app name, + the RDP host (and a non-default port) or the local client window's owner. + Window titles and credentials are left out: a title can carry a patient + or document name, and a URL's userinfo is a secret. ``None`` for ``web``, + which is named by its browser origin instead. + """ + from urllib.parse import quote, urlsplit + + kind = str(getattr(backend, "kind", None) or "web").strip().lower() + kind = kind.replace("_", "-") + kind = "rdp" if kind == "remote-display" else kind + if kind == "web": + return None + if kind == "windows": + parts = urlsplit(str(getattr(backend, "agent_url", "") or "")) + if parts.username or parts.password: + raise ValueError( + "backend.agent_url must not carry credentials; use agent_token" + ) + target = parts.hostname or "" + if parts.port: + target += f":{parts.port}" + elif kind == "macos": + target = str(getattr(backend, "macos_app", "") or "") + elif kind == "linux": + target = str(getattr(backend, "linux_app", "") or "") + elif kind == "rdp" and getattr(backend, "rdp_host", None): + port = int(getattr(backend, "rdp_port", None) or 3389) + target = str(backend.rdp_host) + ("" if port == 3389 else f":{port}") # type: ignore[attr-defined] + else: + target = str(getattr(backend, "rdp_window", None) or "workspace") + return f"{kind}://{quote(target.lower(), safe='.:-_ ')}" diff --git a/packages/flow/tests/surface_kiosk/__init__.py b/packages/flow/tests/surface_kiosk/__init__.py new file mode 100644 index 000000000..a79acb359 --- /dev/null +++ b/packages/flow/tests/surface_kiosk/__init__.py @@ -0,0 +1,133 @@ +"""A browser page shown to Flow as a pixel-only remote-desktop window. + +:class:`PageWindowClient` is a +:class:`~openadapt_flow.backends.remote_display.WindowClient` over a +Playwright page. Flow's real ``RemoteDisplayBackend`` (``rdp`` with a local +client window) and ``CitrixWorkspaceBackend`` (``citrix``) drive it exactly as +they drive an RDP or Citrix client: they capture the window's pixels and send +pointer and key input at window coordinates. They never see the page's DOM. + +This is the same stand-in the Citrix Workspace benchmark uses +(``benchmark/citrix_workspace/``), without Docker or noVNC, so it runs in the +unit suite. It is not ICA/HDX or a real RDP session. +""" + +from __future__ import annotations + +import io +from typing import Any, Optional + +from PIL import Image + +from openadapt_flow.backends.remote_display import WindowInfo + +#: Named keys the client can send, in the order of their fake key codes. +_KEYS = ( + "Enter", + "Tab", + "Escape", + "Backspace", + "Delete", + "ArrowUp", + "ArrowDown", + "ArrowLeft", + "ArrowRight", + "Home", + "End", + "PageUp", + "PageDown", + "Space", +) +_KEY_TOKENS = {name.lower(): i for i, name in enumerate(_KEYS)} +_KEY_TOKENS.update({"return": 0, "esc": 2, "up": 5, "down": 6, "left": 7, "right": 8}) +_FLAGS = {"shift": "Shift", "control": "Control", "ctrl": "Control", "alt": "Alt"} + + +class PageWindowClient: + """One fixed window, owner ``owner`` and title ``title``, over ``page``.""" + + def __init__( + self, page: Any, *, owner: str, title: str, width: int = 1280, height: int = 800 + ) -> None: + self.page = page + self._w, self._h = width, height + self.window = WindowInfo( + window_id=7, + owner=owner, + title=title, + pid=4343, + bounds=(0.0, 0.0, float(width), float(height)), + on_screen=True, + ) + self.inputs = 0 + + # -- WindowClient -------------------------------------------------------- + + def input_trusted(self) -> bool: + return True + + def frontmost_pid(self) -> Optional[int]: + return self.window.pid + + def find_windows(self, owner: str, title: Optional[str]) -> list[WindowInfo]: + if owner.lower() not in self.window.owner.lower(): + return [] + if title is not None and title not in self.window.title: + return [] + return [self.window] + + def key_window_id(self, pid: int) -> Optional[int]: + return self.window.window_id if pid == self.window.pid else None + + def window_at_point(self, x: float, y: float) -> Optional[int]: + return self.window.window_id + + def capture(self, window_id: int) -> tuple[bytes, int, int]: + raw = self.page.screenshot( + clip={"x": 0, "y": 0, "width": self._w, "height": self._h} + ) + img = Image.open(io.BytesIO(raw)).convert("RGB") + out = io.BytesIO() + img.save(out, format="PNG") + return out.getvalue(), img.width, img.height + + def activate(self, pid: int) -> None: + return None + + def mouse_move(self, x: float, y: float) -> None: + self.page.mouse.move(x, y) + + def mouse( + self, x: float, y: float, *, button: str, down: bool, click_count: int = 1 + ) -> None: + self.inputs += 1 + self.page.mouse.move(x, y) + if down: + self.page.mouse.down(button=button, click_count=click_count) + else: + self.page.mouse.up(button=button, click_count=click_count) + + def type_chars(self, text: str) -> None: + self.inputs += 1 + self.page.keyboard.type(text) + + def key(self, keycode: int, *, down: bool, flags: list[str]) -> None: + self.inputs += 1 + mods = [_FLAGS[f] for f in flags if f in _FLAGS] + name = _KEYS[keycode] + if down: + for mod in mods: + self.page.keyboard.down(mod) + self.page.keyboard.down(name) + else: + self.page.keyboard.up(name) + for mod in reversed(mods): + self.page.keyboard.up(mod) + + def scroll(self, dx: int, dy: int) -> None: + self.inputs += 1 + self.page.mouse.wheel(dx, dy) + + def resolve_key(self, token: str) -> Optional[tuple[int, bool]]: + code = _KEY_TOKENS.get(token.strip().lower()) + return None if code is None else (code, False) diff --git a/packages/flow/tests/test_jit_surface_app.py b/packages/flow/tests/test_jit_surface_app.py new file mode 100644 index 000000000..bd2dd95e0 --- /dev/null +++ b/packages/flow/tests/test_jit_surface_app.py @@ -0,0 +1,339 @@ +"""openadapt_flow.jit on Windows, macOS, Linux, RDP and Citrix apps. + +:class:`~openadapt_flow.jit.SurfaceApp` names the app the way a deployment +file's ``backend`` section does. These tests check that it is validated with +named refusals, that recording, canaries and runs open the backend the section +names, and that an admitted program runs only on the surface its canary ran +on. No browser and no model: runs are stubbed as in test_jit_lifecycle.py. +tests/test_jit_surface_pixels.py records through the real RDP and Citrix +backends. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest + +from openadapt_flow.deployment import BackendConfig +from openadapt_flow.jit import ( + SURFACE_KINDS, + AdmissionError, + SurfaceApp, + TaskSignature, + WebApp, + compile_agent, +) +from tests.test_jit_lifecycle import ( # noqa: F401 - _policy_passes is autouse + REVIEWER, + IndependentVerifier, + _ci, + _jit, + _policy_passes, + _report, + _rerun, + _review, + _stub_runs, +) + +SECTIONS: dict[str, dict[str, Any]] = { + "windows": {"kind": "windows", "agent_url": "http://127.0.0.1:5001"}, + "macos": { + "kind": "macos", + "macos_app": "TextEdit", + "macos_window_title": "Visit note for Jane Sample", + }, + "linux": { + "kind": "linux", + "linux_app": "patient-notes", + "linux_window_title": "Visit note for Jane Sample", + }, + "rdp": {"kind": "rdp", "rdp_host": "10.0.0.5"}, + "citrix": { + "kind": "citrix", + "rdp_window": "Citrix Viewer", + "rdp_window_title": "Visit note for Jane Sample", + "rdp_readiness_text": "Patient Notes", + }, +} +ORIGINS = { + "windows": "windows://127.0.0.1:5001", + "macos": "macos://textedit", + "linux": "linux://patient-notes", + "rdp": "rdp://10.0.0.5", + "citrix": "citrix://citrix viewer", +} + + +def test_every_non_web_backend_kind_is_a_surface_app() -> None: + assert set(SURFACE_KINDS) == set(SECTIONS) + + +@pytest.mark.parametrize("kind", sorted(SECTIONS)) +def test_a_section_names_its_surface_without_window_titles(kind: str) -> None: + app = SurfaceApp(SECTIONS[kind]) + assert app.kind == kind + assert app.url is None and app.entry_url is None + assert app.origin == ORIGINS[kind] + # A window title can carry a patient or document name; it is never part + # of the origin a canary, an admission policy or a report records. + if "Jane" in str(SECTIONS[kind]): + assert "jane" not in app.origin + assert app.run_refusal() is None + + +def test_a_deployment_file_names_the_app(tmp_path: Path) -> None: + deploy = tmp_path / "deploy.yaml" + deploy.write_text("name: notes\nbackend:\n kind: macos\n macos_app: TextEdit\n") + app = SurfaceApp.from_deployment(deploy) + assert app.kind == "macos" and app.origin == "macos://textedit" + assert isinstance(app.backend, BackendConfig) + + +@pytest.mark.parametrize( + ("section", "reason"), + [ + ({"kind": "web", "url": "https://emr.example/"}, "use WebApp"), + ({"kind": "windows"}, "backend.agent_url"), + ({"kind": "macos"}, "backend.macos_app"), + ({"kind": "linux", "linux_app": "gedit"}, "linux_window_title"), + ({"kind": "rdp"}, "exactly one of backend.rdp_host or backend.rdp_window"), + ( + {"kind": "rdp", "rdp_host": "h", "rdp_window": "mstsc"}, + "exactly one of", + ), + ({"kind": "citrix", "rdp_host": "h"}, "local Workspace window"), + ({"kind": "vnc"}, "unknown backend.kind"), + ( + {"kind": "windows", "agent_url": "https://u:secret@vm.example:5001/"}, + "must not carry credentials", + ), + ], +) +def test_a_section_without_its_target_is_refused(section: dict, reason: str) -> None: + with pytest.raises(ValueError, match=reason): + SurfaceApp(section) + + +def test_the_playwright_mcp_dialect_is_refused_off_the_browser(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="CSS selectors in a browser page"): + compile_agent( + task=TaskSignature("t", "x", {}), + app=SurfaceApp(SECTIONS["citrix"]), + verifier=IndependentVerifier(), + policy="clinical-write", + cache_dir=tmp_path, + dialect="playwright_mcp", + ) + + +def test_open_builds_the_backend_the_section_names(monkeypatch) -> None: + import openadapt_flow.backends.factory as factory + + built: list[BackendConfig] = [] + + class Backend: + closed = False + + def close(self) -> None: + Backend.closed = True + + def build_backend(cfg: BackendConfig, **_: Any) -> Backend: + built.append(cfg) + return Backend() + + monkeypatch.setattr(factory, "build_backend", build_backend) + app = SurfaceApp(SECTIONS["linux"]) + backend, close = app.open() + assert isinstance(backend, Backend) and built == [app.backend] + close() + assert Backend.closed + + +def test_citrix_runs_need_a_readiness_marker_before_any_backend_opens( + tmp_path: Path, +) -> None: + opened: list = [] + section = dict(SECTIONS["citrix"], rdp_readiness_text=None) + app = SurfaceApp(section, open_backend=opened.append) + assert "rdp_readiness_text" in (app.run_refusal() or "") + jit = _rerun(_jit(tmp_path), app=app) + detail, report = jit._run_governed( + None, "digest", {}, tmp_path / "run", attended_by=REVIEWER + ) + assert report is None and "refused before any action" in (detail or "") + assert opened == [] + + +@pytest.mark.parametrize("kind", sorted(SECTIONS)) +def test_an_admitted_surface_program_runs_only_on_that_surface( + tmp_path: Path, monkeypatch, kind: str +) -> None: + app = SurfaceApp(SECTIONS[kind]) + jit = _rerun(_jit(tmp_path), app=app) + _review(jit) + _stub_runs(jit, monkeypatch, _report()) + canary = jit.canary(principal_id=REVIEWER) + assert canary.status == "VERIFIED" + manifest = jit.admit_with_key(_ci(jit, monkeypatch))["manifest"] + assert manifest["canary"]["target_origin"] == ORIGINS[kind] + + hit = jit.run() + assert hit.status == "VERIFIED" and hit.cache == "hit" and not hit.agent_called + + for other in ( + WebApp(url="https://emr.example/"), + SurfaceApp(SECTIONS["macos" if kind != "macos" else "linux"]), + ): + moved = _rerun(jit, app=other) + calls = _stub_runs(moved, monkeypatch, _report()) + refused = moved.run() + assert refused.status == "refused" + assert f"the canary ran on {ORIGINS[kind]}" in (refused.detail or "") + assert calls == [] + + +def test_a_draft_still_needs_a_person_on_a_surface(tmp_path: Path) -> None: + jit = _rerun(_jit(tmp_path), app=SurfaceApp(SECTIONS["windows"])) + with pytest.raises(AdmissionError, match="review the draft"): + jit.canary(principal_id="op:someone-else") + + +def test_the_same_kind_on_another_target_is_another_origin() -> None: + a = SurfaceApp(SECTIONS["rdp"]) + b = SurfaceApp(dict(SECTIONS["rdp"], rdp_host="10.0.0.6")) + c = SurfaceApp(dict(SECTIONS["rdp"], rdp_port=3390)) + assert len({a.origin, b.origin, c.origin}) == 3 + + +def test_an_admission_policy_names_a_surface_with_its_kind() -> None: + from types import SimpleNamespace + + from openadapt_flow.jit.authority import scope_problems + + def policy(*hosts: str) -> Any: + return SimpleNamespace( + allowed_hosts=hosts, + allowed_actions=(), + allowed_effects=(), + flow_policies=(), + ) + + empty = SimpleNamespace(steps=[]) + rdp = SurfaceApp(SECTIONS["rdp"]).origin + # A web host entry never covers the RDP desktop on the same address. + assert scope_problems(policy("10.0.0.5"), empty, {}, rdp) + assert scope_problems(policy("rdp://10.0.0.5"), empty, {}, rdp) == [] + # Nor does a surface entry cover the web app on that host. + assert scope_problems(policy("rdp://10.0.0.5"), empty, {}, "http://10.0.0.5") + assert scope_problems(policy("10.0.0.5"), empty, {}, "http://10.0.0.5") == [] + # The kind is part of the name: a Citrix window is not a macOS app. + citrix = SurfaceApp(SECTIONS["citrix"]).origin + assert scope_problems(policy("macos://citrix viewer"), empty, {}, citrix) + + +@pytest.mark.parametrize("kind", sorted(SECTIONS)) +def test_a_canary_and_a_run_drive_the_surface_through_the_replayer( + tmp_path: Path, monkeypatch, kind: str +) -> None: + """Everything but the gate and the replay itself is the real path.""" + + from types import SimpleNamespace + + import openadapt_flow.run_gate as run_gate + import openadapt_flow.runtime as runtime + + opened: list = [] + closed: list = [] + runs: list[dict[str, Any]] = [] + + class Backend: + def close(self) -> None: + closed.append(kind) + + def open_backend(cfg: BackendConfig) -> Backend: + opened.append(cfg.kind) + return Backend() + + class Replayer: + def __init__(self, backend: Any, **kwargs: Any) -> None: + assert isinstance(backend, Backend) + self.kwargs = kwargs + + def run(self, workflow: Any, **kwargs: Any) -> Any: + runs.append({**kwargs, "replayer": self.kwargs}) + return _report() + + monkeypatch.setattr( + run_gate, + "evaluate_run_gate", + lambda *a, **k: SimpleNamespace(passed=True, render=lambda: ""), + ) + monkeypatch.setattr(run_gate, "build_runtime_authorization", lambda *a, **k: "auth") + monkeypatch.setattr(runtime, "Replayer", Replayer) + + app = SurfaceApp(SECTIONS[kind], open_backend=open_backend) + jit = _rerun(_jit(tmp_path), app=app) + _review(jit) + assert jit.canary(principal_id=REVIEWER).status == "VERIFIED" + jit.admit_with_key(_ci(jit, monkeypatch)) + assert jit.run().status == "VERIFIED" + + assert opened == closed == [kind, kind] + for run in runs: + assert run["execution_target_kind"] == kind + assert run["execution_origin"] is None and run["execution_entry_url"] is None + assert run["execution_target_origin"] == ORIGINS[kind] + assert run["replayer"]["deployment_surface"] is None + assert runs[0]["replayer"]["attended_draft_by"] == REVIEWER + + +def test_the_replayer_matches_an_admitted_surface_program_by_its_origin( + tmp_path: Path, monkeypatch +) -> None: + from openadapt_flow.ir import Workflow + from openadapt_flow.jit.lifecycle import DraftNotAdmitted, ensure_runnable + from openadapt_flow.runtime.replayer import Replayer + from tests.test_replayer import FakeBackend + + app = SurfaceApp(SECTIONS["macos"]) + jit = _rerun(_jit(tmp_path), app=app) + _review(jit) + _stub_runs(jit, monkeypatch, _report()) + jit.canary(principal_id=REVIEWER) + jit.admit_with_key(_ci(jit, monkeypatch)) + workflow = Workflow.load(jit.bundle_dir) + + backend = FakeBackend() + with pytest.raises(DraftNotAdmitted, match="this run is on linux://patient-notes"): + Replayer( + backend, effect_verifier=IndependentVerifier(), draft_key_dir=jit.key_dir + ).run( + workflow, + bundle_dir=jit.bundle_dir, + run_dir=tmp_path / "other", + execution_target_kind="linux", + execution_target_origin=ORIGINS["linux"], + ) + assert backend.actions == [] + ensure_runnable( + jit.bundle_dir, + workflow=workflow, + key_dir=jit.key_dir, + execution_origin=ORIGINS["macos"], + effect_verifier=IndependentVerifier(), + at_run=True, + ) + + +def test_the_cli_names_a_desktop_run_for_the_admission_match() -> None: + from openadapt_flow.__main__ import _surface_target_origin_or_none + + assert _surface_target_origin_or_none(BackendConfig(kind="web")) is None + assert ( + _surface_target_origin_or_none(BackendConfig(**SECTIONS["macos"])) + == ORIGINS["macos"] + ) + leaky = BackendConfig(kind="windows", agent_url="https://u:p@vm:5001") + assert _surface_target_origin_or_none(leaky) is None diff --git a/packages/flow/tests/test_jit_surface_pixels.py b/packages/flow/tests/test_jit_surface_pixels.py new file mode 100644 index 000000000..02a389bb1 --- /dev/null +++ b/packages/flow/tests/test_jit_surface_pixels.py @@ -0,0 +1,176 @@ +"""compile_agent records an agent's run through the real RDP and Citrix backends. + +MockMed runs in a browser that Flow never reads: tests/surface_kiosk presents +the page to ``RemoteDisplayBackend`` (``rdp``, local client window) and +``CitrixWorkspaceBackend`` (``citrix``) as a remote-desktop client window, so +Flow sees only its pixels, as it would over RDP or Citrix. The agent replays +the recorded Anthropic trajectory used by tests/e2e/test_jit_cua_e2e.py. + +What it checks: + +* a cache miss records through the surface backend and compiles a DRAFT bound + to that surface, with the typed note lifted to a parameter and the save's + record-written effect mined from the system of record; +* the canary is refused by the standard run gate before any input reaches the + window, because a pixel-only click carries no verified record identity. + ``run --profile standard`` refuses the same bundle the same way; see + docs/BACKEND_PARITY.md. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Iterator + +import pytest + +pytest.importorskip("playwright.sync_api") + +from openadapt_flow.jit import SurfaceApp, TaskSignature, compile_agent # noqa: E402 +from openadapt_flow.jit.recorded import RecordedTrajectoryAgent # noqa: E402 +from openadapt_flow.mockmed.fault_server import serve # noqa: E402 +from openadapt_flow.runtime.effects import RestRecordVerifier # noqa: E402 +from openadapt_flow.tutorial import _http_json, _records # noqa: E402 +from tests.surface_kiosk import PageWindowClient # noqa: E402 + +FIXTURES = Path(__file__).resolve().parent / "fixtures" / "jit" +TASK = TaskSignature( + "mockmed.add_note", + "Open the first referral, start a Triage encounter, and save this note: {note}", + {"note": str}, +) +TITLE = "MockMed kiosk" +SECTIONS = { + "rdp": { + "kind": "rdp", + "rdp_window": "mstsc", + "rdp_window_title": TITLE, + "rdp_max_frame_age_s": 30.0, + }, + "citrix": { + "kind": "citrix", + "rdp_window": "Citrix Viewer", + "rdp_window_title": TITLE, + "rdp_readiness_text": "MockMed", + "rdp_max_frame_age_s": 30.0, + }, +} + + +@pytest.fixture(scope="module") +def mockmed() -> Iterator[str]: + base, _db, stop = serve() + yield base.rstrip("/") + stop() + + +@pytest.fixture(scope="module") +def browser() -> Iterator[Any]: + from playwright.sync_api import sync_playwright + + from openadapt_flow._browser_setup import launch_chromium + + with sync_playwright() as pw: + try: + chromium = launch_chromium(pw.chromium, headless=True) + except Exception as exc: # noqa: BLE001 - no browser on this host + pytest.skip(f"chromium unavailable: {exc}") + yield chromium + chromium.close() + + +def _opener(browser: Any, base: str, clients: list[PageWindowClient]): + """Build each backend through the factory, as ``SurfaceApp.open`` does, + with the page injected as the client window.""" + + from openadapt_flow.backends.factory import build_backend + + def open_backend(cfg: Any) -> Any: + page = browser.new_page( + viewport={"width": 1280, "height": 800}, device_scale_factor=1 + ) + page.goto(f"{base}/?fault=ok&idempotency=demo#tasks") + page.wait_for_timeout(300) + client = PageWindowClient(page, owner=cfg.rdp_window, title=TITLE) + clients.append(client) + backend = build_backend(cfg, window_client=client) + backend.close = page.close # type: ignore[attr-defined] + return backend + + return open_backend + + +@pytest.mark.timeout(300) +@pytest.mark.parametrize("kind", ["rdp", "citrix"]) +def test_an_agent_run_on_a_remote_window_compiles_and_fails_closed( + mockmed: str, browser: Any, tmp_path: Path, kind: str +) -> None: + clients: list[PageWindowClient] = [] + built: list[Any] = [] + opener = _opener(browser, mockmed, clients) + + def open_backend(cfg: Any) -> Any: + built.append(opener(cfg)) + return built[-1] + + app = SurfaceApp( + SECTIONS[kind], + record_reader=lambda: _records(mockmed), + reset=lambda: _http_json(f"{mockmed}/api/reset", method="POST", body={}), + open_backend=open_backend, + ) + agent = RecordedTrajectoryAgent.from_file(FIXTURES / "anthropic_add_note.json") + jit = compile_agent( + agent, + task=TASK, + app=app, + verifier=RestRecordVerifier( + mockmed, + records_path="/api/db", + records_key="records", + timeout_s=2.0, + poll_interval_s=0.05, + ), + policy="clinical-write", + cache_dir=tmp_path / "cache", + key_dir=tmp_path / "keys", + on_warning=lambda text: None, + ) + + miss = jit.run(note="Synthetic follow-up in two weeks") + assert miss.status == "needs_human_admit", miss.detail + assert agent.calls == 1 and len(clients) == 1 and clients[0].inputs > 0 + from openadapt_flow.backends.citrix_workspace import CitrixWorkspaceBackend + from openadapt_flow.backends.remote_display import RemoteDisplayBackend + + expected = CitrixWorkspaceBackend if kind == "citrix" else RemoteDisplayBackend + assert type(built[0]) is expected + candidate = miss.candidate + assert candidate is not None + assert [p["name"] for p in candidate["params"]["lifted"]] == ["note"] + assert [e["kind"] for e in candidate["effects"]] == ["record_written"] + + workflow = json.loads((jit.bundle_dir / "workflow.json").read_text()) + assert workflow["surface"] == kind + assert workflow["backend_hints"]["rdp_window_title"] == TITLE + meta = json.loads((Path(candidate["recording"]) / "meta.json").read_text()) + assert meta["app_url"] is None and meta["surface"] == kind + assert meta["demonstrated_by"] == "agent" + assert meta["authoring"] == {"input": "remote_client_window", "screen_only": True} + + confirmed = jit.confirm_params( + principal_id="Dana Reviewer", accept_params=["note"], confirm="ok" + ) + jit.review( + principal_id="Dana Reviewer", reviewed_steps=confirmed["steps"], confirm="ok" + ) + saved = _records(mockmed) + canary = jit.canary(principal_id="Dana Reviewer", note="Canary test note") + assert canary.status == "refused" + assert "Identity coverage" in (canary.detail or "") + assert "Nothing was executed" in (canary.detail or "") + # The gate refused before a backend opened: no second window, no input. + assert len(clients) == 1 + assert _records(mockmed) == saved + assert canary.inbox_item is not None and canary.inbox_item.is_file()