From dea95bf526c74bbfa02cfd71711e818c6cab6f52 Mon Sep 17 00:00:00 2001
From: Vuong Hoang <vh@phasefinal.com>
Date: Fri, 24 Apr 2026 10:09:16 -0700
Subject: [PATCH] elway: add mini playbook runner + smoke playbook
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`scripts/elway` is a ~600-line Python tool (stdlib + python3-yaml) for
driving one-off ssh commands, ad-hoc file uploads, and YAML playbooks
against a single host. Fills the gap between "single ssh one-liner"
and "reach for Ansible."

Highlights:
  - Three invocation modes: --shell, --upload (LOCAL:REMOTE[:MODE]),
    and --playbook <path>
  - Playbook schema: inline vars, list of steps, optional verify block.
    Template via {{ var }}; CLI --var overrides inline defaults
  - stop_on_fail global (default on), per-step override. Verify phase
    always runs, even after a halt — you see end-state regardless
  - Sudo handled once: probes NOPASSWD; if not, prompts locally via
    getpass, validates up-front, then feeds via `sudo -S` per step.
    Password never written to disk/logs. Upload-with-sudo stages to
    /tmp then sudo-mv + sudo-chmod
  - SSH connection reuse via ControlMaster (60s persist) keeps
    multi-step playbooks responsive (~30ms/step reuse vs ~550ms cold)
  - Live interleaved stdout/stderr with per-step prefix and colored
    pass/fail summary. --dry-run prints the plan without executing
  - Shebang pinned to /usr/bin/python3 to bypass venv-shadowing
    when python3-yaml lives in the system site-packages

Smoke test (playbooks/elway-smoke.yaml) covers vars + upload + verify;
drove out a YAML-scalar-coercion bug before first commit (`shell: false`
parsed to Python bool, crashed the templater — now coerced to string
at load time with a clear error on nulls).
---
 playbooks/elway-smoke.yaml |  33 ++
 scripts/elway              | 620 +++++++++++++++++++++++++++++++++++++
 2 files changed, 653 insertions(+)
 create mode 100644 playbooks/elway-smoke.yaml
 create mode 100755 scripts/elway

diff --git a/playbooks/elway-smoke.yaml b/playbooks/elway-smoke.yaml
new file mode 100644
index 0000000..73ef356
--- /dev/null
+++ b/playbooks/elway-smoke.yaml
@@ -0,0 +1,33 @@
+# elway's own smoke-test playbook. Exercises: vars (inline + CLI override),
+# multi-step flow, upload with mode, stop-on-error semantics, verify phase.
+# Target: any Linux host reachable over ssh that has /tmp and curl.
+
+vars:
+  scratch_dir: /tmp/elway-smoke
+  greeting: hello from elway
+
+steps:
+  - name: Clean any prior scratch dir
+    shell: rm -rf {{ scratch_dir }}
+
+  - name: Create scratch dir
+    shell: mkdir -p {{ scratch_dir }}
+
+  - name: Drop a greeting file
+    shell: echo "{{ greeting }}" > {{ scratch_dir }}/greeting.txt
+
+  - name: Upload this playbook into the scratch dir
+    upload:
+      src: playbooks/elway-smoke.yaml
+      dest: "{{ scratch_dir }}/uploaded.yaml"
+      mode: "0644"
+
+verify:
+  - name: Greeting file has the expected content
+    shell: grep -q "{{ greeting }}" {{ scratch_dir }}/greeting.txt
+
+  - name: Uploaded file is a non-empty yaml
+    shell: test -s {{ scratch_dir }}/uploaded.yaml && head -1 {{ scratch_dir }}/uploaded.yaml
+
+  - name: Scratch dir exists and is listable
+    shell: ls -la {{ scratch_dir }}
diff --git a/scripts/elway b/scripts/elway
new file mode 100755
index 0000000..88fd133
--- /dev/null
+++ b/scripts/elway
@@ -0,0 +1,620 @@
+#!/usr/bin/python3
+# Pinned to /usr/bin/python3 rather than `env python3` so the script is
+# immune to an active user virtualenv shadowing system site-packages
+# (where PyYAML lives on Debian — `apt install python3-yaml`).
+"""
+elway — a mini playbook runner over SSH.
+
+Named for John Elway: quarterbacks run plays. You hand it a play (a single
+shell command or an upload), or a playbook (a YAML list of steps + an
+optional verify phase), and it shoots them across the ssh link with live
+output, structured reporting, and stop-on-error semantics.
+
+Conventions:
+  * Single host per run. Fleet-wide = shell loop outside.
+  * Leverages your ~/.ssh/config aliases. No inventory file.
+  * SSH connection reuse via ControlMaster for multi-step speed.
+  * Sudo password (if needed) prompted once at start, reused via `sudo -S`,
+    never written to disk, never logged.
+  * Simple `{{ var }}` substitution. No Jinja, no filters, no loops.
+
+Quick starts:
+  elway irv-ml1 --shell 'docker compose ls'
+  elway irv-ml1 --shell 'mkdir -p /worktank/foo' --sudo
+  elway irv-ml1 --upload stacks/comfyui/compose.yaml:/opt/docker/compose/comfyui/compose.yaml
+  elway irv-ml1 --playbook playbooks/deploy-stack.yaml --var stack=comfyui
+
+Playbook schema (YAML):
+  vars:                       # optional inline defaults; --var CLI overrides
+    stack: comfyui
+  steps:                      # required list
+    - name: <label>           # required
+      shell: <command>        # exactly one of shell / upload
+      # or
+      upload:
+        src: <local path>
+        dest: <remote path>
+        mode: "0644"          # optional
+      sudo: false             # default false
+      stop_on_fail: true      # optional per-step override of global flag
+  verify:                     # optional; always runs after steps (even on fail)
+    - name: Port responds
+      shell: curl -sf http://localhost:8188/ >/dev/null
+
+Exit codes:
+  0 — every step (+ verify) passed or was ignored
+  1 — a non-ignored step failed
+  2 — usage / config error before execution began
+"""
+
+from __future__ import annotations
+
+import argparse
+import dataclasses
+import getpass
+import json
+import os
+import re
+import shlex
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+from typing import Optional
+
+try:
+    import yaml
+except ImportError:
+    sys.stderr.write(
+        "elway: PyYAML is required. Install with:\n"
+        "    sudo apt install python3-yaml\n"
+        "or, inside a venv:\n"
+        "    pip install pyyaml\n"
+    )
+    sys.exit(2)
+
+
+# ─── Terminal output helpers ──────────────────────────────────────────────
+
+IS_TTY = sys.stdout.isatty() and os.environ.get("NO_COLOR") != "1"
+
+
+def c(code: str, s: str) -> str:
+    return f"\033[{code}m{s}\033[0m" if IS_TTY else s
+
+
+GREEN = lambda s: c("32", s)
+RED = lambda s: c("31", s)
+YELLOW = lambda s: c("33", s)
+CYAN = lambda s: c("36", s)
+DIM = lambda s: c("2", s)
+BOLD = lambda s: c("1", s)
+
+
+# ─── Data model ────────────────────────────────────────────────────────────
+
+
+@dataclasses.dataclass
+class UploadSpec:
+    src: str
+    dest: str
+    mode: Optional[str] = None  # octal string like "0644"
+
+
+@dataclasses.dataclass
+class Step:
+    name: str
+    shell: Optional[str] = None
+    upload: Optional[UploadSpec] = None
+    sudo: bool = False
+    stop_on_fail: Optional[bool] = None  # None = inherit global
+
+    def kind(self) -> str:
+        return "shell" if self.shell is not None else "upload"
+
+
+@dataclasses.dataclass
+class Playbook:
+    vars: dict
+    steps: list[Step]
+    verify: list[Step]
+
+
+@dataclasses.dataclass
+class StepResult:
+    step: Step
+    phase: str  # "steps" or "verify"
+    index: int
+    total: int
+    exit_code: int
+    duration_s: float
+    ignored: bool = False
+
+    @property
+    def passed(self) -> bool:
+        return self.exit_code == 0
+
+    @property
+    def effectively_passed(self) -> bool:
+        return self.passed or self.ignored
+
+
+# ─── Template substitution ─────────────────────────────────────────────────
+
+VAR_RE = re.compile(r"\{\{\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*\}\}")
+
+
+def substitute(text: str, vars: dict, where: str) -> str:
+    def sub(m: re.Match) -> str:
+        key = m.group(1)
+        if key not in vars:
+            raise SystemExit(
+                f"elway: undefined variable {{{{ {key} }}}} in {where} "
+                f"(known: {sorted(vars.keys()) or 'none'})"
+            )
+        return str(vars[key])
+
+    return VAR_RE.sub(sub, text)
+
+
+def substitute_step(step: Step, vars: dict, where: str) -> Step:
+    new = dataclasses.replace(step)
+    new.name = substitute(step.name, vars, f"{where} name")
+    if step.shell is not None:
+        new.shell = substitute(step.shell, vars, f"{where} shell")
+    if step.upload is not None:
+        new.upload = UploadSpec(
+            src=substitute(step.upload.src, vars, f"{where} upload.src"),
+            dest=substitute(step.upload.dest, vars, f"{where} upload.dest"),
+            mode=step.upload.mode,
+        )
+    return new
+
+
+# ─── Playbook loading ──────────────────────────────────────────────────────
+
+
+def _as_str(value, where: str) -> str:
+    # YAML parses `false` / `true` / `null` / numbers as typed scalars. For fields
+    # that are meant to be shell text or paths, always present them as strings so
+    # the templater (and eventually the shell) sees what the author intended.
+    if value is None:
+        raise SystemExit(f"elway: {where} is null; expected a string")
+    return str(value)
+
+
+def _load_step(raw: dict, where: str) -> Step:
+    if not isinstance(raw, dict):
+        raise SystemExit(f"elway: {where} must be a mapping, got {type(raw).__name__}")
+    if "name" not in raw:
+        raise SystemExit(f"elway: {where} missing required 'name'")
+    has_shell = "shell" in raw
+    has_upload = "upload" in raw
+    if has_shell == has_upload:
+        raise SystemExit(f"elway: {where} must have exactly one of 'shell' or 'upload'")
+    upload = None
+    shell = None
+    if has_shell:
+        shell = _as_str(raw["shell"], f"{where} shell")
+    if has_upload:
+        u = raw["upload"]
+        if not isinstance(u, dict) or "src" not in u or "dest" not in u:
+            raise SystemExit(f"elway: {where} upload must have src and dest")
+        mode = u.get("mode")
+        upload = UploadSpec(
+            src=_as_str(u["src"], f"{where} upload.src"),
+            dest=_as_str(u["dest"], f"{where} upload.dest"),
+            mode=str(mode) if mode is not None else None,
+        )
+    return Step(
+        name=_as_str(raw["name"], f"{where} name"),
+        shell=shell,
+        upload=upload,
+        sudo=bool(raw.get("sudo", False)),
+        stop_on_fail=raw.get("stop_on_fail"),
+    )
+
+
+def load_playbook(path: Path) -> Playbook:
+    try:
+        doc = yaml.safe_load(path.read_text())
+    except yaml.YAMLError as e:
+        raise SystemExit(f"elway: failed to parse {path}: {e}")
+    if doc is None:
+        raise SystemExit(f"elway: {path} is empty")
+    if not isinstance(doc, dict):
+        raise SystemExit(f"elway: {path} top-level must be a mapping with 'steps'")
+
+    vars_ = doc.get("vars") or {}
+    if not isinstance(vars_, dict):
+        raise SystemExit(f"elway: 'vars' in {path} must be a mapping")
+
+    steps_raw = doc.get("steps")
+    if not steps_raw:
+        raise SystemExit(f"elway: {path} has no 'steps'")
+    steps = [_load_step(s, f"{path}:steps[{i}]") for i, s in enumerate(steps_raw)]
+
+    verify_raw = doc.get("verify") or []
+    verify = [_load_step(s, f"{path}:verify[{i}]") for i, s in enumerate(verify_raw)]
+
+    return Playbook(vars=vars_, steps=steps, verify=verify)
+
+
+# ─── SSH / SCP wrappers with ControlMaster ─────────────────────────────────
+
+
+@dataclasses.dataclass
+class SSHContext:
+    host: str
+    control_path: str
+    verbose: bool
+    sudo_password: Optional[str]
+
+    def _base_ssh_opts(self) -> list[str]:
+        return [
+            "-o",
+            "ControlMaster=auto",
+            "-o",
+            f"ControlPath={self.control_path}",
+            "-o",
+            "ControlPersist=60s",
+            "-o",
+            "BatchMode=no",
+        ]
+
+    def ssh_cmd(self, remote_cmd: str) -> list[str]:
+        return ["ssh", *self._base_ssh_opts(), self.host, remote_cmd]
+
+    def scp_cmd(self, src: str, dest: str) -> list[str]:
+        return ["scp", *self._base_ssh_opts(), src, f"{self.host}:{dest}"]
+
+    def close_master(self) -> None:
+        # Best-effort: terminate the control socket so we don't leak processes.
+        try:
+            subprocess.run(
+                ["ssh", "-O", "exit", "-o", f"ControlPath={self.control_path}", self.host],
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+                timeout=5,
+            )
+        except Exception:
+            pass
+
+
+def probe_sudo(ctx: SSHContext) -> Optional[str]:
+    """Return a sudo password if one is needed, else None. Prompts locally."""
+    # Is sudo configured NOPASSWD for this user? `sudo -n -v` exits 0 if so.
+    p = subprocess.run(
+        ctx.ssh_cmd("sudo -n -v"),
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    if p.returncode == 0:
+        return None
+    pw = getpass.getpass(f"[elway] sudo password for {ctx.host}: ")
+    # Validate it works; don't silently carry a wrong password through the whole run.
+    validation = subprocess.run(
+        ctx.ssh_cmd("sudo -S -p '' -v"),
+        input=pw + "\n",
+        text=True,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.PIPE,
+    )
+    if validation.returncode != 0:
+        raise SystemExit("elway: sudo password validation failed; aborting before anything runs.")
+    return pw
+
+
+# ─── Step execution ────────────────────────────────────────────────────────
+
+
+def _stream_process(cmd: list[str], stdin_data: Optional[str], prefix: str) -> int:
+    """Run `cmd`, streaming stdout+stderr line-by-line with `prefix`. Return exit code."""
+    proc = subprocess.Popen(
+        cmd,
+        stdin=subprocess.PIPE if stdin_data is not None else subprocess.DEVNULL,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,  # interleave — user confirmed ok
+        text=True,
+        bufsize=1,
+    )
+    if stdin_data is not None:
+        assert proc.stdin is not None
+        try:
+            proc.stdin.write(stdin_data)
+            proc.stdin.close()
+        except BrokenPipeError:
+            pass
+    assert proc.stdout is not None
+    for line in proc.stdout:
+        print(f"{prefix}{line.rstrip()}")
+    return proc.wait()
+
+
+def run_shell_step(ctx: SSHContext, step: Step, prefix: str) -> int:
+    """Run a shell step on the remote host. Returns exit code."""
+    cmd_inner = step.shell or ""
+    if step.sudo:
+        # bash -s eats stdin for script; sudo -S reads password from stdin.
+        # We prepend the password line ourselves and pipe cmd_inner via bash -c after.
+        # Using `sudo -S -p '' bash -c <script>` lets sudo consume its password line,
+        # then runs the script. The password arrives as one line on stdin.
+        remote = "sudo -S -p '' bash -c " + shlex.quote(cmd_inner)
+        stdin_data = (ctx.sudo_password or "") + "\n"
+    else:
+        remote = "bash -c " + shlex.quote(cmd_inner)
+        stdin_data = None
+    cmd = ctx.ssh_cmd(remote)
+    if ctx.verbose:
+        print(DIM(f"{prefix}$ {' '.join(shlex.quote(c) for c in cmd)}"))
+    return _stream_process(cmd, stdin_data, prefix)
+
+
+def run_upload_step(ctx: SSHContext, step: Step, prefix: str) -> int:
+    """Run an upload step. Auto-mkdirs parent dir, preserves/applies mode, sudo-capable."""
+    assert step.upload is not None
+    up = step.upload
+    src = Path(up.src)
+    if not src.exists():
+        print(f"{prefix}{RED('ERROR')} local file not found: {src}")
+        return 2
+
+    parent = os.path.dirname(up.dest) or "/"
+    mode_clause = f" && chmod {up.mode} {shlex.quote(up.dest)}" if up.mode else ""
+
+    if not step.sudo:
+        # Non-sudo path: mkdir parent, scp direct, chmod.
+        mkdir = f"mkdir -p {shlex.quote(parent)}"
+        if ctx.verbose:
+            print(DIM(f"{prefix}$ ssh {ctx.host} {mkdir!r}"))
+        rc = _stream_process(ctx.ssh_cmd(mkdir), None, prefix)
+        if rc != 0:
+            return rc
+        if ctx.verbose:
+            print(DIM(f"{prefix}$ scp {up.src} {ctx.host}:{up.dest}"))
+        rc = _stream_process(ctx.scp_cmd(str(src), up.dest), None, prefix)
+        if rc != 0:
+            return rc
+        if up.mode:
+            if ctx.verbose:
+                print(DIM(f"{prefix}$ ssh {ctx.host} chmod {up.mode} {up.dest}"))
+            rc = _stream_process(
+                ctx.ssh_cmd(f"chmod {up.mode} {shlex.quote(up.dest)}"), None, prefix
+            )
+        return rc
+
+    # Sudo path: scp to /tmp as user, then sudo mv + chmod server-side.
+    staging = f"/tmp/elway.{os.getpid()}.{int(time.time() * 1000)}.{src.name}"
+    if ctx.verbose:
+        print(DIM(f"{prefix}$ scp {up.src} {ctx.host}:{staging}"))
+    rc = _stream_process(ctx.scp_cmd(str(src), staging), None, prefix)
+    if rc != 0:
+        return rc
+    script = (
+        f"mkdir -p {shlex.quote(parent)} "
+        f"&& mv -f {shlex.quote(staging)} {shlex.quote(up.dest)}"
+        f"{mode_clause}"
+    )
+    remote = "sudo -S -p '' bash -c " + shlex.quote(script)
+    if ctx.verbose:
+        print(DIM(f"{prefix}$ ssh {ctx.host} {remote}"))
+    return _stream_process(
+        ctx.ssh_cmd(remote),
+        (ctx.sudo_password or "") + "\n",
+        prefix,
+    )
+
+
+# ─── Driver / reporting ────────────────────────────────────────────────────
+
+
+def run_phase(
+    ctx: SSHContext,
+    phase: str,
+    steps: list[Step],
+    vars: dict,
+    global_stop_on_fail: bool,
+    dry_run: bool,
+) -> list[StepResult]:
+    results: list[StepResult] = []
+    halted = False
+    for i, step in enumerate(steps, start=1):
+        resolved = substitute_step(step, vars, where=f"{phase}[{i}]")
+        header = f"[{ctx.host}] {phase} {i}/{len(steps)}  {BOLD(resolved.name)}"
+        if resolved.sudo:
+            header += f"  {YELLOW('(sudo)')}"
+        print(header)
+
+        if halted:
+            print(f"  {DIM('SKIPPED (earlier failure)')}")
+            results.append(
+                StepResult(resolved, phase, i, len(steps), exit_code=0, duration_s=0.0, ignored=True)
+            )
+            continue
+
+        if dry_run:
+            if resolved.kind() == "shell":
+                print(f"  {DIM('dry-run:')} shell: {resolved.shell}")
+            else:
+                u = resolved.upload
+                mode = f" mode={u.mode}" if u.mode else ""
+                print(f"  {DIM('dry-run:')} upload: {u.src} → {u.dest}{mode}")
+            results.append(
+                StepResult(resolved, phase, i, len(steps), exit_code=0, duration_s=0.0, ignored=True)
+            )
+            continue
+
+        prefix = f"  {DIM(f'[{phase[0]}{i}]')} "
+        t0 = time.monotonic()
+        if resolved.kind() == "shell":
+            rc = run_shell_step(ctx, resolved, prefix)
+        else:
+            rc = run_upload_step(ctx, resolved, prefix)
+        elapsed = time.monotonic() - t0
+
+        if rc == 0:
+            status = GREEN("✓ PASS")
+        else:
+            status = RED(f"✗ FAIL (rc={rc})")
+
+        print(f"  {status}  {DIM(f'{elapsed:.2f}s')}")
+        results.append(StepResult(resolved, phase, i, len(steps), exit_code=rc, duration_s=elapsed))
+
+        if rc != 0:
+            step_stop = resolved.stop_on_fail
+            effective_stop = global_stop_on_fail if step_stop is None else step_stop
+            if effective_stop and phase == "steps":
+                halted = True
+                # Verify still runs in the caller, even after halt.
+    return results
+
+
+def print_summary(step_results: list[StepResult], verify_results: list[StepResult]) -> int:
+    def summarize(phase: str, results: list[StepResult]) -> tuple[int, int, int]:
+        passed = sum(1 for r in results if r.effectively_passed and not r.ignored)
+        failed = sum(1 for r in results if not r.effectively_passed)
+        skipped = sum(1 for r in results if r.ignored)
+        return passed, failed, skipped
+
+    print()
+    print(BOLD("── summary ────────────────────────────────────────────"))
+    for phase, res in [("steps", step_results), ("verify", verify_results)]:
+        if not res:
+            continue
+        passed, failed, skipped = summarize(phase, res)
+        color = GREEN if failed == 0 else RED
+        print(
+            f"  {phase:<8} "
+            f"{color(f'{passed} passed')}, "
+            f"{RED(f'{failed} failed') if failed else DIM('0 failed')}, "
+            f"{DIM(f'{skipped} skipped')}"
+        )
+    any_fail = any(not r.effectively_passed for r in step_results + verify_results)
+    exit_code = 1 if any_fail else 0
+    overall = RED("FAILED") if any_fail else GREEN("OK")
+    print(f"  overall: {overall}")
+    return exit_code
+
+
+# ─── CLI ───────────────────────────────────────────────────────────────────
+
+
+def parse_var_args(raw_list: list[str]) -> dict:
+    out = {}
+    for raw in raw_list:
+        if "=" not in raw:
+            raise SystemExit(f"elway: --var must be KEY=VALUE, got '{raw}'")
+        k, v = raw.split("=", 1)
+        out[k.strip()] = v
+    return out
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    p = argparse.ArgumentParser(
+        prog="elway",
+        description="Run a shell command, file upload, or YAML playbook over SSH.",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog=__doc__,
+    )
+    p.add_argument("host", help="ssh alias or user@host")
+    mode = p.add_mutually_exclusive_group(required=True)
+    mode.add_argument("--shell", "-s", help="ad-hoc shell command to run on the remote host")
+    mode.add_argument(
+        "--upload",
+        "-u",
+        help="ad-hoc upload as LOCAL:REMOTE (optionally :MODE — e.g. compose.yaml:/opt/x/compose.yaml:0644)",
+    )
+    mode.add_argument("--playbook", "-p", help="path to a YAML playbook")
+    p.add_argument("--sudo", action="store_true", help="run ad-hoc --shell/--upload with sudo")
+    p.add_argument(
+        "--var",
+        action="append",
+        default=[],
+        help="KEY=VALUE (repeatable). Overrides playbook `vars:` defaults.",
+    )
+    p.add_argument(
+        "--continue-on-error",
+        action="store_true",
+        help="Do not halt on first failure (default: halt).",
+    )
+    p.add_argument("--dry-run", action="store_true", help="Print the plan without executing.")
+    p.add_argument("-v", "--verbose", action="store_true", help="Show ssh/scp invocations.")
+    return p.parse_args(argv)
+
+
+def build_adhoc_playbook(args: argparse.Namespace) -> Playbook:
+    if args.shell:
+        step = Step(name="ad-hoc shell", shell=args.shell, sudo=args.sudo)
+    else:
+        assert args.upload
+        parts = args.upload.split(":")
+        if len(parts) == 2:
+            src, dest = parts
+            mode = None
+        elif len(parts) == 3:
+            src, dest, mode = parts
+        else:
+            raise SystemExit(
+                "elway: --upload expected LOCAL:REMOTE or LOCAL:REMOTE:MODE"
+            )
+        step = Step(
+            name="ad-hoc upload",
+            upload=UploadSpec(src=src, dest=dest, mode=mode),
+            sudo=args.sudo,
+        )
+    return Playbook(vars={}, steps=[step], verify=[])
+
+
+def main(argv: Optional[list[str]] = None) -> int:
+    args = parse_args(argv if argv is not None else sys.argv[1:])
+
+    if args.playbook:
+        pb = load_playbook(Path(args.playbook))
+    else:
+        pb = build_adhoc_playbook(args)
+
+    # Merge vars: playbook defaults, overridden by CLI.
+    merged_vars = dict(pb.vars)
+    merged_vars.update(parse_var_args(args.var))
+
+    needs_sudo = any(s.sudo for s in pb.steps + pb.verify)
+
+    # ControlMaster socket in a temp dir scoped to this run.
+    with tempfile.TemporaryDirectory(prefix="elway-") as tmp:
+        control_path = os.path.join(tmp, "cm-%r@%h:%p")
+        ctx = SSHContext(
+            host=args.host,
+            control_path=control_path,
+            verbose=args.verbose,
+            sudo_password=None,
+        )
+        if needs_sudo and not args.dry_run:
+            ctx.sudo_password = probe_sudo(ctx)
+        try:
+            print(BOLD(f"── {args.host} ─ playbook: {args.playbook or 'ad-hoc'}  "
+                      f"({len(pb.steps)} steps"
+                      + (f", {len(pb.verify)} verify" if pb.verify else "")
+                      + ")"))
+            step_results = run_phase(
+                ctx, "steps", pb.steps, merged_vars,
+                global_stop_on_fail=not args.continue_on_error,
+                dry_run=args.dry_run,
+            )
+            verify_results = []
+            if pb.verify:
+                # Verify ALWAYS runs, even if main steps failed — the point
+                # of a verify phase is to characterize end state, not to be
+                # skipped when things are most interesting.
+                verify_results = run_phase(
+                    ctx, "verify", pb.verify, merged_vars,
+                    global_stop_on_fail=not args.continue_on_error,
+                    dry_run=args.dry_run,
+                )
+            return print_summary(step_results, verify_results)
+        finally:
+            ctx.close_master()
+
+
+if __name__ == "__main__":
+    sys.exit(main())