car-server-core 0.16.0

//! Smoke tests for embedders of `car-server-core`.
//!
//! These tests are written from the perspective of a downstream
//! consumer (the future `tokhn-daemon`). They prove:
//!
//! 1. An embedder can construct `ServerState` directly and exchange a
//!    JSON-RPC frame end-to-end through `run_dispatch`.
//! 2. `memory.add_fact` succeeds through the new
//!    `engine.lock().await` path and the fact is queryable.
//! 3. Two `ServerState`s with distinct memgine engines do not share
//!    state.
//! 4. Constructing a `ServerState` does NOT spawn a dream loop —
//!    `car_scheduler::ensure_dream_task` is never called from inside
//!    the library.
//! 5. A panicking dispatcher handler aborts only the offending
//!    connection's task; the engine remains usable for sibling
//!    sessions (tokio::sync::Mutex does not poison).

use car_memgine::MemgineEngine;
use car_server_core::{run_dispatch, ServerState, ServerStateConfig};
use futures::{SinkExt, StreamExt};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::sync::Arc;
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio::sync::Mutex;
use tokio_tungstenite::{accept_async, connect_async, tungstenite::Message};

fn loopback_state(journal_dir: std::path::PathBuf) -> Arc<ServerState> {
    let engine = Arc::new(Mutex::new(MemgineEngine::new(None)));
    let cfg = ServerStateConfig::new(journal_dir).with_shared_memgine(engine);
    Arc::new(ServerState::with_config(cfg))
}

/// Spin up a one-off loopback listener that runs `run_dispatch` against
/// the first incoming connection. Returns the bound address so a client
/// can connect.
async fn spawn_one_shot_dispatcher(state: Arc<ServerState>) -> SocketAddr {
    let listener = TcpListener::bind(SocketAddr::V4(SocketAddrV4::new(
        Ipv4Addr::new(127, 0, 0, 1),
        0,
    )))
    .await
    .expect("bind loopback");
    let addr = listener.local_addr().expect("local_addr");

    tokio::spawn(async move {
        let (stream, peer) = listener.accept().await.expect("accept");
        let ws = accept_async(stream).await.expect("ws handshake");
        let (write, read) = ws.split();
        let _ = run_dispatch(read, Box::pin(write), peer.to_string(), state).await;
    });

    addr
}

/// Connect to a loopback dispatcher, send one JSON-RPC request, await
/// the response. Panics if the response cannot be parsed as JSON.
async fn rpc_one(addr: SocketAddr, request: serde_json::Value) -> serde_json::Value {
    let url = format!("ws://{}", addr);
    let (mut ws, _resp) = connect_async(&url).await.expect("ws client connect");
    let body = serde_json::to_string(&request).expect("request to_string");
    ws.send(Message::Text(body.into()))
        .await
        .expect("send request");
    let resp = ws
        .next()
        .await
        .expect("response frame")
        .expect("response frame ok");
    let text = match resp {
        Message::Text(t) => t.to_string(),
        other => panic!("expected Text frame, got {:?}", other),
    };
    serde_json::from_str(&text).expect("parse response JSON")
}

#[tokio::test]
async fn session_init_round_trip_through_run_dispatch() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_one_shot_dispatcher(state).await;

    let resp = rpc_one(
        addr,
        serde_json::json!({
            "jsonrpc": "2.0",
            "id": "init-1",
            "method": "session.init",
            "params": { "client_id": "embed-smoke", "tools": [], "policies": [] }
        }),
    )
    .await;

    assert_eq!(resp.get("id").and_then(|v| v.as_str()), Some("init-1"));
    let result = resp
        .get("result")
        .unwrap_or_else(|| panic!("session.init should succeed; full response: {}", resp));
    assert!(result.get("session_id").is_some());
    assert_eq!(
        result.get("tools_registered").and_then(|v| v.as_u64()),
        Some(0)
    );
    assert_eq!(
        result.get("policies_registered").and_then(|v| v.as_u64()),
        Some(0)
    );
}

#[tokio::test]
async fn a2ui_round_trips_through_run_dispatch() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_one_shot_dispatcher(state).await;

    let url = format!("ws://{}", addr);
    let (mut ws, _) = connect_async(&url).await.expect("ws client connect");

    async fn call(
        ws: &mut tokio_tungstenite::WebSocketStream<
            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
        >,
        id: &str,
        method: &str,
        params: serde_json::Value,
    ) -> serde_json::Value {
        ws.send(Message::Text(
            serde_json::json!({
                "jsonrpc": "2.0",
                "id": id,
                "method": method,
                "params": params
            })
            .to_string()
            .into(),
        ))
        .await
        .expect("send request");
        let text = ws
            .next()
            .await
            .expect("response frame")
            .expect("response frame ok")
            .into_text()
            .expect("text response")
            .to_string();
        let resp: serde_json::Value = serde_json::from_str(&text).expect("parse response");
        assert!(
            resp.get("error").is_none(),
            "{} should succeed; full response: {}",
            method,
            resp
        );
        resp["result"].clone()
    }

    let capabilities = call(&mut ws, "cap", "a2ui.capabilities", serde_json::json!({})).await;
    assert!(capabilities["components"]
        .as_array()
        .unwrap()
        .iter()
        .any(|component| component == "Button"));

    call(
        &mut ws,
        "create",
        "a2ui.apply",
        serde_json::json!({
            "version": "v0.9",
            "createSurface": { "surfaceId": "approval" }
        }),
    )
    .await;
    call(
        &mut ws,
        "components",
        "a2ui.apply",
        serde_json::json!({
            "version": "v0.9",
            "updateComponents": {
                "surfaceId": "approval",
                "components": [
                    { "id": "root", "component": "Button", "text": "Approve", "action": { "name": "approve" } }
                ]
            }
        }),
    )
    .await;

    let surfaces = call(&mut ws, "list", "a2ui.surfaces", serde_json::json!({})).await;
    assert_eq!(surfaces[0]["surfaceId"], "approval");
    assert_eq!(surfaces[0]["components"]["root"]["component"], "Button");

    let action = call(
        &mut ws,
        "action",
        "a2ui.action",
        serde_json::json!({
            "name": "approve",
            "surfaceId": "approval",
            "sourceComponentId": "root",
            "timestamp": "2026-05-08T00:00:00Z",
            "context": {}
        }),
    )
    .await;
    assert_eq!(action["route"]["delivered"], false);

    let reap = call(&mut ws, "reap", "a2ui.reap", serde_json::json!({})).await;
    assert_eq!(reap["removed"].as_array().unwrap().len(), 0);
}

#[tokio::test]
async fn event_log_can_be_truncated_and_cleared_through_run_dispatch() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_one_shot_dispatcher(state).await;

    let url = format!("ws://{}", addr);
    let (mut ws, _) = connect_async(&url).await.expect("ws client connect");

    async fn call(
        ws: &mut tokio_tungstenite::WebSocketStream<
            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
        >,
        id: &str,
        method: &str,
        params: serde_json::Value,
    ) -> serde_json::Value {
        ws.send(Message::Text(
            serde_json::json!({
                "jsonrpc": "2.0",
                "id": id,
                "method": method,
                "params": params
            })
            .to_string()
            .into(),
        ))
        .await
        .expect("send request");
        let text = ws
            .next()
            .await
            .expect("response frame")
            .expect("response frame ok")
            .into_text()
            .expect("text response")
            .to_string();
        let resp: serde_json::Value = serde_json::from_str(&text).expect("parse response");
        assert!(
            resp.get("error").is_none(),
            "{} should succeed; full response: {}",
            method,
            resp
        );
        resp["result"].clone()
    }

    call(
        &mut ws,
        "init",
        "session.init",
        serde_json::json!({ "client_id": "event-log-smoke", "tools": [], "policies": [] }),
    )
    .await;
    call(
        &mut ws,
        "proposal",
        "proposal.submit",
        serde_json::json!({
            "proposal": {
                "id": "event-log-proposal",
                "source": "test",
                "actions": [{
                    "id": "write-x",
                    "type": "state_write",
                    "parameters": { "key": "x", "value": 1 }
                }]
            }
        }),
    )
    .await;
    let stats = call(&mut ws, "stats", "events.stats", serde_json::json!({})).await;
    assert!(stats["events"].as_u64().unwrap() > 0);
    assert!(stats["spans"].as_u64().unwrap() > 0);

    let truncated = call(
        &mut ws,
        "truncate",
        "events.truncate",
        serde_json::json!({ "maxEvents": 1, "maxSpans": 1 }),
    )
    .await;
    assert!(truncated["removedEvents"].as_u64().unwrap() > 0);
    assert_eq!(truncated["stats"]["events"], 1);
    assert_eq!(truncated["stats"]["spans"], 1);

    let cleared = call(&mut ws, "clear", "events.clear", serde_json::json!({})).await;
    assert_eq!(cleared["stats"]["events"], 0);
    assert_eq!(cleared["stats"]["spans"], 0);
}

#[tokio::test]
async fn a2a_send_rejects_non_loopback_endpoint_without_opt_in() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_one_shot_dispatcher(state).await;

    let resp = rpc_one(
        addr,
        serde_json::json!({
            "jsonrpc": "2.0",
            "id": "a2a-send",
            "method": "a2a.send",
            "params": {
                "endpoint": "http://198.51.100.7:8080",
                "message": {
                    "messageId": "m1",
                    "role": "user",
                    "parts": [{ "kind": "text", "text": "hello" }]
                }
            }
        }),
    )
    .await;

    let error = resp
        .get("error")
        .and_then(|err| err.get("message"))
        .and_then(|msg| msg.as_str())
        .unwrap_or("");
    assert!(
        error.contains("allowUntrustedEndpoint"),
        "expected endpoint rejection, got {}",
        resp
    );
}

#[tokio::test]
async fn a2ui_update_without_fresh_route_auth_clears_stale_auth() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_one_shot_dispatcher(state.clone()).await;

    let url = format!("ws://{}", addr);
    let (mut ws, _) = connect_async(&url).await.expect("ws client connect");

    async fn call(
        ws: &mut tokio_tungstenite::WebSocketStream<
            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
        >,
        id: &str,
        method: &str,
        params: serde_json::Value,
    ) -> serde_json::Value {
        ws.send(Message::Text(
            serde_json::json!({
                "jsonrpc": "2.0",
                "id": id,
                "method": method,
                "params": params
            })
            .to_string()
            .into(),
        ))
        .await
        .expect("send request");
        let text = ws
            .next()
            .await
            .expect("response frame")
            .expect("response frame ok")
            .into_text()
            .expect("text response")
            .to_string();
        let resp: serde_json::Value = serde_json::from_str(&text).expect("parse response");
        assert!(
            resp.get("error").is_none(),
            "{} should succeed; full response: {}",
            method,
            resp
        );
        resp["result"].clone()
    }

    call(
        &mut ws,
        "ingest",
        "a2ui.ingest",
        serde_json::json!({
            "endpoint": "http://127.0.0.1:9109",
            "routeAuth": { "type": "bearer", "token": "secret-token" },
            "owner": { "kind": "a2a", "taskId": "task-1", "contextId": "ctx-1" },
            "payload": {
                "version": "v0.9",
                "createSurface": { "surfaceId": "auth-test" }
            }
        }),
    )
    .await;
    assert!(state.a2ui_route_auth.lock().await.contains_key("auth-test"));

    call(
        &mut ws,
        "update",
        "a2ui.apply",
        serde_json::json!({
            "version": "v0.9",
            "updateComponents": {
                "surfaceId": "auth-test",
                "components": [
                    { "id": "root", "component": "Text", "text": "updated" }
                ]
            }
        }),
    )
    .await;
    assert!(!state.a2ui_route_auth.lock().await.contains_key("auth-test"));
}

#[tokio::test]
async fn memory_add_fact_then_query_through_tokio_lock_path() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_one_shot_dispatcher(state).await;

    let url = format!("ws://{}", addr);
    let (mut ws, _) = connect_async(&url).await.expect("ws client connect");

    // 1. session.init so the runtime exists.
    ws.send(Message::Text(
        serde_json::to_string(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": "init",
            "method": "session.init",
            "params": { "client_id": "embed-smoke", "tools": [], "policies": [] }
        }))
        .unwrap()
        .into(),
    ))
    .await
    .unwrap();
    let _init = ws.next().await.unwrap().unwrap();

    // 2. memory.add_fact — proves engine.lock().await works.
    ws.send(Message::Text(
        serde_json::to_string(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": "add",
            "method": "memory.add_fact",
            "params": {
                "subject": "smoke-test",
                "body": "embedders can write facts",
                "kind": "pattern"
            }
        }))
        .unwrap()
        .into(),
    ))
    .await
    .unwrap();
    let add_resp = ws.next().await.unwrap().unwrap();
    let add_text = match add_resp {
        Message::Text(t) => t.to_string(),
        other => panic!("unexpected: {:?}", other),
    };
    let add_json: serde_json::Value = serde_json::from_str(&add_text).unwrap();
    assert!(
        add_json.get("result").is_some(),
        "memory.add_fact should succeed; got {}",
        add_text
    );

    // 3. memory.query — should see the fact.
    ws.send(Message::Text(
        serde_json::to_string(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": "q",
            "method": "memory.query",
            "params": { "query": "smoke-test", "k": 5 }
        }))
        .unwrap()
        .into(),
    ))
    .await
    .unwrap();
    let q_resp = ws.next().await.unwrap().unwrap();
    let q_text = match q_resp {
        Message::Text(t) => t.to_string(),
        other => panic!("unexpected: {:?}", other),
    };
    let q_json: serde_json::Value = serde_json::from_str(&q_text).unwrap();
    let result_arr = q_json
        .get("result")
        .and_then(|v| v.as_array())
        .expect("memory.query result array");
    assert!(
        !result_arr.is_empty(),
        "memory.query should retrieve the just-added fact; got {}",
        q_text
    );
}

#[tokio::test]
async fn distinct_states_do_not_share_memgine() {
    let tmp1 = TempDir::new().unwrap();
    let tmp2 = TempDir::new().unwrap();

    let engine_a = Arc::new(Mutex::new(MemgineEngine::new(None)));
    let engine_b = Arc::new(Mutex::new(MemgineEngine::new(None)));

    let state_a = Arc::new(ServerState::with_config(
        ServerStateConfig::new(tmp1.path().join("a")).with_shared_memgine(engine_a.clone()),
    ));
    let state_b = Arc::new(ServerState::with_config(
        ServerStateConfig::new(tmp2.path().join("b")).with_shared_memgine(engine_b.clone()),
    ));

    // Write a fact directly into engine_a (skipping the WS round-trip
    // for simplicity — the goal of this test is to prove the two
    // ServerStates carry independent engines).
    {
        let mut guard = engine_a.lock().await;
        guard.ingest_fact(
            "f-a",
            "marker",
            "in engine A only",
            "user",
            "peer",
            chrono::Utc::now(),
            "global",
            None,
            vec![],
            false,
        );
    }

    // engine_b must be empty.
    let count_b = engine_b.lock().await.valid_fact_count();
    assert_eq!(
        count_b, 0,
        "two distinct ServerStates must not share memgine state"
    );
    let count_a = engine_a.lock().await.valid_fact_count();
    assert_eq!(count_a, 1, "engine_a should hold the fact we wrote");

    // And the ServerStates themselves point at the configured engines.
    assert!(state_a.shared_memgine.is_some());
    assert!(state_b.shared_memgine.is_some());
    assert!(
        Arc::ptr_eq(state_a.shared_memgine.as_ref().unwrap(), &engine_a),
        "state_a should hold the engine_a handle"
    );
    assert!(
        Arc::ptr_eq(state_b.shared_memgine.as_ref().unwrap(), &engine_b),
        "state_b should hold the engine_b handle"
    );
}

#[tokio::test]
async fn library_does_not_spawn_a_dream_loop_at_boot() {
    // Contract: per the U1 boundary, constructing a ServerState (and
    // accepting one connection) must NOT call
    // `car_scheduler::spawn_dream_loop`. Note that car-server-core
    // DOES depend on car-scheduler — the JSON-RPC `scheduler.*`
    // methods (`scheduler.create`, `scheduler.run`, `scheduler.run_loop`)
    // are user-driven and need access to the scheduler types. The
    // distinction matters: a JSON-RPC dispatcher entry that builds a
    // scheduler.Task for a client is fine; an automatic background
    // dream-loop spawn at library load is not.
    //
    // Verifying "spawn_dream_loop is never called from the library"
    // is partially a code-review property and partially a runtime
    // property. We exercise the runtime path here: build state, run
    // dispatch, never observe the dream loop's side effects (no task
    // file or trajectory dir created in the temp tree).
    let tmp = TempDir::new().unwrap();
    let engine = Arc::new(Mutex::new(MemgineEngine::new(None)));
    let cfg =
        ServerStateConfig::new(tmp.path().join("journals")).with_shared_memgine(engine.clone());
    let state = Arc::new(ServerState::with_config(cfg));
    let addr = spawn_one_shot_dispatcher(state.clone()).await;

    // Drive a single round-trip so accept() and create_session() run.
    let _ = rpc_one(
        addr,
        serde_json::json!({
            "jsonrpc": "2.0",
            "id": "init",
            "method": "session.init",
            "params": { "client_id": "no-dream-loop", "tools": [], "policies": [] }
        }),
    )
    .await;

    // The dream loop persists state to disk under `<car_dir>/tasks/`
    // and `<car_dir>/trajectories/` when wired up in the binary. Since
    // we never set CAR_HOME and only handed in `tmp/journals` for the
    // event log, neither path should exist anywhere we control.
    assert!(
        !tmp.path().join("tasks").exists(),
        "library spawned a dream/scheduler task store; it must not"
    );
    assert!(
        !tmp.path().join("trajectories").exists(),
        "library wrote dream-loop trajectories; it must not"
    );
}

#[tokio::test]
async fn handler_panic_does_not_poison_engine_for_other_connections() {
    // tokio::sync::Mutex does not poison on panic. We assert the
    // observable effect: after a connection's task panics while
    // holding the engine lock (simulated below), a fresh acquisition
    // of the same engine still succeeds.
    let engine = Arc::new(Mutex::new(MemgineEngine::new(None)));
    let cloned = engine.clone();

    // Spawn a task that grabs the lock and then panics. With tokio
    // Mutex, the lock is released when the guard is dropped during
    // unwind — and the Mutex itself never enters a poisoned state.
    let join = tokio::spawn(async move {
        let _guard = cloned.lock().await;
        panic!("simulated handler panic while holding engine lock");
    });
    let result = join.await;
    assert!(
        result.is_err(),
        "panicking task should report JoinError, got {:?}",
        result
    );

    // The mutex must still be usable from another task.
    let _guard = engine.lock().await;
    // If we reach here without deadlock, the engine is not poisoned.
}

#[tokio::test]
async fn disconnect_removes_session_from_registry() {
    // MULTI-4 / WS-3 regression: after a client disconnects, its entry
    // in `state.sessions` MUST be removed. Pre-fix, every connection
    // we ever accepted leaked an `Arc<ClientSession>` indefinitely.
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_one_shot_dispatcher(state.clone()).await;

    // Connect, perform one round-trip so create_session has fired,
    // then drop the WS to trigger disconnect.
    let url = format!("ws://{}", addr);
    {
        let (mut ws, _) = connect_async(&url).await.expect("ws client connect");
        ws.send(Message::Text(
            serde_json::to_string(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": "init",
                "method": "session.init",
                "params": { "client_id": "ws3-regression", "tools": [], "policies": [] }
            }))
            .unwrap()
            .into(),
        ))
        .await
        .unwrap();
        let _init = ws.next().await.unwrap().unwrap();

        // Confirm session was inserted into the registry.
        let count_before = state.sessions.lock().await.len();
        assert_eq!(
            count_before, 1,
            "session.init should have inserted exactly one session"
        );

        // Send a close frame; ws drops at end of scope.
        let _ = ws.send(Message::Close(None)).await;
    }

    // Poll for up to 500ms: the dispatcher's disconnect cleanup runs
    // on the spawned task, so session removal is asynchronous from
    // this test thread's perspective.
    let mut removed = false;
    for _ in 0..50 {
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        if state.sessions.lock().await.is_empty() {
            removed = true;
            break;
        }
    }
    assert!(
        removed,
        "session must be removed from state.sessions within 500ms of disconnect"
    );
}

/// End-to-end agent loop: client creates a small-list surface, fires
/// a render_report describing it, and the embedded `UIImprovementAgent`
/// auto-applies a `patchComponents` envelope tagging the surface with
/// `x-renderer.domain = "small_list"`. Verifies the full handler.rs
/// integration: report dispatch → agent decision → apply_a2ui_envelope
/// → broadcasted surface mutation visible to the same WS client.
#[tokio::test]
async fn ui_agent_round_trips_through_run_dispatch() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_one_shot_dispatcher(state).await;

    let url = format!("ws://{}", addr);
    let (mut ws, _) = connect_async(&url).await.expect("ws client connect");

    async fn call(
        ws: &mut tokio_tungstenite::WebSocketStream<
            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
        >,
        id: &str,
        method: &str,
        params: serde_json::Value,
    ) -> serde_json::Value {
        ws.send(Message::Text(
            serde_json::json!({
                "jsonrpc": "2.0",
                "id": id,
                "method": method,
                "params": params
            })
            .to_string()
            .into(),
        ))
        .await
        .expect("send request");
        // Drain frames until we find the matching `id` — render_report
        // also fans out `a2ui.event` notifications which can interleave
        // ahead of the JSON-RPC response.
        loop {
            let text = ws
                .next()
                .await
                .expect("response frame")
                .expect("response frame ok")
                .into_text()
                .expect("text response")
                .to_string();
            let resp: serde_json::Value = serde_json::from_str(&text).expect("parse response");
            if resp.get("id").and_then(|v| v.as_str()) == Some(id) {
                assert!(
                    resp.get("error").is_none(),
                    "{} should succeed; full response: {}",
                    method,
                    resp
                );
                return resp["result"].clone();
            }
            // Otherwise it's a notification — keep draining.
        }
    }

    // Build a small-list surface: root component is a List with two
    // items, no x-renderer hint. This matches `TagSmallListDomain`'s
    // trigger (structure=list, cardinality_band=2-5, domain=None).
    call(
        &mut ws,
        "create",
        "a2ui.apply",
        serde_json::json!({
            "version": "v0.9",
            "createSurface": { "surfaceId": "inbox" }
        }),
    )
    .await;
    call(
        &mut ws,
        "components",
        "a2ui.apply",
        serde_json::json!({
            "version": "v0.9",
            "updateComponents": {
                "surfaceId": "inbox",
                "components": [
                    {
                        "id": "root",
                        "component": "List",
                        "props": { "items": { "path": "/items" } }
                    }
                ]
            }
        }),
    )
    .await;

    // Send a render report matching the agent's TagSmallListDomain trigger.
    call(
        &mut ws,
        "report",
        "a2ui.render_report",
        serde_json::json!({
            "surfaceId": "inbox",
            "signature": {
                "structure": "list",
                "cardinality_band": "2-5",
                "action_count_band": "none"
            },
            "density": "comfortable",
            "visibleFieldCount": 2,
            "actionCount": 0,
            "viewport": { "width": 400, "height": 600, "scale": 1.0 },
            "hiddenViaOverflow": [],
            "timestamp": "2026-05-11T12:00:00Z",
            "reportId": "rpt-1",
            "sequence": 1,
            "placeholderCount": 0,
            "signatureStableSince": 1
        }),
    )
    .await;

    // The agent runs synchronously inside the handler before it
    // returns, so by the time `report` resolves the patch has been
    // applied. Verify via a2ui.surfaces.
    let surfaces = call(&mut ws, "list", "a2ui.surfaces", serde_json::json!({})).await;
    let inbox = surfaces
        .as_array()
        .expect("surfaces is array")
        .iter()
        .find(|s| s["surfaceId"] == "inbox")
        .expect("inbox surface present");
    let domain_hint = &inbox["components"]["root"]["props"]["x-renderer"]["domain"];
    assert_eq!(
        domain_hint, "small_list",
        "agent should have tagged root with x-renderer.domain=small_list; \
         got: {}",
        inbox["components"]["root"]
    );
}

#[tokio::test]
async fn ui_agent_no_patch_when_domain_already_set() {
    // Same setup as the round-trip test but the surface is created
    // with the domain hint already present. The agent should run
    // and decide no change — surface stays exactly as-is.
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_one_shot_dispatcher(state).await;

    let url = format!("ws://{}", addr);
    let (mut ws, _) = connect_async(&url).await.expect("ws client connect");

    async fn call(
        ws: &mut tokio_tungstenite::WebSocketStream<
            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
        >,
        id: &str,
        method: &str,
        params: serde_json::Value,
    ) -> serde_json::Value {
        ws.send(Message::Text(
            serde_json::json!({
                "jsonrpc": "2.0",
                "id": id,
                "method": method,
                "params": params
            })
            .to_string()
            .into(),
        ))
        .await
        .expect("send");
        loop {
            let text = ws
                .next()
                .await
                .expect("frame")
                .expect("frame ok")
                .into_text()
                .expect("text")
                .to_string();
            let resp: serde_json::Value = serde_json::from_str(&text).expect("json");
            if resp.get("id").and_then(|v| v.as_str()) == Some(id) {
                assert!(resp.get("error").is_none(), "{}: {}", method, resp);
                return resp["result"].clone();
            }
        }
    }

    call(
        &mut ws,
        "create",
        "a2ui.apply",
        serde_json::json!({
            "version": "v0.9",
            "createSurface": { "surfaceId": "inbox" }
        }),
    )
    .await;
    call(
        &mut ws,
        "components",
        "a2ui.apply",
        serde_json::json!({
            "version": "v0.9",
            "updateComponents": {
                "surfaceId": "inbox",
                "components": [
                    {
                        "id": "root",
                        "component": "List",
                        "props": {
                            "items": { "path": "/items" },
                            "x-renderer": { "domain": "small_list" }
                        }
                    }
                ]
            }
        }),
    )
    .await;

    // Snapshot before the report.
    let before = call(&mut ws, "list1", "a2ui.surfaces", serde_json::json!({})).await;
    let before_root = before[0]["components"]["root"].clone();

    call(
        &mut ws,
        "report",
        "a2ui.render_report",
        serde_json::json!({
            "surfaceId": "inbox",
            "signature": {
                "structure": "list",
                "cardinality_band": "2-5",
                "action_count_band": "none",
                "domain": "small_list"
            },
            "density": "comfortable",
            "visibleFieldCount": 2,
            "actionCount": 0,
            "viewport": { "width": 400, "height": 600, "scale": 1.0 },
            "hiddenViaOverflow": [],
            "timestamp": "2026-05-11T12:00:00Z",
            "reportId": "rpt-1",
            "sequence": 1,
            "placeholderCount": 0,
            "signatureStableSince": 1
        }),
    )
    .await;

    let after = call(&mut ws, "list2", "a2ui.surfaces", serde_json::json!({})).await;
    let after_root = after[0]["components"]["root"].clone();
    assert_eq!(
        before_root, after_root,
        "agent should not have mutated a domain-tagged surface"
    );
}