From fa48cb33606e93596b7b42778b8d4dfadb2a9d37 Mon Sep 17 00:00:00 2001 From: abrichr Date: Sun, 18 Oct 2026 14:50:49 -0400 Subject: [PATCH 1/3] fix(flow): emit, serve-execute, bench and resume stop assuming a browser - emit-mcp: a bundle recorded off the browser gets a server that takes no URL. Started with --config deploy.yaml (or OPENADAPT_DEPLOYMENT), it drives the app the backend: section names and refuses one that names another surface. - emit-skill: the SKILL.md for such a bundle runs it with --config, not --url. - serve-execute refuses a non-web bundle before any effect, and bench refuses one instead of replaying it against the web demo. - The attended console refuses Citrix actions without backend.rdp_readiness_text, as run and resume already do. - resume refuses a backend other than the bundle's recorded surface unless --allow-surface-override is passed, as run already does. Co-Authored-By: Claude Opus 5.5 --- packages/flow/README.md | 5 +- packages/flow/openadapt_flow/__main__.py | 17 ++ packages/flow/openadapt_flow/emit/mcp_tool.py | 158 +++++++++++-- packages/flow/openadapt_flow/emit/skill.py | 26 ++- .../flow/openadapt_flow/execute/dispatch.py | 11 + .../runtime/durable/attended_service.py | 7 + .../tests/test_backend_parity_refusals.py | 214 ++++++++++++++++++ packages/flow/tests/test_cli_new_commands.py | 32 +++ 8 files changed, 452 insertions(+), 18 deletions(-) create mode 100644 packages/flow/tests/test_backend_parity_refusals.py diff --git a/packages/flow/README.md b/packages/flow/README.md index 6a9f8a792a..bf2aca9b98 100644 --- a/packages/flow/README.md +++ b/packages/flow/README.md @@ -333,7 +333,10 @@ 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`, and it drives the app that file's `backend:` section names. ### From Claude Code diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index 3effbf62de..7eda2fbfbe 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -3429,6 +3429,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 @@ -3870,9 +3873,23 @@ 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"): + 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 `openadapt-flow 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 66e9b1858f..7f0e238abd 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" @@ -103,6 +103,119 @@ 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), so no URL is passed. +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 os +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 + " Nothing was executed.", + }} + + +@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. + """ + from openadapt_flow.backends.factory import build_backend + from openadapt_flow.deployment import load_deployment + from openadapt_flow.ir import Workflow + from openadapt_flow.runtime import Replayer + + 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." + ) + section = load_deployment(DEPLOYMENT).backend + kind = (section.kind or "").strip().lower().replace("_", "-") + kind = "rdp" if kind == "remote-display" else kind + if kind != SURFACE: + return _refused( + f"the workflow was recorded on {{SURFACE}} and the deployment " + f"file names {{kind!r}}." + ) + workflow = Workflow.load(BUNDLE_DIR) + params = {params_dict} + run_dir = Path(tempfile.mkdtemp(prefix="openadapt_flow_run_")) + backend = build_backend(section) + try: + report = Replayer( + backend, deployment_surface=section.deployment_surface + ).run( + workflow, + params=params, + bundle_dir=BUNDLE_DIR, + run_dir=run_dir, + execution_target_kind=SURFACE, + ) + finally: + close = getattr(backend, "close", None) + if callable(close): + close() + return {{ + "success": ( + report.execution_outcome == "VERIFIED" + and report.transaction_outcome == "VERIFIED" + ), + "outcome": report.execution_outcome, + "transaction_outcome": report.transaction_outcome, + "production_eligible": report.production_eligible, + "run_dir": str(run_dir), + "heal_count": report.heal_count, + "model_calls": report.model_calls, + "total_ms": report.total_ms, + }} + + +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 +235,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 +298,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 51edacbe58..a3b9c33718 100644 --- a/packages/flow/openadapt_flow/emit/skill.py +++ b/packages/flow/openadapt_flow/emit/skill.py @@ -32,6 +32,16 @@ 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) + 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 +178,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 +247,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 d269b53862..89271f1bf0 100644 --- a/packages/flow/openadapt_flow/execute/dispatch.py +++ b/packages/flow/openadapt_flow/execute/dispatch.py @@ -144,6 +144,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 abf0d7e9f9..e20be7c7e8 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/tests/test_backend_parity_refusals.py b/packages/flow/tests/test_backend_parity_refusals.py new file mode 100644 index 0000000000..e978ff1864 --- /dev/null +++ b/packages/flow/tests/test_backend_parity_refusals.py @@ -0,0 +1,214 @@ +"""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, RunReport, 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_or_with_the_wrong_deployment( + tmp_path: Path, monkeypatch +) -> None: + import openadapt_flow.backends.factory as factory + + monkeypatch.setattr( + factory, "build_backend", lambda *a, **k: pytest.fail("must not build") + ) + 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"] + + 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) + out = server.run_triage_note(note="synthetic") + assert out["outcome"] == "REFUSED" + assert "recorded on macos" in out["detail"] and "'linux'" in out["detail"] + + +def test_a_desktop_mcp_tool_drives_the_app_its_deployment_names( + tmp_path: Path, monkeypatch +) -> None: + import openadapt_flow.backends.factory as factory + from openadapt_flow.runtime import Replayer + + seen: dict[str, Any] = {} + + class Backend: + def close(self) -> None: + seen["closed"] = True + + def build_backend(cfg: Any, **_: Any) -> Backend: + seen["cfg"] = cfg + return Backend() + + def run(self: Any, workflow: Workflow, **kwargs: Any) -> RunReport: + seen["kind"] = kwargs["execution_target_kind"] + seen["params"] = kwargs["params"] + return RunReport( + workflow_name=workflow.name, + started_at="2026-10-18T00:00:00Z", + success=True, + execution_profile="standard", + execution_outcome="VERIFIED", + transaction_outcome="VERIFIED", + ) + + monkeypatch.setattr(factory, "build_backend", build_backend) + monkeypatch.setattr(Replayer, "run", run) + server = _load_server(tmp_path, monkeypatch, "citrix") + deploy = tmp_path / "deploy.yaml" + deploy.write_text( + "backend:\n kind: citrix\n rdp_window: wfica32\n" + " rdp_readiness_text: Patient Notes\n" + ) + server.DEPLOYMENT = str(deploy) + out = server.run_triage_note(note="synthetic") + assert out["success"] is True and out["outcome"] == "VERIFIED" + assert seen["cfg"].kind == "citrix" and seen["cfg"].rdp_window == "wfica32" + assert seen["kind"] == "citrix" and seen["params"] == {"note": "synthetic"} + assert seen["closed"] is True + + +# -- 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 31b62cef66..d4441024b7 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"), @@ -854,3 +856,33 @@ 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 From 436dc738eb749805e73d5bd927f7b45e52e4f41f Mon Sep 17 00:00:00 2001 From: abrichr Date: Sun, 18 Oct 2026 15:08:30 -0400 Subject: [PATCH 2/3] fix(flow): the desktop MCP server runs the governed CLI; resume records an override Review fixes: - The emitted server for a non-web bundle now runs openadapt-flow run bundle --config per call (params in an owner-only file, deleted after), so backend hints, the Citrix readiness refusal, the record check, policy and approvals all apply, and run refusals come back as REFUSED. Parameters named like locals (kind, section, workflow) reach the run unchanged; the web template builds params first too. - resume threads surface_override into the resumed report, as run does. - bundle_surface reads a multi-surface bundle's primary surface. Co-Authored-By: Claude Opus 5.5 --- packages/flow/README.md | 4 +- packages/flow/openadapt_flow/__main__.py | 1 + packages/flow/openadapt_flow/emit/mcp_tool.py | 109 +++++++------- packages/flow/openadapt_flow/emit/skill.py | 4 + .../openadapt_flow/runtime/durable/resume.py | 7 + .../tests/test_backend_parity_refusals.py | 134 ++++++++++++------ packages/flow/tests/test_cli_new_commands.py | 51 ++++++- 7 files changed, 210 insertions(+), 100 deletions(-) diff --git a/packages/flow/README.md b/packages/flow/README.md index bf2aca9b98..efa10eb910 100644 --- a/packages/flow/README.md +++ b/packages/flow/README.md @@ -336,7 +336,9 @@ browser. 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`, and it drives the app that file's `backend:` section names. +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 7eda2fbfbe..cf4034dc61 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -3472,6 +3472,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: diff --git a/packages/flow/openadapt_flow/emit/mcp_tool.py b/packages/flow/openadapt_flow/emit/mcp_tool.py index 7f0e238abd..802e433a8a 100644 --- a/packages/flow/openadapt_flow/emit/mcp_tool.py +++ b/packages/flow/openadapt_flow/emit/mcp_tool.py @@ -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) @@ -108,7 +108,9 @@ def {func_name}(url: str{param_sig}) -> dict: 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), so no URL is passed. +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) @@ -117,7 +119,10 @@ def {func_name}(url: str{param_sig}) -> dict: from __future__ import annotations import argparse +import json import os +import subprocess +import sys import tempfile from pathlib import Path @@ -136,7 +141,55 @@ def _refused(detail: str) -> dict: "outcome": "REFUSED", "transaction_outcome": None, "production_eligible": False, - "detail": detail + " Nothing was executed.", + "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"), }} @@ -154,55 +207,7 @@ def {func_name}({param_sig_first}) -> dict: is RECONCILIATION_REQUIRED a write may have landed, so check the record before you call this tool again. ``REFUSED`` means nothing ran. """ - from openadapt_flow.backends.factory import build_backend - from openadapt_flow.deployment import load_deployment - from openadapt_flow.ir import Workflow - from openadapt_flow.runtime import Replayer - - 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." - ) - section = load_deployment(DEPLOYMENT).backend - kind = (section.kind or "").strip().lower().replace("_", "-") - kind = "rdp" if kind == "remote-display" else kind - if kind != SURFACE: - return _refused( - f"the workflow was recorded on {{SURFACE}} and the deployment " - f"file names {{kind!r}}." - ) - workflow = Workflow.load(BUNDLE_DIR) - params = {params_dict} - run_dir = Path(tempfile.mkdtemp(prefix="openadapt_flow_run_")) - backend = build_backend(section) - try: - report = Replayer( - backend, deployment_surface=section.deployment_surface - ).run( - workflow, - params=params, - bundle_dir=BUNDLE_DIR, - run_dir=run_dir, - execution_target_kind=SURFACE, - ) - finally: - close = getattr(backend, "close", None) - if callable(close): - close() - return {{ - "success": ( - report.execution_outcome == "VERIFIED" - and report.transaction_outcome == "VERIFIED" - ), - "outcome": report.execution_outcome, - "transaction_outcome": report.transaction_outcome, - "production_eligible": report.production_eligible, - "run_dir": str(run_dir), - "heal_count": report.heal_count, - "model_calls": report.model_calls, - "total_ms": report.total_ms, - }} + return _run_governed({params_dict}) if __name__ == "__main__": diff --git a/packages/flow/openadapt_flow/emit/skill.py b/packages/flow/openadapt_flow/emit/skill.py index a3b9c33718..d596c92903 100644 --- a/packages/flow/openadapt_flow/emit/skill.py +++ b/packages/flow/openadapt_flow/emit/skill.py @@ -36,6 +36,10 @@ 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) diff --git a/packages/flow/openadapt_flow/runtime/durable/resume.py b/packages/flow/openadapt_flow/runtime/durable/resume.py index 3dbb8f7b87..75ac08ab71 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 index e978ff1864..758b003068 100644 --- a/packages/flow/tests/test_backend_parity_refusals.py +++ b/packages/flow/tests/test_backend_parity_refusals.py @@ -19,7 +19,7 @@ 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, RunReport, Step, Workflow +from openadapt_flow.ir import ActionKind, BackendHints, Step, Workflow def _bundle(tmp_path: Path, surface: str | None) -> Path: @@ -101,71 +101,119 @@ def test_a_desktop_mcp_tool_takes_no_url(tmp_path: Path, monkeypatch) -> None: assert "playwright" not in (tmp_path / "mcp" / "server.py").read_text() -def test_a_desktop_mcp_tool_refuses_without_or_with_the_wrong_deployment( +def test_a_desktop_mcp_tool_refuses_without_a_deployment( tmp_path: Path, monkeypatch ) -> None: - import openadapt_flow.backends.factory as factory + import subprocess monkeypatch.setattr( - factory, "build_backend", lambda *a, **k: pytest.fail("must not build") + 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"] - 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) - out = server.run_triage_note(note="synthetic") - assert out["outcome"] == "REFUSED" - assert "recorded on macos" in out["detail"] and "'linux'" in out["detail"] - -def test_a_desktop_mcp_tool_drives_the_app_its_deployment_names( +def test_a_desktop_mcp_tool_runs_the_governed_cli_with_every_param_intact( tmp_path: Path, monkeypatch ) -> None: - import openadapt_flow.backends.factory as factory - from openadapt_flow.runtime import Replayer + """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] = {} - class Backend: - def close(self) -> None: - seen["closed"] = True - - def build_backend(cfg: Any, **_: Any) -> Backend: - seen["cfg"] = cfg - return Backend() - - def run(self: Any, workflow: Workflow, **kwargs: Any) -> RunReport: - seen["kind"] = kwargs["execution_target_kind"] - seen["params"] = kwargs["params"] - return RunReport( - workflow_name=workflow.name, - started_at="2026-10-18T00:00:00Z", - success=True, - execution_profile="standard", - execution_outcome="VERIFIED", - transaction_outcome="VERIFIED", + 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") - monkeypatch.setattr(factory, "build_backend", build_backend) - monkeypatch.setattr(Replayer, "run", run) - server = _load_server(tmp_path, monkeypatch, "citrix") + 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: citrix\n rdp_window: wfica32\n" - " rdp_readiness_text: Patient Notes\n" + "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 True and out["outcome"] == "VERIFIED" - assert seen["cfg"].kind == "citrix" and seen["cfg"].rdp_window == "wfica32" - assert seen["kind"] == "citrix" and seen["params"] == {"note": "synthetic"} - assert seen["closed"] is True + 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 -------------------------------------- diff --git a/packages/flow/tests/test_cli_new_commands.py b/packages/flow/tests/test_cli_new_commands.py index d4441024b7..4979de390d 100644 --- a/packages/flow/tests/test_cli_new_commands.py +++ b/packages/flow/tests/test_cli_new_commands.py @@ -568,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() @@ -623,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( @@ -713,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() ), @@ -776,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() ), @@ -886,3 +888,44 @@ def test_resume_refuses_another_surface_than_the_bundle_was_recorded_on( 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 From 24ea10e1eaeb397bb611161b3f44a79e1b94db98 Mon Sep 17 00:00:00 2001 From: abrichr Date: Sun, 18 Oct 2026 20:18:39 -0400 Subject: [PATCH 3/3] fix(flow): bench refusal names the run command through cli_hints The printed hint hard-coded the engine spelling, which test_cli_hints forbids: the launcher (openadapt flow) and the engine (openadapt-flow) print their own spelling. Co-Authored-By: Claude Opus 5.5 --- packages/flow/openadapt_flow/__main__.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index cf4034dc61..6eea45e08a 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -3884,11 +3884,13 @@ def _cmd_bench(args: argparse.Namespace) -> int: 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 `openadapt-flow run BUNDLE --config " - "deploy.yaml`. Nothing was executed." + "on its own app with repeated " + f"`{command('run BUNDLE --config deploy.yaml')}`. Nothing was executed." ) return 2 url, stop = serve(port=0)