#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = ["playwright"]
# ///
"""Run scenario tests against a live device via HTTP.

Same JSON format as the in-process runner. Executes steps via REST API
and collects per-step performance measurements.
"""

import argparse
import atexit
import copy
import signal
import json
import os
import re
import statistics
import subprocess
import sys
import time
import urllib.request
import urllib.error
import urllib.parse
from pathlib import Path


def _mod_path(name: str) -> str:
    """`/api/modules/<name>` with the name URL-encoded. A module name a user chose can
    contain a space, which urllib rejects in a raw URL: encode so delete/replace/clear
    can address it."""
    return "/api/modules/" + urllib.parse.quote(name, safe="")

ROOT = Path(__file__).resolve().parent.parent.parent
SCENARIOS_DIR = ROOT / "test" / "scenarios"

# Reuse the shared test-metadata parser so scenario discovery stays in one place.
sys.path.insert(0, str(ROOT / "moondeck"))
from _host import desktop_target  # noqa: E402
sys.path.insert(0, str(ROOT / "moondeck" / "docs"))
import _test_metadata as test_meta  # noqa: E402
sys.path.insert(0, str(ROOT / "moondeck" / "scenario"))
import _observed  # noqa: E402


# A REDIRECTED Windows stdout takes the locale encoding, cp1252, and this runner prints an arrow
# per step, so a scenario that passed would be recorded as FAILED by the print rather than by the
# device. An attached console is UTF-8 since Python 3.6 (PEP 528), so the failing mode is a pipe
# or a CI log, which is exactly how a gate runs it. errors=replace so a stray glyph never costs
# a result.
if sys.stdout is not None:
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
if sys.stderr is not None:
    sys.stderr.reconfigure(encoding="utf-8", errors="replace")


class Client:
    # Mutating ops (add/delete/replace/control) trigger a full prepareTree on the
    # device: at 128x128 that frees/reallocates a large buffer + LUT and can
    # take several seconds on a busy ESP32. 5s was too tight (deletes timed out
    # mid-teardown, leaving a half-mutated tree). 15s clears the worst case while
    # still catching a genuinely hung device.
    TIMEOUT_S = 15

    def __init__(self, host: str):
        self.base = f"http://{host}"
        # Whether the device is a process on THIS machine. A desktop build elsewhere is named after
        # its own platform and restarts itself, where a local one is named after this host and is
        # relaunched from here. Decided by name alone, so this machine reached by its own LAN
        # address or hostname counts as another one.
        name = urllib.parse.urlsplit(f"//{host}").hostname or host
        self.local = name in ("localhost", "127.0.0.1", "::1")

    def _send(self, req):
        # A mutating call triggers prepareTree; while the device is mid-rebuild it
        # can drop the TCP connection (ConnectionResetError / "remote end closed")
        # or briefly refuse one. The device recovers in well under a second, so a
        # single transient drop shouldn't cascade-fail the run: retry once after
        # a short settle. A genuine HTTPError (4xx/5xx from the handler) is a real
        # result and is NOT retried; it propagates to the caller.
        for attempt in range(2):
            try:
                with urllib.request.urlopen(req, timeout=self.TIMEOUT_S) as resp:
                    return json.loads(resp.read())
            except urllib.error.HTTPError:
                raise  # a real handler response: let the caller decide
            except (urllib.error.URLError, ConnectionError, OSError):
                if attempt == 0:
                    time.sleep(1.0)
                    continue
                raise

    def get(self, path: str):
        return self._send(urllib.request.Request(f"{self.base}{path}"))

    def post(self, path: str, data: dict):
        body = json.dumps(data).encode()
        return self._send(urllib.request.Request(
            f"{self.base}{path}", data=body,
            headers={"Content-Type": "application/json"}))

    def get_text(self, path: str) -> str:
        """GET a raw body. /api/file returns the file's CONTENTS, not JSON, so it cannot go
        through get() which parses what it reads."""
        req = urllib.request.Request(f"{self.base}{path}")
        with urllib.request.urlopen(req, timeout=self.TIMEOUT_S) as resp:
            return resp.read().decode("utf-8", "replace")

    def post_text(self, path: str, text: str):
        """POST a raw text body. /api/file takes the file's CONTENTS, not JSON: the body IS
        the payload, which is why it cannot go through post() like every control write."""
        return self._send(urllib.request.Request(
            f"{self.base}{path}", data=text.encode("utf-8"),
            headers={"Content-Type": "text/plain"}))

    def delete(self, path: str):
        return self._send(urllib.request.Request(f"{self.base}{path}", method="DELETE"))

    def patch(self, path: str, data: dict):
        return self._send(urllib.request.Request(
            f"{self.base}{path}", data=json.dumps(data).encode(), method="PATCH",
            headers={"Content-Type": "application/json"}))


def _error_body(he: urllib.error.HTTPError) -> dict:
    """An error response's JSON object, or an empty one when the body is not JSON, such as a proxy's HTML page."""
    try:
        body = json.loads(he.read() or b"{}")
    except ValueError:
        return {}
    return body if isinstance(body, dict) else {}


def _today_iso() -> str:
    """ISO date stamp for set_by fields. Local timezone is fine: set_by is a
    coarse "around when did this contract get blessed" marker, not a timestamp."""
    import datetime
    return datetime.date.today().isoformat()


def collect_metrics(client: Client, settle_s: float = 1.5) -> dict:
    """Wait for system to settle, then collect metrics. Augments /api/system with
    `dynamicBytesTotal` (sum from the module tree) so callers can compare the
    model's allocation prediction against the observed free heap."""
    time.sleep(settle_s)
    metrics = client.get("/api/system")
    try:
        state = client.get("/api/state")
        metrics["dynamicBytesTotal"] = _sum_dynamic_bytes(state)
        metrics["minStackFree"], metrics["minStackTask"] = _min_stack_free(state)
    except Exception:
        metrics["dynamicBytesTotal"] = None
    return metrics


def _min_stack_free(state: dict):
    """The task with the least stack headroom on the Tasks card, as (bytes, name), or (None, None) where the card lists none."""
    least = (None, None)
    def walk(modules):
        nonlocal least
        for m in modules:
            for c in m.get("controls", []) if m.get("type") == "TasksModule" else []:
                if c.get("name") != "tasks" or not isinstance(c.get("value"), list):
                    continue
                for t in c["value"]:
                    free = t.get("stack")
                    if isinstance(free, int) and (least[0] is None or free < least[0]):
                        least = (free, t.get("name"))
            walk(m.get("children", []))
    walk(state.get("modules", []))
    return least


_browser = {}


def _browser_page():
    """The one browser page the run's `browser` steps share, started on first use; Playwright and Chromium are MoonDeck's, installed by install_playwright.py."""
    if "page" not in _browser:
        from playwright.sync_api import sync_playwright
        _browser["pw"] = sync_playwright().start()
        _browser["browser"] = _browser["pw"].chromium.launch()
        _browser["page"] = _browser["browser"].new_page()
    return _browser["page"]


def _browser_close():
    """Close the browser, if a `browser` step opened one; also run at exit, so a failed run leaves no viewer on the device."""
    if "browser" in _browser:
        _browser["browser"].close()
        _browser["pw"].stop()
    _browser.clear()


atexit.register(_browser_close)


def _floor_verdict(metric: str, value, floor, tolerance_pct: float):
    """A contract floor: (passed, line). The device must deliver at least `floor`, less by more than the tolerance failing; a value of None or 0 against a positive floor fails, since the device then reports none."""
    if not value or value <= 0:
        return False, f"  FAIL  {metric} {value} (the device reports none) vs contract {floor}"
    drop_pct = (floor - value) * 100.0 / floor if value < floor else 0
    if drop_pct > tolerance_pct:
        return False, f"  FAIL  {metric} {value} dropped {drop_pct:.1f}% below contract {floor}"
    return True, f"  PASS  {metric} {value} >= contract {floor} (within -{tolerance_pct}% tolerance)"


def _control_value(module: dict, name: str):
    """Return the value of a named control on a module, or None.

    `enabled` and `status` are fields of the module itself rather than controls, so they are read from there.
    """
    if name in ("enabled", "status"):
        return module.get(name)
    for ctrl in module.get("controls", []):
        if ctrl.get("name") == name:
            return ctrl.get("value")
    return None


def _detect_target(state: dict, local: bool = True) -> str:
    """Identify the build target so per-step contract values can be looked up.

    ESP32: read FirmwareUpdateModule.firmware (`esp32`, `esp32-eth`, `esp32-eth-wifi`,
    `esp32s3-n16r8`, …): set at compile time from MM_FIRMWARE_NAME and
    exposed through the `firmware` control. Desktop: same key but reports
    `unknown`, so we substitute desktop-<host-os> using the runtime os name (still
    distinguishes macOS vs Linux vs Windows builds, which can differ in tick
    noticeably). See docs/explanation/architecture/index.md § Firmware vs board.

    A desktop build on ANOTHER machine is named after the platform it reports about itself
    (`desktop-docker-arm64`), because this host's name says nothing about it: a NanoPi measured
    from a Mac was recorded as `desktop-macos`, into the Mac's own figures.
    """
    firmware = model = None
    for m in state.get("modules", []):
        for c in m.get("controls", []):
            if m.get("type") == "FirmwareUpdateModule" and c.get("name") == "firmware":
                firmware = c.get("value")
            elif m.get("type") == "SystemModule" and c.get("name") == "deviceModel":
                model = c.get("value")
    if firmware and firmware != "unknown":
        return firmware
    if not local:
        return "desktop-" + str(model or "remote").strip().lower().replace(" ", "-")
    # Desktop fallback, from the one home every script shares (moondeck/_host.py).
    return desktop_target()


# --- The host's own WiFi, for a scenario that walks a device's first setup from its access point. ---
# The host joins the device's access point, loses its own network meanwhile, and is always put back:
# a step does it, and atexit does it again for a run that stops in between.

REGISTRY_FILE = ROOT / "moondeck" / "moondeck.json"
# The access point's address, which the firmware's captive portal fixes (src/core/util/CaptivePortal.h).
ACCESS_POINT_ADDRESS = "4.3.2.1"


def _registry_network(name: str) -> dict:
    """A network from the bench registry, `{name, ssid, password}`; the registry is not in git, so credentials stay local."""
    with open(REGISTRY_FILE, encoding="utf-8") as f:
        for net in json.load(f).get("networks", []):
            if net.get("name") == name:
                wifi = net.get("wifi") or {}
                return {"name": name, "ssid": wifi.get("ssid", ""), "password": wifi.get("password", "")}
    raise KeyError(f"no network named {name!r} in {REGISTRY_FILE.name}")


def _wifi_device() -> str:
    """The host's WiFi interface (en0 on most Macs), read rather than assumed."""
    out = subprocess.run(["networksetup", "-listallhardwareports"], capture_output=True, text=True, timeout=30).stdout
    lines = out.splitlines()
    for i, line in enumerate(lines):
        if line.strip() in ("Hardware Port: Wi-Fi", "Hardware Port: AirPort") and i + 1 < len(lines):
            return lines[i + 1].split(":", 1)[1].strip()
    raise RuntimeError("no WiFi interface found (this op needs macOS networksetup)")


def _reachable(url: str, timeout_s: float) -> bool:
    """Whether `url` answers within the time, polled, since a join takes a few seconds to route."""
    end = time.time() + timeout_s
    while time.time() < end:
        try:
            with urllib.request.urlopen(url, timeout=3):
                return True
        except urllib.error.HTTPError:
            return True   # any HTTP answer is an answer
        except Exception:
            time.sleep(1)
    return False


def _join_failed(out) -> bool:
    """Whether networksetup reported a failure: it exits 0 either way, so its words decide."""
    text = out.stdout + out.stderr
    return out.returncode != 0 or any(w in text for w in ("Could not find network", "Failed", "Error"))


def _host_join(ssid: str, password: str, probe_url: str, timeout_s: float) -> str:
    """Join `ssid` and wait until `probe_url` answers: "" on success, else why.

    The join is retried only while networksetup reports a failure, since an access point still starting is not in the scan yet; once it took, the probe is polled rather than the join repeated, which would drop the link it made. The password rides networksetup's arguments, which is the one way that tool takes it.
    """
    device = _wifi_device()
    end = time.time() + timeout_s
    last = ""
    while time.time() < end:
        cmd = ["networksetup", "-setairportnetwork", device, ssid] + ([password] if password else [])
        try:
            out = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
        except subprocess.TimeoutExpired:
            last = "networksetup did not answer"
            continue
        last = (out.stdout + out.stderr).strip()
        if not _join_failed(out):
            if _reachable(probe_url, max(1.0, end - time.time())):
                return ""
            return f"joined {ssid!r}, but {probe_url} did not answer in time"
        time.sleep(2)
    return f"could not join {ssid!r} ({last or 'no message'})"


class HostWifi:
    """Where the host's WiFi is, so a run that moved it to a device's access point always moves it back."""

    def __init__(self):
        self.home = None   # the registry network to return to, set once the host leaves it

    def restore(self):
        """Rejoin the home network, retried as a join is, since macOS often misses a network the moment it leaves an access point; home is kept until it took, so the exit hook can try again."""
        if not self.home:
            return
        home = self.home
        print(f"  HOST  back to {home['ssid']}")
        end = time.time() + 30
        last = ""
        while time.time() < end:
            try:
                out = subprocess.run(["networksetup", "-setairportnetwork", _wifi_device(), home["ssid"], home["password"]],
                                     capture_output=True, text=True, timeout=30)
            except subprocess.TimeoutExpired:
                last = "networksetup did not answer"
                continue
            if not _join_failed(out):
                self.home = None
                return
            last = (out.stdout + out.stderr).strip()
            time.sleep(2)
        print(f"  HOST  WARNING: could not rejoin {home['ssid']} ({last or 'no message'}); rejoin it by hand")


def _fill(value, ctx: dict):
    """Fill `{network.ssid}`, `{network.password}` and `{device}` in a step's strings, so a scenario names no credential."""
    if isinstance(value, str):
        for key, sub in ctx.items():
            value = value.replace("{" + key + "}", sub)
        return value
    if isinstance(value, dict):
        return {k: _fill(v, ctx) for k, v in value.items()}
    if isinstance(value, list):
        return [_fill(v, ctx) for v in value]
    return value


def _http_request(url: str, base: str, method: str = "GET") -> urllib.request.Request:
    """The request an `expect_http` step sends: a path starting with / goes to the device under test, and a POST carries an empty body, as the UI's buttons send."""
    if url.startswith("/"):
        url = base.rstrip("/") + url
    return urllib.request.Request(url, data=b"" if method == "POST" else None, method=method)


def _typed_names(doc) -> list:
    """The module names a state document can create: every object carrying a `type`, depth first."""
    names = []
    if isinstance(doc, dict):
        for key, value in doc.items():
            if isinstance(value, dict):
                if "type" in value:
                    names.append(key)
                names.extend(_typed_names(value))
    return names


def _list_rows(client, module_id: str, key: str) -> list:
    mod = client.get(_mod_path(module_id))
    for c in mod.get("controls") or []:
        if c.get("name") == key:
            return c.get("value") or []
    raise KeyError(f"{module_id} has no list {key!r}")


def _uptime_seconds(client):
    """The System card's uptime as seconds, or None when it cannot be read.

    The value is `H:MM:SS`, and its only use here is comparing two readings across a restart.
    """
    try:
        mod = client.get(_mod_path("System"))
    except Exception:
        return None
    for c in (mod.get("controls") or []):
        if c.get("name") == "uptime":
            parts = str(c.get("value") or "").split(":")
            if len(parts) != 3:
                return None
            try:
                h, m, sec = (int(x) for x in parts)
            except ValueError:
                return None
            return h * 3600 + m * 60 + sec
    return None


def _restarted(before: int, now: int, elapsed: float) -> bool:
    """Whether an uptime reading proves a restart happened since `before` was read, `elapsed` seconds ago.

    A device that kept running has gained as many seconds of uptime as the wall clock has, so one
    that gained fewer restarted in between. Comparing against the wall clock rather than against
    `before` alone is what makes this hold when the device booted moments ago: `now < before` reads
    a fresh boot as "did not restart" whenever the new uptime has already passed the old one, which
    is every time two scenarios restart a board back to back. The two seconds cover the uptime being
    whole seconds on both readings.
    """
    return now + 2 < before + elapsed


def _remember_control(prior: dict, client, mod_id: str, key: str) -> None:
    """Note what a control held before a scenario first writes it, so the run can put it back.

    Only the first write counts: a later one would record the scenario's own value.
    A control that cannot be read is left out, and the scenario's value then stays.
    A password is left out too: the module read returns it obfuscated, and writing that back would set the obfuscated text as the password.
    """
    if (mod_id, key) in prior:
        return
    try:
        module = client.get(_mod_path(mod_id))
    except Exception:
        return
    if any(c.get("name") == key and c.get("type") == "Password" for c in module.get("controls", [])):
        return
    value = _control_value(module, key)
    if value is not None:
        prior[(mod_id, key)] = value


def _restore_controls(client, prior: dict) -> int:
    """Set every control a scenario wrote back to what it held before, newest write first.

    A scenario that names a device to prove the name survives a restart otherwise leaves that name on every device it runs on, and two bench devices answering to one name is how a restore landed on the wrong one.
    A control whose module the scenario itself created is gone by now, and that 404 stays quiet; any other failure is reported.
    Returns how many the device confirmed.
    """
    done = 0
    for (mod_id, key), value in reversed(list(prior.items())):
        try:
            if client.post("/api/control", {"module": mod_id, "control": key, "value": value}).get("ok"):
                done += 1
            else:
                print(f"  WARN  restore: the device declined {mod_id}.{key}={value!r}")
        except urllib.error.HTTPError as e:
            if e.code != 404:
                print(f"  WARN  restore: could not set {mod_id}.{key}={value!r}: {e}")
        except Exception as e:
            print(f"  WARN  restore: could not set {mod_id}.{key}={value!r}: {e}")
    return done


def _reboot_and_wait(client, target: str, timeout_s: float = 60.0) -> str:
    """Restart the device and wait for it to answer again, so a scenario can prove what survives.

    On a board this is the reboot the endpoint performs. On a desktop the same endpoint EXITS the
    process and nothing restarts it, so the runner relaunches the binary itself, with the data
    directory the exiting instance was using: a restart that came back on different files would
    prove nothing about persistence. Returns "" on success, or the reason it did not come back.
    """
    data_dir = os.environ.get("MM_DATA_DIR")
    # Only a desktop build on THIS machine is relaunched from here. One on another host is brought
    # back by whatever runs it there, a container's restart policy or a service manager, and
    # launching a local binary for it would start a second, unrelated device on this machine.
    is_desktop = target.startswith("desktop-") and client.local

    # Uptime before the restart, so the recovered instance can be told from the one still running.
    # Without it the first answering /api/state is accepted, which on a board is routinely the OLD
    # instance replying before it goes down: every persistence assertion after that proves nothing.
    before = _uptime_seconds(client)
    read_at = time.time()

    try:
        client.post("/api/reboot", {})
    except urllib.error.HTTPError as re_:
        # A refused reboot is a failed step: the device is up, so the poll below would pass at once.
        return f"/api/reboot returned HTTP {re_.code}"
    except Exception:
        pass                  # the device goes away mid-response, which is the expected shape

    if is_desktop:
        # The same resolver run_desktop.py uses, which picks the NEWEST candidate rather than the
        # first that exists: picking by existence served a stale build whose changes read as no-ops,
        # and it names the per-host directory, so this works on Linux and Windows too.
        sys.path.insert(0, str(ROOT / "moondeck" / "run"))
        from run_desktop import _resolve_executable            # noqa: E402
        binary = _resolve_executable()
        if not binary.exists():
            return f"no desktop binary to relaunch (looked for {binary})"
        time.sleep(1.0)       # let the old process release the port before the new one binds it
        env = dict(os.environ)
        if data_dir:
            env["MM_DATA_DIR"] = data_dir
        subprocess.Popen([str(binary)], cwd=str(ROOT), env=env,
                         stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    # Evidence of a restart, not merely of an answer: either the device went away and came back, or
    # it answers with an uptime lower than before. One of the two must hold before this reports success.
    deadline = time.time() + timeout_s
    went_away = False
    while time.time() < deadline:
        try:
            client.get("/api/state")
        except Exception:
            went_away = True          # the old instance is gone, so the next answer is the new one
            time.sleep(1.0)
            continue
        if went_away:
            return ""
        now = _uptime_seconds(client)
        if before is not None and now is not None and _restarted(before, now, time.time() - read_at):
            return ""                 # the clock fell behind the wall, which only a reboot does
        time.sleep(1.0)
    if not went_away:
        if before is None:
            return (f"still answering after {timeout_s:.0f}s, and the uptime could not be read before "
                    "the restart, so nothing here can tell whether it happened")
        return (f"still answering after {timeout_s:.0f}s with no uptime reset: "
                "the reboot did not take effect")
    return f"did not answer within {timeout_s:.0f}s"


def _read_values(client, metrics: dict, names: list, boards: dict) -> dict:
    """The values a measure step names: "fps" and "Module.tick" from /api/system, "Module.control" from /api/state, "Module.status[unit]" for the number before that unit in its status; "board:" in front reads another board."""
    ticks = {m.get("name"): m.get("us") for m in metrics.get("modules", []) or []}
    out = {}
    state = None
    for name in names:
        board, _, rest = name.rpartition(":")
        if board:
            try:
                out[name] = _read_values(boards[board], boards[board].get("/api/system"), [rest], {})[rest]
            except Exception as e:
                out[name] = f"? ({type(e).__name__})"
            continue
        if name == "fps":
            out[name] = metrics.get("fps", "-")
            continue
        mod, _, key = name.partition(".")
        if key == "tick":
            out[name] = ticks.get(mod, "-")
            continue
        if state is None:
            try:
                state = client.get("/api/state")
            except Exception as e:   # a device too busy to answer whole reads as unknown, and the run goes on to its restore
                state = {"error": f"? ({type(e).__name__})"}
        out[name] = state.get("error", "-")
        stack = list(state.get("modules", []))
        while stack:
            m = stack.pop()
            if m.get("name") == mod:
                unit = re.fullmatch(r"status\[(.+)\]", key)
                out[name] = status_number(m.get("status"), unit.group(1)) if unit else _control_value(m, key)
                break
            stack.extend(m.get("children", []) or [])
    return out


def median_reads(reads: list) -> dict:
    """Each value's median over several reads, where every read of it is a number; else its last read."""
    out = {}
    for key in reads[-1]:
        values = [r.get(key) for r in reads]
        numbers = [v for v in values if isinstance(v, (int, float)) and not isinstance(v, bool)]
        out[key] = statistics.median(numbers) if len(numbers) == len(values) else values[-1]
    return out


def status_number(status, unit: str):
    """The number in front of `unit` in a status line, such as 41 from "41 fps, 2197 packets/s", or "-" when it has none."""
    found = re.search(rf"(\d+) {re.escape(unit)}(?!\w)", status or "")
    return int(found.group(1)) if found else "-"


def _sum_dynamic_bytes(state: dict) -> int:
    """Sum dynamicBytes across the live module tree. Returned by /api/state per
    module; the sum is the model's prediction for how much heap the tree owns.

    NOT the same as (boot_heap - free_heap): the framework (lwIP, WiFi stack,
    FreeRTOS, HTTP server kernel buffers) consumes heap outside the model.
    Printed alongside the contract for sanity-checking: a regression here
    means a module started allocating something the contract didn't budget for.
    """
    total = 0
    def walk(modules):
        nonlocal total
        for m in modules:
            try:
                total += int(m.get("dynamicBytes", 0))
            except (TypeError, ValueError):
                pass
            walk(m.get("children", []))
    walk(state.get("modules", []))
    return total


def _collect_module_names(state: dict) -> set:
    """Collect every module name in the live tree, including nested children.
    Used by mutate scenarios to pre-flight that every id they touch is wired."""
    names = set()
    def walk(modules):
        for m in modules:
            n = m.get("name")
            if n:
                names.add(n)
            walk(m.get("children", []))
    walk(state.get("modules", []))
    return names


# The containers whose content a scenario may change: their documents, read before the run and applied back after it, leave the board as found.
_SNAPSHOT_CONTAINERS = ("Layouts", "Effects", "Drivers", "Services")


def _snapshot_documents(client) -> list:
    """Each snapshot container as its state document, the shape `PATCH /api/state` takes back.

    A container's document says its children are exactly the ones it lists, so applying it removes what a scenario added, re-creates what it removed, re-types what it replaced and restores every value, in one request per container.
    A container the device does not have is left out.
    """
    docs = []
    for name in _SNAPSHOT_CONTAINERS:
        try:
            docs.append(client.get(_mod_path(name) + "/document"))
        except Exception:
            pass
    return docs


def _merge(into: dict, doc: dict):
    """Deep-merge a state document into another, the later value winning."""
    for k, v in doc.items():
        if isinstance(v, dict) and isinstance(into.get(k), dict):
            _merge(into[k], v)
        else:
            into[k] = v


def _fixture_canvas(fixture: list) -> list:
    """The fixture as one document per container it names, each replacing the container's children with exactly the fixture's.

    Only a container the snapshot restores may be replaced, so a fixture naming another raises rather than leaving the bench changed.
    """
    merged: dict = {}
    for step in fixture:
        if step.get("op") == "apply_state":
            _merge(merged, copy.deepcopy(step.get("document", {})))
    outside = sorted(set(merged) - set(_SNAPSHOT_CONTAINERS))
    if outside:
        raise ValueError(f"the fixture names {', '.join(outside)}, which the run does not restore")
    return [{name: {"$patch": "replace", **body}} for name, body in merged.items() if isinstance(body, dict)]


def run_scenario(client: Client, scenario_path: Path, settle_s: float = 1.5,
                 update_contract: bool = False, update_observed: bool = False,
                 update_reason: str | None = None, named: bool = True,
                 network: str | None = None, boards: dict | None = None, keep: bool = False) -> dict:
    """Run a scenario against a live device and return results.

    Mode handling (see docs/reference/testing.md § Scenario modes):
      construct : scenario builds the pipeline from scratch. Live device's
                   main.cpp owns the top-level shape, so construct scenarios
                   only run in-process. Skip here with a clear note.
      mutate    : scenario assumes a wired pipeline. The fixture is its canvas:
                   each container the fixture names is replaced with exactly
                   the fixture's children before the steps, whatever the bench
                   carried, and the snapshot restore puts the device back.
                   Steps that touch ids neither on the device nor added before
                   them hard-fail.
    """
    # encoding="utf-8" EXPLICITLY, and it is not cosmetic here: _observed.py writes this file back
    # on --update-observed or --update-contract, so reading it as cp1252 on Windows and writing it as
    # UTF-8 double-encodes every non-ASCII character in the scenario's own prose. One run turned
    # "16²→32²" into "16²â†→32²" in every file it touched.
    with open(scenario_path, encoding="utf-8") as f:
        scenario = json.load(f)

    name = scenario.get("name", scenario_path.stem)
    mode = scenario.get("mode", "construct")  # back-compat default
    print(f"\n=== Scenario: {name} ===")
    print(scenario.get("description", ""))

    results = {"name": name, "steps": [], "passed": True, "skipped": False}
    created_modules = []  # mutate scenarios rarely add modules but the existing cleanup path is still useful
    created_files = []    # files a write_file step created, removed at the end even when a later step stopped the run
    prior_files = {}      # files a write_file step overwrote, as they were before its first write, written back at the end
    wrote_observations = [False]  # sentinel; flipped by each measure step that runs
    # Keyed by (step, target): the original contract block before --update-contract
    # mutated it (or None if no prior block existed). Used by the post-run gate to
    # roll mutations back when the run fails, so a renegotiated promise only
    # lands on a clean run.
    pending_contract_originals: dict = {}

    if scenario.get("on_request") and not named:
        print("\n  SKIP (on_request: runs only when named, since it takes the host off its network)")
        results["skipped"] = True
        return results
    # A scenario over several boards names them in `boards`; the command line maps each name to a device with --board.
    boards = boards or {}
    wanted = scenario.get("boards", [])
    absent_boards = [b for b in wanted if b not in boards]
    if absent_boards:
        print(f"\n  SKIP (boards: runs only with {', '.join(absent_boards)} named by --board NAME=HOST)")
        results["skipped"] = True
        return results
    boards = {b: boards[b] for b in wanted}
    if mode == "construct":
        print("\n  SKIP (mode=construct: runs in-process only; the live device's "
              "main.cpp owns the top-level shape)")
        results["skipped"] = True
        return results
    if mode != "mutate":
        print(f"\n  FAIL: unknown mode: {mode!r} (expected construct or mutate)")
        results["passed"] = False
        return results

    # Pre-flight: every id a step sets must be reachable: either already on the
    # device, OR created by an earlier state document in this scenario. A
    # canvas-preparing scenario clears the containers and builds its own tree, so
    # its set_control ids won't exist on the device yet; they're created mid-run.
    # A still-unreachable id is a real typo / wrong-wiring bug.
    target = "unknown"
    live_state = None   # bound before the try so the snapshot below can't NameError if /api/state fails
    try:
        live_state = client.get("/api/state")
        target = _detect_target(live_state, client.local)
        # Walk the steps in order, growing the reachable set as state documents
        # create ids. The containers (Layouts/Effects/Drivers) are always present.
        reachable = _collect_module_names(live_state)
        # The FIXTURE runs before the steps and creates the wired pipeline, so the ids it adds
        # are reachable by the time any step runs. Walking only `steps` reported a fixture-added
        # module as missing the moment a scenario's first step targeted one.
        for step in scenario.get("fixture", []):
            if step.get("op") == "apply_state":
                reachable.update(_typed_names(step.get("document")))
        missing = []
        for step in scenario.get("steps", []):
            if step.get("device"):
                continue   # its ids live on another board, which the pre-flight does not read
            sid = step.get("id")
            opn = step.get("op")
            if opn == "apply_state":
                reachable.update(_typed_names(step.get("document")))
            elif opn == "set_control" and sid:
                # `optional` steps are best-effort (e.g. shrink the grid before a
                # clear, if a grid exists): the executor skips them on a missing
                # target, so they don't count as a wiring bug in the pre-flight.
                if sid not in reachable and not step.get("optional"):
                    missing.append(sid)
        # Reset-block ids must exist before steps run (no add can precede them).
        for r in scenario.get("reset", []):
            if r.get("op") == "set_control" and r.get("id") and r["id"] not in reachable:
                missing.append(r["id"])
        missing = sorted(set(missing))
        if missing:
            print(f"\n  FAIL: scenario references ids that are neither on the live "
                  f"device nor added by an earlier step: {', '.join(missing)}. "
                  f"Fix the wiring or add the module first.")
            results["passed"] = False
            return results
    except Exception as e:
        print(f"\n  WARN: couldn't pre-flight live module names: {e}")
    print(f"  Target: {target}")
    results["target"] = target

    # A scenario that moves the host's WiFi runs on a registry network the host and the device share, named by --network, and its steps fill in from it.
    ctx: dict = {}
    host = HostWifi()
    home_base = client.base
    net = None
    if scenario.get("host_network"):
        try:
            if not network:
                raise KeyError("this scenario moves the host's WiFi: name the network it returns to with --network")
            net = _registry_network(network)
        except Exception as e:
            print(f"\n  FAIL: {e}")
            results["passed"] = False
            return results
        ctx["network.ssid"], ctx["network.password"] = net["ssid"], net["password"]
        for m in (live_state or {}).get("modules", []):
            for c in m.get("controls", []):
                if m.get("type") == "SystemModule" and c.get("name") == "deviceName":
                    ctx["device"] = str(c.get("value") or "")
            for ch in m.get("children", []) or []:
                for c in ch.get("controls", []):
                    if ch.get("type") == "AccessPointModule" and c.get("name") == "name":
                        ctx["access_point"] = str(c.get("value") or "")
        atexit.register(host.restore)   # a run that stops on the access point still puts the host back
        # MoonDeck's Stop sends SIGTERM, whose default skips atexit; exiting through it runs the hook.
        signal.signal(signal.SIGTERM, lambda *_: sys.exit(143))

    # Snapshot the containers a scenario may clear or rebuild, as documents, so the board ends the run as found.
    prior_controls: dict = {}   # (module, control) -> the value it held before this scenario wrote it
    snapshot_docs = _snapshot_documents(client)
    # Every other board is restored the same way, from its own snapshot, controls and created modules.
    home = (client, prior_controls, created_modules)
    board_state = {b: (c, {}, []) for b, c in boards.items()}
    board_snapshots = {b: _snapshot_documents(c) for b, c in boards.items()}
    obs_target = target
    if boards:
        # A step names a board by its address as `{host}` or `{<name>}`, as a sender's `hosts` does.
        ctx["host"] = urllib.parse.urlsplit(home_base).hostname or ""
        for b, c in boards.items():
            ctx[b] = urllib.parse.urlsplit(c.base).hostname or ""
        # The measurements belong to the combination, so wifi-to-wifi and ethernet-to-wifi keep apart.
        kinds = [f"host={target}"]
        for b, c in boards.items():
            try:
                kinds.append(f"{b}={_detect_target(c.get('/api/state'), c.local)}")
            except Exception:
                kinds.append(f"{b}=unknown")
        obs_target = " ".join(kinds)
        print(f"  Boards: {obs_target}")

    # The fixture is the canvas the steps expect, on a bench that may carry another effect or grid.
    try:
        canvas = _fixture_canvas(scenario.get("fixture", []))
    except ValueError as e:
        print(f"\n  FAIL: {e}")
        results["passed"] = False
        return results
    for doc in canvas:
        try:
            client.patch("/api/state", doc)
            print(f"  CANVAS {next(iter(doc))}: replaced with the fixture")
        except Exception as e:
            print(f"\n  FAIL: couldn't lay the fixture's canvas on {next(iter(doc))}: {e}")
            results["passed"] = False
            for snap in snapshot_docs:
                try:
                    client.patch("/api/state", snap)
                except Exception:
                    pass
            return results

    # Reset block: scenarios that mutate shared controls (Mirror toggles, grid
    # size, Preview detail, …) declare a `reset` array of set_control steps that
    # restores those controls to production defaults BEFORE the scenario runs.
    # Without this each scenario's measurements depend on whatever the previous
    # scenario left behind, so contract assertions become coupled to run order.
    # Reset failures fail-fast: a swallowed reset means the measurements reflect
    # the wrong state, which silently produces false-positive contract passes
    # (or false-negative failures) downstream. Better to abort cleanly here.
    reset_steps = scenario.get("reset", [])
    if reset_steps:
        print(f"\n  --- reset ({len(reset_steps)} steps) ---")
        for r_step in reset_steps:
            if r_step.get("op") != "set_control":
                continue
            try:
                client.post("/api/control", {
                    "module": r_step["id"],
                    "control": r_step["key"],
                    "value": r_step["value"]
                })
                print(f"  SET   {r_step.get('id','?')}.{r_step.get('key','?')} = {r_step.get('value','?')}")
            except Exception as e:
                print(f"  FAIL  reset {r_step.get('name','?')}: {e}", file=sys.stderr)
                results["passed"] = False
                results["reset_failed"] = f"{r_step.get('name','?')}: {e}"
                # Stop the scenario before collect_metrics: the reading would
                # otherwise reflect an unknown/partial state. No cleanup
                # needed: created_modules only fills inside the steps loop
                # below, which hasn't run yet.
                return results

    # The starting reading, taken AFTER reset so it reflects the normalized state.
    start = collect_metrics(client, settle_s=settle_s)
    print(f"\n  Start: tick={start.get('tickTimeUs', '?')}us (FPS={start.get('fps', '?')})  heap={start.get('freeHeap', '?')}")

    # ids whose optional state document was skipped (a platform-gated module absent on this
    # target: e.g. the Parlio driver on a non-P4 board). A later optional measure or control
    # that names a skipped id is itself skipped, so an absent driver leaves no trace rather
    # than failing the run. (perf_full's add/measure/remove driver triples are all optional.)
    skipped_ids = set()

    for step_index, step in enumerate(scenario.get("steps", [])):
        step_name = step.get("name", "?")
        op = step.get("op", "")
        fstep = _fill(step, ctx) if ctx else step   # a filled copy, so credentials never reach the file the run writes back
        step_result = {"name": step_name, "op": op}

        # A step naming a board runs there, with that board's restore bookkeeping.
        dev = step.get("device")
        if dev and (dev not in board_state or op not in ("apply_state", "set_control", "measure")):
            print(f"  FAIL  {step_name}: device {dev!r} is not in `boards`, or {op} does not run on another board")
            results["passed"] = False
            break
        client, prior_controls, created_modules = board_state[dev] if dev else home

        # An optional measure/control on a module whose optional add was skipped is a
        # no-op: the module isn't there to measure. Skip before any REST call.
        if step.get("optional") and step.get("id") in skipped_ids and op in ("measure", "set_control"):
            step_result["status"] = "ok"
            print(f"  {op:5} {step.get('id','?')}: skipped (optional, module not present on {target})")
            results["steps"].append(step_result)
            continue

        try:
            if op == "write_file":
                # Stage a script file, the same op the desktop runner has. Without it a migrated
                # scenario ran its set_control against a file that was never written, and every
                # script step failed on hardware while passing on the desktop.
                # A failed write FAILS THE SCENARIO. Every later step runs against a stale or
                # absent file, so reporting PASSED afterwards is the silent pass this op exists
                # to prevent: the same rule the desktop runner applies.
                path_ = step.get("path")
                # `from` stages a file of the repository, such as a shipped script, so its source keeps one home.
                body = (ROOT / step["from"]).read_text(encoding="utf-8") if "from" in step else step.get("value", "")
                if not path_:
                    print(f"  WRITE {step_name}: missing path")
                    step_result["status"] = "skipped" if step.get("optional") else "error"
                    if not step.get("optional"):
                        step_result["error"] = "write_file step has no `path`"
                        results["passed"] = False
                else:
                    try:
                        # Only a file the scenario creates is removed at the end: one that was already there is the device's own.
                        try:
                            before = client.get_text(f"/api/file?path={urllib.parse.quote(path_)}")
                            existed = True
                            prior_files.setdefault(path_, before)
                        except urllib.error.HTTPError as he:
                            existed = he.code != 404
                        resp = client.post_text(f"/api/file?path={urllib.parse.quote(path_)}", body)
                        # The device answers {"ok":true}; anything else is a failure, including a
                        # body that is not the JSON object this endpoint documents.
                        ok = isinstance(resp, dict) and resp.get("ok") is True
                        if ok and not existed:
                            created_files.append(path_)
                        step_result["status"] = "ok" if ok else "error"
                        if ok:
                            print(f"  WRITE {path_} ({len(body)} bytes)")
                        else:
                            step_result["error"] = f"unexpected response: {resp!r}"
                            print(f"  WRITE {path_}: FAILED: unexpected response {resp!r}")
                            results["passed"] = False
                    except Exception as we:
                        step_result["status"] = "error"
                        step_result["error"] = str(we)
                        print(f"  WRITE {path_}: FAILED: {we}")
                        results["passed"] = False

            elif op == "delete_file":
                # Remove a file the scenario staged, so a run leaves the device as it found it.
                # A file already gone counts as removed, the same rule the desktop runner applies.
                path_ = step.get("path")
                if not path_:
                    print(f"  DELETE {step_name}: missing path")
                    step_result["status"] = "error"
                    step_result["error"] = "delete_file step has no `path`"
                    results["passed"] = False
                else:
                    if path_ in created_files:
                        created_files.remove(path_)   # removed here, so the end-of-run cleanup leaves it be
                    try:
                        client.delete(f"/api/dir?path={urllib.parse.quote(path_)}")
                    except Exception:
                        pass   # a 500 for a file that was never there is the case this tolerates
                    # Only a 404 proves it gone: an existing file answers with its text, which is not JSON.
                    gone = False
                    try:
                        client.get(f"/api/file?path={urllib.parse.quote(path_)}")
                    except urllib.error.HTTPError as he:
                        gone = he.code == 404
                    except Exception:
                        pass
                    if gone:
                        step_result["status"] = "ok"
                        print(f"  DELETE {path_}")
                    else:
                        step_result["status"] = "error"
                        step_result["error"] = f"{path_} is still there"
                        print(f"  DELETE {path_}: FAILED: still there")
                        results["passed"] = False

            elif op == "reboot" and step.get("wait") is False:
                # Restart without waiting, for a device that comes back somewhere the host cannot reach yet, such as on its own access point.
                # A refusal is retried, since an answering device is up; its own words are kept for the report.
                refusal = ""
                for _ in range(3):
                    try:
                        client.post("/api/reboot", {})
                        refusal = ""
                        break
                    except urllib.error.HTTPError as re_:
                        refusal = f"/api/reboot returned HTTP {re_.code}: {re_.read().decode('utf-8', 'replace')[:120]}"
                        time.sleep(1)
                    except Exception:
                        refusal = ""
                        break   # the device goes away mid-response
                if refusal:
                    raise RuntimeError(refusal)
                step_result["status"] = "ok"
                print("  REBOOT: not waiting")

            elif op == "reboot":
                # Restart and wait, so a later expect_control proves what SURVIVED rather than what
                # is merely still in memory. The one op that can tell a written setting from a kept one.
                why = _reboot_and_wait(client, target, float(step.get("timeout", 60)))
                if why:
                    step_result["status"] = "error"
                    print(f"  REBOOT: {why}")
                    results["passed"] = False
                else:
                    step_result["status"] = "ok"
                    print("  REBOOT: back up")

            elif op == "expect_file":
                # Read a file back, which is the only way to prove a write reached the filesystem
                # rather than a cache: the card that owns the filesystem is proven by its contents.
                path_ = step["path"]
                # `contains` matches a substring, `equals` the whole file: the two tiers agree, and
                # neither accepts a step that names neither, which would pass on any file at all.
                exact = "equals" in step
                if not exact and "contains" not in step:
                    step_result["status"] = "error"
                    print(f"  EXPECT {path_}: needs `contains` or `equals`")
                    results["passed"] = False
                    continue
                want = str(step["equals"] if exact else step["contains"])
                try:
                    raw = client.get_text(f"/api/file?path={urllib.parse.quote(path_, safe='/')}")
                except Exception as fe:
                    step_result["status"] = "error"
                    print(f"  EXPECT {path_}: could not read ({fe})")
                    results["passed"] = False
                else:
                    holds = (raw == want) if exact else (bool(want) and want in raw)
                    step_result["status"] = "ok" if holds else "error"
                    verb = "is" if exact else "holds"
                    print(f"  EXPECT {path_} {verb if holds else 'does not ' + verb} {want!r}")
                    if not holds:
                        results["passed"] = False

            elif op == "apply_state":
                # A state document in one PATCH /api/state; `error` and `at` name the failure a step expects, and without them it must apply.
                # The modules it creates join the cleanup, so a live device is left as found outside the snapshot containers too.
                absent = []
                for name in _typed_names(fstep["document"]):
                    try:
                        client.get(_mod_path(name))
                    except urllib.error.HTTPError:
                        absent.append(name)
                try:
                    resp = client.patch("/api/state", fstep["document"])
                    got_error, got_at = None, None
                except urllib.error.HTTPError as he:
                    body = _error_body(he)
                    got_error, got_at = body.get("error") or f"HTTP {he.code}", body.get("at")
                for name in absent:
                    try:
                        client.get(_mod_path(name))
                        created_modules.append(name)
                    except urllib.error.HTTPError:
                        pass   # the document did not get as far as creating it
                # An optional document names a type this target may lack, such as another chip's peripheral: a skip, and its modules' later optional steps skip too.
                if step.get("optional") and "error" not in step and got_error == "unknown type":
                    skipped_ids.update(_typed_names(fstep["document"]))
                    step_result["status"] = "ok"
                    print(f"  STATE {step['name']}: skipped (optional, a type unavailable on {target})")
                    results["steps"].append(step_result)
                    continue
                if "error" in step:
                    holds = got_error == step["error"] and ("at" not in step or got_at == step["at"])
                else:
                    holds = got_error is None
                step_result["status"] = "ok" if holds else "error"
                print(f"  STATE {step['name']}: {'applied' if got_error is None else got_error + ' at ' + str(got_at)}{'' if holds else '  (expected ' + str(step.get('error', 'success')) + ')'}")
                if not holds:
                    results["passed"] = False

            elif op == "round_trip_state":
                # Every card read back as its document and sent as one: what the device shows applies as it is, and the body outgrows the request buffer, so the streaming route runs.
                whole = {}
                for mod in client.get("/api/state").get("modules", []):
                    whole.update(client.get(_mod_path(mod["name"]) + "/document"))
                try:
                    client.patch("/api/state", whole)
                    failed = None
                except urllib.error.HTTPError as he:
                    body = _error_body(he)
                    failed = f"{body.get('error') or 'HTTP ' + str(he.code)} at {body.get('at')}"
                step_result["status"] = "ok" if failed is None else "error"
                print(f"  STATE {step['name']}: {len(whole)} cards, {len(json.dumps(whole))} bytes, {'applied' if failed is None else failed}")
                if failed is not None:
                    results["passed"] = False

            elif op == "expect_control":
                # Assert a control reads what the scenario says it must, the only op that fails a
                # scenario on a VALUE rather than on a timing contract. It exists for the strings a
                # rename would change silently, where nothing else can see the break.
                # Read through /api/modules/<id>, the same view a client gets, so an assertion can
                # never pass against a value the device would report differently.
                mod_id, key = step["id"], step["key"]
                # JSON spells a boolean `true`, Python spells it `True`, and the in-process runner
                # renders the JSON form: comparing str() of either would make one tier disagree
                # with the other about the same scenario.
                def _as_written(v):
                    return {True: "true", False: "false"}.get(v, str(v)) if isinstance(v, bool) else str(v)
                # `not_equals` pins a value that moves per release, where the only stable claim is
                # that it is not the empty string a missing source would render. Both tiers agree.
                negated = "equals" not in step and "not_equals" in step
                if "equals" not in step and not negated:
                    step_result["status"] = "error"
                    print(f"  EXPECT {mod_id}.{key}: needs `equals` or `not_equals`")
                    results["passed"] = False
                    continue
                want = _as_written(fstep["not_equals" if negated else "equals"])
                # `within` polls until it holds, for a value the device reaches on its own time, such as after a join.
                deadline = time.time() + float(step.get("within", 0))
                while True:
                    read_error = None
                    try:
                        got = _control_value(client.get(_mod_path(mod_id)), key)   # a control, or the module's own `status` or `enabled`
                    except Exception as ce:
                        got, read_error = None, ce
                    holds = got is not None and ((_as_written(got) != want) if negated else (_as_written(got) == want))
                    if holds or time.time() >= deadline:
                        break
                    time.sleep(1)
                if read_error is not None:
                    step_result["status"] = "error"
                    print(f"  EXPECT {mod_id}.{key}: could not read ({read_error})")
                    results["passed"] = False
                    continue
                if got is None:
                    step_result["status"] = "error"
                    print(f"  EXPECT {mod_id}.{key}: no such control")
                    results["passed"] = False
                elif (_as_written(got) != want) if negated else (_as_written(got) == want):
                    step_result["status"] = "ok"
                    print(f"  EXPECT {mod_id}.{key} {'!=' if negated else '=='} {want}")
                else:
                    step_result["status"] = "error"
                    print(f'  EXPECT {mod_id}.{key} is "{_as_written(got)}", '
                          f'expected {"not " if negated else ""}"{want}"')
                    results["passed"] = False

            elif op == "set_control":
                data = {"module": fstep["id"], "control": fstep["key"],
                        "value": fstep["value"]}
                _remember_control(prior_controls, client, step["id"], step["key"])
                try:
                    resp = client.post("/api/control", data)
                    step_result["status"] = "ok" if resp.get("ok") else "error"
                    print(f"  SET   {step.get('id', '?')}.{step.get('key', '?')} = {step.get('value', '?')}")
                except urllib.error.HTTPError as ce:
                    if step.get("optional") and ce.code == 404:
                        # An `optional` set_control on a missing module (e.g. shrink
                        # a grid a prior run's cleanup removed) is a no-op, not a fail.
                        step_result["status"] = "ok"
                        print(f"  SET   {step.get('id','?')}.{step.get('key','?')}: skipped (optional, not present)")
                    elif step.get("optional") and ce.code == 400:
                        # An `optional` set_control the device REJECTS with 400 (e.g.
                        # "value out of range") is a not-available-here skip, not a
                        # failure: e.g. selecting a `peripheral` value this chip doesn't
                        # offer (Parlio on an S3, MoonI80 on a classic): the Select's max
                        # is the board-filtered option count, so the value is out of range
                        # and returns 400. A REQUIRED set_control that 400s still fails.
                        step_result["status"] = "skipped"
                        # And the MODULE is unavailable from here on. Marking only the step left
                        # every later measure running against a module configured for a peripheral
                        # this chip does not have: the numbers came out, looked like data, and
                        # described a configuration that never applied. skipped_ids is the same set
                        # an optional add uses, so the measures already know to skip it.
                        skipped_ids.add(step.get("id", ""))
                        print(f"  SET   {step.get('id','?')}.{step.get('key','?')} = {step.get('value','?')}: skipped (optional, value not offered on this target; later steps on it skip too)")
                    elif ce.code == 404:
                        # Transient: a set_control issued right after a structural
                        # change (replace/add) can race the device's prepareTree and
                        # briefly see "module not found" while the tree rebuilds.
                        # Settle and retry once before treating it as a real failure.
                        time.sleep(1.0)
                        resp = client.post("/api/control", data)
                        step_result["status"] = "ok" if resp.get("ok") else "error"
                        print(f"  SET   {step.get('id','?')}.{step.get('key','?')} = {step.get('value','?')} (retried)")
                    else:
                        raise
                # If this step doesn't measure (so `collect_metrics` won't wait
                # for us), still give the device a moment: a set_control that
                # triggers prepareTree briefly mutates the module tree, and the
                # very next API call can hit a transient "module not found".
                # 500 ms is empirically enough on the classic board; cheap insurance.
                if not (step.get("measure") or op == "measure"):
                    time.sleep(0.5)

            elif op == "host_wifi":
                # Move the host onto the device's access point or back to the registry network, and the runner's target with it.
                if net is None:
                    raise RuntimeError("host_wifi needs `host_network` on the scenario and --network on the run")
                if step.get("join") == "access_point":
                    if not ctx.get("access_point"):
                        raise RuntimeError("the access point's name is unknown, so it cannot be joined")
                    host.home = host.home or net
                    why = _host_join(ctx.get("access_point", ""), fstep.get("password", ""), f"http://{ACCESS_POINT_ADDRESS}/api/system",
                                     float(step.get("timeout", 60)))
                    if not why:
                        client.base = f"http://{ACCESS_POINT_ADDRESS}"
                else:
                    client.base = home_base
                    why = _host_join(net["ssid"], net["password"], f"{home_base}/api/system",
                                     float(step.get("timeout", 60)))
                    if not why:
                        host.home = None   # home: nothing left for the exit hook to restore
                step_result["status"] = "error" if why else "ok"
                print(f"  HOST  {'FAILED: ' + why if why else 'on ' + (ctx.get('access_point', '') if step.get('join') == 'access_point' else net['ssid'])}")
                if why:
                    results["passed"] = False

            elif op == "list_row" and "add" in step:
                # Add a row with the given fields, and with `to` move it to that position, as the card's Add and drag do.
                mod_id, key = step["id"], step["key"]
                route = f"/api/list/{urllib.parse.quote(mod_id, safe='')}/{urllib.parse.quote(key, safe='')}"
                row_id = client.post(route, {}).get("id")
                if row_id is None:
                    raise RuntimeError(f"{mod_id}.{key} gave no id for the new row")
                for field, value in fstep["add"].items():
                    client.patch(f"{route}/{row_id}", {"field": field, "value": value})
                if "to" in step:
                    client.patch(f"{route}/{row_id}", {"to": step["to"]})
                step_result["status"] = "ok"
                print(f"  ROW   {mod_id}.{key} added {step['add']}{' at ' + str(step['to']) if 'to' in step else ''}")

            elif op == "list_row":
                # Find a row by its fields, then write one of its fields, press one of its buttons, apply it, delete it, or, with none of those, only expect it.
                mod_id, key, match = step["id"], step["key"], fstep["match"]
                deadline = time.time() + float(step.get("within", 0))
                while True:
                    try:
                        rows = _list_rows(client, mod_id, key)
                    except Exception:
                        rows = []
                    row = next((r for r in rows if all(str(r.get(k)) == str(v) for k, v in match.items())), None)
                    if row is not None or time.time() >= deadline:
                        break
                    time.sleep(1)
                where = f"{mod_id}.{key}[{step.get('match')}]"
                if row is None:
                    step_result["status"] = "ok" if step.get("optional") else "error"
                    print(f"  ROW   {where}: {'absent, skipped (optional)' if step.get('optional') else 'no such row'}")
                    if not step.get("optional"):
                        results["passed"] = False
                else:
                    route = f"/api/list/{urllib.parse.quote(mod_id, safe='')}/{urllib.parse.quote(key, safe='')}/{row['id']}"
                    if step.get("delete"):
                        client.delete(route)
                        print(f"  ROW   {where} deleted")
                    elif step.get("apply"):
                        client.post(f"{route}/apply", {})
                        print(f"  ROW   {where} applied")
                    elif "field" not in step:
                        print(f"  ROW   {where} present")
                    else:
                        client.patch(route, {"field": step["field"], "value": fstep.get("value", "")})
                        print(f"  ROW   {where}.{step['field']} set")
                    step_result["status"] = "ok"

            elif op == "expect_http":
                # Redirects are not followed, so a captive portal's own answer is what is checked.
                class _NoRedirect(urllib.request.HTTPRedirectHandler):
                    def redirect_request(self, *args, **kwargs):
                        return None
                url = fstep["url"]
                request = _http_request(url, client.base, step.get("method", "GET"))
                if step.get("resolve") == "access_point":
                    # Resolve at the access point's own DNS, as a phone does: macOS keeps a captive network's answers from ordinary apps until its sign-in completes.
                    parts = urllib.parse.urlsplit(url)
                    dig = subprocess.run(["dig", "+time=2", "+tries=3", "+short", f"@{ACCESS_POINT_ADDRESS}", parts.hostname or ""],
                                         capture_output=True, text=True)
                    resolved = (dig.stdout.strip().splitlines() or ["0.0.0.0"])[-1]
                    print(f"  DNS   {ACCESS_POINT_ADDRESS} resolves {parts.hostname} to {resolved}")
                    request = urllib.request.Request(urllib.parse.urlunsplit(parts._replace(netloc=resolved)),
                                                     data=request.data, method=request.get_method(),
                                                     headers={"Host": parts.netloc})
                # `within` polls too: a host that has joined a network resolves names only once its resolver took that network's server.
                deadline = time.time() + float(step.get("within", 0))
                while True:
                    body = ""
                    try:
                        with urllib.request.build_opener(_NoRedirect).open(request, timeout=10) as resp:
                            status, location = resp.status, resp.headers.get("Location", "")
                            body = resp.read().decode("utf-8", "replace")
                    except urllib.error.HTTPError as he:
                        status, location = he.code, he.headers.get("Location", "")
                    except Exception as ue:
                        status, location = 0, str(ue)
                    # `body_matches` and `body_lacks` are regular expressions the response must, or must not, contain.
                    holds = (status == step["status"] and ("location" not in step or location == fstep["location"])
                             and ("body_matches" not in step or re.search(step["body_matches"], body) is not None)
                             and ("body_lacks" not in step or re.search(step["body_lacks"], body) is None))
                    if holds or time.time() >= deadline:
                        break
                    time.sleep(1)
                if not holds and status == 0:
                    # A name that did not resolve: show which servers the host asks, and what the device's own address answers.
                    name = urllib.parse.urlsplit(url).hostname or ""
                    servers = subprocess.run(["scutil", "--dns"], capture_output=True, text=True).stdout
                    print("  DNS   host asks: " + ", ".join(sorted({l.split(":", 1)[1].strip() for l in servers.splitlines() if "nameserver[" in l})))
                    dig = subprocess.run(["dig", "+time=2", "+tries=1", "+short", f"@{ACCESS_POINT_ADDRESS}", name], capture_output=True, text=True)
                    print(f"  DNS   {ACCESS_POINT_ADDRESS} answers {name}: {(dig.stdout or dig.stderr).strip() or 'nothing'}")
                step_result["status"] = "ok" if holds else "error"
                print(f"  HTTP  {step.get('method', 'GET')} {url} → {status}{' ' + location if location else ''}{'' if holds else '  (expected ' + str(step['status']) + ' ' + fstep.get('location', '') + ')'}")
                if not holds:
                    results["passed"] = False

            elif op == "browser":
                # A real browser on the device's UI, for what a page load and an open card cost the device: open, refresh, close.
                action = step.get("action", "open")
                if action == "close":
                    _browser_close()
                else:
                    page = _browser_page()
                    if action == "open":
                        page.goto(client.base + "/", wait_until="domcontentloaded")
                    else:
                        page.reload(wait_until="domcontentloaded")
                    page.wait_for_timeout(int(float(step.get("settle", 5)) * 1000))
                print(f"  BROWSER {action}")
                step_result["status"] = "ok"

            elif op == "measure":
                # Pure measurement step (introduced for the build-up scenario shape).
                # No REST call; the measure block below picks it up via step["measure"]
                # or the implicit-measure clause we add to the same dispatcher.
                step_result["status"] = "ok"

            else:
                step_result["status"] = "skipped"
                print(f"  SKIP  {step_name} (unknown op: {op})")

        except urllib.error.HTTPError as e:
            # Read the JSON error body for a friendly message
            try:
                body = json.loads(e.read())
                msg = body.get("error", str(e))
            except Exception:
                msg = str(e)
            step_result["status"] = "error"
            step_result["error"] = msg
            # Every rejected step is a real failure: a WARN would turn a rejected
            # step into a meaningless pass.
            print(f"  FAIL  {step_name}: {msg}")
            results["passed"] = False
        except Exception as e:
            step_result["status"] = "error"
            step_result["error"] = str(e)
            print(f"  FAIL  {step_name}: {e}")
            results["passed"] = False

        # Measure after this step if requested (explicit "measure": true OR op == "measure").
        # Skip the measurement block when the step itself failed: writing observed
        # values from a failed-step state would persist garbage as the "latest
        # reading" and the contract assertion would compare against an
        # untrustworthy measurement.
        if step_result.get("status") == "error":
            results["steps"].append(step_result)
            continue
        if step.get("measure") or op == "measure":
            # collect_metrics hits /api/state: a missing measurement is a
            # failed run, not a no-op to skip. Silent-skip would let a broken
            # device pass a scenario that asserts on observed/contract data
            # the step never gathered. Fail loudly, record the error on the
            # step, and break out of the step loop so end-of-run cleanup
            # (delete created modules) still fires.
            try:
                metrics = collect_metrics(client, float(step.get("settle", settle_s)))   # a step that needs longer to reach its steady state says so
            except Exception as e:
                print(f"  FAIL  {step_name}: collect_metrics failed: {e}")
                step_result["status"] = "error"
                step_result["error"] = f"collect_metrics: {e}"
                step_result["metrics"] = {}
                results["passed"] = False
                results["steps"].append(step_result)
                break  # stop step loop; cleanup runs below
            step_result["metrics"] = metrics
            # `read` names extra values to report: "Module.control" from the state, or "Module.tick" for a module's tick time; `samples` reads them that many seconds and keeps each one's median.
            if step.get("read"):
                boards_now = {b: s[0] for b, s in board_state.items()}
                reads = [_read_values(client, metrics, step["read"], boards_now)]
                for _ in range(int(step.get("samples", 1)) - 1):
                    time.sleep(1.0)
                    try:
                        system = client.get("/api/system")
                    except Exception:
                        system = {}
                    reads.append(_read_values(client, system, step["read"], boards_now))
                step_result["read"] = median_reads(reads)
                print("  READ     " + "  ".join(f"{k}={v}" for k, v in step_result["read"].items()))
            tick_us = metrics.get("tickTimeUs", 0)
            fps = 1000000 // tick_us if tick_us > 0 else metrics.get("fps", 0)
            heap = metrics.get("freeHeap", 0)
            max_block = metrics.get("maxBlock", 0)
            model_bytes = metrics.get("dynamicBytesTotal")
            print(f"  MEASURE  tick={tick_us}us (FPS={fps})  heap={heap}  internal={metrics.get('freeInternal')}  "
                  f"block={max_block}  stack={metrics.get('minStackFree')} ({metrics.get('minStackTask')})  model={model_bytes}")

            # Per-step contract: { "contract": { "<target>": { "tick_us": N,
            #   "free_heap": M, "tick_tolerance_pct": P, "heap_tolerance_pct": Q,
            #   "set_by": "YYYY-MM-DD", "reason": "..." } } }
            # Contracts are hand-set promises: see docs/reference/testing.md § Performance
            # contracts. `--update-contract --reason "..."` rewrites them.
            contract_block = step.get("contract", {}).get(obs_target) if step.get("contract") else None
            if contract_block:
                # Defaults reflect run-to-run variance, not "I don't care":
                #   pc-*      : multi-process OS jitter, 20% pct + 200us absolute
                #                floor. The floor dominates below ~1ms tick (the
                #                realistic case for desktop scenarios today).
                #   esp32-*   : bounded RTOS but lwIP/EMAC jitter, 10% pct + 5us
                #                absolute floor.
                # KEEP IN SYNC: the in-process runner re-declares the same defaults
                # at test/scenario_runner.cpp contract-block handler: tuning one
                # without the other silently desyncs the two tiers.
                is_desktop = target.startswith("desktop-")
                tick_tol_pct = contract_block.get("tick_tolerance_pct",
                                                  20 if is_desktop else 10)
                heap_tol_pct = contract_block.get("heap_tolerance_pct",
                                                  20 if is_desktop else 10)
                tol_us_abs = contract_block.get("tolerance_us", 200 if is_desktop else 5)
                exp_tick = contract_block.get("tick_us")
                exp_heap = contract_block.get("free_heap")
                if exp_tick is not None and exp_tick > 0:
                    # tick contract is a *ceiling*: faster than contract is good
                    # news (mirror of heap being a floor). Tolerance absorbs
                    # upward jitter only; speedups never fail.
                    overshoot = tick_us - exp_tick
                    allowed = max(exp_tick * tick_tol_pct / 100.0, tol_us_abs)
                    if overshoot <= 0:
                        print(f"  PASS  tick {tick_us}us <= contract {exp_tick}us "
                              f"(margin {-overshoot:.0f}us)")
                    elif overshoot > allowed:
                        print(f"  FAIL  tick {tick_us}us vs contract {exp_tick}us "
                              f"(over by {overshoot:.0f}us > allowed {allowed:.0f}us)")
                        results["passed"] = False
                    else:
                        print(f"  PASS  tick {tick_us}us vs contract {exp_tick}us "
                              f"(over by {overshoot:.0f}us within {allowed:.0f}us)")
                # Floors: the device must deliver at least this much, less by more than the tolerance being a regression.
                # Free heap and the largest block drift with lwIP's pools, so they carry the heap tolerance; a stack's headroom is exact.
                # The block floor is what an allocator needing one contiguous chunk depends on, and free internal RAM is what WiFi and HTTP need on a board with PSRAM.
                floors = (("free_heap", heap, exp_heap, heap_tol_pct),
                          ("max_alloc_block", max_block, contract_block.get("max_alloc_block"), heap_tol_pct),
                          ("free_internal", metrics.get("freeInternal"), contract_block.get("free_internal"), heap_tol_pct),
                          ("min_stack_free", metrics.get("minStackFree"), contract_block.get("min_stack_free"), 0))
                for metric, value, floor, tol in floors:
                    if floor is None or floor <= 0:
                        continue
                    ok, line = _floor_verdict(metric, value, floor, tol)
                    print(line + (f" ({metrics.get('minStackTask')})" if metric == "min_stack_free" else ""))
                    if not ok:
                        results["passed"] = False

            sample = {
                "tick_us": int(tick_us),
                "free_heap": int(heap),
                "max_alloc_block": int(max_block),
            }
            for metric, key in (("free_internal", "freeInternal"), ("min_stack_free", "minStackFree")):
                if metrics.get(key) is not None:
                    sample[metric] = int(metrics[key])
            # The window is this machine's, kept in the local store; the scenario file takes the summary only when asked.
            local = _observed.record(scenario.get("name", scenario_path.stem), step.get("name", "?"), obs_target, sample, _today_iso(), reset_window=update_contract)
            if update_observed or update_contract:
                step.setdefault("observed", {})[obs_target] = _observed.summary(local)
                wrote_observations[0] = True

            # --update-contract: rewrite the contract in the scenario JSON for the
            # active target. This is *renegotiating* a contract, not refreshing a
            # last-reading baseline: set_by + reason are stamped so the diff
            # records when and why the promise changed. Caller is responsible for
            # committing the diff intentionally.
            #
            # Originals are stashed in pending_contract_originals so the
            # post-run gate (see below) can roll the in-memory tree back to
            # disk shape if the run failed: only successful runs get to
            # commit a renegotiated promise.
            if update_contract:
                # Preserve any per-step tolerance overrides already in place.
                # Key by step INDEX, not the step dict: a dict is unhashable, so
                # `(step, target)` as a key raised TypeError (this whole path is
                # only reached with --update-contract, which the gates don't pass).
                existing = step.get("contract", {}).get(obs_target, {})
                if (step_index, obs_target) not in pending_contract_originals:
                    # Deep enough copy: existing is a flat dict of scalars.
                    pending_contract_originals[(step_index, obs_target)] = (
                        dict(existing) if existing else None
                    )
                new_block = {
                    "tick_us": int(tick_us),
                    "free_heap": int(heap),
                    "set_by": _today_iso(),
                    "reason": update_reason or existing.get("reason", "updated"),
                }
                for k in ("tick_tolerance_pct", "heap_tolerance_pct", "tolerance_us"):
                    if k in existing:
                        new_block[k] = existing[k]
                # max_alloc_block: opt-in (only carry it over if the existing
                # contract had it), but refresh the value from this run rather
                # than copying the stale one. Mirrors run_scenario.py's update
                # path: keep both files in sync if you change one.
                if "max_alloc_block" in existing:
                    new_block["max_alloc_block"] = int(max_block)
                # The internal-RAM and stack floors are requirements a person sets, not readings, so a renegotiation keeps them.
                for k in ("free_internal", "min_stack_free"):
                    if k in existing:
                        new_block[k] = existing[k]
                step.setdefault("contract", {})[obs_target] = new_block

        results["steps"].append(step_result)

    # The host back on its own network before anything else talks to the device, which is then there.
    if host.home:
        host.restore()
    client, prior_controls, created_modules = home
    client.base = home_base

    # The other boards first, so a sender stops before this device is put back.
    for b, (bc, b_prior, b_created) in board_state.items():
        for module_id in reversed(b_created):
            try:
                bc.delete(_mod_path(module_id))
                print(f"  -     {module_id} ({b} cleanup)")
            except Exception:
                pass
        for doc in board_snapshots[b]:
            try:
                bc.patch("/api/state", doc)
            except Exception as e:
                print(f"  WARN: couldn't restore {b}'s {next(iter(doc), '?')}: {e}")
        put_back = _restore_controls(bc, b_prior)
        if put_back:
            print(f"  restored {put_back} control(s) on {b}")

    # `--keep` leaves the device as the last step set it, files and all, to watch or deploy what the scenario built.
    if not keep:
        # Cleanup: delete modules that were created by this scenario
        for module_id in reversed(created_modules):
            try:
                client.delete(_mod_path(module_id))
                print(f"  -     {module_id} (cleanup)")
            except Exception:
                pass

        # The controls it wrote, put back first: a control's prior value is read at its first write, after the fixture's canvas, so inside a snapshot container it is the fixture's value.
        put_back = _restore_controls(client, prior_controls)
        if put_back:
            print(f"  restored {put_back} control(s) the scenario had set")
        # Then each container's document as it was, last since it holds every value of its tree: it removes, re-creates and re-types until the tree is the one the run started with.
        for doc in snapshot_docs:
            try:
                client.patch("/api/state", doc)
            except Exception as e:
                print(f"  WARN: couldn't restore {next(iter(doc), '?')}: {e}")
        # And the files it created that no step removed, after the controls, so nothing still selects a file as it goes.
        for path_ in created_files:
            try:
                client.delete(f"/api/dir?path={urllib.parse.quote(path_)}")
                print(f"  -     {path_} (cleanup)")
            except Exception:
                pass
        for path_, before in prior_files.items():
            try:
                client.post_text(f"/api/file?path={urllib.parse.quote(path_)}", before)
                print(f"  ~     {path_} (restored)")
            except Exception as e:
                print(f"  WARN  restore of {path_} failed: {e}")

    # Write the scenario JSON back if anything changed:
    #   - observed.<target> was updated by any measure step (every run); OR
    #   - --update-contract renegotiated the contract: AND the run passed
    #     (don't persist a renegotiated promise from a half-broken run; the
    #     observed values still land so drift is visible either way).
    # If the contract was renegotiated but the run failed, the in-memory
    # mutations stay in `scenario` only until the process exits; the on-disk
    # contract is preserved.
    contract_safe_to_write = update_contract and results["passed"]
    if not contract_safe_to_write and update_contract:
        # Revert any contract mutations we made to the in-memory tree so the
        # JSON write below (for observed) doesn't leak them to disk.
        for step_index, step in enumerate(scenario.get("steps", [])):
            contract = step.get("contract")
            if contract and obs_target in contract:
                if (step_index, obs_target) in pending_contract_originals:
                    orig = pending_contract_originals[(step_index, obs_target)]
                    if orig is None:
                        del contract[obs_target]
                    else:
                        contract[obs_target] = orig
        print(f"  contract[{obs_target}] NOT written (run failed)")

    if wrote_observations[0] or contract_safe_to_write:
        _observed.save_scenario(scenario_path, scenario)
        what = []
        if wrote_observations[0]:
            what.append(f"observed[{obs_target}]")
        if contract_safe_to_write:
            what.append(f"contract[{obs_target}]")
        print(f"  WROTE  {scenario_path.name} ({' + '.join(what)})")

    # The values each measure step read, one row per step, for comparing configurations at a glance.
    rows = [s for s in results["steps"] if s.get("read")]
    if rows:
        keys = list(dict.fromkeys(k for s in rows for k in s["read"]))
        width = max(len(s["name"]) for s in rows)
        print("\n  " + "step".ljust(width) + "  tick_us  " + "  ".join(keys))
        for s in rows:
            tick = (s.get("metrics") or {}).get("tickTimeUs", "?")
            print("  " + s["name"].ljust(width) + f"  {str(tick):>7}  " + "  ".join(str(s["read"].get(k, "-")).rjust(len(k)) for k in keys))

    # Summary
    print("\n---")
    if results["passed"]:
        print("PASSED")
    else:
        print("FAILED")

    return results


def main():
    parser = argparse.ArgumentParser(description="Run live scenario tests")
    parser.add_argument("--host", default="localhost:8080",
                        help="Device host:port (default: localhost:8080)")
    parser.add_argument("--name", default=None,
                        help="Scenario name (without .json), or several separated by commas. Runs all if omitted.")
    parser.add_argument("--module", default=None,
                        help="Filter to scenarios whose top-level module / also matches.")
    parser.add_argument("--network", default=None,
                        help="Registry network (moondeck.json) the host and device share, for a scenario that moves the host's WiFi.")
    parser.add_argument("--board", action="append", default=[], metavar="NAME=HOST",
                        help="Another board a scenario names in `boards`, such as sender=192.168.1.156; repeat for each.")
    parser.add_argument("--keep", action="store_true",
                        help="Leave the device as the last step set it, rather than restoring it: to watch or deploy what a scenario built.")
    parser.add_argument("--settle", type=float, default=3.0,
                        help="Settle time in seconds between step and measurement")
    parser.add_argument("--update-observed", action="store_true",
                        help=("Write the statistics of this machine's sample window into the scenario file as its published summary. "
                              "Without it a run keeps its sample window on this machine and leaves the file alone."))
    parser.add_argument("--update-contract", action="store_true",
                        help=("Renegotiate the per-step performance contract: write "
                              "observed tick/heap into contract[<target>] and stamp "
                              "set_by + reason. Requires --reason. Overwrites existing "
                              "values for the active target only; other targets untouched."))
    parser.add_argument("--reason", default=None,
                        help=("Why the contract is being renegotiated (required with "
                              "--update-contract). Examples: 'tighter Layer LUT copy', "
                              "'accepted DMX driver overhead'. Written into each updated "
                              "contract block."))
    args = parser.parse_args()

    if args.update_contract and not args.reason:
        parser.error("--update-contract requires --reason "
                     "(e.g. --reason 'tightened after Layer optimization')")

    client = Client(args.host)
    boards = {}
    for spec in args.board:
        bname, sep, bhost = spec.partition("=")
        if not sep or not bname or not bhost:
            parser.error(f"--board takes NAME=HOST, got {spec!r}")
        boards[bname] = Client(bhost)

    # Verify connection
    try:
        state = client.get("/api/state")
        module_count = len(state.get("modules", []))
        print(f"Connected to {args.host} ({module_count} modules)")
    except Exception as e:
        print(f"Cannot connect to {args.host}: {e}")
        sys.exit(1)

    # Find scenarios via the shared metadata module (recursive: scenarios live under core/, light/, …)
    if args.name:
        paths = []
        for name in (n.strip() for n in args.name.split(",") if n.strip()):
            match = test_meta.find_scenario_path(name)
            if not match:
                print(f"Scenario not found: {name}.json under {test_meta.SCENARIO_DIR}")
                sys.exit(1)
            paths.append(match)
    else:
        paths = [s["path"] for s in test_meta.collect_scenario_files()]

    if args.module and args.module.lower() != "all":
        module_paths = set(test_meta.paths_for_module(args.module))
        filtered = [p for p in paths if p in module_paths]
        if not filtered:
            print(f"No scenarios match module: {args.module}")
            sys.exit(1)
        paths = filtered
        print(f"Module filter: {args.module} ({len(paths)} scenario(s))")

    if not paths:
        print("No scenarios found")
        sys.exit(1)

    # Run scenarios
    all_results = {}
    all_passed = True
    for path in paths:
        if not path.exists():
            print(f"Scenario not found: {path}")
            continue
        result = run_scenario(client, path, args.settle,
                              update_contract=args.update_contract, update_observed=args.update_observed,
                              update_reason=args.reason, named=bool(args.name), network=args.network,
                              boards=boards, keep=args.keep)
        all_results[result["name"]] = result
        if not result["passed"]:
            all_passed = False

    print(f"\n=== {len(all_results)} scenario(s), "
          f"{'all passed' if all_passed else 'SOME FAILED'} ===")
    sys.exit(0 if all_passed else 1)


if __name__ == "__main__":
    main()