#![cfg(not(target_arch = "wasm32"))]
//! Socket tests for the F2 lane: instance/run/value routes over real HTTP,
//! against `FlowServer::start` with fake seams injected via
//! `FlowServerConfig::with_seams` (§12 headless acceptance, adapted to the
//! Gates list in `local/agent_state/flow/briefs/F2-routes.md`).

mod support;

use makepad_ai_hub::discovery::DEFAULT_FLEET;
use makepad_ai_hub::download::Downloader;
use makepad_ai_hub::peer_serve::PeerOptions;
use makepad_ai_hub::registry::{Domain, ModelSpec, Registry};
use makepad_ai_hub::server::{start_service, ServiceConfig};
use makepad_ai_hub::sha256::sha256_hex;
use makepad_flow::engine::FixedGen;
use makepad_flow::host::{FlowServer, FlowServerConfig};
use makepad_flow::{
    BatchMutationResponse, CreateBatchRequest, CreateBatchResponse, CreateInstanceRequest,
    CreateInstanceResponse, CreateRunRequest, CreateRunResponse, EventsResponse, InputValueDto,
    InstanceRow, NodeState, PortType, PutSourceRequest, PutValueResponse, RunRowDto, RunState, Seams,
    SetInputsResponse,
};
use makepad_micro_serde::{DeJson, JsonValue, SerJson};
use std::collections::HashMap;
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpStream};
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use support::{seams, FakeChat, FakeGen, FakeHttp};

struct TempRoot(PathBuf);

impl TempRoot {
    fn new(label: &str) -> Self {
        let nonce = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
        let path = std::env::temp_dir().join(format!(
            "makepad-flow-host-runs-{}-{}-{nonce}",
            std::process::id(),
            label
        ));
        std::fs::create_dir_all(&path).unwrap();
        Self(path)
    }
}

impl Drop for TempRoot {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.0);
    }
}

struct HttpResponse {
    status: u16,
    /// Lossily decoded for the common case (JSON, error messages).
    body: String,
    /// The exact response bytes, for the one route (`GET /v1/values/…`)
    /// whose body is not text.
    bytes: Vec<u8>,
}

fn request(
    address: SocketAddr,
    method: &str,
    target: &str,
    token: Option<&str>,
    body: &str,
) -> HttpResponse {
    let mut stream = TcpStream::connect(address).unwrap();
    stream.set_read_timeout(Some(Duration::from_secs(35))).unwrap();
    let authorization = token
        .map(|token| format!("Authorization: Bearer {token}\r\n"))
        .unwrap_or_default();
    write!(
        stream,
        "{method} {target} HTTP/1.1\r\nHost: localhost\r\n{authorization}Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
        body.len()
    )
    .unwrap();
    let mut raw = Vec::new();
    stream.read_to_end(&mut raw).unwrap();
    let boundary = raw.windows(4).position(|window| window == b"\r\n\r\n").expect("head/body boundary");
    let head = std::str::from_utf8(&raw[..boundary]).expect("response head is not utf-8");
    let bytes = raw[boundary + 4..].to_vec();
    let status = head.lines().next().unwrap().split_whitespace().nth(1).unwrap().parse().unwrap();
    let body = String::from_utf8_lossy(&bytes).into_owned();
    HttpResponse { status, body, bytes }
}

fn start(root: &Path, seams: Seams) -> FlowServer {
    let mut config = FlowServerConfig::new(root.to_path_buf()).with_seams(seams);
    config.watch_interval_ms = 25;
    config.janitor_sweep_secs = 1;
    config.log = Box::new(|_| {});
    FlowServer::start(config).unwrap()
}

fn put_flow(address: SocketAddr, token: &str, name: &str, source: &str) {
    let response = request(
        address,
        "PUT",
        &format!("/v1/flows/{name}"),
        Some(token),
        &PutSourceRequest { source: source.to_string() }.serialize_json(),
    );
    assert_eq!(response.status, 200, "put_flow {name}: {}", response.body);
}

fn cursor(address: SocketAddr, token: &str, topic: &str) -> String {
    let response = request(address, "GET", &format!("/v1/events?topic={topic}"), Some(token), "");
    assert_eq!(response.status, 200, "{}", response.body);
    EventsResponse::deserialize_json(&response.body).unwrap().cursor
}

/// Poll `/v1/events` on `topic` until `stop` returns true on one of the
/// events seen so far, or a 5 s deadline elapses. Returns every event kind
/// string observed, in order.
fn poll_events_until(
    address: SocketAddr,
    token: &str,
    topic: &str,
    mut cursor: String,
    mut stop: impl FnMut(&JsonValue) -> bool,
) -> Vec<JsonValue> {
    let deadline = Instant::now() + Duration::from_secs(5);
    let mut seen = Vec::new();
    loop {
        let response = request(
            address,
            "GET",
            &format!("/v1/events?cursor={cursor}&wait=500&topic={topic}"),
            Some(token),
            "",
        );
        assert_eq!(response.status, 200, "{}", response.body);
        let page = EventsResponse::deserialize_json(&response.body).unwrap();
        cursor = page.cursor;
        for event in page.events {
            let done = stop(&event);
            seen.push(event);
            if done {
                return seen;
            }
        }
        assert!(Instant::now() < deadline, "event deadline elapsed; saw {seen:#?}");
    }
}

fn event_kind(event: &JsonValue) -> &str {
    event.key("kind").and_then(JsonValue::string).map(String::as_str).unwrap_or("")
}

fn event_str<'a>(event: &'a JsonValue, key: &str) -> Option<&'a str> {
    event.key(key).and_then(JsonValue::string).map(String::as_str)
}

/// `RunState`/`NodeState`/`PortType` are fieldless enums; this crate's
/// `SerJson` derive still wraps them as a single-key object (`{"Done":[]}`)
/// rather than a bare string, so raw-event assertions match on the key.
fn variant_name(value: Option<&JsonValue>) -> Option<&str> {
    match value {
        Some(JsonValue::Object(fields)) if fields.len() == 1 => {
            fields.keys().next().map(String::as_str)
        }
        _ => None,
    }
}

fn text_input(text: &str) -> InputValueDto {
    InputValueDto { ty: PortType::Text, text: Some(text.to_string()), json: None, digest: None }
}

fn image_digest_input(digest: &str) -> InputValueDto {
    InputValueDto {
        ty: PortType::Image,
        text: None,
        json: None,
        digest: Some(digest.to_string()),
    }
}

fn one_input(node: &str, port: &str, value: InputValueDto) -> HashMap<String, HashMap<String, InputValueDto>> {
    let mut ports = HashMap::new();
    ports.insert(port.to_string(), value);
    let mut inputs = HashMap::new();
    inputs.insert(node.to_string(), ports);
    inputs
}

const ECHO_FLOW: &str = r#"use mod.flow.*
let prompt = Input{type: @text value: "x"}
let echo = Fn{in: {text: prompt.text()} out: [@text] run: |i| {{text: i.text}}}
let result = Output{type: @text value: echo.text()}
Flow{prompt, echo, result}
"#;

/// `expand` never finishes while the injected `FakeChat` is `pending`, so a
/// run on this flow stays `running` until cancelled — exactly what the
/// concurrency-queue and cancel tests need to observe deterministically.
const STALLED_FLOW: &str = r#"use mod.flow.*
let prompt = Input{type: @text default: "x"}
let expand = Llm{prompt: prompt.text()}
let result = Output{type: @text value: expand.text()}
Flow{concurrency: 1 prompt, expand, result}
"#;

const ASK_FLOW: &str = r#"use mod.flow.*
let which = Ask{question: "Which?" type: @text options: ["first", "second"]}
let echo = Fn{in: {text: which.text()} out: [@text] run: |i| {{text: i.text}}}
let result = Output{type: @text value: echo.text()}
Flow{which, echo, result}
"#;

const IMAGE_ECHO_FLOW: &str = r#"use mod.flow.*
let picture = Input{type: @image default: nil}
let result = Output{type: @image value: picture.image()}
Flow{picture, result}
"#;

#[test]
fn design_value_defaults_and_external_overrides_never_rewrite_the_file() {
    let root = TempRoot::new("design-value-override");
    let server = start(
        &root.0,
        seams(
            FakeChat::done("unused"),
            FakeGen::done(),
            FakeHttp::json(200, "{}"),
        ),
    );
    let endpoints = server.endpoints();
    put_flow(endpoints.control, &endpoints.token, "echo", ECHO_FLOW);
    let path = root.0.join("flows/echo.splash");
    let before = std::fs::read_to_string(&path).unwrap();
    assert!(before.contains("value: \"x\""));

    let design = request(
        endpoints.control,
        "POST",
        "/v1/flows/echo/instances",
        Some(&endpoints.token),
        &CreateInstanceRequest::default().serialize_json(),
    );
    assert_eq!(design.status, 201, "{}", design.body);
    let design_id = CreateInstanceResponse::deserialize_json(&design.body)
        .unwrap()
        .instance;
    let design_row = request(
        endpoints.control,
        "GET",
        &format!("/v1/instances/{design_id}"),
        Some(&endpoints.token),
        "",
    );
    assert_eq!(
        InstanceRow::deserialize_json(&design_row.body)
            .unwrap()
            .input_text("prompt", "text")
            .as_deref(),
        Some("x")
    );

    let external = request(
        endpoints.control,
        "POST",
        "/v1/flows/echo/instances",
        Some(&endpoints.token),
        &CreateInstanceRequest {
            inputs: Some(one_input("prompt", "text", text_input("external"))),
            ..CreateInstanceRequest::default()
        }
        .serialize_json(),
    );
    assert_eq!(external.status, 201, "{}", external.body);
    let external_id = CreateInstanceResponse::deserialize_json(&external.body)
        .unwrap()
        .instance;
    let external_row = request(
        endpoints.control,
        "GET",
        &format!("/v1/instances/{external_id}"),
        Some(&endpoints.token),
        "",
    );
    assert_eq!(
        InstanceRow::deserialize_json(&external_row.body)
            .unwrap()
            .input_text("prompt", "text")
            .as_deref(),
        Some("external")
    );

    let run_cursor = cursor(endpoints.control, &endpoints.token, "run");
    let started = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{external_id}/runs"),
        Some(&endpoints.token),
        &CreateRunRequest::default().serialize_json(),
    );
    assert_eq!(started.status, 202, "{}", started.body);
    let run_id = CreateRunResponse::deserialize_json(&started.body)
        .unwrap()
        .run_id;
    poll_events_until(
        endpoints.control,
        &endpoints.token,
        "run",
        run_cursor,
        |event| {
            event_kind(event) == "run.finished"
                && event_str(event, "run_id") == Some(run_id.as_str())
        },
    );

    assert_eq!(std::fs::read_to_string(path).unwrap(), before);
    server.shutdown();
}

// ---------------------------------------------------------------------------
// 1. §12 headless acceptance: the full instance/run/value pipeline
// ---------------------------------------------------------------------------

#[test]
fn full_run_streams_the_design_pipeline_and_serves_the_picture() {
    let root = TempRoot::new("pipeline");
    let chat = FakeChat::done("a moody paragraph");
    let chat_requests = chat.requests.clone();
    let server = start(
        &root.0,
        seams(chat, FakeGen::done(), FakeHttp::json(200, "{}")),
    );
    let endpoints = server.endpoints();
    put_flow(
        endpoints.control,
        &endpoints.token,
        "demo",
        include_str!("fixtures/prompt_image.splash"),
    );

    let create = request(
        endpoints.control,
        "POST",
        "/v1/flows/demo/instances",
        Some(&endpoints.token),
        &CreateInstanceRequest::default().serialize_json(),
    );
    assert_eq!(create.status, 201, "{}", create.body);
    let instance = CreateInstanceResponse::deserialize_json(&create.body).unwrap().instance;

    let set_inputs = request(
        endpoints.control,
        "PUT",
        &format!("/v1/instances/{instance}/inputs"),
        Some(&endpoints.token),
        &one_input("prompt", "text", text_input("a lighthouse at dusk")).serialize_json(),
    );
    assert_eq!(set_inputs.status, 200, "{}", set_inputs.body);

    let run_cursor = cursor(endpoints.control, &endpoints.token, "run");
    let start_run = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{instance}/runs"),
        Some(&endpoints.token),
        &CreateRunRequest::default().serialize_json(),
    );
    assert_eq!(start_run.status, 202, "{}", start_run.body);
    let run_id = CreateRunResponse::deserialize_json(&start_run.body).unwrap().run_id;
    assert_eq!(CreateRunResponse::deserialize_json(&start_run.body).unwrap().queued, 0);

    let events = poll_events_until(endpoints.control, &endpoints.token, "run", run_cursor, |event| {
        event_kind(event) == "run.finished" && event_str(event, "run_id") == Some(run_id.as_str())
    });
    assert_eq!(
        chat_requests.lock().unwrap().as_slice(),
        &[(
            "Rewrite the prompt as one vivid paragraph for an image model.\n             \
             Keep the subject. Add light, lens, material, mood. No lists."
                .to_string(),
            "a lighthouse at dusk".to_string(),
            String::new(),
            None,
            Some(false),
        )]
    );
    let kinds: Vec<&str> = events.iter().map(event_kind).collect();

    // DESIGN.md §12's exact sequence, with the granularity the brief lists.
    assert_eq!(kinds.first(), Some(&"run.started"));
    assert!(kinds.contains(&"node.done"), "{kinds:#?}");
    assert!(kinds.contains(&"node.started"), "{kinds:#?}");
    assert!(kinds.contains(&"node.delta"), "{kinds:#?}");
    assert_eq!(kinds.last(), Some(&"run.finished"));

    let node_done_events: Vec<&JsonValue> =
        events.iter().filter(|event| event_kind(event) == "node.done").collect();
    assert!(
        node_done_events.iter().any(|event| event_str(event, "node") == Some("prompt")),
        "{kinds:#?}"
    );
    assert!(
        node_done_events.iter().any(|event| event_str(event, "node") == Some("expand")),
        "{kinds:#?}"
    );
    assert!(
        node_done_events.iter().any(|event| event_str(event, "node") == Some("styled")),
        "{kinds:#?}"
    );
    let image_done = node_done_events
        .iter()
        .find(|event| event_str(event, "node") == Some("image"))
        .unwrap_or_else(|| panic!("no node.done(image): {kinds:#?}"));
    let outputs = match image_done.key("outputs") {
        Some(JsonValue::Array(items)) => items,
        other => panic!("expected outputs array, got {other:?}"),
    };
    let picture_port = outputs
        .iter()
        .find(|entry| entry.key("port").and_then(JsonValue::string).map(String::as_str) == Some("image"))
        .expect("an `image` output port");
    let picture_value = picture_port.key("value").expect("output port value");
    assert_eq!(
        picture_value.key("content_type").and_then(JsonValue::string).map(String::as_str),
        Some("image/png")
    );
    let digest =
        picture_value.key("digest").and_then(JsonValue::string).cloned().expect("digest field");

    let finished = events.last().unwrap();
    assert_eq!(variant_name(finished.key("state")), Some("Done"));

    let fetched = request(endpoints.data, "GET", &format!("/v1/values/{digest}"), Some(&endpoints.token), "");
    assert_eq!(fetched.status, 200, "{}", fetched.body);
    assert_eq!(sha256_hex(&fetched.bytes), digest.trim_start_matches("sha256:"));

    let get_run = request(endpoints.control, "GET", &format!("/v1/runs/{run_id}"), Some(&endpoints.token), "");
    assert_eq!(get_run.status, 200, "{}", get_run.body);
    let run_row = RunRowDto::deserialize_json(&get_run.body).unwrap();
    assert!(run_row.outputs.contains_key("picture"), "{run_row:#?}");
    assert_eq!(run_row.outputs["picture"].digest.trim_start_matches("sha256:"), digest);

    // Same instance, same inputs -> the fake gen's fixed bytes dedupe to the
    // same digest, exercising the value-store path a second time.
    let run_cursor2 = cursor(endpoints.control, &endpoints.token, "run");
    let second = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{instance}/runs"),
        Some(&endpoints.token),
        &CreateRunRequest::default().serialize_json(),
    );
    assert_eq!(second.status, 202, "{}", second.body);
    let second_run_id = CreateRunResponse::deserialize_json(&second.body).unwrap().run_id;
    poll_events_until(endpoints.control, &endpoints.token, "run", run_cursor2, |event| {
        event_kind(event) == "run.finished" && event_str(event, "run_id") == Some(second_run_id.as_str())
    });
    let second_row = request(endpoints.control, "GET", &format!("/v1/runs/{second_run_id}"), Some(&endpoints.token), "");
    let second_row = RunRowDto::deserialize_json(&second_row.body).unwrap();
    assert_eq!(second_row.outputs["picture"].digest, digest);

    server.shutdown();
}

// ---------------------------------------------------------------------------
// 2. concurrent instances with different outputs; concurrency queues a run
// ---------------------------------------------------------------------------

#[test]
fn two_instances_finish_with_different_digests() {
    let root = TempRoot::new("concurrent");
    let server = start(&root.0, seams(FakeChat::done("unused"), FakeGen::done(), FakeHttp::json(200, "{}")));
    let endpoints = server.endpoints();
    put_flow(endpoints.control, &endpoints.token, "echo", ECHO_FLOW);

    let mut digests = Vec::new();
    for prompt in ["alpha", "beta"] {
        let create = request(
            endpoints.control,
            "POST",
            "/v1/flows/echo/instances",
            Some(&endpoints.token),
            &CreateInstanceRequest::default().serialize_json(),
        );
        let instance = CreateInstanceResponse::deserialize_json(&create.body).unwrap().instance;
        let set_inputs = request(
            endpoints.control,
            "PUT",
            &format!("/v1/instances/{instance}/inputs"),
            Some(&endpoints.token),
            &one_input("prompt", "text", text_input(prompt)).serialize_json(),
        );
        assert_eq!(set_inputs.status, 200, "{}", set_inputs.body);
        let run_cursor = cursor(endpoints.control, &endpoints.token, "run");
        let start_run = request(
            endpoints.control,
            "POST",
            &format!("/v1/instances/{instance}/runs"),
            Some(&endpoints.token),
            &CreateRunRequest::default().serialize_json(),
        );
        assert_eq!(start_run.status, 202, "{}", start_run.body);
        let run_id = CreateRunResponse::deserialize_json(&start_run.body).unwrap().run_id;
        poll_events_until(endpoints.control, &endpoints.token, "run", run_cursor, |event| {
            event_kind(event) == "run.finished" && event_str(event, "run_id") == Some(run_id.as_str())
        });
        let row = request(endpoints.control, "GET", &format!("/v1/runs/{run_id}"), Some(&endpoints.token), "");
        let row = RunRowDto::deserialize_json(&row.body).unwrap();
        digests.push(row.outputs["result"].digest.clone());
    }
    assert_ne!(digests[0], digests[1]);

    server.shutdown();
}

#[test]
fn concurrency_one_queues_the_second_run() {
    let root = TempRoot::new("queue");
    let server = start(
        &root.0,
        seams(
            FakeChat { pending: true, ..FakeChat::done("unused") },
            FakeGen::done(),
            FakeHttp::json(200, "{}"),
        ),
    );
    let endpoints = server.endpoints();
    put_flow(endpoints.control, &endpoints.token, "stalled", STALLED_FLOW);
    let create = request(
        endpoints.control,
        "POST",
        "/v1/flows/stalled/instances",
        Some(&endpoints.token),
        &CreateInstanceRequest::default().serialize_json(),
    );
    let instance = CreateInstanceResponse::deserialize_json(&create.body).unwrap().instance;

    let first = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{instance}/runs"),
        Some(&endpoints.token),
        &CreateRunRequest::default().serialize_json(),
    );
    assert_eq!(first.status, 202, "{}", first.body);
    assert_eq!(CreateRunResponse::deserialize_json(&first.body).unwrap().queued, 0);

    let second = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{instance}/runs"),
        Some(&endpoints.token),
        &CreateRunRequest::default().serialize_json(),
    );
    assert_eq!(second.status, 202, "{}", second.body);
    assert_eq!(CreateRunResponse::deserialize_json(&second.body).unwrap().queued, 1);

    let get = request(endpoints.control, "GET", &format!("/v1/instances/{instance}"), Some(&endpoints.token), "");
    let row = InstanceRow::deserialize_json(&get.body).unwrap();
    assert_eq!(row.state, "running");

    server.shutdown();
}

// ---------------------------------------------------------------------------
// 3. the Ask flow: parks, is listed waiting, is answered, wrong node is 409
// ---------------------------------------------------------------------------

#[test]
fn ask_flow_parks_lists_waiting_and_is_answered() {
    let root = TempRoot::new("ask");
    let server = start(&root.0, seams(FakeChat::done("unused"), FakeGen::done(), FakeHttp::json(200, "{}")));
    let endpoints = server.endpoints();
    put_flow(endpoints.control, &endpoints.token, "ask", ASK_FLOW);
    let create = request(
        endpoints.control,
        "POST",
        "/v1/flows/ask/instances",
        Some(&endpoints.token),
        &CreateInstanceRequest::default().serialize_json(),
    );
    let instance = CreateInstanceResponse::deserialize_json(&create.body).unwrap().instance;

    // The Ask node exists but nothing is waiting there yet: 409.
    let too_early = request(
        endpoints.control,
        "PUT",
        &format!("/v1/instances/{instance}/inputs"),
        Some(&endpoints.token),
        &one_input("which", "text", text_input("first")).serialize_json(),
    );
    assert_eq!(too_early.status, 409, "{}", too_early.body);

    let instance_cursor = cursor(endpoints.control, &endpoints.token, "instance");
    let start_run = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{instance}/runs"),
        Some(&endpoints.token),
        &CreateRunRequest::default().serialize_json(),
    );
    assert_eq!(start_run.status, 202, "{}", start_run.body);
    let run_id = CreateRunResponse::deserialize_json(&start_run.body).unwrap().run_id;

    let waiting_events = poll_events_until(endpoints.control, &endpoints.token, "instance", instance_cursor, |event| {
        event_kind(event) == "node.waiting"
    });
    assert!(waiting_events.iter().any(|event| event_kind(event) == "node.waiting"), "{waiting_events:#?}");

    let waiting = request(endpoints.control, "GET", "/v1/instances?waiting=1", Some(&endpoints.token), "");
    let rows = Vec::<InstanceRow>::deserialize_json(&waiting.body).unwrap();
    assert!(rows.iter().any(|row| row.instance == instance), "{rows:#?}");
    let row = rows.iter().find(|row| row.instance == instance).unwrap();
    assert_eq!(row.waiting.as_ref().unwrap().node, "which");

    let run_cursor = cursor(endpoints.control, &endpoints.token, "run");
    let answer = request(
        endpoints.control,
        "PUT",
        &format!("/v1/instances/{instance}/inputs?actor=tab"),
        Some(&endpoints.token),
        &one_input("which", "text", text_input("first")).serialize_json(),
    );
    assert_eq!(answer.status, 200, "{}", answer.body);

    let events = poll_events_until(endpoints.control, &endpoints.token, "run", run_cursor, |event| {
        event_kind(event) == "run.finished" && event_str(event, "run_id") == Some(run_id.as_str())
    });
    let answered = events
        .iter()
        .find(|event| event_kind(event) == "node.answered")
        .unwrap_or_else(|| panic!("no node.answered: {events:#?}"));
    assert_eq!(event_str(answered, "by"), Some("tab"));
    assert_eq!(events.last().map(event_kind), Some("run.finished"));

    server.shutdown();
}

// ---------------------------------------------------------------------------
// 4. flow.changed re-evaluates a live instance, a pinned one ignores it;
//    flow.removed drops instances
// ---------------------------------------------------------------------------

#[test]
fn flow_changed_updates_live_instances_and_flow_removed_drops_them() {
    let root = TempRoot::new("changed");
    let server = start(&root.0, seams(FakeChat::done("unused"), FakeGen::done(), FakeHttp::json(200, "{}")));
    let endpoints = server.endpoints();
    put_flow(endpoints.control, &endpoints.token, "echo", ECHO_FLOW);

    let live = request(
        endpoints.control,
        "POST",
        "/v1/flows/echo/instances",
        Some(&endpoints.token),
        &CreateInstanceRequest::default().serialize_json(),
    );
    let live_id = CreateInstanceResponse::deserialize_json(&live.body).unwrap().instance;

    let pinned = request(
        endpoints.control,
        "POST",
        "/v1/flows/echo/instances",
        Some(&endpoints.token),
        &CreateInstanceRequest { pin: Some(true), ..CreateInstanceRequest::default() }.serialize_json(),
    );
    let pinned_id = CreateInstanceResponse::deserialize_json(&pinned.body).unwrap().instance;

    let changed_source = r#"use mod.flow.*
let prompt = Input{type: @text default: "x"}
let extra = Input{type: @text default: "new"}
let echo = Fn{in: {text: prompt.text()} out: [@text] run: |i| {{text: i.text}}}
let result = Output{type: @text value: echo.text()}
Flow{prompt, extra, echo, result}
"#;
    put_flow(endpoints.control, &endpoints.token, "echo", changed_source);

    let live_row = request(endpoints.control, "GET", &format!("/v1/instances/{live_id}"), Some(&endpoints.token), "");
    let live_row = InstanceRow::deserialize_json(&live_row.body).unwrap();
    assert_eq!(live_row.revision, 2);
    assert!(live_row.inputs.contains_key("extra"), "{live_row:#?}");
    assert!(live_row.inputs.contains_key("prompt"), "{live_row:#?}");

    let pinned_row =
        request(endpoints.control, "GET", &format!("/v1/instances/{pinned_id}"), Some(&endpoints.token), "");
    let pinned_row = InstanceRow::deserialize_json(&pinned_row.body).unwrap();
    assert_eq!(pinned_row.revision, 1);
    assert!(!pinned_row.inputs.contains_key("extra"), "{pinned_row:#?}");

    let delete = request(endpoints.control, "DELETE", "/v1/flows/echo", Some(&endpoints.token), "");
    assert_eq!(delete.status, 204, "{}", delete.body);

    let gone_live = request(endpoints.control, "GET", &format!("/v1/instances/{live_id}"), Some(&endpoints.token), "");
    assert_eq!(gone_live.status, 404);
    let gone_pinned =
        request(endpoints.control, "GET", &format!("/v1/instances/{pinned_id}"), Some(&endpoints.token), "");
    assert_eq!(gone_pinned.status, 404);

    server.shutdown();
}

// ---------------------------------------------------------------------------
// 5. DELETE mid-run and POST cancel both finish the run cancelled
// ---------------------------------------------------------------------------

#[test]
fn delete_instance_and_cancel_route_both_cancel_the_run() {
    let root = TempRoot::new("cancel");
    let server = start(
        &root.0,
        seams(
            FakeChat { pending: true, ..FakeChat::done("unused") },
            FakeGen::done(),
            FakeHttp::json(200, "{}"),
        ),
    );
    let endpoints = server.endpoints();
    put_flow(endpoints.control, &endpoints.token, "stalled", STALLED_FLOW);

    // DELETE mid-run.
    let create = request(
        endpoints.control,
        "POST",
        "/v1/flows/stalled/instances",
        Some(&endpoints.token),
        &CreateInstanceRequest::default().serialize_json(),
    );
    let instance = CreateInstanceResponse::deserialize_json(&create.body).unwrap().instance;
    let run_cursor = cursor(endpoints.control, &endpoints.token, "run");
    let start_run = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{instance}/runs"),
        Some(&endpoints.token),
        &CreateRunRequest::default().serialize_json(),
    );
    let run_id = CreateRunResponse::deserialize_json(&start_run.body).unwrap().run_id;
    let delete = request(endpoints.control, "DELETE", &format!("/v1/instances/{instance}"), Some(&endpoints.token), "");
    assert_eq!(delete.status, 204, "{}", delete.body);
    let events = poll_events_until(endpoints.control, &endpoints.token, "run", run_cursor, |event| {
        event_kind(event) == "run.finished" && event_str(event, "run_id") == Some(run_id.as_str())
    });
    let finished = events.last().unwrap();
    assert_eq!(variant_name(finished.key("state")), Some("Cancelled"));

    // POST .../cancel on a second, independent run.
    let create2 = request(
        endpoints.control,
        "POST",
        "/v1/flows/stalled/instances",
        Some(&endpoints.token),
        &CreateInstanceRequest::default().serialize_json(),
    );
    let instance2 = CreateInstanceResponse::deserialize_json(&create2.body).unwrap().instance;
    let run_cursor2 = cursor(endpoints.control, &endpoints.token, "run");
    let start_run2 = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{instance2}/runs"),
        Some(&endpoints.token),
        &CreateRunRequest::default().serialize_json(),
    );
    let run_id2 = CreateRunResponse::deserialize_json(&start_run2.body).unwrap().run_id;
    let cancel = request(endpoints.control, "POST", &format!("/v1/runs/{run_id2}/cancel"), Some(&endpoints.token), "");
    assert_eq!(cancel.status, 200, "{}", cancel.body);
    let events2 = poll_events_until(endpoints.control, &endpoints.token, "run", run_cursor2, |event| {
        event_kind(event) == "run.finished" && event_str(event, "run_id") == Some(run_id2.as_str())
    });
    let finished2 = events2.last().unwrap();
    assert_eq!(variant_name(finished2.key("state")), Some("Cancelled"));

    server.shutdown();
}

#[test]
fn clear_instance_forgets_generated_state_but_keeps_inputs_and_refuses_live_runs() {
    let root = TempRoot::new("clear-instance");
    let server = start(
        &root.0,
        seams(
            FakeChat { pending: true, ..FakeChat::done("unused") },
            FakeGen::done(),
            FakeHttp::json(200, "{}"),
        ),
    );
    let endpoints = server.endpoints();
    put_flow(endpoints.control, &endpoints.token, "echo", ECHO_FLOW);
    put_flow(endpoints.control, &endpoints.token, "stalled-clear", STALLED_FLOW);

    let create = request(
        endpoints.control,
        "POST",
        "/v1/flows/echo/instances",
        Some(&endpoints.token),
        &CreateInstanceRequest {
            inputs: Some(one_input("prompt", "text", text_input("keep me"))),
            ..CreateInstanceRequest::default()
        }
        .serialize_json(),
    );
    let instance = CreateInstanceResponse::deserialize_json(&create.body).unwrap().instance;
    let run_cursor = cursor(endpoints.control, &endpoints.token, "run");
    let started = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{instance}/runs"),
        Some(&endpoints.token),
        &CreateRunRequest::default().serialize_json(),
    );
    let run_id = CreateRunResponse::deserialize_json(&started.body).unwrap().run_id;
    poll_events_until(endpoints.control, &endpoints.token, "run", run_cursor, |event| {
        event_kind(event) == "run.finished" && event_str(event, "run_id") == Some(run_id.as_str())
    });
    let before = request(
        endpoints.control,
        "GET",
        &format!("/v1/instances/{instance}"),
        Some(&endpoints.token),
        "",
    );
    assert!(!InstanceRow::deserialize_json(&before.body).unwrap().outputs.is_empty());

    let instance_cursor = cursor(endpoints.control, &endpoints.token, "instance");
    let clear = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{instance}/clear"),
        Some(&endpoints.token),
        "",
    );
    assert_eq!(clear.status, 200, "{}", clear.body);
    let after = request(
        endpoints.control,
        "GET",
        &format!("/v1/instances/{instance}"),
        Some(&endpoints.token),
        "",
    );
    let after = InstanceRow::deserialize_json(&after.body).unwrap();
    assert_eq!(after.input_text("prompt", "text").as_deref(), Some("keep me"));
    assert!(after.outputs.is_empty());
    assert_eq!(after.state, "idle");
    let forgotten = request(
        endpoints.control,
        "GET",
        &format!("/v1/runs/{run_id}"),
        Some(&endpoints.token),
        "",
    );
    assert_eq!(forgotten.status, 404, "{}", forgotten.body);
    let events = poll_events_until(
        endpoints.control,
        &endpoints.token,
        "instance",
        instance_cursor,
        |event| event_kind(event) == "instance.cleared",
    );
    assert_eq!(event_str(events.last().unwrap(), "instance"), Some(instance.as_str()));

    let create_live = request(
        endpoints.control,
        "POST",
        "/v1/flows/stalled-clear/instances",
        Some(&endpoints.token),
        &CreateInstanceRequest::default().serialize_json(),
    );
    let live = CreateInstanceResponse::deserialize_json(&create_live.body).unwrap().instance;
    let started_live = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{live}/runs"),
        Some(&endpoints.token),
        &CreateRunRequest::default().serialize_json(),
    );
    assert_eq!(started_live.status, 202, "{}", started_live.body);
    let busy = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{live}/clear"),
        Some(&endpoints.token),
        "",
    );
    assert_eq!(busy.status, 409, "{}", busy.body);

    server.shutdown();
}

// ---------------------------------------------------------------------------
// 6. values: PUT bytes -> digest -> usable as an image input; TTL expiry
// ---------------------------------------------------------------------------

#[test]
fn value_upload_is_usable_as_an_image_input() {
    let root = TempRoot::new("value-upload");
    let server = start(&root.0, seams(FakeChat::done("unused"), FakeGen::done(), FakeHttp::json(200, "{}")));
    let endpoints = server.endpoints();
    put_flow(endpoints.control, &endpoints.token, "image-echo", IMAGE_ECHO_FLOW);

    let put_value = request(
        endpoints.data,
        "PUT",
        "/v1/values?type=image&content_type=image/png",
        Some(&endpoints.token),
        "not-really-a-png-but-real-bytes",
    );
    assert_eq!(put_value.status, 201, "{}", put_value.body);
    let digest = PutValueResponse::deserialize_json(&put_value.body).unwrap().digest;

    let create = request(
        endpoints.control,
        "POST",
        "/v1/flows/image-echo/instances",
        Some(&endpoints.token),
        &CreateInstanceRequest::default().serialize_json(),
    );
    assert_eq!(create.status, 201, "{}", create.body);
    let instance = CreateInstanceResponse::deserialize_json(&create.body).unwrap().instance;
    let set_inputs = request(
        endpoints.control,
        "PUT",
        &format!("/v1/instances/{instance}/inputs"),
        Some(&endpoints.token),
        &one_input("picture", "image", image_digest_input(&digest)).serialize_json(),
    );
    assert_eq!(set_inputs.status, 200, "{}", set_inputs.body);
    let inputs = SetInputsResponse::deserialize_json(&set_inputs.body).unwrap().inputs;
    assert_eq!(inputs["picture"]["image"].digest, digest);

    let run_cursor = cursor(endpoints.control, &endpoints.token, "run");
    let start_run = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{instance}/runs"),
        Some(&endpoints.token),
        &CreateRunRequest::default().serialize_json(),
    );
    let run_id = CreateRunResponse::deserialize_json(&start_run.body).unwrap().run_id;
    poll_events_until(endpoints.control, &endpoints.token, "run", run_cursor, |event| {
        event_kind(event) == "run.finished" && event_str(event, "run_id") == Some(run_id.as_str())
    });
    let row = request(endpoints.control, "GET", &format!("/v1/runs/{run_id}"), Some(&endpoints.token), "");
    let row = RunRowDto::deserialize_json(&row.body).unwrap();
    assert_eq!(row.outputs["result"].digest, digest);

    server.shutdown();
}

#[test]
fn value_ttl_expiry_returns_404() {
    let root = TempRoot::new("value-ttl");
    let mut config = FlowServerConfig::new(root.0.clone())
        .with_seams(seams(FakeChat::done("unused"), FakeGen::done(), FakeHttp::json(200, "{}")));
    // Force an immediate spill (so TTL, not the RAM cache, governs this
    // value's lifetime) and sweep fast enough for a test to observe it.
    config.values_ram_budget = 1;
    config.value_ttl_secs = 1;
    config.janitor_sweep_secs = 1;
    config.log = Box::new(|_| {});
    let server = FlowServer::start(config).unwrap();
    let endpoints = server.endpoints();

    let put_value = request(
        endpoints.data,
        "PUT",
        "/v1/values?type=bytes&content_type=application/octet-stream",
        Some(&endpoints.token),
        "scratch bytes nobody keeps",
    );
    assert_eq!(put_value.status, 201, "{}", put_value.body);
    let digest = PutValueResponse::deserialize_json(&put_value.body).unwrap().digest;

    let immediate = request(endpoints.data, "GET", &format!("/v1/values/{digest}"), Some(&endpoints.token), "");
    assert_eq!(immediate.status, 200);

    std::thread::sleep(Duration::from_millis(2_500));

    let expired = request(endpoints.data, "GET", &format!("/v1/values/{digest}"), Some(&endpoints.token), "");
    assert_eq!(expired.status, 404, "{}", expired.body);

    server.shutdown();
}

// ---------------------------------------------------------------------------
// 7. the real-fleet path is selected by default; the testpattern hub works
//    through the HTTP routes with `FixedGen` injected
// ---------------------------------------------------------------------------

#[test]
fn default_seams_are_the_real_fleet_path() {
    let root = TempRoot::new("default-seams");
    // No `with_seams`: `FlowServer::start` must build `FleetGen` + `HubChat`
    // + `HubHttp` on its own and still come up cleanly.
    let mut config = FlowServerConfig::new(root.0.clone());
    config.log = Box::new(|_| {});
    let server = FlowServer::start(config).unwrap();
    let endpoints = server.endpoints();
    let health = request(endpoints.control, "GET", "/v1/health", None, "");
    assert_eq!(health.status, 200, "{}", health.body);
    server.shutdown();
}

#[test]
fn testpattern_hub_service_runs_through_the_http_routes() {
    let cache_dir = std::env::temp_dir().join(format!(
        "makepad-flow-host-runs-testpattern-{}-{}",
        std::process::id(),
        SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos()
    ));
    std::fs::create_dir_all(&cache_dir).unwrap();
    let service = start_service(ServiceConfig {
        host: "127.0.0.1".to_string(),
        port: 0,
        cache_dir: cache_dir.clone(),
        registry: Registry {
            models: vec![ModelSpec {
                id: "testpattern".to_string(),
                domain: Domain::Image,
                backend: "testpattern".to_string(),
                available: true,
                gated: false,
                vram_gb: Some(0.0),
                min_vram_gb: None,
                min_compute_cap: None,
                note: None,
                license: None,
                files: Vec::new(),
            }],
        },
        downloader: Downloader::new("http://127.0.0.1:1", None).unwrap(),
        peer: PeerOptions { serve: Some(false), sources: Some(Vec::new()), ..Default::default() },
        fleet: DEFAULT_FLEET.to_string(),
    });
    let handle = match service {
        Ok(handle) => handle,
        Err(error) if error.to_string().contains("Operation not permitted") => {
            let _ = std::fs::remove_dir_all(&cache_dir);
            eprintln!("skipping real hub service: loopback bind is forbidden by this sandbox");
            return;
        }
        Err(error) => panic!("start real hub service: {error}"),
    };
    let base_url = format!("http://{}", handle.addr);
    std::mem::forget(handle);

    let root = TempRoot::new("testpattern-route");
    let server = start(
        &root.0,
        Seams {
            chat: std::sync::Arc::new(FakeChat::done("unused")),
            gen: std::sync::Arc::new(FixedGen(base_url)),
            http: std::sync::Arc::new(FakeHttp::json(200, "{}")),
        },
    );
    let endpoints = server.endpoints();
    let source = r#"use mod.flow.*
let prompt = Input{default: "same prompt"}
let image = Image{prompt: prompt.text() width: 64 height: 48 steps: 4 seed: 42 model: "testpattern"}
let picture = Output{type: @image value: image.image()}
Flow{prompt, image, picture}
"#;
    put_flow(endpoints.control, &endpoints.token, "gen-demo", source);
    let create = request(
        endpoints.control,
        "POST",
        "/v1/flows/gen-demo/instances",
        Some(&endpoints.token),
        &CreateInstanceRequest::default().serialize_json(),
    );
    let instance = CreateInstanceResponse::deserialize_json(&create.body).unwrap().instance;
    let run_cursor = cursor(endpoints.control, &endpoints.token, "run");
    let start_run = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{instance}/runs"),
        Some(&endpoints.token),
        &CreateRunRequest::default().serialize_json(),
    );
    assert_eq!(start_run.status, 202, "{}", start_run.body);
    let run_id = CreateRunResponse::deserialize_json(&start_run.body).unwrap().run_id;
    let events = poll_events_until(endpoints.control, &endpoints.token, "run", run_cursor, |event| {
        event_kind(event) == "run.finished" && event_str(event, "run_id") == Some(run_id.as_str())
    });
    assert_eq!(
        variant_name(events.last().and_then(|event| event.key("state"))),
        Some("Done"),
        "{events:#?}"
    );
    let row = request(endpoints.control, "GET", &format!("/v1/runs/{run_id}"), Some(&endpoints.token), "");
    let row = RunRowDto::deserialize_json(&row.body).unwrap();
    let digest = row.outputs["picture"].digest.clone();
    let fetched = request(endpoints.data, "GET", &format!("/v1/values/{digest}"), Some(&endpoints.token), "");
    assert_eq!(fetched.status, 200);
    assert_eq!(&fetched.bytes[..8], &[0x89, b'P', b'N', b'G', b'\r', b'\n', 0x1a, b'\n']);
    assert_eq!(sha256_hex(&fetched.bytes), digest.trim_start_matches("sha256:"));

    server.shutdown();
}

#[test]
fn batch_route_creates_slices_and_cancel_all_cancels_each() {
    let root = TempRoot::new("batch-cancel");
    let mut stalled = FakeChat::done("unused");
    stalled.pending = true;
    let server = start(
        &root.0,
        seams(stalled, FakeGen::done(), FakeHttp::json(200, "{}")),
    );
    let endpoints = server.endpoints();
    put_flow(
        endpoints.control,
        &endpoints.token,
        "batch-demo",
        STALLED_FLOW,
    );

    let created = request(
        endpoints.control,
        "POST",
        "/v1/flows/batch-demo/batches",
        Some(&endpoints.token),
        &CreateBatchRequest {
            parallel: 3,
            inputs: Some(one_input("prompt", "text", text_input("batch text"))),
        }
        .serialize_json(),
    );
    assert_eq!(created.status, 202, "{}", created.body);
    let batch = CreateBatchResponse::deserialize_json(&created.body).unwrap();
    assert_eq!(batch.runs.len(), 3);

    let instances = request(
        endpoints.control,
        "GET",
        "/v1/instances?flow=batch-demo",
        Some(&endpoints.token),
        "",
    );
    let instances = Vec::<InstanceRow>::deserialize_json(&instances.body).unwrap();
    assert_eq!(instances.len(), 3);
    for row in &instances {
        assert!(
            row.label
                .as_deref()
                .is_some_and(|label| label.starts_with(&format!("batch-{}#", batch.batch)))
        );
        assert_eq!(row.input_text("prompt", "text"), Some("batch text".to_string()));
    }

    let one = request(
        endpoints.control,
        "POST",
        &format!("/v1/runs/{}/cancel", batch.runs[0].run_id),
        Some(&endpoints.token),
        "",
    );
    assert_eq!(one.status, 200, "{}", one.body);
    wait_run_state(
        endpoints.control,
        &endpoints.token,
        &batch.runs[0].run_id,
        RunState::Cancelled,
    );
    let sibling = request(
        endpoints.control,
        "GET",
        &format!("/v1/instances/{}", batch.runs[1].instance),
        Some(&endpoints.token),
        "",
    );
    assert_eq!(sibling.status, 200, "single-slice cancel touched a sibling");

    let cancelled = request(
        endpoints.control,
        "POST",
        &format!("/v1/batches/{}/cancel", batch.batch),
        Some(&endpoints.token),
        "",
    );
    assert_eq!(cancelled.status, 200, "{}", cancelled.body);
    assert_eq!(
        BatchMutationResponse::deserialize_json(&cancelled.body)
            .unwrap()
            .runs,
        3
    );
    for run in &batch.runs {
        let row = wait_run_state(
            endpoints.control,
            &endpoints.token,
            &run.run_id,
            RunState::Cancelled,
        );
        assert_eq!(row.batch.as_deref(), Some(batch.batch.as_str()));
        let instance = request(
            endpoints.control,
            "GET",
            &format!("/v1/instances/{}", run.instance),
            Some(&endpoints.token),
            "",
        );
        assert_eq!(instance.status, 404);
    }

    let cleared = request(
        endpoints.control,
        "DELETE",
        &format!("/v1/batches/{}", batch.batch),
        Some(&endpoints.token),
        "",
    );
    assert_eq!(cleared.status, 200, "{}", cleared.body);
    for run in &batch.runs {
        let row = request(
            endpoints.control,
            "GET",
            &format!("/v1/runs/{}", run.run_id),
            Some(&endpoints.token),
            "",
        );
        assert_eq!(row.status, 404);
    }
    server.shutdown();
}

#[test]
fn random_seed_is_distinct_per_batch_slice_and_recorded() {
    let root = TempRoot::new("batch-seeds");
    let gen = FakeGen::done();
    let requests = gen.requests.clone();
    let server = start(
        &root.0,
        seams(FakeChat::done("unused"), gen, FakeHttp::json(200, "{}")),
    );
    let endpoints = server.endpoints();
    put_flow(
        endpoints.control,
        &endpoints.token,
        "seed-demo",
        r#"use mod.flow.*
let image = Image{prompt: "a cube" seed: @random}
let picture = Output{type: @image value: image.image()}
Flow{image, picture}
"#,
    );
    let created = request(
        endpoints.control,
        "POST",
        "/v1/flows/seed-demo/batches",
        Some(&endpoints.token),
        &CreateBatchRequest {
            parallel: 2,
            inputs: None,
        }
        .serialize_json(),
    );
    assert_eq!(created.status, 202, "{}", created.body);
    let batch = CreateBatchResponse::deserialize_json(&created.body).unwrap();
    let rows: Vec<_> = batch
        .runs
        .iter()
        .map(|run| {
            wait_run_state(
                endpoints.control,
                &endpoints.token,
                &run.run_id,
                RunState::Done,
            )
        })
        .collect();
    let mut seeds: Vec<u64> = requests
        .lock()
        .unwrap()
        .iter()
        .map(|(_, request)| request.seed.expect("resolved random seed"))
        .collect();
    assert_eq!(seeds.len(), 2);
    assert_ne!(seeds[0], seeds[1]);
    let mut recorded: Vec<u64> = rows
        .iter()
        .map(|row| {
            row.nodes["image"]
            .outputs
            .iter()
            .find(|output| output.port == "seed_used")
            .and_then(|output| output.value.preview.as_ref())
            .and_then(|literal| match literal {
                makepad_flow::Literal::Str(value) => value.parse::<u64>().ok(),
                _ => None,
            })
            .expect("seed_used output")
        })
        .collect();
    seeds.sort_unstable();
    recorded.sort_unstable();
    assert_eq!(recorded, seeds);
    server.shutdown();
}

fn wait_run_state(
    address: SocketAddr,
    token: &str,
    run_id: &str,
    wanted: RunState,
) -> RunRowDto {
    let deadline = Instant::now() + Duration::from_secs(5);
    loop {
        let response = request(
            address,
            "GET",
            &format!("/v1/runs/{run_id}"),
            Some(token),
            "",
        );
        assert_eq!(response.status, 200, "{}", response.body);
        let row = RunRowDto::deserialize_json(&response.body).unwrap();
        if row.state == wanted {
            return row;
        }
        assert!(
            Instant::now() < deadline,
            "run {run_id} stayed {:?}",
            row.state
        );
        std::thread::sleep(Duration::from_millis(10));
    }
}

#[test]
fn a_cancelled_run_marks_its_live_nodes_cancelled() {
    let root = TempRoot::new("cancel-marks-nodes");
    let mut stalled = FakeChat::done("unused");
    stalled.pending = true;
    let server = start(
        &root.0,
        seams(stalled, FakeGen::done(), FakeHttp::json(200, "{}")),
    );
    let endpoints = server.endpoints();
    put_flow(endpoints.control, &endpoints.token, "stalled", STALLED_FLOW);
    let created = request(
        endpoints.control,
        "POST",
        "/v1/flows/stalled/instances",
        Some(&endpoints.token),
        &CreateInstanceRequest::default().serialize_json(),
    );
    assert_eq!(created.status, 201, "{}", created.body);
    let instance = CreateInstanceResponse::deserialize_json(&created.body)
        .unwrap()
        .instance;
    let run_cursor = cursor(endpoints.control, &endpoints.token, "run");
    let started = request(
        endpoints.control,
        "POST",
        &format!("/v1/instances/{instance}/runs"),
        Some(&endpoints.token),
        &CreateRunRequest::default().serialize_json(),
    );
    assert_eq!(started.status, 202, "{}", started.body);
    let run_id = CreateRunResponse::deserialize_json(&started.body)
        .unwrap()
        .run_id;
    poll_events_until(
        endpoints.control,
        &endpoints.token,
        "run",
        run_cursor,
        |event| event_kind(event) == "node.started" && event_str(event, "node") == Some("expand"),
    );
    let cancelled = request(
        endpoints.control,
        "POST",
        &format!("/v1/runs/{run_id}/cancel"),
        Some(&endpoints.token),
        "",
    );
    assert_eq!(cancelled.status, 200, "{}", cancelled.body);
    let row = wait_run_state(
        endpoints.control,
        &endpoints.token,
        &run_id,
        RunState::Cancelled,
    );
    // The stalled chat node was mid-flight; the row must not keep it Running.
    assert_eq!(row.nodes["expand"].state, NodeState::Cancelled);
    assert!(row.nodes.values().all(|node| !matches!(
        node.state,
        NodeState::Pending | NodeState::Running | NodeState::Waiting
    )));
    server.shutdown();
}