tropel-engine 0.6.0

//! TR-405: the localhost agent server. knockport's desktop transport reaches
//! the SAME engine over this socket — one engine from Send to 10 000 VU.
//!
//! The agent is a plain HTTP server on a loopback address (refuses any
//! non-loopback bind), token-authenticated and rate-limited, because it is an
//! arbitrary-request-execution endpoint reachable from any local process.
//!
//! Endpoints:
//!   POST /execute   { method, url, headers, body, follow_redirects } →
//!                   { status, status_text, headers, body, timings, error }
//!                   — a single request with full sub-timings (the same code
//!                   path a request under load takes).
//!   GET  /version   → the agent's tropel version (TR-406 handshake).

use std::collections::HashMap;
use std::net::IpAddr;
use std::sync::Arc;
use std::time::Instant;

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

use tropel_sdk::types::{Body, Method, Request as TropelRequest, ResponseType};
use tropel_sdk::TropelError;

/// Max requests per second per connection (a crude rate limit for the
/// arbitrary-request-execution endpoint).
const RATE_LIMIT_PER_SEC: u64 = 200;

/// Shared agent state: the auth token and the engine's HTTP client.
/// Records every request a SCRIPT issued, in call order (TR-477).
///
/// The app renders these as timeline arms, the same view the main request
/// gets. Without them a script-issued send is invisible on the agent tier —
/// which is the KP-413 defect ("script-issued sends were invisible") landing
/// again, one tier over.
///
/// Recorded at the HTTP CLIENT, not in the shims. `pm.sendRequest` and
/// `fetch` are two spellings that both end at `__tropel_trp_send_request`,
/// which ends here — so one wrapper catches both, and a third spelling added
/// later is caught without anyone remembering to wire it up. Wrapping the
/// shims instead would have meant one recorder per spelling.
#[derive(Default)]
struct ScriptSendLog {
    sends: std::sync::Mutex<Vec<serde_json::Value>>,
}

impl ScriptSendLog {
    fn record(&self, entry: serde_json::Value) {
        let mut g = match self.sends.lock() {
            Ok(g) => g,
            Err(e) => e.into_inner(),
        };
        g.push(entry);
    }
    fn drain(&self) -> Vec<serde_json::Value> {
        let g = match self.sends.lock() {
            Ok(g) => g,
            Err(e) => e.into_inner(),
        };
        g.clone()
    }
}

/// A `DriverHttpClient` that records what it sent and defers to the real one.
struct RecordingHttpClient {
    inner: Arc<dyn tropel_sdk::traits::DriverHttpClient>,
    log: Arc<ScriptSendLog>,
}

#[async_trait::async_trait]
impl tropel_sdk::traits::DriverHttpClient for RecordingHttpClient {
    async fn execute(
        &self,
        req: &tropel_sdk::Request,
    ) -> tropel_sdk::Result<tropel_sdk::types::Response> {
        let started = std::time::Instant::now();
        let out = self.inner.execute(req).await;
        // A FAILED script send is still a send. Dropping it here would make a
        // script whose request never connected look like a script that never
        // made one (invariant 7).
        let (status, error) = match &out {
            Ok(res) => (res.status_code, None),
            Err(e) => (0u16, Some(e.to_string())),
        };
        self.log.record(serde_json::json!({
            "source": "sendRequest",
            "kind": "ad-hoc",
            "method": req.method.to_string(),
            "url": req.url,
            "status": status,
            "responseTime": started.elapsed().as_millis() as u64,
            "error": error,
        }));
        out
    }
}

/// The per-script cookie jar (TR-476).
///
/// A SNAPSHOT, deliberately, not the authority. The caller owns the real jar;
/// it seeds this one for the duration of the script and replays the recorded
/// ops afterwards. That keeps one jar authoritative instead of two that drift,
/// and it is why reads here need only be good enough to serve the script that
/// is running — not to re-implement the caller's jar.
#[derive(Default)]
struct ScriptCookies {
    jar: std::sync::Mutex<Vec<serde_json::Value>>,
    /// What the script DID, in order. Returned as `cookieOps` so the caller
    /// can apply the same changes to the jar that outlives the script.
    ops: std::sync::Mutex<Vec<serde_json::Value>>,
}

impl ScriptCookies {
    fn lock_jar(&self) -> std::sync::MutexGuard<'_, Vec<serde_json::Value>> {
        match self.jar.lock() {
            Ok(g) => g,
            Err(e) => e.into_inner(),
        }
    }
    fn lock_ops(&self) -> std::sync::MutexGuard<'_, Vec<serde_json::Value>> {
        match self.ops.lock() {
            Ok(g) => g,
            Err(e) => e.into_inner(),
        }
    }
    fn record(&self, op: serde_json::Value) {
        self.lock_ops().push(op);
    }
}

/// Does `cookie` apply to `url`?
///
/// Host suffix + path prefix, which is the matching a script needs to read
/// back what it and the caller put in. NOT a full RFC 6265 implementation and
/// not trying to be: the caller's jar decides what is actually sent, and this
/// snapshot only has to answer the script honestly for the values it holds.
fn cookie_matches_url(cookie: &serde_json::Value, url: &str) -> bool {
    let host = url
        .split("://")
        .nth(1)
        .unwrap_or(url)
        .split('/')
        .next()
        .unwrap_or("")
        .split(':')
        .next()
        .unwrap_or("");
    let path = {
        let after = url.split("://").nth(1).unwrap_or(url);
        match after.find('/') {
            Some(i) => after[i..].split('?').next().unwrap_or("/").to_string(),
            None => "/".to_string(),
        }
    };
    let domain = cookie
        .get("domain")
        .and_then(|d| d.as_str())
        .unwrap_or("")
        .trim_start_matches('.');
    let c_path = cookie.get("path").and_then(|p| p.as_str()).unwrap_or("/");
    let domain_ok = domain.is_empty() || host == domain || host.ends_with(&format!(".{domain}"));
    domain_ok && path.starts_with(c_path)
}

/// One script→host call parked until the caller answers it (TR-474).
#[derive(Clone, serde::Serialize)]
struct PendingHostCall {
    #[serde(rename = "callId")]
    call_id: u64,
    kind: &'static str,
    path: String,
}

/// The bidirectional half of `/script`.
///
/// TR-474. `pm.sendRequest` and `fetch` are servable by the agent's own HTTP
/// client (TR-472) — on desktop the agent IS the local transport. But
/// `bru.runRequest` resolves a request BY NAME out of a collection the agent
/// has never seen, and re-enters the caller's own pipeline (auth, variables,
/// its recursion guard). No widening of the request body can carry that: the
/// script has to be able to call BACK, mid-execution.
///
/// `POST /script` is one request and one reply, so the call-back rides two
/// extra endpoints instead: the caller parks on `/script/callback/next` for
/// work, and answers on `/script/callback/reply`.
///
/// The two halves deliberately use different primitives. The JS host function
/// is SYNCHRONOUS — QuickJS gives it no way to suspend — so it blocks on a
/// std channel. The long-poll is async and waits on a tokio `Notify`. Mixing
/// them is the point: each side blocks in the way its own runtime allows.
#[derive(Default)]
struct RunCallbacks {
    pending: std::sync::Mutex<std::collections::VecDeque<PendingHostCall>>,
    ready: tokio::sync::Notify,
    replies: std::sync::Mutex<HashMap<u64, std::sync::mpsc::Sender<String>>>,
    next_id: std::sync::atomic::AtomicU64,
    /// The script finished. Parks the long-poll on 204 instead of hanging
    /// until its timeout — a caller that keeps polling a finished run would
    /// otherwise look like a stalled agent.
    done: std::sync::atomic::AtomicBool,
}

impl RunCallbacks {
    /// Mark finished and wake every parked poller. Called on EVERY exit path
    /// of a run, success or failure — a run that ended by erroring must not
    /// leave its caller parked.
    fn finish(&self) {
        self.done.store(true, std::sync::atomic::Ordering::SeqCst);
        self.ready.notify_waiters();
    }
}

struct AgentState {
    token: Option<String>,
    client: tropel_http::HttpClient,
    /// Origins allowed to reach this agent from a browser (TR-459).
    ///
    /// An ALLOWLIST, never `*`, and empty by default. The agent holds
    /// collection variables and OAuth client secrets and will execute any
    /// request it is handed — so echoing an arbitrary `Origin` would let any
    /// page the user happens to have open drive their local agent. The token
    /// is not a substitute: a browser attaches it automatically once CORS
    /// permits the call.
    allowed_origins: Vec<String>,
    /// TR-474: in-flight `/script` runs that opted into host callbacks,
    /// keyed by the caller's `runId`. Empty for every run that did not.
    runs: std::sync::Mutex<HashMap<String, Arc<RunCallbacks>>>,
}

/// CORS headers for a request carrying `origin`, or `None` when the browser
/// must not be told it may proceed.
///
/// TR-459: KT-402 (the website talking to a local agent) needs two things that
/// are easy to get almost-right:
///
///   1. `http://localhost` is a *potentially trustworthy* origin per W3C
///      secure-contexts, so an HTTPS page may fetch it without mixed-content
///      blocking. That part is the browser's doing and needs nothing here.
///   2. Chrome additionally requires PRIVATE NETWORK ACCESS: a preflight
///      carrying `Access-Control-Request-Private-Network: true` must be
///      answered with `Access-Control-Allow-Private-Network: true`. Omit it
///      and this works everywhere except Chrome — the horrible-to-find-late
///      bug the plan calls out by name.
fn cors_headers(state: &AgentState, origin: Option<&str>, is_preflight: bool) -> Option<String> {
    let origin = origin?;
    if !state.allowed_origins.iter().any(|o| o == origin) {
        return None;
    }
    let mut h = format!("Access-Control-Allow-Origin: {origin}\r\n");
    // The allowlist is per-origin, so caches must key on it.
    h.push_str("Vary: Origin\r\n");
    // The token rides as a header, not a cookie, so credentials stay off —
    // turning them on would let a page reuse the user's ambient session.
    if is_preflight {
        h.push_str("Access-Control-Allow-Methods: GET, POST, OPTIONS\r\n");
        h.push_str("Access-Control-Allow-Headers: Authorization, Content-Type\r\n");
        h.push_str("Access-Control-Max-Age: 600\r\n");
    }
    Some(h)
}

/// Start the agent server. Refuses a non-loopback bind address (the register:
/// "Refuses to start with an obviously-wrong bind address rather than exposing
/// an execution endpoint to the network").
pub async fn run_agent(
    port: u16,
    bind: &str,
    token: Option<&str>,
    allowed_origins: &[String],
    exit_with_parent: bool,
) -> tropel_sdk::Result<()> {
    let ip: IpAddr = bind
        .parse()
        .map_err(|_| TropelError::Other(format!("invalid bind address: {bind}")))?;
    if !ip.is_loopback() {
        return Err(TropelError::Other(format!(
            "refusing to bind {bind}: the agent is a localhost-only execution endpoint (TR-405)"
        )));
    }

    // TR-471: die with the process that spawned us.
    //
    // The desktop shell kills the agent on window-destroy, but that handler
    // does not run when the shell is SIGKILLed, crashes, or is restarted by a
    // dev rebuild — and every one of those leaves an agent listening on a
    // loopback port, holding the collection variables and OAuth client
    // secrets it was sent, with nothing left to talk to it. Observed twice in
    // one session: two agents reparented to init while the app was gone.
    //
    // A parent-side handler cannot close this on its own (it is precisely the
    // cases where the parent runs no code), so the agent has to notice. On
    // Unix an orphan is reparented, so the parent id CHANGING is the signal —
    // no pid to pass in, and no way to mistake a recycled pid for the parent
    // still being alive.
    if exit_with_parent {
        // Watch stdin for EOF, on a dedicated thread.
        //
        // The FIRST attempt at this compared parent process ids — an orphan is
        // reparented, so a changed parent id looks like a death signal. It
        // failed its own test: a process can be reparented before it ever
        // reads its parent id, in which case "changed" never happens and the
        // watchdog is a no-op that looks armed. That is the wrong failure
        // direction for something whose whole job is to not leak.
        //
        // A pipe cannot be fooled that way. The spawning process hands us a
        // stdin pipe and simply holds it; when it dies for ANY reason — exit,
        // crash, or SIGKILL, which runs no cleanup on either side — the OS
        // closes the write end and this read returns 0. It needs no pid, wins
        // no race, and works the same on Windows.
        //
        // The contract this places on the caller is real and worth stating:
        // stdin MUST be a pipe the parent keeps open. Handed /dev/null or a
        // closed descriptor, the read returns 0 at once and the agent exits
        // immediately — which is why the flag says so, and why the exit says
        // which of the two happened rather than vanishing silently.
        std::thread::spawn(|| {
            use std::io::Read;
            let mut stdin = std::io::stdin();
            let mut buf = [0u8; 256];
            loop {
                match stdin.read(&mut buf) {
                    Ok(0) | Err(_) => {
                        // Deliberately ONE message naming both causes. An
                        // earlier version tried to tell them apart with a
                        // "did we ever read a byte" flag, which is wrong: a
                        // parent that holds the pipe open and never writes to
                        // it — the normal case — looks identical to a parent
                        // that never gave us a pipe at all. A diagnostic that
                        // confidently names the wrong cause is worse than one
                        // that names both.
                        tracing::info!(
                            "tropel agent exiting: stdin reached EOF — either the process that \
                             spawned it is gone, or stdin was not a pipe held open by it \
                             (--exit-with-parent requires one)"
                        );
                        std::process::exit(0);
                    }
                    Ok(_) => {}
                }
            }
        });
    }

    let addr = format!("{bind}:{port}");
    let listener = TcpListener::bind(&addr).await.map_err(TropelError::Io)?;
    tracing::info!(
        "tropel agent listening on http://{addr} (token auth {})",
        if token.is_some() { "on" } else { "off" }
    );

    let http_config = tropel_http::HttpConfig::default();
    let client = tropel_http::HttpClient::new(&http_config)
        .map_err(|e| TropelError::Other(format!("http client init failed: {e}")))?;
    let state = Arc::new(AgentState {
        token: token.map(str::to_string),
        client,
        allowed_origins: allowed_origins.to_vec(),
        runs: std::sync::Mutex::new(HashMap::new()),
    });

    loop {
        let (mut sock, peer) = listener.accept().await.map_err(TropelError::Io)?;
        tracing::debug!("agent: connection from {peer}");
        let state = state.clone();
        tokio::spawn(async move {
            if let Err(e) = handle_connection(&mut sock, state).await {
                tracing::debug!("agent: connection error: {e}");
            }
        });
    }
}

async fn handle_connection(sock: &mut TcpStream, state: Arc<AgentState>) -> tropel_sdk::Result<()> {
    // Read the HTTP request head (bounded buffer; we only support simple
    // POST/GET with a JSON body).
    let mut buf = vec![0u8; 64 * 1024];
    let n = sock.read(&mut buf).await.map_err(TropelError::Io)?;
    let raw = String::from_utf8_lossy(&buf[..n]);

    // TR-445: any body bytes that arrived in the SAME read as the head.
    //
    // This single `read` routinely returns head AND body together — a small
    // JSON POST is one TCP segment, which is the normal case, not an edge
    // one. Every handler below then called `read_exact(content_length)` and
    // waited for bytes that had already been delivered, so the connection
    // hung until the client gave up. It was masked because clients that write
    // the head and body in separate calls (curl, reqwest) happen to split the
    // segments.
    let prefetched_body: Vec<u8> = raw
        .find("\r\n\r\n")
        .map(|i| buf[i + 4..n].to_vec())
        .unwrap_or_default();

    // Parse the request line, path, and headers.
    let mut lines = raw.lines();
    let request_line = lines.next().unwrap_or("");
    let mut parts = request_line.split_whitespace();
    let method = parts.next().unwrap_or("");
    let path = parts.next().unwrap_or("/");
    let mut content_length = 0usize;
    let mut auth_header = String::new();
    let mut origin: Option<String> = None;
    let mut wants_private_network = false;
    for line in lines {
        if line.is_empty() {
            break;
        }
        if let Some((k, v)) = line.split_once(':') {
            let key = k.trim().to_ascii_lowercase();
            if key == "content-length" {
                content_length = v.trim().parse().unwrap_or(0);
            } else if key == "authorization" {
                auth_header = v.trim().to_string();
            } else if key == "origin" {
                origin = Some(v.trim().to_string());
            } else if key == "access-control-request-private-network" {
                wants_private_network = v.trim().eq_ignore_ascii_case("true");
            }
        }
    }

    let cors = cors_headers(&state, origin.as_deref(), false);

    // Rate limit on every request (a fresh limiter per connection — good
    // enough for the localhost boundary).
    //
    // TR-459 moved this AFTER the head is parsed. It costs no extra I/O — the
    // head is already in the buffer from the single read above — and it means
    // a rate-limited browser gets a 429 it can actually READ. Answered before
    // the Origin was known, the reply carried no CORS header, so the page saw
    // a CORS failure and the real cause never reached the user.
    {
        let mut limiter = RateLimiter::new();
        if limiter.allow().is_err() {
            return respond_raw_cors(sock, cors.as_deref(), 429, "rate limit exceeded").await;
        }
    }

    // TR-459: the CORS preflight, answered BEFORE the auth check — a browser
    // never sends `Authorization` on a preflight, so requiring the token here
    // would refuse every cross-origin call the allowlist was meant to permit.
    if method == "OPTIONS" {
        let Some(mut headers) = cors_headers(&state, origin.as_deref(), true) else {
            // Named, not a bare 403. "The agent is not running" and "the agent
            // is running and does not trust this page" are different problems
            // and the page can only tell them apart if we say so.
            return respond(
                sock,
                403,
                &error_body(&format!(
                    "origin {:?} is not allowed to reach this agent \u{2014} start it with --allow-origin {}",
                    origin.as_deref().unwrap_or("(none)"),
                    origin.as_deref().unwrap_or("<origin>")
                )),
            )
            .await;
        };
        // Chrome's Private Network Access. A public page reaching 127.0.0.1
        // gets a preflight carrying `Access-Control-Request-Private-Network`,
        // and it must be answered explicitly. Answered only when ASKED, so
        // the header never appears on a same-origin or non-Chrome preflight
        // that did not request it.
        if wants_private_network {
            headers.push_str("Access-Control-Allow-Private-Network: true\r\n");
        }
        return respond_raw(sock, 204, "", Some(&headers)).await;
    }

    if let Some(expected) = &state.token {
        if auth_header != format!("Bearer {expected}") {
            return respond_raw_cors(sock, cors.as_deref(), 401, r#"{"error":"unauthorized"}"#)
                .await;
        }
    }

    match (method, path) {
        ("GET", "/version") => {
            let body = format!(r#"{{"version":"{}"}}"#, env!("CARGO_PKG_VERSION"));
            respond_raw_cors(sock, cors.as_deref(), 200, &body).await
        }
        // ── TR-445 · the RULES endpoints ─────────────────────────────────────
        //
        // The agent exposed request EXECUTION only, which is why every
        // core-tier method in knockport's `native-agent.ts` USED TO throw
        // `TropelCoreUnavailableError` naming this gap. Desktop ships no wasm,
        // so without these the only way to resolve a variable or sign a
        // request there was a TypeScript re-implementation — invariant #3, and
        // the most expensive recurring bug class in both repos.
        //
        // TR-464: those methods FORWARD now (knockport KP-209), batched at the
        // provider so a burst of 33 template resolutions costs one round trip
        // rather than 33. The present tense here described the state this
        // endpoint set was built to end.
        //
        // These are pure functions over JSON: same Rust the wasm tier calls,
        // reached over the loopback socket instead of a wasm boundary.
        ("POST", "/resolve/batch") => {
            // TR-448 (knockport KP-209): resolve MANY templates in one call.
            //
            // `/resolve` takes a single template, and knockport's
            // `resolveRequest` walks ~33 of them per request — url, headers,
            // params, auth fields, body. Per-call that is 33 loopback round
            // trips at 70 us each: 2.3 ms of pure overhead on every send,
            // measured. Batched it is one trip, 0.07 ms.
            //
            // That difference is the whole reason the desktop tier can use
            // the agent at all instead of shipping a second copy of this Rust
            // as wasm.
            let Some(payload) =
                read_json_body(sock, content_length, 8 * 1024 * 1024, &prefetched_body).await?
            else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    400,
                    r#"{"error":"invalid JSON body"}"#,
                )
                .await;
            };
            let vars: HashMap<String, String> = payload
                .get("variables")
                .and_then(|v| serde_json::from_value(v.clone()).ok())
                .unwrap_or_default();
            let items: Vec<BatchResolveItem> =
                match serde_json::from_value(payload.get("items").cloned().unwrap_or_default()) {
                    Ok(v) => v,
                    Err(e) => {
                        return respond_raw_cors(
                            sock,
                            cors.as_deref(),
                            400,
                            &error_body(&format!("invalid items: {e}")),
                        )
                        .await
                    }
                };
            // ORDER IS THE CONTRACT: the caller re-assembles its request by
            // index, so a reordered or short reply would put a header's value
            // in a param. One output per input, always, even for the ones
            // that fail.
            // TR-449: each item reports `hitCap` and `unresolved`, not just a
            // string. KnockPort's `resolveVariables` uses them to tell a CYCLE
            // (`{{a}}` -> `{{b}}` -> `{{a}}`, a failed send) from an UNKNOWN
            // NAME (the user's typo, left visible and sent). Both leave a
            // literal `{{…}}` in the text, so a bare string cannot distinguish
            // them — and only the resolver's own loop knows which happened.
            let scope = tropel_variables::VariableScope {
                env: vars.clone(),
                ..Default::default()
            };
            let resolver = tropel_variables::VariableResolver::new();
            let mut out = Vec::with_capacity(items.len());
            for item in &items {
                let mode = item.mode.as_deref().unwrap_or("plain");
                // TR-449: `deep: false` is REFUSED here, not ignored. The
                // batched path exists to serve `resolveTemplateDetailed`,
                // whose report only means something for a chain the resolver
                // ran to settlement: a shallow pass stops BY DESIGN, so it has
                // no cap to hit. Emulating one with `max_passes = 1` would
                // report `{{a}}` -> `{{b}}` as a CYCLE, and silently upgrading
                // to deep is the worse half of the same trade — the caller
                // asked for one pass, got twenty, and cannot tell. `POST
                // /resolve` still answers a shallow single resolve.
                if item.deep == Some(false) {
                    out.push(serde_json::json!({
                        "error": "deep: false is not supported by POST /resolve/batch \u{2014} the batched reply reports hitCap/unresolved, which only a chain resolved to settlement has; use POST /resolve for a shallow resolve"
                    }));
                    continue;
                }
                match resolver.resolve_reporting(
                    &item.template,
                    &scope,
                    tropel_variables::MAX_VARIABLE_RESOLUTION_PASSES,
                    mode,
                ) {
                    Ok(outcome) => out.push(serde_json::json!({
                        "value": outcome.value,
                        "hitCap": outcome.hit_cap,
                        "unresolved": outcome.unresolved,
                    })),
                    // A per-item failure does NOT fail the batch: one bad
                    // escape mode must not lose the other 32 resolutions, and
                    // the caller can still see exactly which item broke.
                    Err(why) => out.push(serde_json::json!({ "error": why })),
                }
            }
            respond_raw_cors(
                sock,
                cors.as_deref(),
                200,
                &serde_json::json!({ "items": out }).to_string(),
            )
            .await
        }

        ("POST", "/resolve") => {
            let Some(payload) =
                read_json_body(sock, content_length, 1024 * 1024, &prefetched_body).await?
            else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    400,
                    r#"{"error":"invalid JSON body"}"#,
                )
                .await;
            };
            let template = payload
                .get("template")
                .and_then(|t| t.as_str())
                .unwrap_or("");
            let vars: HashMap<String, String> = payload
                .get("variables")
                .and_then(|v| serde_json::from_value(v.clone()).ok())
                .unwrap_or_default();
            let mode = payload
                .get("mode")
                .and_then(|m| m.as_str())
                .unwrap_or("none");
            let deep = payload
                .get("deep")
                .and_then(|d| d.as_bool())
                .unwrap_or(true);
            match tropel_variables::resolve_template_for_host(template, &vars, mode, deep) {
                Ok(resolved) => {
                    respond(
                        sock,
                        200,
                        &serde_json::json!({ "resolved": resolved }).to_string(),
                    )
                    .await
                }
                // A typo'd mode is a NAMED 400, never a silent fallback to
                // plain — that is how a quote-bearing value corrupts a body.
                Err(why) => respond_raw_cors(sock, cors.as_deref(), 400, &error_body(&why)).await,
            }
        }

        ("POST", "/assert") => {
            let Some(payload) =
                read_json_body(sock, content_length, 8 * 1024 * 1024, &prefetched_body).await?
            else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    400,
                    r#"{"error":"invalid JSON body"}"#,
                )
                .await;
            };
            let target: tropel_variables::assertions::AssertionTarget = match serde_json::from_value(
                payload.get("response").cloned().unwrap_or_default(),
            ) {
                Ok(t) => t,
                Err(e) => {
                    return respond_raw_cors(
                        sock,
                        cors.as_deref(),
                        400,
                        &error_body(&format!("invalid response: {e}")),
                    )
                    .await
                }
            };
            let specs: Vec<AgentAssertionSpec> = match serde_json::from_value(
                payload.get("assertions").cloned().unwrap_or_default(),
            ) {
                Ok(v) => v,
                Err(e) => {
                    return respond_raw_cors(
                        sock,
                        cors.as_deref(),
                        400,
                        &error_body(&format!("invalid assertions: {e}")),
                    )
                    .await
                }
            };
            // A native agent CAN link a regex engine — unlike the wasm tier,
            // where TR-434 removed it and the host's RegExp is injected. Using
            // Rust's `regex` here would make `matches` behave differently on
            // desktop than in the browser, which is precisely the divergence
            // this endpoint exists to prevent. So it is left unwired and the
            // outcome says so BY NAME.
            let outcomes: Vec<_> = specs
                .iter()
                .map(|spec| {
                    let name = spec
                        .name
                        .clone()
                        .unwrap_or_else(|| format!("{} {}", spec.target, spec.operator));
                    match tropel_variables::assertions::resolve_assertion_target(
                        &spec.target,
                        &target,
                    ) {
                        Ok(actual) => tropel_variables::assertions::assert_evaluate(
                            &name,
                            &spec.target,
                            &actual,
                            &spec.operator,
                            &spec.expected,
                            None,
                        ),
                        Err(why) => tropel_variables::assertions::AssertionOutcome {
                            name,
                            passed: false,
                            unsupported: Some(why),
                            message: None,
                        },
                    }
                })
                .collect();
            respond(
                sock,
                200,
                &serde_json::to_string(&outcomes).unwrap_or_default(),
            )
            .await
        }

        ("POST", "/variables/dynamic/batch") => {
            // TR-452: the batched twin, for the same reason /resolve has one.
            // KnockPort's `resolveVariables` calls the dynamic pass for EVERY
            // template, unconditionally, before it looks at the `{{var}}` map
            // — so without this the desktop tier pays a round trip per field.
            //
            // A client-side "skip it unless the text contains `{{$`" would
            // remove those trips without an endpoint, and it is exactly the
            // shortcut not to take: `{{ $guid }}` with spaces is a dynamic
            // token that such a test would miss, and deciding what counts as
            // one is the catalogue's job, not the caller's (invariant #3).
            let Some(payload) =
                read_json_body(sock, content_length, 8 * 1024 * 1024, &prefetched_body).await?
            else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    400,
                    r#"{"error":"invalid JSON body"}"#,
                )
                .await;
            };
            let items: Vec<BatchResolveItem> = payload
                .get("items")
                .and_then(|v| serde_json::from_value(v.clone()).ok())
                .unwrap_or_default();
            // ORDER IS THE CONTRACT, as in /resolve/batch: the caller
            // re-assembles by index, so one output per input, always.
            let catalog = tropel_variables::DynamicCatalog::new();
            let mut out = Vec::with_capacity(items.len());
            for item in &items {
                match catalog.resolve(&item.template) {
                    Ok(value) => out.push(serde_json::json!({ "value": value })),
                    Err(why) => out.push(serde_json::json!({ "error": why })),
                }
            }
            respond_raw_cors(
                sock,
                cors.as_deref(),
                200,
                &serde_json::json!({ "items": out }).to_string(),
            )
            .await
        }

        ("POST", "/variables/dynamic") => {
            // TR-451: `{{$guid}}`, `{{$timestamp}}`, `{{$randomInt}}` — the
            // predefined catalogue. SEPARATE from /resolve on purpose, and
            // that separation is the contract, not an accident: /resolve
            // substitutes the embedder's `{{var}}` map and leaves `{{$…}}`
            // alone, while this one generates a FRESH value per occurrence
            // and leaves plain `{{var}}` alone. KnockPort runs them in that
            // order (`resolveVariables` in packages/core/src/utils.ts), so
            // folding them together here would change which of the two saw a
            // `{{$guid}}` produced by a variable's value.
            let Some(payload) =
                read_json_body(sock, content_length, 1024 * 1024, &prefetched_body).await?
            else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    400,
                    r#"{"error":"invalid JSON body"}"#,
                )
                .await;
            };
            let template = payload
                .get("template")
                .and_then(|t| t.as_str())
                .unwrap_or("");
            let catalog = tropel_variables::DynamicCatalog::new();
            match catalog.resolve(template) {
                Ok(value) => {
                    respond(
                        sock,
                        200,
                        &serde_json::json!({ "value": value }).to_string(),
                    )
                    .await
                }
                // The 16 MiB total-output cap (TR-403). A NAMED failure, never
                // a truncated body: `{{$randomLoremParagraphs}}` in a loop is
                // the shape that hits it, and silently returning half of it is
                // the data loss invariant #7 forbids.
                Err(why) => respond_raw_cors(sock, cors.as_deref(), 400, &error_body(&why)).await,
            }
        }

        ("GET", "/constants") => {
            // TR-451: the values that CANNOT drift between the two hosts, in
            // one fetch at handshake.
            //
            // The pass cap is here because KnockPort was carrying its own
            // `MAX_VARIABLE_RESOLUTION_PASSES_FALLBACK` on the desktop path —
            // a SECOND ceiling, which is the exact duplication KP-424 removed
            // from the resolver itself. A host that stops at 20 talking to an
            // agent that stops at 25 disagrees about which chains are cyclic,
            // and a cycle is a failed send.
            //
            // The predefined catalogue rides along rather than getting its own
            // route: it is equally constant, the editor needs it at the same
            // moment, and one fetch cannot half-succeed the way two can.
            let variables: Vec<serde_json::Value> = tropel_variables::PREDEFINED_VARIABLE_META
                .iter()
                .map(|m| serde_json::json!({ "name": m.name, "description": m.description }))
                .collect();
            let body = serde_json::json!({
                "maxVariableResolutionPasses": tropel_variables::MAX_VARIABLE_RESOLUTION_PASSES,
                "predefinedVariables": variables,
            });
            respond_raw_cors(sock, cors.as_deref(), 200, &body.to_string()).await
        }

        ("GET", "/operators") => {
            // The assertion vocabulary, so a desktop editor renders the SAME
            // dropdown the evaluator dispatches on.
            let body = serde_json::to_string(tropel_variables::assertions::ASSERTION_OPERATORS)
                .unwrap_or_default();
            respond_raw_cors(sock, cors.as_deref(), 200, &body).await
        }

        ("POST", "/auth/sign") => {
            // TR-445: the four request signers, so the desktop tier does not
            // re-implement them in TypeScript. Same shape as `core-wasm`'s
            // exports — RAW request components in, finished headers out — so
            // the AWS service derivation, the S3 double-encoding rule, the
            // RFC 5849 base-string URI and the digest challenge parse all stay
            // on this side (TR-428..TR-431).
            let Some(payload) =
                read_json_body(sock, content_length, 8 * 1024 * 1024, &prefetched_body).await?
            else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    400,
                    r#"{"error":"invalid JSON body"}"#,
                )
                .await;
            };
            let scheme = payload.get("scheme").and_then(|s| s.as_str()).unwrap_or("");
            let params = payload.get("params").cloned().unwrap_or_default();
            match sign_with_scheme(scheme, &params) {
                Ok(headers) => {
                    respond(
                        sock,
                        200,
                        &serde_json::to_string(&headers).unwrap_or_default(),
                    )
                    .await
                }
                Err(why) => respond_raw_cors(sock, cors.as_deref(), 400, &error_body(&why)).await,
            }
        }

        ("POST", "/script") => {
            // TR-446 (KT-203 `run_script`): run a pre/post-request script and
            // return its effects. Same realm a load run uses, so a script that
            // behaves one way in the app behaves the same way under load.
            let Some(payload) =
                read_json_body(sock, content_length, 4 * 1024 * 1024, &prefetched_body).await?
            else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    400,
                    r#"{"error":"invalid JSON body"}"#,
                )
                .await;
            };
            // TR-469: a MISSING or non-string `code` is a caller bug, not an
            // empty script. `.unwrap_or("")` ran nothing and answered 200 with
            // a success-shaped body, so a client that misspelled the field saw
            // "the script ran and did nothing" — the silent success invariant 8
            // forbids. An empty STRING stays legal (a stage with no script).
            let Some(code) = payload.get("code").and_then(|c| c.as_str()) else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    400,
                    r#"{"error":"`code` is required and must be a string"}"#,
                )
                .await;
            };
            // TR-469: the embedder's canonical namespace. pm.js installs the
            // name named here (P4b `__tropel_sandbox_config`); absent a config
            // the stock install applies, whose canonical name is `trp`.
            //
            // The API client's realm is built with namespace "kp", so WITHOUT
            // this the agent realm had no `kp` at all: a `kp.test(...)` script
            // passed in the app and died with "kp is not defined" on desktop —
            // two realms for one script, disagreeing invisibly. tropel stays
            // product-neutral; the client declares its own namespace and sends
            // the SAME value it builds its local realm from.
            let sandbox_cfg = payload
                .get("sandbox")
                .map(|v| tropel_sandbox::config::SandboxConfig {
                    namespace: v
                        .get("namespace")
                        .and_then(|n| n.as_str())
                        .unwrap_or("trp")
                        .to_string(),
                    aliases: v
                        .get("aliases")
                        .and_then(|a| a.as_array())
                        .map(|a| {
                            a.iter()
                                .filter_map(|x| x.as_str().map(str::to_string))
                                .collect()
                        })
                        .unwrap_or_default(),
                })
                .unwrap_or_default();
            let scopes = ScriptScopes::from_payload(&payload);
            // TR-467: the request the script may mutate. Optional, so a
            // caller that only needs environment effects (a bare `pm.test`)
            // keeps working unchanged.
            let script_request: Option<TropelRequest> = payload
                .get("request")
                .and_then(|v| serde_json::from_value(v.clone()).ok());
            let script_response: Option<tropel_sdk::types::Response> = payload
                .get("response")
                .and_then(|v| serde_json::from_value(v.clone()).ok());
            // TR-474: opt-in host callbacks. The CALLER supplies the id, so
            // it can start polling before this request returns — the agent
            // cannot hand one back in a reply that only arrives at the end.
            let run_id = payload
                .get("runId")
                .and_then(|v| v.as_str())
                .map(str::to_string);
            // TR-476: an explicit `cookies` array opts the jar in — even an
            // empty one. Absent entirely means "no jar", which the shim
            // refuses by name rather than reporting as an empty one.
            let cookies = payload
                .get("cookies")
                .and_then(|c| c.as_array())
                .map(|arr| {
                    let sc = ScriptCookies::default();
                    *sc.lock_jar() = arr.clone();
                    Arc::new(sc)
                });

            let callbacks = run_id.as_ref().map(|id| {
                let cb = Arc::new(RunCallbacks::default());
                let mut runs = match state.runs.lock() {
                    Ok(g) => g,
                    Err(e) => e.into_inner(),
                };
                runs.insert(id.clone(), cb.clone());
                cb
            });

            let outcome = run_script_once(
                code,
                scopes,
                script_request,
                script_response,
                sandbox_cfg,
                ScriptHost {
                    http: state.client.clone(),
                    callbacks: callbacks.clone(),
                    cookies: cookies.clone(),
                },
            )
            .await;

            // Deregister on EVERY exit path. A run that ended by erroring
            // must still wake its poller, or the caller waits out the full
            // long-poll on a run that is already gone.
            if let (Some(id), Some(cb)) = (run_id, callbacks) {
                cb.finish();
                let mut runs = match state.runs.lock() {
                    Ok(g) => g,
                    Err(e) => e.into_inner(),
                };
                runs.remove(&id);
            }

            match outcome {
                Ok(out) => respond_raw_cors(sock, cors.as_deref(), 200, &out.to_string()).await,
                Err(why) => respond_raw_cors(sock, cors.as_deref(), 500, &error_body(&why)).await,
            }
        }

        // ── TR-474 · the host-callback channel ───────────────────────────
        //
        // Two endpoints rather than a path parameter, because this dispatch
        // is exact-match on (method, path) and a `/script/{id}/…` route would
        // be the only prefix match in it. The run id rides the body.
        ("POST", "/script/callback/next") => {
            // Park until this run has work, the run ends, or the poll ages
            // out. The caller re-polls on 204; that is the normal idle path,
            // not an error.
            let Some(payload) =
                read_json_body(sock, content_length, 64 * 1024, &prefetched_body).await?
            else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    400,
                    r#"{"error":"invalid JSON body"}"#,
                )
                .await;
            };
            let Some(run_id) = payload.get("runId").and_then(|v| v.as_str()) else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    400,
                    r#"{"error":"`runId` is required"}"#,
                )
                .await;
            };
            let cb = {
                let runs = match state.runs.lock() {
                    Ok(g) => g,
                    Err(e) => e.into_inner(),
                };
                runs.get(run_id).cloned()
            };
            // An unknown run is 204, not 404: the run may have finished
            // between the script returning and this poll arriving, and that
            // race is ordinary rather than an error.
            let Some(cb) = cb else {
                return respond_raw_cors(sock, cors.as_deref(), 204, "").await;
            };

            let deadline = std::time::Duration::from_secs(20);
            let started = std::time::Instant::now();
            loop {
                // The waiter is created BEFORE the check. Reversed, a call
                // enqueued in between would be missed and the poll would park
                // until it aged out, with work already waiting.
                let waiting = cb.ready.notified();
                if let Some(call) = {
                    let mut pending = match cb.pending.lock() {
                        Ok(g) => g,
                        Err(e) => e.into_inner(),
                    };
                    pending.pop_front()
                } {
                    let body = serde_json::to_string(&call).unwrap_or_default();
                    return respond_raw_cors(sock, cors.as_deref(), 200, &body).await;
                }
                if cb.done.load(std::sync::atomic::Ordering::SeqCst) {
                    return respond_raw_cors(sock, cors.as_deref(), 204, "").await;
                }
                let left = deadline.saturating_sub(started.elapsed());
                if left.is_zero() {
                    return respond_raw_cors(sock, cors.as_deref(), 204, "").await;
                }
                tokio::select! {
                    _ = waiting => {}
                    _ = tokio::time::sleep(left) => {}
                }
            }
        }

        ("POST", "/script/callback/reply") => {
            // Deliver one answer. The parked host function is holding the
            // realm's thread, so this must never block on it.
            let Some(payload) =
                read_json_body(sock, content_length, 8 * 1024 * 1024, &prefetched_body).await?
            else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    400,
                    r#"{"error":"invalid JSON body"}"#,
                )
                .await;
            };
            let run_id = payload.get("runId").and_then(|v| v.as_str()).unwrap_or("");
            let call_id = payload.get("callId").and_then(|v| v.as_u64());
            let Some(call_id) = call_id else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    400,
                    r#"{"error":"`callId` is required and must be a number"}"#,
                )
                .await;
            };
            let cb = {
                let runs = match state.runs.lock() {
                    Ok(g) => g,
                    Err(e) => e.into_inner(),
                };
                runs.get(run_id).cloned()
            };
            let Some(cb) = cb else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    404,
                    r#"{"error":"no such run — it may have already finished"}"#,
                )
                .await;
            };
            let tx = {
                let mut replies = match cb.replies.lock() {
                    Ok(g) => g,
                    Err(e) => e.into_inner(),
                };
                replies.remove(&call_id)
            };
            // Answering twice, or answering a call that timed out, is a
            // NAMED 404 rather than a silent success — the caller would
            // otherwise believe the script received something it never did.
            let Some(tx) = tx else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    404,
                    r#"{"error":"no such callId — already answered, or it timed out"}"#,
                )
                .await;
            };
            let result = payload
                .get("result")
                .cloned()
                .unwrap_or(serde_json::Value::Null);
            let _ = tx.send(result.to_string());
            respond_raw_cors(sock, cors.as_deref(), 200, r#"{"delivered":true}"#).await
        }

        ("POST", "/auth/oauth2") => {
            // TR-447: the OAuth2/JWT/WSSE family. Closes the last arm of the
            // gap `native-agent.ts` documents — every method on its
            // `TropelAuthProvider` threw because the agent exposed none of
            // this, and desktop ships no wasm to fall back on.
            let Some(payload) =
                read_json_body(sock, content_length, 1024 * 1024, &prefetched_body).await?
            else {
                return respond_raw_cors(
                    sock,
                    cors.as_deref(),
                    400,
                    r#"{"error":"invalid JSON body"}"#,
                )
                .await;
            };
            let op = payload.get("op").and_then(|o| o.as_str()).unwrap_or("");
            let params = payload.get("params").cloned().unwrap_or_default();
            match oauth2_dispatch(op, &params) {
                Ok(out) => respond_raw_cors(sock, cors.as_deref(), 200, &out.to_string()).await,
                Err(why) => respond_raw_cors(sock, cors.as_deref(), 400, &error_body(&why)).await,
            }
        }

        ("POST", "/execute") => {
            let body_buf = read_body(sock, content_length, 64 * 1024, &prefetched_body).await?;
            let req: serde_json::Value = match serde_json::from_slice(&body_buf) {
                Ok(v) => v,
                Err(_) => {
                    return respond_raw_cors(
                        sock,
                        cors.as_deref(),
                        400,
                        r#"{"error":"invalid JSON body"}"#,
                    )
                    .await
                }
            };
            let out = execute_single(&state, &req).await;
            respond_raw_cors(sock, cors.as_deref(), 200, &out.to_string()).await
        }
        ("POST", "/run") => {
            // TR-411: the relay is explicitly NOT a load transport — the agent
            // refuses a load dispatch that arrives over the relay. A web relay
            // request would be CORS-bridged and would misreport percentiles
            // (the browser tier must not be able to report them). Detect the
            // relay by the header the relay always sets (`X-Tropel-Relay` or
            // the legacy `X-Knockport-Relay` / `Via: relay`) and refuse with a
            // 403 so the client can surface "relay cannot run loads — use the
            // desktop transport or the native CLI".
            let raw_lower = raw.to_ascii_lowercase();
            if raw_lower.contains("x-tropel-relay")
                || raw_lower.contains("x-knockport-relay")
                || raw_lower.contains("via: relay")
                || raw_lower.contains("x-relay-transport")
            {
                return respond(
                    sock,
                    403,
                    r#"{"error":"relay is not a load transport — POST /run refused (TR-411); use the desktop tauri transport or the native CLI"}"#,
                )
                .await;
            }
            // TR-411: a load run — a collection (scenario JSON) plus a load
            // block (iterations). Runs each item through the SAME engine HTTP
            // client, bounded by `iterations`, and returns the aggregated
            // raw samples. NO percentiles (TR-411 — the browser tier cannot
            // report them; this endpoint matches that contract).
            let body_buf =
                read_body(sock, content_length, 4 * 1024 * 1024, &prefetched_body).await?;
            let payload: serde_json::Value = match serde_json::from_slice(&body_buf) {
                Ok(v) => v,
                Err(_) => {
                    return respond_raw_cors(
                        sock,
                        cors.as_deref(),
                        400,
                        r#"{"error":"invalid JSON body"}"#,
                    )
                    .await
                }
            };
            let scenario_json = payload
                .get("scenario")
                .and_then(|s| s.as_str())
                .unwrap_or("");
            let iterations = payload
                .get("iterations")
                .and_then(|i| i.as_u64())
                .unwrap_or(1)
                .min(1000); // bounded — a load run is not an unbounded loop
            let scenario: tropel_sdk::scenario::Scenario = match serde_json::from_str(scenario_json)
            {
                Ok(s) => s,
                Err(e) => {
                    return respond(
                        sock,
                        400,
                        &format!(r#"{{"error":"invalid scenario: {e}"}}"#),
                    )
                    .await
                }
            };
            // TR-411: optional thresholds map — evaluated against the run's
            // http_reqs count + http_req_failed rate; the verdict is returned
            // so the client can use it as the exit code.
            let thresholds: std::collections::HashMap<String, String> = payload
                .get("thresholds")
                .and_then(|t| t.as_object())
                .map(|o| {
                    o.iter()
                        .map(|(k, v)| (k.clone(), v.as_str().unwrap_or("").to_string()))
                        .collect()
                })
                .unwrap_or_default();
            // TR-411: `stream: true` streams each iteration's samples as a
            // chunked response (live metrics) instead of one batched JSON.
            let stream = payload
                .get("stream")
                .and_then(|s| s.as_bool())
                .unwrap_or(false);
            if stream {
                return run_load_streaming(sock, &state, &scenario, iterations, &thresholds).await;
            }
            let out = run_load(&state, &scenario, iterations, &thresholds).await;
            respond_raw_cors(sock, cors.as_deref(), 200, &out.to_string()).await
        }
        _ => respond_raw_cors(sock, cors.as_deref(), 404, r#"{"error":"not found"}"#).await,
    }
}

/// Evaluate a simple `<metric> <op> <value>` threshold against a single
/// number. Supported metrics: `http_reqs` (count), `http_req_failed` (rate).
fn eval_threshold(expr: &str, reqs: u64, failed: u64) -> Result<bool, String> {
    let parts: Vec<&str> = expr.split_whitespace().collect();
    if parts.len() != 3 {
        return Err(format!(
            "invalid threshold '{expr}': expected '<metric> <op> <value>'"
        ));
    }
    let actual = match parts[0] {
        "http_reqs" => reqs as f64,
        "http_req_failed" => {
            if reqs == 0 {
                0.0
            } else {
                failed as f64 / reqs as f64
            }
        }
        other => return Err(format!("unsupported threshold metric '{other}'")),
    };
    let threshold: f64 = parts[2]
        .parse()
        .map_err(|_| format!("invalid threshold value '{}'", parts[2]))?;
    let passed = match parts[1] {
        "<" => actual < threshold,
        "<=" => actual <= threshold,
        ">" => actual > threshold,
        ">=" => actual >= threshold,
        "==" | "===" => (actual - threshold).abs() < f64::EPSILON,
        "!=" => (actual - threshold).abs() > f64::EPSILON,
        other => return Err(format!("unknown operator '{other}'")),
    };
    Ok(passed)
}

/// Run a load run: walk the scenario items `iterations` times, executing each
/// request through the shared engine HTTP client with full sub-timings, and
/// aggregate the raw samples. No percentiles (TR-411).
async fn run_load(
    state: &AgentState,
    scenario: &tropel_sdk::scenario::Scenario,
    iterations: u64,
    thresholds: &std::collections::HashMap<String, String>,
) -> serde_json::Value {
    let mut samples: Vec<serde_json::Value> = Vec::new();
    let mut total_failures = 0u64;
    let mut unsupported_errors: Vec<String> = Vec::new();
    for it in 0..iterations {
        for item in &scenario.items {
            let Some(request) = item.request.as_ref() else {
                continue;
            };
            // TR-409: resolve the signer and surface `unsupported` as a hard
            // failure rather than sending the request unauthenticated. A
            // request that declares `ntlm`/`akamai-edgegrid`/`jwt`/`wsse` on a
            // transport that cannot sign it must fail loudly (the TR-004 shape
            // would be a 200 with no Authorization header and a green run).
            let signer_opt = match &request.auth {
                Some(auth) => match state.client.get_signer(auth) {
                    Ok(s) => s,
                    Err(e) => {
                        let msg = e.to_string();
                        if !unsupported_errors.contains(&msg) {
                            unsupported_errors.push(msg.clone());
                        }
                        total_failures += 1;
                        let elapsed_ms = 0.0;
                        samples.push(serde_json::json!({
                            "metric": "http_reqs",
                            "iteration": it,
                            "url": request.url,
                            "status": 0,
                            "duration_ms": elapsed_ms,
                            "error": msg,
                        }));
                        continue;
                    }
                },
                None => None,
            };
            let start = Instant::now();
            let result = state.client.execute(request, signer_opt.as_deref()).await;
            let elapsed_ms = start.elapsed().as_millis() as f64;
            let (status, ok) = match &result {
                Ok(resp) => (resp.status_code, (200..400).contains(&resp.status_code)),
                Err(_) => (0, false),
            };
            if !ok {
                total_failures += 1;
            }
            let mut sample = serde_json::json!({
                "metric": "http_reqs",
                "iteration": it,
                "url": request.url,
                "status": status,
                "duration_ms": elapsed_ms,
            });
            if let Err(e) = &result {
                sample["error"] = serde_json::Value::String(e.to_string());
            }
            samples.push(sample);
        }
    }
    let mut out = serde_json::json!({
        "iterations": iterations,
        "samples": samples,
        "failures": total_failures,
        "has_failures": total_failures > 0,
        "thresholds": threshold_verdict(thresholds, iterations, total_failures),
    });
    if !unsupported_errors.is_empty() {
        out["unsupported_auth"] = serde_json::Value::Array(
            unsupported_errors
                .into_iter()
                .map(serde_json::Value::String)
                .collect(),
        );
    }
    out
}

/// TR-411: STREAMING load run — writes a chunked HTTP response and emits
/// each iteration's samples as a chunk, so the client sees live metrics
/// during the run instead of a single batched JSON at the end.
async fn run_load_streaming(
    sock: &mut TcpStream,
    state: &AgentState,
    scenario: &tropel_sdk::scenario::Scenario,
    iterations: u64,
    thresholds: &std::collections::HashMap<String, String>,
) -> tropel_sdk::Result<()> {
    // Chunked HTTP head.
    let head = "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n";
    sock.write_all(head.as_bytes())
        .await
        .map_err(TropelError::Io)?;

    let mut total_failures = 0u64;
    for it in 0..iterations {
        let mut batch: Vec<serde_json::Value> = Vec::new();
        for item in &scenario.items {
            let Some(request) = item.request.as_ref() else {
                continue;
            };
            let signer_opt = match &request.auth {
                Some(auth) => match state.client.get_signer(auth) {
                    Ok(s) => s,
                    Err(e) => {
                        total_failures += 1;
                        batch.push(serde_json::json!({
                            "metric": "http_reqs",
                            "iteration": it,
                            "url": request.url,
                            "status": 0,
                            "duration_ms": 0.0,
                            "error": e.to_string(),
                        }));
                        continue;
                    }
                },
                None => None,
            };
            let start = Instant::now();
            let result = state.client.execute(request, signer_opt.as_deref()).await;
            let elapsed_ms = start.elapsed().as_millis() as f64;
            let (status, ok) = match &result {
                Ok(resp) => (resp.status_code, (200..400).contains(&resp.status_code)),
                Err(_) => (0, false),
            };
            if !ok {
                total_failures += 1;
            }
            let mut sample = serde_json::json!({
                "metric": "http_reqs",
                "iteration": it,
                "url": request.url,
                "status": status,
                "duration_ms": elapsed_ms,
            });
            if let Err(e) = &result {
                sample["error"] = serde_json::Value::String(e.to_string());
            }
            batch.push(sample);
        }
        let chunk = serde_json::json!({
            "iteration": it,
            "samples": batch,
            "failures": total_failures,
        })
        .to_string();
        write_chunk(sock, &chunk).await?;
    }
    // Final verdict chunk + the terminating chunk.
    let verdict = serde_json::json!({
        "done": true,
        "iterations": iterations,
        "failures": total_failures,
        "has_failures": total_failures > 0,
        "thresholds": threshold_verdict(thresholds, iterations, total_failures),
    })
    .to_string();
    write_chunk(sock, &verdict).await?;
    sock.write_all(b"0\r\n\r\n").await.map_err(TropelError::Io)
}

/// Write one HTTP/1.1 chunk: `<hex-size>\r\n<data>\r\n`.
async fn write_chunk(sock: &mut TcpStream, data: &str) -> tropel_sdk::Result<()> {
    let size = format!("{:x}\r\n", data.len());
    sock.write_all(size.as_bytes())
        .await
        .map_err(TropelError::Io)?;
    sock.write_all(data.as_bytes())
        .await
        .map_err(TropelError::Io)?;
    sock.write_all(b"\r\n").await.map_err(TropelError::Io)
}

/// Evaluate the run's thresholds and produce the verdict (TR-411).
fn threshold_verdict(
    thresholds: &std::collections::HashMap<String, String>,
    iterations: u64,
    failures: u64,
) -> serde_json::Value {
    let mut results: Vec<serde_json::Value> = Vec::new();
    let mut all_passed = true;
    for (name, expr) in thresholds {
        let passed = match eval_threshold(expr, iterations, failures) {
            Ok(p) => p,
            Err(e) => {
                all_passed = false;
                results.push(serde_json::json!({
                    "name": name, "expression": expr, "passed": false,
                    "error": e, "actual": null, "threshold": null,
                }));
                continue;
            }
        };
        if !passed {
            all_passed = false;
        }
        let actual = if expr.starts_with("http_req_failed") && iterations > 0 {
            failures as f64 / iterations as f64
        } else {
            iterations as f64
        };
        results.push(serde_json::json!({
            "name": name, "expression": expr, "passed": passed,
            "actual": actual,
            "threshold": expr.split_whitespace().nth(2).and_then(|v| v.parse::<f64>().ok()),
        }));
    }
    results.sort_by(|a, b| a["name"].as_str().cmp(&b["name"].as_str()));
    serde_json::json!({
        "results": results,
        "passed": all_passed,
    })
}

async fn respond(sock: &mut TcpStream, status: u16, body: &str) -> tropel_sdk::Result<()> {
    respond_raw(sock, status, body, None).await
}

/// `respond`, with the connection's CORS headers attached.
///
/// TR-459: a separate name rather than a fourth argument on `respond`,
/// because the argument would sit between the socket and the status at every
/// one of these call sites and read as noise. The CORS value is computed once
/// per connection and is `None` for every non-browser caller, which is all of
/// them today.
async fn respond_raw_cors(
    sock: &mut TcpStream,
    cors: Option<&str>,
    status: u16,
    body: &str,
) -> tropel_sdk::Result<()> {
    respond_raw(sock, status, body, cors).await
}

/// `respond`, plus any extra headers the caller needs on the wire.
///
/// TR-459: the CORS headers have to go on the ACTUAL response, not only on the
/// preflight — a browser that gets a clean preflight and then a reply with no
/// `Access-Control-Allow-Origin` still refuses to hand the body to the page.
async fn respond_raw(
    sock: &mut TcpStream,
    status: u16,
    body: &str,
    extra_headers: Option<&str>,
) -> tropel_sdk::Result<()> {
    let status_text = match status {
        200 => "OK",
        204 => "No Content",
        400 => "Bad Request",
        401 => "Unauthorized",
        403 => "Forbidden",
        404 => "Not Found",
        429 => "Too Many Requests",
        _ => "Error",
    };
    let extra = extra_headers.unwrap_or("");
    let resp = format!(
        "HTTP/1.1 {status} {status_text}\r\nContent-Type: application/json\r\nContent-Length: {}\r\n{extra}Connection: close\r\n\r\n{body}",
        body.len()
    );
    sock.write_all(resp.as_bytes())
        .await
        .map_err(TropelError::Io)
}

/// Execute a single request with full sub-timings — the SAME engine code path
/// a request under load takes. Returns a JSON response payload.
/// One template in a batch resolve (TR-448).
#[derive(serde::Deserialize)]
struct BatchResolveItem {
    template: String,
    #[serde(default)]
    mode: Option<String>,
    #[serde(default)]
    deep: Option<bool>,
}

/// One assertion as the desktop tier sends it.
#[derive(serde::Deserialize)]
struct AgentAssertionSpec {
    #[serde(default)]
    name: Option<String>,
    target: String,
    operator: String,
    #[serde(default)]
    expected: serde_json::Value,
}

/// Read and parse a JSON request body, bounded.
///
/// Returns `Ok(None)` for malformed JSON so the caller answers 400 rather
/// than dropping the connection — a desktop shell debugging its own payload
/// needs the status code, not a closed socket.
async fn read_json_body(
    sock: &mut TcpStream,
    content_length: usize,
    max: usize,
    prefetched: &[u8],
) -> Result<Option<serde_json::Value>, TropelError> {
    Ok(serde_json::from_slice(&read_body(sock, content_length, max, prefetched).await?).ok())
}

/// Read a request body, using whatever already arrived with the head.
///
/// TR-445: the fix for the hang described in `handle_connection`. Reads only
/// the REMAINDER, and returns immediately when the body was fully prefetched.
async fn read_body(
    sock: &mut TcpStream,
    content_length: usize,
    max: usize,
    prefetched: &[u8],
) -> Result<Vec<u8>, TropelError> {
    let want = content_length.min(max);
    let mut body = prefetched.to_vec();
    body.truncate(want);
    if body.len() < want {
        let mut rest = vec![0u8; want - body.len()];
        sock.read_exact(&mut rest).await.map_err(TropelError::Io)?;
        body.extend_from_slice(&rest);
    }
    Ok(body)
}

fn error_body(message: &str) -> String {
    serde_json::json!({ "error": message }).to_string()
}

/// Dispatch a signing request to the ungated `tropel-auth` builders.
///
/// TR-445: this is deliberately ONE endpoint with a `scheme` discriminant
/// rather than four routes. The desktop tier calls it from one place, and a
/// new scheme is a match arm here instead of a new URL its client must learn.
///
/// An unknown scheme is refused BY NAME. Falling back to "no headers" would
/// send the request unsigned while the config says it is authenticated —
/// invariant #7, silent data loss.
fn sign_with_scheme(
    scheme: &str,
    p: &serde_json::Value,
) -> Result<Vec<tropel_auth::builders::HeaderOut>, String> {
    let s = |k: &str| p.get(k).and_then(|v| v.as_str()).unwrap_or("").to_string();
    let opt = |k: &str| {
        p.get(k)
            .and_then(|v| v.as_str())
            .filter(|v| !v.is_empty())
            .map(str::to_string)
    };
    let pairs = |k: &str| -> Vec<(String, String)> {
        p.get(k)
            .and_then(|v| serde_json::from_value(v.clone()).ok())
            .unwrap_or_default()
    };

    match scheme {
        "digest" => {
            let challenge = s("wwwAuthenticate");
            let Some(c) = tropel_auth::builders::find_digest_challenge(&challenge) else {
                // No Digest challenge in the header — the caller must not
                // re-send. Distinct from a signing failure.
                return Err("the WWW-Authenticate header carries no Digest challenge".to_string());
            };
            let get = |k: &str| c.get(k).map(String::as_str);
            Ok(vec![tropel_auth::builders::digest_build_authorization(
                &tropel_auth::builders::DigestBuildParams {
                    username: &s("username"),
                    password: &s("password"),
                    method: &s("method"),
                    uri: &s("uri"),
                    realm: get("realm").unwrap_or(""),
                    nonce: get("nonce").unwrap_or(""),
                    nc: p.get("nc").and_then(|v| v.as_u64()).unwrap_or(1),
                    cnonce: &s("cnonce"),
                    qop: get("qop"),
                    algorithm: get("algorithm"),
                    opaque: get("opaque"),
                },
            )])
        }
        "hawk" => Ok(vec![tropel_auth::builders::hawk_build_header(
            &tropel_auth::builders::HawkBuildParams {
                method: &s("method"),
                resource: &s("resource"),
                host: &s("host"),
                port: p.get("port").and_then(|v| v.as_u64()).unwrap_or(443) as u16,
                id: &s("id"),
                key: &s("key"),
                algorithm: opt("algorithm").as_deref(),
                ts: &s("ts"),
                nonce: &s("nonce"),
                ext: &s("ext"),
            },
        )]),
        "awsSigV4" => {
            let host = s("host");
            let region = opt("region").unwrap_or_else(|| "us-east-1".to_string());
            let service =
                opt("service").unwrap_or_else(|| tropel_auth::builders::default_service(&host));
            let signing_service = tropel_auth::builders::signing_name(&service);
            let path = s("path");
            let canonical_uri = tropel_auth::builders::sigv4_canonical_uri(&path, &service);
            let secret = s("secretKey");
            let date_stamp = s("dateStamp");
            let key = tropel_auth::builders::derive_signing_key(
                &secret,
                &date_stamp,
                &region,
                signing_service,
            );
            let body = match opt("bodyBase64") {
                Some(b64) => Some(
                    base64_decode(&b64).map_err(|e| format!("bodyBase64 is not base64: {e}"))?,
                ),
                None => None,
            };
            let headers = pairs("headers");
            let out = tropel_auth::builders::aws_sigv4_build_headers(
                &tropel_auth::builders::AwsSigV4BuildParams {
                    method: &s("method"),
                    path: &path,
                    query: &s("query"),
                    host: &tropel_auth::builders::bracket_host(&host),
                    headers: &headers,
                    body: body.as_deref(),
                    access_key: &s("accessKey"),
                    secret_key: &secret,
                    session_token: opt("sessionToken").as_deref(),
                    region: &region,
                    service: &service,
                    amz_date: &s("amzDate"),
                    date_stamp: &date_stamp,
                },
                &canonical_uri,
                signing_service,
                &key,
            );
            Ok(out.headers)
        }
        "oauth1" => {
            let base_uri = tropel_auth::builders::oauth1_base_uri(
                &s("scheme"),
                &s("host"),
                p.get("port").and_then(|v| v.as_u64()).map(|v| v as u16),
                &s("path"),
            );
            let mut params = pairs("queryParams");
            if let Some(form) = opt("formBody") {
                params.extend(tropel_auth::builders::parse_form(form.as_bytes()));
            }
            let method = s("signatureMethod");
            tropel_auth::builders::oauth1_build_header(&tropel_auth::builders::OAuth1BuildParams {
                method: &s("method"),
                base_uri: &base_uri,
                request_params: &params,
                consumer_key: &s("consumerKey"),
                consumer_secret: &s("consumerSecret"),
                token: opt("token").as_deref(),
                token_secret: opt("tokenSecret").as_deref(),
                signature_method: &method,
                nonce: &s("nonce"),
                timestamp: &s("timestamp"),
            })
            .map(|o| vec![o.header])
            .ok_or_else(|| {
                format!(
                    "unsupported OAuth1 signature_method '{method}' — supported: {}",
                    tropel_auth::builders::OAUTH1_SIGNATURE_METHODS.join(", ")
                )
            })
        }
        "akamai-edgegrid" => {
            let body = opt("body").map(|b| b.into_bytes());
            let headers_to_sign: Vec<String> = p
                .get("headersToSign")
                .or_else(|| p.get("headers_to_sign"))
                .and_then(|v| serde_json::from_value(v.clone()).ok())
                .unwrap_or_default();
            let params = tropel_auth::edgegrid::EdgeGridBuildParams {
                method: s("method"),
                url: s("url"),
                headers_to_sign,
                body,
                access_token: s("accessToken"),
                client_token: s("clientToken"),
                client_secret: s("clientSecret"),
                // Absent means GENERATE, not empty. A caller pinning either
                // is a test reproducing a vector; a caller omitting both
                // wants a fresh nonce and the host clock, and passing empty
                // strings through would produce a signature Akamai rejects
                // for a stale timestamp.
                nonce: opt("nonce"),
                timestamp: opt("timestamp"),
                max_body: p
                    .get("maxBody")
                    .or_else(|| p.get("max_body"))
                    .and_then(|v| v.as_u64())
                    .map(|n| n as usize)
                    .unwrap_or(tropel_auth::edgegrid::DEFAULT_MAX_BODY),
            };
            // The VALUES of the headers being signed, which the canonical
            // string needs and `headers_to_sign` only names.
            let signed_headers = pairs("headers");
            tropel_auth::edgegrid::edgegrid_build_header(&params, &signed_headers)
                .map(|value| {
                    vec![tropel_auth::builders::HeaderOut {
                        name: "Authorization".to_string(),
                        value,
                    }]
                })
                .map_err(|e| e.to_string())
        }
        "wsse" => {
            let signed = tropel_auth::oauth::sign_wsse(&tropel_auth::oauth::WsseParams {
                username: s("username"),
                password: s("password"),
                nonce: s("nonce"),
                created: s("created"),
            })
            .map_err(|e| e.to_string())?;
            // BOTH headers, which is the WSSE UsernameToken profile's wire
            // shape: the token rides `X-WSSE` and the profile marker rides
            // `Authorization`. Returning only one would serve half the
            // servers that implement the profile, and the caller has no way
            // to know which half.
            Ok(vec![
                tropel_auth::builders::HeaderOut {
                    name: "X-WSSE".to_string(),
                    value: signed.authorization,
                },
                tropel_auth::builders::HeaderOut {
                    name: "Authorization".to_string(),
                    value: "WSSE profile=\"UsernameToken\"".to_string(),
                },
            ])
        }
        other => Err(format!(
            "unknown auth scheme '{other}' — supported: {}",
            AUTH_SIGN_SCHEMES.join(", ")
        )),
    }
}

/// The schemes `sign_with_scheme` answers for, in one place.
///
/// It used to be a hardcoded string in the error arm, and it drifted the
/// moment a scheme was added — the message still read "digest, hawk,
/// awsSigV4, oauth1" while `build_auth_signer` had grown EdgeGrid and WSSE.
/// A caller reading that message would conclude the agent could not sign
/// something it could, so the list is derived from one declaration and
/// asserted against the dispatcher below.
pub const AUTH_SIGN_SCHEMES: &[&str] = &[
    "digest",
    "hawk",
    "awsSigV4",
    "oauth1",
    "akamai-edgegrid",
    "wsse",
];

/// Base64 encode, beside the decoder — the same engine, so a round trip
/// through the agent cannot disagree with itself.
fn base64_encode(bytes: &[u8]) -> String {
    use base64::Engine as _;
    base64::engine::general_purpose::STANDARD.encode(bytes)
}

/// Base64 decode without pulling a new dependency into this crate.
fn base64_decode(s: &str) -> Result<Vec<u8>, String> {
    use base64::Engine as _;
    base64::engine::general_purpose::STANDARD
        .decode(s)
        .map_err(|e| e.to_string())
}

/// Run a script in a fresh realm and report what it did.
///
/// TR-446, completing KT-203's `run_script`. This is the entry point the
/// desktop tier and the third differential leg both need: the SAME realm a
/// load run uses (deep-equal + k6-core + the `pm`/`trp` shims), driven once,
/// with its effects returned as data.
///
/// A FRESH realm per call is deliberate. Scripts mutate globals, and a shared
/// context would let one request's script change the next one's behaviour —
/// a bug that reproduces only under a specific ordering, which is the worst
/// kind to chase. The agent is a per-request ABI, not a session.
/// The four variable scopes a script reads and writes.
///
/// TR-473: `/script` carried only `environment`. The realm has always had the
/// other three — `pm.collectionVariables`, `pm.globals` and `pm.variables` are
/// separate stores with a defined precedence (local > data > env > collection)
/// — so a script reading any of them saw an empty scope, and anything it wrote
/// to them was discarded on the way back. Both directions silently, which is
/// the shape invariant 7 forbids.
///
/// A struct rather than four more positional parameters: the call already took
/// six, and `environment`/`collection`/`globals`/`variables` are four maps that
/// would be trivially transposable at a call site.
#[derive(Default)]
struct ScriptScopes {
    environment: HashMap<String, String>,
    collection: HashMap<String, serde_json::Value>,
    globals: HashMap<String, serde_json::Value>,
    variables: HashMap<String, serde_json::Value>,
}

impl ScriptScopes {
    /// Read the scopes off a `/script` payload. Every one is optional — a
    /// caller that sends none keeps the previous behaviour exactly.
    fn from_payload(payload: &serde_json::Value) -> Self {
        fn map_of(payload: &serde_json::Value, key: &str) -> HashMap<String, serde_json::Value> {
            payload
                .get(key)
                .and_then(|v| v.as_object())
                .map(|o| o.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
                .unwrap_or_default()
        }
        Self {
            environment: payload
                .get("environment")
                .and_then(|e| serde_json::from_value(e.clone()).ok())
                .unwrap_or_default(),
            collection: map_of(payload, "collectionVariables"),
            globals: map_of(payload, "globals"),
            variables: map_of(payload, "variables"),
        }
    }
}

/// What the HOST lends the realm for one script run.
///
/// A struct because the alternative was eight positional parameters, three of
/// them `Option<Arc<…>>` sitting next to each other — the exact shape where a
/// call site transposes two and the compiler is happy.
struct ScriptHost {
    http: tropel_http::HttpClient,
    /// TR-474: the bidirectional channel. `None` = no `bru.runRequest`.
    callbacks: Option<Arc<RunCallbacks>>,
    /// TR-476: the per-script cookie jar. `None` = no `bru.cookies`.
    cookies: Option<Arc<ScriptCookies>>,
}

async fn run_script_once(
    code: &str,
    scopes: ScriptScopes,
    request: Option<TropelRequest>,
    response: Option<tropel_sdk::types::Response>,
    sandbox: tropel_sandbox::config::SandboxConfig,
    host: ScriptHost,
) -> Result<serde_json::Value, String> {
    let ScriptHost {
        http,
        callbacks,
        cookies,
    } = host;
    // Captured BEFORE `request` is moved into the realm state below. The
    // cookie bindings need the executing URL to scope reads and writes.
    let request_url = request.as_ref().map(|r| r.url.clone()).unwrap_or_default();
    let mut ctx = tropel_js::JsContext::new(None, Some(std::time::Duration::from_secs(10)))
        .await
        .map_err(|e| format!("js context: {e:?}"))?;

    // TR-469: BEFORE the bundle — pm.js's install tail reads this global to
    // decide the canonical binding name and its aliases, so a preamble
    // evaluated afterwards would be read too late and silently leave the
    // stock `trp` install in place.
    ctx.eval(&sandbox.render_js_preamble())
        .await
        .map_err(|e| format!("sandbox config preamble: {e:?}"))?;

    ctx.eval(include_str!("../js/shared/deep-equal.js"))
        .await
        .map_err(|e| format!("deep-equal shim: {e:?}"))?;
    // TR-465: the SHARED bundle, not a hand-rolled list.
    //
    // This used to eval `k6-core.js` + `pm.js` inline, which quietly gave
    // /script a SMALLER surface than a load run: `typeof bru === "undefined"`
    // here and an object there, so a Bruno-style script worked in the app and
    // failed on the agent. Measured, not guessed — a realm probe across both
    // engines is what surfaced it.
    //
    // `js_bootstrap` exists precisely because this went wrong once before: two
    // hand-maintained shim lists drifted and "bru.js was compiled into the
    // binary but NEVER evaluated". Re-deriving the list here re-opened that
    // exact hole, one endpoint over.
    for entry in crate::js_bootstrap::ShimBundle::default().0 {
        ctx.eval(&entry.1)
            .await
            .map_err(|e| format!("{} shim: {e:?}", entry.0))?;
    }

    let state = tropel_sandbox::state::SharedPmState::default();
    {
        // A poisoned lock here means a previous script panicked mid-mutation.
        // Recovering the guard is right: the agent is per-request, the state
        // is fresh, and refusing would strand the caller on someone else's
        // panic.
        let mut st = state.lock().unwrap_or_else(|e| e.into_inner());
        st.environment = scopes.environment;
        // TR-473: the other three stores. `collection_vars`/`globals` are
        // Arc'd (they are cloned per build_scope), `local_vars` is not.
        st.collection_vars = std::sync::Arc::new(scopes.collection);
        st.globals = std::sync::Arc::new(scopes.globals);
        st.local_vars = scopes.variables;
        // TR-467: seed the REQUEST the script is about to mutate.
        //
        // Without it `pm.request.headers.add(...)` had nothing to write to, so
        // a pre-request script ran, appeared to succeed, and its header never
        // reached the wire — the silent no-op invariant #7 forbids. The state
        // has always carried the field; /script simply never filled it.
        st.request = request;
        // TR-468: and the RESPONSE a test script asserts against. Without it
        // `pm.response.code` was undefined and every `pm.test` that looked at
        // the response silently failed on a response that was never there —
        // the test stage cannot run on this realm at all until it is seeded.
        st.response = response;
    }
    // TR-472: the bridge gets the agent's HTTP client, so `pm.sendRequest`
    // actually sends.
    //
    // `TrpBridge::new` leaves `http_client: None`, and the send-request
    // binding is still INSTALLED in that state — so `typeof pm.sendRequest`
    // was "function" and calling it handed the script
    // `Error: pm.sendRequest unavailable in this build (no HTTP client)`.
    // Declared, present, and non-functional (invariant 4); at least it named
    // itself rather than failing silently.
    //
    // Through `new_arc` rather than a plain `Arc::new`, which publishes this
    // client's cookie jar against it (TR-233) — a plain Arc leaves the jar
    // unreachable and the cookie surface degrades to a no-op shim.
    // TR-477: everything the script sends is recorded here, at the ONE place
    // both `pm.sendRequest` and `fetch` end up.
    let sends = Arc::new(ScriptSendLog::default());
    let vu_client = tropel_http::VuCookieClient::new(http);
    let base_client: std::sync::Arc<dyn tropel_sdk::traits::DriverHttpClient> =
        crate::vu_loop::DriverHttpClientImpl::new_arc(vu_client);
    let driver_client: std::sync::Arc<dyn tropel_sdk::traits::DriverHttpClient> =
        Arc::new(RecordingHttpClient {
            inner: base_client,
            log: sends.clone(),
        });
    tropel_sandbox::bindings::trp::TrpBridge::with_http_client(state.clone(), driver_client)
        .install(&mut ctx)
        .map_err(|e| format!("bridge install: {e:?}"))?;

    // TR-476: the cookie jar, installed ONLY when the caller supplied one.
    // Absent, the bindings do not exist and `bru.cookies` refuses by name —
    // "no jar" and "an empty jar" must not look the same to a script.
    if let Some(cookies) = cookies.clone() {
        let url_for_current = request_url.clone();
        let installed: Result<(), String> = ctx.with_ctx(|rq| {
            let globals = rq.globals();
            let mut fail: Option<String> = None;

            let u = url_for_current.clone();
            if let Err(e) = globals.set(
                "__tropel_cookies_current_url",
                rquickjs::function::Func::from(move || -> String { u.clone() }),
            ) {
                fail = Some(e.to_string());
            }

            let c = cookies.clone();
            if let Err(e) = globals.set(
                "__tropel_cookies_all",
                rquickjs::function::Func::from(move |url: String| -> String {
                    let jar = c.lock_jar();
                    let matching: Vec<&serde_json::Value> = jar
                        .iter()
                        .filter(|ck| cookie_matches_url(ck, &url))
                        .collect();
                    serde_json::to_string(&matching).unwrap_or_else(|_| "[]".to_string())
                }),
            ) {
                fail = Some(e.to_string());
            }

            let c = cookies.clone();
            if let Err(e) = globals.set(
                "__tropel_cookies_set",
                rquickjs::function::Func::from(move |url: String, cookie_json: String| {
                    let Ok(cookie) = serde_json::from_str::<serde_json::Value>(&cookie_json) else {
                        return;
                    };
                    let name = cookie
                        .get("key")
                        .or_else(|| cookie.get("name"))
                        .and_then(|v| v.as_str())
                        .unwrap_or("")
                        .to_string();
                    {
                        // Read-your-writes: the script must see what it just
                        // set, not the seeded snapshot.
                        let mut jar = c.lock_jar();
                        jar.retain(|ck| {
                            ck.get("key").and_then(|v| v.as_str()).unwrap_or("") != name
                        });
                        jar.push(cookie.clone());
                    }
                    c.record(serde_json::json!({ "op": "set", "url": url, "cookie": cookie }));
                }),
            ) {
                fail = Some(e.to_string());
            }

            let c = cookies.clone();
            if let Err(e) = globals.set(
                "__tropel_cookies_delete",
                rquickjs::function::Func::from(move |url: String, name: String| {
                    {
                        let mut jar = c.lock_jar();
                        jar.retain(|ck| {
                            ck.get("key").and_then(|v| v.as_str()).unwrap_or("") != name
                        });
                    }
                    c.record(serde_json::json!({ "op": "delete", "url": url, "name": name }));
                }),
            ) {
                fail = Some(e.to_string());
            }

            let c = cookies.clone();
            if let Err(e) = globals.set(
                "__tropel_cookies_clear",
                rquickjs::function::Func::from(move |url: String| {
                    c.lock_jar().clear();
                    c.record(serde_json::json!({ "op": "clear", "url": url }));
                }),
            ) {
                fail = Some(e.to_string());
            }

            match fail {
                Some(e) => Err(e),
                None => Ok(()),
            }
        });
        installed.map_err(|e| format!("cookie bridge: {e}"))?;
    }

    // TR-474: `bru.runRequest`'s host half, installed ONLY when the caller
    // opted in with a `runId`. Absent, `__tropel_trp_run_request` stays
    // undefined and the shim refuses by name — the agent never pretends to
    // offer a callback nobody is listening for (invariant 4).
    if let Some(cb) = callbacks.clone() {
        let sends_in_cb = sends.clone();
        let installed: Result<(), String> = ctx.with_ctx(|rq| {
            rq.globals()
                .set(
                    "__tropel_trp_run_request",
                    rquickjs::function::Func::from(move |path: String| -> String {
                        let sends_for_run = sends_in_cb.clone();
                        let requested_path = path.clone();
                        // SYNCHRONOUS on purpose: QuickJS gives a host
                        // function no way to suspend, so the realm's thread
                        // parks here until the caller answers. The reply
                        // arrives on a DIFFERENT connection, hence a
                        // different tokio task — blocking this one does not
                        // stop it being served.
                        let call_id = cb
                            .next_id
                            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                        let (tx, rx) = std::sync::mpsc::channel::<String>();
                        {
                            let mut replies = match cb.replies.lock() {
                                Ok(g) => g,
                                Err(e) => e.into_inner(),
                            };
                            replies.insert(call_id, tx);
                        }
                        {
                            let mut pending = match cb.pending.lock() {
                                Ok(g) => g,
                                Err(e) => e.into_inner(),
                            };
                            pending.push_back(PendingHostCall {
                                call_id,
                                kind: "runRequest",
                                path,
                            });
                        }
                        cb.ready.notify_waiters();

                        // `block_in_place`, not a bare blocking recv.
                        //
                        // QuickJS gives a host function no way to suspend, so
                        // this thread MUST block. Blocking it silently would
                        // hold a tokio worker for the whole round trip — and
                        // the answer arrives on another connection, i.e.
                        // another task, which then needs a worker to be
                        // served. Enough concurrent scripts and every worker
                        // is parked waiting for replies that cannot be
                        // delivered. `block_in_place` tells the runtime to
                        // move the rest of this worker's queue elsewhere
                        // first, so the reply path always has somewhere to run.
                        //
                        // It requires a MULTI-THREADED runtime: it panics on a
                        // current-thread one. The agent builds a multi-thread
                        // runtime (4 workers by default), and the tests below
                        // ask for one explicitly.
                        let started = std::time::Instant::now();
                        match tokio::task::block_in_place(|| {
                            rx.recv_timeout(std::time::Duration::from_secs(30))
                        }) {
                            Ok(json) => {
                                // TR-477: a runRequest is a script-issued send
                                // too — the HOST performed it, but the SCRIPT
                                // caused it, and the timeline shows it either
                                // way. Recorded here rather than at the client
                                // because it never touches the agent's client.
                                let parsed: serde_json::Value =
                                    serde_json::from_str(&json).unwrap_or(serde_json::Value::Null);
                                sends_for_run.record(serde_json::json!({
                                    "source": "runRequest",
                                    "kind": "collection",
                                    "requestName": requested_path,
                                    "method": parsed.get("method")
                                        .and_then(|v| v.as_str()).unwrap_or(""),
                                    "url": parsed.get("url")
                                        .and_then(|v| v.as_str()).unwrap_or(""),
                                    "status": parsed.get("status")
                                        .and_then(|v| v.as_u64()).unwrap_or(0),
                                    "responseTime": started.elapsed().as_millis() as u64,
                                    "error": parsed.get("error").cloned(),
                                }));
                                json
                            }
                            Err(_) => {
                                // Drop the slot so a late reply cannot land on
                                // a call nobody is waiting for any more.
                                if let Ok(mut r) = cb.replies.lock() {
                                    r.remove(&call_id);
                                }
                                // A REFUSAL, not a fabricated response: the
                                // script must not read a timeout as a request
                                // that ran and returned nothing (invariant 8).
                                r#"{"error":"runRequest timed out: the host did not answer within 30s"}"#
                                    .to_string()
                            }
                        }
                    }),
                )
                .map_err(|e| e.to_string())
        });
        installed.map_err(|e| format!("run_request bridge: {e}"))?;
    }

    // A THROWING script is not an agent error — it is a result. The caller
    // needs the message and whatever ran before the throw, exactly as the
    // in-app runner reports it; a 500 here would lose both.
    //
    // The throw is caught in JS rather than in Rust because the Rust side
    // only sees "Exception generated by QuickJS" — the actual message lives
    // on the JS exception object, and losing it leaves a user staring at a
    // failed script with no reason.
    //
    // MESSAGE FIRST, then the stack. QuickJS's `e.stack` carries only the
    // frames, so preferring it (the obvious `e.stack || e.message`) drops the
    // one line a user actually reads. A `try`/`catch` around the body keeps the
    // message; the `await` wrapper keeps top-level `await` working, which a
    // bare try/catch would break.
    let wrapped = format!(
        "globalThis.__tropel_script_error = null;\n\
         (async function () {{ try {{\n{code}\n}} catch (e) {{ \
           globalThis.__tropel_script_error = \
           (e && e.message ? e.message : String(e)) + \
           (e && e.stack ? \"\\n\" + e.stack : \"\"); \
         }} }})();"
    );
    let script_error = match ctx.eval(&wrapped).await {
        Ok(_) => {
            let raw = ctx
                .eval("globalThis.__tropel_script_error === null ? \"\" : String(globalThis.__tropel_script_error)")
                .await
                .unwrap_or_default();
            if raw.is_empty() {
                None
            } else {
                Some(raw)
            }
        }
        // A failure of the WRAPPER itself (a syntax error in the user's code,
        // which `try` cannot catch) still has to be reported.
        Err(e) => Some(format!("{e:?}")),
    };

    let st = state.lock().unwrap_or_else(|e| e.into_inner());
    // Individual checks are reconstructed from the `checks` samples — that is
    // where `record_test_tagged` puts them, tagged with the raw check name.
    let tests: Vec<serde_json::Value> = st
        .samples
        .iter()
        .filter(|s| s.metric == "checks")
        .map(|s| {
            serde_json::json!({
                "name": s.tags.get("check").unwrap_or_default(),
                "passed": s.value != 0.0,
            })
        })
        .collect();

    Ok(serde_json::json!({
        "tests": tests,
        "assertions": {
            "total": st.assertions.total,
            "passed": st.assertions.passed,
            "failed": st.assertions.failed,
        },
        // The MUTATIONS: what the script left behind. The caller merges these
        // into its own scope — the agent holds no session state.
        "environment": st.environment,
        // TR-473: the scopes AS THE SCRIPT LEFT THEM. Returned even when the
        // caller sent none, so a `pm.globals.set` in a script that seeded only
        // `environment` still reaches the caller rather than dying with the
        // realm.
        "collectionVariables": *st.collection_vars,
        "globals": *st.globals,
        "variables": st.local_vars,
        // TR-467: the request AS THE SCRIPT LEFT IT. The bridges mutate
        // `st.request` in place, so every `pm.request.headers.add/upsert/
        // remove`, URL change and body change is already recorded here — it
        // was simply never returned, which made every one of them a no-op
        // from the caller's point of view.
        "request": st.request,
        // TR-476: what the script DID to the jar, in order. The caller's jar
        // stays authoritative — it replays these rather than being replaced,
        // so a cookie the agent never saw is not lost.
        // TR-477: what the SCRIPT sent, in call order. Absent sends are an
        // empty list, never null — "the script sent nothing" is an answer,
        // and it must not read the same as "this tier does not record".
        "scriptSends": sends.drain(),
        "cookieOps": cookies
            .as_ref()
            .map(|c| serde_json::Value::Array(c.lock_ops().clone()))
            .unwrap_or(serde_json::Value::Null),
        "scriptError": script_error,
    }))
}

/// Dispatch an OAuth2/JWT/WSSE operation to the ungated `tropel-auth::oauth`.
///
/// TR-447, the last of KT-203's auth gap. Like `/auth/sign`, this is ONE
/// endpoint with an `op` discriminant rather than nine routes: the desktop
/// tier calls it from one place, and a new operation is a match arm instead
/// of a new URL its client has to learn.
///
/// Every arm is a straight call into the same functions the wasm tier
/// exports. No wrapper logic — a second implementation of PKCE, token
/// building or JWT signing is the invariant #3 failure this endpoint exists
/// to prevent, and D4 names signing specifically ("a signing byte-difference
/// is a 403 that takes a day to find").
/// Serialise a builder's output for the wire.
///
/// A free generic fn rather than a closure: `impl Trait` is not allowed in
/// closure parameters, and the alternative — a `dyn Serialize` box per call —
/// would allocate on a path that runs per request.
fn as_json<T: serde::Serialize>(v: &T) -> Result<serde_json::Value, String> {
    serde_json::to_value(v).map_err(|e| e.to_string())
}

fn oauth2_dispatch(op: &str, p: &serde_json::Value) -> Result<serde_json::Value, String> {
    use tropel_auth::oauth;
    let as_str = |k: &str| p.get(k).and_then(|v| v.as_str()).unwrap_or("");
    let opt = |k: &str| {
        p.get(k)
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
            .map(str::to_string)
    };

    match op {
        "buildAuthorizeUrl" => {
            let params: oauth::AuthorizeParams =
                serde_json::from_value(p.clone()).map_err(|e| e.to_string())?;
            as_json(&oauth::build_authorize_url(&params).map_err(|e| e.to_string())?)
        }
        "buildTokenRequest" => {
            let params: oauth::TokenRequestParams =
                serde_json::from_value(p.clone()).map_err(|e| e.to_string())?;
            as_json(&oauth::build_token_request(&params).map_err(|e| e.to_string())?)
        }
        "parseTokenResponse" => {
            as_json(&oauth::parse_token_response(as_str("body")).map_err(|e| e.to_string())?)
        }
        "attachToken" => {
            // The placement vocabulary is the Rust's, not a string the caller
            // invents — an unknown placement must be refused, not defaulted
            // to header, or a token silently stops reaching a query-auth API.
            let placement = match p.get("placement").and_then(|v| v.as_str()) {
                Some("header") | None => oauth::TokenPlacement::Header,
                Some("query") => oauth::TokenPlacement::Query,
                Some(other) => {
                    return Err(format!(
                        "unknown token placement '{other}' — expected header or query"
                    ))
                }
            };
            as_json(&oauth::attach_token(
                as_str("token"),
                opt("tokenType").as_deref(),
                placement,
                opt("headerPrefix").as_deref(),
                opt("queryKey").as_deref(),
            ))
        }
        "decodeJwt" => as_json(&oauth::decode_jwt(as_str("token")).map_err(|e| e.to_string())?),
        "jwtExpiresAt" => {
            let exp = oauth::jwt_expires_at(as_str("token")).map_err(|e| e.to_string())?;
            Ok(serde_json::json!({ "expiresAt": exp }))
        }
        "signJwt" => {
            let algorithm = match p.get("algorithm").and_then(|v| v.as_str()) {
                Some("HS256") | None => oauth::JwtAlgorithm::Hs256,
                Some("HS384") => oauth::JwtAlgorithm::Hs384,
                Some("HS512") => oauth::JwtAlgorithm::Hs512,
                // Never downgrade to HS256: the config would say one thing
                // and the wire another (the TR-004/TR-409 shape).
                Some(other) => {
                    return Err(format!(
                        "unsupported JWT algorithm '{other}' — supported: HS256, HS384, HS512"
                    ))
                }
            };
            let payload = p.get("payload").cloned().unwrap_or_default();
            let header = p.get("header").cloned().filter(|h| !h.is_null());
            let token = oauth::sign_jwt(header.as_ref(), &payload, algorithm, as_str("secret"))
                .map_err(|e| e.to_string())?;
            Ok(serde_json::json!({ "token": token }))
        }
        "wsseSign" => {
            let params: oauth::WsseParams =
                serde_json::from_value(p.clone()).map_err(|e| e.to_string())?;
            as_json(&oauth::sign_wsse(&params).map_err(|e| e.to_string())?)
        }
        "codeChallengeS256" => Ok(serde_json::json!({
            "codeChallenge": oauth::code_challenge_s256(as_str("verifier")),
            "codeChallengeMethod": "S256",
        })),
        other => Err(format!(
            "unknown oauth2 op '{other}' — supported: buildAuthorizeUrl, buildTokenRequest, \
             parseTokenResponse, attachToken, decodeJwt, jwtExpiresAt, signJwt, wsseSign, \
             codeChallengeS256"
        )),
    }
}

async fn execute_single(state: &AgentState, req: &serde_json::Value) -> serde_json::Value {
    let method = req.get("method").and_then(|m| m.as_str()).unwrap_or("GET");
    let url = req.get("url").and_then(|u| u.as_str()).unwrap_or("");
    let follow = req
        .get("follow_redirects")
        .and_then(|f| f.as_bool())
        .unwrap_or(true);

    // TR-463: headers arrive as an ARRAY of pairs, with the old object form
    // still accepted.
    //
    // A JSON object cannot hold two entries with the same key, so the object
    // form silently collapsed duplicate header names — one of two `Set-Cookie`
    // or `Accept` rows reached the wire and nothing reported the other. That
    // is the same silent-loss class as TR-462's mangled bodies, refusing to
    // send data the caller asked for rather than corrupting what came back.
    //
    // The relay has always used a pair list for exactly this reason
    // (`duplicateNames: true` in its capability descriptor); /execute now
    // matches, so a transport built on it can declare the same.
    let headers: Vec<(String, String)> = match req.get("headers") {
        Some(serde_json::Value::Array(rows)) => rows
            .iter()
            .filter_map(|row| match row {
                // ["Name", "value"]
                serde_json::Value::Array(pair) if pair.len() == 2 => Some((
                    pair[0].as_str()?.to_string(),
                    pair[1].as_str().unwrap_or("").to_string(),
                )),
                // {"name": "...", "value": "..."}
                serde_json::Value::Object(o) => Some((
                    o.get("name")?.as_str()?.to_string(),
                    o.get("value")
                        .and_then(|v| v.as_str())
                        .unwrap_or("")
                        .to_string(),
                )),
                _ => None,
            })
            .collect(),
        Some(serde_json::Value::Object(o)) => o
            .iter()
            .map(|(k, v)| (k.clone(), v.as_str().unwrap_or("").to_string()))
            .collect(),
        _ => Vec::new(),
    };

    // TR-463: the fields the engine has always supported and the wire format
    // dropped on the floor. Each was hard-coded to its empty value, so a
    // client asking for a client certificate, a Host override, a cookie or a
    // timeout was ignored WITHOUT being told — the request went out missing
    // what it asked for. `tropel_sdk::Request` carries every one of them.
    let certificate: Option<tropel_sdk::types::CertificateConfig> = req
        .get("certificate")
        .and_then(|v| serde_json::from_value(v.clone()).ok());
    // Ask 17's per-request half, read off the wire like `certificate` above.
    // KnockPort declares every `proxy.*` capability false with "the agent's
    // client is built once from HttpConfig::default() and /execute carries no
    // proxy field" as the stated reason — this is that field. Additive, so a
    // client that sends no `proxy` is unaffected.
    //
    // `.ok()` rather than a hard error, matching `certificate`: a malformed
    // block is the same class of caller mistake and the request is refused
    // downstream by name (`fixed` with no host, a bypass typo) rather than
    // failing to parse with a serde message about a field the caller cannot
    // see.
    let proxy: Option<tropel_sdk::types::ProxyConfig> = req
        .get("proxy")
        .and_then(|v| serde_json::from_value(v.clone()).ok());
    let host: Option<String> = req.get("host").and_then(|h| h.as_str()).map(str::to_string);
    let cookies: Vec<tropel_sdk::types::RequestCookie> = req
        .get("cookies")
        .and_then(|v| serde_json::from_value(v.clone()).ok())
        .unwrap_or_default();
    let timeout = req
        .get("timeout_ms")
        .and_then(|t| t.as_u64())
        .map(std::time::Duration::from_millis);
    let query_params: HashMap<String, String> = req
        .get("query_params")
        .and_then(|v| serde_json::from_value(v.clone()).ok())
        .unwrap_or_default();

    let method_parsed = Method::parse(method).unwrap_or(Method::GET);
    // TR-409: parse optional `auth` field (`AuthConfig` JSON) so the single
    // request path (`POST /execute`) and the load path (`POST /run`) share the
    // same signer builder. Unsupported schemes are reported, not degraded.
    let auth: Option<tropel_sdk::types::AuthConfig> = req
        .get("auth")
        .and_then(|v| serde_json::from_value(v.clone()).ok());
    let request = TropelRequest {
        url: url.to_string(),
        method: method_parsed,
        headers,
        query_params,
        body: req
            .get("body")
            .and_then(|b| b.as_str())
            .map(|s| Body::Raw(s.to_string())),
        auth: auth.clone(),
        certificate,
        proxy,
        follow_redirects: follow,
        host,
        cookies,
        timeout,
        response_type: ResponseType::Text,
    };

    // TR-409: surface unsupported auth as a transport error rather than
    // sending the request without an Authorization header (the TR-004 shape).
    let signer_opt = match &request.auth {
        Some(a) => match state.client.get_signer(a) {
            Ok(s) => s,
            Err(e) => {
                return serde_json::json!({
                    "status": 0,
                    "status_text": "Unsupported Auth",
                    "headers": {},
                    "body": "",
                    "timings": {
                        "blocked": 0.0, "dns": 0.0, "connecting": 0.0, "tls_handshaking": 0.0,
                        "sending": 0.0, "waiting": 0.0, "receiving": 0.0, "duration": 0.0,
                    },
                    "error": e.to_string(),
                });
            }
        },
        None => None,
    };

    let start = Instant::now();
    let result = state.client.execute(&request, signer_opt.as_deref()).await;
    let elapsed_ms = start.elapsed().as_millis() as f64;

    match result {
        Ok(resp) => {
            let waiting = resp
                .timings
                .as_ref()
                .map(|t| t.waiting.as_millis() as f64)
                .unwrap_or(0.0);
            let receiving = resp
                .timings
                .as_ref()
                .map(|t| t.receiving.as_millis() as f64)
                .unwrap_or(0.0);
            // TR-462: a response body that is not valid UTF-8 comes back as
            // base64, and says so.
            //
            // This used to be `String::from_utf8_lossy`, which replaces every
            // invalid byte with U+FFFD and reports nothing. A PNG, a protobuf
            // or a gzip payload fetched through the agent arrived CORRUPTED,
            // and no field on the reply said so — the silent data loss
            // invariant #7 forbids, on the single-request path a desktop or
            // website transport uses for every send.
            //
            // `bodyEncoding` is always present so a caller never has to guess,
            // and a caller that ignores it now sees obvious base64 rather than
            // subtle mojibake — a failure that is visible instead of one that
            // looks like a server bug.
            let (body, body_encoding) = match std::str::from_utf8(&resp.body) {
                Ok(text) => (text.to_string(), "utf8"),
                Err(_) => (base64_encode(&resp.body), "base64"),
            };
            serde_json::json!({
                "status": resp.status_code,
                "status_text": resp.status_text,
                "headers": resp.headers,
                "body": body,
                "bodyEncoding": body_encoding,
                "timings": {
                    "blocked": 0.0, "dns": 0.0, "connecting": 0.0, "tls_handshaking": 0.0,
                    "sending": 0.0, "waiting": waiting, "receiving": receiving,
                    "duration": elapsed_ms,
                },
                "error": null,
            })
        }
        Err(e) => {
            serde_json::json!({
                "status": 0,
                "status_text": "Transport Error",
                "headers": {},
                "body": "",
                "timings": {
                    "blocked": 0.0, "dns": 0.0, "connecting": 0.0, "tls_handshaking": 0.0,
                    "sending": 0.0, "waiting": 0.0, "receiving": 0.0, "duration": elapsed_ms,
                },
                "error": e.to_string(),
            })
        }
    }
}

/// A rolling-window rate limiter per connection.
struct RateLimiter {
    window_start: Instant,
    count: u64,
}

impl RateLimiter {
    fn new() -> Self {
        Self {
            window_start: Instant::now(),
            count: 0,
        }
    }

    fn allow(&mut self) -> Result<(), ()> {
        if self.window_start.elapsed().as_secs() >= 1 {
            self.window_start = Instant::now();
            self.count = 0;
        }
        self.count += 1;
        if self.count > RATE_LIMIT_PER_SEC {
            return Err(());
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn refuses_non_loopback_bind_is_enforced() {
        // The loopback check must reject a public address — the execution
        // endpoint must never be reachable off-box.
        let ip: IpAddr = "0.0.0.0".parse().unwrap();
        assert!(!ip.is_loopback(), "0.0.0.0 must be rejected");
        let ip2: IpAddr = "127.0.0.1".parse().unwrap();
        assert!(ip2.is_loopback(), "127.0.0.1 must be accepted");
    }

    #[test]
    fn threshold_verdict_evaluates_and_verdicts() {
        // TR-411: http_reqs < N and http_req_failed <= rate thresholds.
        use std::collections::HashMap;
        let mut t = HashMap::new();
        t.insert("reqs_ok".into(), "http_reqs < 100".into());
        t.insert("fail_rate_ok".into(), "http_req_failed <= 0.1".into());

        // 10 iterations, 1 failure → both pass.
        let v = threshold_verdict(&t, 10, 1);
        assert!(v["passed"].as_bool().unwrap(), "all must pass: {v}");
        let results = v["results"].as_array().unwrap();
        assert_eq!(results.len(), 2);
        assert!(results.iter().all(|r| r["passed"].as_bool().unwrap()));

        // 200 iterations, 1 failure → http_reqs < 100 fails.
        let v2 = threshold_verdict(&t, 200, 1);
        assert!(!v2["passed"].as_bool().unwrap(), "reqs threshold must fail");
        let reqs = v2["results"]
            .as_array()
            .unwrap()
            .iter()
            .find(|r| r["name"] == "reqs_ok")
            .unwrap();
        assert!(!reqs["passed"].as_bool().unwrap());

        // A malformed threshold reports an error, not a silent pass.
        let mut bad = HashMap::new();
        bad.insert("bogus".into(), "http_reqs".into());
        let v3 = threshold_verdict(&bad, 10, 0);
        assert!(!v3["passed"].as_bool().unwrap(), "malformed must fail");
        assert!(
            v3["results"][0]["error"].as_str().is_some(),
            "the malformed threshold must report its error"
        );
    }
    /// TR-445: the rules endpoints, driven over a REAL loopback socket.
    ///
    /// Unit-testing the handler functions would not prove the thing that
    /// matters — knockport's desktop tier reaches these through HTTP, and the
    /// bug class this closes (every core-tier method throwing
    /// `TropelCoreUnavailableError`) is about the WIRE contract, not the Rust.
    #[tokio::test]
    async fn the_rules_endpoints_answer_over_the_socket() {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let port = listener.local_addr().expect("addr").port();
        let state = Arc::new(AgentState {
            token: None,
            client: tropel_http::HttpClient::new(&tropel_http::config::HttpConfig::default())
                .expect("http client"),
            runs: std::sync::Mutex::new(HashMap::new()),
            // No browser origin: these drive the socket directly, and an
            // empty allowlist is the default a real agent starts with.
            allowed_origins: vec![],
        });
        tokio::spawn(async move {
            while let Ok((mut sock, _)) = listener.accept().await {
                let st = state.clone();
                tokio::spawn(async move {
                    let _ = handle_connection(&mut sock, st).await;
                });
            }
        });

        let post = |path: &'static str, body: String| async move {
            let mut s = TcpStream::connect(("127.0.0.1", port))
                .await
                .expect("connect");
            let req = format!(
                "POST {path} HTTP/1.1\r\nHost: localhost\r\nContent-Length: {}\r\n\r\n{body}",
                body.len()
            );
            s.write_all(req.as_bytes()).await.expect("write");
            let mut out = Vec::new();
            s.read_to_end(&mut out).await.expect("read");
            String::from_utf8_lossy(&out).to_string()
        };

        // ── /resolve ──
        let raw = post(
            "/resolve",
            serde_json::json!({
                "template": "{{base}}/v1", "variables": {"base": "https://x.test"},
                "mode": "plain", "deep": true
            })
            .to_string(),
        )
        .await;
        assert!(raw.contains("https://x.test/v1"), "{raw}");

        // A typo'd mode is a NAMED 400, never a silent fallback to plain —
        // that is how a quote-bearing value corrupts a JSON body.
        let raw = post(
            "/resolve",
            serde_json::json!({"template": "x", "variables": {}, "mode": "jsonn"}).to_string(),
        )
        .await;
        assert!(raw.starts_with("HTTP/1.1 400"), "{raw}");
        assert!(raw.contains("unknown mode"), "{raw}");

        // ── /assert ──
        let raw = post(
            "/assert",
            serde_json::json!({
                "response": {
                    "status": 200, "status_text": "OK",
                    "headers": [["Content-Type", "application/json"]],
                    "body": "{\"count\":2}", "response_time": 5.0, "size": 12,
                    "cookies": []
                },
                "assertions": [
                    {"name": "ok", "target": "status", "operator": "eq", "expected": 200},
                    {"target": "json.count", "operator": "eq", "expected": 99}
                ]
            })
            .to_string(),
        )
        .await;
        assert!(raw.contains(r#""name":"ok""#), "{raw}");
        assert!(raw.contains(r#""passed":true"#), "{raw}");
        // A FAILING row explains itself, and names the TARGET not the row name.
        assert!(
            raw.contains("expected target json.count equals 99"),
            "{raw}"
        );

        // `matches` is UNSUPPORTED here on purpose: a native agent could link
        // Rust's regex, but that would make the operator behave differently on
        // desktop than in the browser (where TR-434 injects the host RegExp) —
        // the exact divergence this endpoint exists to prevent.
        let raw = post(
            "/assert",
            serde_json::json!({
                "response": {
                    "status": 200, "status_text": "OK", "headers": [],
                    "body": "abc", "response_time": 1.0, "size": 3, "cookies": []
                },
                "assertions": [{"target": "body", "operator": "matches", "expected": "^a"}]
            })
            .to_string(),
        )
        .await;
        assert!(raw.contains("regex matcher"), "{raw}");

        // ── /operators ──
        let mut s = TcpStream::connect(("127.0.0.1", port))
            .await
            .expect("connect");
        s.write_all(b"GET /operators HTTP/1.1\r\nHost: localhost\r\n\r\n")
            .await
            .expect("write");
        let mut out = Vec::new();
        s.read_to_end(&mut out).await.expect("read");
        let raw = String::from_utf8_lossy(&out).to_string();
        assert!(raw.contains(r#""name":"eq""#), "{raw}");
        assert!(raw.contains(r#""arity":"unary""#), "{raw}");

        // ── /variables/dynamic (TR-451) ──
        // Each occurrence must generate a FRESH value — that is the whole
        // point of the catalogue, and a cached-once implementation would put
        // the same "unique" id on every request in a run.
        let raw = post(
            "/variables/dynamic",
            serde_json::json!({"template": "{{$guid}}|{{$guid}}"}).to_string(),
        )
        .await;
        let body = raw.split("\r\n\r\n").nth(1).unwrap_or_default().to_string();
        let parsed: serde_json::Value = serde_json::from_str(&body).expect("json body");
        let value = parsed["value"].as_str().expect("value");
        let (first, second) = value.split_once('|').expect("two guids");
        assert_ne!(
            first, second,
            "each occurrence generates a fresh value: {value}"
        );
        assert_eq!(first.len(), 36, "a v4 GUID, not a placeholder: {value}");

        // Plain `{{var}}` is left ALONE here. /resolve owns that map, and
        // KnockPort runs the two in order — folding them together would
        // change which pass saw a `{{$guid}}` that came out of a variable.
        let raw = post(
            "/variables/dynamic",
            serde_json::json!({"template": "{{base}}/{{$timestamp}}"}).to_string(),
        )
        .await;
        assert!(
            raw.contains("{{base}}"),
            "a plain variable must survive the dynamic pass untouched: {raw}"
        );

        // ── /variables/dynamic/batch (TR-452) ──
        // One output per input, in order: the caller re-assembles its request
        // by index, so a short or reordered reply puts a header's value in a
        // param.
        let raw = post(
            "/variables/dynamic/batch",
            serde_json::json!({
                "items": [
                    {"template": "{{$guid}}"},
                    {"template": "no tokens here"},
                    {"template": "{{$timestamp}}"}
                ]
            })
            .to_string(),
        )
        .await;
        let body = raw.split("\r\n\r\n").nth(1).unwrap_or_default().to_string();
        let parsed: serde_json::Value = serde_json::from_str(&body).expect("json body");
        let items = parsed["items"].as_array().expect("items");
        assert_eq!(items.len(), 3, "one output per input, always: {body}");
        assert_eq!(
            items[1]["value"], "no tokens here",
            "a template with nothing to resolve comes back unchanged: {body}"
        );
        assert_ne!(
            items[0]["value"], items[2]["value"],
            "different tokens, different values: {body}"
        );

        // ── /constants (TR-451) ──
        let mut s = TcpStream::connect(("127.0.0.1", port))
            .await
            .expect("connect");
        s.write_all(b"GET /constants HTTP/1.1\r\nHost: localhost\r\n\r\n")
            .await
            .expect("write");
        let mut out = Vec::new();
        s.read_to_end(&mut out).await.expect("read");
        let raw = String::from_utf8_lossy(&out).to_string();
        let body = raw.split("\r\n\r\n").nth(1).unwrap_or_default().to_string();
        let parsed: serde_json::Value = serde_json::from_str(&body).expect("json body");
        // Read from the resolver's own constant, never a literal here: a
        // second ceiling is what KP-424 removed, and pinning 20 in this test
        // would quietly re-introduce one.
        assert_eq!(
            parsed["maxVariableResolutionPasses"],
            tropel_variables::MAX_VARIABLE_RESOLUTION_PASSES,
            "the cap must come from the resolver: {body}"
        );
        let vars = parsed["predefinedVariables"]
            .as_array()
            .expect("predefinedVariables array");
        assert_eq!(
            vars.len(),
            tropel_variables::PREDEFINED_VARIABLE_META.len(),
            "the whole catalogue, not a subset: {body}"
        );
        assert!(
            vars.iter()
                .any(|v| v["name"] == "$guid" && v["description"].is_string()),
            "names AND descriptions — the editor renders both: {body}"
        );
    }

    /// KT-202 — the THIRD differential leg: the same corpus, over the socket.
    ///
    /// `packages/core-wasm/fixtures/resolve-corpus.json` already runs through
    /// two paths: native Rust (`tropel-core-wasm`'s `conformance_corpus`) and
    /// the real wasm (`packages/core-wasm/smoke.mjs`). Those are the two tiers
    /// a BROWSER host can be served by. The agent is the third, and since
    /// KP-209 it is the one KnockPort's DESKTOP tier actually runs on — so an
    /// agent that resolved `{{base-url}}` differently would put literal text
    /// on the wire for desktop users only, which is the single divergence
    /// this corpus was written to catch in the first place.
    ///
    /// It reads the SAME bytes rather than restating the cases. A copied
    /// corpus drifts exactly the way the two resolvers did.
    ///
    /// This deviates from KT-202's literal wording ("run the corpus through
    /// the TypeScript host"): driving KnockPort's TypeScript from a Rust test
    /// would need Node and a second checkout inside tropel's CI. The agent leg
    /// covers what that was for — the desktop tier's rules — with no
    /// cross-repo dependency, and the browser tier is already covered by
    /// smoke.mjs, which IS the TypeScript leg.
    #[tokio::test]
    async fn the_resolution_corpus_agrees_over_the_socket() {
        const CORPUS: &str = include_str!("../testdata/resolve-corpus.json");

        /// Mirrors `tropel-core-wasm`'s `generated_vars` — same kind, same
        /// construction from the SAME constant, so the two legs cannot drift
        /// into testing different chains.
        fn generated_vars(kind: &str) -> serde_json::Value {
            match kind {
                "chain_longer_than_cap" => {
                    let cap = tropel_variables::MAX_VARIABLE_RESOLUTION_PASSES;
                    let mut m = serde_json::Map::new();
                    for i in 0..=cap {
                        m.insert(
                            format!("v{i}"),
                            serde_json::json!(format!("{{{{v{}}}}}", i + 1)),
                        );
                    }
                    m.insert(format!("v{}", cap + 1), serde_json::json!("end"));
                    serde_json::Value::Object(m)
                }
                other => panic!("unknown vars_generated kind: {other}"),
            }
        }

        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let port = listener.local_addr().expect("addr").port();
        let state = Arc::new(AgentState {
            token: None,
            client: tropel_http::HttpClient::new(&tropel_http::config::HttpConfig::default())
                .expect("http client"),
            runs: std::sync::Mutex::new(HashMap::new()),
            // No browser origin: these drive the socket directly, and an
            // empty allowlist is the default a real agent starts with.
            allowed_origins: vec![],
        });
        tokio::spawn(async move {
            while let Ok((mut sock, _)) = listener.accept().await {
                let st = state.clone();
                tokio::spawn(async move {
                    let _ = handle_connection(&mut sock, st).await;
                });
            }
        });

        let doc: serde_json::Value = serde_json::from_str(CORPUS).expect("corpus is valid JSON");
        let cases = doc["cases"].as_array().expect("cases array");
        assert!(!cases.is_empty(), "an empty corpus asserts nothing");

        for case in cases {
            let name = case["name"].as_str().expect("every case is named");
            let template = case["template"].as_str().expect("template");
            let mode = case["mode"].as_str().expect("mode");
            let vars = match case.get("vars_generated") {
                Some(kind) => generated_vars(kind.as_str().unwrap()),
                None => case["vars"].clone(),
            };

            // Through /resolve/batch, because that is the endpoint the desktop
            // tier actually calls — a leg that exercised a different route
            // would not be testing the path users get.
            let body = serde_json::json!({
                "variables": vars,
                "items": [{"template": template, "mode": mode}],
            })
            .to_string();
            let mut s = TcpStream::connect(("127.0.0.1", port))
                .await
                .expect("connect");
            let req = format!(
                "POST /resolve/batch HTTP/1.1\r\nHost: localhost\r\nContent-Length: {}\r\n\r\n{body}",
                body.len()
            );
            s.write_all(req.as_bytes()).await.expect("write");
            let mut out = Vec::new();
            s.read_to_end(&mut out).await.expect("read");
            let raw = String::from_utf8_lossy(&out).to_string();
            let reply = raw.split("\r\n\r\n").nth(1).unwrap_or_default().to_string();
            let parsed: serde_json::Value =
                serde_json::from_str(&reply).unwrap_or_else(|e| panic!("{name}: {e} — {reply}"));
            let item = &parsed["items"][0];
            assert!(
                item["error"].is_null(),
                "{name}: the agent refused a corpus case: {item}"
            );
            let value = item["value"]
                .as_str()
                .unwrap_or_else(|| panic!("{name}: no value"));

            // The same assertions the native leg makes, in the same order.
            if let Some(expected) = case.get("expect").and_then(|v| v.as_str()) {
                assert_eq!(value, expected, "{name}");
            }
            if case.get("parses_as_json").and_then(|v| v.as_bool()) == Some(true) {
                serde_json::from_str::<serde_json::Value>(value).unwrap_or_else(|e| {
                    panic!("{name}: result must stay parseable JSON: {e} — {value}")
                });
            }
            if let Some(expected) = case.get("expect_hit_cap").and_then(|v| v.as_bool()) {
                assert_eq!(item["hitCap"], expected, "{name}: hitCap");
            }
            if let Some(expected) = case.get("expect_unresolved") {
                assert_eq!(&item["unresolved"], expected, "{name}: unresolved");
            }
            if let Some(required) = case
                .get("expect_unresolved_contains")
                .and_then(|v| v.as_array())
            {
                let got = item["unresolved"].as_array().unwrap();
                for n in required {
                    assert!(
                        got.contains(n),
                        "{name}: unresolved must contain {n} — got {got:?}"
                    );
                }
            }
        }
    }

    /// TR-459 — the CORS + Private Network Access preflight (KT-402).
    ///
    /// Driven over a real socket because the failure this guards is a HEADER
    /// that is absent, and a unit test on the builder would pass while the
    /// router never called it.
    #[tokio::test]
    async fn the_preflight_answers_cors_and_private_network_for_allowed_origins() {
        async fn agent_with(origins: Vec<String>) -> u16 {
            let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
            let port = listener.local_addr().expect("addr").port();
            let state = Arc::new(AgentState {
                token: Some("s3cret".into()),
                client: tropel_http::HttpClient::new(&tropel_http::config::HttpConfig::default())
                    .expect("http client"),
                runs: std::sync::Mutex::new(HashMap::new()),
                allowed_origins: origins,
            });
            tokio::spawn(async move {
                while let Ok((mut sock, _)) = listener.accept().await {
                    let st = state.clone();
                    tokio::spawn(async move {
                        let _ = handle_connection(&mut sock, st).await;
                    });
                }
            });
            port
        }

        async fn preflight(port: u16, origin: &str, ask_pna: bool) -> String {
            let mut s = TcpStream::connect(("127.0.0.1", port))
                .await
                .expect("connect");
            let pna = if ask_pna {
                "Access-Control-Request-Private-Network: true\r\n"
            } else {
                ""
            };
            let req = format!(
                "OPTIONS /resolve/batch HTTP/1.1\r\nHost: localhost\r\nOrigin: {origin}\r\n\
                 Access-Control-Request-Method: POST\r\n{pna}\r\n"
            );
            s.write_all(req.as_bytes()).await.expect("write");
            let mut out = Vec::new();
            s.read_to_end(&mut out).await.expect("read");
            String::from_utf8_lossy(&out).to_string()
        }

        let allowed = "https://app.knockport.dev";
        let port = agent_with(vec![allowed.to_string()]).await;

        // THE Chrome-only bug this exists to prevent. Everything works in
        // Firefox and Safari without this header; Chrome refuses a public
        // page's request to 127.0.0.1 unless the preflight says so.
        let raw = preflight(port, allowed, true).await;
        assert!(raw.starts_with("HTTP/1.1 204"), "{raw}");
        assert!(
            raw.contains("Access-Control-Allow-Private-Network: true"),
            "Chrome's PNA preflight must be answered or this breaks in Chrome ONLY: {raw}"
        );
        assert!(
            raw.contains(&format!("Access-Control-Allow-Origin: {allowed}")),
            "{raw}"
        );
        // Cached per-origin, so a proxy cannot serve one origin's answer to
        // another.
        assert!(raw.contains("Vary: Origin"), "{raw}");

        // The PNA header appears only when ASKED — a browser that did not
        // request private-network access should not be handed the grant.
        let raw = preflight(port, allowed, false).await;
        assert!(raw.starts_with("HTTP/1.1 204"), "{raw}");
        assert!(
            !raw.contains("Access-Control-Allow-Private-Network"),
            "the grant must not be volunteered: {raw}"
        );

        // An origin nobody allowed is REFUSED BY NAME. "The agent is not
        // running" and "the agent is running and does not trust this page"
        // are different problems, and a page can only tell them apart if the
        // reply says which.
        let raw = preflight(port, "https://evil.test", true).await;
        assert!(raw.starts_with("HTTP/1.1 403"), "{raw}");
        assert!(
            raw.contains("--allow-origin"),
            "the refusal must say how to fix it: {raw}"
        );
        assert!(
            !raw.contains("Access-Control-Allow-Origin"),
            "a refused origin must NOT be handed a grant: {raw}"
        );

        // Default is deny: an agent started without --allow-origin is not
        // reachable from any page at all.
        let closed = agent_with(vec![]).await;
        let raw = preflight(closed, allowed, true).await;
        assert!(raw.starts_with("HTTP/1.1 403"), "{raw}");
    }

    /// TR-459 — the preflight is answered BEFORE the token check.
    ///
    /// A browser never sends `Authorization` on a preflight. Requiring the
    /// token there would 401 every cross-origin call, and the page would see
    /// a CORS error rather than an auth one — the wrong problem to debug.
    #[tokio::test]
    async fn the_preflight_does_not_require_the_token() {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let port = listener.local_addr().expect("addr").port();
        let state = Arc::new(AgentState {
            token: Some("s3cret".into()),
            client: tropel_http::HttpClient::new(&tropel_http::config::HttpConfig::default())
                .expect("http client"),
            runs: std::sync::Mutex::new(HashMap::new()),
            allowed_origins: vec!["https://app.knockport.dev".into()],
        });
        tokio::spawn(async move {
            while let Ok((mut sock, _)) = listener.accept().await {
                let st = state.clone();
                tokio::spawn(async move {
                    let _ = handle_connection(&mut sock, st).await;
                });
            }
        });

        let mut s = TcpStream::connect(("127.0.0.1", port))
            .await
            .expect("connect");
        // No Authorization header, exactly as a browser sends it.
        s.write_all(
            b"OPTIONS /resolve HTTP/1.1\r\nHost: localhost\r\nOrigin: https://app.knockport.dev\r\n\r\n",
        )
        .await
        .expect("write");
        let mut out = Vec::new();
        s.read_to_end(&mut out).await.expect("read");
        let raw = String::from_utf8_lossy(&out).to_string();
        assert!(
            raw.starts_with("HTTP/1.1 204"),
            "a preflight must not be 401'd: {raw}"
        );

        // But a REAL request still needs the token — the preflight carve-out
        // must not become an auth hole.
        let mut s = TcpStream::connect(("127.0.0.1", port))
            .await
            .expect("connect");
        s.write_all(
            b"GET /version HTTP/1.1\r\nHost: localhost\r\nOrigin: https://app.knockport.dev\r\n\r\n",
        )
        .await
        .expect("write");
        let mut out = Vec::new();
        s.read_to_end(&mut out).await.expect("read");
        let raw = String::from_utf8_lossy(&out).to_string();
        assert!(raw.starts_with("HTTP/1.1 401"), "{raw}");
    }

    /// TR-468 — a test script sees the response it is asserting against.
    ///
    /// `st.response` was never seeded either, so `pm.response.code` was
    /// undefined and every `pm.test` that looked at the response asserted
    /// against nothing. The test stage could not run on this realm at all —
    /// which is the stage KT-404 moves to it.
    #[tokio::test]
    async fn a_test_script_sees_the_response() {
        let response = tropel_sdk::types::Response {
            url: "https://api.test/v1".into(),
            status_code: 201,
            status_text: "Created".into(),
            protocol: "HTTP/1.1".into(),
            headers: HashMap::from([("content-type".to_string(), "application/json".to_string())]),
            body: br#"{"id":7}"#.to_vec(),
            text_cache: std::sync::OnceLock::new(),
            json_cache: std::sync::OnceLock::new(),
            response_time: std::time::Duration::from_millis(1),
            timings: None,
            cookies: Vec::new(),
            size: 8,
            redirects: Vec::new(),
            request_body_size: 0,
        };

        let out = run_script_once(
            "pm.test('status', () => pm.response.code === 201);\n             pm.test('body', () => pm.response.json().id === 7);",
            ScriptScopes::default(),
            None,
            Some(response),
            tropel_sandbox::config::SandboxConfig::default(),
            ScriptHost { http: test_http_client(), callbacks: None, cookies: None },
        )
        .await
        .expect("the realm runs");

        assert!(
            out.get("scriptError").is_some_and(|e| e.is_null()),
            "the script must not error: {out}"
        );
        let tests = out["tests"].as_array().expect("tests array");
        assert_eq!(tests.len(), 2, "both checks must have run: {out}");
        for t in tests {
            assert_eq!(
                t["passed"], true,
                "a check asserting against the seeded response must PASS — a \
                 failing one means the response was not there: {t}"
            );
        }
    }

    /// TR-467 — a pre-request script's header reaches the caller.
    ///
    /// Before this, `/script` never seeded `st.request`, so
    /// `pm.request.headers.add(...)` wrote to nothing and the reply carried no
    /// request at all. The script ran, reported success, and its header simply
    /// did not exist — a silent no-op, on the stage whose entire job is
    /// mutating the request (invariant #7).
    ///
    /// KT-404 makes this load-bearing: desktop runs pre-request scripts in
    /// THIS realm, so a header added by a script would have vanished on
    /// desktop while working in the app.
    #[tokio::test]
    async fn a_pre_request_script_mutation_comes_back() {
        let request = TropelRequest {
            url: "https://api.test/v1".into(),
            method: Method::GET,
            headers: vec![("Accept".into(), "application/json".into())],
            query_params: HashMap::new(),
            body: None,
            auth: None,
            certificate: None,
            proxy: None,
            follow_redirects: true,
            host: None,
            cookies: Vec::new(),
            timeout: None,
            response_type: ResponseType::Text,
        };

        let out = run_script_once(
            "pm.request.headers.add({ key: 'X-Trace', value: 'abc' });",
            ScriptScopes::default(),
            Some(request),
            None,
            tropel_sandbox::config::SandboxConfig::default(),
            ScriptHost {
                http: test_http_client(),
                callbacks: None,
                cookies: None,
            },
        )
        .await
        .expect("the realm runs");

        assert!(
            out.get("scriptError").is_some_and(|e| e.is_null()),
            "the script must not error: {out}"
        );
        let headers = out
            .get("request")
            .and_then(|r| r.get("headers"))
            .and_then(|h| h.as_array())
            .unwrap_or_else(|| panic!("the mutated request must come back: {out}"));
        let rendered = format!("{headers:?}");
        assert!(
            rendered.contains("X-Trace") && rendered.contains("abc"),
            "the header the script added must survive the round trip: {rendered}"
        );
        // And the header it started with must still be there — a reply that
        // returned ONLY the additions would silently drop the rest.
        assert!(
            rendered.contains("Accept"),
            "the original headers must survive too: {rendered}"
        );
    }

    /// TR-469: the embedder's canonical namespace reaches the realm.
    ///
    /// The API client builds its LOCAL realm with namespace `kp`, but the
    /// agent applied tropel's stock install (canonical `trp`, no aliases)
    /// because `/script` never rendered a `SandboxConfig` preamble. So the
    /// very same script passed in the app and died with "kp is not defined"
    /// on the desktop tier — two realms for one script, disagreeing where
    /// nothing looked. The D4 question ("can two implementations disagree
    /// invisibly?") answered yes, and no test asked it.
    ///
    /// Both halves are asserted deliberately: that the configured name works,
    /// AND that the stock install genuinely lacks it. Without the second, a
    /// future default of `kp` everywhere would make this test vacuous while
    /// still passing.
    /// A real client for the script realm's `pm.sendRequest`. These tests do
    /// not send anywhere; it exists so the bridge is built the way production
    /// builds it, rather than in the `http_client: None` state that made
    /// `pm.sendRequest` a declared-but-dead binding (TR-472).
    fn test_http_client() -> tropel_http::HttpClient {
        tropel_http::HttpClient::new(&tropel_http::config::HttpConfig::default())
            .expect("test http client")
    }

    #[tokio::test]
    async fn the_embedder_namespace_reaches_the_script_realm() {
        let code = "kp.environment.set('viaKp', '2');";

        let configured = run_script_once(
            code,
            ScriptScopes::default(),
            None,
            None,
            tropel_sandbox::config::SandboxConfig {
                namespace: "kp".into(),
                aliases: Vec::new(),
            },
            ScriptHost {
                http: test_http_client(),
                callbacks: None,
                cookies: None,
            },
        )
        .await
        .expect("the realm runs");
        assert!(
            configured
                .get("scriptError")
                .map(|e| e.is_null())
                .unwrap_or(false),
            "a kp.* script must run when the caller declares the kp namespace: {configured}"
        );
        assert_eq!(
            configured
                .get("environment")
                .and_then(|e| e.get("viaKp"))
                .and_then(|v| v.as_str()),
            Some("2"),
            "the effect must come back, not just the absence of an error: {configured}"
        );

        let stock = run_script_once(
            code,
            ScriptScopes::default(),
            None,
            None,
            tropel_sandbox::config::SandboxConfig::default(),
            ScriptHost {
                http: test_http_client(),
                callbacks: None,
                cookies: None,
            },
        )
        .await
        .expect("the realm runs");
        let err = stock
            .get("scriptError")
            .and_then(|e| e.as_str())
            .unwrap_or("");
        assert!(
            err.contains("kp is not defined"),
            "the stock install must NOT bind kp — if it does, this test no \
             longer proves the preamble is what carries the namespace: {stock}"
        );
    }

    /// TR-474: a script calls OUT of the realm and resumes with the answer.
    ///
    /// This is the one thing no amount of payload widening could do:
    /// `bru.runRequest` resolves a name out of a collection the agent has
    /// never seen. The realm parks mid-script, the host answers, the script
    /// carries on with the value.
    ///
    /// Driven at the channel rather than over HTTP so it is hermetic and
    /// fast; the two endpoints on top of it are exercised end to end in the
    /// PR description against a live agent.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_script_calls_back_into_the_host_and_resumes() {
        let cb = Arc::new(RunCallbacks::default());

        // The host half: wait for the parked call, answer it.
        let host = {
            let cb = cb.clone();
            tokio::spawn(async move {
                for _ in 0..200 {
                    let call = {
                        let mut p = cb.pending.lock().unwrap();
                        p.pop_front()
                    };
                    if let Some(call) = call {
                        let tx = cb.replies.lock().unwrap().remove(&call.call_id);
                        if let Some(tx) = tx {
                            let _ = tx.send(
                                serde_json::json!({ "status": 201, "body": call.path }).to_string(),
                            );
                        }
                        return true;
                    }
                    tokio::time::sleep(std::time::Duration::from_millis(10)).await;
                }
                false
            })
        };

        let out = run_script_once(
            "var r = bru.runRequest('Login');\
             kp.environment.set('status', String(r.status));\
             kp.environment.set('echoed', String(r.body));",
            ScriptScopes::default(),
            None,
            None,
            tropel_sandbox::config::SandboxConfig {
                namespace: "kp".into(),
                aliases: Vec::new(),
            },
            ScriptHost {
                http: test_http_client(),
                callbacks: Some(cb.clone()),
                cookies: None,
            },
        )
        .await
        .expect("the realm runs");

        assert!(host.await.unwrap_or(false), "the host never saw the call");
        assert!(
            out.get("scriptError").map(|e| e.is_null()).unwrap_or(false),
            "the script must not error: {out}"
        );
        let env = out.get("environment").expect("environment comes back");
        assert_eq!(
            env.get("status").and_then(|v| v.as_str()),
            Some("201"),
            "the script must resume with the HOST's answer, not a placeholder: {out}"
        );
        assert_eq!(
            env.get("echoed").and_then(|v| v.as_str()),
            Some("Login"),
            "the path the script asked for must reach the host verbatim: {out}"
        );
    }

    /// TR-474: without a channel, `bru.runRequest` REFUSES BY NAME.
    ///
    /// A load run has no collection to re-enter. Returning undefined there
    /// would read as "the request ran and gave nothing back" — the silent
    /// failure invariant 8 forbids — and the binding must not exist at all
    /// when nothing is listening for it (invariant 4).
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn run_request_refuses_when_no_channel_is_open() {
        let out = run_script_once(
            "bru.runRequest('Login');",
            ScriptScopes::default(),
            None,
            None,
            tropel_sandbox::config::SandboxConfig::default(),
            ScriptHost {
                http: test_http_client(),
                callbacks: None,
                cookies: None,
            },
        )
        .await
        .expect("the realm runs");

        let err = out
            .get("scriptError")
            .and_then(|e| e.as_str())
            .unwrap_or("");
        assert!(
            err.contains("not available here"),
            "it must refuse by name rather than return undefined: {out}"
        );
    }

    /// TR-477: a script-issued send is RECORDED, including one that failed.
    ///
    /// The app renders these as timeline arms. Without them a script send is
    /// invisible on this tier — the KP-413 defect ("script-issued sends were
    /// invisible") landing again, one tier over.
    ///
    /// The failed send is the point. Recording only successes would make a
    /// script whose request never connected look like a script that never
    /// made one, which is the silent loss invariant 7 forbids. Hermetic: it
    /// sends to a port nothing listens on, so the failure is the assertion
    /// rather than the network's mood.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_failed_script_send_is_still_recorded() {
        let out = run_script_once(
            "pm.sendRequest('http://127.0.0.1:1/nope', function () {});",
            ScriptScopes::default(),
            None,
            None,
            tropel_sandbox::config::SandboxConfig {
                namespace: "kp".into(),
                aliases: Vec::new(),
            },
            ScriptHost {
                http: test_http_client(),
                callbacks: None,
                cookies: None,
            },
        )
        .await
        .expect("the realm runs");

        let sends = out
            .get("scriptSends")
            .and_then(|v| v.as_array())
            .expect("scriptSends is always an array, never null");
        assert_eq!(
            sends.len(),
            1,
            "the send must be recorded even though it failed: {out}"
        );
        let only = &sends[0];
        assert_eq!(
            only.get("url").and_then(|v| v.as_str()),
            Some("http://127.0.0.1:1/nope"),
            "the recorded URL must be the one the script asked for: {out}"
        );
        assert_eq!(
            only.get("status").and_then(|v| v.as_u64()),
            Some(0),
            "a send that never connected has no status — 0, not a fabricated one: {out}"
        );
        assert!(
            only.get("error").map(|e| !e.is_null()).unwrap_or(false),
            "the failure must be NAMED on the record, not implied by status 0: {out}"
        );
    }

    /// TR-477: no sends is an EMPTY list, not null.
    ///
    /// "The script sent nothing" is an answer. It must not read the same as
    /// "this tier does not record sends".
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_script_that_sends_nothing_reports_an_empty_list() {
        let out = run_script_once(
            "kp.environment.set('x', '1');",
            ScriptScopes::default(),
            None,
            None,
            tropel_sandbox::config::SandboxConfig::default(),
            ScriptHost {
                http: test_http_client(),
                callbacks: None,
                cookies: None,
            },
        )
        .await
        .expect("the realm runs");
        assert_eq!(
            out.get("scriptSends")
                .and_then(|v| v.as_array())
                .map(|a| a.len()),
            Some(0),
            "an empty array, not null: {out}"
        );
    }

    /// TR-478: `bru.getTestResults` answers; `getAssertionResults` refuses.
    ///
    /// Both lived only in the API client's own prelude, so a script calling
    /// either worked in the app and died on a bare ReferenceError here — the
    /// same one-script-two-answers split `fetch` had before TR-475.
    ///
    /// Only ONE of them can live here, and the asymmetry is the point:
    /// `getTestResults` reports what THIS realm recorded, which is ours to
    /// answer. `getAssertionResults` evaluates the caller's declarative
    /// assertion grammar through its operator table — a vocabulary this
    /// runtime does not have, and re-implementing it would be a second
    /// semantics for one language.
    ///
    /// So the second REFUSES BY NAME. An empty array would be an answer —
    /// "nothing failed" — and would be a lie.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn test_results_answer_and_assertion_results_refuse() {
        let out = run_script_once(
            "kp.test('first', function () {});\
             kp.test('second', function () { throw new Error('no'); });\
             var r = bru.getTestResults();\
             kp.environment.set('count', String(r.length));\
             kp.environment.set('shape', r.map(function (x) { return x.name + ':' + x.passed; }).join(','));",
            ScriptScopes::default(),
            None,
            None,
            tropel_sandbox::config::SandboxConfig {
                namespace: "kp".into(),
                aliases: Vec::new(),
            },
            ScriptHost { http: test_http_client(), callbacks: None, cookies: None },
        )
        .await
        .expect("the realm runs");

        let env = out.get("environment").expect("environment comes back");
        assert_eq!(
            env.get("count").and_then(|v| v.as_str()),
            Some("2"),
            "both checks must be reported, not just the passing one: {out}"
        );
        assert_eq!(
            env.get("shape").and_then(|v| v.as_str()),
            Some("first:true,second:false"),
            "each result must carry its own name AND outcome — a count alone \
             cannot tell a passing suite from a failing one: {out}"
        );

        let refused = run_script_once(
            "bru.getAssertionResults();",
            ScriptScopes::default(),
            None,
            None,
            tropel_sandbox::config::SandboxConfig::default(),
            ScriptHost {
                http: test_http_client(),
                callbacks: None,
                cookies: None,
            },
        )
        .await
        .expect("the realm runs");
        let err = refused
            .get("scriptError")
            .and_then(|e| e.as_str())
            .unwrap_or("");
        assert!(
            err.contains("not available here"),
            "it must refuse by NAME rather than return an empty list, which would \
             read as \"no assertions failed\": {refused}"
        );
    }

    /// TR-476: the cookie jar round-trips, and refuses when absent.
    ///
    /// Three properties in one, because they only mean anything together:
    /// the script READS what the caller seeded, it READS BACK its own writes
    /// (not the seeded snapshot), and the caller gets the ops in order so it
    /// can replay them onto the jar that outlives the script.
    #[tokio::test]
    async fn the_cookie_jar_round_trips_and_refuses_when_absent() {
        let jar = Arc::new(ScriptCookies::default());
        *jar.lock_jar() = vec![serde_json::json!({
            "key": "sid", "value": "abc", "domain": "api.example.test", "path": "/"
        })];

        let out = run_script_once(
            "kp.environment.set('seeded', String(bru.cookies.get('sid')));\
             bru.cookies.add({ key: 'new', value: 'n1', domain: 'api.example.test', path: '/' });\
             kp.environment.set('readback', String(bru.cookies.get('new')));\
             bru.cookies.remove('sid');\
             kp.environment.set('count', String(bru.cookies.count()));",
            ScriptScopes::default(),
            Some(TropelRequest {
                url: "https://api.example.test/things".to_string(),
                ..Default::default()
            }),
            None,
            tropel_sandbox::config::SandboxConfig {
                namespace: "kp".into(),
                aliases: Vec::new(),
            },
            ScriptHost {
                http: test_http_client(),
                callbacks: None,
                cookies: Some(jar.clone()),
            },
        )
        .await
        .expect("the realm runs");

        let env = out.get("environment").expect("environment comes back");
        assert_eq!(
            env.get("seeded").and_then(|v| v.as_str()),
            Some("abc"),
            "the script must read what the caller seeded: {out}"
        );
        assert_eq!(
            env.get("readback").and_then(|v| v.as_str()),
            Some("n1"),
            "read-your-writes: a cookie the script just set must be visible to it, \
             not shadowed by the seeded snapshot: {out}"
        );
        assert_eq!(
            env.get("count").and_then(|v| v.as_str()),
            Some("1"),
            "the removed cookie must be gone from the script's view too: {out}"
        );

        let ops = out
            .get("cookieOps")
            .and_then(|o| o.as_array())
            .expect("cookieOps come back");
        let kinds: Vec<&str> = ops
            .iter()
            .filter_map(|o| o.get("op").and_then(|v| v.as_str()))
            .collect();
        assert_eq!(
            kinds,
            vec!["set", "delete"],
            "the caller replays these onto the real jar, so ORDER is part of the \
             contract, not just the set of ops: {out}"
        );
    }

    /// TR-476: no jar supplied is a NAMED refusal, not an empty jar.
    ///
    /// "No cookies" and "no cookie jar" must not look the same to a script —
    /// the first is an answer, the second is a missing capability.
    #[tokio::test]
    async fn bru_cookies_refuses_when_no_jar_was_supplied() {
        let out = run_script_once(
            "bru.cookies.get('sid');",
            ScriptScopes::default(),
            None,
            None,
            tropel_sandbox::config::SandboxConfig::default(),
            ScriptHost {
                http: test_http_client(),
                callbacks: None,
                cookies: None,
            },
        )
        .await
        .expect("the realm runs");

        let err = out
            .get("scriptError")
            .and_then(|e| e.as_str())
            .unwrap_or("");
        assert!(
            err.contains("not available here"),
            "it must refuse by name rather than read as an empty jar: {out}"
        );
        assert!(
            out.get("cookieOps").map(|v| v.is_null()).unwrap_or(false),
            "no jar means no ops, not an empty op list: {out}"
        );
    }

    /// TR-475: `fetch` exists in the realm and rides the SAME client.
    ///
    /// QuickJS ships none, so a script written the way every modern one is
    /// written — `await fetch(...)` — died on a bare ReferenceError, while
    /// the identical call through `pm.sendRequest` worked. Two spellings of
    /// one capability, one of which was a crash.
    ///
    /// Hermetic, like the send-request test: it fetches a port nothing
    /// listens on, so the call MUST fail, and asserts WHICH failure. A
    /// connection error proves the binding reached the client; "not
    /// available here" proves it did not.
    #[tokio::test]
    async fn fetch_is_bound_to_the_same_host_client() {
        let out = run_script_once(
            "(async function () {\
               try {\
                 var r = await fetch('http://127.0.0.1:1/');\
                 kp.environment.set('status', String(r.status));\
               } catch (e) {\
                 kp.environment.set('err', String(e && e.message));\
               }\
             })();",
            ScriptScopes::default(),
            None,
            None,
            tropel_sandbox::config::SandboxConfig {
                namespace: "kp".into(),
                aliases: Vec::new(),
            },
            ScriptHost {
                http: test_http_client(),
                callbacks: None,
                cookies: None,
            },
        )
        .await
        .expect("the realm runs");

        let env = out.get("environment").expect("environment comes back");
        let err = env.get("err").and_then(|v| v.as_str()).unwrap_or("");
        assert!(
            !err.contains("not available here"),
            "fetch must be bound to a real client, not refuse for lack of one: {out}"
        );
        assert!(
            !err.is_empty() || env.get("status").is_some(),
            "the call must have gone somewhere — neither a result nor an error \
             means `fetch` never ran: {out}"
        );
    }

    /// TR-473: all FOUR scopes reach the realm and come back.
    ///
    /// `/script` carried only `environment`. The realm has always had four
    /// separate stores with a defined precedence, so a script reading
    /// `pm.collectionVariables` / `pm.globals` / `pm.variables` saw an empty
    /// scope, and anything it wrote to them was dropped on the way out.
    ///
    /// Both directions are asserted for each scope. Seeding alone would pass
    /// against a realm that read the seed and discarded every write; returning
    /// alone would pass against one that started empty. It is the round trip
    /// that pins the behaviour.
    #[tokio::test]
    async fn every_variable_scope_round_trips() {
        let mut scopes = ScriptScopes::default();
        scopes.environment.insert("envIn".into(), "e".into());
        scopes
            .collection
            .insert("colIn".into(), serde_json::json!("c"));
        scopes
            .globals
            .insert("gloIn".into(), serde_json::json!("g"));
        scopes
            .variables
            .insert("varIn".into(), serde_json::json!("v"));

        let out = run_script_once(
            "kp.environment.set('envSeen', String(kp.environment.get('envIn')));\
             kp.collectionVariables.set('colSeen', String(kp.collectionVariables.get('colIn')));\
             kp.globals.set('gloSeen', String(kp.globals.get('gloIn')));\
             kp.variables.set('varSeen', String(kp.variables.get('varIn')));",
            scopes,
            None,
            None,
            tropel_sandbox::config::SandboxConfig {
                namespace: "kp".into(),
                aliases: Vec::new(),
            },
            ScriptHost {
                http: test_http_client(),
                callbacks: None,
                cookies: None,
            },
        )
        .await
        .expect("the realm runs");

        assert!(
            out.get("scriptError").map(|e| e.is_null()).unwrap_or(false),
            "the script must not error: {out}"
        );
        for (scope, seen, expected) in [
            ("environment", "envSeen", "e"),
            ("collectionVariables", "colSeen", "c"),
            ("globals", "gloSeen", "g"),
            ("variables", "varSeen", "v"),
        ] {
            let got = out
                .get(scope)
                .and_then(|m| m.get(seen))
                .and_then(|v| v.as_str());
            assert_eq!(
                got,
                Some(expected),
                "`{scope}` must be seeded AND returned — the script read \
                 `{expected}` from it and wrote `{seen}` back: {out}"
            );
        }
    }

    /// TR-472: `pm.sendRequest` is WIRED, not merely present.
    ///
    /// The bridge was built with `TrpBridge::new`, which leaves
    /// `http_client: None` — and the send-request binding is installed in
    /// that state anyway. So `typeof pm.sendRequest` was "function" and
    /// calling it handed the script
    /// `Error: pm.sendRequest unavailable in this build (no HTTP client)`:
    /// declared, present and dead (invariant 4).
    ///
    /// Hermetic on purpose. It sends to a port nothing listens on, so the
    /// call MUST fail — the assertion is on WHICH failure. A connection
    /// error proves the client was reached; the build-state error proves it
    /// was not. Asserting success would need the network and would pin
    /// somebody else's uptime instead of our wiring.
    #[tokio::test]
    async fn send_request_reaches_a_real_http_client() {
        let out = run_script_once(
            "pm.sendRequest('http://127.0.0.1:1/', function (err, res) {\
               kp.environment.set('err', String(err));\
               kp.environment.set('code', String(res && res.code));\
             });",
            ScriptScopes::default(),
            None,
            None,
            tropel_sandbox::config::SandboxConfig {
                namespace: "kp".into(),
                aliases: Vec::new(),
            },
            ScriptHost {
                http: test_http_client(),
                callbacks: None,
                cookies: None,
            },
        )
        .await
        .expect("the realm runs");

        let err = out
            .get("environment")
            .and_then(|e| e.get("err"))
            .and_then(|v| v.as_str())
            .unwrap_or("");
        assert!(
            !err.contains("unavailable in this build"),
            "the send-request bridge must hold a real client, not be installed \
             over `http_client: None`: {out}"
        );
        assert_ne!(
            err, "null",
            "port 1 cannot have answered — a success here means the call never \
             left the realm: {out}"
        );
    }

    /// TR-470: the QuickJS realm has no host escape.
    ///
    /// The desktop tier runs user scripts HERE rather than in the webview,
    /// and the sandbox is the reason that decision was taken: the API
    /// client's own realm gate states it "is NOT airtight" against an
    /// obfuscated escape and names this realm as the airtight path. That
    /// claim rested on READING the gate rather than attacking it — and an
    /// escape here sits next to the user's filesystem, where in a tab it
    /// would be tab-scoped.
    ///
    /// The indirect-eval case is the one that matters most: it is exactly
    /// what the textual gate admits it cannot catch, so this is the test
    /// that has to hold when that one does not.
    #[tokio::test]
    async fn the_script_realm_has_no_host_escape() {
        fn kp() -> tropel_sandbox::config::SandboxConfig {
            tropel_sandbox::config::SandboxConfig {
                namespace: "kp".into(),
                aliases: Vec::new(),
            }
        }

        // Host capabilities a script must not be able to name. Read back
        // through the environment rather than asserted on a throw, so a probe
        // that never ran cannot be mistaken for a probe that found nothing.
        // TR-475 removed `fetch` from this list, deliberately. It is now a
        // BOUND capability — the same audited host client `pm.sendRequest`
        // already rode — not an ambient one the realm happened to expose.
        // The realm was never network-isolated; `pm.sendRequest` predates
        // this test. Keeping `fetch` here would have asserted an isolation
        // the realm did not have, which is worse than not asserting it.
        //
        // What remains is the set that would reach the HOST PROCESS: the
        // module loader, `process`, and the Function-constructor route to a
        // global. Those are the escape; a guarded HTTP call is not.
        let probe = "kp.environment.set('process', typeof process);\
                     kp.environment.set('require', typeof require);\
                     kp.environment.set('module', typeof module);\
                     kp.environment.set('viaFn', typeof Function('return this')().process);";
        let out = run_script_once(
            probe,
            ScriptScopes::default(),
            None,
            None,
            kp(),
            ScriptHost {
                http: test_http_client(),
                callbacks: None,
                cookies: None,
            },
        )
        .await
        .expect("the realm runs");
        let env = out.get("environment").expect("the environment comes back");
        for name in ["process", "require", "module", "viaFn"] {
            assert_eq!(
                env.get(name).and_then(|v| v.as_str()),
                Some("undefined"),
                "`{name}` must not be reachable from a script: {out}"
            );
        }

        // And no host module loader is wired — including through an indirect
        // eval, which hides the syntax from any textual gate.
        for (label, code) in [
            ("direct", "import('fs');"),
            ("indirect eval", "var e = eval; e(\"import\" + \"('fs')\");"),
        ] {
            let result = run_script_once(
                code,
                ScriptScopes::default(),
                None,
                None,
                kp(),
                ScriptHost {
                    http: test_http_client(),
                    callbacks: None,
                    cookies: None,
                },
            )
            .await;
            let refused = match &result {
                Err(why) => why.contains("module"),
                Ok(out) => out
                    .get("scriptError")
                    .and_then(|e| e.as_str())
                    .is_some_and(|e| e.contains("module")),
            };
            assert!(
                refused,
                "a {label} import must not load a host module: {result:?}"
            );
        }
    }

    /// TR-465 / KT-404 — the QuickJS half of the script-realm corpus.
    ///
    /// The corpus is a COMMITTED, MEASURED table of what a user script can
    /// rely on in each realm — KnockPort's host-JS one and this one. Both
    /// repos read the same file, so a probe whose answer changes fails here
    /// AND there until the table is updated. That is the whole design: the
    /// divergence stops being folklore and becomes something a test owns.
    ///
    /// It runs through `run_script_once`, the production /script path, rather
    /// than a hand-built context — a realm assembled just for the test could
    /// pass while the one users reach is missing a shim, which is exactly the
    /// bug the `bru` probe found.
    #[tokio::test]
    async fn the_script_realm_matches_the_committed_corpus() {
        const CORPUS: &str = include_str!("../testdata/script-realm-corpus.json");
        let doc: serde_json::Value = serde_json::from_str(CORPUS).expect("corpus is valid JSON");
        let probes = doc["probes"].as_array().expect("probes array");
        assert!(!probes.is_empty(), "an empty corpus asserts nothing");

        // One script sets one environment key per probe, so a single realm
        // answers all of them — and the realm is built exactly once, as a
        // caller's would be.
        let script = probes
            .iter()
            .map(|p| {
                let name = p["name"].as_str().expect("name");
                let expr = p["expression"].as_str().expect("expression");
                format!("pm.environment.set({name:?}, String({expr}));")
            })
            .collect::<Vec<_>>()
            .join("\n");

        let out = run_script_once(
            &script,
            ScriptScopes::default(),
            None,
            None,
            tropel_sandbox::config::SandboxConfig::default(),
            ScriptHost {
                http: test_http_client(),
                callbacks: None,
                cookies: None,
            },
        )
        .await
        .expect("the realm runs");

        let mut wrong: Vec<String> = Vec::new();
        for probe in probes {
            let name = probe["name"].as_str().unwrap();
            let want = probe["quickJs"].as_str().unwrap();
            let got = out
                .get("environment")
                .and_then(|e| e.get(name))
                .and_then(|v| v.as_str())
                .unwrap_or("(probe did not run)");
            if got != want {
                wrong.push(format!(
                    "  {name}: corpus says {want:?}, realm answered {got:?}"
                ));
            }
        }
        assert!(
            wrong.is_empty(),
            "the QuickJS realm no longer matches the committed corpus.\n{}\n\n\
             If the realm CHANGED on purpose, update \
             packages/shims/fixtures/script-realm-corpus.json — and update the \
             `why` line too, because KnockPort's half of this corpus asserts \
             the same file and a user reads those lines to know what their \
             script can use.",
            wrong.join("\n")
        );
    }

    /// TR-463 — duplicate header names survive `/execute`.
    ///
    /// The object form cannot hold two entries with the same key, so
    /// `{"Accept": "a", "Accept": "b"}` is not even expressible — one row is
    /// gone before the agent sees it. Two `Set-Cookie` rows, or the `Accept`
    /// pair an API needs, arrived as one and nothing reported the other.
    ///
    /// Pins the ARRAY form carrying both, and the object form still working,
    /// because breaking the old shape would break every existing caller.
    #[test]
    fn the_execute_wire_carries_a_per_request_proxy() {
        // Exactly the parsing branch `execute_single` runs for `proxy`.
        // KnockPort declares every `proxy.*` capability false with "the
        // agent's client is built once from HttpConfig::default() and
        // /execute carries no proxy field" as its stated reason — this is
        // that field, so the reason has to stop being true.
        fn parse(v: &serde_json::Value) -> Option<tropel_sdk::types::ProxyConfig> {
            v.get("proxy")
                .and_then(|p| serde_json::from_value(p.clone()).ok())
        }

        let cfg = parse(&serde_json::json!({
            "proxy": {"mode": "fixed", "protocol": "http", "host": "p.internal",
                      "port": 3128, "username": "u", "password": "p",
                      "bypass": ["localhost", "*.internal"]}
        }))
        .expect("a fixed proxy reads");
        assert_eq!(cfg.mode, tropel_sdk::types::ProxyMode::Fixed);
        assert_eq!(cfg.fixed_url().as_deref(), Some("http://p.internal:3128"));
        assert_eq!(cfg.bypass.len(), 2);

        // camelCase, which is what KnockPort writes.
        let pac = parse(&serde_json::json!({
            "proxy": {"mode": "pac", "pacUrl": "http://wpad/proxy.pac"}
        }))
        .expect("a pac proxy reads");
        assert_eq!(pac.pac_url.as_deref(), Some("http://wpad/proxy.pac"));

        // ABSENT is the common case and must stay `None` — additive means a
        // client that sends no `proxy` is unaffected.
        assert!(parse(&serde_json::json!({"url": "https://x/y"})).is_none());

        // A malformed block is `None` rather than a parse failure, matching
        // `certificate`: the request is then refused downstream BY NAME
        // (`fixed` with no host, a bypass typo) rather than with a serde
        // message about a field the caller cannot see.
        assert!(parse(&serde_json::json!({"proxy": "http://p:3128"})).is_none());
    }

    #[test]
    fn duplicate_header_names_survive_the_execute_wire_format() {
        // Exactly the parsing branch `execute_single` runs.
        fn parse(v: &serde_json::Value) -> Vec<(String, String)> {
            match v.get("headers") {
                Some(serde_json::Value::Array(rows)) => rows
                    .iter()
                    .filter_map(|row| match row {
                        serde_json::Value::Array(pair) if pair.len() == 2 => Some((
                            pair[0].as_str()?.to_string(),
                            pair[1].as_str().unwrap_or("").to_string(),
                        )),
                        serde_json::Value::Object(o) => Some((
                            o.get("name")?.as_str()?.to_string(),
                            o.get("value")
                                .and_then(|v| v.as_str())
                                .unwrap_or("")
                                .to_string(),
                        )),
                        _ => None,
                    })
                    .collect(),
                Some(serde_json::Value::Object(o)) => o
                    .iter()
                    .map(|(k, v)| (k.clone(), v.as_str().unwrap_or("").to_string()))
                    .collect(),
                _ => Vec::new(),
            }
        }

        // Pair form: BOTH rows survive, in order.
        let pairs = parse(&serde_json::json!({
            "headers": [["Accept", "application/json"], ["Accept", "text/plain"]]
        }));
        assert_eq!(
            pairs,
            vec![
                ("Accept".to_string(), "application/json".to_string()),
                ("Accept".to_string(), "text/plain".to_string()),
            ],
            "a duplicate header name must reach the wire twice"
        );

        // Object form: still accepted, so existing callers keep working.
        let obj = parse(&serde_json::json!({"headers": {"Accept": "application/json"}}));
        assert_eq!(
            obj,
            vec![("Accept".to_string(), "application/json".to_string())]
        );

        // The {name, value} row shape too — what a KnockPort KeyValuePair
        // serialises to, so a caller does not have to reshape it first.
        let named = parse(&serde_json::json!({
            "headers": [{"name": "X-A", "value": "1"}, {"name": "X-A", "value": "2"}]
        }));
        assert_eq!(named.len(), 2, "{named:?}");

        // And the loss the object form CANNOT avoid, pinned so the reason
        // this changed is visible: serde keeps the last of two equal keys.
        let collapsed: serde_json::Value =
            serde_json::from_str(r#"{"headers":{"Accept":"a","Accept":"b"}}"#).unwrap();
        assert_eq!(
            parse(&collapsed).len(),
            1,
            "the object form loses one row before the agent ever sees it"
        );
    }

    /// TR-462 — a non-UTF-8 response body survives, and says how.
    ///
    /// Not a socket test: `/execute` needs a live upstream, and what is under
    /// test is the ENCODING DECISION, not the HTTP plumbing. Driving the same
    /// bytes through the same branch is the honest way to pin it — and the
    /// round trip through `base64_decode` proves the two halves of this file
    /// agree, which is the property that actually matters to a caller.
    #[test]
    fn a_binary_response_body_is_base64_not_mojibake() {
        // A PNG header: valid bytes, invalid UTF-8. `from_utf8_lossy` turns
        // every one of the high bytes into U+FFFD and reports nothing, so the
        // caller receives a corrupted image that looks like a server bug.
        //
        // Built through a function rather than as a literal: clippy
        // const-folds `from_utf8` on a literal and warns that it "always
        // returns an error", which is the very property under test.
        fn png_header() -> Vec<u8> {
            vec![0x89, b'P', b'N', b'G', 0x0D, 0x0A, 0x1A, 0x0A, 0xFF, 0xD8]
        }
        let png = png_header();
        assert!(
            std::str::from_utf8(&png).is_err(),
            "the fixture must actually be invalid UTF-8 or this test proves nothing"
        );

        let (body, encoding) = match std::str::from_utf8(&png) {
            Ok(text) => (text.to_string(), "utf8"),
            Err(_) => (base64_encode(&png), "base64"),
        };
        assert_eq!(encoding, "base64");
        assert_eq!(
            base64_decode(&body).expect("round trips"),
            png,
            "the bytes must come back EXACTLY — lossy is the bug"
        );

        // What the old code did, pinned so the difference is visible rather
        // than asserted: every high byte became the replacement character.
        let lossy = String::from_utf8_lossy(&png);
        assert!(
            lossy.contains('\u{FFFD}'),
            "the old path really did corrupt these bytes: {lossy:?}"
        );
        assert_ne!(
            lossy.as_bytes(),
            png,
            "and the corruption was unrecoverable — no field said so"
        );

        // Text is untouched: the common case must not start arriving as
        // base64, which would break every existing caller.
        fn json_body() -> Vec<u8> {
            br#"{"ok":true}"#.to_vec()
        }
        let text = json_body();
        let (body, encoding) = match std::str::from_utf8(&text) {
            Ok(t) => (t.to_string(), "utf8"),
            Err(_) => (base64_encode(&text), "base64"),
        };
        assert_eq!(encoding, "utf8");
        assert_eq!(body, "{\"ok\":true}");
    }

    /// TR-445: `/auth/sign`, over the socket.
    ///
    /// This is the endpoint that lets knockport's DESKTOP tier stop throwing
    /// `TropelAuthUnavailableError`. What it must prove is not "signing
    /// works" — `tropel-auth` has its own vectors for that — but that the
    /// desktop tier gets the SAME rules the browser does, applied here rather
    /// than re-derived in TypeScript.
    #[tokio::test]
    async fn the_auth_sign_endpoint_applies_the_rules_server_side() {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let port = listener.local_addr().expect("addr").port();
        let state = Arc::new(AgentState {
            token: None,
            client: tropel_http::HttpClient::new(&tropel_http::config::HttpConfig::default())
                .expect("http client"),
            runs: std::sync::Mutex::new(HashMap::new()),
            // No browser origin: these drive the socket directly, and an
            // empty allowlist is the default a real agent starts with.
            allowed_origins: vec![],
        });
        tokio::spawn(async move {
            while let Ok((mut sock, _)) = listener.accept().await {
                let st = state.clone();
                tokio::spawn(async move {
                    let _ = handle_connection(&mut sock, st).await;
                });
            }
        });
        let sign = |body: String| async move {
            let mut s = TcpStream::connect(("127.0.0.1", port))
                .await
                .expect("connect");
            let req = format!(
                "POST /auth/sign HTTP/1.1\r\nHost: localhost\r\nContent-Length: {}\r\n\r\n{body}",
                body.len()
            );
            s.write_all(req.as_bytes()).await.expect("write");
            let mut out = Vec::new();
            s.read_to_end(&mut out).await.expect("read");
            String::from_utf8_lossy(&out).to_string()
        };

        // SigV4: the service must derive to `s3` from a VIRTUAL-HOSTED bucket
        // host. A desktop tier deriving it itself would take the first DNS
        // label — the bug `default_service`'s comment records, a 403 on every
        // virtual-hosted-S3 request (TR-428).
        let raw = sign(
            serde_json::json!({
                "scheme": "awsSigV4",
                "params": {
                    "method": "GET", "host": "examplebucket.s3.amazonaws.com",
                    "path": "/test.txt", "accessKey": "AKID", "secretKey": "SECRET",
                    "region": "us-east-1", "amzDate": "20130524T000000Z",
                    "dateStamp": "20130524"
                }
            })
            .to_string(),
        )
        .await;
        assert!(
            raw.contains("/20130524/us-east-1/s3/aws4_request"),
            "service must derive to s3, not the bucket: {raw}"
        );
        // Every header that is IN the signature must come back, or the caller
        // sends a valid-looking Authorization and gets a 403.
        assert!(raw.contains("x-amz-date"), "{raw}");
        assert!(raw.contains("x-amz-content-sha256"), "{raw}");

        // Digest: the challenge is parsed HERE — multi-scheme and quoted qop,
        // the two things a naive client parser gets wrong (TR-429).
        let raw = sign(
            serde_json::json!({
                "scheme": "digest",
                "params": {
                    "wwwAuthenticate": "Basic realm=\"b\", Digest realm=\"r\", qop=\"auth, auth-int\", nonce=\"n\"",
                    "username": "u", "password": "p", "method": "GET",
                    "uri": "/dir/index.html", "nc": 1, "cnonce": "0a4f113b"
                }
            })
            .to_string(),
        )
        .await;
        assert!(raw.contains("Digest "), "{raw}");
        assert!(raw.contains(r#"realm=\"r\""#), "{raw}");

        // A header with no Digest challenge is a NAMED 400, not an unsigned
        // 200 — the caller must not re-send.
        let raw = sign(
            serde_json::json!({
                "scheme": "digest",
                "params": {"wwwAuthenticate": "Basic realm=\"b\"", "username": "u"}
            })
            .to_string(),
        )
        .await;
        assert!(raw.starts_with("HTTP/1.1 400"), "{raw}");
        assert!(raw.contains("no Digest challenge"), "{raw}");

        // OAuth1: the IPv6 host is bracketed HERE. A client forwarding
        // `URL.hostname` cannot know whether to add them (TR-431).
        let raw = sign(
            serde_json::json!({
                "scheme": "oauth1",
                "params": {
                    "method": "POST", "scheme": "http", "host": "::1", "port": 8080,
                    "path": "/request", "formBody": "c2=&a3=2+q",
                    "consumerKey": "ck", "consumerSecret": "cs",
                    "signatureMethod": "HMAC-SHA1", "nonce": "n", "timestamp": "1"
                }
            })
            .to_string(),
        )
        .await;
        assert!(raw.contains("oauth_signature="), "{raw}");

        // An unsupported signature method is refused BY NAME, never
        // downgraded to HMAC-SHA1 (TR-409).
        let raw = sign(
            serde_json::json!({
                "scheme": "oauth1",
                "params": {
                    "method": "GET", "scheme": "https", "host": "x.test", "path": "/",
                    "consumerKey": "ck", "consumerSecret": "cs",
                    "signatureMethod": "RSA-SHA1", "nonce": "n", "timestamp": "1"
                }
            })
            .to_string(),
        )
        .await;
        assert!(raw.starts_with("HTTP/1.1 400"), "{raw}");
        assert!(raw.contains("RSA-SHA1"), "{raw}");

        // EdgeGrid: the signature covers method, url, the NAMED headers and
        // the body, so the whole lot goes over the wire. Nonce and timestamp
        // are pinned here to make the header reproducible; omitting them
        // generates both, which is what a real caller wants.
        let raw = sign(
            serde_json::json!({
                "scheme": "akamai-edgegrid",
                "params": {
                    "method": "GET", "url": "https://akaa-x.luna.akamaiapis.net/diagnostic/v1/x",
                    "clientToken": "ct", "accessToken": "at", "clientSecret": "cs",
                    "nonce": "nnn", "timestamp": "20260909T12:00:00+0000"
                }
            })
            .to_string(),
        )
        .await;
        assert!(raw.contains("EG1-HMAC-SHA256 "), "{raw}");
        assert!(raw.contains("client_token=ct"), "{raw}");
        assert!(raw.contains("nonce=nnn"), "{raw}");
        // The secret must never appear in a response the caller logs.
        assert!(
            !raw.contains("cs\""),
            "the client secret must not echo: {raw}"
        );

        // A missing credential is a NAMED 400. Three opaque tokens are easy
        // to paste into the wrong field and Akamai's 401 will not say which.
        let raw = sign(
            serde_json::json!({
                "scheme": "akamai-edgegrid",
                "params": {"method": "GET", "url": "https://x/y", "clientToken": "ct"}
            })
            .to_string(),
        )
        .await;
        assert!(raw.starts_with("HTTP/1.1 400"), "{raw}");
        assert!(raw.contains("access_token"), "{raw}");
        assert!(raw.contains("client_secret"), "{raw}");

        // WSSE returns BOTH headers of the UsernameToken profile: the token
        // on X-WSSE and the profile marker on Authorization. Returning one
        // would serve half the servers implementing it, and the caller has
        // no way to tell which half.
        let raw = sign(
            serde_json::json!({
                "scheme": "wsse",
                "params": {"username": "u", "password": "p", "nonce": "n", "created": "2026-01-01T00:00:00Z"}
            })
            .to_string(),
        )
        .await;
        assert!(raw.contains("X-WSSE"), "{raw}");
        assert!(raw.contains("UsernameToken"), "{raw}");
        assert!(raw.contains("PasswordDigest"), "{raw}");

        // An unknown scheme is refused, not silently unsigned — sending a
        // request the config calls authenticated with no Authorization is
        // invariant #7's silent data loss.
        let raw = sign(serde_json::json!({"scheme": "ntlm", "params": {}}).to_string()).await;
        assert!(raw.starts_with("HTTP/1.1 400"), "{raw}");
        assert!(raw.contains("unknown auth scheme"), "{raw}");
        // And the refusal names what IS supported, from the one declaration.
        // The hardcoded list drifted the moment a scheme was added: it still
        // read "digest, hawk, awsSigV4, oauth1" while both arms above worked.
        for scheme in AUTH_SIGN_SCHEMES {
            assert!(
                raw.contains(scheme),
                "the refusal must name {scheme}: {raw}"
            );
        }
    }
    /// TR-446: `POST /script`, over the socket.
    ///
    /// This is KT-203's `run_script`. What it must prove is that the desktop
    /// tier gets the SAME realm a load run uses — a script behaving one way
    /// in the app and another under load is the divergence this whole
    /// workstream exists to remove.
    #[tokio::test]
    async fn the_script_endpoint_runs_in_the_same_realm_as_a_load_run() {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let port = listener.local_addr().expect("addr").port();
        let state = Arc::new(AgentState {
            token: None,
            client: tropel_http::HttpClient::new(&tropel_http::config::HttpConfig::default())
                .expect("http client"),
            runs: std::sync::Mutex::new(HashMap::new()),
            // No browser origin: these drive the socket directly, and an
            // empty allowlist is the default a real agent starts with.
            allowed_origins: vec![],
        });
        tokio::spawn(async move {
            while let Ok((mut sock, _)) = listener.accept().await {
                let st = state.clone();
                tokio::spawn(async move {
                    let _ = handle_connection(&mut sock, st).await;
                });
            }
        });
        let run = |body: String| async move {
            let mut s = TcpStream::connect(("127.0.0.1", port))
                .await
                .expect("connect");
            let req = format!(
                "POST /script HTTP/1.1\r\nHost: localhost\r\nContent-Length: {}\r\n\r\n{body}",
                body.len()
            );
            s.write_all(req.as_bytes()).await.expect("write");
            let mut out = Vec::new();
            s.read_to_end(&mut out).await.expect("read");
            String::from_utf8_lossy(&out).to_string()
        };

        // Checks are reported individually, not just counted — the results
        // panel names each one.
        let raw = run(serde_json::json!({
            "code": r#"
                    pm.test("passes", function () { pm.expect(1).to.eql(1); });
                    pm.test("fails", function () { pm.expect(1).to.eql(2); });
                "#
        })
        .to_string())
        .await;
        assert!(raw.contains(r#""name":"passes""#), "{raw}");
        assert!(raw.contains(r#""name":"fails""#), "{raw}");
        assert!(raw.contains(r#""passed":1"#), "{raw}");
        assert!(raw.contains(r#""failed":1"#), "{raw}");

        // Variable MUTATIONS come back — the agent holds no session, so the
        // caller merges them into its own scope. Without this the desktop
        // tier could run a pre-request script and lose everything it set.
        let raw = run(serde_json::json!({
            "code": r#"pm.environment.set("token", "abc123");"#,
            "environment": {"seeded": "yes"}
        })
        .to_string())
        .await;
        assert!(raw.contains(r#""token":"abc123""#), "{raw}");
        assert!(
            raw.contains(r#""seeded":"yes""#),
            "seed must survive: {raw}"
        );

        // A THROWING script is a RESULT, not a 500 — the caller needs the
        // message and whatever ran before the throw, exactly as the in-app
        // runner reports it.
        let raw = run(serde_json::json!({
            "code": r#"pm.test("ran", function () {}); throw new Error("boom");"#
        })
        .to_string())
        .await;
        assert!(raw.starts_with("HTTP/1.1 200"), "{raw}");
        assert!(raw.contains("scriptError"), "{raw}");
        assert!(raw.contains("boom"), "{raw}");
        assert!(
            raw.contains(r#""name":"ran""#),
            "what ran before the throw survives: {raw}"
        );

        // Each call gets a FRESH realm: a global left by one script must not
        // be visible to the next. A shared context would let one request
        // change the next one's behaviour — a bug that reproduces only under
        // a specific ordering.
        let _ = run(serde_json::json!({"code": "globalThis.__leak = 1;"}).to_string()).await;
        let raw = run(serde_json::json!({
            "code": r#"pm.test("isolated", function () {
                    pm.expect(typeof globalThis.__leak).to.eql("undefined");
                });"#
        })
        .to_string())
        .await;
        assert!(
            raw.contains(r#""passed":1"#),
            "realms must not share globals: {raw}"
        );
    }
    /// TR-447: `POST /auth/oauth2`, over the socket.
    ///
    /// The last arm of the gap `native-agent.ts` documents. What this must
    /// prove is that the desktop tier gets the SAME OAuth2/JWT/WSSE the
    /// browser does — D4 names signing specifically ("a signing
    /// byte-difference is a 403 that takes a day to find").
    #[tokio::test]
    async fn the_oauth2_endpoint_serves_the_whole_family() {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let port = listener.local_addr().expect("addr").port();
        let state = Arc::new(AgentState {
            token: None,
            client: tropel_http::HttpClient::new(&tropel_http::config::HttpConfig::default())
                .expect("http client"),
            runs: std::sync::Mutex::new(HashMap::new()),
            // No browser origin: these drive the socket directly, and an
            // empty allowlist is the default a real agent starts with.
            allowed_origins: vec![],
        });
        tokio::spawn(async move {
            while let Ok((mut sock, _)) = listener.accept().await {
                let st = state.clone();
                tokio::spawn(async move {
                    let _ = handle_connection(&mut sock, st).await;
                });
            }
        });
        let call = |op: &'static str, params: serde_json::Value| async move {
            let body = serde_json::json!({ "op": op, "params": params }).to_string();
            let mut s = TcpStream::connect(("127.0.0.1", port))
                .await
                .expect("connect");
            let req = format!(
                "POST /auth/oauth2 HTTP/1.1\r\nHost: localhost\r\nContent-Length: {}\r\n\r\n{body}",
                body.len()
            );
            s.write_all(req.as_bytes()).await.expect("write");
            let mut out = Vec::new();
            s.read_to_end(&mut out).await.expect("read");
            String::from_utf8_lossy(&out).to_string()
        };

        // PKCE: the challenge must be computed HERE, so it matches the token
        // request the same tier builds. A client deriving it separately is
        // how a verifier and challenge stop agreeing.
        let raw = call(
            "codeChallengeS256",
            serde_json::json!({"verifier": "abc123"}),
        )
        .await;
        assert!(raw.contains("codeChallenge"), "{raw}");
        assert!(raw.contains(r#""codeChallengeMethod":"S256""#), "{raw}");

        // JWT signing, and the algorithm is NEVER downgraded — a config
        // saying HS512 while the wire carries HS256 is the TR-004/TR-409
        // shape.
        let raw = call(
            "signJwt",
            serde_json::json!({"payload": {"sub": "u1"}, "algorithm": "HS512", "secret": "s"}),
        )
        .await;
        assert!(raw.contains(r#""token":"#), "{raw}");
        let raw = call(
            "signJwt",
            serde_json::json!({"payload": {"sub": "u1"}, "algorithm": "RS256", "secret": "s"}),
        )
        .await;
        assert!(raw.starts_with("HTTP/1.1 400"), "{raw}");
        assert!(raw.contains("RS256"), "refused by name: {raw}");

        // Token placement is the Rust's vocabulary. An unknown value must be
        // REFUSED, not defaulted to header — defaulting silently stops a
        // token reaching a query-auth API.
        let raw = call(
            "attachToken",
            serde_json::json!({"token": "t", "placement": "cookie"}),
        )
        .await;
        assert!(raw.starts_with("HTTP/1.1 400"), "{raw}");
        assert!(raw.contains("unknown token placement"), "{raw}");

        let raw = call(
            "attachToken",
            serde_json::json!({"token": "t", "tokenType": "Bearer", "placement": "header"}),
        )
        .await;
        assert!(raw.contains("Bearer"), "{raw}");

        // WSSE, and an unknown op is refused listing what IS supported.
        let raw = call(
            "wsseSign",
            serde_json::json!({"username": "u", "password": "p"}),
        )
        .await;
        assert!(raw.starts_with("HTTP/1.1 200"), "{raw}");
        let raw = call("nope", serde_json::json!({})).await;
        assert!(raw.starts_with("HTTP/1.1 400"), "{raw}");
        assert!(raw.contains("unknown oauth2 op"), "{raw}");
        assert!(raw.contains("signJwt"), "it lists the alternatives: {raw}");
    }
    /// TR-448: `POST /resolve/batch`.
    ///
    /// The endpoint that lets the DESKTOP tier use the agent instead of
    /// shipping a second copy of this Rust as wasm. Per-call resolution is 33
    /// loopback round trips per request — 2.3 ms measured, against 0.07 ms
    /// batched.
    #[tokio::test]
    async fn the_batch_resolve_endpoint_preserves_order_and_isolates_failures() {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let port = listener.local_addr().expect("addr").port();
        let state = Arc::new(AgentState {
            token: None,
            client: tropel_http::HttpClient::new(&tropel_http::config::HttpConfig::default())
                .expect("http client"),
            runs: std::sync::Mutex::new(HashMap::new()),
            // No browser origin: these drive the socket directly, and an
            // empty allowlist is the default a real agent starts with.
            allowed_origins: vec![],
        });
        tokio::spawn(async move {
            while let Ok((mut sock, _)) = listener.accept().await {
                let st = state.clone();
                tokio::spawn(async move {
                    let _ = handle_connection(&mut sock, st).await;
                });
            }
        });
        let post = |body: String| async move {
            let mut s = TcpStream::connect(("127.0.0.1", port))
                .await
                .expect("connect");
            let req = format!(
                "POST /resolve/batch HTTP/1.1\r\nHost: localhost\r\nContent-Length: {}\r\n\r\n{body}",
                body.len()
            );
            s.write_all(req.as_bytes()).await.expect("write");
            let mut out = Vec::new();
            s.read_to_end(&mut out).await.expect("read");
            String::from_utf8_lossy(&out).to_string()
        };

        let raw = post(
            serde_json::json!({
                "variables": {"base": "https://api.test", "tok": "abc", "n": "2"},
                "items": [
                    {"template": "{{base}}/v{{n}}"},
                    {"template": "Bearer {{tok}}"},
                    {"template": "{\"t\":\"{{tok}}\"}", "mode": "json"},
                    // A bad mode: this item fails, the others must not.
                    {"template": "{{tok}}", "mode": "nope"},
                    {"template": "{{base}}"}
                ]
            })
            .to_string(),
        )
        .await;

        let body = raw.split("\r\n\r\n").nth(1).unwrap_or_default().to_string();
        let parsed: serde_json::Value = serde_json::from_str(&body).expect("json body");
        let items = parsed["items"].as_array().expect("items array");

        // ORDER IS THE CONTRACT — the caller re-assembles its request by index,
        // so a reordered or short reply would put a header's value in a param.
        assert_eq!(items.len(), 5, "one output per input, always: {body}");
        assert_eq!(items[0]["value"], "https://api.test/v2");
        assert_eq!(items[1]["value"], "Bearer abc");
        assert_eq!(items[2]["value"], "{\"t\":\"abc\"}");
        // A per-item failure does NOT fail the batch: one bad escape mode must
        // not lose the other four resolutions.
        assert!(
            items[3]["error"].is_string(),
            "item 3 should carry an error: {body}"
        );
        assert!(items[3]["value"].is_null());
        assert_eq!(
            items[4]["value"], "https://api.test",
            "later items still resolve"
        );

        // TR-449: a CYCLE and an UNKNOWN NAME both leave a literal `{{…}}`,
        // and only the resolver's own loop can tell them apart. KnockPort
        // turns `hitCap` into a failed send and an unresolved name into a
        // visible typo, so collapsing them to a bare string would make a
        // cyclic chain look like a harmless placeholder.
        let raw = post(
            serde_json::json!({
                "variables": {"a": "{{b}}", "b": "{{a}}"},
                "items": [
                    {"template": "{{a}}"},
                    {"template": "{{nosuchvar}}"}
                ]
            })
            .to_string(),
        )
        .await;
        let body = raw.split("\r\n\r\n").nth(1).unwrap_or_default().to_string();
        let parsed: serde_json::Value = serde_json::from_str(&body).expect("json body");
        let items = parsed["items"].as_array().expect("items");
        assert_eq!(
            items[0]["hitCap"], true,
            "a cycle must report hitCap: {body}"
        );
        assert_eq!(
            items[1]["hitCap"], false,
            "an unknown name is NOT a cycle: {body}"
        );
        assert!(
            items[1]["unresolved"]
                .as_array()
                .is_some_and(|u| u.iter().any(|n| n == "nosuchvar")),
            "the unknown name must be reported so the user sees their typo: {body}"
        );

        // TR-449: a shallow item is REFUSED by name. Silently resolving it
        // deep would hand the caller twenty passes when it asked for one,
        // with nothing in the reply to say so — the D4 failure this seam
        // exists to prevent.
        let raw = post(
            serde_json::json!({
                "variables": {"a": "{{b}}", "b": "final"},
                "items": [{"template": "{{a}}", "deep": false}]
            })
            .to_string(),
        )
        .await;
        let body = raw.split("\r\n\r\n").nth(1).unwrap_or_default().to_string();
        let parsed: serde_json::Value = serde_json::from_str(&body).expect("json body");
        let item = &parsed["items"][0];
        assert!(
            item["value"].is_null(),
            "a refused item carries no value: {body}"
        );
        assert!(
            item["error"]
                .as_str()
                .is_some_and(|e| e.contains("deep: false") && e.contains("/resolve")),
            "the refusal must name the field AND the endpoint that serves it: {body}"
        );

        // An empty batch is a valid batch — a request with no templates is not
        // an error, and returning one would make the caller special-case it.
        let raw = post(serde_json::json!({"variables": {}, "items": []}).to_string()).await;
        assert!(raw.starts_with("HTTP/1.1 200"), "{raw}");
        assert!(raw.contains(r#""items":[]"#), "{raw}");
    }
}