# steward/ansible/executor.py
from __future__ import annotations
import asyncio
import json
import logging
import os
import re
import shutil
import signal
import tempfile
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import TYPE_CHECKING

from steward.core.crypto import is_encrypted as _crypto_is_encrypted

if TYPE_CHECKING:
    from quart import Quart

logger = logging.getLogger(__name__)

_OUTPUT_CAP_BYTES = 1024 * 1024  # 1 MB DB cap (full output also goes to a log artifact)
_FLUSH_LINES = 50
_FLUSH_SECS = 5.0
_MEM_LINES_CAP = 5000   # bounded per-run in-memory replay buffer (SSE late-joiners)
_MEM_TTL_SECS = 300     # drop in-memory state this long after a run completes
_FAILURE_CAP = 50       # max failed-task lines captured into structured results
_READ_CHUNK = 65536     # output is read in chunks, never a whole line at once (#5156)
_MAX_LINE_BYTES = 65536  # a longer line is cut here and marked, not allowed to stall

# Full run logs persist here (survive restart + the 1 MB DB cap). Override via env.
ARTIFACT_DIR = os.environ.get("STEWARD_ANSIBLE_LOG_DIR", "/data/ansible/runs")


def artifact_path(run_id: str) -> str:
    return os.path.join(ARTIFACT_DIR, f"{run_id}.log")


# ── One run per host (#5158) ──────────────────────────────────────────────────
# Two runs against the same host would race each other's changes, so a run
# claims every host it touches before it takes a run slot, and waits (queued)
# while another run holds one. The claim is all-or-nothing, so two runs over
# overlapping hosts can't each hold half and wait on the other.

HOST_WAIT_SECONDS = 3600.0  # a run waiting longer than this for a host fails


class HostBusy(Exception):
    """A run gave up waiting for a host another run held."""


def run_hosts(inventory_content: str | None, inventory_path: str | None) -> set[str]:
    """What a run touches, as claim keys: the address of each host in the YAML
    inventory Steward wrote for it, or, for a source's own inventory file
    (whose hosts Steward doesn't read), that file."""
    if not inventory_content:
        return {f"inventory:{inventory_path}"} if inventory_path else set()
    import yaml
    try:
        data = yaml.safe_load(inventory_content)
    except yaml.YAMLError:
        return set()  # ansible-playbook will refuse it and say why
    # A host listed again under a group carries no vars there, so its address
    # is whichever listing gives one.
    addresses: dict[str, str | None] = {}

    def walk(group) -> None:
        if not isinstance(group, dict):
            return
        members = group.get("hosts")
        if isinstance(members, dict):
            for name, hostvars in members.items():
                addr = hostvars.get("ansible_host") if isinstance(hostvars, dict) else None
                addresses[str(name)] = addr or addresses.get(str(name))
        children = group.get("children")
        if isinstance(children, dict):
            for child in children.values():
                walk(child)

    if isinstance(data, dict):
        for group in data.values():
            walk(group)
    return {str(addr or name).strip().lower() for name, addr in addresses.items()}


def _host_gate(app: "Quart") -> dict:
    """The claims and the wake-up event, bound to the running loop like the
    run-slot semaphore (one loop in prod, one per test)."""
    loop = asyncio.get_running_loop()
    gate = getattr(app, "_ansible_host_gate", None)
    if gate is None or gate["loop"] is not loop:
        gate = {"loop": loop, "held": {}, "changed": asyncio.Event()}
        app._ansible_host_gate = gate
    return gate


async def _claim_hosts(app: "Quart", run_id: str, hosts: set[str]) -> str | None:
    """Hold every host in hosts for run_id, waiting while another run holds
    any of them. Returns the line saying what it waited for, or None when it
    didn't wait. Raises _Aborted when the run is cancelled while waiting,
    HostBusy after HOST_WAIT_SECONDS."""
    from steward.models.ansible import AnsibleRunStatus

    gate = _host_gate(app)
    held: dict[str, str] = gate["held"]
    give_up = time.monotonic() + HOST_WAIT_SECONDS
    waited: str | None = None
    while True:
        busy = sorted({held[h] for h in hosts if h in held and held[h] != run_id})
        if not busy:
            for h in hosts:
                held[h] = run_id
            return waited
        if run_id in _cancelled:
            raise _Aborted
        remaining = give_up - time.monotonic()
        if remaining <= 0:
            names = ", ".join(sorted(h for h in hosts if h in held))
            raise HostBusy(f"gave up after {HOST_WAIT_SECONDS:.0f} s waiting for "
                           f"{names}, held by run {', '.join(busy)}")
        if waited is None:
            waited = (f"[queued — waiting for run {', '.join(busy)} "
                      "to finish on the same host]")
            await _set_status(app, run_id, AnsibleRunStatus.queued)
            _broadcast(run_id, waited)
        changed = gate["changed"]
        try:
            # Woken by a release; the short timeout also notices a cancel.
            await asyncio.wait_for(changed.wait(), timeout=min(remaining, 2.0))
        except asyncio.TimeoutError:
            pass


def _release_hosts(app: "Quart", run_id: str, hosts: set[str]) -> None:
    gate = getattr(app, "_ansible_host_gate", None)
    if gate is None:
        return
    for h in hosts:
        if gate["held"].get(h) == run_id:
            del gate["held"][h]
    changed, gate["changed"] = gate["changed"], asyncio.Event()
    changed.set()


@dataclass
class _Done:
    status: str


class _Aborted(Exception):
    """Raised internally when a queued run is cancelled before it launches."""


# Per-run broadcast state (lives until shortly after completion, then GC'd)
_run_lines: dict[str, list[str]] = {}          # bounded tail of output lines
_run_done: dict[str, _Done] = {}               # set when run completes
_run_listeners: dict[str, list[asyncio.Queue]] = {}  # active SSE client queues

# Running subprocesses by run_id (for cancellation) + run_ids an operator cancelled.
_run_procs: dict[str, asyncio.subprocess.Process] = {}
_cancelled: set[str] = set()


def register_listener(run_id: str) -> asyncio.Queue:
    """Create a listener queue for a run. Replays existing lines immediately."""
    q: asyncio.Queue = asyncio.Queue()
    for line in _run_lines.get(run_id, []):
        q.put_nowait(line)
    done = _run_done.get(run_id)
    if done:
        q.put_nowait(done)
    else:
        _run_listeners.setdefault(run_id, []).append(q)
    return q


def deregister_listener(run_id: str, q: asyncio.Queue) -> None:
    listeners = _run_listeners.get(run_id, [])
    if q in listeners:
        listeners.remove(q)


def _schedule_cleanup(run_id: str) -> None:
    """Drop a finished run's in-memory state after a grace window."""
    async def _cleanup() -> None:
        await asyncio.sleep(_MEM_TTL_SECS)
        _run_lines.pop(run_id, None)
        _run_done.pop(run_id, None)
    try:
        asyncio.create_task(_cleanup())
    except RuntimeError:
        pass  # no running loop (e.g. tests) — fine to skip


def _broadcast(run_id: str, item: str | _Done) -> None:
    if isinstance(item, str):
        buf = _run_lines.setdefault(run_id, [])
        buf.append(item)
        if len(buf) > _MEM_LINES_CAP:  # keep only the most recent lines in memory
            del buf[: len(buf) - _MEM_LINES_CAP]
    else:
        _run_done[run_id] = item
    for q in _run_listeners.get(run_id, []):
        q.put_nowait(item)
    if isinstance(item, _Done):
        _run_listeners.pop(run_id, None)
        _schedule_cleanup(run_id)


def build_ansible_command(
    playbook_path: str,
    inventory_path: str,
    params: dict | None = None,
) -> list[str]:
    """Build the ansible-playbook argv from a run's optional params.

    Pure (no IO). Everything is passed as argv — never shell-interpolated — so
    extra-var values and limits can contain arbitrary characters safely.
    params keys: extra_vars (list of "key=value"), limit, tags, check (bool).
    """
    cmd = ["ansible-playbook", playbook_path, "-i", inventory_path]
    params = params or {}
    for ev in params.get("extra_vars") or []:
        cmd += ["-e", ev]
    if params.get("limit"):
        cmd += ["--limit", params["limit"]]
    if params.get("tags"):
        cmd += ["--tags", params["tags"]]
    if params.get("check"):
        cmd += ["--check", "--diff"]
    return cmd


def build_credentials(creds: dict | None, tmpdir: str) -> tuple[list[str], list[tuple[str, str]]]:
    """Given global Ansible creds + a temp dir, return (extra argv, files to write).

    Pure: decides flags + file contents; the caller writes the files (mode 0600).
    Secrets are always passed via files — never on argv / the process list.
    creds keys: ssh_private_key, vault_password, become_password.
    """
    args: list[str] = []
    files: list[tuple[str, str]] = []
    creds = creds or {}

    from steward.core.crypto import is_encrypted
    key = creds.get("ssh_private_key")
    # A still-encrypted value means decryption failed (app secret key changed) —
    # writing the ciphertext as a key file just yields a cryptic libcrypto error,
    # so skip it; start_run surfaces a clear message instead.
    if key and not is_encrypted(key):
        path = os.path.join(tmpdir, "id_key")
        files.append((path, key if key.endswith("\n") else key + "\n"))
        args += ["--private-key", path]

    vault = creds.get("vault_password")
    if vault:
        path = os.path.join(tmpdir, "vault_pass")
        files.append((path, vault + "\n"))
        args += ["--vault-password-file", path]

    become = creds.get("become_password")
    if become:
        path = os.path.join(tmpdir, "become.yml")
        # A JSON string is valid YAML; keeps the password out of argv.
        files.append((path, "ansible_become_password: " + json.dumps(become) + "\n"))
        args += ["-e", "@" + path]

    return args, files


def build_bootstrap(connection: dict | None, tmpdir: str) -> tuple[list[str], list[tuple[str, str]]]:
    """Per-run connection override for first-contact provisioning.

    Pure: returns (extra argv, files to write). The bootstrap user/password are
    written to a temp vars file (``-e @file``) — never argv, never the DB — so
    they don't leak via the process list or persist on the AnsibleRun row.
    A bare connection password doubles as the become password unless one is
    given explicitly. connection keys: user, password, become_password.
    """
    args: list[str] = []
    files: list[tuple[str, str]] = []
    connection = connection or {}
    user = (connection.get("user") or "").strip()
    password = connection.get("password") or ""
    if not user and not password:
        return args, files

    lines: dict[str, str] = {}
    if user:
        lines["ansible_user"] = user
    if password:
        lines["ansible_password"] = password
        lines["ansible_become_password"] = connection.get("become_password") or password
    # JSON values are valid YAML and keep arbitrary characters safe.
    content = "".join(f"{k}: {json.dumps(v)}\n" for k, v in lines.items())
    path = os.path.join(tmpdir, "bootstrap.yml")
    files.append((path, content))
    args += ["-e", "@" + path]
    return args, files


def build_extra_vars_file(merged: dict | None, tmpdir: str) -> tuple[list[str], list[tuple[str, str]]]:
    """Run-time variables → a JSON extra-vars file (``-e @file``).

    Pure: returns (argv, files). JSON is valid YAML to Ansible and, unlike
    ``-e key=value`` strings (which Ansible shlex-splits on whitespace), keeps
    values with spaces/quotes intact. Used for both operator-entered playbook
    variables and the unpersisted secret subset (merged by the caller).
    """
    if not merged:
        return [], []
    path = os.path.join(tmpdir, "extravars.json")
    return ["-e", "@" + path], [(path, json.dumps(merged))]


def ansible_env(creds: dict | None, base_env) -> dict:
    """Return a copy of base_env with ANSIBLE_HOST_KEY_CHECKING set from creds,
    and the interpreter found quietly (#5334): Ansible still discovers each
    host's Python, but no longer warns about it on every run. A target or group
    that sets ansible_python_interpreter still wins."""
    env = dict(base_env)
    env["ANSIBLE_HOST_KEY_CHECKING"] = "True" if (creds or {}).get("host_key_checking") else "False"
    env.setdefault("ANSIBLE_PYTHON_INTERPRETER", "auto_silent")
    return env


# ── Structured results (parsed from the streamed output) ──────────────────────

_RECAP_RE = re.compile(
    r"^(?P<host>\S+)\s*:\s*ok=(?P<ok>\d+)\s+changed=(?P<changed>\d+)\s+"
    r"unreachable=(?P<unreachable>\d+)\s+failed=(?P<failed>\d+)"
    r"(?:\s+skipped=(?P<skipped>\d+))?"
)


def parse_recap(lines: list[str]) -> dict[str, dict[str, int]]:
    """Parse Ansible's PLAY RECAP into {host: {ok,changed,unreachable,failed,skipped}}.

    Streaming-friendly: we keep the default stdout callback (live output) and
    derive the per-host summary from the recap rather than swapping to the json
    callback (which would buffer everything to the end and kill live streaming).
    """
    hosts: dict[str, dict[str, int]] = {}
    in_recap = False
    for line in lines:
        if "PLAY RECAP" in line:
            in_recap = True
            continue
        if not in_recap:
            continue
        m = _RECAP_RE.match(line.strip())
        if m:
            hosts[m.group("host")] = {
                "ok": int(m.group("ok")),
                "changed": int(m.group("changed")),
                "unreachable": int(m.group("unreachable")),
                "failed": int(m.group("failed")),
                "skipped": int(m.group("skipped") or 0),
            }
    return hosts


# ── Cancellation ──────────────────────────────────────────────────────────────

async def _set_status(app: "Quart", run_id: str, status) -> None:
    from sqlalchemy import update
    from steward.models.ansible import AnsibleRun
    async with app.db_sessionmaker() as session:
        async with session.begin():
            await session.execute(
                update(AnsibleRun).where(AnsibleRun.id == run_id).values(status=status)
            )


def _signal_group(proc: asyncio.subprocess.Process, sig: int) -> None:
    """Send sig to the run's whole process group (ansible-playbook and the
    ssh and worker processes it forks), or to the process alone when it has
    no group of its own."""
    try:
        os.killpg(proc.pid, sig)
    except ProcessLookupError:
        pass
    except OSError:
        try:
            proc.send_signal(sig)
        except ProcessLookupError:
            pass


async def _terminate_then_kill(proc: asyncio.subprocess.Process, grace: float = 10.0) -> None:
    """SIGTERM the run's process group, then SIGKILL it if it outlives grace.
    The group is killed even once the leader has exited, so no worker it
    forked is left behind (#5156)."""
    if proc.returncode is None:
        _signal_group(proc, signal.SIGTERM)
        try:
            await asyncio.wait_for(proc.wait(), timeout=grace)
        except asyncio.TimeoutError:
            pass
    _signal_group(proc, signal.SIGKILL)


async def _lines(stream: asyncio.StreamReader):
    """The stream's lines, read in chunks so no line length can stall or break
    the reader: a line past _MAX_LINE_BYTES is cut there and marked, and the
    rest of it is dropped up to its newline (#5156)."""
    pending = b""
    skipping = False
    while True:
        chunk = await stream.read(_READ_CHUNK)
        if not chunk:
            break
        pending += chunk
        while True:
            nl = pending.find(b"\n")
            if nl < 0:
                break
            line, pending = pending[:nl], pending[nl + 1:]
            if skipping:
                skipping = False
                continue
            yield line
        if len(pending) > _MAX_LINE_BYTES and not skipping:
            yield pending[:_MAX_LINE_BYTES] + b" [line cut at %d bytes]" % _MAX_LINE_BYTES
            pending, skipping = b"", True
        elif skipping:
            pending = b""
    if pending and not skipping:
        yield pending


def cancel_run(run_id: str) -> bool:
    """Request cancellation of a run. Terminates the process if it's running;
    a queued run is flagged so start_run aborts before launch. Returns False
    only when the run isn't tracked in this process (already finished)."""
    if run_id not in _run_procs and run_id not in _cancelled:
        # Not obviously in-flight. Still flag it (covers a just-queued run whose
        # proc hasn't been registered yet); caller should gate on run status.
        _cancelled.add(run_id)
        return run_id in _run_procs or True
    _cancelled.add(run_id)
    proc = _run_procs.get(run_id)
    if proc is not None:
        asyncio.create_task(_terminate_then_kill(proc))
    return True


# ── Run execution ─────────────────────────────────────────────────────────────

# ── Secret scrubbing ──────────────────────────────────────────────────────────
# Ansible echoes values back: a failed task prints its args, -v prints vars,
# --diff prints templated files. Each line is masked before it is broadcast,
# written to the log artifact or stored on the run, so a secret the run was
# handed never leaves it.
MASK = "***"
_MIN_SECRET_LEN = 4  # masking "on" or "1" would mangle the output and hide nothing


def _strings(value) -> list[str]:
    if isinstance(value, str):
        return [value]
    if isinstance(value, dict):
        return [s for v in value.values() for s in _strings(v)]
    if isinstance(value, (list, tuple, set)):
        return [s for v in value for s in _strings(v)]
    return []


def run_secrets(creds: dict | None, connection: dict | None,
                secret_vars: dict | None, redact: list[str] | None) -> list[str]:
    """Every secret value this run is handed: the global become/vault
    passwords, the bootstrap passwords, the sensitive run variables and any
    the caller names (a per-host agent token)."""
    creds, connection = creds or {}, connection or {}
    values = [creds.get("become_password"), creds.get("vault_password"),
              connection.get("password"), connection.get("become_password"),
              *_strings(secret_vars), *(redact or [])]
    found = {v for v in values if isinstance(v, str) and len(v) >= _MIN_SECRET_LEN}
    return sorted(found, key=len, reverse=True)  # longest first: no partial masks


def scrub(line: str, secrets: list[str]) -> str:
    for value in secrets:
        if value in line:
            line = line.replace(value, MASK)
    return line


async def start_run(
    app: "Quart",
    run_id: str,
    playbook_path: str,
    inventory_path: str,
    source_path: str,
    params: dict | None = None,
    inventory_content: str | None = None,
    connection: dict | None = None,
    secret_vars: dict | None = None,
    redact: list[str] | None = None,
) -> None:
    """Execute ansible-playbook as a subprocess and update the DB run row.

    Waits (as 'queued') while another run holds one of its hosts (#5158), then
    for a slot under the global concurrency cap (app._ansible_semaphore).
    Supports cancellation, a
    persistent full-log artifact, and a parsed per-host result summary.

    connection is an optional per-run SSH override (user/password) for
    first-contact provisioning. It is applied to the subprocess only — it is
    never persisted on the AnsibleRun row, never placed on argv, and not logged.
    secret_vars are run-time playbook variables flagged sensitive; like
    connection they reach the subprocess (merged into the extra-vars file) but
    are never persisted. Every value in run_secrets() — redact names more — is
    masked in each output line before it goes anywhere.
    """
    from steward.models.ansible import AnsibleRunStatus

    secrets = run_secrets(app.config.get("ANSIBLE", {}), connection, secret_vars, redact)
    _run_lines[run_id] = []
    db_output = ""
    truncated = False
    lines_since_flush = 0
    last_flush_at = time.monotonic()
    failures: list[str] = []

    async def _flush(final_status: str | None = None, results: dict | None = None) -> None:
        nonlocal lines_since_flush, last_flush_at
        from sqlalchemy import update
        from steward.models.ansible import AnsibleRun

        values: dict = {"output": db_output}
        if results is not None:
            values["results"] = results
        if final_status is not None:
            values["status"] = AnsibleRunStatus(final_status)
            values["finished_at"] = datetime.now(timezone.utc)
        async with app.db_sessionmaker() as session:
            async with session.begin():
                await session.execute(
                    update(AnsibleRun).where(AnsibleRun.id == run_id).values(**values)
                )
        lines_since_flush = 0
        last_flush_at = time.monotonic()

    # ── Concurrency gate (semaphore lazily bound to the running loop so the
    # cap is honoured in prod's single loop without breaking multi-loop tests).
    loop = asyncio.get_running_loop()
    sem = getattr(app, "_ansible_semaphore", None)
    if sem is None or getattr(app, "_ansible_sem_loop", None) is not loop:
        cap = int((app.config.get("ANSIBLE") or {}).get("max_concurrent_runs", 3) or 3)
        sem = asyncio.Semaphore(max(1, cap))
        app._ansible_semaphore = sem
        app._ansible_sem_loop = loop
    hosts = run_hosts(inventory_content, inventory_path)

    final_status = "failed"
    logf = None
    proc = None
    interrupted = False
    slot_held = False
    try:
        # The hosts first, so a run waiting on a host doesn't sit on a run slot.
        # What it waited for goes in the stored output too, so a run that sat
        # queued says why after the live view is gone.
        waits: list[str] = []
        host_wait = await _claim_hosts(app, run_id, hosts)
        if host_wait:
            waits.append(host_wait)
        if sem.locked():
            if not waits:
                await _set_status(app, run_id, AnsibleRunStatus.queued)
            waits.append("[queued — waiting for an Ansible run slot]")
            _broadcast(run_id, waits[-1])
        queued = bool(waits)
        db_output = "\n".join(waits)
        await sem.acquire()
        slot_held = True
        if run_id in _cancelled:  # cancelled while queued — abort before launching
            raise _Aborted
        if queued:
            await _set_status(app, run_id, AnsibleRunStatus.running)

        cwd = source_path if Path(source_path).exists() else None
        creds = app.config.get("ANSIBLE", {})
        try:
            os.makedirs(ARTIFACT_DIR, exist_ok=True)
            logf = open(artifact_path(run_id), "w", encoding="utf-8")
        except OSError:
            logf = None  # artifact is best-effort; the DB still holds capped output

        # Temp dir holds the ephemeral inventory (if any) + credential files.
        tmpdir = tempfile.mkdtemp(prefix="steward-ansible-")
        os.chmod(tmpdir, 0o700)
        try:
            if inventory_content:
                # .yml so Ansible's yaml inventory plugin reliably claims it
                # (the plugin's verify_file checks the extension).
                inv_target = os.path.join(tmpdir, "inventory.yml")
                with open(inv_target, "w", encoding="utf-8") as invf:
                    invf.write(inventory_content)
            else:
                inv_target = inventory_path
            cmd = build_ansible_command(playbook_path, inv_target, params)

            # First-contact provisioning supplies a password and a bootstrap
            # user; the managed key isn't on the host yet, so don't offer it
            # (avoids a failed key attempt before the password fallback).
            conn = connection or {}
            effective_creds = dict(creds)
            if conn.get("password"):
                effective_creds.pop("ssh_private_key", None)
            elif _crypto_is_encrypted(effective_creds.get("ssh_private_key") or ""):
                _broadcast(run_id,
                           "[run error] The managed SSH key could not be decrypted "
                           "(the app secret key changed). Regenerate it in "
                           "Settings → Ansible and re-provision the host(s).")

            cred_args, cred_files = build_credentials(effective_creds, tmpdir)
            boot_args, boot_files = build_bootstrap(conn, tmpdir)
            # Operator-entered run-time vars (persisted) + secret vars (not).
            merged_vars = {**((params or {}).get("extra_vars_map") or {}), **(secret_vars or {})}
            ev_args, ev_files = build_extra_vars_file(merged_vars, tmpdir)
            for cred_path, content in cred_files + boot_files + ev_files:
                with open(cred_path, "w", encoding="utf-8") as cf:
                    cf.write(content)
                os.chmod(cred_path, 0o600)

            # Steady-state floor: connect as the managed account unless a target
            # var or the bootstrap override (extra-vars, higher precedence) wins.
            user_args: list[str] = []
            if not conn.get("user"):
                ssh_user = (creds.get("ssh_user") or "").strip()
                if ssh_user:
                    user_args += ["--user", ssh_user]

            # Its own session, so the whole group (ssh, forks) can be signalled.
            proc = await asyncio.create_subprocess_exec(
                *cmd, *cred_args, *boot_args, *user_args, *ev_args,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.STDOUT,
                cwd=cwd,
                env=ansible_env(creds, os.environ),
                start_new_session=True,
            )
            _run_procs[run_id] = proc
            assert proc.stdout is not None

            async for raw_line in _lines(proc.stdout):
                line = scrub(raw_line.decode(errors="replace").rstrip("\n\r"), secrets)
                _broadcast(run_id, line)
                if logf is not None:
                    logf.write(line + "\n")
                if (line.startswith("fatal:") or "FAILED!" in line) and len(failures) < _FAILURE_CAP:
                    failures.append(line[:500])

                if not truncated:
                    candidate = (db_output + "\n" + line) if db_output else line
                    if len(candidate.encode()) > _OUTPUT_CAP_BYTES:
                        db_output += "\n[output truncated — download the full log]"
                        truncated = True
                    else:
                        db_output = candidate

                lines_since_flush += 1
                elapsed = time.monotonic() - last_flush_at
                if lines_since_flush >= _FLUSH_LINES or elapsed >= _FLUSH_SECS:
                    await _flush()

            await proc.wait()
            recap = parse_recap(_run_lines.get(run_id, []))
            # A cancel that lands after the run finished doesn't undo its result.
            if run_id in _cancelled and proc.returncode != 0:
                final_status = "cancelled"
            elif proc.returncode != 0:
                final_status = "failed"
            elif not recap:
                # ansible-playbook exits 0 on "no hosts matched" / empty inventory
                # — nothing actually ran, so don't report it as a success.
                final_status = "failed"
                if len(failures) < _FAILURE_CAP:
                    failures.append(
                        "No hosts matched — nothing executed "
                        "(check the host's Ansible target and inventory).")
            else:
                final_status = "success"
        finally:
            # Whatever ended the read (the run finishing, an error, the task
            # being cancelled), nothing the run started outlives it.
            if proc is not None:
                await asyncio.shield(_terminate_then_kill(proc, grace=5.0))
            shutil.rmtree(tmpdir, ignore_errors=True)

    except _Aborted:
        final_status = "cancelled"
    except asyncio.CancelledError:
        # The task itself was cancelled (shutdown): record that the run did
        # not finish rather than leave it marked running (#5156).
        final_status, interrupted = "cancelled", True
        msg = "[run error] Steward stopped the run before it finished."
        _broadcast(run_id, msg)
        db_output = (db_output + "\n" + msg) if db_output else msg
    except Exception as exc:
        logger.exception("Ansible run %s failed with exception", run_id)
        final_status = "cancelled" if run_id in _cancelled else "failed"
        # Surface WHY in the run output instead of an empty "failed" (e.g. ENOSPC
        # when the temp dir can't be created — the run never reaches Ansible).
        msg = scrub(f"[run error] {type(exc).__name__}: {exc}", secrets)
        _broadcast(run_id, msg)
        if len(failures) < _FAILURE_CAP:
            failures.append(msg[:500])
        db_output = (db_output + "\n" + msg) if db_output else msg
    finally:
        if logf is not None:
            logf.close()
        _run_procs.pop(run_id, None)
        _cancelled.discard(run_id)
        if slot_held:
            sem.release()
        _release_hosts(app, run_id, hosts)

    results = {"hosts": parse_recap(_run_lines.get(run_id, [])), "failures": failures}
    try:
        await asyncio.shield(_flush(final_status, results=results))
    except Exception:
        logger.exception("Ansible run %s: could not record its final status %s",
                         run_id, final_status)
    _broadcast(run_id, _Done(status=final_status))
    if interrupted:
        raise asyncio.CancelledError