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


# ─── Run logging ──────────────────────────────────────────────────────────
#
# Every run tees output to two files in addition to the terminal so the
# operator (or a sibling tool) can re-read what happened without scrolling
# back. Two locations:
#
#   /tmp/elway-last.log                — overwritten per run; the easy
#                                        "what just happened" target
#   ~/.cache/elway/runs/<ts>-<host>-<playbook>.log
#                                      — permanent timestamped record;
#                                        accumulates over time
#
# ANSI color codes are kept in the file (modern less/tail handle them);
# strip with `sed 's/\x1b\[[0-9;]*m//g'` if pasting somewhere that doesn't.

LAST_LOG_PATH = Path("/tmp/elway-last.log")
LOG_ROOT = Path.home() / ".cache" / "elway" / "runs"


class _Tee:
    """File-like that fans writes out to one or more streams."""

    def __init__(self, *streams) -> None:
        self._streams = [s for s in streams if s is not None]

    def write(self, data) -> int:
        for s in self._streams:
            try:
                s.write(data)
            except (BrokenPipeError, ValueError):
                pass
        return len(data)

    def flush(self) -> None:
        for s in self._streams:
            try:
                s.flush()
            except (BrokenPipeError, ValueError):
                pass

    def isatty(self) -> bool:
        # Preserve color decisions: if the *first* stream (the real
        # terminal) is a tty, the Tee reports as one. The on-disk files
        # get the same colored output, which is fine.
        return bool(self._streams) and getattr(self._streams[0], "isatty", lambda: False)()


def _open_run_logs(host: str, playbook: Optional[str], explicit: Optional[Path]) -> tuple[Optional[Path], list]:
    """Open log files for this run. Returns (permanent_log_path, [open_handles])."""
    handles = []
    perm_path = None

    safe_host = re.sub(r"[^a-zA-Z0-9._-]", "_", host)
    pb_label = "ad-hoc" if not playbook else re.sub(r"[^a-zA-Z0-9._-]", "_", Path(playbook).stem)

    if explicit is not None:
        # Caller forced a specific path — only open that one (no /tmp tee).
        try:
            explicit.parent.mkdir(parents=True, exist_ok=True)
            handles.append(open(explicit, "w", encoding="utf-8"))
            perm_path = explicit
        except OSError as exc:
            sys.stderr.write(f"[elway] could not open --log {explicit}: {exc}\n")
        return perm_path, handles

    # Default: tee both /tmp/elway-last.log and a timestamped permanent log.
    try:
        handles.append(open(LAST_LOG_PATH, "w", encoding="utf-8"))
    except OSError as exc:
        sys.stderr.write(f"[elway] could not open {LAST_LOG_PATH}: {exc}\n")

    try:
        LOG_ROOT.mkdir(parents=True, exist_ok=True)
        ts = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
        perm_path = LOG_ROOT / f"{ts}-{safe_host}-{pb_label}.log"
        handles.append(open(perm_path, "w", encoding="utf-8"))
    except OSError as exc:
        sys.stderr.write(f"[elway] could not open permanent log under {LOG_ROOT}: {exc}\n")
        perm_path = None

    return perm_path, handles


# ─── 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] = None
    sudo_probed: bool = False   # so we probe at most once per run, lazily

    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 ensure_sudo(ctx: SSHContext) -> None:
    """Idempotent lazy sudo probe. Called before the first sudo step that will
    actually run — not at startup — so playbooks whose sudo steps all skip
    never prompt for a password.

    Sets ctx.sudo_password to None if NOPASSWD sudo works, or to the validated
    password the user entered. Marks ctx.sudo_probed=True so subsequent sudo
    steps reuse the already-acquired credential.
    """
    if ctx.sudo_probed:
        return
    ctx.sudo_probed = True
    # Is sudo configured NOPASSWD for this user?
    #
    # ⚠ This MUST be `sudo -n true`, never `sudo -n -v`. `-v` refreshes the
    # auth timestamp, and a NOPASSWD-only rule creates no timestamp to refresh,
    # so on **sudo >= 1.9.15** `-n -v` returns non-zero even though every real
    # command runs passwordless. Measured 2026-09-01:
    #
    #   pfi-gx10    sudo 1.9.15p5   sudo -n -v rc=1   sudo -n true rc=0
    #   nh3-docker  sudo 1.9.13p3   sudo -n -v rc=0   sudo -n true rc=0
    #   ana-docker  sudo 1.9.13p3   sudo -n -v rc=0   sudo -n true rc=0
    #   irv-ml1     sudo 1.9.13p3   sudo -n -v rc=0   sudo -n true rc=0
    #
    # The old probe cost a whole session on gx10: elway prompted for a password
    # on a host that had working NOPASSWD sudo the entire time, and in a
    # non-interactive run that prompt is an EOFError mid-playbook. Every host
    # will hit this as it moves past sudo 1.9.13. Assert the effective
    # behaviour (can I run a command?), not a proxy for it.
    p = subprocess.run(
        ctx.ssh_cmd("sudo -n true"),
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    if p.returncode == 0:
        ctx.sudo_password = None
        return
    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.")
    ctx.sudo_password = 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:
        # Lazy: probe (and prompt if needed) right before the first sudo step
        # actually executes. Skipped sudo steps never trigger a prompt.
        ensure_sudo(ctx)
        # `sudo -S -p '' bash -c <script>` lets sudo consume its password
        # line from stdin, then runs the script. Password comes as one line.
        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: probe lazily, then scp to /tmp as user, then sudo mv + chmod server-side.
    ensure_sudo(ctx)
    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


# ─── ops log ───────────────────────────────────────────────────────────────
#
# `infra-ops` and `infra-hermes` are two agents behind one OS identity, so a
# change made over ssh is otherwise fingerprint-less. elway is the sanctioned
# way to CHANGE things on a host, which makes it the right place to record
# them — automatically, because a log you have to remember to write is the
# same class of instrument as a health check that passes in both states.
# See scripts/ops-log and docs/pfi/ops-log.md.

OPS_LOG_BIN = Path(__file__).resolve().parent / "ops-log"


def record_to_ops_log(host: str, playbook: Optional[str], adhoc: Optional[str],
                      step_results: list, verify_results: list) -> None:
    """Best-effort. A failure here must never change elway's exit code or
    output — the audit trail is not allowed to become a new failure mode on
    the path that fixes things."""
    if not OPS_LOG_BIN.is_file():
        return
    tallies = []
    for phase, res in (("steps", step_results), ("verify", verify_results)):
        if not res:
            continue
        t = _tally(res)
        tallies.append(f"{phase} {t['ok']} ok/{t['changed']} changed/"
                       f"{t['failed']} failed/{t['skipped']} skipped")
    all_res = step_results + verify_results
    if any(r.state == "failed" for r in all_res):
        outcome = "failed"
    elif any(r.state == "changed" for r in all_res):
        outcome = "changed"
    else:
        outcome = "ok"
    target = Path(playbook).stem if playbook else "ad-hoc"
    detail = "; ".join(tallies) or "no steps ran"
    if adhoc:
        # Truncated: the ops log is an index of what happened, not a
        # transcript — the full command is in the elway run log.
        detail += f" | {adhoc[:160]}"
    try:
        subprocess.run(
            [str(OPS_LOG_BIN), "record", "--host", host, "--action", "elway",
             "--target", target, "--outcome", outcome, "--detail", detail, "-q"],
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10)
    except Exception:
        pass


# ─── 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.")
    p.add_argument(
        "--log",
        type=Path,
        default=None,
        help="Override log file path. Default: tee to /tmp/elway-last.log + ~/.cache/elway/runs/<ts>-<host>-<playbook>.log",
    )
    p.add_argument(
        "--no-log",
        action="store_true",
        help="Don't write run logs to disk; terminal-only.",
    )
    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))

    # Set up run logging — tee stdout to log file(s) so subsequent reads
    # don't have to scroll the terminal. `--no-log` disables.
    log_handles: list = []
    perm_log: Optional[Path] = None
    real_stdout = sys.stdout
    if not args.no_log:
        perm_log, log_handles = _open_run_logs(args.host, args.playbook, args.log)
        if log_handles:
            sys.stdout = _Tee(real_stdout, *log_handles)

    # 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,
        )
        # No upfront sudo probe — lazily triggered from run_shell_step /
        # run_upload_step the first time a sudo step actually runs (i.e.
        # after its when:/creates: skip conditions pass). Playbooks whose
        # sudo steps all skip never prompt.
        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 "")
                      + ")"))
            if perm_log is not None:
                print(DIM(f"   log: {perm_log}"))
            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,
                )
            rc = print_summary(step_results, verify_results)
            if not args.dry_run:
                record_to_ops_log(args.host, args.playbook,
                                  args.shell or args.upload, step_results,
                                  verify_results)
            if perm_log is not None:
                print(DIM(f"log saved: {perm_log}"))
            return rc
        finally:
            ctx.close_master()
            sys.stdout = real_stdout
            for h in log_handles:
                try:
                    h.close()
                except Exception:
                    pass


if __name__ == "__main__":
    sys.exit(main())