mubit-sdk 0.6.0

#[path = "support/utils.rs"]
mod _utils;

use _utils::{boxed_error, cleanup_run, create_client, new_run_id, print_summary, require};
use mubit_sdk::TransportMode;
use serde_json::{json, Value};
use std::error::Error;
use std::time::Instant;

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn Error>> {
    let name = "control_batch_insert_semantic_direct";
    let started = Instant::now();
    let client = create_client().await?;
    let run_id = new_run_id("control_batch_insert_semantic_direct");
    client.set_run_id(Some(run_id.clone()));
    client.set_transport(TransportMode::Http);

    let mut passed = true;
    let mut detail = "validated direct semantic control batch insert".to_string();
    let mut metrics = json!({});

    let scenario = async {
        let response = client
            .batch_insert(json!({
                "run_id": run_id,
                "deduplicate": true,
                "items": [
                    {
                        "item_id": "release-checkpoint-1",
                        "text": "Release checkpoint: API gateway patch r2.14 reduced 95th percentile response time from 410ms to 260ms during canary in us-east.",
                        "metadata_json": "{\"source\":\"release-engineering\",\"region\":\"us-east\",\"team\":\"gateway\"}",
                        "source": "rust-sdk-batch",
                        "embedding": [],
                    },
                    {
                        "item_id": "risk-register-2",
                        "text": "Risk register update: vendor dependency for payment retries has elevated failure probability after certificate chain rotation.",
                        "metadata_json": "{\"source\":\"risk-office\",\"domain\":\"payments\",\"severity\":\"high\"}",
                        "source": "rust-sdk-batch",
                        "embedding": [],
                    },
                    {
                        "item_id": "warehouse-health-3",
                        "text": "Warehouse health: lane C7 temperature deviation stabilized after recalibration; quality hold can be released after two clean cycles.",
                        "metadata_json": "{\"source\":\"iot-telemetry\",\"facility\":\"atl-03\",\"lane\":\"C7\"}",
                        "source": "rust-sdk-batch",
                        "embedding": [],
                    },
                    {
                        "item_id": "warehouse-health-3-duplicate",
                        "text": "Warehouse health: lane C7 temperature deviation stabilized after recalibration; quality hold can be released after two clean cycles.",
                        "metadata_json": "{\"source\":\"iot-telemetry\",\"facility\":\"atl-03\",\"lane\":\"C7\"}",
                        "source": "rust-sdk-batch",
                        "embedding": [],
                    }
                ]
            }))
            .await?;

        let count = response.get("count").and_then(Value::as_u64).unwrap_or(0);
        let node_ids = response
            .get("node_ids")
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_default();
        let item_results = response
            .get("item_results")
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_default();

        require(
            count >= 3,
            format!("expected at least 3 inserted items: {response}"),
        )?;
        require(
            node_ids.len() == count as usize,
            format!("node_ids mismatch: {response}"),
        )?;
        require(
            item_results.len() == 4,
            format!("item_results mismatch: {response}"),
        )?;

        let successful_items = item_results
            .iter()
            .filter(|item| {
                item.get("success")
                    .and_then(Value::as_bool)
                    .unwrap_or(false)
            })
            .count();

        require(
            successful_items >= 3,
            format!("expected at least 3 successful item inserts: {response}"),
        )?;

        let query = client
            .query(json!({
                "run_id": run_id,
                "query": "summarize release, risk, and warehouse updates for this run",
                "schema": "",
                "mode": "agent_routed",
                "direct_lane": "semantic_search",
                "include_linked_runs": false,
                "limit": 6,
                "embedding": [],
            }))
            .await?;

        require(
            query.get("mode").and_then(Value::as_str) == Some("agent_routed"),
            format!("unexpected query mode: {query}"),
        )?;
        require(
            query
                .get("final_answer")
                .and_then(Value::as_str)
                .map(str::trim)
                .map(|value| !value.is_empty())
                .unwrap_or(false),
            format!("final_answer should be non-empty: {query}"),
        )?;
        let evidence_count = query
            .get("evidence")
            .and_then(Value::as_array)
            .map(|items| items.len())
            .unwrap_or(0);
        require(
            evidence_count > 0,
            format!("expected non-empty evidence: {query}"),
        )?;

        metrics = json!({
            "run_id": run_id,
            "count": count,
            "successful_items": successful_items,
            "node_ids": node_ids,
            "evidence_count": evidence_count,
        });

        Ok::<(), Box<dyn Error>>(())
    }
    .await;

    if let Err(err) = scenario {
        passed = false;
        detail = err.to_string();
    }

    let cleanup_ok = cleanup_run(&client, &run_id).await;
    if !cleanup_ok {
        passed = false;
        detail = format!("{detail} | cleanup failures");
    }

    print_summary(
        name,
        passed,
        &detail,
        &metrics,
        started.elapsed().as_secs_f64(),
        cleanup_ok,
    );

    if passed {
        Ok(())
    } else {
        Err(boxed_error(detail))
    }
}