fix(flow): emit, serve-execute, bench and resume stop assuming a browser

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
 

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)
 

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 <file>``, 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)

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 <APP_URL>"
+    surface = bundle_surface(workflow)
+    target = "--url <APP_URL>" if surface == "web" else "--config <DEPLOYMENT_FILE>"
+    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 `<APP_URL>` with the URL of the "
-        f"running target app. "
+        f"`{_BUNDLE_SUBDIR}/`. "
     )
+    if surface == "web":
+        usage_note += "Replace `<APP_URL>` with the URL of the running target app. "
+    else:
+        usage_note += (
+            f"This workflow was recorded on the `{surface}` surface. Replace "
+            "`<DEPLOYMENT_FILE>` 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 `<name>` placeholder "

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:

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(