ignition-cli 1.6.2

//! Opt-in LIVE gate for `ign webdev` (05-03) — dogfoods the BUILT
//! `ign` binary through the full deploy→status→redeploy→scriptExec
//! loop against a real commissioned Ignition 8.3+ gateway.
//!
//! ```text
//! cargo test -p ignition-cli --test e2e_webdev -- --ignored
//! ```
//!
//! Skip behavior (the 02-01 live-suite convention, e2e_projects
//! precedent): every test reads its env vars at start and returns
//! quietly when they are absent — `-- --ignored` with no envs set is
//! a GREEN no-op. The loop MUTATES the gateway (it imports the
//! CLI-owned `ign-cli` project wholesale), so it additionally
//! requires the explicit `IGNITION_LIVE_MUTATIONS=1` opt-in.
//!
//! ## Environment
//!
//! | var | required by | meaning |
//! |---|---|---|
//! | `IGNITION_LIVE_URL` | every test | base URL, e.g. `http://localhost:18088` |
//! | `IGNITION_LIVE_TOKEN` | every test | full `name:key` API-token string |
//! | `IGNITION_LIVE_MUTATIONS` | the loop test + the two 11-06 mutation gates | `1` to allow the deploys / tag-subtree seeding |
//!
//! The 13-04 historian binding gate (`live_tags_history_bindings`)
//! also MUTATES (provisions a unique `p13hist<run>` InternalHistorian
//! plus a bound tag) — `IGNITION_LIVE_MUTATIONS=1` required, and
//! it re-proves the 13-01 closure recipe end-to-end (bind → TAGS-13
//! render → written value proven in history).
//!
//! The 11-06 bulk-transfer gates: `live_tags_xml_fidelity_roundtrip`
//! and `live_tags_csv_roundtrip` MUTATE the gateway (they seed
//! `P11Live*`/`P11Csv*` subtrees and import) — `IGNITION_LIVE_MUTATIONS=1`
//! required, same as the loop. `live_tags_loss_gate_refusal` is READ-ONLY:
//! its refusal is pre-resolution (zero wire work by construction), so it
//! runs on URL+TOKEN alone — no mutations env. The mutation gates
//! self-deploy the current bundle FIRST (the redeploy-sequencing lesson,
//! research Pitfall 5 — any new-format call against a stale 1.x deploy
//! hits the version-drift refusal).
//!
//! ## The loop's contract pins
//!
//! 1. **status-before-deploy is DATA** — exit 0 whether the routes
//!    are absent (fresh gateway) or present (a re-run); the LOCK from
//!    the contract goldens, witnessed live.
//! 2. **deploy → status all-present** — the version handshake reads
//!    back `ROUTE_BUNDLE_VERSION` on every always-on route.
//! 3. **redeploy clean-replaces** — a second deploy (overwrite=true)
//!    leaves the sweep all-present.
//! 4. **scriptExec ships ONLY on request, gated by the secret** —
//!    `--with-script-exec` generates + persists the secret (0600);
//!    RAW wire probes then witness the security posture: no secret
//!    header → `secret_required`, wrong secret → `secret_mismatch`,
//!    the stored secret → the version handshake (the same posture
//!    the research rig witnessed live).
//! 5. **redaction** — the stored secret appears in NO stdout/stderr
//!    of any spawn (the binary-level proof; the value lives in
//!    exactly one place: the baked zip member + the 0600 config).

use std::path::{Path, PathBuf};
use std::process::Output;

use assert_cmd::Command;
use ignition_core::client::install_crypto_provider;
use serde_json::Value;

/// The gathered env contract for one live-gateway run (the
/// e2e_projects shape verbatim).
struct LiveEnv {
    url: String,
    token: String,
}

fn live_url() -> Option<String> {
    std::env::var("IGNITION_LIVE_URL")
        .ok()
        .filter(|value| !value.trim().is_empty())
}

fn live_token() -> Option<String> {
    std::env::var("IGNITION_LIVE_TOKEN")
        .ok()
        .filter(|value| !value.trim().is_empty())
}

fn skip(message: &str) {
    eprintln!("skipping: {message}");
}

fn live_env_mutations() -> Option<LiveEnv> {
    if std::env::var("IGNITION_LIVE_MUTATIONS").as_deref() != Ok("1") {
        return None;
    }
    match (live_url(), live_token()) {
        (Some(url), Some(token)) => Some(LiveEnv { url, token }),
        _ => None,
    }
}

/// ONE isolated config for the WHOLE loop — the scriptExec secret is
/// generated by the `--with-script-exec` deploy and must persist into
/// the later status/probe steps (that persistence is itself under
/// test).
fn isolated_live_config(env: &LiveEnv) -> (tempfile::TempDir, PathBuf) {
    let dir = tempfile::tempdir().expect("tempdir");
    let config = dir.path().join("config.toml");
    std::fs::write(
        &config,
        format!(
            "active = \"dev\"\n\n[profiles.dev]\nurl = \"{}\"\nauth = {{ token_env = \"IGNITION_TOKEN\" }}\n",
            env.url
        ),
    )
    .expect("write config");
    (dir, config)
}

/// Spawn the built `ign` binary at the live gateway with args.
fn ign(config: &Path, env: &LiveEnv, args: &[&str]) -> Output {
    spawn_ign(config, env, args, None, None)
}

/// `ign` with a JSON document piped to stdin (`--file -`).
fn ign_stdin(config: &Path, env: &LiveEnv, args: &[&str], stdin: &str) -> Output {
    spawn_ign(config, env, args, Some(stdin), None)
}

/// The one spawn primitive: env pinning, optional stdin, optional
/// pinned working directory (the default-file export path).
fn spawn_ign(
    config: &Path,
    env: &LiveEnv,
    args: &[&str],
    stdin: Option<&str>,
    cwd: Option<&Path>,
) -> Output {
    let mut command = Command::cargo_bin("ign").expect("binary 'ign' not found");
    command
        .env("IGNITION_CLI_CONFIG", config)
        .env("IGNITION_TOKEN", &env.token)
        .env("IGNITION_URL", &env.url);
    if let Some(dir) = cwd {
        command.current_dir(dir);
    }
    command.args(args);
    if let Some(text) = stdin {
        command.write_stdin(text);
    }
    command.output().expect("spawn ign")
}

/// Parse the compact JSON FAILURE envelope from stderr (the
/// first-`{` convention — log-tolerant).
fn err_envelope(out: &Output) -> Value {
    let stderr = String::from_utf8_lossy(&out.stderr);
    let from = stderr.find('{').expect("envelope starts somewhere");
    serde_json::from_str(&stderr[from..]).expect("envelope parses")
}

fn expect_ok(what: &str, out: &Output) {
    assert!(
        out.status.success(),
        "{what} failed (exit {:?}):\nstdout: {}\nstderr: {}",
        out.status.code(),
        String::from_utf8_lossy(&out.stdout),
        String::from_utf8_lossy(&out.stderr)
    );
}

/// Parse the compact JSON success envelope from stdout.
fn data_envelope(out: &Output) -> Value {
    let stdout = String::from_utf8_lossy(&out.stdout);
    serde_json::from_str(stdout.trim())
        .unwrap_or_else(|err| panic!("stdout envelope parses ({err}): {stdout}"))
}

/// The row for one route out of a status envelope.
fn status_row<'a>(envelope: &'a Value, route: &str) -> &'a Value {
    envelope["data"]["routes"]
        .as_array()
        .unwrap_or_else(|| panic!("routes[] present: {envelope}"))
        .iter()
        .find(|row| row["route"] == route)
        .unwrap_or_else(|| panic!("{route} row: {envelope}"))
}

/// Read the stored webdev secret back out of the config (test eyes
/// only — the raw-probe oracle; the same read the contract redaction
/// test performs).
fn stored_secret(config: &Path) -> String {
    ignition_core::config::load(config)
        .expect("config reloads")
        .profiles
        .get("dev")
        .and_then(|profile| profile.webdev_secret.clone())
        .expect("the --with-script-exec deploy persisted a secret")
}

/// ONE raw wire probe against the deployed scriptExec route — POST
/// the version action with the gateway token auth and a CHOICE of
/// secret presentation: None (no secret header), Some(secret). The
/// route's answer is the 200 body envelope (denials ride 200 — the
/// seam's first wire rule); the function returns `error.code` on a
/// denial or panics with the body when ok:true.
async fn raw_scriptexec_probe(env: &LiveEnv, secret: Option<&str>) -> Result<(), String> {
    install_crypto_provider();
    let client = reqwest::Client::new();
    let url = format!("{}/system/webdev/ign-cli/cli/scriptExec", env.url);
    let mut request = client
        .post(&url)
        .header("X-Ignition-API-Token", &env.token)
        .json(&serde_json::json!({"action": "version"}));
    if let Some(secret) = secret {
        request = request.header("X-Ignition-CLI-Secret", secret);
    }
    let response = request.send().await.expect("probe reaches the gateway");
    assert_eq!(
        response.status().as_u16(),
        200,
        "the webdev servlet answers 200 (denials ride the body)"
    );
    let body: Value = response.json().await.expect("envelope parses");
    if body["ok"] == Value::Bool(true) {
        assert_eq!(
            body["data"]["routeVersion"],
            ignition_core::webdev::ROUTE_BUNDLE_VERSION,
            "the handshake version: {body}"
        );
        Ok(())
    } else {
        Err(body["error"]["code"]
            .as_str()
            .unwrap_or("(no code)")
            .to_string())
    }
}

const ALWAYS_ON: [&str; 4] = ["tags", "tagConfig", "alarms", "tagHistory"];

/// THE live loop — order is the contract (see the module docs).
#[tokio::test]
#[ignore = "opt-in e2e: set IGNITION_LIVE_URL + IGNITION_LIVE_TOKEN + IGNITION_LIVE_MUTATIONS=1"]
async fn live_webdev_deploy_status_scriptexec_loop() {
    let Some(env) = live_env_mutations() else {
        skip(
            "IGNITION_LIVE_MUTATIONS=1 (with URL+TOKEN) not set — refusing to touch a live gateway",
        );
        return;
    };
    let (_dir, config) = isolated_live_config(&env);
    // THE serializer: one live gate at a time (shared gateway state).
    let _gate = LIVE_GATE.lock().await;
    let mut all_output = String::new();

    // (1) status BEFORE any deploy: a READ — exit 0 whether the
    // routes are absent (fresh gateway) or present (a re-run against
    // an already-deployed gateway). Degradation is data.
    let out = ign(&config, &env, &["webdev", "status", "--compact"]);
    all_output.push_str(&String::from_utf8_lossy(&out.stdout));
    all_output.push_str(&String::from_utf8_lossy(&out.stderr));
    expect_ok("status before deploy (a read — degradation is data)", &out);
    let envelope = data_envelope(&out);
    for route in ALWAYS_ON {
        let row = status_row(&envelope, route);
        let status = row["status"].as_str().unwrap_or_default();
        assert!(
            [
                "present",
                "absent",
                "unlicensed",
                "auth_gated",
                "version_mismatch"
            ]
            .contains(&status),
            "{route}: a known matrix state, got {row}"
        );
    }

    // (2) deploy → the 4 always-on routes install; status verifies
    // the version handshake on every one.
    let out = ign(&config, &env, &["webdev", "deploy", "--compact"]);
    all_output.push_str(&String::from_utf8_lossy(&out.stdout));
    all_output.push_str(&String::from_utf8_lossy(&out.stderr));
    expect_ok("deploy", &out);
    let envelope = data_envelope(&out);
    assert_eq!(envelope["data"]["routes"].as_array().map(Vec::len), Some(4));
    assert_eq!(envelope["data"]["script_exec"], Value::Bool(false));

    let out = ign(&config, &env, &["webdev", "status", "--compact"]);
    all_output.push_str(&String::from_utf8_lossy(&out.stdout));
    all_output.push_str(&String::from_utf8_lossy(&out.stderr));
    expect_ok("status after deploy", &out);
    let mut envelope = data_envelope(&out);
    // LIVE-TRUTH TOLERANCE (11-06 rig runs): the FIRST deploy on a
    // freshly commissioned gateway can answer the immediate status
    // sweep with an all-absent read (`ok:false`) — the webdev servlet
    // mounts the new project asynchronously (405=absent until it
    // does). MEASURED first-mount windows: 8.3.6 < 3s; 8.3.3 warm
    // ~30–35s; 8.3.3 FIRST-EVER servlet activation >90s (routes
    // served at ~95–105s on the fresh-rig run — hence the 240s
    // budget; every bump carries its measurement, no blind bumps —
    // the 09/10 Jetty-retry discipline). The gate still REQUIRES the
    // all-present handshake; it just allows the gateway its mount
    // window.
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(240);
    loop {
        let all_present = envelope["data"]["ok"] == Value::Bool(true)
            && ALWAYS_ON
                .iter()
                .all(|route| status_row(&envelope, route)["status"] == "present");
        if all_present {
            break;
        }
        if std::time::Instant::now() >= deadline {
            panic!("status after deploy: routes never became visible within 240s: {envelope}");
        }
        eprintln!("status after deploy: servlet mount not yet visible, retrying (bounded 240s)…");
        std::thread::sleep(std::time::Duration::from_secs(3));
        let out = ign(&config, &env, &["webdev", "status", "--compact"]);
        expect_ok("status after deploy (retry)", &out);
        envelope = data_envelope(&out);
    }
    assert_eq!(envelope["data"]["ok"], Value::Bool(true), "{envelope}");
    for route in ALWAYS_ON {
        let row = status_row(&envelope, route);
        assert_eq!(row["status"], "present", "{route}: {row}");
        assert_eq!(
            row["deployed_version"], row["expected_version"],
            "{route}: the handshake matches: {row}"
        );
    }

    // (3) redeploy → clean replace (overwrite=true); still all-present.
    let out = ign(&config, &env, &["webdev", "deploy", "--compact"]);
    all_output.push_str(&String::from_utf8_lossy(&out.stdout));
    all_output.push_str(&String::from_utf8_lossy(&out.stderr));
    expect_ok("redeploy (clean replace)", &out);
    let out = ign(&config, &env, &["webdev", "status", "--compact"]);
    all_output.push_str(&String::from_utf8_lossy(&out.stdout));
    all_output.push_str(&String::from_utf8_lossy(&out.stderr));
    expect_ok("status after redeploy", &out);
    let envelope = data_envelope(&out);
    assert_eq!(envelope["data"]["ok"], Value::Bool(true), "{envelope}");

    // (4) deploy --with-script-exec → 5 routes + a GENERATED secret
    // (fresh config → secret_rotated true); status sweeps 5 rows
    // (scriptExec probed WITH the stored secret).
    let out = ign(
        &config,
        &env,
        &["webdev", "deploy", "--with-script-exec", "--compact"],
    );
    all_output.push_str(&String::from_utf8_lossy(&out.stdout));
    all_output.push_str(&String::from_utf8_lossy(&out.stderr));
    expect_ok("deploy --with-script-exec", &out);
    let envelope = data_envelope(&out);
    assert_eq!(
        envelope["data"]["routes"].as_array().map(Vec::len),
        Some(5),
        "{envelope}"
    );
    assert_eq!(envelope["data"]["script_exec"], Value::Bool(true));
    assert_eq!(envelope["data"]["secret_rotated"], Value::Bool(true));

    let out = ign(&config, &env, &["webdev", "status", "--compact"]);
    all_output.push_str(&String::from_utf8_lossy(&out.stdout));
    all_output.push_str(&String::from_utf8_lossy(&out.stderr));
    expect_ok("status after scriptExec deploy", &out);
    let envelope = data_envelope(&out);
    let script_exec = status_row(&envelope, "scriptExec");
    assert_eq!(script_exec["status"], "present", "{script_exec}");

    // (5) the security posture, RAW on the wire: no secret header →
    // secret_required; wrong secret → secret_mismatch; the STORED
    // secret → the version handshake answers ok.
    let code = raw_scriptexec_probe(&env, None)
        .await
        .expect_err("no header refuses");
    assert_eq!(code, "secret_required", "the fail-closed gate");
    let code = raw_scriptexec_probe(&env, Some("deadbeef-wrong-secret"))
        .await
        .expect_err("wrong secret refuses");
    assert_eq!(code, "secret_mismatch", "constant-time mismatch refusal");
    raw_scriptexec_probe(&env, Some(&stored_secret(&config)))
        .await
        .expect("the stored secret opens the version handshake");

    // (6) redaction — the binary-level proof across EVERY spawn of
    // this loop: the stored secret appears in no output stream.
    let secret = stored_secret(&config);
    assert!(
        !all_output.contains(&secret),
        "redaction: the secret leaked into a spawn's output"
    );
}

/// Best-effort pre-clean (idempotent re-runs): delete a provider
/// with the guarded verb, IGNORE failures (an absent provider is the
/// usual case; the OWN-OCCUPANT convention — a mid-run panic must
/// not poison the next run against the same rig).
fn clean_provider(config: &Path, env: &LiveEnv, provider: &str) {
    let _ = ign(
        config,
        env,
        &["tags", "provider", "delete", provider, "--yes", "--compact"],
    );
}

/// Best-effort pre-clean: delete tag configs, ignoring failures.
fn clean_tag_configs(config: &Path, env: &LiveEnv, paths: &[&str]) {
    let mut args: Vec<&str> = vec!["tags", "config", "delete"];
    args.extend_from_slice(paths);
    args.push("--yes");
    args.push("--compact");
    let _ = ign(config, env, &args);
}

/// Import with the fresh-provider MOUNT-RACE tolerance (11-06 rig
/// run, live-truth delta): importTagsFile against a JUST-CREATED
/// provider intermittently answers per-row
/// `Error_Exception("…TagPath.getPathLength()… \"cleanPath\" is null")`
/// elements — the provider's tag model is still mounting (flaky 2/3
/// on immediate-import trials on 8.3.6; same lag class as the
/// deploy-loop first-sweep). Unlike parse-phase errors (wholesale
/// abort, nothing lands — probe 5's all-or-nothing), this
/// path-resolution-phase error CAN partially land; a re-import under
/// the same (overwrite) policy converges the subtree, so the helper
/// retries the SAME import within a bounded 30s window until an
/// attempt answers `failed: []`. Any non-transient failure (or a
/// non-ok exit) returns immediately — the caller's asserts fire on
/// the last envelope.
fn import_with_mount_tolerance(config: &Path, env: &LiveEnv, args: &[&str]) -> Output {
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
    let mut out = ign(config, env, args);
    loop {
        let clean = out.status.success()
            && data_envelope(&out)["data"]["failed"]
                .as_array()
                .map(Vec::len)
                == Some(0);
        if clean {
            return out;
        }
        let transient = out.status.success() && {
            let envelope = data_envelope(&out);
            let failed = envelope["data"]["failed"]
                .as_array()
                .cloned()
                .unwrap_or_default();
            !failed.is_empty()
                && failed
                    .iter()
                    .all(|f| f.as_str().map(|s| s.contains("cleanPath")).unwrap_or(false))
        };
        if !transient || std::time::Instant::now() >= deadline {
            return out;
        }
        eprintln!("import: provider mount race (cleanPath), retrying (bounded 30s)…");
        std::thread::sleep(std::time::Duration::from_secs(3));
        out = ign(config, env, args);
    }
}

/// Read with the fresh-provider RESOLUTION tolerance (11-06 rig run,
/// Rig B 8.3.3): an export against a JUST-CREATED provider can
/// transiently fail with the route's `Provider not found: <name>`
/// (the tagConfig/getConfiguration path resolves the provider before
/// its registration fully propagates) — the provider demonstrably
/// exists (a moments-later manual list shows it). Reads are
/// side-effect-free, so the helper re-runs the SAME read within a
/// bounded 30s window while stderr carries that shape; any other
/// outcome returns immediately for the caller's asserts.
fn read_with_provider_tolerance(config: &Path, env: &LiveEnv, args: &[&str]) -> Output {
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
    let mut out = ign(config, env, args);
    while !out.status.success() {
        let transient = String::from_utf8_lossy(&out.stderr).contains("Provider not found:");
        if !transient || std::time::Instant::now() >= deadline {
            return out;
        }
        eprintln!("read: provider resolution race, retrying (bounded 30s)…");
        std::thread::sleep(std::time::Duration::from_secs(3));
        out = ign(config, env, args);
    }
    out
}

/// THE live-gate serializer: every live test mutates SHARED gateway
/// state (the CLI-owned ign-cli deploy project — five concurrent
/// overwrite-imports race each other — plus tag providers), so the
/// gates hold this lock for their whole body and run ONE AT A TIME
/// within the process (cargo's default parallelism would 409-race
/// the deploys and provider creates; live-proven 05-06).
static LIVE_GATE: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

/// THE tags live loop (05-04) — the first route-consuming family,
/// self-contained: deploy the routes (browse/read/write need them),
/// then provider create → browse root sees it → read/write on a
/// NONEXISTENT path assert the QUALITY passthrough (quality IS data
/// — `Bad_NotFound`/`Bad` come back as exit-0 data, the honest
/// oracle verifiable without mutation side effects) → provider
/// delete cleans up. Provider verbs themselves ride the NATIVE REST
/// (no route dependency).
#[tokio::test]
#[ignore = "opt-in e2e: set IGNITION_LIVE_URL + IGNITION_LIVE_TOKEN + IGNITION_LIVE_MUTATIONS=1"]
async fn live_tags_provider_browse_read_write_loop() {
    let Some(env) = live_env_mutations() else {
        skip(
            "IGNITION_LIVE_MUTATIONS=1 (with URL+TOKEN) not set — refusing to touch a live gateway",
        );
        return;
    };
    let (_dir, config) = isolated_live_config(&env);
    // THE serializer: one live gate at a time (shared gateway state).
    let _gate = LIVE_GATE.lock().await;

    // Routes first: browse/read/write refuse exit 6 without them.
    let out = ign(&config, &env, &["webdev", "deploy", "--compact"]);
    expect_ok("deploy (the tags route's precondition)", &out);

    // Pre-clean a leftover occupant from a prior aborted run.
    clean_provider(&config, &env, "p5e2e");

    // Provider create (NATIVE REST): a STANDARD provider named p5e2e.
    let out = ign(
        &config,
        &env,
        &["tags", "provider", "create", "p5e2e", "--compact"],
    );
    expect_ok("provider create p5e2e", &out);
    let envelope = data_envelope(&out);
    assert_eq!(envelope["data"]["name"], "p5e2e");

    // Provider list sees it (native), browse root sees it (route).
    let out = ign(&config, &env, &["tags", "provider", "list", "--compact"]);
    expect_ok("provider list", &out);
    let envelope = data_envelope(&out);
    assert!(
        envelope["data"]["providers"]
            .as_array()
            .expect("providers[]")
            .iter()
            .any(|row| row["name"] == "p5e2e"),
        "p5e2e in the native list: {envelope}"
    );
    let out = ign(&config, &env, &["tags", "browse", "--compact"]);
    expect_ok("browse root", &out);
    let envelope = data_envelope(&out);
    assert!(
        envelope["data"]["entries"]
            .as_array()
            .expect("entries[]")
            .iter()
            .any(|entry| entry["path"] == "[p5e2e]"),
        "browse root contains the fresh provider: {envelope}"
    );

    // Read a NONEXISTENT path: exit 0, the quality passthrough IS
    // the honest oracle (Bad_NotFound(...) — quality never parsed, so
    // a missing tag is DATA, not an error; the LIVE string carries
    // embedded detail: `Bad_NotFound("Path ... not found.")`).
    let out = ign(
        &config,
        &env,
        &["tags", "read", "[p5e2e]DoesNotExist", "--compact"],
    );
    expect_ok("read nonexistent path (quality is data)", &out);
    let envelope = data_envelope(&out);
    assert!(
        envelope["data"]["results"][0]["quality"]
            .as_str()
            .unwrap_or_default()
            .starts_with("Bad_NotFound"),
        "quality is data with embedded detail: {envelope}"
    );

    // Write to the nonexistent path likewise returns a Bad quality
    // (verifiable without mutation side effects on a fresh
    // provider).
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "write",
            "[p5e2e]DoesNotExist",
            "--value",
            "7",
            "--compact",
        ],
    );
    expect_ok("write nonexistent path (quality is data)", &out);
    let envelope = data_envelope(&out);
    assert!(
        envelope["data"]["quality"]
            .as_str()
            .expect("quality present")
            .starts_with("Bad"),
        "a nonexistent target writes Bad quality: {envelope}"
    );

    // Cleanup: the guarded, signature-chained native delete.
    let out = ign(
        &config,
        &env,
        &["tags", "provider", "delete", "p5e2e", "--yes", "--compact"],
    );
    expect_ok("provider delete p5e2e (guarded)", &out);
    let envelope = data_envelope(&out);
    assert_eq!(envelope["data"]["deleted"], "p5e2e");
}

/// THE tagConfig live round-trip (05-05) — the research-proven loop
/// that closes TAGS-05/09: config create an Int4 memory tag (value
/// 123) → read back Good/123 (05-04's read IS the oracle) → export
/// the tag subtree → import into a fresh provider (abort first: the
/// second import refuses `tag_collision`, then overwrite --yes
/// replaces) → read the IMPORTED tag == 123/Good (values intact —
/// TAGS-09's live proof) → config delete cleanup → provider deletes.
/// (11-06 reconciliation: the export rides the `[p5e2e]T1` SUBTREE —
/// the 1.2.0 route contract-refuses provider-root exports, so the
/// Phase-5 provider-scope wrapper workflow is gone by design.)
#[tokio::test]
#[ignore = "opt-in e2e: set IGNITION_LIVE_URL + IGNITION_LIVE_TOKEN + IGNITION_LIVE_MUTATIONS=1"]
async fn live_tags_config_export_import_roundtrip() {
    let Some(env) = live_env_mutations() else {
        skip(
            "IGNITION_LIVE_MUTATIONS=1 (with URL+TOKEN) not set — refusing to touch a live gateway",
        );
        return;
    };
    let (_dir, config) = isolated_live_config(&env);
    // THE serializer: one live gate at a time (shared gateway state).
    let _gate = LIVE_GATE.lock().await;

    // Routes first: every tagConfig verb refuses exit 6 without them.
    let out = ign(&config, &env, &["webdev", "deploy", "--compact"]);
    expect_ok("deploy (the tagConfig route's precondition)", &out);

    // Pre-clean leftovers from a prior aborted run (idempotent).
    for provider in ["p5e2e", "p5import"] {
        clean_provider(&config, &env, provider);
    }

    // Fresh providers for the round-trip (native REST, pre-cleaned).
    for provider in ["p5e2e", "p5import"] {
        let out = ign(
            &config,
            &env,
            &["tags", "provider", "create", provider, "--compact"],
        );
        expect_ok(&format!("provider create {provider}"), &out);
    }

    // Config create: an Int4 memory tag, value 123 (stdin definition —
    // the surgical edit loop's write half).
    let out = ign_stdin(
        &config,
        &env,
        &[
            "tags",
            "config",
            "create",
            "[p5e2e]T1",
            "--file",
            "-",
            "--compact",
        ],
        r#"{"tagType": "AtomicTag", "dataType": "Int4", "value": 123}"#,
    );
    expect_ok("config create [p5e2e]T1", &out);
    let envelope = data_envelope(&out);
    assert_eq!(envelope["data"]["quality"], "Good");

    // Read back: Good/123 — 05-04's read verb is the oracle.
    let out = ign(&config, &env, &["tags", "read", "[p5e2e]T1", "--compact"]);
    expect_ok("read [p5e2e]T1", &out);
    let envelope = data_envelope(&out);
    assert_eq!(envelope["data"]["results"][0]["quality"], "Good");
    assert_eq!(envelope["data"]["results"][0]["value"], 123);

    // Export the tag subtree → a JSON file. SUBTREE path, not the
    // provider root: the 1.2.0 route codified the 07-06/8.3.3 truth
    // (getConfiguration/exportTags need an RpcContext WebDev threads
    // don't carry) into a deterministic provider-root refusal — the
    // Phase-5 provider-scope export this gate originally rode is
    // route-refused by contract now (first live proof: the 11-06
    // rig run).
    let tmp = tempfile::tempdir().expect("tempdir");
    let export_file = tmp.path().join("p5e2e.json");
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "export",
            "[p5e2e]T1",
            "-o",
            export_file.to_str().expect("path"),
            "--compact",
        ],
    );
    expect_ok("export [p5e2e]T1", &out);
    let envelope = data_envelope(&out);
    assert_eq!(envelope["data"]["tag_count"], 1, "one top-level subtree");
    assert_eq!(envelope["data"]["file"], export_file.display().to_string());

    // Import #1 (abort, clean target): the browse pre-check passes,
    // configure 'a' imports.
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "import",
            "--file",
            export_file.to_str().expect("path"),
            "--provider",
            "p5import",
            "--compact",
        ],
    );
    expect_ok("import #1 (abort, clean target)", &out);
    let envelope = data_envelope(&out);
    // The export is the T1 SUBTREE (named) — it lands itself at the
    // target; the effective top-level count is 1 (the provider-
    // wrapper workflow the original Phase-5 count rode is route-
    // refused by contract now).
    assert_eq!(envelope["data"]["imported"], 1, "{envelope}");

    // LANDING VERIFICATION (11-06 Rig A live truth): a clean answer
    // against a JUST-CREATED provider can be a SILENT NO-OP (see the
    // 11-06 gates' comments) — and import #2's collision refusal
    // PRESUPPOSES the landing. Verify by read-back and re-import
    // (abort policy, clean target — state-safe) bounded.
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
    loop {
        let out = read_with_provider_tolerance(
            &config,
            &env,
            &["tags", "read", "[p5import]T1", "--compact"],
        );
        let landed =
            out.status.success() && data_envelope(&out)["data"]["results"][0]["quality"] == "Good";
        if landed {
            break;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "import #1 never landed (the read-back kept missing T1)"
        );
        eprintln!("import: clean answer but nothing landed, re-importing (bounded 30s)…");
        let out = ign(
            &config,
            &env,
            &[
                "tags",
                "import",
                "--file",
                export_file.to_str().expect("path"),
                "--provider",
                "p5import",
                "--compact",
            ],
        );
        if out.status.code() == Some(6) {
            // A collision refusal is itself LANDING EVIDENCE — the
            // read-back raced the model, but the tag demonstrably
            // exists (11-06 live truth: the read raced, the abort
            // import then saw the existing T1).
            break;
        }
        expect_ok("import #1 (re-import)", &out);
    }

    // Import #2 (abort again): the collision refusal — exit 6,
    // tag_collision, BEFORE any write.
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "import",
            "--file",
            export_file.to_str().expect("path"),
            "--provider",
            "p5import",
            "--compact",
        ],
    );
    assert_eq!(
        out.status.code(),
        Some(6),
        "collision refuses exit 6; stdout: {} stderr: {}",
        String::from_utf8_lossy(&out.stdout),
        String::from_utf8_lossy(&out.stderr)
    );
    let envelope = err_envelope(&out);
    assert_eq!(envelope["error"]["code"], "tag_collision");

    // Import #3 (overwrite --yes): the guarded replace.
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "import",
            "--file",
            export_file.to_str().expect("path"),
            "--provider",
            "p5import",
            "--collision-policy",
            "overwrite",
            "--yes",
            "--compact",
        ],
    );
    expect_ok("import #3 (overwrite --yes)", &out);

    // THE round-trip proof: the imported tag reads back 123/Good —
    // values intact across the export/import interchange.
    let out = ign(
        &config,
        &env,
        &["tags", "read", "[p5import]T1", "--compact"],
    );
    expect_ok("read [p5import]T1 (the round-trip oracle)", &out);
    let envelope = data_envelope(&out);
    assert_eq!(envelope["data"]["results"][0]["quality"], "Good");
    assert_eq!(envelope["data"]["results"][0]["value"], 123);

    // Cleanup: config delete both tags (guarded), then the providers.
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "config",
            "delete",
            "[p5e2e]T1",
            "[p5import]T1",
            "--yes",
            "--compact",
        ],
    );
    expect_ok("config delete cleanup", &out);
    for provider in ["p5e2e", "p5import"] {
        let out = ign(
            &config,
            &env,
            &["tags", "provider", "delete", provider, "--yes", "--compact"],
        );
        expect_ok(&format!("provider delete {provider} (guarded)"), &out);
    }
}

// ---- 05-06 live fixtures: the historian/binding spike + the alarm
// lifecycle (TAGS-07/08) ----

/// Wall-clock epoch milliseconds.
fn now_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .expect("clock after epoch")
        .as_millis() as i64
}

/// The InternalHistorian create path (native config-resource REST —
/// NO database needed, the 05-RESEARCH live-proven recipe).
const HISTORIAN_PROVIDER_PATH: &str =
    "/data/api/v1/resources/com.inductiveautomation.historian/historian-provider";

/// Provision an InternalHistorian via native REST (token auth, the
/// ARRAY body — the tag-provider create precedent). Idempotent: a
/// create that comes back non-OK with an "already exists"-flavored
/// body passes (re-runs against a provisioned rig).
async fn provision_internal_historian(env: &LiveEnv, name: &str) {
    install_crypto_provider();
    let client = reqwest::Client::new();
    let url = format!("{}{HISTORIAN_PROVIDER_PATH}", env.url);
    let body = serde_json::json!([{
        "name": name,
        "type": "com.inductiveautomation.historian/historian-provider",
        "collection": "core",
        "enabled": true,
        "config": {"profile": {"type": "InternalHistorian"}, "settings": {}}
    }]);
    let response = client
        .post(&url)
        .header("X-Ignition-API-Token", &env.token)
        .json(&body)
        .send()
        .await
        .expect("historian create reaches the gateway");
    let status = response.status().as_u16();
    let text = response.text().await.unwrap_or_default();
    assert!(
        status == 200
            || text.to_lowercase().contains("exist")
            || text.to_lowercase().contains("duplicate"),
        "InternalHistorian create failed (HTTP {status}): {text}"
    );
    println!("historian {name} provisioned (HTTP {status})");
}

/// Delete the InternalHistorian — the CORRECTED resources-API chain
/// (13-01's live-proven route finding): find is the GENERIC mount
/// `GET /data/api/v1/resources/find/{module}/{type}/{name}` (the
/// provider-prefixed `…/resources/{module}/{type}/find/{name}` shape
/// used here before 13-04 returns Jetty-404 "No route match" on every
/// rig — the old non-200→"already gone" tolerance made this delete a
/// SILENT NO-OP for its whole life), and delete takes the signature:
/// `DELETE /data/api/v1/resources/{module}/{type}/{name}/{signature}`.
/// The delete is VERIFIED: re-find must answer 404 — the
/// teardown-clean contract, no more silent no-ops. A 404 on the
/// first find means it is already gone.
async fn delete_internal_historian(env: &LiveEnv, name: &str) {
    install_crypto_provider();
    let client = reqwest::Client::new();
    let find_url = format!(
        "{}/data/api/v1/resources/find/com.inductiveautomation.historian/historian-provider/{name}",
        env.url
    );
    let response = client
        .get(&find_url)
        .header("X-Ignition-API-Token", &env.token)
        .send()
        .await
        .expect("historian find reaches the gateway");
    let status = response.status().as_u16();
    if status == 404 {
        println!("historian {name} already gone (find → 404)");
        return;
    }
    assert_eq!(status, 200, "historian find answered HTTP {status}");
    let record: Value = response.json().await.expect("find record parses");
    let Some(signature) = record.get("signature").and_then(Value::as_str) else {
        println!("historian {name} find record carried no signature — leaving it (manual cleanup)");
        return;
    };
    let delete_url = format!("{}{HISTORIAN_PROVIDER_PATH}/{name}/{signature}", env.url);
    let response = client
        .delete(&delete_url)
        .header("X-Ignition-API-Token", &env.token)
        .send()
        .await
        .expect("historian delete reaches the gateway");
    let status = response.status().as_u16();
    assert_eq!(
        status, 200,
        "historian {name} delete answered HTTP {status}"
    );
    // VERIFY gone: the re-find must answer 404 (the 13-01 cleanup
    // chain's proof, now in-harness).
    let response = client
        .get(&find_url)
        .header("X-Ignition-API-Token", &env.token)
        .send()
        .await
        .expect("historian re-find reaches the gateway");
    assert_eq!(
        response.status().as_u16(),
        404,
        "historian {name} still findable after delete — the teardown was NOT clean"
    );
    println!("historian {name} deleted + verified gone (find → 404)");
}

/// Do the history rows carry ANY non-null value cell (beyond the
/// t_stamp column)? The spike's data oracle.
fn history_has_data(envelope: &Value) -> bool {
    let Some(rows) = envelope["data"]["rows"].as_array() else {
        return false;
    };
    rows.iter().any(|row| {
        row.as_array()
            .map(|cells| cells.iter().skip(1).any(|cell| !cell.is_null()))
            .unwrap_or(false)
    })
}

/// Do the history rows carry a SPECIFIC numeric value anywhere beyond
/// the t_stamp column? The closure data oracle (13-01's proof shape:
/// the WRITTEN value proven in history — stronger than any-non-null,
/// which the binding's initial 0 row would satisfy). Int-matched with
/// a float fallback (wire serialization may differ across 8.3.x).
fn history_rows_contain(envelope: &Value, value: i64) -> bool {
    envelope["data"]["rows"]
        .as_array()
        .map(|rows| {
            rows.iter().any(|row| {
                row.as_array()
                    .map(|cells| {
                        cells.iter().skip(1).any(|cell| {
                            cell.as_i64() == Some(value) || cell.as_f64() == Some(value as f64)
                        })
                    })
                    .unwrap_or(false)
            })
        })
        .unwrap_or(false)
}

/// expect_ok's Result twin for gate bodies that must ALWAYS reach
/// their cleanup (the finally pattern): same failure detail, no
/// panic — the caller cleans up, THEN propagates.
fn ensure_ok(what: &str, out: &Output) -> Result<(), String> {
    if out.status.success() {
        return Ok(());
    }
    Err(format!(
        "{what} FAILED (exit {:?}):\n--- stdout ---\n{}\n--- stderr ---\n{}",
        out.status.code(),
        String::from_utf8_lossy(&out.stdout),
        String::from_utf8_lossy(&out.stderr)
    ))
}

/// THE tag-history live fixture + binding spike (05-06, TAGS-08 —
/// 05-RESEARCH Open Question 1, bounded ≤30 min by construction):
///
/// 1. provision `InternalHistorian` `p5hist` via native REST (no
///    database, the live-proven recipe),
/// 2. configure one Int4 tag with `historyEnabled` +
///    `historicalProvider` through the deployed tagConfig route,
/// 3. write values, then query `ign tags history query`,
/// 4. THE SPIKE: null values trigger the three documented
///    candidates in order — (1) execution scan-class keys in the
///    configure shape, (2) wider windows + aggregationMode
///    variations, (3) a browseHistoricalTags cross-check. A winner
///    is printed + pinned as a comment; NO winner asserts the
///    STRUCTURAL outcome only (columns present, no error, null
///    values — the research pre-cleared this fallback; query
///    capability is the phase criterion, never hang on the spike).
#[tokio::test]
#[ignore = "opt-in e2e: set IGNITION_LIVE_URL + IGNITION_LIVE_TOKEN + IGNITION_LIVE_MUTATIONS=1"]
async fn live_tags_history_historian_and_binding_spike() {
    let Some(env) = live_env_mutations() else {
        skip(
            "IGNITION_LIVE_MUTATIONS=1 (with URL+TOKEN) not set — refusing to touch a live gateway",
        );
        return;
    };
    let (_dir, config) = isolated_live_config(&env);
    // THE serializer: one live gate at a time (shared gateway state).
    let _gate = LIVE_GATE.lock().await;
    let tag = "[default]P5H/T1";

    // Pre-clean leftovers from a prior aborted run (idempotent):
    // the tag first (create over an existing node would abort), then
    // the historian.
    clean_tag_configs(&config, &env, &[tag]);
    delete_internal_historian(&env, "p5hist").await;

    // Routes first: history query refuses exit 6 without them.
    let out = ign(&config, &env, &["webdev", "deploy", "--compact"]);
    expect_ok("deploy (the tagHistory route's precondition)", &out);

    // (1) The historian — native REST, no database.
    provision_internal_historian(&env, "p5hist").await;

    // (2) The history-enabled Int4 tag (the research's stored shape:
    // historyEnabled + historicalProvider BOTH at the top level).
    let out = ign_stdin(
        &config,
        &env,
        &["tags", "config", "create", tag, "--file", "-", "--compact"],
        r#"{"tagType": "AtomicTag", "dataType": "Int4", "value": 0, "historyEnabled": true, "historicalProvider": "p5hist"}"#,
    );
    expect_ok("config create the history-enabled tag", &out);

    // (3) Write values (spaced so a sampling historian can catch
    // distinct values), then query.
    let start_ms = now_ms() - 5_000;
    for value in [7, 8, 9] {
        let out = ign(
            &config,
            &env,
            &[
                "tags",
                "write",
                tag,
                "--value",
                &value.to_string(),
                "--compact",
            ],
        );
        expect_ok("write a history value", &out);
        tokio::time::sleep(std::time::Duration::from_millis(1_100)).await;
    }
    let end_ms = now_ms() + 5_000;
    // The query argv builder (String args — the value-typed flags
    // need formatting; spawn converts to &str).
    let query_args = |start_ms: i64, end_ms: i64, aggregation: Option<&str>| {
        let mut args: Vec<String> = vec![
            "tags".into(),
            "history".into(),
            "query".into(),
            tag.into(),
            "--start".into(),
            start_ms.to_string(),
            "--end".into(),
            end_ms.to_string(),
        ];
        if let Some(mode) = aggregation {
            args.push("--aggregation".into());
            args.push(mode.into());
        }
        args.push("--compact".into());
        args
    };
    let query = |config: &std::path::Path,
                 env: &LiveEnv,
                 start_ms: i64,
                 end_ms: i64,
                 aggregation: Option<&str>| {
        let args = query_args(start_ms, end_ms, aggregation);
        let refs: Vec<&str> = args.iter().map(String::as_str).collect();
        ign(config, env, &refs)
    };
    let out = query(&config, &env, start_ms, end_ms, None);
    expect_ok("history query (base shape)", &out);
    let envelope = data_envelope(&out);
    let columns = envelope["data"]["columns"]
        .as_array()
        .cloned()
        .unwrap_or_default();
    assert_eq!(
        columns.first().and_then(Value::as_str),
        Some("t_stamp"),
        "{envelope}"
    );
    // LIVE-DISCOVERED (05-06): history tag columns ride PROVIDER-RELATIVE
    // (the bracket is stripped — `[default]P5H/T1` surfaces as `P5H/T1`);
    // t_stamp is preserved exactly, never renamed.
    let relative = tag
        .trim_start_matches('[')
        .split_once(']')
        .map(|(_, rest)| rest);
    assert!(
        columns
            .iter()
            .any(|column| column.as_str() == relative && relative.is_some()),
        "the tag column rides provider-relative: {envelope}"
    );

    if history_has_data(&envelope) {
        // SPIKE RESOLVED at the base shape.
        println!(
            "SPIKE RESOLVED (base shape): historyEnabled + historicalProvider on an Int4 tag \
             with a REST-provisioned InternalHistorian produces DATA — pin this shape \
             (see the comment at this fixture in e2e_webdev.rs)"
        );
        // PIN: the base binding shape is
        //   {tagType: AtomicTag, dataType: Int4, historyEnabled: true,
        //    historicalProvider: <name>} + InternalHistorian via native REST.
    } else {
        println!("SPIKE: base shape returned null values — trying the documented candidates");
        let mut resolved: Option<String> = None;

        // Candidate 1: execution scan-class keys in the configure
        // shape (edit = full node config under 'o' — carry the base
        // keys, add the execution block).
        let out = ign_stdin(
            &config,
            &env,
            &["tags", "config", "edit", tag, "--file", "-", "--compact"],
            r#"{"tagType": "AtomicTag", "dataType": "Int4", "value": 9, "historyEnabled": true, "historicalProvider": "p5hist", "execution": {"mode": "TagGroup", "rate": 1000}}"#,
        );
        expect_ok("candidate 1: execution scan-class edit", &out);
        let out = ign(
            &config,
            &env,
            &["tags", "write", tag, "--value", "11", "--compact"],
        );
        expect_ok("candidate 1: write", &out);
        tokio::time::sleep(std::time::Duration::from_millis(2_500)).await;
        let out = query(&config, &env, start_ms, end_ms, None);
        expect_ok("candidate 1: query", &out);
        let envelope = data_envelope(&out);
        if history_has_data(&envelope) {
            resolved = Some("candidate 1: execution scan-class keys in the configure shape".into());
        }

        // Candidate 2: wider window + aggregationMode variations.
        if resolved.is_none() {
            let wide_start = now_ms() - 600_000;
            let wide_end = now_ms() + 5_000;
            for mode in ["Average", "MinMax", "DurationOn"] {
                let out = query(&config, &env, wide_start, wide_end, Some(mode));
                expect_ok(&format!("candidate 2: query ({mode})"), &out);
                let envelope = data_envelope(&out);
                if history_has_data(&envelope) {
                    resolved = Some(format!(
                        "candidate 2: aggregationMode {mode} on a 10-minute window"
                    ));
                    break;
                }
            }
        }

        // Candidate 3: browseHistoricalTags cross-check — the
        // historian registered? (Diagnostic: proves the provider is
        // alive even when the tag binding is not.)
        let out = ign(
            &config,
            &env,
            &["tags", "browse", "histprov:p5hist", "--compact"],
        );
        let historian_registered = out.status.success()
            && data_envelope(&out)["data"]["entries"]
                .as_array()
                .is_some_and(|entries| !entries.is_empty());
        println!(
            "candidate 3 cross-check: histprov:p5hist browse {}",
            if historian_registered {
                "ANSWERS (the historian is registered)"
            } else {
                "answered empty (the historian is registered but holds no tags)"
            }
        );

        match resolved {
            Some(winner) => println!("SPIKE RESOLVED: {winner} — pin the winning shape"),
            None => {
                // The pre-cleared fallback: the STRUCTURAL outcome is
                // the assertion (t_stamp + tag columns, no error);
                // null values are the documented limitation.
                println!(
                    "SPIKE OUTCOME (no winner): history query is structurally working \
                     (t_stamp + tag columns, exit 0) but the tag↔historian DATA binding \
                     remains unresolved after all documented candidates — recorded as a \
                     documented limitation (README 'Tag history'); the Designer-diff \
                     follow-up (create one history tag by hand in the Designer, getConfig \
                     it via this CLI, diff the shapes) is the resolution path"
                );
            }
        }
    }

    // Cleanup (self-cleaning): the tag, then the historian.
    let out = ign(
        &config,
        &env,
        &["tags", "config", "delete", tag, "--yes", "--compact"],
    );
    expect_ok("config delete the history tag", &out);
    delete_internal_historian(&env, "p5hist").await;
}

/// THE historian binding live gate (13-04, TAGS-14 CLOSURE branch —
/// the 13-01 spike recipe re-proven end-to-end; field set
/// capture-locked to 13-LIVE-CAPTURES.md §Field-set table):
///
/// 1. provision `InternalHistorian` with a UNIQUE `p13hist<run>`
///    name via native REST (no database — the 05-RESEARCH recipe),
/// 2. create the bound tag — the CLOSURE RECIPE: the COMPLETE node
///    shape (`tags config` writes REPLACE the whole node — 13-01
///    wire truth) plus the captured three-key field set
///    (`historyEnabled` + `historyProvider` + `sampleMode:
///    "TagGroup"`; the historical-group key is NOT needed — the
///    gateway's DEFAULT group binds, 13-01 bonus finding),
/// 3. TAGS-13 live assert: `ign tags config get` HUMAN output
///    carries the additive `history:` block naming the live provider,
/// 4. write 42/43/44, then poll `tags history query` (bounded — the
///    tag-group scan stores at 40–50 s cadence, 13-01 data probe)
///    until the WRITTEN 44 appears: DATA rows are the closure
///    evidence. A null-data outcome on this branch is a DRIFT
///    FINDING (the gate fails loudly; 13-LIVE-GATE.md records it —
///    capture wins, never code around it),
/// 5. FINALLY: tag + historian deletion runs regardless of the
///    body's outcome, and the historian delete VERIFIES gone
///    (re-find → 404 — the corrected harness chain).
#[tokio::test]
#[ignore = "opt-in e2e: set IGNITION_LIVE_URL + IGNITION_LIVE_TOKEN + IGNITION_LIVE_MUTATIONS=1"]
async fn live_tags_history_bindings() {
    let Some(env) = live_env_mutations() else {
        skip(
            "IGNITION_LIVE_MUTATIONS=1 (with URL+TOKEN) not set — refusing to touch a live gateway",
        );
        return;
    };
    let (_dir, config) = isolated_live_config(&env);
    // THE serializer: one live gate at a time (shared gateway state).
    let _gate = LIVE_GATE.lock().await;
    // UNIQUE historian AND tag name per run (the 11-06 discipline —
    // dozens of same-name cycles degrade a rig's tag model; virgin
    // namespaces. The 13-04 rig-A runs proved the class applies to
    // TAG paths too: a delete/recreated [default]P13H/T1 never
    // re-registered with the historian while a fresh path flowed on
    // the first scan).
    let suffix = now_ms();
    let historian = format!("p13hist{suffix}");
    let tag = format!("[default]P13H/T{suffix}");

    // Pre-clean leftovers (idempotent re-runs): the tag first (create
    // over an existing node would abort), then the historian.
    clean_tag_configs(&config, &env, &[&tag]);
    delete_internal_historian(&env, &historian).await;

    // Routes first: history query refuses exit 6 without them. The
    // import can race a partial route mount on a fresh gateway (the
    // 13-04 rig-A run: three routes landed, `tags` answered 405 — the
    // documented 11-06 mount-race class; the redeploy-overwrite heals
    // it), so the step VERIFIES the sweep afterwards and redeploys
    // ONCE on a miss.
    let out = ign(&config, &env, &["webdev", "deploy", "--compact"]);
    expect_ok("deploy (the tagHistory route's precondition)", &out);
    // The sweep itself is POLLED (30 s): a just-imported servlet's
    // first activation can lag seconds-to-minutes (the 11-06
    // first-activation tolerance) — a single-shot sweep would fail
    // the heal path spuriously (13-04 rig-B run 1).
    let sweep_all_present = |config: &Path, env: &LiveEnv| -> bool {
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
        loop {
            let out = ign(config, env, &["webdev", "status", "--compact"]);
            let healthy = out.status.success()
                && ["tags", "tagConfig", "alarms", "tagHistory"]
                    .iter()
                    .all(|route| {
                        data_envelope(&out)["data"]["routes"]
                            .as_array()
                            .is_some_and(|rows| {
                                rows.iter().any(|row| {
                                    row["route"].as_str() == Some(*route)
                                        && row["status"].as_str() == Some("present")
                                })
                            })
                    });
            if healthy || std::time::Instant::now() >= deadline {
                return healthy;
            }
            std::thread::sleep(std::time::Duration::from_secs(3));
        }
    };
    if !sweep_all_present(&config, &env) {
        eprintln!("deploy: mount race (a core route absent) — redeploying once (11-06 lesson)");
        let out = ign(&config, &env, &["webdev", "deploy", "--compact"]);
        expect_ok("redeploy (mount-race heal)", &out);
        assert!(
            sweep_all_present(&config, &env),
            "core routes still absent after the heal redeploy"
        );
    }

    let outcome = historian_binding_body(&config, &env, &historian, &tag).await;

    // FINALLY: self-cleaning regardless of the body's outcome (the
    // corrected historian chain verifies gone).
    clean_tag_configs(&config, &env, &[&tag]);
    delete_internal_historian(&env, &historian).await;

    if let Err(message) = outcome {
        panic!("{message}");
    }
}

/// The gate's asserting body, Result-shaped so the FINALLY above
/// always runs (ensure_ok instead of expect_ok inside).
///
/// The bind→write→prove sequence runs inside a 3-cycle SELF-HEALING
/// loop (the 11-06 live-truth recovery, proven again on 13-04 rig A):
/// a tag created in the deploy import's model-rebuild tail can read
/// back its config, answer writes Good, and yet NEVER register with
/// the historian (rows stay empty from birth — rig A runs 2–4, while
/// identical probes created OUTSIDE that window flowed on the first
/// write). The proven recovery is delete → settle → recreate.
async fn historian_binding_body(
    config: &Path,
    env: &LiveEnv,
    historian: &str,
    tag: &str,
) -> Result<(), String> {
    // (1) The historian — native REST, no database.
    provision_internal_historian(env, historian).await;

    let mut last_miss = String::new();
    for cycle in 1..=3 {
        if cycle > 1 {
            eprintln!(
                "bind cycle {cycle}: prior cycle stored nothing — delete → settle → recreate"
            );
        }
        // Settle FIRST (the deploy import's model-rebuild tail on
        // cycle 1; the recreate tail on later cycles — 30 s, rig B's
        // just-RUNNING gateway orphaned every 15 s settle), then
        // clear any prior cycle's node.
        tokio::time::sleep(std::time::Duration::from_secs(30)).await;
        clean_tag_configs(config, env, std::slice::from_ref(&tag));

        // (2) The bound tag — the closure recipe, complete node shape
        // + the captured three-key field set.
        let out = ign_stdin(
            config,
            env,
            &["tags", "config", "create", tag, "--file", "-", "--compact"],
            &format!(
                r#"{{"tagType": "AtomicTag", "dataType": "Int4", "value": 0, "historyEnabled": true, "historyProvider": "{historian}", "sampleMode": "TagGroup"}}"#
            ),
        );
        ensure_ok("config create the bound tag (closure recipe)", &out)?;

        // (3) TAGS-13 live assert — the HUMAN render carries the
        // additive history block with the LIVE provider name (no
        // --compact).
        let out = ign(config, env, &["tags", "config", "get", tag]);
        ensure_ok("config get the bound tag", &out)?;
        let stdout = String::from_utf8_lossy(&out.stdout);
        for needle in [
            "history:",
            "  enabled: true",
            &format!("  provider: {historian}"),
            "  sample mode: TagGroup",
        ] {
            if !stdout.contains(needle) {
                return Err(format!("TAGS-13 live render missing {needle:?}:\n{stdout}"));
            }
        }
        println!("TAGS-13 live: history block present for {tag} (provider {historian})");

        // (4) Write values (spaced), then poll the query until the
        // WRITTEN 44 lands — the stored row can ride the write
        // directly (~seconds, live-proven) or the tag-group scan
        // (40–50 s cadence, the 13-01 data probe), so the per-cycle
        // poll is bounded and patient.
        let start_ms = now_ms() - 60_000;
        for value in [42, 43, 44] {
            let out = ign(
                config,
                env,
                &[
                    "tags",
                    "write",
                    tag,
                    "--value",
                    &value.to_string(),
                    "--compact",
                ],
            );
            ensure_ok("write a history value", &out)?;
            tokio::time::sleep(std::time::Duration::from_millis(1_100)).await;
        }
        let end_ms = now_ms() + 5_000;
        let query_args = |start_ms: i64, end_ms: i64| {
            vec![
                "tags".to_string(),
                "history".to_string(),
                "query".to_string(),
                tag.to_string(),
                "--start".to_string(),
                start_ms.to_string(),
                "--end".to_string(),
                end_ms.to_string(),
                "--compact".to_string(),
            ]
        };
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(75);
        loop {
            let args = query_args(start_ms, end_ms);
            let refs: Vec<&str> = args.iter().map(String::as_str).collect();
            let out = ign(config, env, &refs);
            ensure_ok("history query", &out)?;
            let envelope = data_envelope(&out);
            // Structural pins while we're here (the 05-06 wire
            // truths): t_stamp leads, the tag column rides
            // provider-relative.
            let columns = envelope["data"]["columns"]
                .as_array()
                .cloned()
                .unwrap_or_default();
            if columns.first().and_then(Value::as_str) != Some("t_stamp") {
                return Err(format!("history query lost the t_stamp column: {envelope}"));
            }
            let relative = tag
                .trim_start_matches('[')
                .split_once(']')
                .map(|(_, rest)| rest)
                .unwrap_or(tag);
            if !columns
                .iter()
                .any(|column| column.as_str() == Some(relative))
            {
                return Err(format!(
                    "the tag column rides provider-relative: {envelope}"
                ));
            }
            if history_rows_contain(&envelope, 44) {
                println!("SC-4 closure evidence: written value 44 proven in history ({historian})");
                return Ok(());
            }
            last_miss = serde_json::to_string(&envelope["data"]["rows"]).unwrap_or_default();
            if std::time::Instant::now() >= deadline {
                break;
            }
            eprintln!("history: no 44 yet (scan cadence 40–50 s) — rows so far: {last_miss}");
            tokio::time::sleep(std::time::Duration::from_secs(15)).await;
        }
    }
    Err(format!(
        "CLOSURE DRIFT FINDING: the 13-01 recipe live-proved data flow, but 3 self-healing \
         cycles of polling never surfaced the written 44 (last rows: {last_miss}) — record \
         in 13-LIVE-GATE.md (capture wins; never code around it)"
    ))
}

/// THE alarm lifecycle live fixture (05-06, TAGS-07) — configure a
/// tag with a LIST-form alarm (AboveValue 100, High — the
/// research-proven configure shape) → write past the setpoint →
/// poll `alarms active` until the event appears (bounded) → ack with
/// note+username (the 3-arg form) → assert the state flip to
/// `Active, Acknowledged` + the remainder shape → the history leg
/// (a default rig asserts the honest `alarm_journal_missing`
/// refusal LIVE — the pinned default-rig path; a journal-provisioned
/// rig asserts rows) → the journal stretch is SKIP-WITH-REASON (the
/// sidecar-postgres chain is documented in the README; the wiremock
/// pins carry the capability proof).
#[tokio::test]
#[ignore = "opt-in e2e: set IGNITION_LIVE_URL + IGNITION_LIVE_TOKEN + IGNITION_LIVE_MUTATIONS=1"]
async fn live_tags_alarm_lifecycle() {
    let Some(env) = live_env_mutations() else {
        skip(
            "IGNITION_LIVE_MUTATIONS=1 (with URL+TOKEN) not set — refusing to touch a live gateway",
        );
        return;
    };
    let (_dir, config) = isolated_live_config(&env);
    // THE serializer: one live gate at a time (shared gateway state).
    let _gate = LIVE_GATE.lock().await;
    // UNIQUE provider name per run (11-06 live-truth): dozens of
    // same-name delete/recreate cycles on one rig eventually degrade
    // that name's tag model (Error_Configuration persisting minutes —
    // seen on both rigs after repeated suite runs). Disposable rigs
    // get disposable names: each run lands in a virgin namespace.
    let provider = format!(
        "p5alarm{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("clock")
            .as_secs()
    );
    let tag = format!("[{provider}]AlarmTag");

    // Routes first: every alarms verb refuses exit 6 without them.
    let out = ign(&config, &env, &["webdev", "deploy", "--compact"]);
    expect_ok("deploy (the alarms route's precondition)", &out);

    // SELF-HEALING SETUP (11-06 live-truth, both rigs): a recreate
    // can race the old provider model's teardown tail — the fresh
    // provider's tags answer Error_Configuration, sometimes for a
    // minute-plus. The observed recovery is simply another
    // delete→settle→recreate cycle (proven live), so the setup loops:
    // clean → settle → create → config-create → readiness poll; a
    // cycle that never reaches Good tears down and retries (3 cycles
    // × 20s readiness window).
    for _ in 0..3 {
        clean_provider(&config, &env, &provider);
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
        loop {
            let out = ign(&config, &env, &["tags", "provider", "list", "--compact"]);
            expect_ok("provider list (settle poll)", &out);
            let envelope = data_envelope(&out);
            let gone = !envelope["data"]["providers"]
                .as_array()
                .expect("providers[]")
                .iter()
                .any(|p| p["name"] == provider);
            if gone || std::time::Instant::now() >= deadline {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
        }
        let out = ign(
            &config,
            &env,
            &["tags", "provider", "create", &provider, "--compact"],
        );
        expect_ok("provider create p5alarm", &out);
        let out = ign_stdin(
            &config,
            &env,
            &["tags", "config", "create", &tag, "--file", "-", "--compact"],
            r#"{"tagType": "AtomicTag", "dataType": "Int4", "value": 0, "alarms": [{"name": "HighLimit", "enabled": true, "mode": "AboveValue", "setpointA": 100, "priority": "High"}]}"#,
        );
        expect_ok("config create the alarmed tag", &out);
        // 60s per cycle: the FIRST provider creation on a fresh
        // 8.3.3 gateway initializes the tag-provider subsystem and
        // can hold Error_Configuration past a 20s window (measured
        // on the fresh-rig run — three 20s cycles all missed; a
        // later attempt answers in seconds).
        if await_tag_ready(&config, &env, &tag, std::time::Duration::from_secs(60)).await {
            break;
        }
        eprintln!("setup: tag model stuck Error_Configuration, tearing down and retrying…");
    }
    // (The readiness guarantee the rest of the gate relies on — the
    // loop above either achieved it or the gate panics here.)
    assert!(
        await_tag_ready(&config, &env, &tag, std::time::Duration::from_secs(1)).await,
        "the alarmed tag never reached a Good model after 3 setup cycles"
    );

    // Write with LAND-VERIFICATION (same live-truth class): the write
    // answering ok is not proof the value landed while the provider's
    // model mounts — verify by read-back and retry (bounded). Before
    // the FIRST write, wait for READS to answer Good: a recreate can
    // race the old model's teardown tail (list-absence is not model-
    // freed), leaving Error_Configuration reads that clear once the
    // fresh model binds.
    async fn await_tag_ready(
        config: &Path,
        env: &LiveEnv,
        tag: &str,
        budget: std::time::Duration,
    ) -> bool {
        let deadline = std::time::Instant::now() + budget;
        loop {
            let out = ign(config, env, &["tags", "read", tag, "--compact"]);
            expect_ok("read (model-ready poll)", &out);
            let envelope = data_envelope(&out);
            let quality = envelope["data"]["results"][0]["quality"]
                .as_str()
                .unwrap_or_default();
            if quality == "Good" {
                return true;
            }
            if std::time::Instant::now() >= deadline {
                return false;
            }
            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
        }
    }
    async fn write_verified(config: &Path, env: &LiveEnv, tag: &str, value: &str, what: &str) {
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
        loop {
            let out = ign(
                config,
                env,
                &["tags", "write", tag, "--value", value, "--compact"],
            );
            expect_ok(what, &out);
            let out = ign(config, env, &["tags", "read", tag, "--compact"]);
            expect_ok("read-back after write", &out);
            let envelope = data_envelope(&out);
            let landed = envelope["data"]["results"][0]["value"].as_i64();
            if landed == value.parse::<i64>().ok() {
                return;
            }
            assert!(
                std::time::Instant::now() < deadline,
                "{what}: the write never landed (read-back kept disagreeing)"
            );
            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
        }
    }

    // (The model-ready gate passed inside the setup loop; writes
    // below assume it.)

    // Trigger: write past the setpoint (land-verified).
    write_verified(&config, &env, &tag, "150", "write past the setpoint").await;

    // Poll `alarms active` until the event appears (bounded).
    let find_event = |envelope: &Value| -> Option<Value> {
        envelope["data"]["alarms"].as_array().and_then(|alarms| {
            alarms
                .iter()
                .find(|alarm| {
                    alarm["source"]
                        .as_str()
                        .is_some_and(|source| source.contains(&provider))
                })
                .cloned()
        })
    };
    let mut event = None;
    // POLL + RE-TRIGGER (11-06 live-truth, Rig B 8.3.3): after
    // creating the provider, the alarm engine can take a
    // LONG bounded window to register the new tag's alarm config —
    // observed: 45s+ of silent ok-polls (the config verifiably landed
    // in the model), with events flowing normally on the next attempt
    // ~50s later; warm rigs answer in 2–4s. The RE-TRIGGER (write
    // below the setpoint, then past it again — both land-verified)
    // keeps a real transition available for whichever poll window the
    // registered engine finally sees. 10 attempts ≈ 100s worst case;
    // warm runs never leave attempt 1.
    for attempt in 0..10 {
        if attempt > 0 {
            write_verified(
                &config,
                &env,
                &tag,
                "0",
                "write below the setpoint (re-trigger clear)",
            )
            .await;
            write_verified(
                &config,
                &env,
                &tag,
                "150",
                "write past the setpoint (re-trigger set)",
            )
            .await;
        }
        for _ in 0..8 {
            let out = ign(&config, &env, &["tags", "alarms", "active", "--compact"]);
            expect_ok("alarms active poll", &out);
            let envelope = data_envelope(&out);
            if let Some(found) = find_event(&envelope) {
                event = Some(found);
                break;
            }
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        }
        if event.is_some() {
            break;
        }
    }
    let event = event.expect("the alarm event appeared (bounded retries)");

    let event_id = event["event_id"]
        .as_str()
        .expect("event_id present")
        .to_string();
    assert!(
        event["state"]
            .as_str()
            .unwrap_or_default()
            .contains("Unacknowledged"),
        "the fresh event is Active, Unacknowledged: {event}"
    );
    assert_eq!(event["priority"], "High", "{event}");
    println!(
        "alarm active: {event_id} {}",
        event["state"].as_str().unwrap_or_default()
    );

    // Acknowledge: the 3-arg form (note + username explicit).
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "alarms",
            "ack",
            &event_id,
            "--note",
            "e2e lifecycle ack",
            "--username",
            "ign-e2e",
            "--compact",
        ],
    );
    expect_ok("alarms ack (3-arg form)", &out);
    let envelope = data_envelope(&out);
    assert_eq!(envelope["data"]["acknowledged"], 1, "{envelope}");
    assert_eq!(
        envelope["data"]["unacknowledged"].as_array().map(Vec::len),
        Some(0),
        "empty remainder: {envelope}"
    );

    // The state flip: poll active until 'Active, Acknowledged'.
    let mut acknowledged = false;
    for _ in 0..20 {
        let out = ign(&config, &env, &["tags", "alarms", "active", "--compact"]);
        expect_ok("alarms active poll (ack)", &out);
        let envelope = data_envelope(&out);
        if let Some(event) = find_event(&envelope)
            && event["state"]
                .as_str()
                .is_some_and(|state| state.contains("Acknowledged"))
        {
            assert!(
                event["state"]
                    .as_str()
                    .unwrap_or_default()
                    .contains("Active"),
                "the acked alarm stays active until cleared: {event}"
            );
            acknowledged = true;
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    }
    assert!(acknowledged, "the event flipped to 'Active, Acknowledged'");
    println!("alarm acked: state flipped to Active, Acknowledged");

    // The history leg: a DEFAULT rig asserts the honest refusal LIVE
    // (exit 6 + alarm_journal_missing — the pinned default-rig
    // path); a journal-provisioned rig asserts rows instead.
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "alarms",
            "history",
            "--start",
            &(now_ms() - 600_000).to_string(),
            "--end",
            &now_ms().to_string(),
            "--compact",
        ],
    );
    if out.status.code() == Some(6) {
        let envelope = err_envelope(&out);
        assert_eq!(
            envelope["error"]["code"], "alarm_journal_missing",
            "the default-rig refusal is the journal slug: {envelope}"
        );
        println!(
            "alarm history: default-rig honest refusal pinned live \
             (alarm_journal_missing, exit 6 — the actionable journal-chain hint)"
        );
    } else {
        expect_ok("alarms history (journal-provisioned rig)", &out);
        let envelope = data_envelope(&out);
        println!(
            "alarm history: journal answered ({} row(s)) — the provisioned path",
            envelope["data"]["count"]
        );
    }

    // The journal stretch: SKIP-WITH-REASON (the documented
    // sidecar-postgres chain lives in the README; the wiremock pins
    // carry the capability proof).
    println!(
        "journal provisioning stretch: skipped — the chain (sidecar postgres container → \
         database-connection resource → alarm-journal profile → general-alarm-settings \
         singleton) is documented in the README 'Alarm history' section; provision it \
         manually and re-run to walk the provisioned path above"
    );

    // Cleanup (self-cleaning): quiet the tag value, delete the tag,
    // delete the provider.
    let _ = ign(
        &config,
        &env,
        &["tags", "write", &tag, "--value", "0", "--compact"],
    );
    let out = ign(
        &config,
        &env,
        &["tags", "config", "delete", &tag, "--yes", "--compact"],
    );
    expect_ok("config delete the alarmed tag", &out);
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "provider",
            "delete",
            &provider,
            "--yes",
            "--compact",
        ],
    );
    expect_ok("provider delete (guarded)", &out);
}

// ---- 11-06 live gates: the XML fidelity round-trip (the
// determinism-capture-selected oracle), the CSV round-trip (the
// coverage table live-diff), and the loss-gate refusal
// (pre-resolution, read-only). Oracle authority:
// .planning/phases/11-tag-bulk-transfer-xml-csv/11-LIVE-CAPTURES.md
// Probe 3. ----

use quick_xml::XmlVersion;
use quick_xml::events::{BytesStart, Event};
use quick_xml::reader::Reader;

/// sha256 hex — the fidelity-oracle evidence values recorded in
/// 11-LIVE-GATE.md (transport A, round-trip B, per rig).
fn sha256_hex(bytes: &[u8]) -> String {
    use sha2::{Digest, Sha256};
    let mut hasher = Sha256::new();
    hasher.update(bytes);
    hasher
        .finalize()
        .iter()
        .map(|b| format!("{b:02x}"))
        .collect()
}

/// One canonical-XML node: element name, SORTED attributes,
/// ORDER-NORMALIZED children, concatenated meaningful text.
/// Whitespace-only text (the 3-space indentation) is dropped — the
/// gateway's own CRLF documents differ only in sibling order after
/// an import (11-01 Probe 3: "sibling order permutes on the model
/// rebuild"), so the structural oracle compares canonical SERIALIZATIONS
/// of order-sorted trees, never raw bytes.
#[derive(Debug, Clone, PartialEq, Eq)]
struct Canon {
    name: String,
    attrs: Vec<(String, String)>,
    children: Vec<Canon>,
    text: String,
}

/// All attributes, normalized + sorted (the tag_loss `collect_attrs`
/// idiom). Sorting makes attribute order irrelevant to identity —
/// gateway serializers are stable, but the oracle should not depend
/// on it.
fn canon_attrs(element: &BytesStart) -> Vec<(String, String)> {
    let mut attrs: Vec<(String, String)> = element
        .attributes()
        .flatten()
        .filter_map(|attr| {
            attr.normalized_value(XmlVersion::Implicit1_0)
                .ok()
                .map(|v| {
                    (
                        String::from_utf8_lossy(attr.key.as_ref()).into_owned(),
                        v.into_owned(),
                    )
                })
        })
        .collect();
    attrs.sort();
    attrs
}

/// The canonical serialization: `<name k="v">text<sorted-children/></name>`
/// (self-closing when empty). Siblings sort by their own serialized
/// form — a total order, so the sort is stable across runs and rigs.
fn serialize_canonical(node: &Canon) -> String {
    let mut out = format!("<{}", node.name);
    for (key, value) in &node.attrs {
        out.push_str(&format!(" {key}=\"{value}\""));
    }
    let mut kids: Vec<String> = node.children.iter().map(serialize_canonical).collect();
    kids.sort();
    if kids.is_empty() && node.text.is_empty() {
        out.push_str("/>");
        return out;
    }
    out.push('>');
    out.push_str(&node.text);
    for kid in kids {
        out.push_str(&kid);
    }
    out.push_str(&format!("</{}>", node.name));
    out
}

/// Parse gateway XML into the canonical tree (the tag_loss Reader
/// idioms; the oracle inputs are gateway exports — a parse error is
/// a gate bug, never an expected case).
fn parse_canonical(raw: &[u8]) -> Canon {
    let mut reader = Reader::from_reader(raw);
    let mut stack: Vec<Canon> = Vec::new();
    let mut buf = Vec::new();
    loop {
        buf.clear();
        match reader.read_event_into(&mut buf) {
            Ok(Event::Start(ref e)) => stack.push(Canon {
                name: String::from_utf8_lossy(e.name().as_ref()).into_owned(),
                attrs: canon_attrs(e),
                children: Vec::new(),
                text: String::new(),
            }),
            Ok(Event::Empty(ref e)) => {
                let node = Canon {
                    name: String::from_utf8_lossy(e.name().as_ref()).into_owned(),
                    attrs: canon_attrs(e),
                    children: Vec::new(),
                    text: String::new(),
                };
                stack
                    .last_mut()
                    .expect("Empty event always has a parent")
                    .children
                    .push(node);
            }
            Ok(Event::End(_)) => {
                let node = stack.pop().expect("balanced document");
                match stack.last_mut() {
                    Some(parent) => parent.children.push(node),
                    None => return node,
                }
            }
            Ok(Event::Text(ref t)) => {
                let Ok(text) = t.decode() else { continue };
                if !text.trim().is_empty()
                    && let Some(parent) = stack.last_mut()
                {
                    parent.text.push_str(text.trim());
                }
            }
            Ok(Event::Eof) => break,
            Ok(_) => {}
            Err(err) => panic!("canonicalizer: oracle inputs are gateway exports: {err}"),
        }
    }
    panic!("canonicalizer: unbalanced document (no root End event)");
}

/// The ORDER-NORMALIZED STRUCTURAL IDENTITY oracle (binds 11-06 per
/// 11-LIVE-CAPTURES.md Probe 3's determinism answer): byte-identity
/// is proven ONLY for unchanged-subtree re-export (capture (a));
/// import→re-export permutes sibling order on the model rebuild
/// (capture (b) — 1128==1128 length, byte-different, structurally
/// identical), so byte-identity is INVALID here by capture. The
/// cheap length pre-check rides at the call site.
fn structurally_identical(a: &[u8], b: &[u8]) -> bool {
    serialize_canonical(&parse_canonical(a)) == serialize_canonical(&parse_canonical(b))
}

/// ONE raw wire probe of the deployed tagConfig route — POST the
/// exportTags action for `paths` (token auth; the secret gates
/// scriptExec ONLY — the always-on routes are token-gated) and
/// return the DECODED payload bytes. This is the transport
/// assertion's cleaner capture choice (per the plan's either/or):
/// the raw route response pins the EXACT `payload_b64` the CLI's own
/// export decodes, so `sha256(file) == sha256(decoded payload_b64)`
/// witnesses the whole CLI transport (route → base64 → decode →
/// file) in one assertion — a second CLI export would only prove
/// two in-process decodes agree.
async fn raw_tagconfig_export_xml(env: &LiveEnv, paths: &[&str]) -> Vec<u8> {
    install_crypto_provider();
    let client = reqwest::Client::new();
    let url = format!("{}/system/webdev/ign-cli/cli/tagConfig", env.url);
    let response = client
        .post(&url)
        .header("X-Ignition-API-Token", &env.token)
        .json(&serde_json::json!({
            "action": "exportTags",
            "paths": paths,
            "format": "xml"
        }))
        .send()
        .await
        .expect("tagConfig exportTags reaches the gateway");
    assert_eq!(
        response.status().as_u16(),
        200,
        "the webdev servlet answers 200"
    );
    let body: Value = response.json().await.expect("envelope parses");
    assert_eq!(body["ok"], Value::Bool(true), "export answered ok: {body}");
    let payload_b64 = body["data"]["payload_b64"]
        .as_str()
        .unwrap_or_else(|| panic!("payload_b64 present: {body}"));
    use base64::Engine as _;
    base64::engine::general_purpose::STANDARD
        .decode(payload_b64)
        .expect("payload_b64 decodes")
}

/// THE XML fidelity round-trip live gate (11-06, TAGS-10) — the
/// roadmap SC-1 proof on a REAL multi-level UDT subtree: seed the
/// probe-3-shaped structure (MotorType UDT definition under
/// `[default]_types_` + a P11Live folder of two instances and a
/// nested Sub third), export XML, and assert the capture-selected
/// oracle at every hop:
///
/// 1. **transport** — `sha256(a.xml) == sha256(decoded route
///    payload_b64)` (raw wire probe, comment at the helper);
/// 2. **refusal** — the loss gate refuses exit 2 WITHOUT `--yes`
///    (gateway XML always carries MinVersion → the
///    `xml_export_edited_only` fact), pre-resolution;
/// 3. **zero mutation** — the re-export after the refusal is
///    BYTE-IDENTICAL (unchanged-subtree export determinism, capture
///    (a), live re-proven — which doubles as the zero-write proof);
/// 4. **import** — `--yes` proceeds; the QualityCode backstop
///    (`data.failed`) stays empty;
/// 5. **round-trip** — the imported subtree re-exports at EQUAL
///    LENGTH and is ORDER-NORMALIZED STRUCTURALLY IDENTICAL (capture
///    (b): byte-identity is invalid after an import — sibling order
///    permutes on the model rebuild; shas printed for the record).
///
/// Cleanup deletes the seeded paths + the UDT type + the scratch
/// provider (the CLI's import target is PROVIDER-scoped — `--provider`,
/// the shipped 11-05 surface — so the import lands at `[p11live]P11Live`
/// and the gate exports that path; the plan's basePath-positional
/// sketch reconciled to the actual CLI, per its own "adjust the
/// exported path to whatever the import actually landed" note).
#[tokio::test]
#[ignore = "opt-in e2e: set IGNITION_LIVE_URL + IGNITION_LIVE_TOKEN + IGNITION_LIVE_MUTATIONS=1"]
async fn live_tags_xml_fidelity_roundtrip() {
    let Some(env) = live_env_mutations() else {
        skip(
            "IGNITION_LIVE_MUTATIONS=1 (with URL+TOKEN) not set — refusing to touch a live gateway",
        );
        return;
    };
    let (_dir, config) = isolated_live_config(&env);
    // THE serializer: one live gate at a time (shared gateway state).
    let _gate = LIVE_GATE.lock().await;

    // (1) Self-deploy FIRST (the redeploy-sequencing pin) with the
    // scriptExec leg so the rig keeps the full standing 5-route
    // bundle — then the version handshake reads the CURRENT bundle
    // version (1.3.0: the 11-06 importTagsFile format-suffix fix) on
    // every route.
    let out = ign(
        &config,
        &env,
        &["webdev", "deploy", "--with-script-exec", "--compact"],
    );
    expect_ok("deploy (the gate's own precondition)", &out);
    let out = ign(&config, &env, &["webdev", "status", "--compact"]);
    expect_ok("status after self-deploy", &out);
    let envelope = data_envelope(&out);
    assert_eq!(envelope["data"]["ok"], Value::Bool(true), "{envelope}");
    for route in ["tags", "tagConfig", "alarms", "tagHistory", "scriptExec"] {
        let row = status_row(&envelope, route);
        assert_eq!(row["status"], "present", "{route}: {row}");
        assert_eq!(
            row["deployed_version"],
            ignition_core::webdev::ROUTE_BUNDLE_VERSION,
            "{route}: the current-bundle handshake pin: {row}"
        );
    }

    // Pre-clean leftovers from a prior aborted run (idempotent): the
    // kept-alive 11-01 rigs carry P11UDT*/P11Seed* probe tags — our
    // namespace is P11Live* (no collision), plus the MotorType the
    // 11-01 probes may have left under _types_.
    clean_tag_configs(
        &config,
        &env,
        &["[default]P11Live", "[default]_types_/MotorType"],
    );
    clean_provider(&config, &env, "p11live");

    // (2) Seed the probe-3 construction (config files via tempfile):
    // the MotorType UDT definition (parameter + OPC child with an
    // alarm + expression child) and the P11Live folder (two
    // instances + nested Sub third). Parameter values ride the
    // `value` key (the captured construction finding — `defaultValue`
    // is silently dropped); the alarm's `mode` key is omitted (it
    // does not bind — captured); the property bindings ride the
    // research-recorded JSON dict form.
    let tmp = tempfile::tempdir().expect("tempdir");
    let type_def = tmp.path().join("motor-type.json");
    std::fs::write(
        &type_def,
        serde_json::json!({
            "tagType": "UdtType",
            "parameters": {"MotorNumber": {"dataType": "Integer", "value": 1}},
            "tags": [
                {
                    "name": "Amps",
                    "tagType": "AtomicTag",
                    "valueSource": "opc",
                    "opcServer": "Ignition OPC-UA Server",
                    "opcItemPath": {
                        "bindType": "parameter",
                        "binding": "ns=1;s=[Dairy]Motor {MotorNumber}/Amps"
                    },
                    "alarms": [
                        {"name": "Low Amps", "priority": "High", "setpointA": 25}
                    ]
                },
                {
                    "name": "Doubled",
                    "tagType": "AtomicTag",
                    "valueSource": "expression",
                    "expression": {"bindType": "parameter", "binding": "{MotorNumber} * 2"}
                }
            ]
        })
        .to_string(),
    )
    .expect("write type def");
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "config",
            "create",
            "[default]_types_/MotorType",
            "--file",
            type_def.to_str().expect("path"),
            "--compact",
        ],
    );
    expect_ok("config create the MotorType UDT definition", &out);

    let folder_def = tmp.path().join("p11live.json");
    std::fs::write(
        &folder_def,
        serde_json::json!({
            "tagType": "Folder",
            "tags": [
                {
                    "name": "M1",
                    "tagType": "UdtInstance",
                    "udtParentType": "[default]_types_/MotorType",
                    "parameters": {"MotorNumber": {"dataType": "Integer", "value": 1}}
                },
                {
                    "name": "M2",
                    "tagType": "UdtInstance",
                    "udtParentType": "[default]_types_/MotorType",
                    "parameters": {"MotorNumber": {"dataType": "Integer", "value": 2}}
                },
                {
                    "name": "Sub",
                    "tagType": "Folder",
                    "tags": [
                        {
                            "name": "M3",
                            "tagType": "UdtInstance",
                            "udtParentType": "[default]_types_/MotorType",
                            "parameters": {"MotorNumber": {"dataType": "Integer", "value": 3}}
                        }
                    ]
                }
            ]
        })
        .to_string(),
    )
    .expect("write folder def");
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "config",
            "create",
            "[default]P11Live",
            "--file",
            folder_def.to_str().expect("path"),
            "--compact",
        ],
    );
    expect_ok("config create the P11Live folder", &out);

    // (3) Export XML (single path — instance-only: the TYPE lives
    // under _types_, so the file carries NO UdtType definition and
    // importTags accepts it; capture (b).2).
    let a_path = tmp.path().join("a.xml");
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "export",
            "[default]P11Live",
            "--format",
            "xml",
            "-o",
            a_path.to_str().expect("path"),
            "--compact",
        ],
    );
    expect_ok("export [default]P11Live as xml", &out);
    let a_bytes = std::fs::read(&a_path).expect("a.xml readable");
    assert!(!a_bytes.is_empty(), "the export is not empty");
    let sha_a = sha256_hex(&a_bytes);

    // ORACLE TIER 1 — transport fidelity: the file's bytes == the
    // route's payload_b64 bytes (the raw probe captures the route
    // response directly — see the helper's comment).
    let route_bytes = raw_tagconfig_export_xml(&env, &["[default]P11Live"]).await;
    let sha_route = sha256_hex(&route_bytes);
    assert_eq!(
        sha_a, sha_route,
        "TRANSPORT FIDELITY: sha256(export file) == sha256(route payload_b64)"
    );
    println!(
        "xml transport fidelity: sha256 {sha_a} ({} bytes)",
        a_bytes.len()
    );

    // (4) The loss gate refuses WITHOUT --yes: exit 2, invalid_input,
    // the report prose on stderr (the MinVersion fact fires on any
    // gateway-export-shaped file) — and PRE-RESOLUTION (profile null,
    // zero wire work).
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "import",
            "--file",
            a_path.to_str().expect("path"),
            "--format",
            "xml",
            "--provider",
            "p11live",
            "--compact",
        ],
    );
    assert_eq!(
        out.status.code(),
        Some(2),
        "the loss gate refuses exit 2; stdout: {} stderr: {}",
        String::from_utf8_lossy(&out.stdout),
        String::from_utf8_lossy(&out.stderr)
    );
    let envelope = err_envelope(&out);
    assert_eq!(envelope["error"]["code"], "invalid_input", "{envelope}");
    assert!(
        envelope["profile"].is_null(),
        "the refusal is PRE-resolution — no profile resolved: {envelope}"
    );
    let message = envelope["error"]["message"].as_str().unwrap_or_default();
    assert!(
        message.contains("loss report (xml)") && message.contains("re-run with --yes"),
        "the refusal message IS the report prose: {envelope}"
    );

    // (5) ZERO-MUTATION proof: re-export and byte-compare — the
    // unchanged-subtree export determinism (capture (a), live
    // re-proven) IS the nothing-landed evidence.
    let a2_path = tmp.path().join("a2.xml");
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "export",
            "[default]P11Live",
            "--format",
            "xml",
            "-o",
            a2_path.to_str().expect("path"),
            "--compact",
        ],
    );
    expect_ok("re-export after refusal (the zero-mutation read)", &out);
    let a2_bytes = std::fs::read(&a2_path).expect("a2.xml readable");
    assert_eq!(
        sha256_hex(&a2_bytes),
        sha_a,
        "ZERO MUTATION: the refusal left the seeded subtree byte-identical"
    );

    // (6) The import target: a dedicated scratch provider (the CLI
    // surface is provider-scoped; native REST create, no route).
    let out = ign(
        &config,
        &env,
        &["tags", "provider", "create", "p11live", "--compact"],
    );
    expect_ok("provider create p11live", &out);

    // THE TYPE DEFINITION RIDES THE TRANSFER (11-06 live-truth, Rig B
    // 8.3.3): importTags SILENTLY DROPS UdtInstance parameter
    // overrides when the referenced type is not resolvable in the
    // TARGET provider — the import answers Good (failed empty) and
    // lands EMPTY `<Parameters/>` blocks (8.3.6 preserves them
    // regardless; overwrite re-import does NOT heal the drop). A
    // UDT-instance migration without its type definition is not a
    // faithful transfer anyway, so the gate provisions MotorType at
    // the target via the config surface — the capture's importTags
    // UdtType REFUSAL applies to type XML inside import files, not to
    // the config-create route (probe 3's construction precedent).
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "config",
            "create",
            "[p11live]_types_/MotorType",
            "--file",
            type_def.to_str().expect("path"),
            "--compact",
        ],
    );
    expect_ok("config create the MotorType at the import target", &out);

    // The guarded import WITH --yes: exit 0, no server-side
    // Bad_Failure backstop, the loss report rides the envelope.
    // (The mount-race tolerance rides the helper — see its comment.)
    let import_args = [
        "tags",
        "import",
        "--file",
        a_path.to_str().expect("path"),
        "--format",
        "xml",
        "--provider",
        "p11live",
        "--yes",
        "--compact",
    ];
    let out = import_with_mount_tolerance(&config, &env, &import_args);
    expect_ok("import --yes (the guarded commit)", &out);
    let envelope = data_envelope(&out);
    assert_eq!(envelope["data"]["format"], "xml", "{envelope}");
    assert_eq!(
        envelope["data"]["failed"].as_array().map(Vec::len),
        Some(0),
        "the QualityCode backstop stays empty: {envelope}"
    );
    assert!(
        envelope["data"]["loss_report"].is_object(),
        "--yes attaches data.loss_report: {envelope}"
    );
    assert!(
        envelope["data"]["top_level_names"]
            .as_array()
            .expect("top_level_names[]")
            .iter()
            .any(|name| name == "P11Live"),
        "the scan named the landing subtree: {envelope}"
    );

    // LANDING VERIFICATION (11-06 Rig B live truth): a clean
    // failed-empty answer against a JUST-CREATED provider can still
    // be a SILENT NO-OP (nothing lands — the import raced the mount;
    // the type=Unknown export shells are the tell). Verify by browse
    // and re-import (state-safe) bounded.
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
    loop {
        let probe = read_with_provider_tolerance(
            &config,
            &env,
            &["tags", "browse", "[p11live]", "--compact"],
        );
        let landed = probe.status.success()
            && data_envelope(&probe)["data"]["entries"]
                .as_array()
                .map(|entries| entries.iter().any(|row| row["name"] == "P11Live"))
                .unwrap_or(false);
        if landed {
            break;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "the xml import never landed (browse kept showing an empty root)"
        );
        eprintln!("import: clean answer but nothing landed, re-importing (bounded 30s)…");
        let out = import_with_mount_tolerance(&config, &env, &import_args);
        expect_ok("import --yes (re-import)", &out);
    }

    // (7) Re-export the IMPORTED subtree (it landed at
    // [p11live]P11Live — provider-root basePath) and run the
    // capture-selected oracle: EQUAL LENGTH pre-check + order-
    // normalized structural identity. Byte-identity is INVALID here
    // by capture (b) — sibling order permutes on the model rebuild —
    // so the shas are printed for the evidence doc, not asserted.
    let b_path = tmp.path().join("b.xml");
    let out = read_with_provider_tolerance(
        &config,
        &env,
        &[
            "tags",
            "export",
            "[p11live]P11Live",
            "--format",
            "xml",
            "-o",
            b_path.to_str().expect("path"),
            "--compact",
        ],
    );
    expect_ok("export the imported [p11live]P11Live", &out);
    let b_bytes = std::fs::read(&b_path).expect("b.xml readable");
    assert_eq!(
        a_bytes.len(),
        b_bytes.len(),
        "LENGTH pre-check (capture (b): 1128 == 1128): a={} b={}",
        a_bytes.len(),
        b_bytes.len()
    );
    let sha_b = sha256_hex(&b_bytes);
    assert!(
        structurally_identical(&a_bytes, &b_bytes),
        "ROUND-TRIP ORACLE: order-normalized structural identity failed \
         (sha_a {sha_a} vs sha_b {sha_b})"
    );
    println!(
        "xml round-trip: sha_a {sha_a} / sha_b {sha_b} ({} bytes each) — \
         structurally identical (byte-{})",
        a_bytes.len(),
        if sha_a == sha_b {
            "identical too"
        } else {
            "different: sibling order permuted on rebuild, per capture (b)"
        }
    );

    // (8) Cleanup: the seeded paths + the UDT type + the scratch
    // provider (the imported copy rides the provider delete).
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "config",
            "delete",
            "[default]P11Live",
            "[default]_types_/MotorType",
            "--yes",
            "--compact",
        ],
    );
    expect_ok("config delete cleanup (seeded paths + type)", &out);
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "provider",
            "delete",
            "p11live",
            "--yes",
            "--compact",
        ],
    );
    expect_ok("provider delete p11live (guarded)", &out);
}

/// THE CSV round-trip live gate (11-06, TAGS-11) — the roadmap SC-2
/// proof: a seeded subtree exports as CLI-GENERATED legacy CSV (the
/// gateway cannot), imports through `importTagsFile`, and the
/// re-exported JSON's field diff matches the documented lossy
/// coverage table (11-LIVE-CAPTURES.md §Probe 5) — the live diff IS
/// the table verification:
///
/// - **LANDED**: memory `value` (42), `tooltip`, `engUnit`, the
///   expression (`1+1`);
/// - **SILENTLY DROPPED**: alarms (the seeded alarm is ABSENT from
///   the imported model — the live proof of "CSV does not include
///   support for alarm configurations", stronger: silent);
/// - **per-column legacy-sheet materialization** (11-06 live-truth
///   refinement of the probe-5 note): every column PRESENT in the
///   CSV header materializes its model default (`deadband`,
///   `rawHigh`, …) that configure-created tags do NOT carry — and
///   the generated header's missing Alert* columns mean NO Alert
///   keys materialize (probe 5's "AlertAckMode … regardless"
///   overgeneralized; its headers carried Alert columns).
#[tokio::test]
#[ignore = "opt-in e2e: set IGNITION_LIVE_URL + IGNITION_LIVE_TOKEN + IGNITION_LIVE_MUTATIONS=1"]
async fn live_tags_csv_roundtrip() {
    let Some(env) = live_env_mutations() else {
        skip(
            "IGNITION_LIVE_MUTATIONS=1 (with URL+TOKEN) not set — refusing to touch a live gateway",
        );
        return;
    };
    let (_dir, config) = isolated_live_config(&env);
    // THE serializer: one live gate at a time (shared gateway state).
    let _gate = LIVE_GATE.lock().await;

    // Self-deploy FIRST (the redeploy-sequencing pin) — keep the
    // standing 5-route bundle.
    let out = ign(
        &config,
        &env,
        &["webdev", "deploy", "--with-script-exec", "--compact"],
    );
    expect_ok("deploy (the gate's own precondition)", &out);

    // Pre-clean leftovers (idempotent re-runs).
    clean_tag_configs(&config, &env, &["[default]P11CsvSeed"]);
    clean_provider(&config, &env, "p11csv");

    // Seed: a folder with one memory tag (tooltip + engUnit), one
    // expression tag, one ALARMED tag — the three coverage-table
    // classes in one subtree.
    let tmp = tempfile::tempdir().expect("tempdir");
    let seed = tmp.path().join("seed.json");
    std::fs::write(
        &seed,
        serde_json::json!({
            "tagType": "Folder",
            "tags": [
                {
                    "name": "TMem",
                    "tagType": "AtomicTag",
                    "dataType": "Int4",
                    "value": 42,
                    "tooltip": "e2e csv",
                    "engUnit": "PSI"
                },
                {
                    "name": "TExpr",
                    "tagType": "AtomicTag",
                    "valueSource": "expression",
                    "expression": "1+1"
                },
                {
                    "name": "TAlarmed",
                    "tagType": "AtomicTag",
                    "dataType": "Int4",
                    "value": 0,
                    "alarms": [
                        {"name": "HighLimit", "mode": "AboveValue", "setpointA": 100, "priority": "High"}
                    ]
                }
            ]
        })
        .to_string(),
    )
    .expect("write seed");
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "config",
            "create",
            "[default]P11CsvSeed",
            "--file",
            seed.to_str().expect("path"),
            "--compact",
        ],
    );
    expect_ok("config create the P11CsvSeed folder", &out);

    // The ORIGINAL JSON export — the legacy-sheet contrast baseline
    // (configure-created tags carry NO AlertAckMode). Same-parent
    // LEAF paths: a multi-path export answers the `{"tags":[...]}`
    // wrapper (single-path would answer the folder subtree itself).
    let orig_path = tmp.path().join("orig.json");
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "export",
            "[default]P11CsvSeed/TMem",
            "[default]P11CsvSeed/TExpr",
            "[default]P11CsvSeed/TAlarmed",
            "--format",
            "json",
            "-o",
            orig_path.to_str().expect("path"),
            "--compact",
        ],
    );
    expect_ok("export the original as json", &out);
    // The export FILE holds the normalized LIST-OF-SUBTREES (the
    // CLI's interchange format — the raw gateway wrapper never
    // reaches the file).
    let orig: Vec<Value> =
        serde_json::from_slice(&std::fs::read(&orig_path).expect("orig.json readable"))
            .expect("orig.json parses as a subtree list");
    let orig_tmem = orig
        .iter()
        .find(|tag| tag["name"] == "TMem")
        .expect("TMem in the original export");
    // Baseline: configure-created tags carry NONE of the legacy
    // sheet's keys.
    for sheet_key in ["AlertAckMode", "deadband", "rawHigh", "tagGroup"] {
        assert!(
            orig_tmem.get(sheet_key).is_none(),
            "baseline: configure-created tags carry no '{sheet_key}': {orig_tmem}"
        );
    }

    // CLI-GENERATED CSV export (the gateway cannot; the lossy
    // warnings ride stderr — the warn-and-continue posture).
    let c_path = tmp.path().join("c.csv");
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "export",
            "[default]P11CsvSeed",
            "--format",
            "csv",
            "-o",
            c_path.to_str().expect("path"),
            "--compact",
        ],
    );
    expect_ok("export the seed as generated csv", &out);
    let csv_text = std::fs::read_to_string(&c_path).expect("c.csv readable");
    assert!(
        csv_text.starts_with("Path,Name,") && csv_text.contains("# version=1"),
        "the generated CSV carries the legacy header + version marker"
    );

    // Scratch provider + the guarded CSV import (--yes: the csv scan
    // ALWAYS reports csv_no_alarms + csv_legacy_columns_only on
    // non-empty files). CAPTURE FACT: the CSV basePath is IGNORED —
    // rows land at the PROVIDER ROOT regardless ([p11csv]TMem, …).
    // (The mount-race tolerance rides the helper — see its comment;
    // the overwrite policy makes each retry converge.)
    let out = ign(
        &config,
        &env,
        &["tags", "provider", "create", "p11csv", "--compact"],
    );
    expect_ok("provider create p11csv", &out);
    let import_args = [
        "tags",
        "import",
        "--file",
        c_path.to_str().expect("path"),
        "--format",
        "csv",
        "--provider",
        "p11csv",
        "--collision-policy",
        "overwrite",
        "--yes",
        "--compact",
    ];
    let out = import_with_mount_tolerance(&config, &env, &import_args);
    expect_ok("import the generated csv --yes", &out);
    let envelope = data_envelope(&out);
    assert_eq!(envelope["data"]["format"], "csv", "{envelope}");
    assert_eq!(
        envelope["data"]["failed"].as_array().map(Vec::len),
        Some(0),
        "all-or-nothing import answered clean: {envelope}"
    );

    // LANDING VERIFICATION (11-06 Rig B live truth): a clean
    // failed-empty answer against a JUST-CREATED provider can still
    // be a SILENT NO-OP — nothing lands (the import raced the
    // provider model mount and lost its write buffer; probe 4's
    // type="Unknown" export shells are the tell). Verify the landing
    // by browse and re-import (overwrite — state-safe) bounded.
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
    loop {
        let probe = read_with_provider_tolerance(
            &config,
            &env,
            &["tags", "browse", "[p11csv]", "--compact"],
        );
        let landed = probe.status.success()
            && data_envelope(&probe)["data"]["entries"]
                .as_array()
                .map(|entries| entries.iter().any(|row| row["name"] == "TMem"))
                .unwrap_or(false);
        if landed {
            break;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "the csv import never landed (browse kept showing an empty root)"
        );
        eprintln!("import: clean answer but nothing landed, re-importing (bounded 30s)…");
        let out = import_with_mount_tolerance(&config, &env, &import_args);
        expect_ok("import the generated csv --yes (re-import)", &out);
    }

    // Export the imported model back as JSON and diff against the
    // coverage table. (The read rides the provider-resolution
    // tolerance — see the helper's comment.)
    let back_path = tmp.path().join("back.json");
    let out = read_with_provider_tolerance(
        &config,
        &env,
        &[
            "tags",
            "export",
            "[p11csv]TMem",
            "[p11csv]TExpr",
            "[p11csv]TAlarmed",
            "--format",
            "json",
            "-o",
            back_path.to_str().expect("path"),
            "--compact",
        ],
    );
    expect_ok("export the imported tags as json", &out);
    let back: Vec<Value> =
        serde_json::from_slice(&std::fs::read(&back_path).expect("back.json readable"))
            .expect("back.json parses as a subtree list");
    let find = |name: &str| -> Value {
        back.iter()
            .find(|tag| tag["name"] == name)
            .unwrap_or_else(|| panic!("{name} in the imported export: {back:?}"))
            .clone()
    };
    // LANDED (coverage table): value + tooltip + engUnit.
    let tmem = find("TMem");
    assert_eq!(tmem["value"], 42, "CSV Value lands: {tmem}");
    assert_eq!(tmem["tooltip"], "e2e csv", "CSV Tooltip lands: {tmem}");
    assert_eq!(tmem["engUnit"], "PSI", "CSV EngUnit lands: {tmem}");
    // LANDED: the expression.
    assert_eq!(find("TExpr")["expression"], "1+1", "CSV Expression lands");
    // SILENTLY DROPPED: the alarm never arrives (csv_no_alarms —
    // the live proof the docs' one-liner understates: silent).
    let tal = find("TAlarmed");
    assert!(
        tal.get("alarms").is_none(),
        "CSV alarms are SILENTLY dropped — none may arrive: {tal}"
    );
    // Legacy-default materialization — LIVE-TRUTH REFINED (11-06 rig
    // run; corrects the 11-01 probe-5 note): the sheet is PER-COLUMN —
    // a column PRESENT in the CSV header materializes its model
    // default; a column absent from the header materializes nothing.
    // The generated 48-col docs-sample header has NO Alert* columns,
    // so AlertAckMode/AlertSendClear never appear (probe 5's
    // "materializes AlertAckMode … regardless" overgeneralized — its
    // probe headers carried Alert*/AlarmStates columns). The
    // deterministic sheet for OUR header set:
    for sheet_key in [
        "deadband",
        "rawHigh",
        "scaledHigh",
        "engHigh",
        "formatString",
        "historicalDeadband",
        "historicalDeadbandStyle",
        "historyMaxAge",
        "historyTagGroup",
        "tagGroup",
    ] {
        assert!(
            tmem.get(sheet_key).is_some(),
            "CSV import materializes '{sheet_key}' (column present in the header): {tmem}"
        );
    }
    assert!(
        tmem.get("AlertAckMode").is_none(),
        "no Alert* column in the header → no Alert key materializes: {tmem}"
    );
    println!(
        "csv round-trip diff vs coverage table: value/tooltip/engUnit/expression LANDED; \
         alarms SILENTLY absent; per-column legacy sheet materialized (no Alert keys — \
         no Alert columns) — live re-proven"
    );

    // Cleanup: both sides of the transfer.
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "config",
            "delete",
            "[default]P11CsvSeed",
            "[p11csv]TMem",
            "[p11csv]TExpr",
            "[p11csv]TAlarmed",
            "--yes",
            "--compact",
        ],
    );
    expect_ok("config delete cleanup (both sides)", &out);
    let out = ign(
        &config,
        &env,
        &["tags", "provider", "delete", "p11csv", "--yes", "--compact"],
    );
    expect_ok("provider delete p11csv (guarded)", &out);
}

/// THE loss-gate refusal live gate (11-06, TAGS-12 / roadmap SC-3) —
/// READ-ONLY by construction: a loss-bearing XML fixture (a
/// gateway-shaped document carrying `MinVersion` + a UdtType
/// definition + a CompoundProperty alarms block) meets
/// `ign tags import` WITHOUT `--yes` and the gate refuses
/// PRE-RESOLUTION: exit 2, `invalid_input`, `profile` null (no
/// resolution ever happened), the report prose on stderr naming the
/// stable scan codes — and ZERO route calls (provable because the
/// refusal fires before any wire work: there is no route-layer
/// envelope or error shape in the answer, and no gateway contact is
/// even required). No seeding, no mutations env.
#[tokio::test]
#[ignore = "opt-in e2e: set IGNITION_LIVE_URL + IGNITION_LIVE_TOKEN (read-only — the refusal is pre-resolution, no mutations env needed)"]
async fn live_tags_loss_gate_refusal() {
    let Some(env) = ({
        match (live_url(), live_token()) {
            (Some(url), Some(token)) => Some(LiveEnv { url, token }),
            _ => None,
        }
    }) else {
        skip(
            "IGNITION_LIVE_URL/IGNITION_LIVE_TOKEN not set — nothing to point the read-only refusal at",
        );
        return;
    };
    let (_dir, config) = isolated_live_config(&env);
    let _gate = LIVE_GATE.lock().await;

    // The loss-bearing fixture: gateway-shaped (MinVersion root attr
    // → xml_export_edited_only), carries a UdtType definition
    // (→ xml_udt_type_definition — definitions DO land via the CLI's
    // provider-root import, routed to _types_; the transfer is not
    // lossless) and a CompoundProperty alarms block.
    let tmp = tempfile::tempdir().expect("tempdir");
    let fixture = tmp.path().join("lossy.xml");
    std::fs::write(
        &fixture,
        r#"<Tags MinVersion="8.0.0" locale="en_US">
   <Tag name="MotorType" type="UdtType">
      <Parameters>
         <Property name="MotorNumber" type="Integer">1</Property>
      </Parameters>
      <Tags>
         <Tag name="Amps" type="AtomicTag">
            <Property name="valueSource">opc</Property>
            <CompoundProperty name="alarms">
               <PropertySet>
                  <Property name="setpointA">25</Property>
                  <Property name="name">Low Amps</Property>
                  <Property name="priority">3</Property>
               </PropertySet>
            </CompoundProperty>
         </Tag>
      </Tags>
   </Tag>
</Tags>
"#,
    )
    .expect("write fixture");

    // NO --yes, default abort policy: the loss gate refuses before
    // resolution — the provider name is irrelevant (never resolved).
    let out = ign(
        &config,
        &env,
        &[
            "tags",
            "import",
            "--file",
            fixture.to_str().expect("path"),
            "--format",
            "xml",
            "--provider",
            "default",
            "--compact",
        ],
    );
    assert_eq!(
        out.status.code(),
        Some(2),
        "the loss gate refuses exit 2; stdout: {} stderr: {}",
        String::from_utf8_lossy(&out.stdout),
        String::from_utf8_lossy(&out.stderr)
    );
    let envelope = err_envelope(&out);
    assert_eq!(envelope["error"]["code"], "invalid_input", "{envelope}");
    assert!(
        envelope["profile"].is_null(),
        "PRE-RESOLUTION: no profile resolved, zero wire work: {envelope}"
    );
    let message = envelope["error"]["message"].as_str().unwrap_or_default();
    assert!(
        message.contains("loss report (xml)"),
        "the report prose leads the refusal: {envelope}"
    );
    assert!(
        message.contains("[xml_export_edited_only]"),
        "the MinVersion fact is named: {envelope}"
    );
    assert!(
        message.contains("[xml_udt_type_definition]"),
        "the UdtType-definition fact is named: {envelope}"
    );
    assert!(
        message.contains("re-run with --yes to import anyway"),
        "the --yes hint closes the prose: {envelope}"
    );
    assert!(
        !message.contains("route_error") && !message.contains("gateway_client_error"),
        "ZERO route calls: no route-layer error shape anywhere: {envelope}"
    );
    println!(
        "loss-gate refusal: exit 2 pre-resolution, report prose carried \
         xml_export_edited_only + xml_udt_type_definition, zero route calls"
    );
}