esh-pfi-infrastructure/scripts/elway

#!/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
      # Idempotency (tier 1): if ANY of these conditions says 'skip',
      # the step is marked `skipped`, not run, not counted as failure.
      when: "<remote shell expr>"     # skip unless expr exits 0
      creates: <remote path>          # skip if this path already exists
      removes: <remote path>          # skip if this path is already absent
      # Idempotency (tier 2): runs on the remote AFTER a successful step.
      # If exit 0, step is `changed`; exit != 0, `ok` (meaning no-op).
      # Without this, successful steps default to `changed` (Ansible's
      # default for shell/command modules).
      changed_when: "<remote shell expr>"
  verify:                     # optional; always runs after steps (even on fail)
    - name: Port responds
      shell: curl -sf http://localhost:8188/ >/dev/null

Per-step state machine:
  ok       — ran successfully, changed_when reported no change
  changed  — ran successfully (default for shell/upload without changed_when)
  skipped  — conditions said skip, OR halted after a prior failure, OR dry-run
  failed   — exit code != 0

Overall outcome / exit code:
  0 if OK (all ok) or CHANGED (some changed, none failed)
  1 if FAILED (any failed step, non-ignored)
  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
    # Tier 1 — conditional skip. Any of these evaluating to "skip" means the
    # step is not run; it counts as `skipped`, not `failed`.
    #   when:     remote shell expr; skip if exit != 0 (run when it returns 0)
    #   creates:  remote path; skip if it already exists
    #   removes:  remote path; skip if it does NOT exist
    when: Optional[str] = None
    creates: Optional[str] = None
    removes: Optional[str] = None
    # Tier 2 — change detection. Evaluated on the remote after a successful
    # run; exit 0 means "something changed", exit != 0 means "no change".
    # `changed_when: "false"` forces ok; `changed_when: "true"` forces changed.
    # Without this field, a successful step is `changed` by default.
    changed_when: Optional[str] = None

    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]


# Step result state machine:
#   ok       — ran successfully, changed_when said "no change"
#   changed  — ran successfully (default when changed_when not set)
#   failed   — exit code != 0 (and stop_on_fail / effective flag engaged)
#   skipped  — skipped due to when/creates/removes OR halted after prior fail
STATES = ("ok", "changed", "failed", "skipped")


@dataclasses.dataclass
class StepResult:
    step: Step
    phase: str  # "steps" or "verify"
    index: int
    total: int
    state: str  # one of STATES
    exit_code: int = 0
    duration_s: float = 0.0
    skip_reason: Optional[str] = None

    @property
    def succeeded(self) -> bool:
        # "succeeded" = didn't fail. Skipped and ok and changed all count.
        return self.state != "failed"


# ─── 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,
        )
    for field in ("when", "creates", "removes", "changed_when"):
        val = getattr(step, field)
        if val is not None:
            setattr(new, field, substitute(val, vars, f"{where} {field}"))
    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,
        )
    def opt_str(field: str) -> Optional[str]:
        v = raw.get(field)
        return None if v is None else _as_str(v, f"{where} {field}")

    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"),
        when=opt_str("when"),
        creates=opt_str("creates"),
        removes=opt_str("removes"),
        changed_when=opt_str("changed_when"),
    )


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


# ─── Skip + change evaluators (tier 1 + tier 2) ────────────────────────────


def _quiet_rc(ctx: SSHContext, remote_cmd: str) -> int:
    """Run a shell expression on the remote under bash -c, discarding output."""
    # Wrapping in bash -c gives `!`, `[[`, pipes, etc. consistent semantics
    # across hosts whose default login shell might be dash (Debian default) or
    # something else weird.
    wrapped = "bash -c " + shlex.quote(remote_cmd)
    return subprocess.run(
        ctx.ssh_cmd(wrapped),
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    ).returncode


def evaluate_skip(ctx: SSHContext, step: Step) -> Optional[str]:
    """If any pre-condition says skip, return a human-readable reason. Else None."""
    if step.when is not None:
        rc = _quiet_rc(ctx, step.when)
        if rc != 0:
            return f"when: expr exited {rc} (needed 0 to run)"
    if step.creates is not None:
        rc = _quiet_rc(ctx, f"test -e {shlex.quote(step.creates)}")
        if rc == 0:
            return f"creates: {step.creates} already exists"
    if step.removes is not None:
        rc = _quiet_rc(ctx, f"test -e {shlex.quote(step.removes)}")
        if rc != 0:
            return f"removes: {step.removes} is already absent"
    return None


def evaluate_changed(ctx: SSHContext, step: Step) -> bool:
    """Post-step: should this count as `changed`? Default True (Ansible shell default)."""
    if step.changed_when is None:
        return True
    rc = _quiet_rc(ctx, step.changed_when)
    # exit 0 = the change-detector expression "fired" = step counts as changed.
    return rc == 0


# ─── Driver / reporting ────────────────────────────────────────────────────


STATE_BADGE = {
    "ok":      lambda: GREEN("○ OK"),
    "changed": lambda: YELLOW("● CHANGED"),
    "failed":  lambda: RED("✗ FAILED"),
    "skipped": lambda: DIM("⏭ SKIPPED"),
}


def _print_status(state: str, exit_code: int, elapsed: float, skip_reason: Optional[str]) -> None:
    badge = STATE_BADGE[state]()
    suffix = ""
    if state == "failed":
        suffix = f" (rc={exit_code})"
    elif state == "skipped" and skip_reason:
        suffix = f" — {skip_reason}"
    tail = f"  {DIM(f'{elapsed:.2f}s')}" if state not in ("skipped",) else ""
    print(f"  {badge}{suffix}{tail}")


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_status("skipped", 0, 0.0, "earlier step failed")
            results.append(StepResult(resolved, phase, i, len(steps),
                                      state="skipped", skip_reason="earlier step failed"))
            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}")
            if resolved.when or resolved.creates or resolved.removes:
                conds = []
                if resolved.when: conds.append(f"when={resolved.when!r}")
                if resolved.creates: conds.append(f"creates={resolved.creates!r}")
                if resolved.removes: conds.append(f"removes={resolved.removes!r}")
                print(f"  {DIM('dry-run:')} conditions: {', '.join(conds)}")
            results.append(StepResult(resolved, phase, i, len(steps), state="skipped",
                                      skip_reason="dry-run"))
            continue

        # Tier 1: conditional skip
        skip_reason = evaluate_skip(ctx, resolved)
        if skip_reason is not None:
            _print_status("skipped", 0, 0.0, skip_reason)
            results.append(StepResult(resolved, phase, i, len(steps),
                                      state="skipped", skip_reason=skip_reason))
            continue

        # Execute
        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:
            state = "failed"
        else:
            # Tier 2: changed_when decides ok vs changed
            state = "changed" if evaluate_changed(ctx, resolved) else "ok"

        _print_status(state, rc, elapsed, None)
        results.append(StepResult(resolved, phase, i, len(steps),
                                  state=state, exit_code=rc, duration_s=elapsed))

        if state == "failed":
            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 _tally(results: list[StepResult]) -> dict[str, int]:
    return {s: sum(1 for r in results if r.state == s) for s in STATES}


def print_summary(step_results: list[StepResult], verify_results: list[StepResult]) -> int:
    print()
    print(BOLD("── summary ────────────────────────────────────────────"))
    for phase, res in [("steps", step_results), ("verify", verify_results)]:
        if not res:
            continue
        t = _tally(res)
        # Color the ok / changed counts only when there's something to report.
        ok_s     = GREEN(f"{t['ok']} ok")
        chg_s    = (YELLOW(f"{t['changed']} changed") if t["changed"] else DIM("0 changed"))
        fail_s   = (RED(f"{t['failed']} failed") if t["failed"] else DIM("0 failed"))
        skip_s   = DIM(f"{t['skipped']} skipped")
        print(f"  {phase:<8} {ok_s}, {chg_s}, {fail_s}, {skip_s}")

    any_fail = any(r.state == "failed" for r in step_results + verify_results)
    any_changed = any(r.state == "changed" for r in step_results + verify_results)
    if any_fail:
        overall, exit_code = RED("FAILED"), 1
    elif any_changed:
        overall, exit_code = YELLOW("CHANGED"), 0
    else:
        overall, exit_code = GREEN("OK"), 0
    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())