greentic-runner-host 1.1.2

//! End-to-end characterization of the `component == "mcp"` flow node
//! (LOCKED ENCODING v2: `server`/`tool`/`arguments`/`output` carried in the
//! node payload/config).
//!
//! This is the flow-execution MCP path (role `flow_editor`), separate from the
//! agent-loop MCP path (role `agentic_worker`). A single-node `.gtpack` flow is
//! built and run through the real [`FlowEngine`], with the per-tenant MCP
//! catalog backed by two wiremock servers: a fake admin
//! (`/api/v1/designer/tenant/me/mcp-servers`) returning one `flow_editor` MCP
//! server, and a fake MCP server speaking the JSON-RPC contract (initialize,
//! initialized, tools/list, tools/call).
//!
//! `FlowEngine::new` constructs the MCP source from the admin endpoint and
//! token env vars, so the success test points those at the admin wiremock. The
//! wiremock admin/MCP mount shapes mirror the in-module tests in the
//! `greentic-aw-runtime` mcp_source module (those helpers are test-only and
//! cannot be reused from here).
//!
//! Two paths are covered. The happy path calls the tool with templated
//! arguments and binds its structured result under the configured `output`
//! state key. The degraded path leaves the admin unconfigured (no env), so the
//! node FAILS GRACEFULLY with a structured error value while the flow still
//! COMPLETES (no panic, no aborted runtime).

#![cfg(feature = "agentic-worker")]
#![allow(clippy::unwrap_used, clippy::expect_used)]

use std::collections::BTreeMap;
use std::fs::File;
use std::io::Write;
use std::path::Path;
use std::str::FromStr;
use std::sync::{Arc, Mutex};

use anyhow::{Context, Result};
use greentic_runner_host::config::{
    FlowRetryConfig, HostConfig, OperatorPolicy, RateLimits, SecretsPolicy, StateStorePolicy,
    WebhookPolicy,
};
use greentic_runner_host::pack::{ComponentResolution, PackRuntime};
use greentic_runner_host::runner::engine::{FlowContext, FlowEngine, FlowStatus};
use greentic_runner_host::runner::flow_adapter::{NodeIR, flow_doc_to_ir};
use greentic_runner_host::trace::TraceConfig;
use greentic_runner_host::validate::ValidationConfig;
use greentic_types::{
    ExtensionInline, ExtensionRef, PackFlowEntry, PackKind, PackManifest, encode_pack_manifest,
};
use once_cell::sync::Lazy;
use semver::Version;
use serde_json::{Value, json};
use tempfile::TempDir;
use wiremock::matchers::{body_partial_json, header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
use zip::write::FileOptions;

const RUNTIME_FLOW_EXTENSION_ID: &str = "greentic.pack.runtime_flow";
const PACK_ID: &str = "mcp.flow.node.test";
const FLOW_ID: &str = "mcp.flow";
const SESSION_HINT: &str = "demo:provider:chan:conv:user";

static RUNTIME: Lazy<&'static tokio::runtime::Runtime> = Lazy::new(|| {
    Box::leak(Box::new(
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("test runtime"),
    ))
});

/// Serializes the env-dependent tests in this file: both read/write the shared
/// process-global `GREENTIC_AW_*` vars that `FlowEngine::new` consumes.
static ENV_GUARD: Mutex<()> = Mutex::new(());

// ── wiremock fakes (shapes mirror mcp_source.rs in-module tests) ──────────────

fn initialize_ok() -> ResponseTemplate {
    ResponseTemplate::new(200)
        .insert_header("Mcp-Session-Id", "sess-1")
        .set_body_json(json!({
            "jsonrpc": "2.0", "id": 1,
            "result": {
                "protocolVersion": "2025-06-18",
                "serverInfo": { "name": "fake", "version": "1.0.0" }
            }
        }))
}

/// Mount the 4-call MCP JSON-RPC contract on a fresh wiremock server, returning
/// `call_result_json` for `tools/call`.
async fn fake_mcp_server(call_result_json: Value) -> MockServer {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(body_partial_json(json!({ "method": "initialize" })))
        .respond_with(initialize_ok())
        .mount(&server)
        .await;
    Mock::given(method("POST"))
        .and(body_partial_json(
            json!({ "method": "notifications/initialized" }),
        ))
        .respond_with(ResponseTemplate::new(202))
        .mount(&server)
        .await;
    Mock::given(method("POST"))
        .and(body_partial_json(json!({ "method": "tools/list" })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "jsonrpc": "2.0", "id": 2,
            "result": { "tools": [
                { "name": "get_issue", "description": "Get an issue",
                  "inputSchema": { "type": "object",
                    "properties": { "id": { "type": "string" } } } }
            ] }
        })))
        .mount(&server)
        .await;
    Mock::given(method("POST"))
        .and(body_partial_json(json!({ "method": "tools/call" })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "jsonrpc": "2.0", "id": 3,
            "result": call_result_json
        })))
        .mount(&server)
        .await;
    server
}

/// Mount the admin `mcp-servers` endpoint returning one `flow_editor` server
/// pointing at `mcp_url`.
async fn fake_admin(mcp_url: &str) -> MockServer {
    let admin = MockServer::start().await;
    Mock::given(method("GET"))
        .and(path("/api/v1/designer/tenant/me/mcp-servers"))
        .and(header("authorization", "Bearer gtc_live_test"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "servers": [
                {
                    "id": "github", "name": "GitHub", "transport_url": mcp_url,
                    "auth_header_name": null, "auth_token": null,
                    "allowed_tools": null, "roles": ["flow_editor"]
                }
            ]
        })))
        .mount(&admin)
        .await;
    admin
}

// ── pack harness (mirrors runtime_call_nodes.rs build_dispatch_pack) ──────────

fn host_config(bindings_path: &Path) -> HostConfig {
    HostConfig {
        tenant: "demo".into(),
        bindings_path: bindings_path.to_path_buf(),
        flow_type_bindings: Default::default(),
        rate_limits: RateLimits::default(),
        retry: FlowRetryConfig::default(),
        http_enabled: false,
        secrets_policy: SecretsPolicy::allow_all(),
        state_store_policy: StateStorePolicy::default(),
        webhook_policy: WebhookPolicy::default(),
        timers: Vec::new(),
        oauth: None,
        mocks: None,
        pack_bindings: Vec::new(),
        env_passthrough: Vec::new(),
        trace: TraceConfig::from_env(),
        validation: ValidationConfig::from_env(),
        operator_policy: OperatorPolicy::allow_all(),
        agents: std::collections::HashMap::new(),
        graphs: std::collections::HashMap::new(),
    }
}

/// Build a `.gtpack` whose only flow is a single `component == "mcp"` node
/// (LOCKED ENCODING v2). `node_input` carries `{ server, tool, arguments,
/// output? }` directly in the node payload/config — `server`/`tool` are the
/// source of truth, NOT an `operation` string. No WASM component is needed:
/// the node is a native runtime dispatch.
///
/// `component = "mcp"` is a valid `greentic_types::ComponentId`, so the
/// runtime-flow pack round-trips it through `ComponentId::from_str` without the
/// earlier `:`/`/` rejection; the flow adapter sees exactly `"mcp"`.
fn build_mcp_pack(pack_path: &Path, node_input: Value) -> Result<()> {
    let mut nodes = serde_json::Map::new();
    nodes.insert(
        "call".to_string(),
        json!({
            "component": "mcp",
            "input": node_input,
            "routing": "end",
        }),
    );

    let runtime_flow = json!({
        "id": FLOW_ID,
        "flow_type": "messaging",
        "start": "call",
        "nodes": Value::Object(nodes),
    });
    let runtime_extension = json!({ "flows": [runtime_flow] });

    let mut extensions = BTreeMap::new();
    extensions.insert(
        RUNTIME_FLOW_EXTENSION_ID.to_string(),
        ExtensionRef {
            kind: RUNTIME_FLOW_EXTENSION_ID.to_string(),
            version: "2.0.0".into(),
            digest: None,
            location: None,
            inline: Some(ExtensionInline::Other(runtime_extension)),
        },
    );

    let manifest = PackManifest {
        schema_version: "1.0".into(),
        pack_id: PACK_ID.parse()?,
        name: None,
        version: Version::parse("0.0.0")?,
        kind: PackKind::Application,
        publisher: "test".into(),
        components: Vec::new(),
        flows: Vec::<PackFlowEntry>::new(),
        dependencies: Vec::new(),
        capabilities: Vec::new(),
        signatures: Default::default(),
        secret_requirements: Vec::new(),
        bootstrap: None,
        agents: BTreeMap::new(),
        extensions: Some(extensions),
    };

    let mut zip = zip::ZipWriter::new(File::create(pack_path).context("create pack archive")?);
    let options: FileOptions<'_, ()> =
        FileOptions::default().compression_method(zip::CompressionMethod::Stored);
    let manifest_bytes = encode_pack_manifest(&manifest)?;
    zip.start_file("manifest.cbor", options)?;
    zip.write_all(&manifest_bytes)?;
    zip.finish().context("finalise pack archive")?;
    Ok(())
}

fn build_engine(
    pack_path: &Path,
    config: Arc<HostConfig>,
) -> Result<(Arc<PackRuntime>, FlowEngine)> {
    let rt = *RUNTIME;
    let pack = Arc::new(rt.block_on(PackRuntime::load(
        pack_path,
        Arc::clone(&config),
        None,
        None,
        None,
        None,
        Arc::new(greentic_runner_host::wasi::RunnerWasiPolicy::new()),
        greentic_runner_host::secrets::default_manager()?,
        None,
        false,
        ComponentResolution::default(),
    ))?);
    let engine = rt.block_on(FlowEngine::new(
        vec![Arc::clone(&pack)],
        Arc::clone(&config),
    ))?;
    Ok((pack, engine))
}

fn flow_ctx<'a>(config: &'a HostConfig, pack_id: &'a str) -> FlowContext<'a> {
    FlowContext {
        tenant: config.tenant.as_str(),
        pack_id,
        flow_id: FLOW_ID,
        node_id: None,
        tool: None,
        action: Some("messaging"),
        session_id: Some(SESSION_HINT),
        provider_id: Some("provider"),
        reply_scope: None,
        retry_config: config.retry.clone().into(),
        attempt: 1,
        observer: None,
        mocks: None,
    }
}

/// Recursively find a value at `key` anywhere in the (possibly envelope-wrapped)
/// output tree.
fn find_key<'a>(value: &'a Value, key: &str) -> Option<&'a Value> {
    match value {
        Value::Object(map) => {
            if let Some(found) = map.get(key) {
                return Some(found);
            }
            map.values().find_map(|v| find_key(v, key))
        }
        Value::Array(items) => items.iter().find_map(|v| find_key(v, key)),
        _ => None,
    }
}

// ── YGTC op-key load path (flow_doc_to_ir) ───────────────────────────────────
//
// The designer's packc injection writes the MCP node as the YGTC op-key shape
//
//   nodes:
//     lookup:
//       mcp:
//         server: <id>
//         tool: <name>
//         arguments: { ... }   # values may be {{ }} templates
//         output: <state_key>
//       routing: [ ... ]
//
// (see greentic-designer `inject_mcp_nodes`). The runner deserialises that into
// `greentic_flow::model::FlowDoc` and lowers it with the REAL flow adapter
// `flow_doc_to_ir`. This is the load primitive the runner uses; the assertions
// below start from the op-key text (NOT a hand-built runtime node with
// `component` pre-set) and prove the op-key passes through verbatim as
// `component == "mcp"` with the payload intact.

/// Build the YGTC op-key node text for `lookup` and lower it through the real
/// runner adapter `flow_doc_to_ir`, returning the derived `(component, payload,
/// routes)` for that node.
///
/// `flow_doc_to_ir` performs STRUCTURAL op-key extraction: it reads the single
/// flattened `raw` key off `NodeDoc` (here `mcp`) and carries it through as the
/// component, independent of `greentic-flow`'s own `classify_node_type` /
/// `NodeKind` lowering. This is the evidence for the release-ordering answer:
/// the runner does not depend on greentic-flow's `NodeKind::Mcp` to recognise
/// the node at runtime.
fn lower_ygtc_mcp_node(node_payload: Value, routing: Value) -> Result<NodeIR> {
    let doc_json = json!({
        "id": FLOW_ID,
        "type": "messaging",
        "start": "lookup",
        "nodes": {
            "lookup": {
                "mcp": node_payload,
                "routing": routing,
            }
        }
    });
    let doc: greentic_flow::model::FlowDoc =
        serde_json::from_value(doc_json).context("deserialise YGTC FlowDoc")?;
    let ir = flow_doc_to_ir(doc).context("flow_doc_to_ir lowering")?;
    let node = ir
        .nodes
        .get("lookup")
        .cloned()
        .context("`lookup` node missing from lowered FlowIR")?;
    Ok(node)
}

/// Build a single-flow `.gtpack` whose `lookup` node is derived END-TO-END from
/// the YGTC op-key shape via [`lower_ygtc_mcp_node`]. The runtime-flow node's
/// `component`/`input`/`routing` are taken from the adapter output (NOT
/// hand-set), so a regression that drops the `mcp` op-key (or wraps it in
/// `component.exec`) would surface here as a non-`mcp` component.
fn build_ygtc_mcp_pack(pack_path: &Path, node_ir: &NodeIR) -> Result<()> {
    // Translate the adapter's `routes` back into the runtime-flow `routing`
    // shape the loader understands: a single terminal route -> `"end"`.
    let routing = if node_ir.routes.iter().all(|r| r.out || r.to.is_none()) {
        json!("end")
    } else {
        serde_json::to_value(&node_ir.routes)?
    };

    let mut nodes = serde_json::Map::new();
    nodes.insert(
        "lookup".to_string(),
        json!({
            "component": node_ir.component,
            "input": node_ir.payload_expr,
            "routing": routing,
        }),
    );

    let runtime_flow = json!({
        "id": FLOW_ID,
        "flow_type": "messaging",
        "start": "lookup",
        "nodes": Value::Object(nodes),
    });
    let runtime_extension = json!({ "flows": [runtime_flow] });

    let mut extensions = BTreeMap::new();
    extensions.insert(
        RUNTIME_FLOW_EXTENSION_ID.to_string(),
        ExtensionRef {
            kind: RUNTIME_FLOW_EXTENSION_ID.to_string(),
            version: "2.0.0".into(),
            digest: None,
            location: None,
            inline: Some(ExtensionInline::Other(runtime_extension)),
        },
    );

    let manifest = PackManifest {
        schema_version: "1.0".into(),
        pack_id: PACK_ID.parse()?,
        name: None,
        version: Version::parse("0.0.0")?,
        kind: PackKind::Application,
        publisher: "test".into(),
        components: Vec::new(),
        flows: Vec::<PackFlowEntry>::new(),
        dependencies: Vec::new(),
        capabilities: Vec::new(),
        signatures: Default::default(),
        secret_requirements: Vec::new(),
        bootstrap: None,
        agents: BTreeMap::new(),
        extensions: Some(extensions),
    };

    let mut zip = zip::ZipWriter::new(File::create(pack_path).context("create pack archive")?);
    let options: FileOptions<'_, ()> =
        FileOptions::default().compression_method(zip::CompressionMethod::Stored);
    let manifest_bytes = encode_pack_manifest(&manifest)?;
    zip.start_file("manifest.cbor", options)?;
    zip.write_all(&manifest_bytes)?;
    zip.finish().context("finalise pack archive")?;
    Ok(())
}

// ── tests ─────────────────────────────────────────────────────────────────────

/// Release-ordering proof: the runner's CURRENT pinned `greentic-flow`
/// (1.1.x-dev, WITHOUT `NodeKind::Mcp`) already lowers the YGTC op-key
/// `{ mcp: { ... }, routing }` into a FlowIR node with `component == "mcp"` and
/// the payload intact — purely via `flow_doc_to_ir`'s structural op-key
/// extraction. No `greentic-flow` PR #235 is required at runtime.
#[test]
fn flow_doc_to_ir_preserves_mcp_op_key_verbatim() -> Result<()> {
    let node = lower_ygtc_mcp_node(
        json!({
            "server": "github",
            "tool": "get_issue",
            "arguments": { "id": "{{ entry.issue_id }}" },
            "output": "issue"
        }),
        json!([{ "out": true }]),
    )?;

    // The op-key survives verbatim as the component...
    assert_eq!(
        node.component, "mcp",
        "expected the `mcp` op-key to pass through `flow_doc_to_ir` verbatim, got `{}`",
        node.component
    );
    // ...and `mcp` is a valid ComponentId, so `flow_ir_to_flow` /
    // `ComponentId::from_str` accepts it at pack load.
    assert!(
        greentic_types::ComponentId::from_str(&node.component).is_ok(),
        "derived component `{}` must be a valid ComponentId",
        node.component
    );
    // ...and the payload (server/tool/arguments/output) is carried intact.
    assert_eq!(node.payload_expr["server"], json!("github"));
    assert_eq!(node.payload_expr["tool"], json!("get_issue"));
    assert_eq!(
        node.payload_expr["arguments"]["id"],
        json!("{{ entry.issue_id }}")
    );
    assert_eq!(node.payload_expr["output"], json!("issue"));
    Ok(())
}

/// End-to-end from the YGTC op-key shape through `flow_doc_to_ir` -> runtime
/// flow -> `FlowEngine`: the lowered MCP node calls the tool with templated
/// arguments rendered from flow state and binds the result under `output`.
#[test]
fn mcp_node_from_ygtc_op_key_calls_tool_and_binds_output() -> Result<()> {
    let _guard = ENV_GUARD.lock().unwrap();
    let rt = *RUNTIME;
    let temp = TempDir::new()?;
    let pack_path = temp.path().join("mcp-ygtc-ok.gtpack");
    let bindings_path = temp.path().join("bindings.yaml");
    std::fs::write(&bindings_path, b"tenant: demo")?;

    let mcp = rt.block_on(fake_mcp_server(
        json!({ "structuredContent": { "title": "Bug" } }),
    ));
    let admin = rt.block_on(fake_admin(&mcp.uri()));

    // Start from the YGTC op-key text and lower it with the REAL adapter.
    let node = lower_ygtc_mcp_node(
        json!({
            "server": "github",
            "tool": "get_issue",
            "arguments": { "id": "{{ entry.issue_id }}" },
            "output": "issue"
        }),
        json!([{ "out": true }]),
    )?;
    assert_eq!(
        node.component, "mcp",
        "YGTC op-key must lower to component `mcp` for the engine to dispatch it"
    );
    build_ygtc_mcp_pack(&pack_path, &node)?;

    let config = Arc::new(host_config(&bindings_path));

    // SAFETY: serialized by ENV_GUARD; cleared after the run below.
    unsafe {
        std::env::set_var("GREENTIC_AW_ADMIN_ENDPOINT", admin.uri());
        std::env::set_var("GREENTIC_AW_ADMIN_TOKEN", "gtc_live_test");
        std::env::remove_var("GREENTIC_AW_MCP");
    }

    let (pack, engine) = build_engine(&pack_path, Arc::clone(&config))?;

    unsafe {
        std::env::remove_var("GREENTIC_AW_ADMIN_ENDPOINT");
        std::env::remove_var("GREENTIC_AW_ADMIN_TOKEN");
    }

    let ctx = flow_ctx(&config, pack.metadata().pack_id.as_str());
    let execution = rt
        .block_on(engine.execute(ctx, json!({ "issue_id": "42" })))
        .context("mcp ygtc flow run")?;

    match execution.status {
        FlowStatus::Completed => {}
        FlowStatus::Waiting(wait) => anyhow::bail!("flow paused unexpectedly: {:?}", wait.reason),
    }

    let issue = find_key(&execution.output, "issue")
        .with_context(|| format!("output key `issue` missing, got {:?}", execution.output))?;
    assert_eq!(
        issue,
        &json!({ "title": "Bug" }),
        "bound output mismatch, got {:?}",
        execution.output
    );
    assert!(
        find_key(&execution.output, "error").is_none(),
        "unexpected error in output: {:?}",
        execution.output
    );
    Ok(())
}

/// Graceful-degrade twin of the YGTC-op-key path: with the admin unconfigured
/// the lowered MCP node fails with a structured error while the flow still
/// COMPLETES (no panic, no aborted runtime).
#[test]
fn mcp_node_from_ygtc_op_key_degrades_gracefully_when_unconfigured() -> Result<()> {
    let _guard = ENV_GUARD.lock().unwrap();
    let rt = *RUNTIME;
    let temp = TempDir::new()?;
    let pack_path = temp.path().join("mcp-ygtc-degraded.gtpack");
    let bindings_path = temp.path().join("bindings.yaml");
    std::fs::write(&bindings_path, b"tenant: demo")?;

    let node = lower_ygtc_mcp_node(
        json!({
            "server": "github",
            "tool": "get_issue",
            "arguments": { "id": "1" },
            "output": "issue"
        }),
        json!([{ "out": true }]),
    )?;
    assert_eq!(node.component, "mcp");
    build_ygtc_mcp_pack(&pack_path, &node)?;

    let config = Arc::new(host_config(&bindings_path));

    // No admin env configured -> MCP source is None -> graceful node error.
    unsafe {
        std::env::remove_var("GREENTIC_AW_ADMIN_ENDPOINT");
        std::env::remove_var("GREENTIC_AW_ADMIN_TOKEN");
        std::env::remove_var("GREENTIC_AW_MCP");
    }

    let (pack, engine) = build_engine(&pack_path, Arc::clone(&config))?;

    let ctx = flow_ctx(&config, pack.metadata().pack_id.as_str());
    let execution = rt
        .block_on(engine.execute(ctx, json!({})))
        .context("mcp ygtc degraded flow run")?;

    match execution.status {
        FlowStatus::Completed => {}
        FlowStatus::Waiting(wait) => anyhow::bail!("flow paused unexpectedly: {:?}", wait.reason),
    }

    let bound = find_key(&execution.output, "issue")
        .with_context(|| format!("output key `issue` missing, got {:?}", execution.output))?;
    let err = bound
        .get("error")
        .and_then(Value::as_str)
        .with_context(|| format!("expected an `error` string under `issue`, got {bound:?}"))?;
    assert!(
        err.contains("MCP is not configured"),
        "unexpected error message: {err}"
    );
    Ok(())
}

#[test]
fn mcp_node_calls_tool_and_binds_output() -> Result<()> {
    let _guard = ENV_GUARD.lock().unwrap();
    let rt = *RUNTIME;
    let temp = TempDir::new()?;
    let pack_path = temp.path().join("mcp-ok.gtpack");
    let bindings_path = temp.path().join("bindings.yaml");
    std::fs::write(&bindings_path, b"tenant: demo")?;

    // Stand up the wiremock pair: MCP server returns a structured payload.
    let mcp = rt.block_on(fake_mcp_server(
        json!({ "structuredContent": { "title": "Bug" } }),
    ));
    let admin = rt.block_on(fake_admin(&mcp.uri()));

    // LOCKED ENCODING v2: the component ref the flow adapter parses is exactly
    // `"mcp"` — a valid `ComponentId` (no `:`/`/` to be rejected). This is what
    // `runtime_flow_to_flow` runs through `ComponentId::from_str` at pack load.
    assert!(
        greentic_types::ComponentId::from_str("mcp").is_ok(),
        "component ref `mcp` must be a valid ComponentId"
    );

    // `arguments` uses a `{{ }}` template against entry state; `output` names
    // the state key the result is bound under. `server`/`tool` live in the
    // payload (the source of truth), NOT in an `operation` string.
    build_mcp_pack(
        &pack_path,
        json!({
            "server": "github",
            "tool": "get_issue",
            "arguments": { "id": "{{ entry.issue_id }}" },
            "output": "issue"
        }),
    )?;

    let config = Arc::new(host_config(&bindings_path));

    // SAFETY: serialized by ENV_GUARD; cleared after the run below.
    unsafe {
        std::env::set_var("GREENTIC_AW_ADMIN_ENDPOINT", admin.uri());
        std::env::set_var("GREENTIC_AW_ADMIN_TOKEN", "gtc_live_test");
        std::env::remove_var("GREENTIC_AW_MCP");
    }

    let (pack, engine) = build_engine(&pack_path, Arc::clone(&config))?;

    unsafe {
        std::env::remove_var("GREENTIC_AW_ADMIN_ENDPOINT");
        std::env::remove_var("GREENTIC_AW_ADMIN_TOKEN");
    }

    let ctx = flow_ctx(&config, pack.metadata().pack_id.as_str());
    let execution = rt
        .block_on(engine.execute(ctx, json!({ "issue_id": "42" })))
        .context("mcp flow run")?;

    match execution.status {
        FlowStatus::Completed => {}
        FlowStatus::Waiting(wait) => anyhow::bail!("flow paused unexpectedly: {:?}", wait.reason),
    }

    // The structured tool result is bound under the `issue` output key.
    let issue = find_key(&execution.output, "issue")
        .with_context(|| format!("output key `issue` missing, got {:?}", execution.output))?;
    assert_eq!(
        issue,
        &json!({ "title": "Bug" }),
        "bound output mismatch, got {:?}",
        execution.output
    );
    // And no error surfaced.
    assert!(
        find_key(&execution.output, "error").is_none(),
        "unexpected error in output: {:?}",
        execution.output
    );
    Ok(())
}

#[test]
fn mcp_node_degrades_gracefully_when_unconfigured() -> Result<()> {
    let _guard = ENV_GUARD.lock().unwrap();
    let rt = *RUNTIME;
    let temp = TempDir::new()?;
    let pack_path = temp.path().join("mcp-degraded.gtpack");
    let bindings_path = temp.path().join("bindings.yaml");
    std::fs::write(&bindings_path, b"tenant: demo")?;

    build_mcp_pack(
        &pack_path,
        json!({
            "server": "github",
            "tool": "get_issue",
            "arguments": { "id": "1" },
            "output": "issue"
        }),
    )?;

    let config = Arc::new(host_config(&bindings_path));

    // No admin env configured -> MCP source is None -> graceful node error.
    unsafe {
        std::env::remove_var("GREENTIC_AW_ADMIN_ENDPOINT");
        std::env::remove_var("GREENTIC_AW_ADMIN_TOKEN");
        std::env::remove_var("GREENTIC_AW_MCP");
    }

    let (pack, engine) = build_engine(&pack_path, Arc::clone(&config))?;

    let ctx = flow_ctx(&config, pack.metadata().pack_id.as_str());
    let execution = rt
        .block_on(engine.execute(ctx, json!({})))
        .context("mcp degraded flow run")?;

    // The runtime must NOT abort: the flow completes...
    match execution.status {
        FlowStatus::Completed => {}
        FlowStatus::Waiting(wait) => anyhow::bail!("flow paused unexpectedly: {:?}", wait.reason),
    }

    // ...and the node binds a structured error under the `issue` output key.
    let bound = find_key(&execution.output, "issue")
        .with_context(|| format!("output key `issue` missing, got {:?}", execution.output))?;
    let err = bound
        .get("error")
        .and_then(Value::as_str)
        .with_context(|| format!("expected an `error` string under `issue`, got {bound:?}"))?;
    assert!(
        err.contains("MCP is not configured"),
        "unexpected error message: {err}"
    );
    Ok(())
}

/// Flow-side local-wasm transport proof: an admin row with
/// `"transport": "local-wasm"`, `"component_ref": "router_echo"`, and
/// `"roles": ["flow_editor"]` causes the flow MCP node to dispatch through
/// `mcp_local::local_call_tool` in-process (no network MCP handshake) and
/// bind the result under the configured `output` state key.
///
/// Self-skips when `router_echo.wasm` is not built, mirroring the agent-loop
/// test in `mcp_local_loop.rs`. To run live:
///   ```
///   GREENTIC_MCP_ROUTER_ECHO_WASM=<path> cargo test -p greentic-runner-host mcp_node_local_wasm
///   ```
#[allow(unsafe_code)]
#[test]
fn mcp_node_local_wasm_calls_echo_tool_in_process() -> Result<()> {
    let _guard = ENV_GUARD.lock().unwrap();

    // Resolve the fixture wasm (self-skip when absent).
    let fixture_path = std::env::var("GREENTIC_MCP_ROUTER_ECHO_WASM")
        .map(std::path::PathBuf::from)
        .unwrap_or_else(|_| {
            std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
                .join("../../../greentic-mcp/target/wasm32-wasip2/release/router_echo.wasm")
        });
    if !fixture_path.exists() {
        eprintln!(
            "SKIP: router_echo.wasm not found; set GREENTIC_MCP_ROUTER_ECHO_WASM to run live"
        );
        return Ok(());
    }

    let rt = *RUNTIME;
    let temp = TempDir::new()?;
    let pack_path = temp.path().join("mcp-local-wasm.gtpack");
    let bindings_path = temp.path().join("bindings.yaml");
    let cache_dir = temp.path().join("mcp-local-cache");
    std::fs::create_dir_all(&cache_dir)?;
    std::fs::write(&bindings_path, b"tenant: demo")?;

    // Copy the fixture into the temp cache dir.
    std::fs::copy(&fixture_path, cache_dir.join("router_echo.wasm"))?;

    // Stand up a fake admin returning a `local-wasm` flow_editor server.
    let admin_server = rt.block_on(MockServer::start());
    rt.block_on(
        Mock::given(method("GET"))
            .and(path("/api/v1/designer/tenant/me/mcp-servers"))
            .and(header("authorization", "Bearer gtc_live_test"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "servers": [
                    {
                        "id": "local_echo",
                        "name": "LocalEcho",
                        "transport_url": "",
                        "auth_header_name": null,
                        "auth_token": null,
                        "allowed_tools": null,
                        "roles": ["flow_editor"],
                        "transport": "local-wasm",
                        "component_ref": "router_echo",
                        "component_version": "1.0.0"
                    }
                ]
            })))
            .mount(&admin_server),
    );

    // Build a pack with an MCP node calling the `echo` tool.
    build_mcp_pack(
        &pack_path,
        json!({
            "server": "local_echo",
            "tool": "echo",
            "arguments": { "message": "hi" },
            "output": "echo_result"
        }),
    )?;

    let config = Arc::new(host_config(&bindings_path));

    // SAFETY: serialised by ENV_GUARD.
    //
    // `GREENTIC_AW_ADMIN_ENDPOINT`/`GREENTIC_AW_ADMIN_TOKEN` are read by
    // `FlowEngine::new` to construct the `McpToolSource` and can be cleared
    // after `build_engine`. `GREENTIC_MCP_LOCAL_CACHE_DIR` is read lazily
    // inside `mcp_local::cache_dir()` at probe / call time (during `execute`),
    // so it must remain set until after the engine runs.
    unsafe {
        std::env::set_var("GREENTIC_AW_ADMIN_ENDPOINT", admin_server.uri());
        std::env::set_var("GREENTIC_AW_ADMIN_TOKEN", "gtc_live_test");
        std::env::remove_var("GREENTIC_AW_MCP");
        std::env::set_var("GREENTIC_MCP_LOCAL_CACHE_DIR", &cache_dir);
    }

    let (pack, engine) = build_engine(&pack_path, Arc::clone(&config))?;

    // Clear only the admin endpoint/token; keep GREENTIC_MCP_LOCAL_CACHE_DIR
    // set until after execute (the cache_dir() call is lazy).
    unsafe {
        std::env::remove_var("GREENTIC_AW_ADMIN_ENDPOINT");
        std::env::remove_var("GREENTIC_AW_ADMIN_TOKEN");
    }

    let ctx = flow_ctx(&config, pack.metadata().pack_id.as_str());
    let execution = rt
        .block_on(engine.execute(ctx, json!({})))
        .context("mcp local-wasm flow run")?;

    unsafe {
        std::env::remove_var("GREENTIC_MCP_LOCAL_CACHE_DIR");
    }

    match execution.status {
        FlowStatus::Completed => {}
        FlowStatus::Waiting(wait) => anyhow::bail!("flow paused unexpectedly: {:?}", wait.reason),
    }

    // The echo result is bound under `echo_result` — no error key.
    let echo_result = find_key(&execution.output, "echo_result").with_context(|| {
        format!(
            "output key `echo_result` missing, got {:?}",
            execution.output
        )
    })?;
    assert!(
        !echo_result.to_string().contains("\"error\""),
        "echo result must not contain an error; got: {echo_result}"
    );
    Ok(())
}