diff --git a/AGENTS.md b/AGENTS.md index 4fffaea2cd..756359bdab 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -209,6 +209,13 @@ Embedded provider webviews **must not** grow new JS injection. No new `.js` unde Modules: `all`, `auth`, `cli`, `dispatch`, `event_bus/`, `jsonrpc`, `logging`, `observability`, `types`, etc. No business logic here. +### Runtime composition — `ServiceSet` + `DomainSet` on `CoreBuilder` + +Two independent runtime axes on `CoreBuilder` (`src/core/runtime/builder.rs`): + +- **`ServiceSet`** selects which *background services / transports* run (`rpc_http`, `socketio`, `cron`, `channels`, `heartbeat`, …). Presets: `desktop()` / `headless_api()` / `none()`. +- **`DomainSet`** selects which *domain families* exist at runtime, one flag per `DomainGroup` (`src/core/all.rs`). Presets: `full()` (default — byte-identical to before #4796), `harness()` (agent + memory + threads + config + security only), `none()`. Every controller is tagged with its `DomainGroup` at the single registration site in `src/core/all.rs`; the live surface (controllers/`/schema`/dispatch, agent tools, stores, subscribers) is filtered by the ambient `CoreContext::domains()`. A gated domain's controllers become unknown-method, its agent tools absent, its stores/subscribers uninitialized. `examples/embed_headless.rs` uses `DomainSet::harness()`. Per-gate Cargo `[features]` (children #4797–#4804) narrow the compile-time surface further; `DomainSet` is the runtime axis they compose with. + ### Event bus (`src/core/event_bus/`) Typed pub/sub + native request/response. Both singletons — use module-level functions. diff --git a/docs/plans/pluggable-core/phase-2-corecontext.md b/docs/plans/pluggable-core/phase-2-corecontext.md index 77182c7462..d7e23130d0 100644 --- a/docs/plans/pluggable-core/phase-2-corecontext.md +++ b/docs/plans/pluggable-core/phase-2-corecontext.md @@ -116,6 +116,32 @@ pub enum StorageBackend { WorkspaceFs } // CoreBuilder::storage(..); only impl | Store trait too narrow, churns later | extract traits from _observed_ handler usage per domain, not speculatively | | Coverage gate on mechanical diffs | signature/registry PRs carry existing tests; per-domain PRs add store-trait unit tests | +## 2.d `DomainSet` — the runtime composition axis (#4796) + +`DomainSet` (in `src/core/runtime/builder.rs`, sibling of `ServiceSet`) is the +**runtime** axis that selects which domain *families* exist, complementing +`ServiceSet` (which selects background services). One flag per `DomainGroup` +(`src/core/all.rs`); presets `full()` (default — byte-identical to pre-#4796), +`harness()` (agent + memory + threads + config + security only), `none()`. + +Mechanism (Shape B — filter seam, **not** the full per-domain +`DomainRegistration` struct collapse, which remains phase-2-deferred): + +- Every controller is tagged with its `DomainGroup` once, at the single + registration site (`build_registered_controllers` / + `build_internal_only_controllers`), via a `GroupedController { group, + controller }` wrapper — the ~109 domain modules keep returning bare + `RegisteredController` lists and never learn about groups. +- The live surface filters by the **ambient** `CoreContext::domains()`: + `all_registered_controllers` / `all_controller_schemas` (so `/schema` omits + gated namespaces), `try_invoke_registered_rpc` (gated method ⇒ `None`, i.e. + unknown-method), the agent tool surface (`tools::ops::all_tools_with_runtime` + post-filter by `tool_group`), workspace-bound store init (`init_stores`), and + domain event subscribers (`register_domain_subscribers`). +- No active context (pre-boot unit tests) ⇒ no filtering (treated as full). +- Per-gate Cargo `[features]` (children #4797–#4804) narrow the *compile-time* + surface further; `DomainSet` is the runtime axis they compose with. + ## Verification - `pnpm test:rust` + `json_rpc_e2e` green after each sub-series. @@ -124,3 +150,7 @@ pub enum StorageBackend { WorkspaceFs } // CoreBuilder::storage(..); only impl - `all::try_invoke_registered_rpc` installs an ambient `CoreContext::scope` around registered handler futures, and tests verify `CoreContext::current()` propagation through registered RPC invocation. +- `DomainSet`: `full_registration_is_byte_identical`, + `harness_excludes_gated_namespaces`, `dispatch_returns_none_for_gated_method`, + `group_mapping_smoke` (`src/core/all_tests.rs`) + + `domain_set_presets_have_expected_flags` (`builder.rs`). diff --git a/examples/embed_headless.rs b/examples/embed_headless.rs index b960d9856f..d40a334994 100644 --- a/examples/embed_headless.rs +++ b/examples/embed_headless.rs @@ -1,9 +1,14 @@ //! Embed the OpenHuman core as a library — no HTTP, no background services. //! -//! Demonstrates the Phase-1 pluggable-core API: build a fully-initialized core -//! with [`ServiceSet::none`] (no ports bound, no cron/channels/heartbeat) and -//! dispatch RPC methods in-process through [`CoreRuntime::invoke`] — the exact -//! same path the HTTP `/rpc` handler and the CLI use. +//! Demonstrates the pluggable-core API: build a fully-initialized core with +//! [`ServiceSet::none`] (no ports bound, no cron/channels/heartbeat) AND +//! [`DomainSet::harness`] (only the agent + memory + threads + config + security +//! domain families are live — the gate families flows/skills/mcp/meet/channels/ +//! web3/voice/media and the catch-all `platform` are off, so their controllers +//! are unknown-method, their agent tools absent, and their stores/subscribers +//! never initialize). Dispatch RPC methods in-process through +//! [`CoreRuntime::invoke`] — the exact same path the HTTP `/rpc` handler and the +//! CLI use. //! //! Run with: //! @@ -16,9 +21,10 @@ //! `CancellationToken`, and call `runtime.serve(None, Some(token)).await` — it //! binds `127.0.0.1:7788` (override with `.host(..)` / `.port(..)` on the //! builder, or `OPENHUMAN_CORE_HOST` / `OPENHUMAN_CORE_PORT`) and serves until -//! the token is cancelled. +//! the token is cancelled. Widen the runtime surface by swapping +//! `DomainSet::harness()` for `DomainSet::full()`. -use openhuman_core::{CoreBuilder, HostKind, ServiceSet}; +use openhuman_core::{CoreBuilder, DomainSet, HostKind, ServiceSet}; #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -27,14 +33,20 @@ async fn main() -> anyhow::Result<()> { let _ = env_logger::builder().is_test(false).try_init(); // Initialize the core against the local workspace. `HostKind::Cli` selects - // the standalone (non-desktop) bootstrap path; `ServiceSet::none()` means no - // transport and no background services are started. + // the standalone (non-desktop) bootstrap path; `DomainSet::harness()` builds + // the embeddable agent core; `ServiceSet::none()` means no transport and no + // background services are started. let runtime = CoreBuilder::new(HostKind::Cli) + .domains(DomainSet::harness()) .services(ServiceSet::none()) .build() .await?; // Dispatch a couple of RPC methods in-process — no network involved. + // `core.version` and `openhuman.ping` (a legacy alias for the built-in + // `core.ping`) are always available regardless of the DomainSet — they are + // transport built-ins, not domain controllers — so they succeed even under + // `harness()`. let version = runtime .invoke("core.version", serde_json::json!({})) .await diff --git a/src/core/all.rs b/src/core/all.rs index 46b17241ce..06a44d060e 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -46,13 +46,91 @@ impl RegisteredController { } } +/// Coarse-grained domain *family* a controller belongs to, used to gate its live +/// surface by the ambient [`crate::core::runtime::DomainSet`] (#4796). +/// +/// Every registered controller is tagged with exactly one group at its single +/// registration site ([`build_registered_controllers`] / +/// [`build_internal_only_controllers`]); the live surface (schema dump, +/// dispatch, agent tools, stores, subscribers) filters by whether the active +/// [`crate::core::runtime::context::CoreContext`]'s `DomainSet` allows that +/// group. `full()` allows every group ⇒ registration is byte-identical to +/// pre-#4796. When no context is active (unit tests before boot) filtering is +/// disabled (treated as full). +/// +/// The harness families (`Agent`/`Memory`/`Threads`/`Config`/`Security`) are on +/// under [`crate::core::runtime::DomainSet::harness`]; the gate families +/// (`Flows`/`Skills`/`Mcp`/`Meet`/`Channels`/`Web3`/`Voice`/`Media`) are the +/// per-feature axes the child issues (#4797–#4804) additionally narrow at +/// compile time. `Platform` is the catch-all for everything not in a named +/// family — always on in `full()`, off in `harness()`/`none()`. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum DomainGroup { + // Harness families — on under `DomainSet::harness()`. + Agent, + Memory, + Threads, + Config, + Security, + // Gate families — off under `harness()`; per-gate Cargo features (#4797–#4804) + // narrow these further at compile time. + Flows, + Skills, + Mcp, + Meet, + Channels, + Web3, + Voice, + Media, + // Everything not in a named family — always on in `full()`, off otherwise. + Platform, +} + +/// A [`RegisteredController`] tagged with the [`DomainGroup`] it belongs to. +/// +/// The registry stores these so the live surface can be filtered by the ambient +/// [`crate::core::runtime::DomainSet`] WITHOUT touching the ~109 domain modules' +/// bare `RegisteredController { schema, handler }` literals — the group tag is +/// attached once, here, at the single registration site. +#[derive(Clone)] +struct GroupedController { + group: DomainGroup, + controller: RegisteredController, +} + +/// Append `items` to `dst`, tagging each with `group`. This is the single seam +/// that attaches a [`DomainGroup`] to every domain's controllers without the +/// domain modules knowing about groups. +fn push(dst: &mut Vec, group: DomainGroup, items: Vec) { + dst.extend( + items + .into_iter() + .map(|controller| GroupedController { group, controller }), + ); +} + +/// The [`DomainSet`](crate::core::runtime::DomainSet) of the ambient dispatch +/// context, if any. `None` before any [`crate::core::runtime::context::CoreContext`] +/// is built (some unit tests) ⇒ callers treat that as "no filtering" (full). +fn active_domain_set() -> Option { + crate::core::runtime::context::CoreContext::current().map(|c| c.domains()) +} + +/// Whether the given [`DomainGroup`] is enabled under the ambient +/// [`DomainSet`](crate::core::runtime::DomainSet). No active context ⇒ `true` +/// (full, no filter) so pre-boot unit tests and non-context callers see every +/// domain, exactly as before #4796. +fn group_allowed(group: DomainGroup) -> bool { + active_domain_set().is_none_or(|s| s.allows(group)) +} + /// The global static registry of all controllers, initialized once on first access. -static REGISTRY: OnceLock> = OnceLock::new(); +static REGISTRY: OnceLock> = OnceLock::new(); /// Internal-only controllers: registered for RPC dispatch but NOT in the agent-facing /// schema catalog. These handlers are callable by trusted callers (e.g. the Tauri scanner) /// but should not be advertised to agents via tool listings or schema discovery. -static INTERNAL_REGISTRY: OnceLock> = OnceLock::new(); +static INTERNAL_REGISTRY: OnceLock> = OnceLock::new(); /// The global static registry of standalone CLI adapters. static CLI_ADAPTERS: OnceLock> = OnceLock::new(); @@ -61,10 +139,13 @@ static CLI_ADAPTERS: OnceLock> = OnceLock::new(); /// /// This function initializes the registry if it hasn't been already, /// performing validation to ensure no duplicates or missing handlers exist. -fn registry() -> &'static [RegisteredController] { +fn registry() -> &'static [GroupedController] { REGISTRY .get_or_init(|| { let registered = build_registered_controllers(); + // Drift guard runs once on the FULL set — validation is independent + // of any ambient DomainSet filter (which only affects the live + // surface, never registry integrity). validate_registry(®istered).unwrap_or_else(|err| { panic!("invalid controller registry: {err}"); }); @@ -77,7 +158,7 @@ fn registry() -> &'static [RegisteredController] { /// /// These controllers are callable over RPC but are NOT included in agent tool listings /// or schema discovery endpoints. -fn internal_registry() -> &'static [RegisteredController] { +fn internal_registry() -> &'static [GroupedController] { INTERNAL_REGISTRY .get_or_init(build_internal_only_controllers) .as_slice() @@ -101,242 +182,609 @@ fn cli_adapters() -> &'static [RegisteredCliAdapter] { /// /// When adding a new domain/namespace, its `all_*_registered_controllers()` /// function must be called here to make it available via RPC and CLI. -fn build_registered_controllers() -> Vec { +fn build_registered_controllers() -> Vec { let mut controllers = Vec::new(); // Application information and capabilities - controllers.extend(crate::openhuman::about_app::all_about_app_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::about_app::all_about_app_registered_controllers(), + ); // AgentBox marketplace adapter status - controllers.extend(crate::openhuman::agentbox::all_agentbox_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::agentbox::all_agentbox_registered_controllers(), + ); // Core application shell state - controllers.extend(crate::openhuman::app_state::all_app_state_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::app_state::all_app_state_registered_controllers(), + ); // Audio generation + podcast-style email delivery - controllers.extend(crate::openhuman::audio_toolkit::all_audio_toolkit_registered_controllers()); + push( + &mut controllers, + DomainGroup::Voice, + crate::openhuman::audio_toolkit::all_audio_toolkit_registered_controllers(), + ); // Composio integration controllers - controllers.extend(crate::openhuman::composio::all_composio_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::composio::all_composio_registered_controllers(), + ); // Recall.ai Calendar V1 (backend-proxied) controllers - controllers - .extend(crate::openhuman::recall_calendar::all_recall_calendar_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::recall_calendar::all_recall_calendar_registered_controllers(), + ); // Scheduled job management - controllers.extend(crate::openhuman::cron::all_cron_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::cron::all_cron_registered_controllers(), + ); // Saved automation workflows (tinyflows graphs): create/get/list/update/delete/run - controllers.extend(crate::openhuman::flows::all_flows_registered_controllers()); + push( + &mut controllers, + DomainGroup::Flows, + crate::openhuman::flows::all_flows_registered_controllers(), + ); // Proactive task ingestion from external tools (github/notion/linear/clickup) - controllers.extend(crate::openhuman::task_sources::all_task_sources_registered_controllers()); - controllers.extend(crate::openhuman::dashboard::all_dashboard_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::task_sources::all_task_sources_registered_controllers(), + ); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::dashboard::all_dashboard_registered_controllers(), + ); // MCP client subsystem: Smithery registry browser, local server install/connect, tool dispatch - controllers.extend(crate::openhuman::mcp_registry::all_mcp_registry_registered_controllers()); + push( + &mut controllers, + DomainGroup::Mcp, + crate::openhuman::mcp_registry::all_mcp_registry_registered_controllers(), + ); // Webview APIs bridge — proxies connector calls (Gmail, …) through // a WebSocket to the Tauri shell so curl reaches the live webview. - controllers.extend(crate::openhuman::webview_apis::all_webview_apis_registered_controllers()); + push( + &mut controllers, + DomainGroup::Channels, + crate::openhuman::webview_apis::all_webview_apis_registered_controllers(), + ); // Agent definition and prompt inspection - controllers.extend(crate::openhuman::agent::all_agent_registered_controllers()); + push( + &mut controllers, + DomainGroup::Agent, + crate::openhuman::agent::all_agent_registered_controllers(), + ); // Read-only agent run replay + status over the durable journal/status seams // (agent_run_events / agent_run_status / agent_runs_active). - controllers - .extend(crate::openhuman::tinyagents::replay::all_agent_replay_registered_controllers()); + push( + &mut controllers, + DomainGroup::Agent, + crate::openhuman::tinyagents::replay::all_agent_replay_registered_controllers(), + ); // Persistent agent profiles (flavours): name, soul, memory sources, skills, MCP, connectors. - controllers.extend(crate::openhuman::profiles::all_profiles_registered_controllers()); + push( + &mut controllers, + DomainGroup::Agent, + crate::openhuman::profiles::all_profiles_registered_controllers(), + ); // User-facing agent registry: defaults, enablement, custom agents, tool policy. - controllers - .extend(crate::openhuman::agent_registry::all_agent_registry_registered_controllers()); + push( + &mut controllers, + DomainGroup::Agent, + crate::openhuman::agent_registry::all_agent_registry_registered_controllers(), + ); // Local procedural operating experience for agent self-learning - controllers - .extend(crate::openhuman::agent_experience::all_agent_experience_registered_controllers()); + push( + &mut controllers, + DomainGroup::Agent, + crate::openhuman::agent_experience::all_agent_experience_registered_controllers(), + ); // System and process health monitoring - controllers.extend(crate::openhuman::health::all_health_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::health::all_health_registered_controllers(), + ); // One-time first-run initialization (Python/spaCy/Node provisioning) - controllers.extend(crate::openhuman::harness_init::all_harness_init_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::harness_init::all_harness_init_registered_controllers(), + ); // Diagnostic tools - controllers.extend(crate::openhuman::doctor::all_doctor_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::doctor::all_doctor_registered_controllers(), + ); // Secret storage and encryption - controllers.extend(crate::openhuman::encryption::all_encryption_registered_controllers()); + push( + &mut controllers, + DomainGroup::Security, + crate::openhuman::encryption::all_encryption_registered_controllers(), + ); // Keyring consent — user approval before local secret storage fallback - controllers - .extend(crate::openhuman::keyring_consent::all_keyring_consent_registered_controllers()); + push( + &mut controllers, + DomainGroup::Security, + crate::openhuman::keyring_consent::all_keyring_consent_registered_controllers(), + ); // Security policy metadata - controllers.extend(crate::openhuman::security::all_security_registered_controllers()); + push( + &mut controllers, + DomainGroup::Security, + crate::openhuman::security::all_security_registered_controllers(), + ); // Interactive approval workflow (#1339 — gate external-effect tool calls) - controllers.extend(crate::openhuman::approval::all_approval_registered_controllers()); + push( + &mut controllers, + DomainGroup::Security, + crate::openhuman::approval::all_approval_registered_controllers(), + ); // Interactive plan-review gate — parks a live turn on a thread-scoped plan - controllers.extend(crate::openhuman::plan_review::all_plan_review_registered_controllers()); + push( + &mut controllers, + DomainGroup::Security, + crate::openhuman::plan_review::all_plan_review_registered_controllers(), + ); // Agent-generated artifact storage, retrieval, and lifecycle management - controllers.extend(crate::openhuman::artifacts::all_artifacts_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::artifacts::all_artifacts_registered_controllers(), + ); // Background heartbeat loop controls - controllers.extend(crate::openhuman::heartbeat::all_heartbeat_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::heartbeat::all_heartbeat_registered_controllers(), + ); // Ad-hoc static directory HTTP hosting for local file sharing / previews - controllers.extend(crate::openhuman::http_host::all_http_host_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::http_host::all_http_host_registered_controllers(), + ); // Token usage and billing cost tracking - controllers.extend(crate::openhuman::cost::all_cost_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::cost::all_cost_registered_controllers(), + ); // x402 machine-payable API payment protocol - controllers.extend(crate::openhuman::x402::all_x402_registered_controllers()); + push( + &mut controllers, + DomainGroup::Web3, + crate::openhuman::x402::all_x402_registered_controllers(), + ); // Inline autocomplete settings - controllers.extend(crate::openhuman::autocomplete::all_autocomplete_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::autocomplete::all_autocomplete_registered_controllers(), + ); // External messaging channels (Web, Telegram, etc.) - controllers.extend( + push( + &mut controllers, + DomainGroup::Channels, crate::openhuman::channels::providers::web::all_web_channel_registered_controllers(), ); - controllers - .extend(crate::openhuman::channels::controllers::all_channels_registered_controllers()); + push( + &mut controllers, + DomainGroup::Channels, + crate::openhuman::channels::controllers::all_channels_registered_controllers(), + ); // Persistent configuration management - controllers.extend(crate::openhuman::config::all_config_registered_controllers()); + push( + &mut controllers, + DomainGroup::Config, + crate::openhuman::config::all_config_registered_controllers(), + ); // Local sidecar reachability + backend Socket.IO state diagnostics (#1527) - controllers.extend(crate::openhuman::connectivity::all_connectivity_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::connectivity::all_connectivity_registered_controllers(), + ); // User credentials and session management - controllers.extend(crate::openhuman::credentials::all_credentials_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::credentials::all_credentials_registered_controllers(), + ); // Desktop service management - controllers.extend(crate::openhuman::service::all_service_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::service::all_service_registered_controllers(), + ); // Data migration utilities - controllers.extend(crate::openhuman::migration::all_migration_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::migration::all_migration_registered_controllers(), + ); // Saved council definitions for the desktop Model Council surface. - controllers - .extend(crate::openhuman::council_registry::all_council_registry_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::council_registry::all_council_registry_registered_controllers(), + ); // Model Council: multi-model deliberation (parallel members + chair synthesis) - controllers.extend(crate::openhuman::model_council::all_model_council_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::model_council::all_model_council_registered_controllers(), + ); // Background command monitors for agent-scoped event sources - controllers.extend(crate::openhuman::monitor::all_monitor_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::monitor::all_monitor_registered_controllers(), + ); // Unified inference domain: text / vision / local runtime / cloud providers. // (Formerly split across inference, local AI, and providers modules.) - controllers.extend(crate::openhuman::inference::all_inference_registered_controllers()); - controllers.extend(crate::openhuman::inference::all_local_inference_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::inference::all_inference_registered_controllers(), + ); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::inference::all_local_inference_registered_controllers(), + ); // Embedding provider configuration and embed RPC. - controllers.extend(crate::openhuman::embeddings::all_embeddings_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::embeddings::all_embeddings_registered_controllers(), + ); // People resolution and interaction scoring - controllers.extend(crate::openhuman::people::all_people_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::people::all_people_registered_controllers(), + ); // Screen capture and UI analysis - controllers.extend( + push( + &mut controllers, + DomainGroup::Platform, crate::openhuman::screen_intelligence::all_screen_intelligence_registered_controllers(), ); // Sandbox execution backends (Docker, local jail, policy, cleanup) - controllers.extend(crate::openhuman::sandbox::all_sandbox_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::sandbox::all_sandbox_registered_controllers(), + ); // Backend Socket.IO bridge + related runtime plumbing - controllers.extend(crate::openhuman::socket::all_socket_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::socket::all_socket_registered_controllers(), + ); // Managed Node.js runtime bridge (tool listing + dispatch) - controllers.extend(crate::openhuman::javascript::all_javascript_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::javascript::all_javascript_registered_controllers(), + ); // Discovered SKILL.md skills and their bundled resources - controllers.extend(crate::openhuman::skills::all_skills_registered_controllers()); + push( + &mut controllers, + DomainGroup::Skills, + crate::openhuman::skills::all_skills_registered_controllers(), + ); // Skill runtime: run/cancel/log skill executions and resolve Node/Python toolchains - controllers.extend(crate::openhuman::skill_runtime::all_skill_runtime_registered_controllers()); + push( + &mut controllers, + DomainGroup::Skills, + crate::openhuman::skill_runtime::all_skill_runtime_registered_controllers(), + ); // Skill registry: browse, search, install from remote registries - controllers - .extend(crate::openhuman::skill_registry::all_skill_registry_registered_controllers()); + push( + &mut controllers, + DomainGroup::Skills, + crate::openhuman::skill_registry::all_skill_registry_registered_controllers(), + ); // User workspace and file management - controllers.extend(crate::openhuman::workspace::all_workspace_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::workspace::all_workspace_registered_controllers(), + ); // Workflow tool registry - controllers.extend(crate::openhuman::tools::all_tools_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::tools::all_tools_registered_controllers(), + ); // Unified read-only registry across MCP stdio tools and controller-backed tools - controllers.extend(crate::openhuman::tool_registry::all_tool_registry_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::tool_registry::all_tool_registry_registered_controllers(), + ); // Document and knowledge graph storage - controllers.extend(crate::openhuman::memory::all_memory_registered_controllers()); + push( + &mut controllers, + DomainGroup::Memory, + crate::openhuman::memory::all_memory_registered_controllers(), + ); // Long-term goals list (editable list + turn-based enrichment agent) - controllers.extend(crate::openhuman::memory_goals::all_memory_goals_registered_controllers()); + push( + &mut controllers, + DomainGroup::Memory, + crate::openhuman::memory_goals::all_memory_goals_registered_controllers(), + ); // Thread-level goal (Codex-style per-thread completion contract) - controllers.extend(crate::openhuman::thread_goals::all_thread_goals_registered_controllers()); + push( + &mut controllers, + DomainGroup::Threads, + crate::openhuman::thread_goals::all_thread_goals_registered_controllers(), + ); // Memory tree ingestion layer (#707 — canonicalised chunks with provenance) - controllers.extend(crate::openhuman::memory_tree::all_memory_tree_registered_controllers()); + push( + &mut controllers, + DomainGroup::Memory, + crate::openhuman::memory_tree::all_memory_tree_registered_controllers(), + ); // Memory tree retrieval layer (#710 — LLM-callable read tools over the tree) - controllers.extend(crate::openhuman::memory_tree::all_retrieval_registered_controllers()); + push( + &mut controllers, + DomainGroup::Memory, + crate::openhuman::memory_tree::all_retrieval_registered_controllers(), + ); // Slack → memory-tree ingestion engine (per-message ingest, no bucketing) - controllers.extend( + push( + &mut controllers, + DomainGroup::Memory, crate::openhuman::composio::providers::slack::all_slack_memory_registered_controllers(), ); // Per-connection memory sync status, controls, and progress (#1136) - controllers.extend( + push( + &mut controllers, + DomainGroup::Memory, crate::openhuman::memory_sync::sync_status::all_memory_sync_status_registered_controllers(), ); // Memory sources — user-configured data connectors registry - controllers - .extend(crate::openhuman::memory_sources::all_memory_sources_registered_controllers()); + push( + &mut controllers, + DomainGroup::Memory, + crate::openhuman::memory_sources::all_memory_sources_registered_controllers(), + ); // Memory diff — snapshot-based change tracking for memory sources - controllers.extend(crate::openhuman::memory_diff::all_memory_diff_registered_controllers()); + push( + &mut controllers, + DomainGroup::Memory, + crate::openhuman::memory_diff::all_memory_diff_registered_controllers(), + ); // Link shortener for long tracking URLs — saves LLM tokens - controllers - .extend(crate::openhuman::redirect_links::all_redirect_links_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::redirect_links::all_redirect_links_registered_controllers(), + ); // Referral and growth tracking - controllers.extend(crate::openhuman::referral::all_referral_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::referral::all_referral_registered_controllers(), + ); // Billing and subscription management - controllers.extend(crate::openhuman::billing::all_billing_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::billing::all_billing_registered_controllers(), + ); // Announcements surfaced on harness init - controllers.extend(crate::openhuman::announcements::all_announcements_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::announcements::all_announcements_registered_controllers(), + ); // Team and role management - controllers.extend(crate::openhuman::team::all_team_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::team::all_team_registered_controllers(), + ); // E2E test support — `openhuman.test_reset` wipes sidecar state in-place. // Gated behind the `e2e-test-support` cargo feature so shipped binaries // never even register the destructive wipe RPC. Flipped on by the E2E // build script (app/scripts/e2e-build.sh). #[cfg(feature = "e2e-test-support")] - controllers.extend(crate::openhuman::test_support::all_test_support_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::test_support::all_test_support_registered_controllers(), + ); // Local wallet metadata and onboarding status - controllers.extend(crate::openhuman::wallet::all_wallet_registered_controllers()); + push( + &mut controllers, + DomainGroup::Web3, + crate::openhuman::wallet::all_wallet_registered_controllers(), + ); // High-level web3 surface (swaps / bridges / dapp calls) over the wallet - controllers.extend(crate::openhuman::web3::all_web3_registered_controllers()); + push( + &mut controllers, + DomainGroup::Web3, + crate::openhuman::web3::all_web3_registered_controllers(), + ); // Local assistive surfaces over third-party provider apps - controllers.extend( + push( + &mut controllers, + DomainGroup::Platform, crate::openhuman::provider_surfaces::all_provider_surfaces_registered_controllers(), ); // OS-level text input interactions - controllers.extend(crate::openhuman::text_input::all_text_input_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::text_input::all_text_input_registered_controllers(), + ); // Voice transcription and synthesis - controllers.extend(crate::openhuman::voice::all_voice_registered_controllers()); + push( + &mut controllers, + DomainGroup::Voice, + crate::openhuman::voice::all_voice_registered_controllers(), + ); // Background awareness and autonomous tasks - controllers.extend(crate::openhuman::subconscious::all_subconscious_registered_controllers()); - controllers.extend( + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::subconscious::all_subconscious_registered_controllers(), + ); + push( + &mut controllers, + DomainGroup::Platform, crate::openhuman::subconscious_triggers::all_subconscious_triggers_registered_controllers(), ); // Webhook tunnel management - controllers.extend(crate::openhuman::webhooks::all_webhooks_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::webhooks::all_webhooks_registered_controllers(), + ); // Core binary update management - controllers.extend(crate::openhuman::update::all_update_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::update::all_update_registered_controllers(), + ); // Hierarchical knowledge summarization - controllers.extend(crate::openhuman::memory_tree::all_tree_summarizer_registered_controllers()); + push( + &mut controllers, + DomainGroup::Memory, + crate::openhuman::memory_tree::all_tree_summarizer_registered_controllers(), + ); // Self-learning and user context enrichment - controllers.extend(crate::openhuman::learning::all_learning_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::learning::all_learning_registered_controllers(), + ); // Conversation thread and message management - controllers.extend(crate::openhuman::threads::all_threads_registered_controllers()); - // TokenJuice content-router debug controllers (detect / compress / cache_stats / retrieve) - controllers.extend(crate::openhuman::tokenjuice::all_tokenjuice_registered_controllers()); + push( + &mut controllers, + DomainGroup::Threads, + crate::openhuman::threads::all_threads_registered_controllers(), + ); + // TokenJuice content-router debug controllers (detect / compress / cache_stats / retrieve). + // Classified Platform (always-on): TokenJuice is the token-compression content + // router that runs on every agent tool output, not a crypto surface — despite + // #4802 listing it under the web3 gate. Flagged for #4802 re-scope. + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::tokenjuice::all_tokenjuice_registered_controllers(), + ); // Per-thread todo list (agent task board CRUD over RPC) - controllers.extend(crate::openhuman::todos::all_todos_registered_controllers()); + push( + &mut controllers, + DomainGroup::Threads, + crate::openhuman::todos::all_todos_registered_controllers(), + ); // Embedded webview native notifications - controllers.extend( + push( + &mut controllers, + DomainGroup::Channels, crate::openhuman::webview_notifications::all_webview_notifications_registered_controllers(), ); // Integration notification ingest, triage, and per-provider settings - controllers.extend(crate::openhuman::notifications::all_notifications_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::notifications::all_notifications_registered_controllers(), + ); // Google Meet call-join request validation (shell handles the webview) - controllers.extend(crate::openhuman::meet::all_meet_registered_controllers()); + push( + &mut controllers, + DomainGroup::Meet, + crate::openhuman::meet::all_meet_registered_controllers(), + ); // Agent meetings — backend-delegated Meet bot via Socket.IO - controllers - .extend(crate::openhuman::agent_meetings::all_agent_meetings_registered_controllers()); + push( + &mut controllers, + DomainGroup::Meet, + crate::openhuman::agent_meetings::all_agent_meetings_registered_controllers(), + ); // Live meet-agent loop: STT/LLM/TTS over the open call's audio. - controllers.extend(crate::openhuman::meet_agent::all_meet_agent_registered_controllers()); + push( + &mut controllers, + DomainGroup::Meet, + crate::openhuman::meet_agent::all_meet_agent_registered_controllers(), + ); // Desktop companion — Clicky-style interaction loop. - controllers.extend( + push( + &mut controllers, + DomainGroup::Platform, crate::openhuman::desktop_companion::all_desktop_companion_registered_controllers(), ); // Structured WhatsApp Web data — agent-facing read-only controllers (list/search). // The write-path ingest controller is registered separately in build_internal_only_controllers. - controllers.extend(crate::openhuman::whatsapp_data::all_whatsapp_data_registered_controllers()); + // Classified Channels (WhatsApp Web messaging surface) — not enumerated in the + // spec Platform list; grouped with the other channel/webview domains. + push( + &mut controllers, + DomainGroup::Channels, + crate::openhuman::whatsapp_data::all_whatsapp_data_registered_controllers(), + ); // Mobile device pairing and management - controllers.extend(crate::openhuman::devices::all_devices_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::devices::all_devices_registered_controllers(), + ); // Durable agent session database — queryable index over transcripts, lineage, tool calls - controllers.extend(crate::openhuman::session_db::all_session_db_registered_controllers()); + push( + &mut controllers, + DomainGroup::Agent, + crate::openhuman::session_db::all_session_db_registered_controllers(), + ); // One-time legacy session import into TinyAgents stores - controllers - .extend(crate::openhuman::session_import::all_session_import_registered_controllers()); + push( + &mut controllers, + DomainGroup::Agent, + crate::openhuman::session_import::all_session_import_registered_controllers(), + ); // Background agent command center — read-only grouped view over the run ledger - controllers - .extend(crate::openhuman::agent_orchestration::all_command_center_registered_controllers()); + push( + &mut controllers, + DomainGroup::Agent, + crate::openhuman::agent_orchestration::all_command_center_registered_controllers(), + ); // Durable dynamic workflow runs — definitions + read surface over the run ledger - controllers - .extend(crate::openhuman::agent_orchestration::all_workflow_run_registered_controllers()); + push( + &mut controllers, + DomainGroup::Agent, + crate::openhuman::agent_orchestration::all_workflow_run_registered_controllers(), + ); // Durable agent-team coordination — teams, members, dependency-aware task claiming, messaging - controllers - .extend(crate::openhuman::agent_orchestration::all_agent_team_registered_controllers()); + push( + &mut controllers, + DomainGroup::Agent, + crate::openhuman::agent_orchestration::all_agent_team_registered_controllers(), + ); // Git-worktree isolation manager — list / status / diff / remove worker worktrees (#3376) - controllers - .extend(crate::openhuman::agent_orchestration::all_worktree_registered_controllers()); + push( + &mut controllers, + DomainGroup::Agent, + crate::openhuman::agent_orchestration::all_worktree_registered_controllers(), + ); // User-driven cancel of detached background sub-agents (#3711) - controllers.extend( + push( + &mut controllers, + DomainGroup::Agent, crate::openhuman::agent_orchestration::all_subagent_control_registered_controllers(), ); controllers @@ -346,37 +794,74 @@ fn build_registered_controllers() -> Vec { /// /// These are write-path or internal-only handlers callable by trusted callers /// (e.g. the Tauri scanner ingest path) that should not appear in agent tool listings. -fn build_internal_only_controllers() -> Vec { +fn build_internal_only_controllers() -> Vec { let mut controllers = Vec::new(); // whatsapp_data ingest: scanner-side write path. Callable over RPC by the // Tauri scanner but excluded from agent-facing schema discovery. - controllers.extend(crate::openhuman::whatsapp_data::all_whatsapp_data_internal_controllers()); + push( + &mut controllers, + DomainGroup::Channels, + crate::openhuman::whatsapp_data::all_whatsapp_data_internal_controllers(), + ); // MCP write audit list: internal-only so the desktop UI/CLI can inspect // local write history without exposing cross-client history as an MCP tool. - controllers.extend(crate::openhuman::mcp_audit::all_mcp_audit_internal_controllers()); + push( + &mut controllers, + DomainGroup::Mcp, + crate::openhuman::mcp_audit::all_mcp_audit_internal_controllers(), + ); // tiny.place A2A social-network integration: renderer-callable via core_rpc_relay // but NOT advertised to agents in tool listings or schema discovery. - controllers.extend(crate::openhuman::tinyplace::all_tinyplace_registered_controllers()); + push( + &mut controllers, + DomainGroup::Platform, + crate::openhuman::tinyplace::all_tinyplace_registered_controllers(), + ); // User-consented tiny.place pairing for wrapped agent sessions: UI-callable // via core_rpc_relay, but excluded from agent tool listings/schema discovery. - controllers.extend(crate::openhuman::agent_orchestration::all_pairing_registered_controllers()); + push( + &mut controllers, + DomainGroup::Agent, + crate::openhuman::agent_orchestration::all_pairing_registered_controllers(), + ); // Orchestration read surface (stage 7): the TinyPlaceOrchestrationTab reads // sessions/messages, sends Master steering DMs, marks read, and polls status. // Renderer-only — not advertised to agents. - controllers.extend(crate::openhuman::orchestration::all_registered_controllers()); + push( + &mut controllers, + DomainGroup::Agent, + crate::openhuman::orchestration::all_registered_controllers(), + ); controllers } /// Returns a vector of all currently registered controllers. +/// +/// Filtered by the ambient [`crate::core::runtime::DomainSet`] (#4796): a +/// controller whose [`DomainGroup`] is disabled under the active context is +/// omitted. With no active context, or under `DomainSet::full()`, this returns +/// the complete set (byte-identical to pre-#4796). pub fn all_registered_controllers() -> Vec { - registry().to_vec() + registry() + .iter() + .filter(|g| group_allowed(g.group)) + .map(|g| g.controller.clone()) + .collect() } /// Returns a vector of all controller schemas, derived from the registered /// controllers (the single source of truth). Kept identical in content to the /// registered set — schemas can no longer drift from handlers (Phase 2). +/// +/// Ambient-filtered by the active [`crate::core::runtime::DomainSet`] just like +/// [`all_registered_controllers`], so `/schema` omits gated namespaces +/// automatically under `harness()`. pub fn all_controller_schemas() -> Vec { - registry().iter().map(|c| c.schema.clone()).collect() + registry() + .iter() + .filter(|g| group_allowed(g.group)) + .map(|g| g.controller.schema.clone()) + .collect() } /// Generates a standardized RPC method name from a controller schema. @@ -559,10 +1044,16 @@ pub fn cli_handler_for_namespace(namespace: &str) -> Option { /// Looks up an RPC method name based on namespace and function. pub fn rpc_method_from_parts(namespace: &str, function: &str) -> Option { + // Searches the FULL (unfiltered) registry: this backs parameter validation + // and CLI routing, which are harmless for an about-to-be-rejected gated + // method — the DomainSet gate is enforced at dispatch + // (`try_invoke_registered_rpc`), not here. See that fn for the rationale. registry() .iter() - .find(|r| r.schema.namespace == namespace && r.schema.function == function) - .map(|r| r.rpc_method_name()) + .find(|g| { + g.controller.schema.namespace == namespace && g.controller.schema.function == function + }) + .map(|g| g.controller.rpc_method_name()) } /// Retrieves the schema for a specific RPC method. @@ -570,11 +1061,19 @@ pub fn rpc_method_from_parts(namespace: &str, function: &str) -> Option /// Checks both the agent-facing registry and the internal registry so that /// parameter validation still applies to internal-only methods (e.g. ingest). pub fn schema_for_rpc_method(method: &str) -> Option { + // DomainSet gate (#4796): a method whose group is disabled under the ambient + // context must be indistinguishable from a genuinely-unregistered method at + // EVERY public lookup, not just at dispatch. Filtering here (identically to + // `try_invoke_registered_rpc`) means `invoke_method_inner` never runs param + // validation against a gated method — otherwise a gated `openhuman.flows_*` + // call with bad params would return the controller's validation error + // instead of method-not-found, leaking the hidden RPC surface. No ambient + // context ⇒ `group_allowed` is `true` ⇒ unfiltered, identical to pre-#4796. registry() .iter() .chain(internal_registry().iter()) - .find(|r| r.rpc_method_name() == method) - .map(|r| r.schema.clone()) + .find(|g| g.controller.rpc_method_name() == method && group_allowed(g.group)) + .map(|g| g.controller.schema.clone()) } /// Validates that the provided parameters match the requirements of the controller schema. @@ -722,11 +1221,24 @@ pub async fn try_invoke_registered_rpc( method: &str, params: Map, ) -> Option> { - let handler = registry() + let grouped = registry() .iter() .chain(internal_registry().iter()) - .find(|c| c.rpc_method_name() == method) - .map(|c| c.handler)?; + .find(|g| g.controller.rpc_method_name() == method)?; + + // DomainSet gate (#4796): a method whose group is disabled under the ambient + // context reports as an unknown method (`None`) — the same result a caller + // gets for a genuinely-unregistered method, so a gated domain's controllers + // are indistinguishable from absent. Enforced HERE (dispatch), not in + // schema/validation lookups, to avoid a validate/dispatch split. + if !group_allowed(grouped.group) { + log::debug!( + "[rpc][domain-gate] method '{method}' suppressed — group {:?} disabled under active DomainSet", + grouped.group + ); + return None; + } + let handler = grouped.controller.handler; // Establish the ambient CoreContext for the duration of the handler so // `CoreContext::current()` resolves inside handler bodies (Phase 2). @@ -760,14 +1272,15 @@ pub async fn try_invoke_registered_rpc( /// - There are no duplicate controllers or RPC methods. /// - Namespaces and functions are not empty. /// - Required input names are unique within a controller. -fn validate_registry(registered: &[RegisteredController]) -> Result<(), String> { +fn validate_registry(registered: &[GroupedController]) -> Result<(), String> { use std::collections::{BTreeMap, BTreeSet}; let mut errors: Vec = Vec::new(); let mut registered_keys = BTreeSet::new(); let mut registered_rpc_methods = BTreeSet::new(); - for controller in registered { + for grouped in registered { + let controller = &grouped.controller; let schema = &controller.schema; let key = format!("{}.{}", schema.namespace, schema.function); if !registered_keys.insert(key.clone()) { diff --git a/src/core/all_tests.rs b/src/core/all_tests.rs index 449e685e8e..044e930e20 100644 --- a/src/core/all_tests.rs +++ b/src/core/all_tests.rs @@ -21,6 +21,20 @@ fn noop_handler(_params: Map) -> ControllerFuture { Box::pin(async { Ok(Value::Null) }) } +/// Wrap raw controllers as [`GroupedController`]s (all `Platform`) so the +/// `validate_registry` unit tests — which build hand-made `RegisteredController` +/// lists — can feed the grouped-registry signature (#4796). The group is +/// irrelevant to `validate_registry`, which only inspects `.controller.schema`. +fn grouped(controllers: Vec) -> Vec { + controllers + .into_iter() + .map(|controller| GroupedController { + group: DomainGroup::Platform, + controller, + }) + .collect() +} + #[test] fn validate_registry_rejects_duplicate_namespace_function() { let declared = vec![schema("dup", "fn", vec![]), schema("dup", "fn", vec![])]; @@ -35,7 +49,7 @@ fn validate_registry_rejects_duplicate_namespace_function() { }, ]; - let err = validate_registry(®istered).expect_err("expected duplicate error"); + let err = validate_registry(&grouped(registered)).expect_err("expected duplicate error"); assert!(err.contains("duplicate registered controller `dup.fn`")); } @@ -64,7 +78,7 @@ fn validate_registry_rejects_duplicate_required_inputs() { handler: noop_handler, }]; - let err = validate_registry(®istered).expect_err("expected duplicate input"); + let err = validate_registry(&grouped(registered)).expect_err("expected duplicate input"); assert!(err.contains("duplicate required input `use_cache` in `doctor.models`")); } @@ -82,7 +96,7 @@ fn validate_registry_accepts_valid_registry() { handler: noop_handler, }) .collect::>(); - assert!(validate_registry(®istered).is_ok()); + assert!(validate_registry(&grouped(registered)).is_ok()); } #[test] @@ -522,7 +536,7 @@ fn validate_registry_rejects_empty_namespace() { schema: declared[0].clone(), handler: noop_handler, }]; - let err = validate_registry(®istered).unwrap_err(); + let err = validate_registry(&grouped(registered)).unwrap_err(); assert!(err.contains("namespace must not be empty")); } @@ -533,7 +547,7 @@ fn validate_registry_rejects_empty_function() { schema: declared[0].clone(), handler: noop_handler, }]; - let err = validate_registry(®istered).unwrap_err(); + let err = validate_registry(&grouped(registered)).unwrap_err(); assert!(err.contains("function must not be empty")); } @@ -546,7 +560,7 @@ fn validate_registry_rejects_whitespace_only_namespace() { schema: declared[0].clone(), handler: noop_handler, }]; - let err = validate_registry(®istered).unwrap_err(); + let err = validate_registry(&grouped(registered)).unwrap_err(); assert!(err.contains("namespace must not be empty")); } @@ -567,7 +581,7 @@ fn validate_registry_rejects_duplicate_registered_controllers() { handler: noop_handler, }, ]; - let err = validate_registry(®istered).unwrap_err(); + let err = validate_registry(&grouped(registered)).unwrap_err(); assert!(err.contains("duplicate registered controller `a.b`")); } @@ -626,3 +640,196 @@ fn every_registered_controller_has_matching_declared_schema() { "registry/schema sets must be identical" ); } + +// --- DomainSet registration filter (#4796) ------------------------------ + +use crate::core::runtime::context::CoreContext; +use crate::core::runtime::DomainSet; + +/// The [`DomainGroup`] a registered controller (agent-facing OR internal) is +/// tagged with, looked up by its namespace. Test-only helper over the private +/// grouped registry. +fn group_for_namespace(ns: &str) -> Option { + registry() + .iter() + .chain(internal_registry().iter()) + .find(|g| g.controller.schema.namespace == ns) + .map(|g| g.group) +} + +#[test] +fn full_registration_is_byte_identical() { + // With no ambient CoreContext (⇒ full, no filter), the public + // `all_registered_controllers()` must equal the raw grouped registry — same + // length AND same rpc-method-name sequence IN ORDER. This is the DoD (1) + // proof that wrapping every entry in a `GroupedController` + filtering by the + // ambient DomainSet changes neither the membership nor the ordering of the + // full() surface. + // + // The baseline is the raw `registry()` view rather than a checked-in method + // snapshot (a #4808 review suggestion): `all_registered_controllers()` and + // `registry()` are DIFFERENT code paths — the former exercises the ambient + // filter (`group_allowed`) and re-collects, the latter is the unfiltered + // source — so this asserts the filter is an order-preserving identity under + // full(). A frozen snapshot would instead ossify the controller list and + // force churn on every legitimate new controller; git history is the + // authoritative pre-#4796 baseline for "did the raw list itself change". + let filtered_methods: Vec = all_registered_controllers() + .iter() + .map(|c| c.rpc_method_name()) + .collect(); + let raw_methods: Vec = registry() + .iter() + .map(|g| g.controller.rpc_method_name()) + .collect(); + + assert_eq!( + filtered_methods.len(), + raw_methods.len(), + "unfiltered all_registered_controllers() must equal raw registry length" + ); + // Ordered comparison — NOT sorted. A reordering (or a drop/add) under full() + // would change dispatch/schema iteration order and must fail here. + assert_eq!( + filtered_methods, raw_methods, + "unfiltered rpc-method sequence must be byte-identical (order + membership) to the raw registry" + ); +} + +#[tokio::test] +async fn harness_excludes_gated_namespaces() { + use std::collections::BTreeSet; + + // Baseline (full, no scope) — every family present. + let full_ns: BTreeSet<&str> = all_controller_schemas() + .iter() + .map(|s| s.namespace) + .collect(); + assert!(full_ns.contains("flows"), "full() must expose flows"); + assert!(full_ns.contains("voice"), "full() must expose voice"); + + let ctx = CoreContext::for_test(DomainSet::harness(), None); + let harness_ns: BTreeSet<&'static str> = + CoreContext::scope(ctx, async { all_controller_schemas() }) + .await + .iter() + .map(|s| s.namespace) + .collect(); + + // Harness families remain. + for present in ["memory", "threads", "config", "security", "agent"] { + assert!( + harness_ns.contains(present), + "harness() must keep the `{present}` namespace" + ); + } + // Gate families + platform-only namespaces are gone. + for absent in [ + "flows", + "voice", + "skills", + "wallet", + "meet", + "mcp_clients", + "health", + ] { + assert!( + !harness_ns.contains(absent), + "harness() must omit the gated/platform `{absent}` namespace" + ); + } + assert!( + harness_ns.len() < full_ns.len(), + "harness() must expose strictly fewer namespaces than full()" + ); +} + +#[tokio::test] +async fn dispatch_returns_none_for_gated_method() { + // A method whose group is gated OFF under the ambient DomainSet must + // dispatch as an unknown method (None) — indistinguishable from absent. + let gated_method = all_registered_controllers() + .into_iter() + .find(|c| c.schema.namespace == "flows") + .map(|c| c.rpc_method_name()) + .expect("a flows.* method exists in the full registry"); + + let ctx = CoreContext::for_test(DomainSet::harness(), None); + let out = CoreContext::scope(ctx, try_invoke_registered_rpc(&gated_method, Map::new())).await; + assert!( + out.is_none(), + "gated method `{gated_method}` must dispatch as None under harness()" + ); + + // A harness-family method still routes (Some) — security.policy_info needs + // no workspace, so it is a clean positive control. + let ctx = CoreContext::for_test(DomainSet::harness(), None); + let out = CoreContext::scope( + ctx, + try_invoke_registered_rpc("openhuman.security_policy_info", Map::new()), + ) + .await; + assert!( + out.is_some(), + "harness-family security.policy_info must still route under harness()" + ); +} + +#[tokio::test] +async fn schema_lookup_is_gated_in_lockstep_with_dispatch() { + // #4808 review: `schema_for_rpc_method` must gate identically to + // `try_invoke_registered_rpc`, otherwise `invoke_method_inner` validates a + // gated method's params BEFORE the dispatch gate fires — returning the + // controller's validation error instead of method-not-found and leaking the + // hidden RPC surface. Prove the schema lookup returns None for a gated + // method under harness() (so no validation runs) while a harness-family + // method still resolves. + let gated_method = all_registered_controllers() + .into_iter() + .find(|c| c.schema.namespace == "flows") + .map(|c| c.rpc_method_name()) + .expect("a flows.* method exists in the full registry"); + + // Full (no scope): the gated method's schema IS visible — proves the None + // below is the gate, not a missing method. + assert!( + schema_for_rpc_method(&gated_method).is_some(), + "under full() the schema for `{gated_method}` must resolve" + ); + + let ctx = CoreContext::for_test(DomainSet::harness(), None); + let gated_schema = + CoreContext::scope(ctx, async { schema_for_rpc_method(&gated_method) }).await; + assert!( + gated_schema.is_none(), + "schema lookup for gated `{gated_method}` must be None under harness() (no param validation, no surface leak)" + ); + + let ctx = CoreContext::for_test(DomainSet::harness(), None); + let kept_schema = CoreContext::scope(ctx, async { + schema_for_rpc_method("openhuman.security_policy_info") + }) + .await; + assert!( + kept_schema.is_some(), + "harness-family security.policy_info schema must still resolve under harness()" + ); +} + +#[test] +fn group_mapping_smoke() { + // Representative controller from each harness family maps to its group… + assert_eq!(group_for_namespace("memory"), Some(DomainGroup::Memory)); + assert_eq!(group_for_namespace("threads"), Some(DomainGroup::Threads)); + assert_eq!(group_for_namespace("config"), Some(DomainGroup::Config)); + assert_eq!(group_for_namespace("security"), Some(DomainGroup::Security)); + assert_eq!(group_for_namespace("agent"), Some(DomainGroup::Agent)); + // …and a representative gated one maps to its gate group. + assert_eq!(group_for_namespace("flows"), Some(DomainGroup::Flows)); + assert_eq!(group_for_namespace("skills"), Some(DomainGroup::Skills)); + assert_eq!(group_for_namespace("voice"), Some(DomainGroup::Voice)); + assert_eq!(group_for_namespace("wallet"), Some(DomainGroup::Web3)); + assert_eq!(group_for_namespace("meet"), Some(DomainGroup::Meet)); + // Internal-only registry is grouped too (mcp_audit → Mcp). + assert_eq!(group_for_namespace("mcp_audit"), Some(DomainGroup::Mcp)); +} diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 729dfc7965..095d03ec53 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -1802,85 +1802,113 @@ async fn run_server_with_services( runtime.serve(ready_tx, shutdown_token).await } -/// Registers all long-lived domain event-bus subscribers exactly once. +/// Per-`DomainGroup` gating decision for each event-bus subscriber that +/// [`register_domain_subscribers`] conditionally registers. Extracted as a +/// pure value so the subscriber→group mapping has a single source of truth +/// that the registrar consumes and tests assert directly — without registering +/// real subscribers or touching the process-global event bus (#4796 DoD item 3). /// -/// Guarded by `std::sync::Once` so repeated calls to `bootstrap_core_runtime` -/// are safe and idempotent. +/// Unlisted subscribers (health, scheduler-gate, TokenJuice content-router, +/// session-token seeding, `SessionExpired`, service restart/shutdown) are +/// always registered as core/platform infra and intentionally absent here. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DomainSubscriberPlan { + /// webhook + notification-bridge + composio trigger + task-sources + device-tunnel. + pub platform: bool, + /// channel-inbound + web-only proactive. + pub channels: bool, + /// flows trigger dispatch. + pub flows: bool, + /// memory conversation-persistence + sync-stage bridge. + pub memory: bool, + /// agent_meetings calendar + meeting-event subscribers. + pub meet: bool, + /// agent handlers + background delivery + run-ledger finalizer + orchestration ingest. + pub agent: bool, + /// mcp_registry lifecycle bus init. + pub mcp: bool, +} + +impl DomainSubscriberPlan { + /// The subscriber-registration plan for `domains`. Pure: no side effects. + pub fn for_domains(domains: crate::core::runtime::DomainSet) -> Self { + use crate::core::all::DomainGroup; + Self { + platform: domains.allows(DomainGroup::Platform), + channels: domains.allows(DomainGroup::Channels), + flows: domains.allows(DomainGroup::Flows), + memory: domains.allows(DomainGroup::Memory), + meet: domains.allows(DomainGroup::Meet), + agent: domains.allows(DomainGroup::Agent), + mcp: domains.allows(DomainGroup::Mcp), + } + } +} + +/// Registers all long-lived domain event-bus subscribers, each group at most +/// once per process. +/// +/// Ungated core/platform infra runs exactly once behind `INFRA: Once`; each +/// gated [`DomainGroup`](crate::core::all::DomainGroup) installs the first time +/// it is enabled (tracked by `group_first_time`), so widening the ambient +/// `DomainSet` on a later call (`harness()` → `full()`) still installs the +/// newly-enabled groups without double-subscribing the ones already registered. fn register_domain_subscribers( workspace_dir: std::path::PathBuf, config: crate::openhuman::config::Config, embedded_core: bool, + domains: crate::core::runtime::DomainSet, ) { - use std::sync::{Arc, Once}; - - static REGISTERED: Once = Once::new(); - REGISTERED.call_once(|| { - // Leak the SubscriptionHandle so the background tasks live for the - // entire process — SubscriptionHandle::drop aborts the task. - if let Some(handle) = crate::core::event_bus::subscribe_global(Arc::new( - crate::openhuman::webhooks::bus::WebhookRequestSubscriber::new(), - )) { - std::mem::forget(handle); - } else { - log::warn!("[event_bus] failed to register webhook subscriber — bus not initialized"); - } - - if let Some(handle) = crate::core::event_bus::subscribe_global(Arc::new( - crate::openhuman::channels::bus::ChannelInboundSubscriber::new(), - )) { - std::mem::forget(handle); - } else { - log::warn!("[event_bus] failed to register channel subscriber — bus not initialized"); - } - - // Flows trigger dispatch (issue B2): maps FlowScheduleTick / - // ComposioTriggerReceived / WebhookIncomingRequest onto enabled flows and - // runs `flows::ops::flows_run`. Registered here (unconditional core boot, - // Once-guarded) rather than under channel startup, so schedule/app-event - // workflows still dispatch when no realtime channel is configured or - // `OPENHUMAN_DISABLE_CHANNEL_LISTENERS` short-circuits `start_channels`. - if let Some(handle) = crate::core::event_bus::subscribe_global(Arc::new( - crate::openhuman::flows::bus::FlowTriggerSubscriber::new(Arc::new(config.clone())), - )) { - std::mem::forget(handle); - } else { - log::warn!("[event_bus] failed to register flows trigger subscriber — bus not initialized"); - } + use crate::core::all::DomainGroup; + use std::collections::HashSet; + use std::sync::{Arc, Mutex, Once, OnceLock}; + + let plan = DomainSubscriberPlan::for_domains(domains); + log::debug!("[event_bus] register_domain_subscribers: domains={domains:?} plan={plan:?}"); + + // Per-group idempotency (#4808 review): the previous single process-wide + // `Once` fixed the subscriber set to the FIRST caller's DomainSet — an + // embedder or test that built `harness()`/`none()` first and later widened + // to `full()` would never install the subscribers skipped on that first + // call, even though those domains' controllers are now exposed. Tracking the + // set of already-registered groups lets a later, wider DomainSet install + // exactly the newly-enabled groups (and no group twice). `insert` returns + // `true` only the first time a group is seen. + fn group_first_time(group: DomainGroup) -> bool { + static DONE: OnceLock>> = OnceLock::new(); + DONE.get_or_init(|| Mutex::new(HashSet::new())) + .lock() + .expect("domain-subscriber registry lock poisoned") + .insert(group) + } + // Ungated core/platform infra — health, scheduler-gate, TokenJuice + // content-router, session-token seeding, the SessionExpired handler, and + // service restart/shutdown. These are DomainSet-independent, so they run + // exactly once on the first call regardless of which composition boots + // first. Registered BEFORE any gated subscriber so the SessionExpired + // handler is live before a gated subscriber could publish a 401-derived + // event. Leaked `SubscriptionHandle`s live for the whole process + // (`SubscriptionHandle::drop` aborts the task). + static INFRA: Once = Once::new(); + INFRA.call_once(|| { crate::openhuman::health::bus::register_health_subscriber(); - crate::openhuman::notifications::register_notification_bridge_subscriber(config.clone()); - crate::openhuman::memory_conversations::register_conversation_persistence_subscriber( - workspace_dir.clone(), - ); - crate::openhuman::memory::sync::register_sync_stage_bridge(&config); - if let Err(error) = crate::openhuman::composio::init_composio_trigger_history( - workspace_dir.clone(), - ) { - log::warn!("[composio][history] failed to initialize trigger archive: {error}"); - } - crate::openhuman::composio::register_composio_trigger_subscriber(); - crate::openhuman::agent_meetings::calendar::register_meet_calendar_subscriber(); - crate::openhuman::agent_meetings::bus::register_meeting_event_subscriber(); - // Orchestration: ingest tiny.place harness session DMs off the stream bus. - crate::openhuman::orchestration::register_orchestration_ingest_subscriber(); - // Task-sources proactive ingestion: connection-created hook + poll. - crate::openhuman::task_sources::bus::register_task_sources_subscriber(); - // Initialise the scheduler gate before any background AI workers - // start so they observe a real policy on their first iteration - // (otherwise they fall back to `Policy::Normal` and miss the - // initial throttle decision on battery-powered hosts). + + // Initialise the scheduler gate before any background AI workers start + // so they observe a real policy on their first iteration (otherwise they + // fall back to `Policy::Normal` and miss the initial throttle decision on + // battery-powered hosts). crate::openhuman::scheduler_gate::init_global(&config); // Install the TokenJuice content-router runtime config (compressor - // toggles + CCR cache limits + optional on-disk tier). Compaction runs - // on every agent's tool output, so this must be set before any agent - // loop executes a tool. + // toggles + CCR cache limits + optional on-disk tier). Compaction runs on + // every agent's tool output, so this must be set before any agent loop + // executes a tool. crate::openhuman::tokenjuice::install_from_config(&config); - // Seed the scheduler-gate signed-out override from the on-disk - // session. Without this, a sidecar that boots with no stored JWT - // would happily spin up cron / channel loops and fire LLM requests - // that all 401 immediately. + // Seed the scheduler-gate signed-out override from the on-disk session. + // Without this, a sidecar that boots with no stored JWT would happily + // spin up cron / channel loops and fire LLM requests that all 401. match crate::api::jwt::get_session_token(&config) { Ok(Some(_)) => { crate::openhuman::scheduler_gate::set_signed_out(false); @@ -1905,9 +1933,9 @@ fn register_domain_subscribers( } } - // Register the SessionExpired handler before any subscribers that - // might publish 401-derived events, so the very first 401 is - // routed through `clear_session` + the scheduler-gate override. + // Register the SessionExpired handler before any subscribers that might + // publish 401-derived events, so the very first 401 is routed through + // `clear_session` + the scheduler-gate override. if let Some(handle) = crate::core::event_bus::subscribe_global(Arc::new( crate::openhuman::credentials::bus::SessionExpiredSubscriber::new(), )) { @@ -1930,46 +1958,154 @@ fn register_domain_subscribers( // subscriber exits the current process after a short grace period. crate::openhuman::service::bus::register_shutdown_subscriber(); } + }); - // Proactive message subscriber (web-only in the desktop runtime — - // no external channel instances are registered here). Uses a - // Once-guarded registrar so domain-level startup can't duplicate it. - crate::openhuman::channels::proactive::register_web_only_proactive_subscriber(); - - // Device tunnel subscriber: handles tunnel:frame handshakes, peer-status - // events, and register acks. Must be registered before any tunnel:frame - // events can arrive. - crate::openhuman::devices::bus::register_device_tunnel_subscriber(); - - // Native request handlers — typed in-process request/response. - // The agent `agent.run_turn` handler is what channel dispatch - // calls instead of importing `run_tool_call_loop` directly. - crate::openhuman::agent::bus::register_agent_handlers(); - - // Background-completion delivery: when a detached sub-agent - // (spawn_async_subagent) finishes, surface its result back into the - // originating chat as an idle-gated, batched, system-injected turn. - crate::openhuman::agent_orchestration::background_delivery::register_background_delivery(); - - // Run-ledger finalizer: detached `spawn_async_subagent` runs outlive - // their parent turn, so their terminal `AgentProgress` never reaches the - // per-turn progress bridge that settles the ledger. This global-bus - // subscriber settles `agent_runs` from `DomainEvent::Subagent{Completed, - // Failed}` (always fired from the detached task), preventing rows from - // leaking as perpetual `running` timeline entries on thread reopen. - crate::openhuman::agent_orchestration::run_ledger_finalize::register_run_ledger_finalize_subscriber(&config); - - // MCP clients lifecycle subscriber: logs McpServer{Installed,Connected, - // Disconnected} + McpClientToolExecuted for observability. The boot-time - // spawn of installed servers (boot::spawn_installed_servers) runs later - // in bootstrap_core_runtime; this subscriber must be live before then so - // those connect events are observed (issue #3039 gap A1). - crate::openhuman::mcp_registry::bus::init(); + // ---- Gated domain subscribers — each group installed at most once, the + // first time its owning DomainGroup is enabled. ------------------------- + + // Platform: webhook + notification bridge + composio trigger + task-sources + // proactive ingestion + device tunnel. + if plan.platform { + if group_first_time(DomainGroup::Platform) { + if let Some(handle) = crate::core::event_bus::subscribe_global(Arc::new( + crate::openhuman::webhooks::bus::WebhookRequestSubscriber::new(), + )) { + std::mem::forget(handle); + } else { + log::warn!( + "[event_bus] failed to register webhook subscriber — bus not initialized" + ); + } + crate::openhuman::notifications::register_notification_bridge_subscriber( + config.clone(), + ); + if let Err(error) = + crate::openhuman::composio::init_composio_trigger_history(workspace_dir.clone()) + { + log::warn!("[composio][history] failed to initialize trigger archive: {error}"); + } + crate::openhuman::composio::register_composio_trigger_subscriber(); + crate::openhuman::task_sources::bus::register_task_sources_subscriber(); + // Device tunnel subscriber: handles tunnel:frame handshakes, + // peer-status events, and register acks. Must be live before any + // tunnel:frame events can arrive. + crate::openhuman::devices::bus::register_device_tunnel_subscriber(); + } + } else { + log::debug!( + "[event_bus] Platform subscribers (webhook/notification/composio/task-sources/device-tunnel) SKIPPED — Platform domain disabled" + ); + } - log::info!( - "[event_bus] domain subscribers registered (webhook, channel, health, conversation, composio, restart, proactive, agent, session_expired, mcp_client)" + // Channels: inbound dispatch + web-only proactive messaging. + if plan.channels { + if group_first_time(DomainGroup::Channels) { + if let Some(handle) = crate::core::event_bus::subscribe_global(Arc::new( + crate::openhuman::channels::bus::ChannelInboundSubscriber::new(), + )) { + std::mem::forget(handle); + } else { + log::warn!( + "[event_bus] failed to register channel subscriber — bus not initialized" + ); + } + // Web-only proactive message subscriber (no external channel + // instances are registered here in the desktop runtime). + crate::openhuman::channels::proactive::register_web_only_proactive_subscriber(); + } + } else { + log::debug!( + "[event_bus] Channels subscribers (inbound + web-only proactive) SKIPPED — Channels domain disabled" ); - }); + } + + // Flows trigger dispatch (issue B2): maps FlowScheduleTick / + // ComposioTriggerReceived / WebhookIncomingRequest onto enabled flows and + // runs `flows::ops::flows_run`, so schedule/app-event workflows still + // dispatch when no realtime channel is configured or + // `OPENHUMAN_DISABLE_CHANNEL_LISTENERS` short-circuits `start_channels`. + if plan.flows { + if group_first_time(DomainGroup::Flows) { + if let Some(handle) = crate::core::event_bus::subscribe_global(Arc::new( + crate::openhuman::flows::bus::FlowTriggerSubscriber::new(Arc::new(config.clone())), + )) { + std::mem::forget(handle); + } else { + log::warn!( + "[event_bus] failed to register flows trigger subscriber — bus not initialized" + ); + } + } + } else { + log::debug!("[event_bus] flows trigger subscriber SKIPPED — Flows domain disabled"); + } + + // Memory: conversation-persistence + sync-stage bridge. + if plan.memory { + if group_first_time(DomainGroup::Memory) { + crate::openhuman::memory_conversations::register_conversation_persistence_subscriber( + workspace_dir.clone(), + ); + crate::openhuman::memory::sync::register_sync_stage_bridge(&config); + } + } else { + log::debug!( + "[event_bus] memory conversation-persistence + sync bridge SKIPPED — Memory domain disabled" + ); + } + + // Meet: calendar + meeting-event subscribers. + if plan.meet { + if group_first_time(DomainGroup::Meet) { + crate::openhuman::agent_meetings::calendar::register_meet_calendar_subscriber(); + crate::openhuman::agent_meetings::bus::register_meeting_event_subscriber(); + } + } else { + log::debug!("[event_bus] agent_meetings subscribers SKIPPED — Meet domain disabled"); + } + + // Agent: orchestration ingest + native agent handlers + background-completion + // delivery + run-ledger finalizer. + if plan.agent { + if group_first_time(DomainGroup::Agent) { + // Orchestration: ingest tiny.place harness session DMs off the stream bus. + crate::openhuman::orchestration::register_orchestration_ingest_subscriber(); + // Native request handlers — the agent `agent.run_turn` handler is + // what channel dispatch calls instead of importing + // `run_tool_call_loop` directly. + crate::openhuman::agent::bus::register_agent_handlers(); + // Background-completion delivery: when a detached sub-agent + // (spawn_async_subagent) finishes, surface its result back into the + // originating chat as an idle-gated, batched, system-injected turn. + crate::openhuman::agent_orchestration::background_delivery::register_background_delivery(); + // Run-ledger finalizer: detached `spawn_async_subagent` runs outlive + // their parent turn, so their terminal `AgentProgress` never reaches + // the per-turn progress bridge that settles the ledger. This + // global-bus subscriber settles `agent_runs` from + // `DomainEvent::Subagent{Completed,Failed}`, preventing rows from + // leaking as perpetual `running` timeline entries on thread reopen. + crate::openhuman::agent_orchestration::run_ledger_finalize::register_run_ledger_finalize_subscriber(&config); + } + } else { + log::debug!( + "[event_bus] agent handlers + background delivery + run-ledger finalizer + orchestration ingest SKIPPED — Agent domain disabled" + ); + } + + // MCP clients lifecycle subscriber: logs McpServer{Installed,Connected, + // Disconnected} + McpClientToolExecuted for observability. The boot-time + // spawn of installed servers (boot::spawn_installed_servers) runs later in + // bootstrap_core_runtime; this subscriber must be live before then so those + // connect events are observed (issue #3039 gap A1). + if plan.mcp { + if group_first_time(DomainGroup::Mcp) { + crate::openhuman::mcp_registry::bus::init(); + } + } else { + log::debug!("[event_bus] mcp_registry bus init SKIPPED — Mcp domain disabled"); + } + + log::info!("[event_bus] domain subscriber registration complete: plan={plan:?}"); } /// Initializes long-lived socket/event-bus infrastructure. @@ -1983,6 +2119,7 @@ fn register_domain_subscribers( pub async fn bootstrap_core_runtime( host_kind: crate::core::types::HostKind, config: Option, + domains: crate::core::runtime::DomainSet, ) { use crate::openhuman::socket::{set_global_socket_manager, SocketManager}; use std::sync::Arc; @@ -2001,10 +2138,12 @@ pub async fn bootstrap_core_runtime( // Ensure the global event bus is initialized (no-op if already done by start_channels). crate::core::event_bus::init_global(crate::core::event_bus::DEFAULT_CAPACITY); crate::openhuman::file_state::init_global(); - // Register domain subscribers for cross-module event handling. - // Uses a Once guard so repeated calls to bootstrap_core_runtime() - // cannot double-subscribe. - register_domain_subscribers(workspace_dir.clone(), cfg.clone(), embedded_core); + // Register domain subscribers for cross-module event handling. Ungated infra + // runs once (INFRA: Once) and each DomainGroup installs at most once via the + // per-group `group_first_time` set, so repeated calls to + // bootstrap_core_runtime() cannot double-subscribe (and a later, wider + // DomainSet still installs its newly-enabled groups). + register_domain_subscribers(workspace_dir.clone(), cfg.clone(), embedded_core, domains); // --- Turn-state recovery ------------------------------------------- // Any per-thread turn snapshots left on disk from a previous process @@ -2068,10 +2207,14 @@ pub async fn bootstrap_core_runtime( // --- x402 payment ledger --- // Initializes the JSONL-backed spending ledger for machine-payable API // payments (x402 protocol). Budget defaults can be overridden via - // the `openhuman.x402_update_budget` RPC. - { + // the `openhuman.x402_update_budget` RPC. Gated on the Web3 domain (#4808 + // review): under `harness()`/`none()` the x402 controllers are absent, so + // their ledger must not initialize either. + if domains.allows(crate::core::all::DomainGroup::Web3) { let x402_session = format!("x402-{}", uuid::Uuid::new_v4()); crate::openhuman::x402::init_ledger(&workspace_dir, &x402_session); + } else { + log::debug!("[boot] x402 payment ledger SKIPPED — Web3 domain disabled"); } // --- Sub-agent definition registry bootstrap --- @@ -2130,7 +2273,13 @@ pub async fn bootstrap_core_runtime( // installs. Idempotent — shares a process-global OnceLock with the // `start_channels` site so it registers exactly once regardless of which // path runs first. (Matching only for now; activation handoff still pending.) - crate::openhuman::skills::bus::ensure_triggered_workflow_subscriber(&workspace_dir); + // Gated on the Skills domain (#4808 review): under `harness()`/`none()` the + // skills controllers are absent, so their trigger subscriber must not install. + if domains.allows(crate::core::all::DomainGroup::Skills) { + crate::openhuman::skills::bus::ensure_triggered_workflow_subscriber(&workspace_dir); + } else { + log::debug!("[boot] triggered-workflow subscriber SKIPPED — Skills domain disabled"); + } // --- Approval gate (#1339) --- // ON by default; opt out with `OPENHUMAN_APPROVAL_GATE=0` (or `false`). diff --git a/src/core/jsonrpc_tests.rs b/src/core/jsonrpc_tests.rs index fedcdd4032..11b18a99cd 100644 --- a/src/core/jsonrpc_tests.rs +++ b/src/core/jsonrpc_tests.rs @@ -8,9 +8,76 @@ use tokio_util::sync::CancellationToken; use super::{ build_http_schema_dump, default_state, escape_html, invoke_method, is_param_validation_error, is_session_expired_error, is_unconfirmed_unauthorized_error, is_wallet_not_configured_error, - params_to_object, parse_json_params, rpc_handler, type_name, + params_to_object, parse_json_params, rpc_handler, type_name, DomainSubscriberPlan, }; +// ---- domain-subscriber gating (#4796 DoD item 3) ---------------------------- +// `register_domain_subscribers` registers on the process-global event bus behind +// a `Once`, so its gating is proven via the pure `DomainSubscriberPlan` the +// registrar consumes — no real subscribers, no bus mutation. + +#[test] +fn domain_subscriber_plan_full_registers_every_gated_subscriber() { + let plan = DomainSubscriberPlan::for_domains(crate::core::runtime::DomainSet::full()); + assert_eq!( + plan, + DomainSubscriberPlan { + platform: true, + channels: true, + flows: true, + memory: true, + meet: true, + agent: true, + mcp: true, + }, + "full() must register every gated domain subscriber" + ); +} + +#[test] +fn domain_subscriber_plan_none_registers_no_gated_subscriber() { + let plan = DomainSubscriberPlan::for_domains(crate::core::runtime::DomainSet::none()); + assert_eq!( + plan, + DomainSubscriberPlan { + platform: false, + channels: false, + flows: false, + memory: false, + meet: false, + agent: false, + mcp: false, + }, + "none() must register no gated domain subscriber (core infra still runs, ungated)" + ); +} + +#[test] +fn domain_subscriber_plan_harness_gates_by_owning_group() { + let plan = DomainSubscriberPlan::for_domains(crate::core::runtime::DomainSet::harness()); + // harness() = agent + memory + threads + config + security. + assert!( + plan.agent, + "harness keeps agent + orchestration subscribers" + ); + assert!( + plan.memory, + "harness keeps memory conversation-persistence + sync bridge" + ); + // Platform / Channels / Flows / Meet / Mcp are NOT in harness. + assert!( + !plan.platform, + "harness must skip webhook/notification/composio/task-sources/device-tunnel" + ); + assert!( + !plan.channels, + "harness must skip channel-inbound + web-only proactive" + ); + assert!(!plan.flows, "harness must skip flows trigger dispatch"); + assert!(!plan.meet, "harness must skip agent_meetings subscribers"); + assert!(!plan.mcp, "harness must skip mcp_registry bus init"); +} + struct EnvVarGuard { old_values: Vec<(&'static str, Option)>, _lock: MutexGuard<'static, ()>, @@ -176,6 +243,41 @@ async fn invoke_doctor_models_rejects_unknown_param() { assert!(err.contains("unknown param 'invalid'")); } +#[tokio::test] +async fn gated_method_is_unknown_at_transport_even_with_malformed_params() { + // #4808 review (CodeRabbit): prove the schema-gate fix at the JSON-RPC + // TRANSPORT layer (`invoke_method`), not only via direct dispatch. Under a + // harness() ambient context a gated method must return an unknown-method + // error for BOTH well-formed and malformed params — never the controller's + // param-validation error, which would leak that the hidden method exists. + use crate::core::runtime::context::CoreContext; + use crate::core::runtime::DomainSet; + + let gated_method = crate::core::all::all_registered_controllers() + .into_iter() + .find(|c| c.schema.namespace == "flows") + .map(|c| c.rpc_method_name()) + .expect("a flows.* method exists in the full registry"); + + for params in [json!({}), json!({ "obviously_not_a_real_param_xyz": true })] { + let ctx = CoreContext::for_test(DomainSet::harness(), None); + let err = CoreContext::scope( + ctx, + invoke_method(default_state(), &gated_method, params.clone()), + ) + .await + .expect_err("gated flows method must error under harness()"); + assert!( + err.contains("unknown method"), + "gated `{gated_method}` with params {params} must be unknown-method at transport, got: {err}" + ); + assert!( + !err.contains("param"), + "gated `{gated_method}` must NOT leak a param-validation error (surface leak), got: {err}" + ); + } +} + #[tokio::test] async fn invoke_config_get_runtime_flags_via_registry() { let result = invoke_method( diff --git a/src/core/runtime/builder.rs b/src/core/runtime/builder.rs index 5ecf718025..d0188971a4 100644 --- a/src/core/runtime/builder.rs +++ b/src/core/runtime/builder.rs @@ -23,6 +23,7 @@ use std::sync::Arc; use tokio_util::sync::CancellationToken; +use crate::core::all::DomainGroup; use crate::core::jsonrpc::{self, EmbeddedReadySignal}; use crate::core::runtime::context::CoreContext; use crate::core::types::HostKind; @@ -110,6 +111,144 @@ impl ServiceSet { } } +/// Selects which domain *families* exist at runtime on a [`CoreRuntime`] (#4796). +/// +/// Sibling of [`ServiceSet`]: where `ServiceSet` selects background services and +/// transports, `DomainSet` selects which controller/tool/store/subscriber +/// surfaces are live. Each flag is an independent [`DomainGroup`]; presets cover +/// the common hosts: +/// [`DomainSet::full`] (every family — today's behavior, the default), +/// [`DomainSet::harness`] (agent + memory + threads + config + security only — +/// the embeddable agent core used by `examples/embed_headless.rs`), and +/// [`DomainSet::none`] (all domain families disabled; transport built-ins and +/// always-on core infrastructure still run). +/// +/// `full()` is byte-identical to pre-#4796 registration, so the desktop shell +/// and standalone CLI are unchanged. Per-gate Cargo `[features]` (children +/// #4797–#4804) narrow the *compile-time* surface further; this struct is the +/// *runtime* axis they compose with. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DomainSet { + /// Agent definition/registry/experience, orchestration, session DB/import. + pub agent: bool, + /// Documents, knowledge graph, memory tree/sources/sync/diff/goals. + pub memory: bool, + /// Conversation threads, per-thread goals, todos. + pub threads: bool, + /// Persisted runtime configuration. + pub config: bool, + /// Encryption, keyring consent, security policy, approval, plan-review. + pub security: bool, + /// Saved automation workflows (tinyflows graphs). + pub flows: bool, + /// SKILL.md skills, skill runtime, skill registry. + pub skills: bool, + /// MCP client subsystem (Smithery registry, local servers, audit). + pub mcp: bool, + /// Google Meet join, agent meetings, live meet-agent loop. + pub meet: bool, + /// Messaging channels + webview bridges (web channel, whatsapp data, …). + pub channels: bool, + /// Wallet, high-level web3 surface, x402 machine payments. + pub web3: bool, + /// Speech-to-text / text-to-speech, audio toolkit. + pub voice: bool, + /// Image/video media generation. NOTE: today this gates only the + /// `media_generate_*` **agent tools** — no controller/store/subscriber is + /// tagged `Media` (there is no `media` RPC namespace yet), so a custom set + /// with `media: false, platform: true` drops the media tools while any + /// future backing controller would stay live. Fold the media-generation + /// controller into this group when it lands. + pub media: bool, + /// Everything not in a named family — always on in `full()`. + pub platform: bool, +} + +impl DomainSet { + /// Every family on — today's behavior and the [`CoreBuilder`] default. + /// Registration is byte-identical to pre-#4796. + pub fn full() -> Self { + Self { + agent: true, + memory: true, + threads: true, + config: true, + security: true, + flows: true, + skills: true, + mcp: true, + meet: true, + channels: true, + web3: true, + voice: true, + media: true, + platform: true, + } + } + + /// The embeddable agent core: agent + memory + threads + config + security. + /// Every gate family AND `platform` are off. Used by + /// `examples/embed_headless.rs`. + pub fn harness() -> Self { + Self { + agent: true, + memory: true, + threads: true, + config: true, + security: true, + flows: false, + skills: false, + mcp: false, + meet: false, + channels: false, + web3: false, + voice: false, + media: false, + platform: false, + } + } + + /// Nothing on — every family disabled. + pub fn none() -> Self { + Self { + agent: false, + memory: false, + threads: false, + config: false, + security: false, + flows: false, + skills: false, + mcp: false, + meet: false, + channels: false, + web3: false, + voice: false, + media: false, + platform: false, + } + } + + /// Whether the given [`DomainGroup`] is enabled in this set. + pub fn allows(&self, group: DomainGroup) -> bool { + match group { + DomainGroup::Agent => self.agent, + DomainGroup::Memory => self.memory, + DomainGroup::Threads => self.threads, + DomainGroup::Config => self.config, + DomainGroup::Security => self.security, + DomainGroup::Flows => self.flows, + DomainGroup::Skills => self.skills, + DomainGroup::Mcp => self.mcp, + DomainGroup::Meet => self.meet, + DomainGroup::Channels => self.channels, + DomainGroup::Web3 => self.web3, + DomainGroup::Voice => self.voice, + DomainGroup::Media => self.media, + DomainGroup::Platform => self.platform, + } + } +} + /// How the per-process RPC bearer token is seeded. pub enum TokenSource { /// An in-memory bearer supplied by the embedder (the Tauri shell hands its @@ -129,18 +268,20 @@ pub struct CoreBuilder { host_kind: HostKind, token: TokenSource, services: ServiceSet, + domains: DomainSet, host: Option, port: Option, } impl CoreBuilder { - /// Start a builder for the given host kind. Defaults: [`TokenSource::EnvOrFile`] - /// and [`ServiceSet::desktop`]. + /// Start a builder for the given host kind. Defaults: [`TokenSource::EnvOrFile`], + /// [`ServiceSet::desktop`], and [`DomainSet::full`]. pub fn new(host_kind: HostKind) -> Self { Self { host_kind, token: TokenSource::EnvOrFile, services: ServiceSet::desktop(), + domains: DomainSet::full(), host: None, port: None, } @@ -152,6 +293,14 @@ impl CoreBuilder { self } + /// Choose which domain families exist at runtime (default [`DomainSet::full`]). + /// `harness()` builds the embeddable agent core; `none()` disables every + /// domain family while retaining transport built-ins and core infrastructure. + pub fn domains(mut self, domains: DomainSet) -> Self { + self.domains = domains; + self + } + /// Choose how the RPC bearer token is seeded. pub fn token(mut self, token: TokenSource) -> Self { self.token = token; @@ -178,7 +327,7 @@ impl CoreBuilder { /// Stage A). pub async fn build(self) -> anyhow::Result { let (ctx, has_operator_token, config) = - CoreContext::init(self.host_kind, &self.token).await?; + CoreContext::init(self.host_kind, &self.token, self.domains).await?; Ok(CoreRuntime { ctx, @@ -424,7 +573,83 @@ impl CoreRuntime { #[cfg(test)] mod tests { - use super::ServiceSet; + use super::{DomainSet, ServiceSet}; + use crate::core::all::DomainGroup; + + #[test] + fn domain_set_presets_have_expected_flags() { + // full() = every family on (byte-identical registration). + let full = DomainSet::full(); + for group in [ + DomainGroup::Agent, + DomainGroup::Memory, + DomainGroup::Threads, + DomainGroup::Config, + DomainGroup::Security, + DomainGroup::Flows, + DomainGroup::Skills, + DomainGroup::Mcp, + DomainGroup::Meet, + DomainGroup::Channels, + DomainGroup::Web3, + DomainGroup::Voice, + DomainGroup::Media, + DomainGroup::Platform, + ] { + assert!(full.allows(group), "full() must allow {group:?}"); + } + + // harness() = exactly agent/memory/threads/config/security on; all gate + // families AND platform off. + let harness = DomainSet::harness(); + for on in [ + DomainGroup::Agent, + DomainGroup::Memory, + DomainGroup::Threads, + DomainGroup::Config, + DomainGroup::Security, + ] { + assert!(harness.allows(on), "harness() must allow {on:?}"); + } + for off in [ + DomainGroup::Flows, + DomainGroup::Skills, + DomainGroup::Mcp, + DomainGroup::Meet, + DomainGroup::Channels, + DomainGroup::Web3, + DomainGroup::Voice, + DomainGroup::Media, + DomainGroup::Platform, + ] { + assert!(!harness.allows(off), "harness() must NOT allow {off:?}"); + } + + // none() = every family off. + let none = DomainSet::none(); + for group in [ + DomainGroup::Agent, + DomainGroup::Memory, + DomainGroup::Threads, + DomainGroup::Config, + DomainGroup::Security, + DomainGroup::Flows, + DomainGroup::Skills, + DomainGroup::Mcp, + DomainGroup::Meet, + DomainGroup::Channels, + DomainGroup::Web3, + DomainGroup::Voice, + DomainGroup::Media, + DomainGroup::Platform, + ] { + assert!(!none.allows(group), "none() must NOT allow {group:?}"); + } + + // Spot-check the field/group wiring is not transposed. + assert!(DomainSet::harness().allows(DomainGroup::Memory)); + assert!(!DomainSet::harness().allows(DomainGroup::Web3)); + } #[test] fn boot_jobs_are_independent_from_runtime_service_flags() { diff --git a/src/core/runtime/context.rs b/src/core/runtime/context.rs index d9b4b6d184..8bf9357f21 100644 --- a/src/core/runtime/context.rs +++ b/src/core/runtime/context.rs @@ -50,6 +50,11 @@ tokio::task_local! { pub struct CoreContext { host_kind: HostKind, workspace_dir: RwLock>, + /// Which domain families are live for this context (#4796). The registry + /// filters its controller/schema/dispatch surface by this set via + /// [`CoreContext::current`] → [`CoreContext::domains`]. `full()` for the + /// desktop shell / standalone CLI (byte-identical to pre-#4796). + domains: crate::core::runtime::DomainSet, } impl CoreContext { @@ -65,11 +70,13 @@ impl CoreContext { pub async fn init( host_kind: HostKind, token: &TokenSource, + domains: crate::core::runtime::DomainSet, ) -> anyhow::Result<( Arc, bool, Option, )> { + log::debug!("[core-context] init: host_kind={host_kind:?} domains={domains:?}"); // 1. Ensure all controllers are registered before anything dispatches. let _ = crate::core::all::all_registered_controllers(); @@ -115,7 +122,7 @@ impl CoreContext { // (memory, attachments, whatsapp, people) with that exact workspace. let config = match crate::openhuman::config::Config::load_or_init().await { Ok(cfg) => { - init_stores(&cfg).await; + init_stores(&cfg, domains).await; Some(cfg) } Err(e) => { @@ -138,11 +145,12 @@ impl CoreContext { // background jobs start later, from CoreRuntime::serve(), after bind // succeeds. let runtime_config = config.clone(); - crate::core::jsonrpc::bootstrap_core_runtime(host_kind, config).await; + crate::core::jsonrpc::bootstrap_core_runtime(host_kind, config, domains).await; let ctx = Arc::new(CoreContext { host_kind, workspace_dir: RwLock::new(workspace_dir), + domains, }); // Register the process default context (first build wins). Dispatch @@ -157,6 +165,13 @@ impl CoreContext { self.host_kind } + /// Which domain families are live for this context (#4796). The controller + /// registry consults this (via [`CoreContext::current`]) to filter its + /// schema/dispatch/tool surface. `full()` for desktop/CLI. + pub fn domains(&self) -> crate::core::runtime::DomainSet { + self.domains + } + /// The resolved per-user workspace directory this context is bound to. pub fn workspace_dir(&self) -> Result { self.workspace_dir @@ -242,6 +257,23 @@ impl CoreContext { pub async fn scope(ctx: Arc, fut: F) -> F::Output { CURRENT_CONTEXT.scope(ctx, fut).await } + + /// Test-only constructor: build a context with an explicit + /// [`DomainSet`](crate::core::runtime::DomainSet) and optional workspace, so + /// cross-module tests (e.g. `core::all`'s registry filter) can exercise the + /// ambient DomainSet gate without going through the full [`CoreContext::init`] + /// boot sequence. + #[cfg(test)] + pub(crate) fn for_test( + domains: crate::core::runtime::DomainSet, + workspace_dir: Option, + ) -> Arc { + Arc::new(CoreContext { + host_kind: HostKind::Cli, + workspace_dir: RwLock::new(workspace_dir), + domains, + }) + } } /// Initialize the global `MemoryClient` and the other workspace-bound stores so @@ -260,41 +292,100 @@ impl CoreContext { /// client not ready" error rather than reading/writing the wrong workspace. The /// server still comes up; the operator sees the loud error and fixes their /// config or sets `OPENHUMAN_WORKSPACE` to a writable path, then restarts. -pub async fn init_stores(cfg: &crate::openhuman::config::Config) { +/// Per-`DomainGroup` gating decision for each workspace-bound store that +/// [`init_stores`] initializes. Extracted as a pure value so the store-gating +/// mapping (which store is owned by which `DomainGroup`) has a single source of +/// truth that `init_stores` consumes and tests assert directly — without +/// touching process-global store state or booting a runtime (#4796 DoD item 3). +/// +/// The keyring-path log and the credentials Sentry bind in `init_stores` are +/// intentionally *not* represented here: they are unguarded core infra every +/// `DomainSet` needs. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct StoreInitPlan { + /// `memory::global` — gated on [`DomainGroup::Memory`]. + pub memory: bool, + /// `agent::multimodal` attachments sidecar dir — gated on [`DomainGroup::Agent`]. + pub agent_attachments: bool, + /// `whatsapp_data::global` — gated on [`DomainGroup::Channels`]. + pub whatsapp_data: bool, + /// `people::store` — gated on [`DomainGroup::Platform`]. + pub people: bool, + /// legacy-workflow prune under `skills::registry` — gated on [`DomainGroup::Skills`]. + pub skills_prune: bool, +} + +impl StoreInitPlan { + /// The store-init plan for `domains`. Pure: no side effects, no globals. + pub fn for_domains(domains: crate::core::runtime::DomainSet) -> Self { + use crate::core::all::DomainGroup; + Self { + memory: domains.allows(DomainGroup::Memory), + agent_attachments: domains.allows(DomainGroup::Agent), + whatsapp_data: domains.allows(DomainGroup::Channels), + people: domains.allows(DomainGroup::Platform), + skills_prune: domains.allows(DomainGroup::Skills), + } + } +} + +pub async fn init_stores( + cfg: &crate::openhuman::config::Config, + domains: crate::core::runtime::DomainSet, +) { + let plan = StoreInitPlan::for_domains(domains); + let keyring_dir = crate::openhuman::keyring::store::workspace_dir_for_file_backend(); + // Keyring path log + credentials Sentry bind (below) are unguarded — they + // are core infra every DomainSet needs. Each workspace-bound store init is + // gated on its owning DomainGroup so an excluded domain's store stays + // uninitialized under `harness()`/`none()` (#4796 DoD item 3). log::info!( - "[boot] paths: config={} workspace={} keyring_dir={} keyring_backend={}", + "[boot] paths: config={} workspace={} keyring_dir={} keyring_backend={} domains={:?}", cfg.config_path.display(), cfg.workspace_dir.display(), keyring_dir.display(), crate::openhuman::keyring::backend_name(), + domains, ); - match crate::openhuman::memory::global::init(cfg.workspace_dir.clone()) { - Ok(_) => log::info!( - "[boot] memory::global initialized (workspace={})", - cfg.workspace_dir.display() - ), - Err(e) => log::warn!("[boot] memory::global init failed: {e}"), + if plan.memory { + match crate::openhuman::memory::global::init(cfg.workspace_dir.clone()) { + Ok(_) => log::info!( + "[boot] memory::global initialized (workspace={})", + cfg.workspace_dir.display() + ), + Err(e) => log::warn!("[boot] memory::global init failed: {e}"), + } + } else { + log::debug!("[boot] memory::global init SKIPPED — Memory domain disabled"); } // Install the on-disk image-attachment sidecar dir so inbound // image markers persist under /attachments/ instead // of an in-memory FIFO (survives restarts + delegation hops). // Also fires a best-effort stale-file sweep. - crate::openhuman::agent::multimodal::init_attachments_dir( - cfg.workspace_dir.join("attachments"), - ); - log::info!( - "[boot] image attachments sidecar dir = {}", - cfg.workspace_dir.join("attachments").display() - ); + if plan.agent_attachments { + crate::openhuman::agent::multimodal::init_attachments_dir( + cfg.workspace_dir.join("attachments"), + ); + log::info!( + "[boot] image attachments sidecar dir = {}", + cfg.workspace_dir.join("attachments").display() + ); + } else { + log::debug!("[boot] image attachments sidecar dir SKIPPED — Agent domain disabled"); + } // Initialize the WhatsApp data store so scanner ingest calls // can write data without requiring a lazy-init fallback. - match crate::openhuman::whatsapp_data::global::init(cfg.workspace_dir.clone()) { - Ok(_) => log::info!( - "[boot] whatsapp_data::global initialized (workspace={})", - cfg.workspace_dir.display() - ), - Err(e) => log::warn!("[boot] whatsapp_data::global init failed: {e}"), + if plan.whatsapp_data { + match crate::openhuman::whatsapp_data::global::init(cfg.workspace_dir.clone()) { + Ok(_) => log::info!( + "[boot] whatsapp_data::global initialized (workspace={})", + cfg.workspace_dir.display() + ), + Err(e) => log::warn!("[boot] whatsapp_data::global init failed: {e}"), + } + } else { + log::debug!("[boot] whatsapp_data::global init SKIPPED — Channels domain disabled"); } // Seed the people store so people controllers + `people_*` // tools can read/write. Without this the process-global stays @@ -302,18 +393,26 @@ pub async fn init_stores(cfg: &crate::openhuman::config::Config) { // initialised" (Sentry TAURI-RUST-8NM). Sits inside this // Ok(cfg) arm so it inherits the wrong-workspace guard above // (never seed against a Config::default fallback). - match crate::openhuman::people::store::init_from_workspace(&cfg.workspace_dir) { - Ok(_) => log::info!( - "[boot] people::store initialized (workspace={})", - cfg.workspace_dir.display() - ), - Err(e) => log::warn!("[boot] people::store init failed: {e}"), + if plan.people { + match crate::openhuman::people::store::init_from_workspace(&cfg.workspace_dir) { + Ok(_) => log::info!( + "[boot] people::store initialized (workspace={})", + cfg.workspace_dir.display() + ), + Err(e) => log::warn!("[boot] people::store init failed: {e}"), + } + } else { + log::debug!("[boot] people::store init SKIPPED — Platform domain disabled"); } // Prune legacy bundled skills (dev-workflow / github-issue-crusher // / pr-review-shepherd) that older builds seeded into // /skills/. OpenHuman no longer ships bundled defaults; // this removes the stale dirs on upgrade. Idempotent. - crate::openhuman::skills::registry::prune_legacy_default_workflows(&cfg.workspace_dir); + if plan.skills_prune { + crate::openhuman::skills::registry::prune_legacy_default_workflows(&cfg.workspace_dir); + } else { + log::debug!("[boot] skills legacy-workflow prune SKIPPED — Skills domain disabled"); + } // Boot-time Sentry user binding — issue #3135. If the user is // already signed in (typical desktop restart), the auth-profile // store has their `user_id` *now*, before any background loop @@ -341,6 +440,7 @@ mod tests { Arc::new(CoreContext { host_kind: HostKind::Cli, workspace_dir: RwLock::new(Some(PathBuf::from(dir))), + domains: crate::core::runtime::DomainSet::full(), }) } @@ -350,6 +450,63 @@ mod tests { // directly (independent of the process DEFAULT_CONTEXT global, since // `current()` inside a scope resolves the scoped value). + // ---- store-init gating (#4796 DoD item 3) -------------------------------- + // `init_stores` side-effects on process globals with no init-state probe, so + // the gating is proven via the pure `StoreInitPlan` the registrar consumes. + + #[test] + fn store_init_plan_full_initializes_every_store() { + let plan = StoreInitPlan::for_domains(crate::core::runtime::DomainSet::full()); + assert_eq!( + plan, + StoreInitPlan { + memory: true, + agent_attachments: true, + whatsapp_data: true, + people: true, + skills_prune: true, + }, + "full() must initialize every workspace-bound store" + ); + } + + #[test] + fn store_init_plan_none_initializes_nothing() { + let plan = StoreInitPlan::for_domains(crate::core::runtime::DomainSet::none()); + assert_eq!( + plan, + StoreInitPlan { + memory: false, + agent_attachments: false, + whatsapp_data: false, + people: false, + skills_prune: false, + }, + "none() must leave every workspace-bound store uninitialized" + ); + } + + #[test] + fn store_init_plan_harness_gates_by_owning_group() { + let plan = StoreInitPlan::for_domains(crate::core::runtime::DomainSet::harness()); + // harness() = agent + memory + threads + config + security. + assert!(plan.memory, "harness keeps memory::global (Memory)"); + assert!( + plan.agent_attachments, + "harness keeps agent attachments sidecar (Agent)" + ); + // Channels / Platform / Skills are NOT in harness → their stores stay off. + assert!( + !plan.whatsapp_data, + "harness must skip whatsapp_data::global (Channels)" + ); + assert!(!plan.people, "harness must skip people::store (Platform)"); + assert!( + !plan.skills_prune, + "harness must skip skills legacy-prune (Skills)" + ); + } + #[tokio::test] async fn scope_sets_current_context() { let a = ctx("/tmp/ctx-a"); @@ -360,6 +517,19 @@ mod tests { assert_eq!(seen, Some(PathBuf::from("/tmp/ctx-a"))); } + #[tokio::test] + async fn scoped_context_exposes_its_domain_set() { + // The ambient `current().domains()` must reflect the scoped context's + // DomainSet — this is the seam the registry filter reads (#4796). + let harness = crate::core::runtime::DomainSet::harness(); + let ctx = CoreContext::for_test(harness, Some(PathBuf::from("/tmp/ctx-domains"))); + let seen = + CoreContext::scope(ctx, async { CoreContext::current().map(|c| c.domains()) }).await; + assert_eq!(seen, Some(harness)); + assert!(seen.unwrap().allows(crate::core::all::DomainGroup::Memory)); + assert!(!seen.unwrap().allows(crate::core::all::DomainGroup::Web3)); + } + #[tokio::test] async fn nested_scope_overrides_then_restores() { let a = ctx("/tmp/ctx-a"); @@ -390,10 +560,12 @@ mod tests { let a = Arc::new(CoreContext { host_kind: HostKind::Cli, workspace_dir: RwLock::new(Some(dir_a.path().to_path_buf())), + domains: crate::core::runtime::DomainSet::full(), }); let b = Arc::new(CoreContext { host_kind: HostKind::Cli, workspace_dir: RwLock::new(Some(dir_b.path().to_path_buf())), + domains: crate::core::runtime::DomainSet::full(), }); let store_a = a.people().expect("open people store for workspace A"); @@ -413,6 +585,7 @@ mod tests { let ctx = CoreContext { host_kind: HostKind::Cli, workspace_dir: RwLock::new(Some(dir_a.path().to_path_buf())), + domains: crate::core::runtime::DomainSet::full(), }; let store_a = ctx.people().expect("open people store for workspace A"); @@ -433,10 +606,12 @@ mod tests { let a = Arc::new(CoreContext { host_kind: HostKind::Cli, workspace_dir: RwLock::new(Some(dir_a.path().to_path_buf())), + domains: crate::core::runtime::DomainSet::full(), }); let b = Arc::new(CoreContext { host_kind: HostKind::Cli, workspace_dir: RwLock::new(Some(dir_b.path().to_path_buf())), + domains: crate::core::runtime::DomainSet::full(), }); let params = serde_json::json!({ @@ -483,6 +658,7 @@ mod tests { let ctx = CoreContext { host_kind: HostKind::Cli, workspace_dir: RwLock::new(None), + domains: crate::core::runtime::DomainSet::full(), }; let err = match ctx.people() { diff --git a/src/core/runtime/mod.rs b/src/core/runtime/mod.rs index bd4a1d683a..4b3e3a314f 100644 --- a/src/core/runtime/mod.rs +++ b/src/core/runtime/mod.rs @@ -28,4 +28,4 @@ pub mod builder; pub mod context; pub mod services; -pub use builder::{CoreBuilder, CoreRuntime, ServiceSet, TokenSource}; +pub use builder::{CoreBuilder, CoreRuntime, DomainSet, ServiceSet, TokenSource}; diff --git a/src/lib.rs b/src/lib.rs index 43231f0111..02ef513268 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -24,7 +24,7 @@ pub use openhuman::memory_store::{MemoryClient, MemoryState}; /// Embeddable core composition API. Host the OpenHuman core in any process — /// the Tauri shell, a CLI, a stdio MCP server, or a cloud/team server — via /// [`CoreBuilder`] → [`CoreRuntime`]. See `docs/plans/pluggable-core/`. -pub use core::runtime::{CoreBuilder, CoreRuntime, ServiceSet, TokenSource}; +pub use core::runtime::{CoreBuilder, CoreRuntime, DomainSet, ServiceSet, TokenSource}; pub use core::types::HostKind; /// Runs the core logic based on the provided command-line arguments. diff --git a/src/openhuman/tools/ops.rs b/src/openhuman/tools/ops.rs index e54b0c76d1..f4c9e9c29d 100644 --- a/src/openhuman/tools/ops.rs +++ b/src/openhuman/tools/ops.rs @@ -1051,7 +1051,154 @@ pub fn all_tools_with_runtime( ); } - tools + // DomainSet post-filter (#4796): drop tools whose DomainGroup is disabled + // under the ambient CoreContext. With no active context, or under + // `DomainSet::full()`, every tool is kept (byte-identical). Under + // `harness()` the gate-family tools (web3/mcp/skills/flows/media/voice/meet) + // are dropped so agent turns can't call a domain that isn't live; only the + // memory + threads tools survive (the mapped harness families) — see + // `tool_group` for the classification and its Platform-default caveat. + let domains = crate::core::runtime::context::CoreContext::current().map(|c| c.domains()); + if let Some(set) = domains { + let before = tools.len(); + let filtered: Vec> = tools + .into_iter() + .filter(|t| set.allows(tool_group(t.name()))) + .collect(); + log::debug!( + "[tools::ops][domain-filter] ambient DomainSet active — {} of {before} tools retained", + filtered.len() + ); + filtered + } else { + // No ambient context (unit tests / pre-boot) ⇒ no filtering. + tools + } +} + +/// Classify an agent tool into its [`DomainGroup`](crate::core::all::DomainGroup) +/// by its `name()`, so [`all_tools_with_runtime`] can drop tools whose family is +/// disabled under the ambient [`DomainSet`](crate::core::runtime::DomainSet). +/// +/// Only the gate families (Web3/Mcp/Skills/Flows/Media/Voice/Meet) and the two +/// mapped harness families (Memory/Threads) are matched; **everything else +/// defaults to `Platform`**. Consequence under `harness()` (platform off): the +/// gate-family tools drop AND the generic Platform tools (shell/file/grep/edit/ +/// screen/billing/team/cron/config/security/agent-orchestration/…) drop too — +/// only memory + thread/todo tools remain. This is the strict #4796 harness +/// surface; an embedder that wants a broader tool set can widen its DomainSet. +/// (Names verified against each Tool impl's `fn name()` on 2026-07-13.) +fn tool_group(name: &str) -> crate::core::all::DomainGroup { + use crate::core::all::DomainGroup; + + // Gate families with a domain-exclusive name prefix are matched by prefix + // (not an exact list) so a NEW tool in the family auto-gates instead of + // silently defaulting to Platform and leaking under a custom DomainSet + // (#4808 maintainer review). Web3 = wallet_/web3_/x402_, Media = media_, + // Mcp = mcp_ (below). Families without a clean prefix (Skills/Flows) keep + // their exact lists; `no_gate_family_tool_silently_defaults_to_platform` + // guards the prefix families. + const SKILLS: &[&str] = &[ + "run_workflow", + "await_workflow", + "list_workflows", + "create_workflow", + "describe_workflow", + "read_workflow_resource", + "list_workflow_runs", + "read_workflow_run_log", + "install_workflow_from_url", + "uninstall_workflow", + "skill_registry_browse", + "skill_registry_search", + "skill_registry_install", + "skill_registry_sources", + "skill_registry_uninstall", + "skill_runtime_resolve_runtimes", + ]; + const FLOWS: &[&str] = &[ + "propose_workflow", + "revise_workflow", + "dry_run_workflow", + "save_workflow", + "suggest_workflows", + "run_flow", + "list_flows", + "get_flow", + "get_flow_run", + "list_flow_connections", + "search_tool_catalog", + "get_tool_contract", + "get_tool_output_sample", + "list_agent_profiles", + ]; + // Voice family agent tools (audio_toolkit) — no `voice_`/`tts_`/`stt_` + // prefix, so they must be listed explicitly or they fall through to + // Platform and stay callable when Voice is gated off (#4808 review). + const VOICE: &[&str] = &[ + "audio_generate_podcast", + "audio_email_podcast", + "audio_generate_and_email_podcast", + ]; + // Threads: thread_* / todo_* handled by prefix below; these are the extras. + const THREADS_EXTRA: &[&str] = &["transcript_search", "goal_get", "goal_set", "goal_complete"]; + // Memory extras not covered by the `memory_`/`goals_` prefixes. + const MEMORY_EXTRA: &[&str] = &[ + "remember_preference", + "save_preference", + "update_memory_md", + "tool_stats", + ]; + + // MCP: every MCP tool name is `mcp_` prefixed (mcp_registry_*, mcp_setup_*, + // mcp_call_tool, mcp_list_servers, mcp_list_tools). + if name.starts_with("mcp_") { + return DomainGroup::Mcp; + } + // Web3: wallet_/web3_/x402_ are all Web3-exclusive prefixes. + if name.starts_with("wallet_") || name.starts_with("web3_") || name.starts_with("x402_") { + return DomainGroup::Web3; + } + if SKILLS.contains(&name) { + return DomainGroup::Skills; + } + if FLOWS.contains(&name) { + return DomainGroup::Flows; + } + // Media generation: `media_` prefix (media_generate_image/video, media_list_models). + if name.starts_with("media_") { + return DomainGroup::Media; + } + // Channels family agent tools: read-only WhatsApp data surface. Gated with + // the other channel/webview domains; without this they fall to Platform and + // stay callable when Channels is gated off (#4808 review). + if name.starts_with("whatsapp_data_") { + return DomainGroup::Channels; + } + // Voice family: explicit audio_* podcast tools plus the defensive + // voice_/tts_/stt_ prefixes for any future tool. Meet has no agent tools in + // the current surface, but the `meet_` prefix is mapped defensively. + if VOICE.contains(&name) + || name.starts_with("voice_") + || name.starts_with("tts_") + || name.starts_with("stt_") + { + return DomainGroup::Voice; + } + if name.starts_with("meet_") { + return DomainGroup::Meet; + } + // Memory family (harness-kept): memory_* store/search/etc + goals_* + extras. + if name.starts_with("memory_") || name.starts_with("goals_") || MEMORY_EXTRA.contains(&name) { + return DomainGroup::Memory; + } + // Threads family (harness-kept): thread_* + todo_* + per-thread goal + search. + if name.starts_with("thread_") || name.starts_with("todo_") || THREADS_EXTRA.contains(&name) { + return DomainGroup::Threads; + } + // Everything else — shell/file/screen/config/security/agent/billing/… — is + // Platform: present under full(), absent under harness()/none(). + DomainGroup::Platform } #[cfg(test)] diff --git a/src/openhuman/tools/ops_tests.rs b/src/openhuman/tools/ops_tests.rs index 0f7cbaed19..1d762f46f3 100644 --- a/src/openhuman/tools/ops_tests.rs +++ b/src/openhuman/tools/ops_tests.rs @@ -2174,3 +2174,98 @@ fn desktop_default_off_tools_retained_when_opted_in() { ); } } + +// --- DomainSet tool classifier (#4796) ---------------------------------- + +#[test] +fn tool_group_classifies_gate_and_harness_families() { + use crate::core::all::DomainGroup; + + // Gate families → their gate group (dropped under harness()). + assert_eq!(tool_group("wallet_status"), DomainGroup::Web3); + assert_eq!(tool_group("web3_swap_quote"), DomainGroup::Web3); + assert_eq!(tool_group("x402_request"), DomainGroup::Web3); + assert_eq!(tool_group("mcp_registry_search"), DomainGroup::Mcp); + assert_eq!(tool_group("mcp_call_tool"), DomainGroup::Mcp); + assert_eq!(tool_group("run_workflow"), DomainGroup::Skills); + assert_eq!(tool_group("skill_registry_browse"), DomainGroup::Skills); + assert_eq!(tool_group("list_workflows"), DomainGroup::Skills); + assert_eq!(tool_group("propose_workflow"), DomainGroup::Flows); + assert_eq!(tool_group("list_flows"), DomainGroup::Flows); + assert_eq!(tool_group("media_generate_image"), DomainGroup::Media); + // Voice audio_* tools have no voice_/tts_/stt_ prefix — must be classified + // explicitly, not fall through to Platform (#4808 review). + assert_eq!(tool_group("audio_generate_podcast"), DomainGroup::Voice); + assert_eq!(tool_group("audio_email_podcast"), DomainGroup::Voice); + assert_eq!( + tool_group("audio_generate_and_email_podcast"), + DomainGroup::Voice + ); + // Channels read-only WhatsApp data tools. + assert_eq!( + tool_group("whatsapp_data_list_chats"), + DomainGroup::Channels + ); + assert_eq!( + tool_group("whatsapp_data_search_messages"), + DomainGroup::Channels + ); + + // Harness-mapped families → kept under harness(). + assert_eq!(tool_group("memory_store"), DomainGroup::Memory); + assert_eq!(tool_group("goals_add"), DomainGroup::Memory); + assert_eq!(tool_group("update_memory_md"), DomainGroup::Memory); + assert_eq!(tool_group("thread_list"), DomainGroup::Threads); + assert_eq!(tool_group("todo_add"), DomainGroup::Threads); + assert_eq!(tool_group("goal_get"), DomainGroup::Threads); + + // Everything else → Platform (dropped under harness()). + assert_eq!(tool_group("shell"), DomainGroup::Platform); + assert_eq!(tool_group("file_read"), DomainGroup::Platform); + assert_eq!(tool_group("config_snapshot"), DomainGroup::Platform); + assert_eq!(tool_group("spawn_subagent"), DomainGroup::Platform); +} + +#[test] +fn tool_group_gate_families_dropped_under_harness_not_full() { + use crate::core::runtime::DomainSet; + + let full = DomainSet::full(); + let harness = DomainSet::harness(); + // Full keeps every family. + for name in ["wallet_status", "run_workflow", "memory_store", "shell"] { + assert!(full.allows(tool_group(name)), "full() keeps {name}"); + } + // Harness keeps memory/threads, drops gate families AND platform. + assert!(harness.allows(tool_group("memory_store"))); + assert!(harness.allows(tool_group("thread_list"))); + assert!(!harness.allows(tool_group("wallet_status"))); + assert!(!harness.allows(tool_group("run_workflow"))); + assert!(!harness.allows(tool_group("shell"))); + // The previously-misclassified gate-family tools now drop under harness. + assert!(!harness.allows(tool_group("audio_generate_podcast"))); + assert!(!harness.allows(tool_group("whatsapp_data_list_chats"))); +} + +#[test] +fn no_gate_family_tool_silently_defaults_to_platform() { + use crate::core::all::DomainGroup; + // #4808 maintainer review: a future tool in a prefix-gated family must NOT + // fall through to Platform — otherwise it would stay callable under a custom + // `DomainSet { platform: true, : false }`, leaking the gated surface. + // These synthetic names match no exact list, only the family prefix. + for name in [ + "wallet_new_thing", + "web3_new_thing", + "x402_new_thing", + "mcp_new_thing", + "media_new_thing", + "whatsapp_data_new_thing", + ] { + assert_ne!( + tool_group(name), + DomainGroup::Platform, + "gate-family tool `{name}` must not silently default to Platform" + ); + } +}