//! HTTP JSON-RPC integration tests against a real axum stack and a mock upstream API. //! //! Isolates config under a temp `HOME` so auth profiles and the OpenHuman provider resolve //! the same state directory. Run with: `cargo test --test json_rpc_e2e` use std::net::SocketAddr; use std::path::Path; use std::sync::{Mutex, OnceLock}; use std::time::Duration; use axum::http::{header::AUTHORIZATION, HeaderMap, StatusCode}; use axum::routing::{get, post}; use axum::{Json, Router}; use futures_util::StreamExt; use serde_json::{json, Value}; use tempfile::tempdir; use openhuman_core::core::jsonrpc::build_core_http_router; use openhuman_core::openhuman::memory::all_memory_tree_registered_controllers; struct EnvVarGuard { key: &'static str, old: Option, } impl EnvVarGuard { fn set_to_path(key: &'static str, path: &Path) -> Self { let old = std::env::var(key).ok(); std::env::set_var(key, path.as_os_str()); Self { key, old } } fn set(key: &'static str, value: &str) -> Self { let old = std::env::var(key).ok(); std::env::set_var(key, value); Self { key, old } } fn unset(key: &'static str) -> Self { let old = std::env::var(key).ok(); std::env::remove_var(key); Self { key, old } } } impl Drop for EnvVarGuard { fn drop(&mut self) { match &self.old { Some(v) => std::env::set_var(self.key, v), None => std::env::remove_var(self.key), } } } /// Serializes tests in this binary: `HOME` / `OPENHUMAN_WORKSPACE` / backend URL overrides are /// process-global, so parallel tests would clobber each other and hit the wrong `config.toml` or /// inherited `VITE_BACKEND_URL`. static JSON_RPC_E2E_ENV_LOCK: OnceLock> = OnceLock::new(); static CHAT_COMPLETION_MODELS: OnceLock>> = OnceLock::new(); fn json_rpc_e2e_env_lock() -> std::sync::MutexGuard<'static, ()> { let mutex = JSON_RPC_E2E_ENV_LOCK.get_or_init(|| Mutex::new(())); // Recover from poison so that a panic in one test does not cascade to all others. match mutex.lock() { Ok(guard) => guard, Err(poisoned) => poisoned.into_inner(), } } fn with_chat_completion_models(f: impl FnOnce(&mut Vec) -> T) -> T { let mutex = CHAT_COMPLETION_MODELS.get_or_init(|| Mutex::new(Vec::new())); match mutex.lock() { Ok(mut guard) => f(&mut guard), Err(poisoned) => { let mut guard = poisoned.into_inner(); f(&mut guard) } } } fn mock_upstream_router() -> Router { const GENERAL_TOKEN: &str = "e2e-test-jwt"; const BILLING_TOKEN: &str = "e2e-billing-jwt"; const TEAM_TOKEN: &str = "e2e-team-jwt"; fn error_json(status: StatusCode, message: &str) -> (StatusCode, Json) { ( status, Json(json!({ "success": false, "error": message, "message": message, })), ) } fn require_bearer( headers: &HeaderMap, expected_token: &str, ) -> Result<(), (StatusCode, Json)> { require_any_bearer(headers, &[expected_token]) } fn require_any_bearer( headers: &HeaderMap, expected_tokens: &[&str], ) -> Result<(), (StatusCode, Json)> { let actual = headers .get(AUTHORIZATION) .and_then(|value| value.to_str().ok()) .map(str::trim); match actual { Some(value) if expected_tokens .iter() .any(|token| value == format!("Bearer {token}")) => { Ok(()) } Some(_) => Err(error_json( StatusCode::UNAUTHORIZED, "invalid Authorization bearer token", )), None => Err(error_json( StatusCode::UNAUTHORIZED, "missing Authorization bearer token", )), } } fn require_string_field<'a>( body: &'a Value, field: &str, ) -> Result<&'a str, (StatusCode, Json)> { body.get(field) .and_then(Value::as_str) .map(str::trim) .filter(|value| !value.is_empty()) .ok_or_else(|| { error_json( StatusCode::BAD_REQUEST, &format!("missing or invalid '{field}'"), ) }) } fn require_positive_f64_field( body: &Value, field: &str, ) -> Result)> { body.get(field) .and_then(Value::as_f64) .filter(|value| value.is_finite() && *value > 0.0) .ok_or_else(|| { error_json( StatusCode::BAD_REQUEST, &format!("missing or invalid '{field}'"), ) }) } // Matches authenticated profile fetches used during session validation. async fn current_user(headers: HeaderMap) -> Result, (StatusCode, Json)> { require_any_bearer(&headers, &[GENERAL_TOKEN, BILLING_TOKEN, TEAM_TOKEN])?; Ok(Json(json!({ "success": true, "data": { "_id": "e2e-user-1", "username": "e2e" } }))) } async fn chat_completions(Json(body): Json) -> Json { if let Some(model) = body.get("model").and_then(Value::as_str) { with_chat_completion_models(|models| models.push(model.to_string())); } Json(json!({ "choices": [{ "message": { "role": "assistant", "content": "Hello from e2e mock agent" } }] })) } // ── Billing mock routes ────────────────────────────────────────────────── async fn stripe_current_plan( headers: HeaderMap, ) -> Result, (StatusCode, Json)> { require_bearer(&headers, BILLING_TOKEN)?; Ok(Json(json!({ "success": true, "data": { "plan": "PRO", "hasActiveSubscription": true, "planExpiry": "2030-01-01T00:00:00.000Z", "subscription": { "id": "sub_mock_123", "status": "active" } } }))) } async fn stripe_purchase_plan( headers: HeaderMap, Json(body): Json, ) -> Result, (StatusCode, Json)> { require_bearer(&headers, BILLING_TOKEN)?; let plan = require_string_field(&body, "plan")?; if !matches!(plan, "basic" | "pro" | "BASIC" | "PRO") { return Err(error_json( StatusCode::BAD_REQUEST, "missing or invalid 'plan'", )); } let checkout_url = "http://127.0.0.1/mock-checkout"; let session_id = "cs_mock_abc"; if checkout_url.is_empty() || session_id.is_empty() { return Err(error_json( StatusCode::BAD_REQUEST, "missing checkoutUrl or sessionId", )); } Ok(Json(json!({ "success": true, "data": { "checkoutUrl": checkout_url, "sessionId": session_id } }))) } async fn stripe_portal(headers: HeaderMap) -> Result, (StatusCode, Json)> { require_bearer(&headers, BILLING_TOKEN)?; let portal_url = "http://127.0.0.1/mock-portal"; if portal_url.is_empty() { return Err(error_json(StatusCode::BAD_REQUEST, "missing portalUrl")); } Ok(Json(json!({ "success": true, "data": { "portalUrl": portal_url } }))) } async fn credits_top_up( headers: HeaderMap, Json(body): Json, ) -> Result, (StatusCode, Json)> { require_bearer(&headers, BILLING_TOKEN)?; let amount_usd = require_positive_f64_field(&body, "amountUsd")?; let gateway = require_string_field(&body, "gateway")?; if !matches!(gateway, "stripe" | "coinbase") { return Err(error_json( StatusCode::BAD_REQUEST, "missing or invalid 'gateway'", )); } Ok(Json(json!({ "success": true, "data": { "url": "http://127.0.0.1/mock-topup", "gatewayTransactionId": "txn_mock_1", "amountUsd": amount_usd, "gateway": gateway } }))) } async fn coinbase_charge( headers: HeaderMap, Json(body): Json, ) -> Result, (StatusCode, Json)> { require_bearer(&headers, BILLING_TOKEN)?; let plan = require_string_field(&body, "plan")?; let interval = body .get("interval") .and_then(Value::as_str) .map(str::trim) .filter(|value| !value.is_empty()) .unwrap_or("annual"); if !matches!(plan, "basic" | "pro" | "BASIC" | "PRO") { return Err(error_json( StatusCode::BAD_REQUEST, "missing or invalid 'plan'", )); } if interval != "annual" { return Err(error_json( StatusCode::BAD_REQUEST, "missing or invalid 'interval'", )); } Ok(Json(json!({ "success": true, "data": { "gatewayTransactionId": "coinbase_mock_1", "hostedUrl": "http://127.0.0.1/mock-coinbase", "status": "NEW", "expiresAt": "2030-01-01T01:00:00.000Z" } }))) } // ── Team mock routes ───────────────────────────────────────────────────── async fn team_members(headers: HeaderMap) -> Result, (StatusCode, Json)> { require_bearer(&headers, TEAM_TOKEN)?; Ok(Json(json!({ "success": true, "data": [ { "id": "user-1", "username": "alice", "role": "ADMIN" }, { "id": "user-2", "username": "bob", "role": "MEMBER" } ] }))) } async fn team_invites_get( headers: HeaderMap, ) -> Result, (StatusCode, Json)> { require_bearer(&headers, TEAM_TOKEN)?; Ok(Json(json!({ "success": true, "data": [ { "id": "inv-1", "code": "ALPHA1", "maxUses": 5, "usedCount": 1, "expiresAt": null } ] }))) } async fn team_invites_post( headers: HeaderMap, Json(body): Json, ) -> Result, (StatusCode, Json)> { require_bearer(&headers, TEAM_TOKEN)?; let max_uses = body .get("maxUses") .and_then(Value::as_u64) .ok_or_else(|| error_json(StatusCode::BAD_REQUEST, "missing or invalid 'maxUses'"))?; let expires_in_days = body .get("expiresInDays") .and_then(Value::as_u64) .ok_or_else(|| { error_json( StatusCode::BAD_REQUEST, "missing or invalid 'expiresInDays'", ) })?; if max_uses == 0 || expires_in_days == 0 { return Err(error_json( StatusCode::BAD_REQUEST, "invite payload values must be greater than zero", )); } Ok(Json(json!({ "success": true, "data": { "id": "inv-new", "code": "NEWCODE", "maxUses": max_uses, "usedCount": 0, "expiresAt": null } }))) } async fn team_member_delete( headers: HeaderMap, ) -> Result, (StatusCode, Json)> { require_bearer(&headers, TEAM_TOKEN)?; Ok(Json(json!({ "success": true, "data": {} }))) } async fn team_member_role_put( headers: HeaderMap, Json(body): Json, ) -> Result, (StatusCode, Json)> { require_bearer(&headers, TEAM_TOKEN)?; let role = require_string_field(&body, "role")?; if !matches!(role, "ADMIN" | "MEMBER" | "OWNER") { return Err(error_json( StatusCode::BAD_REQUEST, "missing or invalid 'role'", )); } Ok(Json(json!({ "success": true, "data": {} }))) } async fn team_invite_delete( headers: HeaderMap, ) -> Result, (StatusCode, Json)> { require_bearer(&headers, TEAM_TOKEN)?; Ok(Json(json!({ "success": true, "data": {} }))) } Router::new() .route("/settings", get(current_user)) .route("/auth/me", get(current_user)) .route("/openai/v1/chat/completions", post(chat_completions)) // billing .route("/payments/stripe/currentPlan", get(stripe_current_plan)) .route("/payments/stripe/purchasePlan", post(stripe_purchase_plan)) .route("/payments/stripe/portal", post(stripe_portal)) .route("/payments/credits/top-up", post(credits_top_up)) .route("/payments/coinbase/charge", post(coinbase_charge)) // team .route("/teams/{team_id}/members", get(team_members)) .route( "/teams/{team_id}/members/{user_id}", axum::routing::delete(team_member_delete), ) .route( "/teams/{team_id}/members/{user_id}/role", axum::routing::put(team_member_role_put), ) .route( "/teams/{team_id}/invites", get(team_invites_get).post(team_invites_post), ) .route( "/teams/{team_id}/invites/{invite_id}", axum::routing::delete(team_invite_delete), ) } async fn serve_on_ephemeral( app: Router, ) -> ( SocketAddr, tokio::task::JoinHandle>, ) { let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await .expect("bind"); let addr = listener.local_addr().expect("addr"); let handle = tokio::spawn(async move { axum::serve(listener, app).await }); (addr, handle) } async fn post_json_rpc(rpc_base: &str, id: i64, method: &str, params: Value) -> Value { let client = reqwest::Client::builder() .timeout(Duration::from_secs(120)) .build() .expect("client"); let body = json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params }); let url = format!("{}/rpc", rpc_base.trim_end_matches('/')); let resp = client .post(&url) .json(&body) .send() .await .unwrap_or_else(|e| panic!("POST {url}: {e}")); assert!( resp.status().is_success(), "HTTP error {} for {}", resp.status(), method ); resp.json::() .await .unwrap_or_else(|e| panic!("json for {method}: {e}")) } #[allow(dead_code)] async fn read_first_sse_event(events_url: &str) -> Value { let client = reqwest::Client::builder() .timeout(Duration::from_secs(120)) .build() .expect("client"); let resp = client .get(events_url) .send() .await .unwrap_or_else(|e| panic!("GET {events_url}: {e}")); assert!( resp.status().is_success(), "SSE HTTP error {} for {}", resp.status(), events_url ); let mut stream = resp.bytes_stream(); let mut buffer = String::new(); while let Some(item) = stream.next().await { let chunk = item.unwrap_or_else(|e| panic!("sse stream read failed: {e}")); let text = std::str::from_utf8(&chunk).unwrap_or(""); buffer.push_str(text); while let Some(idx) = buffer.find("\n\n") { let block = buffer[..idx].to_string(); buffer = buffer[idx + 2..].to_string(); let mut data_lines = Vec::new(); for line in block.lines() { if let Some(data) = line.strip_prefix("data:") { data_lines.push(data.trim_start()); } } if !data_lines.is_empty() { let payload = data_lines.join("\n"); let value: Value = serde_json::from_str(&payload) .unwrap_or_else(|e| panic!("invalid sse data json: {e}")); return value; } } } panic!("SSE stream ended before any event payload"); } /// Read SSE events until one matches the given `event` field value, skipping /// progress events (inference_start, iteration_start, etc.) that precede the /// terminal event. async fn read_sse_event_by_type(events_url: &str, target_event: &str) -> Value { let client = reqwest::Client::builder() .timeout(Duration::from_secs(120)) .build() .expect("client"); let resp = client .get(events_url) .send() .await .unwrap_or_else(|e| panic!("GET {events_url}: {e}")); assert!( resp.status().is_success(), "SSE HTTP error {} for {}", resp.status(), events_url ); let mut stream = resp.bytes_stream(); let mut buffer = String::new(); while let Some(item) = stream.next().await { let chunk = item.unwrap_or_else(|e| panic!("sse stream read failed: {e}")); let text = std::str::from_utf8(&chunk).unwrap_or(""); buffer.push_str(text); while let Some(idx) = buffer.find("\n\n") { let block = buffer[..idx].to_string(); buffer = buffer[idx + 2..].to_string(); let mut data_lines = Vec::new(); for line in block.lines() { if let Some(data) = line.strip_prefix("data:") { data_lines.push(data.trim_start()); } } if !data_lines.is_empty() { let payload = data_lines.join("\n"); let value: Value = serde_json::from_str(&payload) .unwrap_or_else(|e| panic!("invalid sse data json: {e}")); if value.get("event").and_then(Value::as_str) == Some(target_event) { return value; } } } } panic!("SSE stream ended before receiving '{target_event}' event"); } fn assert_no_jsonrpc_error<'a>(v: &'a Value, context: &str) -> &'a Value { if let Some(err) = v.get("error") { panic!("{context}: JSON-RPC error: {err}"); } v.get("result") .unwrap_or_else(|| panic!("{context}: missing result: {v}")) } fn extract_string_outcome(result: &Value) -> String { if let Some(s) = result.as_str() { return s.to_string(); } if let Some(inner) = result.get("result").and_then(Value::as_str) { return inner.to_string(); } panic!("expected string or {{result: string}}, got {result}"); } fn write_min_config(openhuman_dir: &Path, api_origin: &str) { // `chat_onboarding_completed = true` bypasses the welcome agent so that // `channel_web_chat` in tests routes straight to the orchestrator. Without // this, the first chat turn goes through the welcome flow whose tool // contract is not modelled by the e2e mock, which closes the SSE stream // mid-response. let cfg = format!( r#"api_url = "{api_origin}" default_model = "e2e-mock-model" default_temperature = 0.7 chat_onboarding_completed = true [secrets] encrypt = false "# ); fn write_config_file(config_dir: &Path, cfg: &str) { std::fs::create_dir_all(config_dir).expect("mkdir openhuman"); let path = config_dir.join("config.toml"); std::fs::write(&path, cfg).expect("write config"); } write_config_file(openhuman_dir, &cfg); // Runtime config resolution is user-scoped before login, so tests that seed // the root `~/.openhuman` directory also need the equivalent pre-login // config under `~/.openhuman/users/local`. if openhuman_dir .file_name() .is_some_and(|name| name == std::ffi::OsStr::new(".openhuman")) { write_config_file(&openhuman_dir.join("users").join("local"), &cfg); } let _: openhuman_core::openhuman::config::Config = toml::from_str(&cfg).expect("config toml must match Config schema"); } fn write_min_config_with_local_ai_disabled(openhuman_dir: &Path, api_origin: &str) { let cfg = format!( r#"api_url = "{api_origin}" default_model = "e2e-mock-model" default_temperature = 0.7 chat_onboarding_completed = true [secrets] encrypt = false [local_ai] enabled = false "# ); fn write_config_file(config_dir: &Path, cfg: &str) { std::fs::create_dir_all(config_dir).expect("mkdir openhuman"); let path = config_dir.join("config.toml"); std::fs::write(&path, cfg).expect("write config"); } write_config_file(openhuman_dir, &cfg); if openhuman_dir .file_name() .is_some_and(|name| name == std::ffi::OsStr::new(".openhuman")) { write_config_file(&openhuman_dir.join("users").join("local"), &cfg); } let _: openhuman_core::openhuman::config::Config = toml::from_str(&cfg).expect("config toml must match Config schema"); } #[tokio::test] async fn json_rpc_protocol_auth_and_agent_hello() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); // Always use the in-process Axum mock for /settings + /openai so this test does not pick up // BACKEND_URL/VITE_BACKEND_URL from the developer shell (e.g. mock-api that returns 401 for // the synthetic JWT used below). let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config(&openhuman_home, &mock_origin); // Pre-create the user-scoped config directory so that when store_session // activates user "e2e-user" and reloads config, it finds the correct // api_url and secrets.encrypt=false (rather than defaults). let user_scoped_dir = openhuman_home.join("users").join("e2e-user"); write_min_config(&user_scoped_dir, &mock_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; // --- core.ping (baseline protocol) --- let ping = post_json_rpc(&rpc_base, 1, "core.ping", json!({})).await; let ping_result = assert_no_jsonrpc_error(&ping, "core.ping"); assert_eq!(ping_result.get("ok"), Some(&json!(true))); // --- unknown method --- let unknown = post_json_rpc(&rpc_base, 2, "core.not_a_real_method", json!({})).await; assert!( unknown.get("error").is_some(), "expected error for unknown method: {unknown}" ); // --- auth: session state (no JWT yet) --- let state_before = post_json_rpc(&rpc_base, 3, "openhuman.auth_get_state", json!({})).await; let state_outer = assert_no_jsonrpc_error(&state_before, "get_state"); let state_body = state_outer.get("result").unwrap_or(state_outer); assert!( state_body.get("isAuthenticated").is_some() || state_body.get("is_authenticated").is_some(), "unexpected auth state shape: {state_body}" ); // --- auth: store session (validates JWT via mock GET /auth/me) --- let store = post_json_rpc( &rpc_base, 4, "openhuman.auth_store_session", json!({ "token": "e2e-test-jwt", "user_id": "e2e-user" }), ) .await; assert_no_jsonrpc_error(&store, "store_session"); // --- agent: single chat turn (mock chat completions) --- let chat = post_json_rpc( &rpc_base, 5, "openhuman.local_ai_agent_chat", json!({ "message": "Hello", }), ) .await; let chat_result = assert_no_jsonrpc_error(&chat, "agent_chat"); let reply = extract_string_outcome(chat_result); assert!( reply.contains("e2e mock") || reply.contains("Hello"), "unexpected agent reply: {reply:?}" ); // --- web channel RPC + SSE loop --- let client_id = "e2e-client-1"; let thread_id = "thread-1"; let events_url = format!("{}/events?client_id={}", rpc_base, client_id); let sse_task = tokio::spawn(async move { read_sse_event_by_type(&events_url, "chat_done").await }); let web_chat = post_json_rpc( &rpc_base, 6, "openhuman.channel_web_chat", json!({ "client_id": client_id, "thread_id": thread_id, "message": "Hello from web channel", "model_override": "e2e-mock-model", }), ) .await; let web_chat_result = assert_no_jsonrpc_error(&web_chat, "channel_web_chat"); assert_eq!( web_chat_result .get("result") .and_then(|v| v.get("accepted")), Some(&json!(true)) ); let sse_event = sse_task.await.expect("sse task join should succeed"); assert_eq!( sse_event.get("event").and_then(Value::as_str), Some("chat_done") ); assert_eq!( sse_event.get("thread_id").and_then(Value::as_str), Some(thread_id) ); assert!( sse_event .get("full_response") .and_then(Value::as_str) .unwrap_or_default() .len() > 0, "expected non-empty chat_done response payload: {sse_event}" ); mock_join.abort(); rpc_join.abort(); } #[tokio::test] async fn json_rpc_memory_tree_end_to_end() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); // Phase 4 (#710): disable strict embedding so ingest falls back to the // Inert (zero-vector) embedder when no Ollama endpoint is reachable. // CI has no local Ollama; without this the `memory_tree_ingest` call // would fail with `embed chunk_id= during ingest` before writing // any chunks. let _embed_strict_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_STRICT", "false"); let _embed_endpoint_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_ENDPOINT", ""); let _embed_model_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_MODEL", ""); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config(&openhuman_home, &mock_origin); let controllers = all_memory_tree_registered_controllers(); let expected_methods = vec![ "openhuman.memory_tree_ingest".to_string(), "openhuman.memory_tree_list_chunks".to_string(), "openhuman.memory_tree_get_chunk".to_string(), ]; assert_eq!(controllers.len(), expected_methods.len()); for method in &expected_methods { assert!( controllers .iter() .any(|controller| controller.rpc_method_name() == *method), "expected memory_tree controller registration for {method}" ); } let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; let ingest = post_json_rpc( &rpc_base, 200, &expected_methods[0], json!({ "source_kind": "document", "source_id": "notion:launch-plan", "owner": "alice@example.com", "tags": ["planning", "launch"], "payload": { "provider": "notion", "title": "Launch Plan", "body": "We decided to ship Phoenix on Friday after reviewing alice@example.com and the migration plan carefully. @bob will coordinate rollout, track #launch-q2 details, and update the Notion launch checklist with staging validation notes.", "modified_at": 1700000000000_i64, "source_ref": " notion://page/launch-plan " } }), ) .await; let ingest_outer = assert_no_jsonrpc_error(&ingest, "memory_tree_ingest"); let ingest_result = ingest_outer.get("result").unwrap_or(ingest_outer); assert_eq!( ingest_result.get("source_id"), Some(&json!("notion:launch-plan")) ); assert_eq!(ingest_result.get("chunks_written"), Some(&json!(1))); assert_eq!(ingest_result.get("chunks_dropped"), Some(&json!(0))); let chunk_ids = ingest_result .get("chunk_ids") .and_then(Value::as_array) .expect("chunk_ids array"); assert_eq!(chunk_ids.len(), 1); let list = post_json_rpc( &rpc_base, 201, &expected_methods[1], json!({ "source_kind": "document", "source_id": "notion:launch-plan", "owner": "alice@example.com", "limit": 0 }), ) .await; let list_outer = assert_no_jsonrpc_error(&list, "memory_tree_list_chunks"); let list_result = list_outer.get("result").unwrap_or(list_outer); let chunks = list_result .get("chunks") .and_then(Value::as_array) .expect("chunks array"); assert_eq!(chunks.len(), 1); let chunk = &chunks[0]; assert_eq!(chunk.get("seq_in_source"), Some(&json!(0))); assert_eq!( chunk.pointer("/metadata/source_ref/value"), Some(&json!("notion://page/launch-plan")) ); let get_chunk = post_json_rpc( &rpc_base, 202, &expected_methods[2], json!({ "id": chunk_ids[0].clone() }), ) .await; let get_outer = assert_no_jsonrpc_error(&get_chunk, "memory_tree_get_chunk"); let get_result = get_outer.get("result").unwrap_or(get_outer); assert_eq!(get_result.pointer("/chunk/id"), Some(&chunk_ids[0])); let invalid_ingest = post_json_rpc( &rpc_base, 203, &expected_methods[0], json!({ "source_kind": "document", "source_id": "notion:bad", "owner": "alice@example.com", "payload": { "provider": "notion", "title": "Bad payload" } }), ) .await; assert!( invalid_ingest.get("error").is_some(), "expected invalid payload JSON-RPC error: {invalid_ingest}" ); let invalid_list = post_json_rpc( &rpc_base, 204, &expected_methods[1], json!({ "source_kind": "not-a-kind" }), ) .await; assert!( invalid_list.get("error").is_some(), "expected invalid source_kind JSON-RPC error: {invalid_list}" ); rpc_join.abort(); let _ = rpc_join.await; mock_join.abort(); let _ = mock_join.await; } #[tokio::test] async fn json_rpc_web_chat_routing_cases_use_expected_backend_models() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config_with_local_ai_disabled(&openhuman_home, &mock_origin); let user_scoped_dir = openhuman_home.join("users").join("e2e-user"); write_min_config_with_local_ai_disabled(&user_scoped_dir, &mock_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; let store = post_json_rpc( &rpc_base, 1, "openhuman.auth_store_session", json!({ "token": "e2e-test-jwt", "user_id": "e2e-user" }), ) .await; assert_no_jsonrpc_error(&store, "store_session"); let routing_cases = [ ("hint:reasoning", "reasoning-v1"), ("hint:agentic", "agentic-v1"), ("hint:coding", "coding-v1"), ("reasoning-v1", "reasoning-v1"), // Web chat forwards lightweight hint overrides as-is for this path, // so the upstream model receives the original hint string. ("hint:reaction", "hint:reaction"), ]; for (idx, (model_override, expected_model)) in routing_cases.iter().enumerate() { with_chat_completion_models(|models| models.clear()); let client_id = format!("routing-case-client-{idx}"); let thread_id = format!("routing-case-thread-{idx}"); let events_url = format!("{}/events?client_id={}", rpc_base, client_id); let sse_task = tokio::spawn(async move { read_sse_event_by_type(&events_url, "chat_done").await }); let web_chat = post_json_rpc( &rpc_base, 100 + idx as i64, "openhuman.channel_web_chat", json!({ "client_id": client_id, "thread_id": thread_id, "message": format!("route case {idx}"), "model_override": model_override, }), ) .await; let web_chat_result = assert_no_jsonrpc_error(&web_chat, "channel_web_chat"); assert_eq!( web_chat_result .get("result") .and_then(|v| v.get("accepted")), Some(&json!(true)) ); let sse_event = tokio::time::timeout(Duration::from_secs(12), sse_task) .await .unwrap_or_else(|_| panic!("timed out waiting for chat_done for case {model_override}")) .expect("sse task join should succeed"); assert_eq!( sse_event.get("event").and_then(Value::as_str), Some("chat_done") ); let mut captured_models: Vec = Vec::new(); for _ in 0..50 { captured_models = with_chat_completion_models(|models| models.clone()); if captured_models.iter().any(|m| m == expected_model) { break; } tokio::time::sleep(Duration::from_millis(20)).await; } assert!( captured_models.iter().any(|m| m == expected_model), "case={model_override} expected={expected_model} captured={captured_models:?}" ); if model_override.starts_with("hint:") && *model_override != "hint:reaction" && *expected_model != *model_override { assert!( !captured_models.iter().any(|m| m == model_override), "hint model should not pass through for case={model_override}: {captured_models:?}" ); } } mock_join.abort(); rpc_join.abort(); } #[tokio::test] async fn json_rpc_rejects_non_object_params_with_clear_error() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config(&openhuman_home, &mock_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; let invalid = post_json_rpc( &rpc_base, 1001, "openhuman.auth_get_state", json!(["invalid", "params"]), ) .await; let err_message = invalid .get("error") .and_then(|e| e.get("message")) .and_then(Value::as_str) .unwrap_or(""); assert!( !err_message.is_empty(), "expected non-empty JSON-RPC error message: {invalid}" ); mock_join.abort(); rpc_join.abort(); } #[tokio::test] async fn json_rpc_screen_intelligence_capture_test_returns_stable_shape() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config(&openhuman_home, &mock_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; let capture = post_json_rpc( &rpc_base, 1002, "openhuman.screen_intelligence_capture_test", json!({}), ) .await; let capture_outer = assert_no_jsonrpc_error(&capture, "screen_intelligence_capture_test"); let capture_result = capture_outer.get("result").unwrap_or(capture_outer); assert!( capture_result.get("ok").and_then(Value::as_bool).is_some(), "expected bool ok field: {capture_result}" ); assert!( matches!( capture_result.get("capture_mode").and_then(Value::as_str), Some("windowed" | "fullscreen") ), "expected capture_mode field: {capture_result}" ); assert!( capture_result .get("timing_ms") .and_then(Value::as_u64) .is_some(), "expected timing_ms field: {capture_result}" ); let ok = capture_result .get("ok") .and_then(Value::as_bool) .expect("ok should be bool"); let image_ref = capture_result.get("image_ref").and_then(Value::as_str); let error = capture_result.get("error").and_then(Value::as_str); if ok { assert!( image_ref .map(|value| value.starts_with("data:image/png;base64,")) .unwrap_or(false), "successful capture should include a PNG data URL: {capture_result}" ); assert!( error.is_none(), "successful capture should not include an error" ); } else { assert!( image_ref.is_none(), "failed capture should not include image data" ); assert!( error.is_some(), "failed capture should include an error message" ); } mock_join.abort(); rpc_join.abort(); } #[tokio::test] async fn json_rpc_screen_intelligence_status_returns_stable_shape() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config(&openhuman_home, &mock_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; let status = post_json_rpc( &rpc_base, 1003, "openhuman.screen_intelligence_status", json!({}), ) .await; let result = assert_no_jsonrpc_error(&status, "screen_intelligence_status"); let status_result = result.get("result").unwrap_or(result); // Required top-level fields assert!( status_result .get("platform_supported") .and_then(Value::as_bool) .is_some(), "expected bool platform_supported: {status_result}" ); assert!( status_result .get("is_context_blocked") .and_then(Value::as_bool) .is_some(), "expected bool is_context_blocked: {status_result}" ); // session block let session = status_result .get("session") .expect("expected session object"); assert!( session.get("active").and_then(Value::as_bool).is_some(), "expected bool session.active: {status_result}" ); assert_eq!( session.get("active").and_then(Value::as_bool), Some(false), "session should not be active without start_session: {status_result}" ); assert!( session .get("capture_count") .and_then(Value::as_u64) .is_some(), "expected u64 session.capture_count: {status_result}" ); assert!( session .get("vision_persist_count") .and_then(Value::as_u64) .is_some(), "expected u64 session.vision_persist_count: {status_result}" ); assert!( session.get("last_vision_persist_error").is_some(), "expected nullable session.last_vision_persist_error: {status_result}" ); // permissions block let perms = status_result .get("permissions") .expect("expected permissions object"); assert!( perms .get("screen_recording") .and_then(Value::as_str) .is_some(), "expected string permissions.screen_recording: {status_result}" ); mock_join.abort(); rpc_join.abort(); } #[tokio::test] async fn json_rpc_app_state_snapshot_returns_runtime_shape() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config(&openhuman_home, &mock_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; let snapshot = post_json_rpc(&rpc_base, 1004, "openhuman.app_state_snapshot", json!({})).await; let result = assert_no_jsonrpc_error(&snapshot, "app_state_snapshot"); let body = result.get("result").unwrap_or(result); assert!( body.get("auth").and_then(Value::as_object).is_some(), "expected auth object: {body}" ); assert!( body.get("localState").and_then(Value::as_object).is_some(), "expected localState object: {body}" ); let runtime = body.get("runtime").expect("expected runtime object"); assert!( runtime .get("screenIntelligence") .and_then(Value::as_object) .is_some(), "expected runtime.screenIntelligence object: {runtime}" ); assert!( runtime.get("localAi").and_then(Value::as_object).is_some(), "expected runtime.localAi object: {runtime}" ); assert!( runtime .get("autocomplete") .and_then(Value::as_object) .is_some(), "expected runtime.autocomplete object: {runtime}" ); assert!( runtime.get("service").and_then(Value::as_object).is_some(), "expected runtime.service object: {runtime}" ); mock_join.abort(); rpc_join.abort(); } #[tokio::test] async fn json_rpc_screen_intelligence_vision_recent_returns_empty_without_session() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config(&openhuman_home, &mock_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; let recent = post_json_rpc( &rpc_base, 1004, "openhuman.screen_intelligence_vision_recent", json!({ "limit": 10 }), ) .await; let result = assert_no_jsonrpc_error(&recent, "screen_intelligence_vision_recent"); let recent_result = result.get("result").unwrap_or(result); let summaries = recent_result .get("summaries") .and_then(Value::as_array) .expect("expected summaries array: {recent_result}"); assert!( summaries.is_empty(), "vision_recent should return empty list without an active session, got {} items", summaries.len() ); mock_join.abort(); rpc_join.abort(); } #[cfg(target_os = "macos")] #[tokio::test] async fn json_rpc_autocomplete_runtime_settings_and_logs_flow() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config_with_local_ai_disabled(&openhuman_home, &mock_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; let set_style = post_json_rpc( &rpc_base, 2001, "openhuman.autocomplete_set_style", json!({ "enabled": true, "debounce_ms": 180, "max_chars": 160, "accept_with_tab": false, "style_preset": "balanced", "style_examples": ["[mail] ...Can you share an update? → Can you share a quick update?"], "disabled_apps": [] }), ) .await; let set_style_outer = assert_no_jsonrpc_error(&set_style, "autocomplete_set_style"); let set_style_payload = set_style_outer.get("result").unwrap_or(set_style_outer); let set_style_logs = set_style_outer .get("logs") .and_then(Value::as_array) .cloned() .unwrap_or_default(); assert_eq!( set_style_payload .get("config") .and_then(|v| v.get("debounce_ms")) .and_then(Value::as_u64), Some(180) ); assert_eq!( set_style_payload .get("config") .and_then(|v| v.get("max_chars")) .and_then(Value::as_u64), Some(160) ); assert!( set_style_logs.iter().any(|entry| { entry .as_str() .map(|s| s.contains("[autocomplete] set_style")) .unwrap_or(false) }), "expected structured set_style log line: {set_style_outer}" ); let cfg = post_json_rpc(&rpc_base, 2002, "openhuman.config_get", json!({})).await; let cfg_outer = assert_no_jsonrpc_error(&cfg, "get_config"); let cfg_payload = cfg_outer.get("result").unwrap_or(cfg_outer); let cfg_autocomplete = cfg_payload .get("config") .and_then(|v| v.get("autocomplete")) .expect("autocomplete config should exist"); assert_eq!( cfg_autocomplete.get("debounce_ms").and_then(Value::as_u64), Some(180) ); assert_eq!( cfg_autocomplete.get("max_chars").and_then(Value::as_u64), Some(160) ); assert_eq!( cfg_autocomplete .get("accept_with_tab") .and_then(Value::as_bool), Some(false) ); let start = post_json_rpc( &rpc_base, 2003, "openhuman.autocomplete_start", json!({ "debounce_ms": 180 }), ) .await; let start_outer = assert_no_jsonrpc_error(&start, "autocomplete_start"); let start_logs = start_outer .get("logs") .and_then(Value::as_array) .cloned() .unwrap_or_default(); assert!( start_logs.iter().any(|entry| { entry .as_str() .map(|s| s.contains("[autocomplete] start")) .unwrap_or(false) }), "expected structured start log line: {start_outer}" ); let status_running = post_json_rpc(&rpc_base, 2004, "openhuman.autocomplete_status", json!({})).await; let status_running_outer = assert_no_jsonrpc_error(&status_running, "autocomplete_status"); let status_running_payload = status_running_outer .get("result") .unwrap_or(status_running_outer); assert_eq!( status_running_payload .get("running") .and_then(Value::as_bool), Some(true) ); assert_eq!( status_running_payload .get("enabled") .and_then(Value::as_bool), Some(true) ); assert_eq!( status_running_payload .get("debounce_ms") .and_then(Value::as_u64), Some(180) ); let current = post_json_rpc( &rpc_base, 2005, "openhuman.autocomplete_current", json!({ "context": "Please review this changeset and" }), ) .await; let current_outer = assert_no_jsonrpc_error(¤t, "autocomplete_current"); let current_payload = current_outer.get("result").unwrap_or(current_outer); let current_logs = current_outer .get("logs") .and_then(Value::as_array) .cloned() .unwrap_or_default(); assert_eq!( current_payload.get("context").and_then(Value::as_str), Some("Please review this changeset and") ); assert!( current_logs.iter().any(|entry| { entry .as_str() .map(|s| s.contains("[autocomplete] current")) .unwrap_or(false) }), "expected structured current log line: {current_outer}" ); let accept = post_json_rpc( &rpc_base, 2006, "openhuman.autocomplete_accept", json!({ "suggestion": " share your thoughts.", "skip_apply": true }), ) .await; let accept_outer = assert_no_jsonrpc_error(&accept, "autocomplete_accept"); let accept_payload = accept_outer.get("result").unwrap_or(accept_outer); let accept_logs = accept_outer .get("logs") .and_then(Value::as_array) .cloned() .unwrap_or_default(); assert_eq!( accept_payload.get("accepted").and_then(Value::as_bool), Some(true) ); assert_eq!( accept_payload.get("applied").and_then(Value::as_bool), Some(false) ); assert!( accept_logs.iter().any(|entry| { entry .as_str() .map(|s| s.contains("[autocomplete] accept")) .unwrap_or(false) }), "expected structured accept log line: {accept_outer}" ); let stop = post_json_rpc( &rpc_base, 2007, "openhuman.autocomplete_stop", json!({ "reason": "json_rpc_e2e" }), ) .await; let stop_outer = assert_no_jsonrpc_error(&stop, "autocomplete_stop"); let stop_payload = stop_outer.get("result").unwrap_or(stop_outer); assert_eq!( stop_payload.get("stopped").and_then(Value::as_bool), Some(true) ); let status_stopped = post_json_rpc(&rpc_base, 2008, "openhuman.autocomplete_status", json!({})).await; let status_stopped_outer = assert_no_jsonrpc_error(&status_stopped, "autocomplete_status"); let status_stopped_payload = status_stopped_outer .get("result") .unwrap_or(status_stopped_outer); assert_eq!( status_stopped_payload .get("running") .and_then(Value::as_bool), Some(false) ); mock_join.abort(); rpc_join.abort(); } // --------------------------------------------------------------------------- // Local AI device profile, presets, and apply preset // --------------------------------------------------------------------------- #[tokio::test] async fn json_rpc_local_ai_device_profile_and_presets() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let _tier_guard = EnvVarGuard::unset("OPENHUMAN_LOCAL_AI_TIER"); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config(&openhuman_home, &mock_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; // --- device_profile --- let profile = post_json_rpc( &rpc_base, 30, "openhuman.local_ai_device_profile", json!({}), ) .await; let profile_result = assert_no_jsonrpc_error(&profile, "device_profile"); assert!( profile_result .get("total_ram_bytes") .and_then(Value::as_u64) .unwrap_or(0) > 0, "expected positive RAM: {profile_result}" ); assert!( profile_result .get("cpu_count") .and_then(Value::as_u64) .unwrap_or(0) > 0, "expected positive CPU count: {profile_result}" ); // --- presets --- let presets = post_json_rpc(&rpc_base, 31, "openhuman.local_ai_presets", json!({})).await; let presets_result = assert_no_jsonrpc_error(&presets, "presets"); let presets_arr = presets_result .get("presets") .and_then(Value::as_array) .expect("presets should be an array"); assert_eq!(presets_arr.len(), 5, "expected 5 presets: {presets_result}"); let recommended = presets_result .get("recommended_tier") .and_then(Value::as_str) .expect("should have recommended_tier"); assert!( [ "ram_1gb", "ram_2_4gb", "ram_4_8gb", "ram_8_16gb", "ram_16_plus_gb", ] .contains(&recommended), "unexpected recommended_tier: {recommended}" ); let current = presets_result .get("current_tier") .and_then(Value::as_str) .expect("should have current_tier"); // Default config uses gemma3:4b-it-qat which now maps to the 8-16 GB tier. assert_eq!( current, "ram_8_16gb", "default config should be the 8-16 GB tier" ); // --- apply_preset (switch to 2-4 GB) --- let apply = post_json_rpc( &rpc_base, 32, "openhuman.local_ai_apply_preset", json!({"tier": "ram_2_4gb"}), ) .await; let apply_result = assert_no_jsonrpc_error(&apply, "apply_preset"); assert_eq!( apply_result.get("applied_tier").and_then(Value::as_str), Some("ram_2_4gb") ); assert_eq!( apply_result.get("chat_model_id").and_then(Value::as_str), Some("gemma3:1b-it-qat") ); assert_eq!( apply_result.get("vision_mode").and_then(Value::as_str), Some("disabled") ); // --- verify presets reflects the change --- let presets_after = post_json_rpc(&rpc_base, 33, "openhuman.local_ai_presets", json!({})).await; let presets_after_result = assert_no_jsonrpc_error(&presets_after, "presets_after"); assert_eq!( presets_after_result .get("current_tier") .and_then(Value::as_str), Some("ram_2_4gb"), "current tier should now be 2-4 GB after apply" ); // --- apply_preset with invalid tier should error --- let bad_apply = post_json_rpc( &rpc_base, 34, "openhuman.local_ai_apply_preset", json!({"tier": "ultra"}), ) .await; assert!( bad_apply.get("error").is_some(), "expected error for invalid tier: {bad_apply}" ); mock_join.abort(); rpc_join.abort(); } // ── Billing & Team E2E tests ────────────────────────────────────────────────── /// End-to-end test for billing RPC methods. /// /// Spins up an in-process Axum mock backend and a real JSON-RPC server, stores a /// session JWT, then exercises every billing controller through the RPC surface /// exactly as the desktop app or a CI script would. #[tokio::test] async fn billing_rpc_e2e() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config(&openhuman_home, &mock_origin); // Pre-create the user-scoped config so store_session finds correct settings. let user_scoped_dir = openhuman_home.join("users").join("e2e-user"); write_min_config(&user_scoped_dir, &mock_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; // Store a session first — all billing methods require it. let store = post_json_rpc( &rpc_base, 1, "openhuman.auth_store_session", json!({ "token": "e2e-billing-jwt", "user_id": "e2e-user" }), ) .await; assert_no_jsonrpc_error(&store, "store_session"); // Helper: the RPC outcome wraps backend data in {result: ..., logs: [...]}. // We peel off the inner "result" field to get the actual backend payload. fn inner(outer: &Value, _ctx: &str) -> Value { outer .get("result") .cloned() .unwrap_or_else(|| outer.clone()) } // --- billing_get_current_plan --- let plan = post_json_rpc( &rpc_base, 2, "openhuman.billing_get_current_plan", json!({}), ) .await; let plan_outer = assert_no_jsonrpc_error(&plan, "billing_get_current_plan"); let plan_result = inner(plan_outer, "billing_get_current_plan"); assert_eq!( plan_result.get("plan").and_then(Value::as_str), Some("PRO"), "expected PRO plan: {plan_result}" ); assert_eq!( plan_result .get("hasActiveSubscription") .and_then(Value::as_bool), Some(true), "expected active subscription: {plan_result}" ); // --- billing_purchase_plan --- let purchase = post_json_rpc( &rpc_base, 3, "openhuman.billing_purchase_plan", json!({ "plan": "pro" }), ) .await; let purchase_outer = assert_no_jsonrpc_error(&purchase, "billing_purchase_plan"); let purchase_result = inner(purchase_outer, "billing_purchase_plan"); assert!( purchase_result .get("checkoutUrl") .and_then(Value::as_str) .is_some(), "expected checkoutUrl: {purchase_result}" ); // --- billing_create_portal_session --- let portal = post_json_rpc( &rpc_base, 4, "openhuman.billing_create_portal_session", json!({}), ) .await; let portal_outer = assert_no_jsonrpc_error(&portal, "billing_create_portal_session"); let portal_result = inner(portal_outer, "billing_create_portal_session"); assert!( portal_result .get("portalUrl") .and_then(Value::as_str) .is_some(), "expected portalUrl: {portal_result}" ); // --- billing_top_up --- let top_up = post_json_rpc( &rpc_base, 5, "openhuman.billing_top_up", json!({ "amountUsd": 10.0, "gateway": "stripe" }), ) .await; let top_up_outer = assert_no_jsonrpc_error(&top_up, "billing_top_up"); let top_up_result = inner(top_up_outer, "billing_top_up"); assert_eq!( top_up_result.get("amountUsd").and_then(Value::as_f64), Some(10.0), "expected amountUsd 10.0: {top_up_result}" ); // --- billing_create_coinbase_charge --- let charge = post_json_rpc( &rpc_base, 6, "openhuman.billing_create_coinbase_charge", json!({ "plan": "pro" }), ) .await; let charge_outer = assert_no_jsonrpc_error(&charge, "billing_create_coinbase_charge"); let charge_result = inner(charge_outer, "billing_create_coinbase_charge"); assert!( charge_result .get("hostedUrl") .and_then(Value::as_str) .is_some(), "expected hostedUrl: {charge_result}" ); assert_eq!( charge_result.get("status").and_then(Value::as_str), Some("NEW"), "expected NEW status: {charge_result}" ); mock_join.abort(); rpc_join.abort(); } /// End-to-end test for team RPC methods. /// /// Spins up an in-process Axum mock backend and a real JSON-RPC server, stores a /// session JWT, then exercises every team controller through the RPC surface. #[tokio::test] async fn team_rpc_e2e() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config(&openhuman_home, &mock_origin); // Pre-create the user-scoped config so store_session finds correct settings. let user_scoped_dir = openhuman_home.join("users").join("e2e-user"); write_min_config(&user_scoped_dir, &mock_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; // Store a session first — all team methods require it. let store = post_json_rpc( &rpc_base, 1, "openhuman.auth_store_session", json!({ "token": "e2e-team-jwt", "user_id": "e2e-user" }), ) .await; assert_no_jsonrpc_error(&store, "store_session"); // Helper: peel off the inner "result" field from the RPC outcome envelope. fn inner(outer: &Value, _ctx: &str) -> Value { outer .get("result") .cloned() .unwrap_or_else(|| outer.clone()) } let team_id = "team-1"; // --- team_list_members --- let members = post_json_rpc( &rpc_base, 2, "openhuman.team_list_members", json!({ "teamId": team_id }), ) .await; let members_outer = assert_no_jsonrpc_error(&members, "team_list_members"); let members_result = inner(members_outer, "team_list_members"); let members_arr = members_result .as_array() .expect("expected array of members"); assert_eq!(members_arr.len(), 2, "expected 2 members: {members_result}"); assert_eq!( members_arr[0].get("username").and_then(Value::as_str), Some("alice") ); // --- team_create_invite --- let invite = post_json_rpc( &rpc_base, 3, "openhuman.team_create_invite", json!({ "teamId": team_id, "maxUses": 3, "expiresInDays": 7 }), ) .await; let invite_outer = assert_no_jsonrpc_error(&invite, "team_create_invite"); let invite_result = inner(invite_outer, "team_create_invite"); assert!( invite_result.get("code").and_then(Value::as_str).is_some(), "expected invite code: {invite_result}" ); // --- team_list_invites --- let invites = post_json_rpc( &rpc_base, 4, "openhuman.team_list_invites", json!({ "teamId": team_id }), ) .await; let invites_outer = assert_no_jsonrpc_error(&invites, "team_list_invites"); let invites_result = inner(invites_outer, "team_list_invites"); let invites_arr = invites_result .as_array() .expect("expected array of invites"); assert!( !invites_arr.is_empty(), "expected at least one invite: {invites_result}" ); // --- team_revoke_invite (no payload to check, just assert no error) --- let revoke = post_json_rpc( &rpc_base, 5, "openhuman.team_revoke_invite", json!({ "teamId": team_id, "inviteId": "inv-1" }), ) .await; assert_no_jsonrpc_error(&revoke, "team_revoke_invite"); // --- team_remove_member --- let remove = post_json_rpc( &rpc_base, 6, "openhuman.team_remove_member", json!({ "teamId": team_id, "userId": "user-2" }), ) .await; assert_no_jsonrpc_error(&remove, "team_remove_member"); // --- team_change_member_role --- let role_change = post_json_rpc( &rpc_base, 7, "openhuman.team_change_member_role", json!({ "teamId": team_id, "userId": "user-1", "role": "MEMBER" }), ) .await; assert_no_jsonrpc_error(&role_change, "team_change_member_role"); mock_join.abort(); rpc_join.abort(); } #[tokio::test] async fn about_app_rpc_list_lookup_and_search() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config(&openhuman_home, &mock_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; fn inner(outer: &Value) -> Value { outer .get("result") .cloned() .unwrap_or_else(|| outer.clone()) } let list = post_json_rpc(&rpc_base, 200, "openhuman.about_app_list", json!({})).await; let list_outer = assert_no_jsonrpc_error(&list, "about_app_list"); let list_result = inner(list_outer); let capabilities = list_result .as_array() .expect("about_app list should return an array"); assert!( capabilities.len() >= 40, "expected large capability catalog, got: {list_result}" ); assert!(capabilities.iter().any(|capability| { capability.get("id").and_then(Value::as_str) == Some("local_ai.download_model") })); let filtered = post_json_rpc( &rpc_base, 201, "openhuman.about_app_list", json!({ "category": "local_ai" }), ) .await; let filtered_outer = assert_no_jsonrpc_error(&filtered, "about_app_list filtered"); let filtered_result = inner(filtered_outer); let filtered_capabilities = filtered_result .as_array() .expect("filtered about_app list should return an array"); assert!( !filtered_capabilities.is_empty(), "expected local_ai capabilities: {filtered_result}" ); assert!(filtered_capabilities.iter().all(|capability| { capability.get("category").and_then(Value::as_str) == Some("local_ai") })); let lookup = post_json_rpc( &rpc_base, 202, "openhuman.about_app_lookup", json!({ "id": "team.generate_invite_codes" }), ) .await; let lookup_outer = assert_no_jsonrpc_error(&lookup, "about_app_lookup"); let lookup_result = inner(lookup_outer); assert_eq!( lookup_result.get("id").and_then(Value::as_str), Some("team.generate_invite_codes") ); assert_eq!( lookup_result.get("category").and_then(Value::as_str), Some("team") ); let search = post_json_rpc( &rpc_base, 203, "openhuman.about_app_search", json!({ "query": "invite" }), ) .await; let search_outer = assert_no_jsonrpc_error(&search, "about_app_search"); let search_result = inner(search_outer); let search_capabilities = search_result .as_array() .expect("about_app search should return an array"); assert!( search_capabilities.iter().any(|capability| { capability.get("id").and_then(Value::as_str) == Some("team.join_via_invite_code") }), "expected invite-related capability in search results: {search_result}" ); assert!( search_capabilities.iter().any(|capability| { capability.get("id").and_then(Value::as_str) == Some("team.generate_invite_codes") }), "expected invite generation capability in search results: {search_result}" ); mock_join.abort(); rpc_join.abort(); } #[tokio::test] async fn voice_status_returns_availability() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let _whisper_guard = EnvVarGuard::unset("WHISPER_BIN"); let _piper_guard = EnvVarGuard::unset("PIPER_BIN"); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config(&openhuman_home, &mock_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; // voice_status does not require auth — it only checks filesystem availability let status = post_json_rpc(&rpc_base, 1, "openhuman.voice_status", json!({})).await; let result = assert_no_jsonrpc_error(&status, "voice_status"); // Without whisper/piper installed in the test env, both should be unavailable assert!( result.get("stt_available").is_some(), "expected stt_available field: {result}" ); assert!( result.get("tts_available").is_some(), "expected tts_available field: {result}" ); assert!( result.get("stt_model_id").is_some(), "expected stt_model_id field: {result}" ); assert!( result.get("tts_voice_id").is_some(), "expected tts_voice_id field: {result}" ); // Verify that without binaries, availability is false assert_eq!( result.get("stt_available").and_then(Value::as_bool), Some(false), "stt should be unavailable without whisper binary" ); assert_eq!( result.get("tts_available").and_then(Value::as_bool), Some(false), "tts should be unavailable without piper binary" ); mock_join.abort(); rpc_join.abort(); } #[tokio::test] async fn notification_settings_roundtrip_and_disabled_ingest_skip() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let openhuman_home = home.join(".openhuman"); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; let mock_origin = format!("http://{}", mock_addr); write_min_config(&openhuman_home, &mock_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{}", rpc_addr); tokio::time::sleep(Duration::from_millis(100)).await; let set = post_json_rpc( &rpc_base, 4001, "openhuman.notification_settings_set", json!({ "provider": "gmail", "enabled": false, "importance_threshold": 0.8, "route_to_orchestrator": false }), ) .await; let set_result = assert_no_jsonrpc_error(&set, "notification_settings_set"); assert_eq!(set_result.get("ok").and_then(Value::as_bool), Some(true)); let get = post_json_rpc( &rpc_base, 4002, "openhuman.notification_settings_get", json!({ "provider": "gmail" }), ) .await; let get_result = assert_no_jsonrpc_error(&get, "notification_settings_get"); let settings = get_result.get("settings").expect("settings object"); assert_eq!( settings.get("enabled").and_then(Value::as_bool), Some(false) ); let threshold = settings .get("importance_threshold") .and_then(Value::as_f64) .unwrap_or_default(); assert!( (threshold - 0.8).abs() < 0.0001, "expected threshold ~= 0.8, got {threshold}" ); assert_eq!( settings .get("route_to_orchestrator") .and_then(Value::as_bool), Some(false) ); let ingest = post_json_rpc( &rpc_base, 4003, "openhuman.notification_ingest", json!({ "provider": "gmail", "account_id": "acct-1", "title": "subject", "body": "body", "raw_payload": { "source": "test" } }), ) .await; let ingest_result = assert_no_jsonrpc_error(&ingest, "notification_ingest"); assert_eq!( ingest_result.get("skipped").and_then(Value::as_bool), Some(true) ); mock_join.abort(); rpc_join.abort(); } /// End-to-end coverage for `openhuman.skills_uninstall`. /// /// Validates that the RPC method is registered, wire-decodes /// `UninstallSkillParams`, resolves the slug against /// `~/.openhuman/skills//`, removes the directory on success, and /// forwards the core error message verbatim for the two documented /// failure modes (missing SKILL.md and path traversal). Previously only /// the `uninstall_skill(...)` helper was tested — the wire layer /// (controller registration, param decoding, response shape) was not. #[tokio::test] async fn skills_uninstall_rpc_e2e() { let _env_lock = json_rpc_e2e_env_lock(); let tmp = tempdir().expect("tempdir"); let home = tmp.path(); let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); let skills_root = home.join(".openhuman").join("skills"); std::fs::create_dir_all(&skills_root).expect("mkdir skills root"); // Seed a skill whose on-disk slug differs from its frontmatter name — // mirrors the bug CodeRabbit flagged for #781: the UI must send the // slug (`SkillSummary.id` / directory name), not the display name. let slug = "weather-helper"; let skill_dir = skills_root.join(slug); std::fs::create_dir_all(&skill_dir).expect("mkdir skill dir"); std::fs::write( skill_dir.join("SKILL.md"), "---\nname: Weather Helper\ndescription: fetches local weather\n---\n# body\n", ) .expect("write SKILL.md"); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{rpc_addr}"); // --- success path ------------------------------------------------------ let ok = post_json_rpc( &rpc_base, 6001, "openhuman.skills_uninstall", json!({ "name": slug }), ) .await; let ok_result = assert_no_jsonrpc_error(&ok, "skills_uninstall success"); assert_eq!( ok_result.get("name").and_then(Value::as_str), Some(slug), "response echoes the slug we passed" ); assert_eq!( ok_result.get("scope").and_then(Value::as_str), Some("user"), "uninstall is user-scope only" ); let removed_path = ok_result .get("removed_path") .and_then(Value::as_str) .expect("removed_path in response"); assert!( removed_path.ends_with(slug) || removed_path.contains(&format!("skills{}{slug}", std::path::MAIN_SEPARATOR)), "removed_path should reference the slug dir, got: {removed_path}" ); assert!( !skill_dir.exists(), "directory must be gone after uninstall" ); // --- not-installed path: core error forwarded verbatim ---------------- let missing = post_json_rpc( &rpc_base, 6002, "openhuman.skills_uninstall", json!({ "name": "does-not-exist" }), ) .await; let err = missing .get("error") .unwrap_or_else(|| panic!("expected error, got {missing}")); let err_msg = err .get("message") .and_then(Value::as_str) .or_else(|| err.get("data").and_then(Value::as_str)) .unwrap_or(""); assert!( err_msg.contains("not installed") || err.to_string().contains("not installed"), "expected verbatim 'not installed' error, got: {err}" ); // --- path-traversal path: core error forwarded verbatim --------------- let traversal = post_json_rpc( &rpc_base, 6003, "openhuman.skills_uninstall", json!({ "name": "../etc" }), ) .await; let traversal_err = traversal .get("error") .unwrap_or_else(|| panic!("expected error, got {traversal}")); let traversal_msg = traversal_err.to_string(); assert!( traversal_msg.contains("path separators") || traversal_msg.contains("path escapes") || traversal_msg.contains("not installed"), "expected traversal rejection error, got: {traversal_err}" ); rpc_join.abort(); }