diff --git a/.github/workflows/ci-lite.yml b/.github/workflows/ci-lite.yml index 4e2afa1c8..4d65bf31d 100644 --- a/.github/workflows/ci-lite.yml +++ b/.github/workflows/ci-lite.yml @@ -430,7 +430,7 @@ jobs: run: | bash scripts/ci-cancel-aware.sh cargo test --manifest-path Cargo.toml \ --no-default-features --features tokenjuice-treesitter --lib -- \ - core::all:: core::cli:: core::jsonrpc:: core::legacy_aliases:: agent_registry::agents::loader:: + core::all:: core::cli:: core::jsonrpc:: core::legacy_aliases:: core::runtime:: agent_registry::agents::loader:: - name: Guard — new feature-gated test modules must be acknowledged # Self-maintaining coverage: the set of source files that #[cfg]-gate a test on @@ -448,6 +448,7 @@ jobs: core/jsonrpc_tests.rs core/legacy_aliases.rs core/runtime/context.rs + core/runtime/services.rs openhuman/agent/harness/builtin_definitions.rs openhuman/agent/harness/definition_tests.rs openhuman/agent/harness/session/tests.rs diff --git a/src/core/runtime/builder.rs b/src/core/runtime/builder.rs index 565a1a475..1c595321e 100644 --- a/src/core/runtime/builder.rs +++ b/src/core/runtime/builder.rs @@ -57,6 +57,12 @@ pub struct ServiceSet { pub skill_catalog_refresh: bool, /// Boot installed MCP servers and supervise reconnects during runtime bootstrap. pub mcp_boot: bool, + /// Composio integration sync: periodic connection sync + one-shot memory-source reconcile. + pub integrations: bool, + /// Workspace memory-source periodic sync — repos, folders, RSS, web pages. + pub memory_sync: bool, + /// Orchestration relay-mailbox drain supervisor. + pub orchestration: bool, } impl ServiceSet { @@ -73,6 +79,9 @@ impl ServiceSet { harness_init: true, skill_catalog_refresh: true, mcp_boot: true, + integrations: true, + memory_sync: true, + orchestration: true, } } @@ -90,6 +99,9 @@ impl ServiceSet { harness_init: false, skill_catalog_refresh: false, mcp_boot: false, + integrations: false, + memory_sync: false, + orchestration: false, } } @@ -107,6 +119,9 @@ impl ServiceSet { harness_init: false, skill_catalog_refresh: false, mcp_boot: false, + integrations: false, + memory_sync: false, + orchestration: false, } } } @@ -732,11 +747,23 @@ mod tests { assert!(!custom.harness_init); assert!(!custom.skill_catalog_refresh); assert!(!custom.mcp_boot); + assert!(!custom.integrations); + assert!(!custom.memory_sync); + assert!(!custom.orchestration); let desktop = ServiceSet::desktop(); assert!(desktop.memory_queue); assert!(desktop.harness_init); assert!(desktop.skill_catalog_refresh); assert!(desktop.mcp_boot); + assert!(desktop.integrations); + assert!(desktop.memory_sync); + assert!(desktop.orchestration); + + // headless_api() runs no bootstrap jobs either. + let headless = ServiceSet::headless_api(); + assert!(!headless.integrations); + assert!(!headless.memory_sync); + assert!(!headless.orchestration); } } diff --git a/src/core/runtime/services.rs b/src/core/runtime/services.rs index 8249194bd..7dd3bddf2 100644 --- a/src/core/runtime/services.rs +++ b/src/core/runtime/services.rs @@ -234,38 +234,110 @@ pub fn spawn_channels_service() { log::debug!("[channels] channels feature disabled at compile time — not spawning listeners"); } +/// Which bootstrap jobs a given [`ServiceSet`] enables — the single source of +/// truth for the flag→job mapping. +/// +/// Computed by [`bootstrap_job_plan`] (a pure fn) so the wiring can be unit +/// tested without spawning the detached, global-state loops that +/// [`start_bootstrap_jobs`] launches. Each field maps 1:1 to one spawn site. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct BootstrapJobPlan { + /// Memory queue ingestion workers (`memory_queue::start`). + pub memory_queue: bool, + /// Composio periodic connection sync (`composio::start_periodic_sync`). + pub composio_integration_sync: bool, + /// Workspace memory-source periodic sync — repos, folders, RSS, web pages + /// (`memory_sync::workspace::start_workspace_periodic_sync`). + pub workspace_memory_sync: bool, + /// Orchestration relay-mailbox drain supervisor + /// (`orchestration::start_message_drain_supervisor`). + pub orchestration_drain: bool, + /// Proactive task pollers (`task_sources::start_periodic_poll` + + /// `agent::task_dispatcher::start_board_poller`). + pub proactive_task_pollers: bool, +} + +/// Pure flag→job mapping for [`start_bootstrap_jobs`]. No side effects. +/// +/// Note the Composio integration sync AND the one-shot Composio source reconcile +/// both ride `services.integrations` — both no-op without active Composio +/// connections, so they share the one concern flag. `channels` gates NO +/// bootstrap job (its remaining meaning is exactly `spawn_channels_service`). +pub(crate) fn bootstrap_job_plan(services: &ServiceSet) -> BootstrapJobPlan { + BootstrapJobPlan { + memory_queue: services.memory_queue, + composio_integration_sync: services.integrations, + workspace_memory_sync: services.memory_sync, + orchestration_drain: services.orchestration, + proactive_task_pollers: services.cron, + } +} + /// Starts legacy bootstrap loops that predate [`ServiceSet`]. /// /// These are separated from pure subscriber registration so a no-background /// runtime can register handlers first without permanently suppressing a later /// desktop/runtime-with-services boot. +/// +/// Selection is computed once via [`bootstrap_job_plan`], then each job spawns +/// behind its own concern flag. The four non-channel jobs used to ride +/// `services.channels` — a channels-off + memory/integrations-on embedder +/// silently lost all of them (#5028) — so they now sit behind `integrations` / +/// `memory_sync` / `orchestration` instead. pub fn start_bootstrap_jobs(services: ServiceSet, config: &Config) { - if services.memory_queue { + let plan = bootstrap_job_plan(&services); + log::debug!("[runtime.bootstrap] starting bootstrap jobs with plan {plan:?}"); + + if plan.memory_queue { + log::debug!("[runtime.bootstrap] starting memory queue workers"); crate::openhuman::memory_queue::start(config.clone()); } else { - log::debug!("[runtime] memory queue workers disabled by ServiceSet"); + log::debug!("[runtime.bootstrap] memory queue workers disabled by ServiceSet"); } - if services.channels { + // Integrations — Composio periodic connection sync + one-shot source + // reconcile. Both no-op without active Composio connections. + if plan.composio_integration_sync { + log::debug!("[runtime.bootstrap] starting composio integration sync + source reconcile"); crate::openhuman::composio::start_periodic_sync(); - // Workspace-kind memory sources (GitHub repos, folders, RSS, web pages) - // get their own cadence loop; the Composio scheduler only walks - // Composio connections. - crate::openhuman::memory_sync::workspace::start_workspace_periodic_sync(); - crate::openhuman::orchestration::start_message_drain_supervisor(); tokio::spawn(async { + log::debug!("[runtime.bootstrap] composio source reconcile started"); crate::openhuman::memory_sources::reconcile::ensure_composio_sources().await; + log::debug!("[runtime.bootstrap] composio source reconcile completed"); }); } else { - log::debug!("[runtime] bootstrap channel/integration pollers disabled by ServiceSet"); + log::debug!( + "[runtime.bootstrap] composio integration sync + source reconcile disabled by ServiceSet" + ); } - if services.cron { + // Memory sync — workspace-kind memory sources (GitHub repos, folders, RSS, + // web pages) get their own cadence loop; the Composio scheduler above only + // walks Composio connections. + if plan.workspace_memory_sync { + log::debug!("[runtime.bootstrap] starting workspace memory-source periodic sync"); + crate::openhuman::memory_sync::workspace::start_workspace_periodic_sync(); + } else { + log::debug!("[runtime.bootstrap] workspace periodic sync disabled by ServiceSet"); + } + + // Orchestration — relay-mailbox drain supervisor. + if plan.orchestration_drain { + log::debug!("[runtime.bootstrap] starting orchestration message drain supervisor"); + crate::openhuman::orchestration::start_message_drain_supervisor(); + } else { + log::debug!("[runtime.bootstrap] message drain supervisor disabled by ServiceSet"); + } + + if plan.proactive_task_pollers { + log::debug!("[runtime.bootstrap] starting proactive task pollers (task sources + board)"); crate::openhuman::task_sources::start_periodic_poll(); crate::openhuman::agent::task_dispatcher::start_board_poller(); } else { - log::debug!("[runtime] bootstrap proactive task pollers disabled by ServiceSet"); + log::debug!("[runtime.bootstrap] proactive task pollers disabled by ServiceSet"); } + + log::debug!("[runtime.bootstrap] bootstrap job dispatch complete"); } /// Starts one-shot boot background work selected by [`ServiceSet`]. @@ -349,3 +421,120 @@ pub fn spawn_socket_auto_connect( log::debug!("[socket] auto-connect disabled by ServiceSet"); } } + +#[cfg(test)] +mod tests { + use super::{bootstrap_job_plan, BootstrapJobPlan}; + use crate::core::runtime::ServiceSet; + + /// desktop() must enable every bootstrap job — proves the un-bundling kept + /// the desktop job set byte-identical. + #[test] + fn desktop_plan_enables_every_job() { + let plan = bootstrap_job_plan(&ServiceSet::desktop()); + assert_eq!( + plan, + BootstrapJobPlan { + memory_queue: true, + composio_integration_sync: true, + workspace_memory_sync: true, + orchestration_drain: true, + proactive_task_pollers: true, + } + ); + } + + /// none() / headless_api() run no bootstrap job at all. + #[test] + fn job_free_presets_enable_nothing() { + let empty = BootstrapJobPlan { + memory_queue: false, + composio_integration_sync: false, + workspace_memory_sync: false, + orchestration_drain: false, + proactive_task_pollers: false, + }; + assert_eq!(bootstrap_job_plan(&ServiceSet::none()), empty); + assert_eq!(bootstrap_job_plan(&ServiceSet::headless_api()), empty); + } + + /// From none(), flipping exactly one concern flag enables exactly its job + /// and nothing else. + #[test] + fn each_concern_flag_enables_exactly_its_job() { + let mut integrations = ServiceSet::none(); + integrations.integrations = true; + let plan = bootstrap_job_plan(&integrations); + assert!(plan.composio_integration_sync); + assert!(!plan.workspace_memory_sync); + assert!(!plan.orchestration_drain); + assert!(!plan.memory_queue); + assert!(!plan.proactive_task_pollers); + + let mut memory_sync = ServiceSet::none(); + memory_sync.memory_sync = true; + let plan = bootstrap_job_plan(&memory_sync); + assert!(plan.workspace_memory_sync); + assert!(!plan.composio_integration_sync); + assert!(!plan.orchestration_drain); + + let mut orchestration = ServiceSet::none(); + orchestration.orchestration = true; + let plan = bootstrap_job_plan(&orchestration); + assert!(plan.orchestration_drain); + assert!(!plan.composio_integration_sync); + assert!(!plan.workspace_memory_sync); + } + + /// From desktop(), disabling exactly one concern flag disables only its job. + #[test] + fn disabling_one_concern_disables_only_its_job() { + let mut services = ServiceSet::desktop(); + services.integrations = false; + let plan = bootstrap_job_plan(&services); + assert!(!plan.composio_integration_sync); + assert!(plan.workspace_memory_sync); + assert!(plan.orchestration_drain); + assert!(plan.memory_queue); + assert!(plan.proactive_task_pollers); + + let mut services = ServiceSet::desktop(); + services.memory_sync = false; + let plan = bootstrap_job_plan(&services); + assert!(!plan.workspace_memory_sync); + assert!(plan.composio_integration_sync); + assert!(plan.orchestration_drain); + + let mut services = ServiceSet::desktop(); + services.orchestration = false; + let plan = bootstrap_job_plan(&services); + assert!(!plan.orchestration_drain); + assert!(plan.composio_integration_sync); + assert!(plan.workspace_memory_sync); + } + + /// The #5028 regression: `channels` gates NO bootstrap job. Turning channels + /// on by itself must enable zero sync jobs, and turning channels off while + /// the new flags stay on must lose nothing. + #[test] + fn channels_flag_gates_no_bootstrap_job() { + // channels=true alone → zero sync jobs. + let mut channels_only = ServiceSet::none(); + channels_only.channels = true; + let plan = bootstrap_job_plan(&channels_only); + assert_eq!( + plan, + bootstrap_job_plan(&ServiceSet::none()), + "channels alone must enable no bootstrap job" + ); + + // channels=false with every new flag on → identical to desktop's plan. + let mut channels_off = ServiceSet::desktop(); + channels_off.channels = false; + assert_eq!( + bootstrap_job_plan(&channels_off), + bootstrap_job_plan(&ServiceSet::desktop()), + "dropping channels must not drop any bootstrap job" + ); + } +}