//! 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::auth::{init_rpc_token, CORE_TOKEN_ENV_VAR}; use openhuman_core::core::jsonrpc::build_core_http_router; use openhuman_core::openhuman::memory::all_memory_tree_registered_controllers; const TEST_RPC_TOKEN: &str = "json-rpc-e2e-local-token"; static JSON_RPC_AUTH_INIT: OnceLock<()> = OnceLock::new(); 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())); } let is_triage_turn = body .get("messages") .and_then(Value::as_array) .map(|messages| { messages.iter().any(|m| { m.get("content") .and_then(Value::as_str) .is_some_and(|content| { content.contains("SOURCE: ") && content.contains("DISPLAY_LABEL: ") && content.contains("PAYLOAD:") }) }) }) .unwrap_or(false); let content = if is_triage_turn { "{\"action\":\"react\",\"reason\":\"e2e triage mock\"}" } else { "Hello from e2e mock agent" }; Json(json!({ "choices": [{ "message": { "role": "assistant", "content": content } }] })) } // ── 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>, ) { ensure_test_rpc_auth(); 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) .header(AUTHORIZATION, format!("Bearer {TEST_RPC_TOKEN}")) .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) .header(AUTHORIZATION, format!("Bearer {TEST_RPC_TOKEN}")) .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) .header(AUTHORIZATION, format!("Bearer {TEST_RPC_TOKEN}")) .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"); } fn ensure_test_rpc_auth() { JSON_RPC_AUTH_INIT.get_or_init(|| { // SAFETY: set_var is inside get_or_init so it runs exactly once across // all test threads. Rust 1.81+ requires unsafe for set_var in // multi-threaded contexts; the OnceLock guard limits the mutation to a // single call at init time, before any concurrent env reads occur. unsafe { std::env::set_var(CORE_TOKEN_ENV_VAR, TEST_RPC_TOKEN) }; let token_dir = std::env::temp_dir().join("openhuman-json-rpc-e2e-auth"); init_rpc_token(&token_dir).expect("init rpc auth token for json_rpc_e2e"); }); } #[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_thread_labels_create_and_update() { 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_url_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); let _api_url_guard = EnvVarGuard::unset("OPENHUMAN_API_URL"); let (api_addr, api_join) = serve_on_ephemeral(mock_upstream_router()).await; let api_origin = format!("http://{api_addr}"); write_min_config(openhuman_home.as_path(), &api_origin); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let rpc_base = format!("http://{rpc_addr}"); // 1. Create a thread with an explicit label. let create = post_json_rpc( &rpc_base, 9001, "openhuman.threads_create_new", json!({ "labels": ["custom"] }), ) .await; let create_outer = assert_no_jsonrpc_error(&create, "threads_create_new with labels"); let created = create_outer .get("data") .expect("data envelope in create response"); let thread_id = created .get("id") .and_then(Value::as_str) .expect("id in created thread"); let created_labels = created .get("labels") .and_then(Value::as_array) .expect("labels in created thread"); assert_eq!( created_labels .iter() .map(|v| v.as_str().unwrap_or("")) .collect::>(), vec!["custom"], "created thread should have labels=[\"custom\"]" ); // 2. Update labels on the thread. let update = post_json_rpc( &rpc_base, 9002, "openhuman.threads_update_labels", json!({ "thread_id": thread_id, "labels": ["work", "briefing"] }), ) .await; let update_outer = assert_no_jsonrpc_error(&update, "threads_update_labels"); let updated = update_outer .get("data") .expect("data envelope in update response"); let updated_labels = updated .get("labels") .and_then(Value::as_array) .expect("labels in updated thread"); assert_eq!( updated_labels .iter() .map(|v| v.as_str().unwrap_or("")) .collect::>(), vec!["work", "briefing"], "updated thread should have labels=[\"work\", \"briefing\"]" ); // 3. Verify the updated labels are reflected in threads_list. let list = post_json_rpc(&rpc_base, 9003, "openhuman.threads_list", json!({})).await; let list_outer = assert_no_jsonrpc_error(&list, "threads_list after label update"); let list_result = list_outer .get("data") .expect("data envelope in list response"); let threads = list_result .get("threads") .and_then(Value::as_array) .expect("threads array in list"); let persisted = threads .iter() .find(|t| t.get("id").and_then(Value::as_str) == Some(thread_id)) .expect("created thread must appear in list"); let persisted_labels = persisted .get("labels") .and_then(Value::as_array) .expect("labels in persisted thread"); assert_eq!( persisted_labels .iter() .map(|v| v.as_str().unwrap_or("")) .collect::>(), vec!["work", "briefing"], "threads_list must reflect the updated labels" ); api_join.abort(); rpc_join.abort(); } #[tokio::test] async fn json_rpc_memory_sync_and_learn() { 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 _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 (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; // ── memory_sync_all: returns requested:true ────────────────────────────── let sync_all = post_json_rpc(&rpc_base, 7001, "openhuman.memory_sync_all", json!({})).await; let sync_all_result = assert_no_jsonrpc_error(&sync_all, "memory_sync_all"); assert_eq!( sync_all_result.get("requested"), Some(&json!(true)), "memory_sync_all must return requested:true" ); // ── memory_sync_channel: echoes channel_id and returns requested:true ───── let sync_ch = post_json_rpc( &rpc_base, 7002, "openhuman.memory_sync_channel", json!({ "channel_id": "test-channel-abc" }), ) .await; let sync_ch_result = assert_no_jsonrpc_error(&sync_ch, "memory_sync_channel"); assert_eq!( sync_ch_result.get("requested"), Some(&json!(true)), "memory_sync_channel must return requested:true" ); assert_eq!( sync_ch_result.get("channel_id").and_then(Value::as_str), Some("test-channel-abc"), "memory_sync_channel must echo channel_id" ); // ── memory_sync_channel: missing channel_id returns a JSON-RPC error ──── let sync_bad = post_json_rpc(&rpc_base, 7003, "openhuman.memory_sync_channel", json!({})).await; assert!( sync_bad.get("error").is_some(), "missing channel_id must return an error, got: {sync_bad}" ); // ── memory_learn_all: no namespaces → zero processed (empty store) ────── let learn_all = post_json_rpc(&rpc_base, 7004, "openhuman.memory_learn_all", json!({})).await; let learn_result = assert_no_jsonrpc_error(&learn_all, "memory_learn_all"); let processed = learn_result .get("namespaces_processed") .and_then(Value::as_u64) .expect("namespaces_processed must be present"); assert_eq!(processed, 0, "no namespaces in a fresh store"); let results_arr = learn_result .get("results") .and_then(Value::as_array) .expect("results array must be present"); assert!( results_arr.is_empty(), "results must be empty when no namespaces" ); // ── memory_learn_all: constrained to non-existent namespace → also zero ── let learn_constrained = post_json_rpc( &rpc_base, 7005, "openhuman.memory_learn_all", json!({ "namespaces": ["does-not-exist"] }), ) .await; let learn_c_result = assert_no_jsonrpc_error(&learn_constrained, "memory_learn_all constrained"); assert_eq!( learn_c_result .get("namespaces_processed") .and_then(Value::as_u64), Some(0), "non-existent namespace must be filtered out" ); 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(), "openhuman.memory_tree_trigger_digest".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}" ); assert_eq!( body.get("onboardingCompleted").and_then(Value::as_bool), Some(false), "expected onboardingCompleted=false default: {body}" ); // Welcome-lockdown frontend gate (#883). `write_min_config` sets // `chat_onboarding_completed = true` so the test harness bypasses the // welcome agent; the snapshot must surface the same camelCase key the // React app reads. assert_eq!( body.get("chatOnboardingCompleted").and_then(Value::as_bool), Some(true), "expected chatOnboardingCompleted=true from test config: {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(); } /// #883 — when `chat_onboarding_completed` is unset in config.toml (fresh /// user), the `openhuman.app_state_snapshot` RPC must surface the flag as /// `false` so the React welcome-lockdown kicks in. #[tokio::test] async fn json_rpc_app_state_snapshot_chat_onboarding_defaults_false() { 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); // Fresh-user config: no `chat_onboarding_completed` key → serde default // of `false`. Cannot reuse `write_min_config` because it hard-codes the // flag to `true` so the e2e mock can bypass the welcome agent. let cfg = format!( r#"api_url = "{mock_origin}" default_model = "e2e-mock-model" default_temperature = 0.7 [secrets] encrypt = false "# ); std::fs::create_dir_all(&openhuman_home).expect("mkdir openhuman"); std::fs::write(openhuman_home.join("config.toml"), &cfg).expect("write config"); std::fs::create_dir_all(openhuman_home.join("users").join("local")).expect("mkdir users/local"); std::fs::write( openhuman_home .join("users") .join("local") .join("config.toml"), &cfg, ) .expect("write user config"); 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, 1005, "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_eq!( body.get("chatOnboardingCompleted").and_then(Value::as_bool), Some(false), "fresh-user config without chat_onboarding_completed must surface chatOnboardingCompleted=false: {body}" ); 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(), 1, "MVP exposes only the 1B preset: {presets_result}" ); assert_eq!( presets_arr[0].get("tier").and_then(Value::as_str), Some("ram_2_4gb"), "only the ram_2_4gb (1B) preset should be exposed: {presets_result}" ); let recommended = presets_result .get("recommended_tier") .and_then(Value::as_str) .expect("should have recommended_tier"); assert_eq!( recommended, "ram_2_4gb", "MVP recommends the only allowed tier: {recommended}" ); let current = presets_result .get("current_tier") .and_then(Value::as_str) .expect("should have current_tier"); // Default config now uses gemma3:1b-it-qat which maps to the only allowed (2-4 GB) tier. assert_eq!( current, "ram_2_4gb", "default config should be the 1B / 2-4 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(); } #[tokio::test] async fn credentials_crud_roundtrip() { // Tests the provider-credential lifecycle over the JSON-RPC transport: // store → list → list-filtered → remove → verify-gone // // Provider credentials are stored locally (auth-profiles.json) and require // no upstream network calls, so no mock session/JWT is needed. 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"); // A mock upstream is required so config validation passes and api_url is // well-formed, even though provider-credential calls don't hit the network. 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; // ── 1. store a provider credential ────────────────────────────────────── let store = post_json_rpc( &rpc_base, 5001, "openhuman.auth_store_provider_credentials", json!({ "provider": "openai", "profile": "default", "token": "sk-e2e-test-key", "setActive": true }), ) .await; // assert_no_jsonrpc_error returns the JSON-RPC `result` field which is the // RpcOutcome envelope: {"logs": [...], "result": { }}. let store_outer = assert_no_jsonrpc_error(&store, "auth_store_provider_credentials"); let store_result = store_outer.get("result").unwrap_or(store_outer); assert_eq!( store_result.get("provider").and_then(Value::as_str), Some("openai"), "stored profile should have provider=openai: {store_result}" ); assert_eq!( store_result.get("profileName").and_then(Value::as_str), Some("default"), "stored profile should have profileName=default: {store_result}" ); assert_eq!( store_result.get("hasToken").and_then(Value::as_bool), Some(true), "stored profile should report hasToken=true: {store_result}" ); // ── 2. list all provider credentials — should find openai ─────────────── let list_all = post_json_rpc( &rpc_base, 5002, "openhuman.auth_list_provider_credentials", json!({}), ) .await; let list_outer = assert_no_jsonrpc_error(&list_all, "auth_list_provider_credentials (all)"); let list_result = list_outer.get("result").unwrap_or(list_outer); let profiles = list_result .as_array() .unwrap_or_else(|| panic!("expected array from list: {list_result}")); assert_eq!(profiles.len(), 1, "expected exactly one stored credential"); assert_eq!( profiles[0].get("provider").and_then(Value::as_str), Some("openai") ); // ── 3. list filtered by provider name ─────────────────────────────────── let list_filtered = post_json_rpc( &rpc_base, 5003, "openhuman.auth_list_provider_credentials", json!({ "provider": "openai" }), ) .await; let filtered_outer = assert_no_jsonrpc_error(&list_filtered, "auth_list_provider_credentials (filtered)"); let filtered_result = filtered_outer.get("result").unwrap_or(filtered_outer); let filtered_profiles = filtered_result .as_array() .unwrap_or_else(|| panic!("expected array from filtered list: {filtered_result}")); assert_eq!( filtered_profiles.len(), 1, "filter by openai should return exactly one entry" ); // ── 4. remove the stored credential ───────────────────────────────────── let remove = post_json_rpc( &rpc_base, 5004, "openhuman.auth_remove_provider_credentials", json!({ "provider": "openai", "profile": "default" }), ) .await; let remove_outer = assert_no_jsonrpc_error(&remove, "auth_remove_provider_credentials"); let remove_result = remove_outer.get("result").unwrap_or(remove_outer); assert_eq!( remove_result.get("removed").and_then(Value::as_bool), Some(true), "remove should report removed=true: {remove_result}" ); // ── 5. verify the credential is gone ──────────────────────────────────── let list_after = post_json_rpc( &rpc_base, 5005, "openhuman.auth_list_provider_credentials", json!({}), ) .await; let after_outer = assert_no_jsonrpc_error(&list_after, "auth_list_provider_credentials (after remove)"); let after_result = after_outer.get("result").unwrap_or(after_outer); let after_profiles = after_result .as_array() .unwrap_or_else(|| panic!("expected array after remove: {after_result}")); assert!( after_profiles.is_empty(), "credentials list should be empty after remove, got {after_profiles:?}" ); 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(); } // --------------------------------------------------------------------------- // Auth middleware tests // --------------------------------------------------------------------------- /// POST /rpc without any Authorization header → 401 with error=unauthorized. #[tokio::test] async fn rpc_rejects_unauthenticated_request() { let _env_lock = json_rpc_e2e_env_lock(); ensure_test_rpc_auth(); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let client = reqwest::Client::new(); let resp = client .post(format!("http://{rpc_addr}/rpc")) .header("Content-Type", "application/json") .body(r#"{"jsonrpc":"2.0","id":1,"method":"core.ping","params":{}}"#) .send() .await .expect("request"); assert_eq!(resp.status(), 401, "missing Authorization must yield 401"); let body: Value = resp.json().await.expect("json body"); assert_eq!( body["error"], "unauthorized", "error field must be 'unauthorized'" ); rpc_join.abort(); } /// POST /rpc with a syntactically valid but wrong bearer token → 401. #[tokio::test] async fn rpc_rejects_wrong_token() { let _env_lock = json_rpc_e2e_env_lock(); ensure_test_rpc_auth(); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let client = reqwest::Client::new(); let resp = client .post(format!("http://{rpc_addr}/rpc")) .header( AUTHORIZATION, "Bearer deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef", ) .header("Content-Type", "application/json") .body(r#"{"jsonrpc":"2.0","id":1,"method":"core.ping","params":{}}"#) .send() .await .expect("request"); assert_eq!(resp.status(), 401, "wrong token must yield 401"); let body: Value = resp.json().await.expect("json body"); assert_eq!(body["error"], "unauthorized"); rpc_join.abort(); } /// Every path in PUBLIC_PATHS must bypass the auth middleware — i.e. never /// return 401 — even without an Authorization header. Some paths return /// non-2xx for other reasons (missing query params, no WebSocket upgrade /// headers) so the assertion is `!= 401`, not `.is_success()`. #[tokio::test] async fn public_paths_accessible_without_token() { let _env_lock = json_rpc_e2e_env_lock(); ensure_test_rpc_auth(); let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let client = reqwest::Client::new(); let base = format!("http://{rpc_addr}"); // Paths that return 200 without any extra params. for path in ["/", "/health", "/schema", "/events/webhooks"] { let resp = client .get(format!("{base}{path}")) .send() .await .unwrap_or_else(|e| panic!("GET {path}: {e}")); assert!( resp.status().is_success(), "public path {path} must return 2xx without auth, got {}", resp.status() ); } // Paths that bypass auth but return non-2xx for unrelated reasons // (missing required query params, no WebSocket upgrade headers, etc.). // The invariant is that the auth middleware does NOT reject them with 401. for path in ["/auth/telegram", "/events", "/ws/dictation"] { let resp = client .get(format!("{base}{path}")) .send() .await .unwrap_or_else(|e| panic!("GET {path}: {e}")); assert_ne!( resp.status(), StatusCode::UNAUTHORIZED, "public path {path} must not be auth-gated (got {})", resp.status() ); } rpc_join.abort(); } /// Simulate an external process using a guessed token — must be rejected. #[tokio::test] async fn external_process_with_guessed_token_is_rejected() { let _env_lock = json_rpc_e2e_env_lock(); ensure_test_rpc_auth(); // server validates against TEST_RPC_TOKEN let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; let client = reqwest::Client::new(); // An attacker process trying a plausible-looking token that isn't the real one. let attacker_token = "aabbccddeeff00112233445566778899aabbccddeeff00112233445566778899"; assert_ne!( attacker_token, TEST_RPC_TOKEN, "attacker token must differ from real one" ); let resp = client .post(format!("http://{rpc_addr}/rpc")) .header(AUTHORIZATION, format!("Bearer {attacker_token}")) .header("Content-Type", "application/json") .body(r#"{"jsonrpc":"2.0","id":1,"method":"core.ping","params":{}}"#) .send() .await .expect("request"); assert_eq!( resp.status(), 401, "external process with wrong token must be rejected" ); rpc_join.abort(); }