refactor(runtime): un-bundle non-channel bootstrap jobs from ServiceSet.channels (#5028) (#5079)

.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

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);
     }
 }

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"
+        );
+    }
+}