node-app-build 6.4.1

//! Live-stack probes: bring-up, status, mine, fund, channel, pay, logs, down.
//!
//! Each function corresponds to one `harness` subcommand.  They all operate
//! against the running stack and share the `HarnessState` index written by
//! `up()`.

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

use anyhow::{Context, Result};
#[cfg(unix)]
use libc;
use serde_json::json;

use crate::commands::dev;
use crate::commands::dev::agent::client::AgentHttpClient;
use crate::commands::dev::agent::session::AgentSession;
use crate::commands::dev::host::{self, DaemonHost, InstanceProfile, Mode};
use crate::commands::harness::bitcoind::Bitcoind;
use crate::commands::harness::state::{BitcoindState, ChannelState, HarnessState, InstanceState, state_path};
use crate::commands::harness::BitcoindMode;

/// Bring up the full harness stack, write `harness-state.json`, and then
/// **block as a foreground supervisor** keeping alice + bob alive.
///
/// Sequence:
///   1. (optional clean) remove any prior state file + wipe partial DBs
///   2. bitcoind — LDK connects to it at daemon boot, so it MUST come first
///   3. build + spawn alice + bob directly via the host layer
///      (`host::for_mode(...).ensure_running()`), which blocks until each
///      daemon's IPC socket + HTTP port are ready
///   4. onboard + cross-seed via `agent::run_agent_setup`
///   5. load the agent sessions, build + save HarnessState (with supervisor pid)
///   6. print state JSON + a readiness line, then block until SIGTERM/SIGINT
///   7. on signal: shut each daemon down, stop bitcoind, return Ok
///
/// Unlike the old `dev::run(once: true)` path (which killed the daemons after
/// onboarding), the daemons stay alive because this process supervises them:
/// the `hosts` Vec (which owns each daemon `Child` + its stdio drain threads)
/// is kept in scope for the whole supervise loop.
///
/// When `--with-channel` is passed, `open_channel_flow` funds alice on-chain,
/// opens an alice→bob channel, and waits for it to become usable before the
/// READY line is printed.  Full `harness down` reuses the `supervisor_pid` written here.
pub fn up(
    monorepo_path: std::path::PathBuf,
    bitcoind_mode: BitcoindMode,
    with_channel: bool,
    clean: bool,
) -> Result<()> {
    if clean {
        // Best-effort removal of a prior state file before re-upping.
        // Also wipe the per-instance node DBs and session files so
        // fresh onboarding succeeds (nodes with existing users reject
        // a new onboarding-challenge with HTTP 409).
        // Full teardown (stopping daemons + container) is added when
        // harness down is implemented.
        if let Ok(path) = state_path() {
            if path.exists() {
                let _ = std::fs::remove_file(&path);
            }
        }
        for instance in ["alice", "bob"] {
            if let Ok(env_dir) = crate::commands::dev::host::monorepo::monorepo_dev_dir(
                &monorepo_path,
                instance,
                None,
            ).map(|dev_dir| dev_dir.parent().unwrap_or(&dev_dir).to_path_buf()) {
                // dev.db (and WAL/SHM) hold the node user account — wipe so
                // fresh onboarding is allowed by the server. lightning.db +
                // ldk_data hold LDK/BDK's persisted chain state (channel_manager
                // best block, header cache). Every `up` starts a fresh `--rm`
                // regtest bitcoind (ephemeral chain from genesis), so stale
                // ldk_data references block hashes that no longer exist on the
                // new chain → initial `synchronize_listeners` fails with
                // `RpcError -5 "Block not found"`, the node never leaves the
                // retry loop, and its chain tip freezes at the old height →
                // channel funding never confirms. Wipe them here so LDK starts
                // in lockstep with the fresh chain (mirrors `down --clean`).
                for name in ["dev.db", "dev.db-shm", "dev.db-wal", "lightning.db"] {
                    let _ = std::fs::remove_file(env_dir.join(name));
                }
                for name in ["ldk_data", "ldk_node_data", "ldk_node_data_backup"] {
                    let _ = std::fs::remove_dir_all(env_dir.join(name));
                }
                // Remove the stale session file so load_for_harness doesn't
                // return a session that references a now-deleted user.
                let dev_dir = env_dir.join("dev-apps");
                let _ = std::fs::remove_file(
                    crate::commands::dev::agent::session::AgentSession::file_path(&dev_dir, instance)
                );
            }
        }
    }

    // Pre-flight: if a node DB exists but the session file does not, we're in
    // a partial state from a prior interrupted run.  Wipe the DB so that
    // `node-server` boots clean and fresh onboarding succeeds.  This is
    // safe because the session file is the only persistent auth artifact
    // the harness relies on; without it the node is effectively unrecoverable
    // for our purposes anyway.
    for instance in ["alice", "bob"] {
        if let Ok(dev_dir) = crate::commands::dev::host::monorepo::monorepo_dev_dir(
            &monorepo_path,
            instance,
            None,
        ) {
            let session_exists =
                crate::commands::dev::agent::session::AgentSession::file_path(&dev_dir, instance).exists();
            let env_dir = dev_dir.parent().unwrap_or(&dev_dir).to_path_buf();
            let db_exists = env_dir.join("dev.db").exists();
            if db_exists && !session_exists {
                eprintln!(
                    "harness: {instance} DB exists but session missing — \
                     wiping DB for clean re-onboard"
                );
                // Wipe LDK/BDK chain state too: even a non-`--clean` up gets a
                // fresh `--rm` genesis chain, so stale ldk_data reproduces the
                // frozen-tip stall documented in the `--clean` block above.
                for name in ["dev.db", "dev.db-shm", "dev.db-wal", "lightning.db"] {
                    let _ = std::fs::remove_file(env_dir.join(name));
                }
                for name in ["ldk_data", "ldk_node_data", "ldk_node_data_backup"] {
                    let _ = std::fs::remove_dir_all(env_dir.join(name));
                }
            }
        }
    }

    // Install the shared SIGINT/SIGTERM handler now, so a signal arriving
    // mid-bring-up is observed rather than terminating the process abruptly
    // and orphaning the daemons.
    dev::install_shutdown_handler();

    // 1. bitcoind first — the daemons' LDK dials it at boot.
    let btc = Bitcoind::ensure(bitcoind_mode).context("bring up regtest bitcoind")?;

    // 2. Build one host per instance and bring each up. `ensure_running()`
    //    BLOCKS until the daemon's IPC socket + HTTP port are ready, and stores
    //    the daemon `Child` (plus its stdio drain threads) inside the host.
    //    node-server is built at most once (guarded by DAEMON_BUILD_ONCE).
    let mode = Mode::Monorepo { path: monorepo_path.clone() };
    let profiles = [InstanceProfile::alice(), InstanceProfile::bob()];
    let hosts: Vec<Box<dyn DaemonHost>> = profiles
        .iter()
        .map(|p| host::for_mode(mode.clone(), p.clone(), None, None, None, false))
        .collect();

    let mut handles = Vec::new();
    for h in &hosts {
        handles.push(h.ensure_running().context("start platform daemon")?);
    }

    // 3. Onboard (BIP39) + cross-seed peer IP pools. Best-effort/advisory.
    dev::agent::run_agent_setup(&handles, None);

    // 4. Read the session files the agent step wrote, build state.
    // Derive the LDK peer address from the InstanceProfile so there is a single
    // source of truth (InstanceProfile::p2p_port) rather than hardcoded literals.
    let mut instances = Vec::new();
    for profile in &profiles {
        let name = profile.name.as_str();
        let ldk_addr = format!("127.0.0.1:{}", profile.p2p_port);
        let session = AgentSession::load_for_harness(&monorepo_path, name)
            .with_context(|| format!("load {name} agent session"))?;
        let dev_dir = crate::commands::dev::host::monorepo::monorepo_dev_dir(
            &monorepo_path,
            name,
            None,
        )?;
        let session_path = AgentSession::file_path(&dev_dir, name);
        instances.push(InstanceState {
            name: name.into(),
            session_path,
            base_url: session.base_url.clone(),
            ldk_addr,
            node_id: session.node_id.clone(),
            pid: None,
        });
    }

    let mut state = HarnessState {
        created_at: now_iso8601(),
        bitcoind: BitcoindState {
            mode: format!("{bitcoind_mode:?}").to_lowercase(),
            rpc_url: btc.rpc_url.clone(),
            rpc_user: crate::commands::harness::bitcoind::RPC_USER.into(),
            container_id: btc.container_id.clone(),
        },
        instances,
        channel: None,
        supervisor_pid: Some(std::process::id()),
    };

    // 4a. Open alice→bob channel if requested.
    if with_channel {
        let ch = open_channel_flow(&state, &btc, "alice", "bob", 100_000)?;
        state.channel = Some(ch);
    }

    // 5. Persist + announce readiness.
    let path = state.save()?;
    println!("{}", serde_json::to_string_pretty(&state)?);
    eprintln!(
        "HARNESS READY — supervising alice+bob; state at {}. \
         Send SIGTERM (or `node-app harness down`) to stop.",
        path.display()
    );

    // 6. Block as supervisor until SIGTERM/SIGINT. `hosts` stays alive in scope
    //    for the whole loop — dropping it would kill the daemons and abandon
    //    the drain threads that keep their piped stdio flowing.
    while !dev::shutdown_requested() {
        std::thread::sleep(Duration::from_millis(500));
    }

    // 7. Clean teardown (a preview of `harness down`): SIGTERM each daemon's
    //    process group, then stop the bitcoind container.
    eprintln!("harness: shutdown signal received — stopping alice+bob…");
    for h in &hosts {
        h.shutdown();
    }
    let _ = btc.stop();
    eprintln!("harness: teardown complete.");
    Ok(())
}

/// Fund alice on-chain, connect her to bob, open a channel, and wait until
/// the channel is usable.  Mirrors the `startWithChannel` sequence in
/// `tests/e2e/src/harness/test-harness.ts`.
///
/// Sequence:
///   1. Get a fresh on-chain address from `from` node.
///   2. Send 1 BTC to it via bitcoind → mine 6 blocks → sleep 3s (confirmations).
///   3. Connect `from` to `to` via `connect_peer`.
///   4. Open channel: capacity = `sats`, push = 10% (rounded down) of `sats`.
///   5. Mine 6 blocks → sleep 5s (funding tx confirmations + LDK chain monitor).
///   6. Poll `list_channels` every 2s (mine 1 block per tick) until
///      `is_usable || is_channel_ready`, or 120s deadline → bail.
pub fn open_channel_flow(
    state: &HarnessState,
    btc: &Bitcoind,
    from: &str,
    to: &str,
    sats: u64,
) -> Result<ChannelState> {
    let from_i = state.instance(from)?;
    let to_i = state.instance(to)?;

    let from_tok = AgentSession::read_token(&from_i.session_path)?;

    let client = AgentHttpClient::new(from_i.base_url.clone());

    // 1. Fund: get address, send 1 BTC, mine 6 confirmation blocks.
    eprintln!("harness: funding {from} on-chain (1 BTC)…");
    let addr = client.new_onchain_address(&from_tok)?;
    btc.send_to_address(&addr, 1.0)?;
    eprintln!("harness: mining 6 confirmation blocks…");
    btc.mine(6)?;
    std::thread::sleep(Duration::from_secs(3));

    // 2. Connect from → to.
    let peer_str = format!("{}@{}", to_i.node_id, to_i.ldk_addr);
    eprintln!("harness: connecting {from} → {to} ({peer_str})…");
    client.connect_peer(&from_tok, &peer_str)?;

    // 3. Open channel with 10% push to counterparty.
    let push_msat = (sats / 10) * 1_000;
    eprintln!(
        "harness: opening channel {from}→{to}: {sats} sats, pushing {} sats to {to}…",
        push_msat / 1_000
    );
    client.open_channel(&from_tok, &peer_str, sats, push_msat)?;

    // 4. Mine 6 blocks so the funding tx gets confirmed; give LDK time to react.
    eprintln!("harness: mining 6 channel confirmation blocks…");
    btc.mine(6)?;
    std::thread::sleep(Duration::from_secs(5));

    // 5. Poll until channel is usable (or timeout after 120s).
    eprintln!("harness: waiting for channel to become usable (120s deadline)…");
    let deadline = Instant::now() + Duration::from_secs(120);
    loop {
        let channels = client.list_channels(&from_tok)?;

        // channels is a JSON Value (array); check each element.
        let is_usable = if let Some(arr) = channels.as_array() {
            arr.iter().any(|ch| {
                ch.get("is_usable").and_then(|v| v.as_bool()).unwrap_or(false)
                    || ch.get("is_channel_ready").and_then(|v| v.as_bool()).unwrap_or(false)
            })
        } else {
            false
        };

        if is_usable {
            eprintln!("harness: channel {from}→{to} is usable");
            return Ok(ChannelState {
                from: from.into(),
                to: to.into(),
                capacity_sats: sats,
                status: "usable".into(),
            });
        }

        if Instant::now() >= deadline {
            anyhow::bail!(
                "channel {from}→{to} not usable within 120s; last: {channels}"
            );
        }

        // Mine 1 block per tick to keep LDK's chain monitor moving.
        let _ = btc.mine(1);
        std::thread::sleep(Duration::from_secs(2));
    }
}

fn now_iso8601() -> String {
    chrono::Utc::now().to_rfc3339()
}

// ─── Probe subcommands ────────────────────────────────────────────────────────

/// `harness status` — print balance/channels/peers for each instance as JSON.
pub fn status() -> Result<()> {
    let state = load_state()?;
    let mut result = serde_json::Map::new();
    for inst in &state.instances {
        let token = AgentSession::read_token(&inst.session_path)
            .with_context(|| format!("read token for {}", inst.name))?;
        let client = AgentHttpClient::new(inst.base_url.clone());

        let balance = client.get_balance(&token)
            .unwrap_or_else(|e| json!({ "error": e.to_string() }));
        let channels = client.list_channels(&token)
            .unwrap_or_else(|e| json!({ "error": e.to_string() }));
        let peers = client.list_peers(&token)
            .unwrap_or_else(|e| json!({ "error": e.to_string() }));

        result.insert(inst.name.clone(), json!({
            "balance": balance,
            "channels": channels,
            "peers": peers,
        }));
    }
    println!("{}", serde_json::to_string_pretty(&serde_json::Value::Object(result))?);
    Ok(())
}

/// `harness mine <blocks>` — mine N regtest blocks and print the new height.
pub fn mine(blocks: u32) -> Result<()> {
    let state = load_state()?;
    let btc = Bitcoind::from_state(&state.bitcoind);
    btc.mine(blocks)?;
    let height = btc.block_count()?;
    println!("{}", json!({ "mined": blocks, "height": height }));
    Ok(())
}

/// `harness fund <node> <btc>` — send BTC to a node's on-chain address,
/// mine 6 confirmations, and print the address + txid.
pub fn fund(node: &str, btc_amount: f64) -> Result<()> {
    let state = load_state()?;
    let inst = state.instance(node)?;
    let token = AgentSession::read_token(&inst.session_path)
        .with_context(|| format!("read token for {node}"))?;
    let client = AgentHttpClient::new(inst.base_url.clone());

    let addr = client.new_onchain_address(&token)?;
    let btc_handle = Bitcoind::from_state(&state.bitcoind);
    let txid = btc_handle.send_to_address(&addr, btc_amount)?;
    btc_handle.mine(6)?;

    println!("{}", json!({
        "address": addr,
        "txid": txid,
        "confirmations": 6,
    }));
    Ok(())
}

/// `harness channel-open <from> <to> <sats>` — open a channel and persist the
/// result into the harness state.
pub fn channel_open(from: &str, to: &str, sats: u64) -> Result<()> {
    let mut state = load_state()?;
    let btc = Bitcoind::from_state(&state.bitcoind);
    let ch = open_channel_flow(&state, &btc, from, to, sats)?;
    let ch_json = serde_json::to_value(&ch)?;
    state.channel = Some(ch);
    state.save()?;
    println!("{}", serde_json::to_string_pretty(&ch_json)?);
    Ok(())
}

/// `harness invoke <node> <capability> <payload>` — invoke a capability on a node.
pub fn invoke(node: &str, capability: &str, payload_str: &str) -> Result<()> {
    let state = load_state()?;
    let inst = state.instance(node)?;
    let token = AgentSession::read_token(&inst.session_path)
        .with_context(|| format!("read token for {node}"))?;
    let client = AgentHttpClient::new(inst.base_url.clone());

    let payload: serde_json::Value = serde_json::from_str(payload_str)
        .with_context(|| format!("parse payload JSON: {payload_str}"))?;

    let response = client.invoke_capability(&token, capability, payload)?;
    println!("{}", serde_json::to_string_pretty(&response)?);
    Ok(())
}

/// `harness logs <node> [--tail]` — print (or tail) the daemon log for a node.
pub fn logs(node: &str, tail: bool) -> Result<()> {
    let state = load_state()?;
    let inst = state.instance(node)?;

    // The session file lives at `<instance-root>/dev-apps/<name>-agent-session.json`.
    // The daemon writes `daemon.log` at `<instance-root>/daemon.log`.
    // Walk up two levels: dev-apps → instance root.
    let instance_root = inst.session_path
        .parent()                       // dev-apps/
        .and_then(|p| p.parent())       // <instance-root>/
        .ok_or_else(|| anyhow::anyhow!("session_path has no grandparent"))?;
    let log_path = instance_root.join("daemon.log");

    if !log_path.exists() {
        anyhow::bail!("log not found: {}", log_path.display());
    }

    let content = std::fs::read_to_string(&log_path)
        .with_context(|| format!("read {}", log_path.display()))?;

    if tail {
        // Print last ~200 lines.
        let lines: Vec<&str> = content.lines().collect();
        let start = lines.len().saturating_sub(200);
        for line in &lines[start..] {
            println!("{}", line);
        }
    } else {
        print!("{}", content);
    }
    Ok(())
}

/// `harness down [--clean]` — stop the harness via the supervisor PID, or
/// best-effort pkill if no supervisor PID is stored.
pub fn down(clean: bool) -> Result<()> {
    let state = match HarnessState::load() {
        Ok(s) => s,
        Err(_) => {
            println!("{}", json!({ "stopped": false, "reason": "no running harness" }));
            return Ok(());
        }
    };

    #[cfg(unix)]
    {
        if let Some(pid) = state.supervisor_pid {
            eprintln!("harness down: sending SIGTERM to supervisor PID {pid}");
            // SAFETY: kill() is always safe to call; the PID might not exist, which
            // just returns ESRCH.
            let rc = unsafe { libc::kill(pid as libc::pid_t, libc::SIGTERM) };
            if rc != 0 {
                // errno is not portable to read inline; just check if the process
                // still exists by probing with signal 0.
                let still_alive = unsafe { libc::kill(pid as libc::pid_t, 0) } == 0;
                if !still_alive {
                    eprintln!("harness down: supervisor {pid} is no longer running");
                } else {
                    eprintln!("harness down: kill({pid}, SIGTERM) returned non-zero");
                }
            } else {
                // Poll up to 40s for the supervisor to exit.
                // MonorepoHost::shutdown() is sequential per host (daemon ~10s + UI
                // ~5s each), so alice+bob worst-case is ~30s.  Give an extra 10s
                // margin before we escalate.
                let deadline = Instant::now() + Duration::from_secs(40);
                let mut timed_out = false;
                loop {
                    std::thread::sleep(Duration::from_millis(300));
                    let probe = unsafe { libc::kill(pid as libc::pid_t, 0) };
                    if probe != 0 {
                        // ESRCH = no such process → supervisor is gone.
                        break;
                    }
                    if Instant::now() >= deadline {
                        eprintln!(
                            "harness down: supervisor {pid} did not exit within 40s — \
                             escalating to SIGKILL"
                        );
                        timed_out = true;
                        break;
                    }
                }

                if timed_out {
                    // SIGKILL the supervisor so it cannot continue leaking daemons.
                    // SAFETY: kill() is always safe; ESRCH is ignored.
                    unsafe { libc::kill(pid as libc::pid_t, libc::SIGKILL) };
                    // Best-effort: also kill any node-server processes that the
                    // supervisor may have spawned but not yet cleaned up.
                    let _ = std::process::Command::new("pkill")
                        .args(["-f", "node-server --env"])
                        .status();
                }
            }
        } else {
            // Stale state without supervisor PID — best-effort cleanup.
            eprintln!("harness down: no supervisor_pid in state, falling back to best-effort teardown");
            let _ = std::process::Command::new("pkill")
                .args(["-f", "node-server --env"])
                .status();
        }
    }

    // Always stop the bitcoind container, regardless of supervisor state.
    // The supervisor's clean-teardown path stops it too, but if the supervisor
    // was already dead (SIGTERM errored / confirmed gone) or failed to exit
    // within the poll window, that stop never ran and the container would leak.
    // Stopping an already-stopped or absent container is a harmless ignored error.
    if let Some(id) = &state.bitcoind.container_id {
        let _ = std::process::Command::new("docker").args(["stop", id]).status();
    }

    // Optional clean: wipe per-instance data files.
    if clean {
        // Determine the cache root so we never wipe outside it.
        let cache_root = state_path()
            .ok()
            .and_then(|p| {
                // Walk up to find the `node-app` ancestor.
                let mut cur = p.parent()?.to_path_buf();
                loop {
                    if cur.file_name().map(|n| n == "node-app").unwrap_or(false) {
                        return Some(cur);
                    }
                    if !cur.pop() {
                        return None;
                    }
                }
            });

        for inst in &state.instances {
            // session_path = <instance-root>/dev-apps/<name>-agent-session.json
            // Data files (dev.db, lightning.db, ldk_data) live at <instance-root>.
            let instance_root = match inst.session_path
                .parent()                  // dev-apps/
                .and_then(|p| p.parent())  // <instance-root>/
            {
                Some(d) => d.to_path_buf(),
                None => continue,
            };

            // Canonicalize so that a path containing `..` components cannot
            // lexically pass the cache-root guard.  The directory must exist
            // at this point (it held the session file), so canonicalize()
            // should succeed; fall back to the raw path if it doesn't.
            let instance_root = instance_root.canonicalize().unwrap_or(instance_root);

            // Guard: only wipe if the path is under the cache root.
            let under_cache = cache_root.as_ref().map(|root| instance_root.starts_with(root)).unwrap_or(false);
            if !under_cache {
                eprintln!(
                    "harness down: skipping clean for {} (not under cache root)",
                    instance_root.display()
                );
                continue;
            }

            // Wipe database files.
            for name in ["dev.db", "dev.db-shm", "dev.db-wal", "lightning.db"] {
                let _ = std::fs::remove_file(instance_root.join(name));
            }

            // Wipe LDK data directories.
            for name in ["ldk_data", "ldk_node_data", "ldk_node_data_backup"] {
                let p = instance_root.join(name);
                if p.is_dir() {
                    let _ = std::fs::remove_dir_all(&p);
                }
            }
        }
    }

    // Remove the state file.
    let _ = std::fs::remove_file(state_path()?);

    println!("{}", json!({ "stopped": true, "cleaned": clean }));
    Ok(())
}

/// `harness pay <from> <to> <sats>` — `to` creates an invoice; `from` pays it.
///
/// Flow:
///   1. `to` creates a BOLT11 invoice for `sats * 1000` msat.
///   2. `from` sends the payment.
///   3. If the pay response is `status == "pending"` (the LDK capability
///      returned before the preimage was extracted), poll
///      `core.lightning.get_payment_status` on `from` up to ~30s.
///   4. Print `{ invoice, status, payment_hash, preimage }` and return.
///
/// A "no route / no channel" failure exits non-zero with a message pointing
/// at `harness channel-open`.
pub fn pay(from: &str, to: &str, sats: u64) -> Result<()> {
    let state = load_state()?;
    let from_i = state.instance(from)?;
    let to_i = state.instance(to)?;

    let from_tok = AgentSession::read_token(&from_i.session_path)
        .with_context(|| format!("read token for {from}"))?;
    let to_tok = AgentSession::read_token(&to_i.session_path)
        .with_context(|| format!("read token for {to}"))?;

    let from_client = AgentHttpClient::new(from_i.base_url.clone());
    let to_client = AgentHttpClient::new(to_i.base_url.clone());

    // 1. `to` creates an invoice.
    let bolt11 = to_client
        .create_invoice(&to_tok, sats * 1_000, "harness pay")
        .with_context(|| format!("{to} create_invoice failed"))?;

    // 2. `from` pays the invoice.
    let pay_resp = from_client
        .pay_invoice(&from_tok, &bolt11)
        .map_err(|e| {
            let msg = e.to_string();
            // Detect "no route" or "channel" errors and hint the user.
            if msg.to_lowercase().contains("route")
                || msg.to_lowercase().contains("channel")
                || msg.to_lowercase().contains("liquidity")
                || msg.to_lowercase().contains("no path")
            {
                anyhow::anyhow!(
                    "payment failed — no usable route/channel: {msg}\n\
                     hint: run `node-app harness channel-open {from} {to} 100000` first"
                )
            } else {
                anyhow::anyhow!("pay_invoice failed: {msg}")
            }
        })?;

    let payment_hash = pay_resp
        .get("payment_hash")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_string();

    let initial_status = pay_resp
        .get("status")
        .and_then(|v| v.as_str())
        .unwrap_or("unknown")
        .to_string();

    let initial_preimage = pay_resp
        .get("preimage")
        .and_then(|v| v.as_str())
        .map(str::to_owned);

    // 3. If still pending, poll get_payment_status up to ~30s.
    let (final_status, final_preimage) = if initial_status == "pending"
        && !payment_hash.is_empty()
    {
        eprintln!("harness pay: payment pending — polling status (30s deadline)…");
        let deadline = Instant::now() + Duration::from_secs(30);
        let mut status = initial_status.clone();
        let mut preimage = initial_preimage.clone();

        loop {
            std::thread::sleep(Duration::from_millis(500));

            match from_client.invoke_capability(
                &from_tok,
                "core.lightning.get_payment_status",
                json!({ "payment_hash": payment_hash }),
            ) {
                Ok(v) => {
                    status = v
                        .get("status")
                        .and_then(|s| s.as_str())
                        .unwrap_or("unknown")
                        .to_string();
                    preimage = v
                        .get("preimage")
                        .and_then(|p| p.as_str())
                        .map(str::to_owned);

                    if status == "succeeded" || status == "failed" {
                        break;
                    }
                }
                Err(e) => {
                    eprintln!("harness pay: get_payment_status error: {e}");
                }
            }

            if Instant::now() >= deadline {
                eprintln!("harness pay: payment still pending after 30s");
                break;
            }
        }

        (status, preimage)
    } else {
        (initial_status, initial_preimage)
    };

    // 4. Print result.
    println!(
        "{}",
        serde_json::to_string_pretty(&json!({
            "invoice": bolt11,
            "status": final_status,
            "payment_hash": payment_hash,
            "preimage": final_preimage,
        }))?
    );

    if !matches!(final_status.as_str(), "succeeded" | "settled") {
        anyhow::bail!(
            "payment did not settle (status: {final_status}); \
             check that a usable channel exists: `node-app harness channel-open {from} {to} 100000`"
        );
    }

    Ok(())
}

/// `harness l402 <from> <to> <route> <payload>` — drive a cross-node L402 paid
/// call where `from`'s daemon proxies to `to` via its L402HttpClient.
///
/// `from` calls `route` on its OWN daemon (authenticated).  Internally the
/// daemon's transport layer looks up `to` in the IP pool and sends the request
/// to `to`'s daemon.  If `to` requires payment, alice's L402HttpClient
/// auto-settles using the channel and retries.
///
/// Outcome:
///   - 200 → `{ "route", "settled": true, "response": … }` + exit 0.
///   - 402 or unconfigured route → `{ "route", "settled": false, "hint": "…" }`
///     + exit non-zero (soft outcome — dev env may have no paid route).
pub fn l402(from: &str, to: &str, route: &str, payload_str: &str) -> Result<()> {
    let state = load_state()?;
    let from_i = state.instance(from)?;
    // Validate `to` is a known instance (used for the hint message only;
    // actual routing is via the IP pool registered during `harness up`).
    let _to_i = state.instance(to)?;

    let from_tok = AgentSession::read_token(&from_i.session_path)
        .with_context(|| format!("read token for {from}"))?;

    let payload: serde_json::Value = serde_json::from_str(payload_str)
        .with_context(|| format!("parse payload JSON: {payload_str}"))?;

    // Force the CROSS-NODE path: without `execution_preference: "remote"`,
    // `from`'s daemon handles the call LOCALLY (no L402 proxy to `to`, no
    // cross-node settlement) and a 200 would be a false positive for a probe
    // named "l402". Inject the field if the payload is a JSON object, without
    // overwriting a caller-supplied value.
    let mut payload = payload;
    if let Some(obj) = payload.as_object_mut() {
        obj.entry("execution_preference".to_string())
            .or_insert(serde_json::Value::String("remote".to_string()));
    }

    let from_client = AgentHttpClient::new(from_i.base_url.clone());

    let (status_code, body) = from_client
        .post_raw_with_status(&from_tok, route, &payload)
        .with_context(|| format!("POST {route} on {from}"))?;

    if status_code == 200 {
        println!(
            "{}",
            serde_json::to_string_pretty(&json!({
                "route": route,
                "settled": true,
                "note": "settled=true means the remote route returned 200; \
                         in a dev env with no paid route configured the daemon \
                         may handle locally",
                "response": body,
            }))?
        );
        Ok(())
    } else {
        // Soft outcome: 402 means no channel / no paid route; other 4xx/5xx
        // also treated as soft so the harness doesn't thrash.
        let hint = if status_code == 402 {
            format!(
                "received 402 — no channel or budget / no paid route configured \
                 between {from} and {to}; \
                 open a channel first: `node-app harness channel-open {from} {to} 100000`"
            )
        } else {
            format!(
                "route returned HTTP {status_code} (may be unconfigured or require \
                 additional setup)"
            )
        };

        println!(
            "{}",
            serde_json::to_string_pretty(&json!({
                "route": route,
                "settled": false,
                "http_status": status_code,
                "hint": hint,
                "body": body,
            }))?
        );

        anyhow::bail!("l402 probe: route {route} did not return 200 (got {status_code})");
    }
}

// ─── Internal helper ──────────────────────────────────────────────────────────

fn load_state() -> Result<HarnessState> {
    HarnessState::load().map_err(|_| {
        anyhow::anyhow!(
            "no harness state found — run `node-app harness up` first"
        )
    })
}