diff --git a/packages/flow/README.md b/packages/flow/README.md index 22bffb2a2..92c85e629 100644 --- a/packages/flow/README.md +++ b/packages/flow/README.md @@ -334,7 +334,12 @@ app and one that drops the save, which return results without opening a browser. `openadapt-flow emit-mcp bundle --out server.py` writes a standalone MCP -server for one bundle. It replays without the record check. +server for one bundle. It replays without the record check. A web bundle's +tool takes the app's URL. A bundle recorded on Windows, macOS, Linux, RDP or +Citrix takes none: start the server with `python server.py --config +deploy.yaml`. Each call then runs `openadapt-flow run bundle --config +deploy.yaml`, the governed path, so the file's app, record check, policy and +approvals all apply. ### From Claude Code diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index 50cb22aa4..b3216b68a 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -3631,6 +3631,9 @@ def _cmd_resume(args: argparse.Namespace) -> int: # the bundle's real substrate rather than always the browser. The default # (web / no flag) reproduces the historical Playwright path below exactly. backend_cfg = _resolve_backend_config(args, cfg, workflow) + refused = _enforce_surface_binding(args, workflow, backend_cfg, operation="resume") + if refused is not None: + return refused if _refuse_missing_citrix_readiness(backend_cfg, operation="Resume"): return 2 @@ -3671,6 +3674,7 @@ def _resume_with(backend: "Backend") -> "RunReport": replayer, key=ckpt_key, execution_target_kind=_report_backend_kind(backend_cfg.kind), + surface_override=bool(getattr(args, "_surface_override", False)), ) try: @@ -4072,9 +4076,25 @@ def _cmd_bench(args: argparse.Namespace) -> int: from contextlib import contextmanager from openadapt_flow.bench import run_bench + from openadapt_flow.emit.skill import bundle_surface + from openadapt_flow.ir import Workflow from openadapt_flow.mockmed.server import serve from openadapt_flow.report import render_bench_report + try: + surface = bundle_surface(Workflow.load(Path(args.bundle))) + except Exception: # noqa: BLE001 - run_bench reports load errors itself + surface = None + if surface not in (None, "web"): + from openadapt_flow.cli_hints import command + + print( + "BENCH REFUSED: bench replays a bundle against the bundled MockMed " + f"web app, and this bundle was recorded on {surface!r}. Measure it " + "on its own app with repeated " + f"`{command('run BUNDLE --config deploy.yaml')}`. Nothing was executed." + ) + return 2 url, stop = serve(port=0) target_url = _with_drift(url, args.drift) diff --git a/packages/flow/openadapt_flow/emit/mcp_tool.py b/packages/flow/openadapt_flow/emit/mcp_tool.py index 66e9b1858..802e433a8 100644 --- a/packages/flow/openadapt_flow/emit/mcp_tool.py +++ b/packages/flow/openadapt_flow/emit/mcp_tool.py @@ -18,7 +18,7 @@ import shutil from pathlib import Path -from openadapt_flow.emit.skill import declared_params +from openadapt_flow.emit.skill import bundle_surface, declared_params from openadapt_flow.ir import Workflow _BUNDLE_SUBDIR = "bundle" @@ -59,6 +59,7 @@ def {func_name}(url: str{param_sig}) -> dict: is RECONCILIATION_REQUIRED a write may have landed, so check the record before you call this tool again. """ + params = {params_dict} from playwright.sync_api import sync_playwright from openadapt_flow._browser_setup import launch_chromium @@ -68,7 +69,6 @@ def {func_name}(url: str{param_sig}) -> dict: from openadapt_flow.runtime import Replayer workflow = Workflow.load(BUNDLE_DIR) - params = {params_dict} run_dir = Path(tempfile.mkdtemp(prefix="openadapt_flow_run_")) with sync_playwright() as p: browser = launch_chromium(p.chromium, headless=True) @@ -103,6 +103,124 @@ def {func_name}(url: str{param_sig}) -> dict: mcp.run() ''' +_SURFACE_SERVER_TEMPLATE = '''\ +"""MCP server exposing the {name!r} workflow as a tool. + +Generated by openadapt-flow. Requires: mcp, openadapt-flow. +The workflow was recorded on the {surface!r} surface. The deployment file +named at start names the app (its ``backend:`` section) and the rest of the +run (record check, policy, approvals), so no URL is passed. Each call runs +``openadapt-flow run bundle --config ``, the governed path. +The workflow bundle lives alongside this file in ./{bundle_subdir}/. +Run with: python {out_name} --config deploy.yaml +(or set OPENADAPT_DEPLOYMENT=deploy.yaml) +""" + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +import tempfile +from pathlib import Path + +from mcp.server.fastmcp import FastMCP + +BUNDLE_DIR = Path(__file__).resolve().parent / {bundle_subdir!r} +SURFACE = {surface!r} +DEPLOYMENT = os.environ.get("OPENADAPT_DEPLOYMENT") + +mcp = FastMCP({name!r}) + + +def _refused(detail: str) -> dict: + return {{ + "success": False, + "outcome": "REFUSED", + "transaction_outcome": None, + "production_eligible": False, + "detail": detail, + }} + + +def _run_governed(params: dict) -> dict: + if not DEPLOYMENT: + return _refused( + f"this server drives a {{SURFACE}} app named by a deployment file; " + "start it with --config deploy.yaml or set OPENADAPT_DEPLOYMENT. " + "Nothing was executed." + ) + work = Path(tempfile.mkdtemp(prefix="openadapt_flow_run_")) + run_dir = work / "run" + params_file = work / "params.json" + fd = os.open(params_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "w") as fh: + json.dump(params, fh) + try: + proc = subprocess.run( + [ + sys.executable, "-m", "openadapt_flow", "run", str(BUNDLE_DIR), + "--config", str(DEPLOYMENT), "--run-dir", str(run_dir), + "--params-file", str(params_file), + ], + capture_output=True, + text=True, + errors="replace", + ) + finally: + params_file.unlink(missing_ok=True) + report_path = run_dir / "report.json" + if not report_path.is_file(): + lines = [line for line in proc.stdout.splitlines() if "REFUSED" in line] + return _refused( + (lines[-1] if lines else f"openadapt-flow run exited {{proc.returncode}}")[:500] + ) + report = json.loads(report_path.read_text()) + return {{ + "success": ( + report.get("execution_outcome") == "VERIFIED" + and report.get("transaction_outcome") == "VERIFIED" + ), + "outcome": report.get("execution_outcome"), + "transaction_outcome": report.get("transaction_outcome"), + "production_eligible": bool(report.get("production_eligible")), + "run_dir": str(run_dir), + "heal_count": report.get("heal_count"), + "model_calls": report.get("model_calls"), + "total_ms": report.get("total_ms"), + }} + + +@mcp.tool() +def {func_name}({param_sig_first}) -> dict: + """{docstring} + + Args: +{param_docs} + Returns: + Dict with the run's evidence-qualified outcome. ``success`` is true + only when ``outcome`` and ``transaction_outcome`` are both VERIFIED: + the change was saved and read back from the system of record. COMPLETED_UNVERIFIED means the steps ran + but the saved result was not confirmed. When ``transaction_outcome`` + is RECONCILIATION_REQUIRED a write may have landed, so check the + record before you call this tool again. ``REFUSED`` means nothing ran. + """ + return _run_governed({params_dict}) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--config", + default=DEPLOYMENT, + help="Deployment file whose backend: section names the app", + ) + DEPLOYMENT = parser.parse_args().config + mcp.run() +''' + def _identifier(name: str) -> str: """Coerce an arbitrary name into a valid Python identifier.""" @@ -122,9 +240,12 @@ def _clean_line(text: str) -> str: def emit_mcp_server(bundle_dir: Path | str, out_path: Path | str) -> Path: """Generate a standalone, portable FastMCP ``server.py`` for the bundle. - The generated module defines one tool named after the workflow, with a - required ``url`` argument plus one required ``str`` argument per - non-secret workflow parameter. A recorded example value is never a + The generated module defines one tool named after the workflow, with + one required ``str`` argument per non-secret workflow parameter. A web + bundle's tool also takes the app's ``url``. A bundle recorded on another + surface (Windows, macOS, Linux, RDP, Citrix) takes no URL: the server is + started with ``--config deploy.yaml`` and drives the app that file's + ``backend:`` section names, refusing one that names another surface. A recorded example value is never a default: it would leak the recorded value into the generated source and write it to the wrong record when a caller omits the argument. The source is validated with :func:`ast.parse` before being written. The workflow @@ -182,16 +303,30 @@ def emit_mcp_server(bundle_dir: Path | str, out_path: Path | str) -> Path: "reviews, canaries and admits the bundle (openadapt-flow jit). " + docstring ) - source = _SERVER_TEMPLATE.format( - name=workflow.name, - out_name=out.name, - bundle_subdir=bundle_subdir, - func_name=func_name, - param_sig=param_sig, - param_docs=param_docs, - params_dict=params_dict, - docstring=docstring, - ) + surface = bundle_surface(workflow) + if surface == "web": + source = _SERVER_TEMPLATE.format( + name=workflow.name, + out_name=out.name, + bundle_subdir=bundle_subdir, + func_name=func_name, + param_sig=param_sig, + param_docs=param_docs, + params_dict=params_dict, + docstring=docstring, + ) + else: + source = _SURFACE_SERVER_TEMPLATE.format( + name=workflow.name, + out_name=out.name, + bundle_subdir=bundle_subdir, + surface=surface, + func_name=func_name, + param_sig_first=param_sig.removeprefix(", "), + param_docs=param_docs or " (none)\n", + params_dict=params_dict, + docstring=docstring, + ) # Guarantee the emitted source is syntactically valid Python. ast.parse(source) diff --git a/packages/flow/openadapt_flow/emit/skill.py b/packages/flow/openadapt_flow/emit/skill.py index 51edacbe5..d596c9290 100644 --- a/packages/flow/openadapt_flow/emit/skill.py +++ b/packages/flow/openadapt_flow/emit/skill.py @@ -32,6 +32,20 @@ def _slugify(name: str) -> str: return slug or "workflow" +def bundle_surface(workflow: Workflow) -> str: + """The surface a bundle was recorded on: ``web`` unless it names another.""" + + surface = getattr(workflow, "surface", None) + surfaces = getattr(workflow, "surfaces", None) or [] + if surface is None and surfaces: + # A multi-surface bundle starts on its first (primary) surface. + surface = getattr(surfaces[0], "kind", None) + hints = getattr(workflow, "backend_hints", None) + if surface is None and hints is not None: + surface = getattr(hints, "backend", None) + return str(surface or "web") + + def declared_params(workflow: Workflow) -> list[str]: """Names of the workflow's non-secret parameters, in declaration order.""" secret = set(workflow.secret_params) | { @@ -168,7 +182,9 @@ def emit_skill(bundle_dir: Path | str, out_dir: Path | str) -> Path: # request, and the governed ``run`` path (not the permissive ``replay``) # is the one an agent should call. param_flags = " ".join(f"--param {name}=<{name}>" for name in params) - invocation = f"openadapt-flow run {_BUNDLE_SUBDIR} --url " + surface = bundle_surface(workflow) + target = "--url " if surface == "web" else "--config " + invocation = f"openadapt-flow run {_BUNDLE_SUBDIR} {target}" if param_flags: invocation += f" {param_flags}" @@ -235,9 +251,17 @@ def emit_skill(bundle_dir: Path | str, out_dir: Path | str) -> Path: usage_note = ( f"Run the command from this skill folder (the `{_BUNDLE_SUBDIR}` " f"path is relative to it), or substitute the absolute path of " - f"`{_BUNDLE_SUBDIR}/`. Replace `` with the URL of the " - f"running target app. " + f"`{_BUNDLE_SUBDIR}/`. " ) + if surface == "web": + usage_note += "Replace `` with the URL of the running target app. " + else: + usage_note += ( + f"This workflow was recorded on the `{surface}` surface. Replace " + "`` with the deployment file whose `backend:` " + f"section names the {surface} app; the operator who set up the " + "app provides it. Don't pass `--url`. " + ) if params: usage_note += ( "Every parameter is required. Replace each `` placeholder " diff --git a/packages/flow/openadapt_flow/execute/dispatch.py b/packages/flow/openadapt_flow/execute/dispatch.py index 628534d4d..1a11efdda 100644 --- a/packages/flow/openadapt_flow/execute/dispatch.py +++ b/packages/flow/openadapt_flow/execute/dispatch.py @@ -145,6 +145,17 @@ def live_replay( f"admitted bundle could not be loaded: {type(exc).__name__}", pre_effect=True, ) from exc + from openadapt_flow.emit.skill import bundle_surface + + surface = bundle_surface(workflow) + if surface != "web": + raise DispatchError( + "the reference Execute worker drives a browser only, and this " + f"bundle was recorded on {surface!r}; run it with `openadapt-flow " + "run BUNDLE --config deploy.yaml` or through `openadapt-flow " + "connector run`. Nothing was executed.", + pre_effect=True, + ) stop: Optional[Callable[[], None]] = None try: if admission.target_url: diff --git a/packages/flow/openadapt_flow/runtime/durable/attended_service.py b/packages/flow/openadapt_flow/runtime/durable/attended_service.py index abf0d7e9f..e20be7c7e 100644 --- a/packages/flow/openadapt_flow/runtime/durable/attended_service.py +++ b/packages/flow/openadapt_flow/runtime/durable/attended_service.py @@ -384,6 +384,13 @@ def __init__( "remove those keys to let the attended service open its own " "visible browser. Nothing was executed." ) + kind = str(backend.kind or "").strip().lower() + if kind == "citrix" and not str(backend.rdp_readiness_text or "").strip(): + raise ValueError( + "attended Citrix actions require a stable current-frame " + "readiness marker before any action: set " + "backend.rdp_readiness_text. Nothing was executed." + ) self._deployment = deployment.model_copy(deep=True) self._key = key self._owner = _ThreadOwnedAttendedExecutor( diff --git a/packages/flow/openadapt_flow/runtime/durable/resume.py b/packages/flow/openadapt_flow/runtime/durable/resume.py index 3dbb8f7b8..75ac08ab7 100644 --- a/packages/flow/openadapt_flow/runtime/durable/resume.py +++ b/packages/flow/openadapt_flow/runtime/durable/resume.py @@ -696,6 +696,7 @@ def resume( worklists: Optional[dict[str, list[dict[str, str]]]] = None, save_healed_to: Optional[Path | str] = None, execution_target_kind: Optional[ExecutionTargetKind] = None, + surface_override: bool = False, now: Optional[datetime] = None, key: Optional[str] = None, ) -> RunReport: @@ -743,6 +744,7 @@ def resume( worklists=worklists, save_healed_to=save_healed_to, execution_target_kind=execution_target_kind, + surface_override=surface_override, now=now, key=resolved_key, ) @@ -760,6 +762,7 @@ def _resume_under_lease( worklists: Optional[dict[str, list[dict[str, str]]]] = None, save_healed_to: Optional[Path | str] = None, execution_target_kind: Optional[ExecutionTargetKind] = None, + surface_override: bool = False, now: Optional[datetime] = None, key: Optional[str] = None, ) -> RunReport: @@ -999,6 +1002,7 @@ def _resume_under_lease( live_bundle_version=live_bundle_version, run_id=(manifest.run_id if manifest is not None else None), execution_target_kind=execution_target_kind, + surface_override=surface_override, manifest=manifest, pending=pending, approval=approved, @@ -1091,6 +1095,7 @@ def _resume_under_lease( run_id=effective_run_id, idempotency_key=(manifest.idempotency_key if manifest is not None else None), execution_target_kind=execution_target_kind, + surface_override=surface_override, prior_screenshots_may_leave_box=( manifest.screenshots_may_leave_box if manifest is not None else False ), @@ -1115,6 +1120,7 @@ def _resume_program( live_bundle_version: str, run_id: str, execution_target_kind: Optional[ExecutionTargetKind], + surface_override: bool, manifest: RunManifest, pending: PendingEscalation, approval: ApprovalRecord, @@ -1237,6 +1243,7 @@ def _resume_program( run_id=effective_run_id, idempotency_key=manifest.idempotency_key, execution_target_kind=execution_target_kind, + surface_override=surface_override, prior_screenshots_may_leave_box=( manifest.screenshots_may_leave_box if manifest is not None else False ), diff --git a/packages/flow/tests/test_backend_parity_refusals.py b/packages/flow/tests/test_backend_parity_refusals.py new file mode 100644 index 000000000..758b00306 --- /dev/null +++ b/packages/flow/tests/test_backend_parity_refusals.py @@ -0,0 +1,262 @@ +"""Commands that used to assume a browser now follow the bundle's surface. + +``emit-mcp`` and ``emit-skill`` name the app with a deployment file for a +bundle recorded off the browser. ``serve-execute``, ``bench`` and the attended +console refuse what they can't drive, with the reason, before any action. See +docs/BACKEND_PARITY.md. +""" + +from __future__ import annotations + +import importlib.util +import shlex +import sys +import types +from pathlib import Path +from typing import Any + +import pytest + +from openadapt_flow.emit import emit_mcp_server, emit_skill +from openadapt_flow.emit.skill import bundle_surface +from openadapt_flow.ir import ActionKind, BackendHints, Step, Workflow + + +def _bundle(tmp_path: Path, surface: str | None) -> Path: + bundle = tmp_path / "bundle" + Workflow( + name="Triage Note", + surface=surface, + params={"note": "x"}, + steps=[Step(id="s0", intent="type note", action=ActionKind.TYPE, param="note")], + ).save(bundle) + return bundle + + +def test_the_bundle_surface_is_web_unless_it_names_another() -> None: + assert bundle_surface(Workflow(name="w", steps=[])) == "web" + assert bundle_surface(Workflow(name="w", steps=[], surface="linux")) == "linux" + hinted = Workflow(name="w", steps=[], backend_hints=BackendHints(backend="citrix")) + assert bundle_surface(hinted) == "citrix" + + +# -- emit-skill ----------------------------------------------------------------- + + +@pytest.mark.parametrize("surface", ["windows", "macos", "linux", "rdp", "citrix"]) +def test_a_desktop_skill_names_a_deployment_file_not_a_url( + tmp_path: Path, surface: str +) -> None: + from openadapt_flow.__main__ import build_parser + + out = emit_skill(_bundle(tmp_path, surface), tmp_path / "skill") + text = (out / "SKILL.md").read_text() + assert "--url" not in text.replace("Don't pass `--url`", "") + assert f"recorded on the `{surface}` surface" in text + line = next( + line for line in text.splitlines() if line.startswith("openadapt-flow run") + ) + args = build_parser().parse_args(shlex.split(line)[1:]) + assert args.command == "run" and args.config == "" + + +def test_a_web_skill_still_takes_the_url(tmp_path: Path) -> None: + out = emit_skill(_bundle(tmp_path, None), tmp_path / "skill") + assert "--url " in (out / "SKILL.md").read_text() + + +# -- emit-mcp ------------------------------------------------------------------- + + +def _load_server( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, surface: str +) -> types.ModuleType: + fastmcp = types.ModuleType("mcp.server.fastmcp") + + class _FakeFastMCP: + def __init__(self, name: str) -> None: + self.name = name + + def tool(self): # noqa: ANN202 - FastMCP's decorator factory + return lambda func: func + + fastmcp.FastMCP = _FakeFastMCP # type: ignore[attr-defined] + for name in ("mcp", "mcp.server"): + monkeypatch.setitem(sys.modules, name, types.ModuleType(name)) + monkeypatch.setitem(sys.modules, "mcp.server.fastmcp", fastmcp) + monkeypatch.delenv("OPENADAPT_DEPLOYMENT", raising=False) + out = emit_mcp_server(_bundle(tmp_path, surface), tmp_path / "mcp" / "server.py") + spec = importlib.util.spec_from_file_location(f"emitted_{surface}", out) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_a_desktop_mcp_tool_takes_no_url(tmp_path: Path, monkeypatch) -> None: + import inspect + + server = _load_server(tmp_path, monkeypatch, "macos") + assert list(inspect.signature(server.run_triage_note).parameters) == ["note"] + assert "playwright" not in (tmp_path / "mcp" / "server.py").read_text() + + +def test_a_desktop_mcp_tool_refuses_without_a_deployment( + tmp_path: Path, monkeypatch +) -> None: + import subprocess + + monkeypatch.setattr( + subprocess, "run", lambda *a, **k: pytest.fail("nothing may run") + ) + server = _load_server(tmp_path, monkeypatch, "macos") + out = server.run_triage_note(note="synthetic") + assert out["success"] is False and out["outcome"] == "REFUSED" + assert "--config deploy.yaml" in out["detail"] + + +def test_a_desktop_mcp_tool_runs_the_governed_cli_with_every_param_intact( + tmp_path: Path, monkeypatch +) -> None: + """A parameter named like a local (``kind``) reaches the run unchanged.""" + + import json + import subprocess + + bundle = tmp_path / "bundle" + Workflow( + name="Triage Note", + surface="citrix", + params={"kind": "x", "section": "y", "workflow": "z"}, + steps=[ + Step(id="s0", intent="type", action=ActionKind.TYPE, param="kind"), + Step(id="s1", intent="type", action=ActionKind.TYPE, param="section"), + Step(id="s2", intent="type", action=ActionKind.TYPE, param="workflow"), + ], + ).save(bundle) + seen: dict[str, Any] = {} + + def run(argv: list[str], **kwargs: Any) -> Any: + args = dict(zip(argv[::1], argv[1:])) + seen["argv"] = argv + seen["params"] = json.loads(Path(args["--params-file"]).read_text()) + run_dir = Path(args["--run-dir"]) + run_dir.mkdir(parents=True) + (run_dir / "report.json").write_text( + json.dumps( + { + "execution_outcome": "VERIFIED", + "transaction_outcome": "VERIFIED", + "production_eligible": True, + "model_calls": 0, + } + ) + ) + return types.SimpleNamespace(returncode=0, stdout="", stderr="") + + fastmcp = types.ModuleType("mcp.server.fastmcp") + fastmcp.FastMCP = lambda name: types.SimpleNamespace( # type: ignore[attr-defined] + tool=lambda: lambda func: func + ) + for name in ("mcp", "mcp.server"): + monkeypatch.setitem(sys.modules, name, types.ModuleType(name)) + monkeypatch.setitem(sys.modules, "mcp.server.fastmcp", fastmcp) + out_path = emit_mcp_server(bundle, tmp_path / "mcp" / "server.py") + spec = importlib.util.spec_from_file_location("emitted_params", out_path) + assert spec is not None and spec.loader is not None + server = importlib.util.module_from_spec(spec) + spec.loader.exec_module(server) + monkeypatch.setattr(subprocess, "run", run) + server.DEPLOYMENT = str(tmp_path / "deploy.yaml") + + out = server.run_triage_note(kind="REAL", section="S", workflow="W") + assert out["success"] is True and out["outcome"] == "VERIFIED" + assert seen["params"] == {"kind": "REAL", "section": "S", "workflow": "W"} + assert seen["argv"][1:5] == [ + "-m", + "openadapt_flow", + "run", + str(out_path.parent / "bundle"), + ] + assert seen["argv"][seen["argv"].index("--config") + 1] == server.DEPLOYMENT + # The params file holds entered values; it is gone after the run. + assert not Path(dict(zip(seen["argv"], seen["argv"][1:]))["--params-file"]).exists() + + +def test_a_desktop_mcp_tool_reports_the_run_refusal( + tmp_path: Path, monkeypatch +) -> None: + """End to end: the real ``openadapt-flow run`` refuses a deployment that + names another surface, and the tool says so. Nothing is built.""" + + import os + + server = _load_server(tmp_path, monkeypatch, "macos") + deploy = tmp_path / "deploy.yaml" + deploy.write_text( + "backend:\n kind: linux\n linux_app: a\n linux_window_title: b\n" + ) + server.DEPLOYMENT = str(deploy) + root = str(Path(__file__).resolve().parents[1]) + monkeypatch.setenv( + "PYTHONPATH", + os.pathsep.join(p for p in (root, os.environ.get("PYTHONPATH")) if p), + ) + out = server.run_triage_note(note="synthetic") + assert out["success"] is False and out["outcome"] == "REFUSED" + assert "REFUSED" in out["detail"] and "'macos'" in out["detail"] + + +def test_a_multi_surface_bundle_is_named_by_its_primary_surface() -> None: + from openadapt_flow.ir import SurfaceSpec + + workflow = Workflow( + name="w", steps=[], surfaces=[SurfaceSpec(id="primary", kind="windows")] + ) + assert bundle_surface(workflow) == "windows" + + +# -- serve-execute, bench, attended console -------------------------------------- + + +@pytest.mark.parametrize("surface", ["windows", "macos", "linux", "rdp", "citrix"]) +def test_the_execute_worker_refuses_a_desktop_bundle_before_any_effect( + tmp_path: Path, surface: str +) -> None: + from openadapt_flow.execute.dispatch import DispatchError, live_replay + from openadapt_flow.execute.models import AdmittedBundle + + admission = AdmittedBundle( + qualification_id="q", + workflow_version="1", + workflow_digest="sha256:" + "0" * 64, + environment_id="env", + minimum_effect_strength="independent_system_of_record", + bundle_dir=str(_bundle(tmp_path, surface)), + ) + with pytest.raises(DispatchError, match=f"recorded on '{surface}'") as caught: + live_replay(admission, None, tmp_path / "run") # type: ignore[arg-type] + assert caught.value.pre_effect is True + + +def test_bench_refuses_a_desktop_bundle(tmp_path: Path, capsys) -> None: + from openadapt_flow.__main__ import main + + rc = main(["bench", str(_bundle(tmp_path, "linux")), "--run-root", str(tmp_path)]) + assert rc == 2 + out = capsys.readouterr().out + assert "BENCH REFUSED" in out and "recorded on 'linux'" in out + + +def test_attended_citrix_actions_need_a_readiness_marker() -> None: + from openadapt_flow.deployment import DeploymentConfig + from openadapt_flow.runtime.durable.attended_service import ( + AttendedActionService, + ) + + with pytest.raises(ValueError, match="rdp_readiness_text"): + AttendedActionService( + DeploymentConfig.model_validate( + {"backend": {"kind": "citrix", "rdp_window": "wfica32"}} + ) + ) diff --git a/packages/flow/tests/test_cli_new_commands.py b/packages/flow/tests/test_cli_new_commands.py index 31b62cef6..4979de390 100644 --- a/packages/flow/tests/test_cli_new_commands.py +++ b/packages/flow/tests/test_cli_new_commands.py @@ -179,6 +179,7 @@ def _paused_run( verified_first: bool = True, backend_hints: BackendHints | None = None, managed: bool = False, + surface: str | None = None, ) -> Path: run = tmp_path / "run" bundle = tmp_path / "b" @@ -186,6 +187,7 @@ def _paused_run( workflow = Workflow( name="w", backend_hints=backend_hints, + surface=surface, steps=[ Step(id="s0", intent="first", action=ActionKind.KEY, key="A"), Step(id="s1", intent="second", action=ActionKind.KEY, key="B"), @@ -566,7 +568,7 @@ def test_resume_web_default_builds_playwright_backend( seen: dict = {} - def _fake_resume(run_dir, replayer, key=None, execution_target_kind=None): + def _fake_resume(run_dir, replayer, key=None, execution_target_kind=None, **_kw): seen["called"] = True seen["execution_target_kind"] = execution_target_kind return _FakeReport() @@ -621,7 +623,9 @@ def _grounder(*, fallback=None): monkeypatch.setattr( durable_mod, "resume", - lambda run_dir, replayer, key=None, execution_target_kind=None: _FakeReport(), + lambda run_dir, replayer, key=None, execution_target_kind=None, **_kw: ( + _FakeReport() + ), ) rc = main( @@ -711,7 +715,7 @@ def __init__(self, backend, **kwargs): monkeypatch.setattr( durable_mod, "resume", - lambda run_dir, replayer, key=None, execution_target_kind=None: ( + lambda run_dir, replayer, key=None, execution_target_kind=None, **_kw: ( captured.update(execution_target_kind=execution_target_kind) or _FakeReport() ), @@ -774,7 +778,7 @@ def fake_build(cfg, **kwargs): monkeypatch.setattr( durable_mod, "resume", - lambda run_dir, replayer, key=None, execution_target_kind=None: ( + lambda run_dir, replayer, key=None, execution_target_kind=None, **_kw: ( captured.update(execution_target_kind=execution_target_kind) or _FakeReport() ), @@ -854,3 +858,74 @@ def test_version_flag_prints_version_and_exits_zero(capsys) -> None: assert out == f"openadapt-flow {_package_version()}" prog, _, ver = out.partition(" ") assert ver and ver[0].isdigit() + + +def test_resume_refuses_another_surface_than_the_bundle_was_recorded_on( + tmp_path: Path, monkeypatch, capsys +) -> None: + run = _paused_run(tmp_path, surface="macos") + main(["approve", str(run)]) + + import openadapt_flow.backends.factory as factory + + monkeypatch.setattr( + factory, + "build_backend", + lambda *args, **kwargs: pytest.fail("backend must not be constructed"), + ) + rc = main( + [ + "resume", + str(run), + "--require-approval", + "--backend", + "windows", + "--agent-url", + "http://127.0.0.1:5001", + ] + ) + assert rc == 2 + out = capsys.readouterr().out + assert "resume REFUSED" in out and "bound to surface 'macos'" in out + assert "Nothing was executed" in out + + +def test_resume_with_a_surface_override_records_it(tmp_path: Path, monkeypatch) -> None: + run = _paused_run(tmp_path, surface="macos") + main(["approve", str(run)]) + + import openadapt_flow.backends.factory as factory + import openadapt_flow.report as report_mod + import openadapt_flow.runtime as runtime_mod + import openadapt_flow.runtime.durable as durable_mod + + seen: dict = {} + + class FakeBackend: + def close(self): + pass + + monkeypatch.setattr(factory, "build_backend", lambda cfg, **kw: FakeBackend()) + monkeypatch.setattr(runtime_mod, "Replayer", lambda backend, **kw: object()) + monkeypatch.setattr(report_mod, "render_run_report", lambda run_dir: "REPORT.md") + + def _fake_resume(run_dir, replayer, key=None, **kw): + seen.update(kw) + return _FakeReport() + + monkeypatch.setattr(durable_mod, "resume", _fake_resume) + rc = main( + [ + "resume", + str(run), + "--require-approval", + "--backend", + "windows", + "--agent-url", + "http://127.0.0.1:5001", + "--allow-surface-override", + ] + ) + assert rc == 0 + assert seen["execution_target_kind"] == "windows" + assert seen["surface_override"] is True