feat(core): runtime DomainSet composition axis — group-tagged registry + ambient filter (#4796) (#4808)

This commit is contained in:
oxoxDev
2026-07-13 19:16:02 +04:00
committed by GitHub
parent d8a706fdb6
commit debca303de
13 changed files with 1967 additions and 304 deletions
+7
View File
@@ -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.
@@ -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`).
+20 -8
View File
@@ -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
+651 -138
View File
File diff suppressed because it is too large Load Diff
+214 -7
View File
@@ -21,6 +21,20 @@ fn noop_handler(_params: Map<String, Value>) -> 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<RegisteredController>) -> Vec<GroupedController> {
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(&registered).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(&registered).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::<Vec<_>>();
assert!(validate_registry(&registered).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(&registered).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(&registered).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(&registered).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(&registered).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<DomainGroup> {
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<String> = all_registered_controllers()
.iter()
.map(|c| c.rpc_method_name())
.collect();
let raw_methods: Vec<String> = 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));
}
+244 -95
View File
@@ -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};
use crate::core::all::DomainGroup;
use std::collections::HashSet;
use std::sync::{Arc, Mutex, Once, OnceLock};
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");
}
let plan = DomainSubscriberPlan::for_domains(domains);
log::debug!("[event_bus] register_domain_subscribers: domains={domains:?} plan={plan:?}");
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");
// 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<Mutex<HashSet<DomainGroup>>> = 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();
// ---- Gated domain subscribers — each group installed at most once, the
// first time its owning DomainGroup is enabled. -------------------------
// Device tunnel subscriber: handles tunnel:frame handshakes, peer-status
// events, and register acks. Must be registered before any tunnel:frame
// events can arrive.
// 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"
);
}
// 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.
// 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}` (always fired from the detached task), preventing rows from
// 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).
// 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 subscribers registered (webhook, channel, health, conversation, composio, restart, proactive, agent, session_expired, mcp_client)"
);
});
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<crate::openhuman::config::Config>,
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.)
// 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`).
+103 -1
View File
@@ -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<OsString>)>,
_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(
+229 -4
View File
@@ -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<String>,
port: Option<u16>,
}
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<CoreRuntime> {
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() {
+180 -4
View File
@@ -50,6 +50,11 @@ tokio::task_local! {
pub struct CoreContext {
host_kind: HostKind,
workspace_dir: RwLock<Option<std::path::PathBuf>>,
/// 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<CoreContext>,
bool,
Option<crate::openhuman::config::Config>,
)> {
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<std::path::PathBuf, String> {
self.workspace_dir
@@ -242,6 +257,23 @@ impl CoreContext {
pub async fn scope<F: Future>(ctx: Arc<CoreContext>, 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<std::path::PathBuf>,
) -> Arc<CoreContext> {
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,15 +292,63 @@ 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,
);
if plan.memory {
match crate::openhuman::memory::global::init(cfg.workspace_dir.clone()) {
Ok(_) => log::info!(
"[boot] memory::global initialized (workspace={})",
@@ -276,10 +356,14 @@ pub async fn init_stores(cfg: &crate::openhuman::config::Config) {
),
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 <workspace>/attachments/ instead
// of an in-memory FIFO (survives restarts + delegation hops).
// Also fires a best-effort stale-file sweep.
if plan.agent_attachments {
crate::openhuman::agent::multimodal::init_attachments_dir(
cfg.workspace_dir.join("attachments"),
);
@@ -287,8 +371,12 @@ pub async fn init_stores(cfg: &crate::openhuman::config::Config) {
"[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.
if plan.whatsapp_data {
match crate::openhuman::whatsapp_data::global::init(cfg.workspace_dir.clone()) {
Ok(_) => log::info!(
"[boot] whatsapp_data::global initialized (workspace={})",
@@ -296,12 +384,16 @@ pub async fn init_stores(cfg: &crate::openhuman::config::Config) {
),
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
// empty and every call fails with "people store not
// 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).
if plan.people {
match crate::openhuman::people::store::init_from_workspace(&cfg.workspace_dir) {
Ok(_) => log::info!(
"[boot] people::store initialized (workspace={})",
@@ -309,11 +401,18 @@ pub async fn init_stores(cfg: &crate::openhuman::config::Config) {
),
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
// <workspace>/skills/. OpenHuman no longer ships bundled defaults;
// this removes the stale dirs on upgrade. Idempotent.
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() {
+1 -1
View File
@@ -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};
+1 -1
View File
@@ -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.
+147
View File
@@ -1051,7 +1051,154 @@ pub fn all_tools_with_runtime(
);
}
// 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<Box<dyn Tool>> = 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)]
+95
View File
@@ -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, <family>: 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"
);
}
}