ryra-test 0.8.5

use std::time::{Duration, Instant};

use anyhow::Result;

use crate::executor::Executor;
use crate::registry::{DiscoveredTest, TestEntry};
use crate::scenario::{Event, EventKind, OnEvent, Outcome, ScenarioResult};
use crate::test_toml::StepDef;

/// Shell expansion for the service-data root: the `RYRA_DATA_DIR` override
/// (the host test sandbox, or a user's relocated services dir) or the default
/// `~/.local/share/services`. The runner and registry tests resolve every
/// service-data path through this so it works under the sandbox, in a VM
/// (where the var is unset and it falls back), and for normal installs.
const DATA_ROOT_SH: &str = "${RYRA_DATA_DIR:-$HOME/.local/share/services}";

fn emit(events: &mut Vec<Event>, ev: Event, on_event: &Option<OnEvent>) {
    if let Some(cb) = on_event {
        cb(&ev);
    }
    events.push(ev);
}

fn print_event_result(prefix: &str, event: &Event) {
    let elapsed = format!("{:.1}s", event.duration.as_secs_f64());
    match &event.outcome {
        Outcome::Passed => println!("{prefix}    ok ({elapsed})"),
        Outcome::Failed(msg) => println!("{prefix}    FAIL ({elapsed}) — {msg}"),
        Outcome::Skipped => println!("{prefix}    skip"),
    }
}

/// Execute a simple (non-lifecycle) test suite inside a VM.
///
/// 1. Deploys registry services with `ryra add` (no explicit init —
///    `ryra add` bootstraps config on its first run)
/// 2. Waits for declared ports
/// 3. If quadlets are present, copies them to systemd dir, reloads, starts them
/// 4. Sources `.env` files
/// 5. Runs each test command via SSH, checks exit code
pub async fn run_registry_test(
    vm: &dyn Executor,
    test: &DiscoveredTest,
    prefixed: bool,
    on_event: Option<OnEvent>,
) -> ScenarioResult {
    let start = Instant::now();
    let name = test.name();
    // Line prefix: `[name] ` when output from parallel tests interleaves on one
    // terminal, empty when a `---- START name ----` banner already scopes the
    // block (serial bare runs). See run_lifecycle_test for the same convention.
    let p = if prefixed {
        format!("[{name}] ")
    } else {
        String::new()
    };
    let mut events = Vec::new();
    let mut failed = false;

    let (services, quadlets) = match test {
        DiscoveredTest::Simple { setup, .. } => (&setup.services, &setup.quadlets),
        DiscoveredTest::Lifecycle { .. } => {
            // Lifecycle tests should use run_lifecycle_test instead
            return ScenarioResult {
                name: name.to_string(),
                events: vec![],
                duration: start.elapsed(),
                outcome: Outcome::Failed("run_registry_test called for lifecycle test".to_string()),
            };
        }
    };

    // Collect env overrides from all tests — these may include values for
    // required env vars that `ryra add` needs to succeed non-interactively.
    let mut add_env_prefix = String::new();
    {
        let mut combined: std::collections::BTreeMap<String, String> =
            std::collections::BTreeMap::new();
        for entry in test.tests() {
            for (key, val) in &entry.env {
                combined.entry(key.clone()).or_insert_with(|| val.clone());
            }
        }
        if !combined.is_empty() {
            let exports: Vec<String> = combined.iter().map(|(k, v)| format!("{k}={v}")).collect();
            add_env_prefix = exports.join(" ") + " ";
        }
    }

    // Deploy each registry service
    if !failed {
        for service in services {
            println!("{p}  ryra add {service}...");
            let step_event = run_event(
                vm,
                EventKind::Step,
                &format!("{add_env_prefix}ryra add {service}"),
                300,
            )
            .await;
            print_event_result(&p, &step_event);

            if step_event.outcome.is_fail() {
                failed = true;
                emit(&mut events, step_event, &on_event);
                break;
            }
            emit(&mut events, step_event, &on_event);

            // Wait for service to be active
            println!("{p}  waiting for {service} to start...");
            let wait_event = wait_for_service(vm, &p, service).await;
            print_event_result(&p, &wait_event);
            if wait_event.outcome.is_fail() {
                failed = true;
                emit(&mut events, wait_event, &on_event);
                break;
            }
            emit(&mut events, wait_event, &on_event);
        }
    }

    // Wait for declared ports to be reachable (services may need startup time
    // beyond what systemd "active" indicates).
    if !failed {
        for service in services {
            let port_cmd =
                format!("grep PORT {DATA_ROOT_SH}/{service}/.env 2>/dev/null | cut -d= -f2");
            if let Ok(out) = vm.exec(&port_cmd).await {
                for port in out.stdout.trim().lines() {
                    let port = port.trim();
                    if port.is_empty() {
                        continue;
                    }
                    println!("{p}  waiting for port {port}...");
                    let port_event = wait_for_port(vm, &p, port).await;
                    print_event_result(&p, &port_event);
                    if port_event.outcome.is_fail() {
                        failed = true;
                        emit(&mut events, port_event, &on_event);
                        break;
                    }
                    emit(&mut events, port_event, &on_event);
                }
            }
            if failed {
                break;
            }
        }
    }

    // Deploy quadlet files if present
    if !failed && !quadlets.is_empty() {
        println!("{p}  deploying quadlet files...");
        let deploy_cmd = "\
            mkdir -p $HOME/.config/containers/systemd && \
            cp /opt/ryra-test-project/*.container $HOME/.config/containers/systemd/ 2>/dev/null; \
            cp /opt/ryra-test-project/*.volume $HOME/.config/containers/systemd/ 2>/dev/null; \
            cp /opt/ryra-test-project/*.network $HOME/.config/containers/systemd/ 2>/dev/null; \
            cp /opt/ryra-test-project/*.pod $HOME/.config/containers/systemd/ 2>/dev/null; \
            systemctl --user daemon-reload";
        let deploy_event = run_event(vm, EventKind::Step, deploy_cmd, 30).await;
        print_event_result(&p, &deploy_event);
        if deploy_event.outcome.is_fail() {
            failed = true;
        }
        emit(&mut events, deploy_event, &on_event);

        // Derive service names from .container file stems, start each
        if !failed {
            let quadlet_services: Vec<&str> = quadlets
                .iter()
                .filter(|q| q.ends_with(".container"))
                .filter_map(|q| q.strip_suffix(".container"))
                .collect();

            for svc in &quadlet_services {
                println!("{p}  starting {svc}...");
                let start_cmd = format!("systemctl --user start {svc}.service");
                let start_event = run_event(vm, EventKind::Step, &start_cmd, 120).await;
                print_event_result(&p, &start_event);
                if start_event.outcome.is_fail() {
                    failed = true;
                    emit(&mut events, start_event, &on_event);
                    break;
                }
                emit(&mut events, start_event, &on_event);

                println!("{p}  waiting for {svc}...");
                let wait_event = wait_for_service(vm, &p, svc).await;
                print_event_result(&p, &wait_event);
                if wait_event.outcome.is_fail() {
                    failed = true;
                    emit(&mut events, wait_event, &on_event);
                    break;
                }
                emit(&mut events, wait_event, &on_event);
            }
        }
    }

    // Build the env sourcing prefix for test commands
    let env_prefix = if !failed {
        match build_env_prefix(vm, test).await {
            Ok(prefix) => prefix,
            Err(e) => {
                failed = true;
                emit(
                    &mut events,
                    Event::bare(
                        "source service env vars".to_string(),
                        EventKind::Step,
                        Outcome::Failed(format!("{e:#}")),
                        Duration::ZERO,
                    ),
                    &on_event,
                );
                String::new()
            }
        }
    } else {
        String::new()
    };

    // Run each test command
    for test_entry in test.tests() {
        if failed {
            emit(
                &mut events,
                Event::bare(
                    format!("test: {}", test_entry.name),
                    EventKind::Assertion,
                    Outcome::Skipped,
                    Duration::ZERO,
                ),
                &on_event,
            );
            println!("{p}  skip  {}", test_entry.name);
            continue;
        }

        println!("{p}  test  {}...", test_entry.name);
        let event = run_test_entry(vm, test_entry, &env_prefix).await;
        print_event_result(&p, &event);
        if event.outcome.is_fail() {
            failed = true;
        }
        emit(&mut events, event, &on_event);
    }

    // Dump diagnostics on failure
    if failed {
        dump_diagnostics(vm, &p, &test.services()).await;
    }

    let outcome = if failed {
        let first_failure = events
            .iter()
            .find_map(|e| match &e.outcome {
                Outcome::Failed(msg) => Some(msg.clone()),
                _ => None,
            })
            .unwrap_or_else(|| "unknown failure".to_string());
        Outcome::Failed(first_failure)
    } else {
        Outcome::Passed
    };

    ScenarioResult {
        name: test.name().to_string(),
        events,
        duration: start.elapsed(),
        outcome,
    }
}

/// Run a single test entry — wraps the `run` command with env sourcing and timeout.
async fn run_test_entry(vm: &dyn Executor, entry: &TestEntry, env_prefix: &str) -> Event {
    let t = Instant::now();

    // Build the full command: env overrides + env sourcing + the test command
    let mut parts = Vec::new();

    // Inline env overrides from the test definition
    for (key, val) in &entry.env {
        parts.push(format!("export {key}={val}"));
    }

    // Source the env files
    if !env_prefix.is_empty() {
        parts.push(env_prefix.to_string());
    }

    // The actual test command
    parts.push(entry.run.clone());

    let full_cmd = parts.join(" && ");

    // Run with timeout
    let timeout = Duration::from_secs(entry.timeout_secs);
    let result = tokio::time::timeout(timeout, vm.exec(&full_cmd)).await;

    let outcome = match result {
        Ok(Ok(_)) => Outcome::Passed,
        Ok(Err(e)) => Outcome::Failed(format!("{e:#}")),
        Err(_) => Outcome::Failed(format!("timed out after {}s", entry.timeout_secs)),
    };

    Event::bare(
        format!("test: {}", entry.name),
        EventKind::Assertion,
        outcome,
        t.elapsed(),
    )
}

/// Shell snippet that loads an .env file into the current shell's exported
/// environment **safely** — `read`-per-line + single split on `=`, so values
/// containing whitespace or shell metacharacters (e.g. supabase's
/// `DB_AFTER_CONNECT_QUERY=SET search_path TO …` or `ERL_AFLAGS=-proto_dist
/// inet_tcp`) don't get parsed as inline commands like `. file` would.
///
/// The .env format ryra writes intentionally avoids quoting so podman's
/// --env-file can consume it verbatim (podman does NOT strip quotes), which
/// means bash `source` / `.` is unsafe against any value with whitespace.
fn load_env_shell(path: &str) -> String {
    format!(
        "while IFS='=' read -r __k __v; do \
         case \"$__k\" in \"\"|\\#*) continue ;; esac; \
         export \"$__k=$__v\"; \
         done < {path}"
    )
}

/// Build a shell snippet that sources all relevant .env files.
///
/// Single-service: loads `<service>/.env` into the current shell unprefixed.
/// Multi-service: reads each .env and exports with SERVICE__ prefix.
async fn build_env_prefix(_vm: &dyn Executor, test: &DiscoveredTest) -> Result<String> {
    match test {
        DiscoveredTest::Simple { setup, .. } => {
            if setup.services.len() == 1 {
                Ok(load_env_shell(&format!(
                    "{DATA_ROOT_SH}/{}/.env",
                    setup.services[0]
                )))
            } else if setup.services.len() > 1 {
                // For multi-service, we generate a script that reads each .env
                // and re-exports vars with the service name prefix
                let mut lines = Vec::new();
                for service in &setup.services {
                    let prefix = service.to_uppercase();
                    lines.push(format!(
                        "while IFS='=' read -r key val; do \
                         case \"$key\" in \"\"|\\#*) continue ;; esac; \
                         export {prefix}__$key=\"$val\"; \
                         done < {DATA_ROOT_SH}/{service}/.env"
                    ));
                }
                Ok(lines.join(" && "))
            } else {
                Ok(String::new())
            }
        }
        DiscoveredTest::Lifecycle { .. } => {
            // Lifecycle tests handle env sourcing within their step commands
            Ok(String::new())
        }
    }
}

/// Wait for a service's systemd unit to become active (default 60s timeout).
async fn wait_for_service(vm: &dyn Executor, prefix: &str, service: &str) -> Event {
    wait_for_service_with_timeout(vm, prefix, service, 60).await
}

/// Wait for a service's systemd unit to become active with a custom timeout.
/// `prefix` is the test's line prefix (`"[name] "` in parallel runs, empty when
/// a banner already scopes the output), used to align the wait's heartbeat with
/// surrounding lines.
async fn wait_for_service_with_timeout(
    vm: &dyn Executor,
    prefix: &str,
    service: &str,
    timeout_secs: u64,
) -> Event {
    let t = Instant::now();
    let timeout = Duration::from_secs(timeout_secs);

    let unit = format!("{service}.service");
    let result = vm
        .wait_for_service(&unit, timeout, &format!("{prefix}    "))
        .await;

    let outcome = match result {
        Ok(()) => Outcome::Passed,
        Err(e) => Outcome::Failed(format!("service didn't start: {e:#}")),
    };

    Event::bare(
        format!("wait for {service}"),
        EventKind::Step,
        outcome,
        t.elapsed(),
    )
}

/// Escape a value for embedding inside a single-quoted shell string.
fn shell_escape(s: &str) -> String {
    s.replace('\'', r"'\''")
}

/// Run a Playwright spec and save the HTML report.
async fn run_browser_step(
    vm: &dyn Executor,
    test_name: &str,
    step_name: &str,
    spec: &str,
    env: &std::collections::BTreeMap<String, String>,
    timeout_secs: u64,
    registry_path: &std::path::Path,
) -> Event {
    let t = Instant::now();

    // Build env exports from the step's env map. Values are shell-quoted.
    let mut env_exports = String::new();
    for (key, val) in env {
        let quoted = shell_escape(val);
        env_exports.push_str(&format!("export {key}='{quoted}' && "));
    }

    let browser_dir = format!("{}/tests/browser", registry_path.display());
    let browser_dir_esc = shell_escape(&browser_dir);
    let spec_esc = shell_escape(spec);
    // Where Playwright writes its HTML report, in the execution environment.
    // Local = the host reports dir directly; VM = a staging path fetched out
    // below. Either way the report ends up under services-test/reports/.
    let out_dir = vm.playwright_out_dir(test_name);
    let out_dir_esc = shell_escape(&out_dir);

    // Shell command:
    // 1. Source all service .env files so port vars (PORT_HTTP etc.) are
    //    available to the spec. Static env from the toml step overrides these.
    // 2. Pre-create the canonical report directory and tell playwright to
    //    emit the HTML report directly there (no intermediate copy step).
    // 3. cd into the browser test dir.
    // 4. Ensure node_modules exists — symlink /opt/playwright/node_modules
    //    in the VM image, or `bun install` on a bare host.
    // 5. Export env vars from the step (overrides sourced vars).
    // 6. Run playwright with the html reporter pointed at the canonical path.
    //    Also use the list reporter so the user sees live progress.
    // 7. Exit with playwright's own exit code.
    // Per-file safe .env load — same rationale as load_env_shell(): raw
    // `.` would choke on values with whitespace (supabase `DB_AFTER_*`, etc.)
    let env_loop = format!(
        "for __f in {DATA_ROOT_SH}/*/.env; do \
           [ -f \"$__f\" ] && {loader}; \
         done",
        loader = load_env_shell("\"$__f\"")
    );
    // Expose inbucket to the spec as INBUCKET_URL when it's installed. Browser
    // specs that read a mailed code (verification/OTT) consume this via the
    // inbucket.ts helper — Playwright has no email primitive, so the runner
    // owns service-layout discovery (which .env, which port) in one place
    // rather than every spec hard-coding ~/.local/share/services/inbucket/.env.
    // `;`-separated (not `&&`) so a missing inbucket leaves INBUCKET_URL unset
    // without aborting the command. A step's own `env` (exported below)
    // overrides this.
    let inbucket_export = format!(
        "INBUCKET_PORT=$(grep SERVICE_PORT_HTTP \
         {DATA_ROOT_SH}/inbucket/.env 2>/dev/null | cut -d= -f2); \
         [ -n \"$INBUCKET_PORT\" ] && \
         export INBUCKET_URL=\"http://127.0.0.1:$INBUCKET_PORT\"; "
    );
    let cmd = format!(
        "{env_loop} && \
         {inbucket_export}\
         DEST={out_dir_esc} && \
         mkdir -p \"$DEST\" && \
         cd '{browser_dir_esc}' && \
         if [ ! -d node_modules ]; then \
           if [ -d /opt/playwright/node_modules ]; then \
             ln -sf /opt/playwright/node_modules .; \
           else \
             bun install playwright @playwright/test 2>&1; \
           fi; \
         fi && \
         export PATH=\"$HOME/.bun/bin:$PATH\" && \
         export PLAYWRIGHT_HTML_REPORT=\"$DEST\" && \
         export PLAYWRIGHT_HTML_OPEN=never && \
         {env_exports}\
         bunx playwright test '{spec_esc}' --reporter=list,html"
    );

    let timeout = Duration::from_secs(timeout_secs);
    let result = tokio::time::timeout(timeout, vm.exec_streaming(&cmd, test_name)).await;

    let outcome = match result {
        Ok(Ok(_)) => Outcome::Passed,
        Ok(Err(e)) => Outcome::Failed(format!("{e:#}")),
        Err(_) => Outcome::Failed(format!("timed out after {timeout_secs}s")),
    };

    // Pull the report into the host's canonical reports dir
    // (services-test/reports/<test>/playwright), regardless of VM or bare mode.
    // On a VM this copies from the staging path; on bare it's a no-op because
    // Playwright already wrote straight there. We always try — even on failure,
    // because traces on failure are the most valuable ones to inspect.
    if let Ok(reports) = crate::reports::reports_dir() {
        let local_dir = reports.join(test_name).join("playwright");
        if let Err(e) = vm.fetch_dir(&out_dir, &local_dir).await {
            eprintln!("warning: failed to fetch playwright report: {e:#}");
        }
    }

    Event::bare(
        format!("browser: {step_name}"),
        EventKind::Assertion,
        outcome,
        t.elapsed(),
    )
}

/// Execute a lifecycle test — interleaved actions and assertions.
///
/// Unlike `run_registry_test`, this processes a sequence of typed steps
/// (add, remove, wait, run, assert) rather than "add all then test".
#[allow(clippy::too_many_arguments)]
pub async fn run_lifecycle_test(
    vm: &dyn Executor,
    name: &str,
    steps: &[StepDef],
    verbose: bool,
    prefixed: bool,
    registry_path: &std::path::Path,
    retest: bool,
    on_event: Option<OnEvent>,
) -> ScenarioResult {
    let start = Instant::now();
    let mut events = Vec::new();
    let mut failed = false;
    // Line prefix: `[name] ` when output from parallel tests interleaves on one
    // terminal, empty when a `---- START name ----` banner already scopes the
    // block (serial bare runs).
    let p = if prefixed {
        format!("[{name}] ")
    } else {
        String::new()
    };
    let stream_prefix = if prefixed { name } else { "" };

    for step in steps {
        // In retest mode, skip setup steps and only run test/assertion steps.
        if retest && step.is_setup() {
            let desc = step.step_name();
            println!("{p}  skip  {desc} (retest)");
            continue;
        }

        if failed {
            let desc = step.step_name();
            emit(
                &mut events,
                Event::bare(
                    desc.clone(),
                    EventKind::Step,
                    Outcome::Skipped,
                    Duration::ZERO,
                ),
                &on_event,
            );
            println!("{p}  skip  {desc}");
            continue;
        }

        match step {
            StepDef::Add {
                service,
                args,
                env,
                timeout,
                project_path,
            } => {
                println!("{p}  ryra add {service}...");
                let mut cmd = String::new();
                for (key, val) in env {
                    let escaped = shell_escape(val);
                    cmd.push_str(&format!("{key}='{escaped}' "));
                }
                // Local projects add by path so `ryra add ./project` reads
                // the project's own service.toml; registry services add by
                // name. Either way the install registers under `service`.
                let add_target = match project_path {
                    Some(p) => shell_escape(&p.display().to_string()),
                    None => service.clone(),
                };
                cmd.push_str(&format!("ryra add {add_target}"));
                if let Some(a) = args.as_deref()
                    && !a.is_empty()
                {
                    cmd.push_str(&format!(" {a}"));
                }
                let event = run_event(vm, EventKind::Step, &cmd, *timeout).await;
                print_event_result(&p, &event);
                if event.outcome.is_fail() {
                    failed = true;
                }
                emit(&mut events, event, &on_event);
            }
            StepDef::Remove { service } => {
                println!("{p}  ryra remove --purge {service}...");
                let event = run_event(
                    vm,
                    EventKind::Step,
                    &format!("ryra remove --purge {service} -y"),
                    120,
                )
                .await;
                print_event_result(&p, &event);
                if event.outcome.is_fail() {
                    failed = true;
                }
                emit(&mut events, event, &on_event);
            }
            StepDef::Wait { service, timeout } => {
                println!("{p}  waiting for {service}...");
                let event = wait_for_service_with_timeout(vm, &p, service, *timeout).await;
                print_event_result(&p, &event);
                if event.outcome.is_fail() {
                    failed = true;
                }
                emit(&mut events, event, &on_event);
            }
            StepDef::Shell {
                name: step_name,
                run,
                timeout,
                poll,
            } => {
                println!("{p}  run: {step_name}...");
                let event = run_step_with_poll(
                    vm,
                    step_name,
                    run,
                    *timeout,
                    poll.as_ref(),
                    verbose,
                    stream_prefix,
                )
                .await;
                print_event_result(&p, &event);
                if event.outcome.is_fail() {
                    failed = true;
                }
                emit(&mut events, event, &on_event);
            }
            StepDef::Http {
                name: http_name,
                url,
                method,
                body,
                content_type,
                headers,
                status,
                service,
                poll,
                timeout,
            } => {
                let step_name = http_name.as_deref().unwrap_or(url);
                println!("{p}  http: {step_name}...");
                // Source service .env files for variable expansion ($SERVICE_PORT_HTTP etc.),
                // follow redirects (-L), skip TLS verification (-k) for self-signed certs.
                // URL uses double quotes so shell variables expand.
                let url_esc = url.replace('"', r#"\""#);
                let env_source = match service {
                    Some(svc) => load_env_shell(&format!("{DATA_ROOT_SH}/{svc}/.env")),
                    None => format!(
                        "for __f in {DATA_ROOT_SH}/*/.env; do [ -f \"$__f\" ] && {}; done",
                        load_env_shell("\"$__f\"")
                    ),
                };
                // Assemble curl. For non-GET methods we prepend a heredoc so
                // the body flows verbatim into a $BODY variable — this
                // dodges all the shell-quoting edge cases of embedding
                // arbitrary JSON/form bodies directly in the command string.
                let verb = method.as_curl_arg();
                let ct_esc = content_type.replace('"', r#"\""#);
                // Extra headers — rendered as a sequence of `-H "K: V"` args.
                // Values go through double-quotes so $VAR expansion against
                // the sourced .env files works (useful for apikey, tokens).
                let header_args = headers
                    .iter()
                    .map(|(k, v)| {
                        let k = k.replace('"', r#"\""#);
                        let v = v.replace('"', r#"\""#);
                        format!(r#" -H "{k}: {v}""#)
                    })
                    .collect::<String>();
                let curl = match body {
                    Some(b) => format!(
                        "BODY=$(cat <<'HTTP_BODY_EOF'\n{b}\nHTTP_BODY_EOF\n) && \
                         HTTP_CODE=$(curl -skL -o /dev/null -w '%{{http_code}}' \
                            -X {verb} \
                            -H \"Content-Type: {ct_esc}\"{header_args} \
                            --data-raw \"$BODY\" \
                            \"{url_esc}\")"
                    ),
                    None => format!(
                        "HTTP_CODE=$(curl -skL -o /dev/null -w '%{{http_code}}' -X {verb}{header_args} \"{url_esc}\")"
                    ),
                };
                let cmd = format!(
                    "set -a && {env_source} && set +a && {curl} && \
                     [ \"$HTTP_CODE\" = \"{status}\" ]"
                );
                let event = run_step_with_poll(
                    vm,
                    step_name,
                    &cmd,
                    *timeout,
                    poll.as_ref(),
                    verbose,
                    stream_prefix,
                )
                .await;
                print_event_result(&p, &event);
                if event.outcome.is_fail() {
                    failed = true;
                }
                emit(&mut events, event, &on_event);
            }
            StepDef::Playwright {
                name: browser_name,
                spec,
                env,
                timeout,
            } => {
                let step_name = browser_name.as_deref().unwrap_or(spec);
                println!("{p}  browser: {step_name}...");
                let event = run_browser_step(
                    vm,
                    name,      // test name (for report paths)
                    step_name, // step name (for event description)
                    spec,
                    env,
                    *timeout,
                    registry_path,
                )
                .await;
                print_event_result(&p, &event);
                if event.outcome.is_fail() {
                    failed = true;
                }
                emit(&mut events, event, &on_event);
            }
            StepDef::Mail {
                name: mail_name,
                mailbox,
                contains,
                poll,
                timeout,
            } => {
                let step_name = mail_name.as_deref().unwrap_or(mailbox);
                println!("{p}  mail: {step_name}...");
                // Single-shot probe: discover inbucket's port, fetch the
                // mailbox JSON, check non-empty + (optional) substring.
                // run_step_with_poll retries this until the mail lands.
                let mailbox_esc = shell_escape(mailbox);
                let contains_check = match contains {
                    Some(c) => {
                        format!(" && echo \"$BODY\" | grep -q -- '{}'", shell_escape(c),)
                    }
                    None => String::new(),
                };
                let cmd = format!(
                    "INBUCKET_PORT=$(grep SERVICE_PORT_HTTP {DATA_ROOT_SH}/inbucket/.env 2>/dev/null | cut -d= -f2); \
                     [ -n \"$INBUCKET_PORT\" ] || {{ echo 'inbucket not installed -- no ~/.local/share/services/inbucket/.env'; exit 2; }}; \
                     BODY=$(curl -sf \"http://127.0.0.1:$INBUCKET_PORT/api/v1/mailbox/{mailbox_esc}\" 2>/dev/null); \
                     [ -n \"$BODY\" ] && [ \"$BODY\" != '[]' ]{contains_check}"
                );
                let event = run_step_with_poll(
                    vm,
                    step_name,
                    &cmd,
                    *timeout,
                    Some(poll),
                    verbose,
                    stream_prefix,
                )
                .await;
                print_event_result(&p, &event);
                if event.outcome.is_fail() {
                    failed = true;
                }
                emit(&mut events, event, &on_event);
            }
        }
    }

    let outcome = if failed {
        let first_failure = events
            .iter()
            .find_map(|e| match &e.outcome {
                Outcome::Failed(msg) => Some(msg.clone()),
                _ => None,
            })
            .unwrap_or_else(|| "unknown failure".to_string());
        Outcome::Failed(first_failure)
    } else {
        Outcome::Passed
    };

    ScenarioResult {
        name: name.to_string(),
        events,
        duration: start.elapsed(),
        outcome,
    }
}

/// Execute a run step, optionally retrying on failure via poll config.
async fn run_step_with_poll(
    vm: &dyn Executor,
    step_name: &str,
    cmd: &str,
    timeout_secs: u64,
    poll: Option<&crate::test_toml::PollConfig>,
    verbose: bool,
    stream_prefix: &str,
) -> Event {
    let t = Instant::now();

    match poll {
        None => {
            // Single execution — same as before
            if verbose {
                run_event_streaming(vm, stream_prefix, EventKind::Step, cmd, timeout_secs).await
            } else {
                run_event(vm, EventKind::Step, cmd, timeout_secs).await
            }
        }
        Some(poll_cfg) => {
            // Retry loop. Print a progress tick every few attempts so long
            // polls (e.g. 80 × 3s = 4min) don't look like the runner hung.
            let mut last_err = String::new();
            let tick_every = (poll_cfg.attempts / 10).max(1);
            for attempt in 1..=poll_cfg.attempts {
                let timeout = Duration::from_secs(timeout_secs);
                let result = tokio::time::timeout(timeout, vm.exec(cmd)).await;

                match result {
                    Ok(Ok(out)) => {
                        return Event {
                            description: format!("run: {step_name}"),
                            kind: EventKind::Step,
                            outcome: Outcome::Passed,
                            duration: t.elapsed(),
                            stdout: out.stdout,
                            stderr: out.stderr,
                        };
                    }
                    Ok(Err(e)) => {
                        last_err = format!("{e:#}");
                    }
                    Err(_) => {
                        last_err = format!("timed out after {timeout_secs}s");
                    }
                }

                if attempt < poll_cfg.attempts {
                    if attempt == 1 || attempt % tick_every == 0 {
                        let line = last_err.lines().next().unwrap_or("").trim();
                        let snippet: String = line.chars().take(80).collect();
                        if stream_prefix.is_empty() {
                            println!("    retrying ({attempt}/{}): {snippet}", poll_cfg.attempts);
                        } else {
                            println!(
                                "[{stream_prefix}]     retrying ({attempt}/{}): {snippet}",
                                poll_cfg.attempts
                            );
                        }
                    }
                    tokio::time::sleep(Duration::from_secs(poll_cfg.interval)).await;
                }
            }

            Event::bare(
                format!("run: {step_name}"),
                EventKind::Step,
                Outcome::Failed(format!(
                    "failed after {} attempts (interval={}s): {last_err}",
                    poll_cfg.attempts, poll_cfg.interval
                )),
                t.elapsed(),
            )
        }
    }
}

/// Wait for a port to accept TCP connections (not just be bound by rootlessport).
///
/// Uses bash `/dev/tcp` to test actual TCP connectivity through to the
/// container, not just whether rootlessport is listening on the host side.
async fn wait_for_port(vm: &dyn Executor, prefix: &str, port: &str) -> Event {
    let t = Instant::now();
    let timeout = Duration::from_secs(60);
    let mut progress =
        ryra_vm::progress::WaitProgress::new(format!("port {port}"), "tcp connect", timeout)
            .with_prefix(format!("{prefix}    "));
    // First few seconds: rootlessport is listening but the container app
    // may not be ready yet. A successful bash /dev/tcp probe means the
    // connection made it all the way to the container.
    loop {
        let cmd = format!("bash -c 'echo > /dev/tcp/127.0.0.1/{port}' 2>/dev/null");
        if vm.exec(&cmd).await.is_ok() {
            return Event::bare(
                format!("port {port} ready"),
                EventKind::Step,
                Outcome::Passed,
                t.elapsed(),
            );
        }

        if progress.timed_out() {
            return Event::bare(
                format!("port {port} ready"),
                EventKind::Step,
                Outcome::Failed(format!(
                    "port {port} not responding after {}s",
                    timeout.as_secs()
                )),
                t.elapsed(),
            );
        }

        progress.tick();
        tokio::time::sleep(Duration::from_secs(3)).await;
    }
}

/// Dump diagnostic info for each service when a test fails. `prefix` is the
/// test's line prefix (see run_registry_test) so diagnostics align with the
/// rest of the test's output.
async fn dump_diagnostics(vm: &dyn Executor, prefix: &str, services: &[&str]) {
    println!("{prefix}--- diagnostics ---");
    for svc in services {
        // Systemd service status
        let cmd = format!("systemctl --user status {svc}.service 2>&1 | head -20 || true");
        if let Ok(out) = vm.exec(&cmd).await {
            let trimmed = out.stdout.trim();
            if !trimmed.is_empty() {
                println!("{prefix}  [{svc}] systemd status:");
                for line in trimmed.lines() {
                    println!("{prefix}    {line}");
                }
            }
        }

        // Container status
        let cmd = "podman ps -a --format '{{.Names}} {{.Status}} {{.Ports}}' 2>&1 || true";
        if let Ok(out) = vm.exec(cmd).await {
            let trimmed = out.stdout.trim();
            if !trimmed.is_empty() {
                println!("{prefix}  [{svc}] containers: {trimmed}");
            } else {
                println!("{prefix}  [{svc}] containers: (none)");
            }
        }

        // Journal logs
        let cmd = format!("journalctl --user -u {svc}.service --no-pager -n 30 2>&1 || true");
        if let Ok(out) = vm.exec(&cmd).await {
            let trimmed = out.stdout.trim();
            if !trimmed.is_empty() {
                println!("{prefix}  [{svc}] logs:");
                for line in trimmed.lines().take(30) {
                    println!("{prefix}    {line}");
                }
            }
        }

        // Env file
        let cmd = format!("cat {DATA_ROOT_SH}/{svc}/.env 2>&1 | grep PORT || true");
        if let Ok(out) = vm.exec(&cmd).await {
            let trimmed = out.stdout.trim();
            if !trimmed.is_empty() {
                println!("{prefix}  [{svc}] ports: {trimmed}");
            }
        }

        // Check quadlet, container internals, and network
        let cmd = format!(
            "echo '=== quadlet ==='; grep -i exec $HOME/.config/containers/systemd/{svc}.container 2>/dev/null || true; \
             echo '=== container process ==='; podman exec {svc} ps aux 2>&1 | head -10 || true; \
             echo '=== container listeners ==='; podman exec {svc} cat /proc/net/tcp6 2>&1 | head -10 || true; \
             echo '=== host listeners ==='; ss -tlnp 2>/dev/null | head -20; \
             echo '=== curl ==='; curl -sv http://127.0.0.1:18789/ 2>&1 | head -10 || true"
        );
        if let Ok(out) = vm.exec(&cmd).await {
            let trimmed = out.stdout.trim();
            println!("{prefix}  [{svc}] network:");
            for line in trimmed.lines() {
                println!("{prefix}    {line}");
            }
        }
    }
    println!("{prefix}--- end diagnostics ---");
}

/// Run a command in the VM with real-time output streaming and return an Event.
async fn run_event_streaming(
    vm: &dyn Executor,
    test_name: &str,
    kind: EventKind,
    cmd: &str,
    timeout_secs: u64,
) -> Event {
    let t = Instant::now();
    let timeout = Duration::from_secs(timeout_secs);
    let result = tokio::time::timeout(timeout, vm.exec_streaming(cmd, test_name)).await;

    let (outcome, stdout, stderr) = match result {
        Ok(Ok(out)) => (Outcome::Passed, out.stdout, out.stderr),
        Ok(Err(e)) => (
            Outcome::Failed(format!("{e:#}")),
            String::new(),
            String::new(),
        ),
        Err(_) => (
            Outcome::Failed(format!("timed out after {timeout_secs}s")),
            String::new(),
            String::new(),
        ),
    };

    Event {
        description: cmd.to_string(),
        kind,
        outcome,
        duration: t.elapsed(),
        stdout,
        stderr,
    }
}

/// Run a command in the VM and return an Event.
async fn run_event(vm: &dyn Executor, kind: EventKind, cmd: &str, timeout_secs: u64) -> Event {
    let t = Instant::now();
    let timeout = Duration::from_secs(timeout_secs);
    let result = tokio::time::timeout(timeout, vm.exec(cmd)).await;

    let (outcome, stdout, stderr) = match result {
        Ok(Ok(out)) => (Outcome::Passed, out.stdout, out.stderr),
        Ok(Err(e)) => (
            Outcome::Failed(format!("{e:#}")),
            String::new(),
            String::new(),
        ),
        Err(_) => (
            Outcome::Failed(format!("timed out after {timeout_secs}s")),
            String::new(),
            String::new(),
        ),
    };

    Event {
        description: cmd.to_string(),
        kind,
        outcome,
        duration: t.elapsed(),
        stdout,
        stderr,
    }
}