car-server-core 0.35.0

//! `memory.utility_set` read-modify-write semantics, end-to-end through
//! `run_dispatch` (Linus round-1 #5 / round-2 blocker).
//!
//! The contract: an omitted field keeps the engine's current value — a partial
//! set must NOT silently zero the field the caller didn't pass. Because the
//! daemon shares one engine across connections (`with_shared_memgine`), three
//! one-shot RPCs against the same state exercise the knob's persistence.

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 shared_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))
}

async fn spawn_one_shot(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
}

async fn rpc(state: &Arc<ServerState>, request: serde_json::Value) -> serde_json::Value {
    let addr = spawn_one_shot(state.clone()).await;
    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");
    let resp = ws.next().await.expect("frame").expect("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")
}

fn utility_set(weight: f64, exploration: Option<f64>) -> serde_json::Value {
    let mut params = serde_json::Map::new();
    params.insert("utility_weight".into(), serde_json::json!(weight));
    if let Some(e) = exploration {
        params.insert("utility_exploration".into(), serde_json::json!(e));
    }
    serde_json::json!({
        "jsonrpc": "2.0",
        "id": "u",
        "method": "memory.utility_set",
        "params": params,
    })
}

#[tokio::test]
async fn utility_set_partial_preserves_omitted_field() {
    let tmp = TempDir::new().unwrap();
    let state = shared_state(tmp.path().join("journals"));

    // Set both.
    let r = rpc(&state, utility_set(0.4, Some(1.2))).await;
    let res = r.get("result").expect("set both ok");
    assert_eq!(
        res.get("utility_weight").and_then(|v| v.as_f64()),
        Some(0.4)
    );
    assert_eq!(
        res.get("utility_exploration").and_then(|v| v.as_f64()),
        Some(1.2)
    );

    // Set ONLY weight (exploration omitted) — must keep 1.2, not reset to 0.
    let r = rpc(&state, utility_set(0.5, None)).await;
    let res = r.get("result").expect("set weight ok");
    assert_eq!(
        res.get("utility_weight").and_then(|v| v.as_f64()),
        Some(0.5)
    );
    assert_eq!(
        res.get("utility_exploration").and_then(|v| v.as_f64()),
        Some(1.2),
        "omitted exploration must be preserved (read-modify-write)"
    );

    // Confirm via the getter on the shared engine.
    let r = rpc(
        &state,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "g", "method": "memory.utility_get", "params": null
        }),
    )
    .await;
    let res = r.get("result").expect("get ok");
    assert_eq!(
        res.get("utility_weight").and_then(|v| v.as_f64()),
        Some(0.5)
    );
    assert_eq!(
        res.get("utility_exploration").and_then(|v| v.as_f64()),
        Some(1.2)
    );
}