//! Central dispatcher for RPC requests. //! //! This module coordinates the routing of incoming requests to either the //! core subsystem or the OpenHuman domain-specific handlers. use crate::core::legacy_aliases::resolve_legacy; use crate::core::rpc_log; use crate::core::types::{AppState, InvocationResult}; use serde_json::{json, Map, Value}; /// Dispatches an RPC method call to the appropriate subsystem. /// /// This is the primary entry point for all RPC calls. It uses a tiered routing /// strategy: /// 1. **Core Subsystem**: Checks for internal methods like `core.ping` or `core.version`. /// 2. **Domain-Specific Handlers**: Delegates to the `openhuman` domain dispatcher /// which handles all registered controllers (memory, skills, etc.). /// /// # Arguments /// /// * `state` - The current application state (e.g., core version). /// * `method` - The name of the RPC method to invoke (e.g., `core.ping`). /// * `params` - The parameters for the method call as a JSON value. /// /// # Returns /// /// A `Result` containing the JSON-formatted response or an error message if /// the method is unknown or invocation fails. pub async fn dispatch( state: AppState, method: &str, params: serde_json::Value, ) -> Result { log::trace!( "[rpc:dispatch] enter method={} params={}", method, rpc_log::redact_params_for_log(¶ms) ); // Tier 0: Rewrite legacy method names to their canonical form before // any subsystem lookup. Symmetric with the frontend's // `normalizeRpcMethod` (`app/src/services/rpcMethods.ts`): the // frontend rewrites outgoing names for clients that just updated, the // core rewrites incoming names for clients that haven't yet. See // `crate::core::legacy_aliases` for the shared table. let resolved = resolve_legacy(method); if resolved != method { // Per-rewrite log at debug to keep the dispatcher hot path quiet // at scale (per graycyrus review on PR #1544). Aggregate // visibility belongs in the observability layer, not here. log::debug!( "[rpc-legacy-alias] rewrite method={} -> canonical={}", method, resolved ); } let method = resolved; // Tier 1: Internal core methods. // These are handled directly within the core module and don't require // a separate controller registration. if let Some(result) = try_core_dispatch(&state, method, params.clone()) { log::debug!("[rpc:dispatch] routed method={} subsystem=core", method); return result.map(crate::core::types::invocation_to_rpc_json); } // Tier 2: Registered domain controllers. if let Some(result) = try_registry_dispatch(method, params.clone()).await { log::debug!( "[rpc:dispatch] routed method={} subsystem=controller_registry", method ); return result; } // Tier 3: Legacy domain-specific dispatcher. if let Some(result) = crate::rpc::try_dispatch(method, params).await { log::debug!( "[rpc:dispatch] routed method={} subsystem=openhuman", method ); return result; } // Tier 4: unrecognised method. The JSON-RPC response is unchanged — the // caller still receives a method-not-found error. Only the *severity* of // how the transport layer records it differs by class (see // `jsonrpc::rpc_handler`): known non-actionable misses // (`is_known_probe_method`) are debug-only and never reach Sentry // (#3567, #3565), while any other unknown method is downgraded to a // warn-level capture (recorded for triage, no page) instead of an error // event. Log here at debug with the method name so the path stays // diagnosable without re-creating the Sentry noise. if is_known_probe_method(method) { log::debug!( "[rpc] unknown_method method={} class=debug_only_known_miss (not reported to Sentry)", method ); } else { log::debug!( "[rpc] unknown_method method={} class=unrecognized (reported to Sentry at warn for triage)", method ); } Err(format!("{UNKNOWN_METHOD_PREFIX}{method}")) } /// Prefix of the error string returned for an unrecognised RPC method. Kept as /// a shared constant so the emit site (above) and the transport-layer /// classifier ([`unknown_method_name`]) cannot drift apart. pub const UNKNOWN_METHOD_PREFIX: &str = "unknown method: "; /// Known non-actionable unknown method names that are debug-only. /// /// Most entries are generic external probes that are never real RPC methods /// and never will be (issue #3567): `rpc.discover` (JSON-RPC service /// discovery), `list_methods`, liveness `status`, `auth.status`, `config/get`. /// This also covers retired feature calls from older clients when no safe /// canonical handler exists (#3565: `openhuman.memory_tree_create_namespace`). /// /// Each miss previously produced recurring Sentry ERROR events with zero user /// impact. The transport layer keeps these debug-only (never captured). The /// matching health-method *aliases* land separately in `legacy_aliases` /// (#3566), which depends on this severity change. const KNOWN_PROBE_METHODS: &[&str] = &[ "rpc.discover", "list_methods", "status", "auth.status", "config/get", "openhuman.memory_tree_create_namespace", ]; /// Returns `true` when `method` is a known non-actionable unknown method name /// from [`KNOWN_PROBE_METHODS`]. Matched against the *resolved* method name /// (after legacy-alias rewrite), i.e. the name embedded in the /// [`UNKNOWN_METHOD_PREFIX`] error string. pub fn is_known_probe_method(method: &str) -> bool { KNOWN_PROBE_METHODS.contains(&method) } /// Extracts the offending method name from an unknown-method error string, or /// `None` if `message` is not an unknown-method error. The transport layer uses /// this to classify the failure for Sentry severity without re-deriving the /// method from the request (which may differ post legacy-alias rewrite). pub fn unknown_method_name(message: &str) -> Option<&str> { message.strip_prefix(UNKNOWN_METHOD_PREFIX) } /// Handles internal core-level RPC methods. /// /// These methods provide basic information about the server and its version. /// /// Currently supported methods: /// - `core.ping`: A simple liveness check. Returns `{ "ok": true }`. /// - `core.version`: Returns the version of the running core binary. fn try_core_dispatch( state: &AppState, method: &str, params: serde_json::Value, ) -> Option> { match method { "core.ping" => Some(InvocationResult::ok(json!({ "ok": true }))), "core.version" => Some(InvocationResult::ok( json!({ "version": state.core_version }), )), "core.events_subscribe_token" => Some(handle_events_subscribe_token(params)), _ => None, } } /// Mint a single-shot bind token for the SSE `/events` stream. /// /// Browser `EventSource` cannot attach an `Authorization` header, so an /// authenticated holder of the per-process RPC bearer first asks for a /// short-lived token here (this RPC is gated by the same bearer-token /// middleware as the rest of `/rpc`) and then opens /// `/events?client_id=&token=`. The `/events` handler removes /// the token from the store on first use, so a leaked URL cannot be /// replayed by a second subscriber. fn handle_events_subscribe_token(params: serde_json::Value) -> Result { let obj = params.as_object(); let client_id = obj .and_then(|m| m.get("client_id")) .and_then(|v| v.as_str()) .map(|s| s.trim()) .filter(|s| !s.is_empty()) .ok_or_else(|| { log::warn!( "[events-bind] reject mint: missing or empty client_id (param_keys={:?})", obj.map(|m| m.keys().collect::>()) ); "missing or empty 'client_id' parameter".to_string() })?; let ttl = obj .and_then(|m| m.get("ttl_secs")) .and_then(|v| v.as_u64()) .map(std::time::Duration::from_secs); let issued = crate::core::event_bind_tokens::issue(client_id.to_string(), ttl).ok_or_else(|| { log::warn!( "[events-bind] reject mint: store at capacity (client_id_len={} ttl_secs={:?})", client_id.len(), ttl.map(|d| d.as_secs()) ); "events bind-token store at capacity; try again shortly".to_string() })?; let ttl_remaining_secs = issued .valid_until .checked_duration_since(std::time::Instant::now()) .unwrap_or_default() .as_secs(); log::debug!( "[events-bind] minted token for client_id_len={} ttl_secs={}", client_id.len(), ttl_remaining_secs ); InvocationResult::ok(json!({ "token": issued.token, "ttl_secs": ttl_remaining_secs, })) } async fn try_registry_dispatch( method: &str, params: Value, ) -> Option> { let schema = crate::core::all::schema_for_rpc_method(method)?; let params_obj = match params_to_object(params) { Ok(params_obj) => params_obj, Err(err) => return Some(Err(err)), }; if let Err(err) = crate::core::all::validate_params(&schema, ¶ms_obj) { return Some(Err(err)); } crate::core::all::try_invoke_registered_rpc(method, params_obj).await } fn params_to_object(params: Value) -> Result, String> { match params { Value::Object(map) => Ok(map), Value::Null => Ok(Map::new()), other => Err(format!( "invalid params: expected object or null, got {}", type_name(&other) )), } } fn type_name(value: &Value) -> &'static str { match value { Value::Null => "null", Value::Bool(_) => "bool", Value::Number(_) => "number", Value::String(_) => "string", Value::Array(_) => "array", Value::Object(_) => "object", } } #[cfg(test)] mod tests { use super::*; use serde_json::json; fn test_state() -> AppState { AppState { core_version: "9.9.9-test".to_string(), } } #[tokio::test] async fn dispatch_core_ping_returns_ok_true() { let out = dispatch(test_state(), "core.ping", json!({})) .await .expect("core.ping should succeed"); assert_eq!(out, json!({ "ok": true })); } #[tokio::test] async fn dispatch_core_version_returns_state_version() { let out = dispatch(test_state(), "core.version", json!({})) .await .expect("core.version should succeed"); assert_eq!(out, json!({ "version": "9.9.9-test" })); } #[tokio::test] async fn dispatch_core_ignores_params() { // Params must be tolerated even when the method takes none. let out = dispatch(test_state(), "core.ping", json!({ "extra": 1 })) .await .expect("core.ping should ignore extra params"); assert_eq!(out, json!({ "ok": true })); } #[tokio::test] async fn dispatch_rewrites_legacy_alias_before_lookup() { // `openhuman.ping` is a legacy alias for `core.ping` in the shared // alias table. Going through the dispatcher must rewrite it and // route successfully to Tier 1 instead of falling through to the // unknown-method error path. let out = dispatch(test_state(), "openhuman.ping", json!({})) .await .expect("legacy alias openhuman.ping must resolve to core.ping"); assert_eq!(out, json!({ "ok": true })); } #[tokio::test] async fn dispatch_unknown_method_returns_error() { let err = dispatch(test_state(), "does.not.exist", json!({})) .await .expect_err("unknown methods must error"); assert!(err.contains("unknown method")); assert!(err.contains("does.not.exist")); } #[tokio::test] async fn dispatch_empty_method_returns_unknown_method_error() { let err = dispatch(test_state(), "", json!({})) .await .expect_err("empty method must error"); assert!(err.contains("unknown method")); } #[tokio::test] async fn dispatch_delegates_to_tier2_for_domain_method() { // Tier 2 dispatcher handles `openhuman.security_policy_info`, so // it must succeed and return a policy object. let out = dispatch(test_state(), "openhuman.security_policy_info", json!({})) .await .expect("security_policy_info should route via tier 2"); // With logs present, payload is wrapped as { result, logs }. assert!(out.get("result").is_some() || out.get("autonomy").is_some()); } #[test] fn try_core_dispatch_returns_none_for_non_core_namespace() { let state = test_state(); assert!(try_core_dispatch(&state, "openhuman.memory_list_namespaces", json!({})).is_none()); assert!(try_core_dispatch(&state, "corez.ping", json!({})).is_none()); } #[test] fn try_core_dispatch_matches_exact_ping_and_version() { let state = test_state(); assert!(try_core_dispatch(&state, "core.ping", json!({})).is_some()); assert!(try_core_dispatch(&state, "core.version", json!({})).is_some()); // Prefix match alone must not count. assert!(try_core_dispatch(&state, "core.pingz", json!({})).is_none()); assert!(try_core_dispatch(&state, "core", json!({})).is_none()); } #[test] fn try_core_dispatch_version_reflects_appstate() { let state = AppState { core_version: "0.0.0-abc".into(), }; let result = try_core_dispatch(&state, "core.version", json!({})) .expect("core.version must be routed") .expect("core.version must produce InvocationResult"); assert_eq!(result.value, json!({ "version": "0.0.0-abc" })); assert!(result.logs.is_empty()); } #[tokio::test] async fn dispatch_legacy_ping_rewrites_and_succeeds() { let out = dispatch(test_state(), "openhuman.ping", json!({})) .await .expect("openhuman.ping should be rewritten to core.ping and succeed"); assert_eq!(out, json!({ "ok": true })); } #[test] fn is_known_probe_method_matches_allow_list_exactly() { // Every allow-listed probe / legacy health name is recognised. for m in [ "rpc.discover", "list_methods", "status", "auth.status", "config/get", "openhuman.memory_tree_create_namespace", ] { assert!( is_known_probe_method(m), "{m} should be a debug-only known miss" ); } // Genuinely-unknown methods and near-misses are NOT allow-listed, so // they stay on the warn-for-triage path rather than being silenced. assert!(!is_known_probe_method("does.not.exist")); assert!(!is_known_probe_method("core.not_a_real_method")); assert!(!is_known_probe_method("Status")); // case-sensitive assert!(!is_known_probe_method("rpc.discover.extra")); // exact match only assert!(!is_known_probe_method("memory_tree_create_namespace")); assert!(!is_known_probe_method("")); } #[tokio::test] async fn dispatch_dotted_channel_list_aliases_route_to_registry() { for method in ["channels.list", "openhuman.channels.list"] { let out = dispatch(test_state(), method, json!({})) .await .unwrap_or_else(|err| panic!("{method} should route via channels_list: {err}")); assert!( out.is_array() || out.get("result").is_some(), "expected {method} to return the channels_list payload, got {out}" ); } } #[test] fn unknown_method_name_extracts_from_error_string_only() { // The classifier round-trips the exact string `dispatch` emits. let err = format!("{UNKNOWN_METHOD_PREFIX}rpc.discover"); assert_eq!(unknown_method_name(&err), Some("rpc.discover")); // Unrelated error strings are not misclassified as unknown-method. assert_eq!(unknown_method_name("unknown param 'x' for ns.fn"), None); assert_eq!(unknown_method_name("Session expired"), None); } #[tokio::test] async fn dispatch_probe_method_still_returns_unknown_method_error() { // Allow-listed probe names must not be silently "handled" — the caller // still gets a method-not-found error. Only the Sentry severity (in the // transport layer) changes; the dispatch contract is unchanged. let err = dispatch(test_state(), "rpc.discover", json!({})) .await .expect_err("probe methods are still unknown to the dispatcher"); assert_eq!(unknown_method_name(&err), Some("rpc.discover")); assert!(is_known_probe_method( unknown_method_name(&err).expect("unknown-method error") )); } #[tokio::test] async fn dispatch_legacy_alias_routes_to_registry() { // openhuman.get_analytics_settings should rewrite to openhuman.config_get_analytics_settings. // This is a read-only call and should succeed if the registry is wired up. let out = dispatch(test_state(), "openhuman.get_analytics_settings", json!({})) .await .expect("openhuman.get_analytics_settings should be rewritten and succeed"); // The registry-wrapped payload has a "result" field. assert!( out.get("enabled").is_some() || out.get("result").is_some(), "Payload should have 'enabled' or 'result', got: {}", out ); } }