From 5ef15425938042e715d22bbd9b41500a71214f04 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Sat, 31 Oct 2026 15:27:32 +0800 Subject: [PATCH 1/5] fix(remote-connect): stop listing the bot host as its own remote device The IM bot device menu added a synthetic local entry and then every online account device, including the host serving the bot, so that host appeared twice. Fold the relay entry whose id matches the local device identity into the local entry, and return a chat that had selected its own host as a remote device to the local target. --- .../remote_connect/bot/command_router.rs | 95 +++++++++++++++++-- 1 file changed, 86 insertions(+), 9 deletions(-) diff --git a/src/crates/assembly/core/src/service/remote_connect/bot/command_router.rs b/src/crates/assembly/core/src/service/remote_connect/bot/command_router.rs index d696769869..69b944beb4 100644 --- a/src/crates/assembly/core/src/service/remote_connect/bot/command_router.rs +++ b/src/crates/assembly/core/src/service/remote_connect/bot/command_router.rs @@ -17,7 +17,7 @@ use crate::service_agent_runtime::{ remote_opened_workspace_catalog, remote_workspace_display_name, remote_workspace_metadata, }; -use log::{error, info}; +use log::{error, info, warn}; use serde_json::Value; use std::sync::{Arc, OnceLock}; @@ -880,21 +880,57 @@ async fn list_devices(state: &mut BotChatState, s: &'static BotStrings) -> Handl } }; - // Build a selectable list: "0. Local" + each online device. - let online: Vec<_> = devices.into_iter().filter(|d| d.online).collect(); - - let mut options: Vec<(String, String)> = - vec![("local".to_string(), s.devices_local.to_string())]; - - for d in &online { - options.push((d.device_id.clone(), d.device_name.clone())); + // The relay lists every account host, including the one serving this bot. + let local_device_id = crate::service::remote_connect::DeviceIdentity::from_current_machine() + .map(|identity| identity.device_id) + .map_err(|e| warn!("Bot list_devices could not resolve the local device id: {e}")) + .ok(); + if let (Some(local_id), Some(active)) = ( + local_device_id.as_deref(), + state.active_remote_device.as_ref(), + ) { + if active.device_id == local_id { + info!("Bot active device is this host; switching back to the local target"); + state.select_local_device(); + } } + let options = device_menu_options(local_device_id.as_deref(), devices, s); let view = device_selection_view(state, &options, s); state.set_pending(PendingAction::SelectDevice { options }); result_from_menu(state, view) } +/// Build the selectable device list: the local host first, then every other +/// online account device. The relay entry for this host is folded into the +/// local entry so the same machine is never offered twice. +fn device_menu_options( + local_device_id: Option<&str>, + devices: Vec, + s: &'static BotStrings, +) -> Vec<(String, String)> { + let mut local_label = s.devices_local.to_string(); + let mut remote = Vec::new(); + for d in devices { + if local_device_id == Some(d.device_id.as_str()) { + let name = d + .device_alias + .as_deref() + .map(str::trim) + .filter(|alias| !alias.is_empty()) + .unwrap_or(d.device_name.as_str()); + if !name.trim().is_empty() { + local_label = format!("{} ({})", s.devices_local, name.trim()); + } + } else if d.online { + remote.push((d.device_id, d.device_name)); + } + } + let mut options = vec![("local".to_string(), local_label)]; + options.extend(remote); + options +} + fn device_selection_view( state: &BotChatState, options: &[(String, String)], @@ -4401,6 +4437,47 @@ mod menu_tests { assert!(text.contains("/menu"), "{text}"); } + #[test] + fn device_menu_folds_this_host_into_the_local_entry() { + let s = strings_for(BotLanguage::ZhCN); + let device = |id: &str, name: &str, online: bool| { + serde_json::from_value::( + serde_json::json!({ + "device_id": id, + "device_name": name, + "online": online, + "last_seen_at": 0, + }), + ) + .expect("device info") + }; + let devices = vec![ + device("self-id", "ecs-openbitfun-personal-lwb", true), + device("mac-id", "MacBook-Pro-2.local", true), + device("offline-id", "old-laptop", false), + ]; + + let options = device_menu_options(Some("self-id"), devices.clone(), s); + assert_eq!( + options, + vec![ + ( + "local".to_string(), + format!("{} (ecs-openbitfun-personal-lwb)", s.devices_local) + ), + ("mac-id".to_string(), "MacBook-Pro-2.local".to_string()), + ] + ); + + // Without a resolvable local id nothing is hidden. + let options = device_menu_options(None, devices, s); + assert_eq!(options.len(), 3); + assert_eq!( + options[0], + ("local".to_string(), s.devices_local.to_string()) + ); + } + #[test] fn device_menu_plain_text_is_reconstructable_without_ids_or_duplicates() { let mut state = BotChatState::new("c".into()); From 9f4ce03c53a5461607496eb8611973cb168a6979 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Sat, 31 Oct 2026 15:27:32 +0800 Subject: [PATCH 2/5] feat(config): reload app.json edits made outside the running host Long-running hosts (CLI daemon, desktop, server) kept the configuration loaded at startup until an in-process reload, so hand edits or another process's writes needed a restart. Track a fingerprint of the content the config manager last loaded or wrote and poll the file every 2 seconds; reload and broadcast only when another writer changed it. Invalid mid-edit content keeps the current configuration, a missing file is never recreated, and reloads do not signal local edits to account sync. --- .../core/src/service/config/global.rs | 43 ++++++++ .../core/src/service/config/manager.rs | 37 +++++++ .../core/src/service/config/service.rs | 103 ++++++++++++++++++ 3 files changed, 183 insertions(+) diff --git a/src/crates/assembly/core/src/service/config/global.rs b/src/crates/assembly/core/src/service/config/global.rs index 412a4c6cd3..e5633a2995 100644 --- a/src/crates/assembly/core/src/service/config/global.rs +++ b/src/crates/assembly/core/src/service/config/global.rs @@ -109,6 +109,9 @@ impl GlobalConfigManager { #[cfg(feature = "model-catalog")] crate::infrastructure::ai::reasoning_catalog::prepare_model_catalog(); + #[cfg(feature = "agent-runtime")] + spawn_config_file_reload_poller(Arc::clone(&config_service)); + info!("Global config service initialized"); Ok(()) @@ -214,6 +217,46 @@ impl GlobalConfigManager { } } +/// Interval between checks of the persisted config file for edits made by +/// another writer (a hand edit, or another OpenBitFun process on this host). +#[cfg(feature = "agent-runtime")] +const CONFIG_FILE_RELOAD_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(2); + +/// Long-running hosts (CLI daemon, desktop, server) otherwise keep serving the +/// configuration they loaded at startup until something reloads in-process. +/// Polling instead of a file-system watch keeps this working for atomic +/// replacement, network home directories, and platforms without inotify. +#[cfg(feature = "agent-runtime")] +fn spawn_config_file_reload_poller(service: Arc) { + let Ok(runtime) = tokio::runtime::Handle::try_current() else { + log::warn!("No async runtime available; config file reload polling is disabled"); + return; + }; + runtime.spawn(async move { + let mut interval = tokio::time::interval(CONFIG_FILE_RELOAD_POLL_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + interval.tick().await; + let mut last_error: Option = None; + loop { + interval.tick().await; + match service.reload_if_changed_on_disk().await { + Ok(_) => last_error = None, + Err(error) => { + // An editor may be mid-write; keep the current config and + // report each distinct failure once. + let message = error.to_string(); + if last_error.as_deref() != Some(message.as_str()) { + log::warn!( + "Config file changed on disk but could not be loaded: {message}" + ); + last_error = Some(message); + } + } + } + } + }); +} + /// Convenience helper: get the global configuration service. pub async fn get_global_config_service() -> OpenBitFunResult> { GlobalConfigManager::get_service().await diff --git a/src/crates/assembly/core/src/service/config/manager.rs b/src/crates/assembly/core/src/service/config/manager.rs index b0308b65fd..a942dea38e 100644 --- a/src/crates/assembly/core/src/service/config/manager.rs +++ b/src/crates/assembly/core/src/service/config/manager.rs @@ -215,6 +215,17 @@ pub struct ConfigManager { backup_count: usize, load_diagnostics: Vec, account_guard_sources: HashMap, + /// Fingerprint of the file content this manager last loaded or wrote, so + /// an on-disk change by another writer can be told apart from its own. + disk_fingerprint: std::sync::Mutex>, +} + +/// Content fingerprint used to detect config files replaced by another writer. +pub(crate) fn config_content_fingerprint(content: &[u8]) -> u64 { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + content.hash(&mut hasher); + hasher.finish() } struct ConfigCandidate { @@ -266,6 +277,7 @@ impl ConfigManager { backup_count, load_diagnostics: Vec::new(), account_guard_sources: HashMap::new(), + disk_fingerprint: std::sync::Mutex::new(None), }; manager.load_or_create_config().await?; @@ -285,6 +297,27 @@ impl ConfigManager { &self.path_manager } + /// Persisted config file owned by this manager. + pub(crate) fn config_file(&self) -> &std::path::Path { + &self.config_file + } + + /// Fingerprint of the content this manager last loaded or wrote. + pub(crate) fn disk_fingerprint(&self) -> Option { + *self + .disk_fingerprint + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn remember_disk_content(&self, content: &str) { + *self + .disk_fingerprint + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = + Some(config_content_fingerprint(content.as_bytes())); + } + /// Reloads from the same storage root while the service holds its write /// lock, so a concurrent save cannot be replaced by an older disk read. pub(crate) async fn reload(&mut self) -> OpenBitFunResult<()> { @@ -459,6 +492,9 @@ impl ConfigManager { "Failed to persist model selection migration: {error}" )) })?; + self.remember_disk_content(&content); + } else { + self.remember_disk_content(&content); } self.config = config; self.load_diagnostics = diagnostics; @@ -612,6 +648,7 @@ impl ConfigManager { self.config_file, e )) })?; + self.remember_disk_content(&content); Ok(account_guard_sources(&value)) } diff --git a/src/crates/assembly/core/src/service/config/service.rs b/src/crates/assembly/core/src/service/config/service.rs index f8871d291b..5c88301778 100644 --- a/src/crates/assembly/core/src/service/config/service.rs +++ b/src/crates/assembly/core/src/service/config/service.rs @@ -565,6 +565,54 @@ impl ConfigService { Ok(()) } + /// Reloads when another writer (a hand edit, another OpenBitFun process) + /// replaced the config file since this service last loaded or wrote it. + /// + /// A missing file is left alone rather than recreated, and a file that + /// fails to load keeps the current in-memory configuration. Returns + /// whether a reload happened. + pub async fn reload_if_changed_on_disk(&self) -> OpenBitFunResult { + let (config_file, known) = { + let manager = self.manager.read().await; + ( + manager.config_file().to_path_buf(), + manager.disk_fingerprint(), + ) + }; + let Some(on_disk) = read_config_fingerprint(&config_file).await? else { + return Ok(false); + }; + if known == Some(on_disk) { + return Ok(false); + } + + { + let mut manager = self.manager.write().await; + // A local save may have landed between the check and the lock. + match read_config_fingerprint(&config_file).await? { + Some(on_disk) if manager.disk_fingerprint() != Some(on_disk) => {} + _ => return Ok(false), + } + manager.reload().await?; + } + + info!( + "Configuration reloaded after an on-disk change: path={}", + config_file.display() + ); + super::global::GlobalConfigManager::broadcast_update( + super::global::ConfigUpdateEvent::ModelConfigurationUpdated, + ) + .await; + super::global::GlobalConfigManager::broadcast_update( + super::global::ConfigUpdateEvent::ConfigReloaded, + ) + .await; + #[cfg(feature = "web-tools")] + self.refresh_web_search_runtime().await; + Ok(true) + } + /// Creates a configuration backup. pub async fn create_backup(&self) -> OpenBitFunResult { let manager = self.manager.read().await; @@ -834,6 +882,17 @@ impl ReconcileModelsReport { } } +async fn read_config_fingerprint(path: &std::path::Path) -> OpenBitFunResult> { + match tokio::fs::read(path).await { + Ok(content) => Ok(Some(super::manager::config_content_fingerprint(&content))), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(OpenBitFunError::config(format!( + "Failed to read config file {}: {error}", + path.display() + ))), + } +} + #[cfg(test)] mod tests { use super::*; @@ -1066,6 +1125,50 @@ mod tests { .contains_key("ResearchSpecialist")); } + #[tokio::test] + async fn reload_if_changed_on_disk_picks_up_external_edits_only() { + let (service, dir) = test_service("reload-on-disk-change").await; + service + .add_ai_model(runtime_model("kept", "fixture-key")) + .await + .unwrap(); + // The service's own write is not an external change. + assert!(!service.reload_if_changed_on_disk().await.unwrap()); + + let config_file = dir + .path() + .join("reload-on-disk-change") + .join("config") + .join("app.json"); + let mut value: serde_json::Value = + serde_json::from_str(&std::fs::read_to_string(&config_file).unwrap()).unwrap(); + value["ai"]["models"] + .as_array_mut() + .unwrap() + .push(serde_json::to_value(runtime_model("added", "fixture-key")).unwrap()); + std::fs::write(&config_file, serde_json::to_string_pretty(&value).unwrap()).unwrap(); + + let mut changes = service.subscribe_local_changes(); + changes.borrow_and_update(); + assert!(service.reload_if_changed_on_disk().await.unwrap()); + let config: GlobalConfig = service.get_config(None).await.unwrap(); + assert!(config.ai.models.iter().any(|model| model.id == "added")); + // A reload is not a local edit and must not echo into account sync. + assert!(!changes.has_changed().unwrap()); + assert!(!service.reload_if_changed_on_disk().await.unwrap()); + + // A half-written edit keeps the loaded configuration. + std::fs::write(&config_file, "{\"ai\": ").unwrap(); + assert!(service.reload_if_changed_on_disk().await.is_err()); + let config: GlobalConfig = service.get_config(None).await.unwrap(); + assert!(config.ai.models.iter().any(|model| model.id == "added")); + + // A deleted file is neither reloaded nor recreated. + std::fs::remove_file(&config_file).unwrap(); + assert!(!service.reload_if_changed_on_disk().await.unwrap()); + assert!(!config_file.exists()); + } + #[tokio::test] async fn model_default_noop_does_not_persist_or_notify() { let (service, dir) = test_service("model-default-noop").await; From 6be14f664ceb869c019806ff334561f5fabbf397 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Sat, 31 Oct 2026 15:27:32 +0800 Subject: [PATCH 3/5] feat(cli): dispatch jobs to other hosts with dispatch outbound The CLI only implemented the target side of Detached Dispatch. Add `openbitfun dispatch outbound targets|submit|status|list|cancel`, a JSON controller over the Core dispatch owner for saved SSH connections and same-account devices, excluding this host. Move the account-device host_invoke envelope and agent-identity translation into Core (AccountRuntimeDeviceDispatchRpc) so Desktop and CLI share one implementation, and admit dispatch-store into the reviewed CLI Core closure. --- docs/architecture/detached-task-dispatch.md | 5 + .../cargo-dependency-boundaries.mjs | 10 +- src/apps/cli/AGENTS.md | 5 +- src/apps/cli/Cargo.toml | 1 + src/apps/cli/README.md | 17 + src/apps/cli/src/dispatch/mod.rs | 1 + src/apps/cli/src/dispatch/outbound.rs | 479 ++++++++++++++++++ src/apps/cli/src/main.rs | 44 +- src/apps/cli/src/root_handlers.rs | 1 + src/apps/desktop/src/api/dispatch_api.rs | 59 +-- .../service/dispatch/account_device_rpc.rs | 143 ++++++ .../assembly/core/src/service/dispatch/mod.rs | 15 + 12 files changed, 726 insertions(+), 54 deletions(-) create mode 100644 src/apps/cli/src/dispatch/outbound.rs create mode 100644 src/crates/assembly/core/src/service/dispatch/account_device_rpc.rs diff --git a/docs/architecture/detached-task-dispatch.md b/docs/architecture/detached-task-dispatch.md index 7e3452a3ca..612bd4911a 100644 --- a/docs/architecture/detached-task-dispatch.md +++ b/docs/architecture/detached-task-dispatch.md @@ -22,6 +22,11 @@ There are three roles: - A **transport adapter** moves the same narrow JSON protocol over SSH or an end-to-end encrypted account-device RPC. +Desktop (`dispatch_*` commands) and the CLI (`openbitfun dispatch outbound`) are +both controllers over the same Core controller. The account-device `host_invoke` +envelope is shared through `AccountRuntimeDeviceDispatchRpc` and its +encode/decode helpers, so the two hosts cannot drift. + The controller is never a runtime or filesystem proxy for a non-local job. It stores only an outbound observer record under `~/.openbitfun/dispatch/outbound/`; it must not create the target session in the diff --git a/scripts/core-boundaries/cargo-dependency-boundaries.mjs b/scripts/core-boundaries/cargo-dependency-boundaries.mjs index b1a4b46d9b..92267963f0 100644 --- a/scripts/core-boundaries/cargo-dependency-boundaries.mjs +++ b/scripts/core-boundaries/cargo-dependency-boundaries.mjs @@ -1469,6 +1469,8 @@ export function findProductEntrypointCoreFeatureViolations( 'plugin-runtime', 'opencode-plugin-host', 'ssh-remote', + // Outbound controller for `openbitfun dispatch outbound`. + 'dispatch-store', ]], ['openbitfun-acp', [...new Set([...acpClientCoreFeatures, ...acpServerCoreFeatures])]], ['openbitfun-app-server', [ @@ -1536,6 +1538,7 @@ export function findProductEntrypointCoreFeatureViolations( 'product-search', 'opencode-plugin-host', 'remote-connect', + 'dispatch-store', ]], ['openbitfun-acp', acpActiveCoreFeatures], ['openbitfun-sdk-host-app', coreCompatibilityActiveFeatures], @@ -1787,9 +1790,10 @@ export function findProductEntrypointCoreFeatureViolations( const corePackage = packages.find((pkg) => pkg.name === 'openbitfun-core'); if (corePackage) { - const forbiddenCoreFeatures = [ + const forbiddenCoreFeaturesFor = (rootName) => [ 'product-full', - 'dispatch-store', + // The CLI is a reviewed dispatch controller; other closures are not. + ...(rootName === 'openbitfun-cli' ? [] : ['dispatch-store']), ]; const reportedUnexpectedFeatures = new Set(); @@ -1930,7 +1934,7 @@ export function findProductEntrypointCoreFeatureViolations( if (targetPackage.name === 'openbitfun-core') { const activeCoreFeatures = targetState.featureState.active; - const unexpected = forbiddenCoreFeatures.find((feature) => + const unexpected = forbiddenCoreFeaturesFor(rootName).find((feature) => activeCoreFeatures.has(feature)) ?? [...activeCoreFeatures] .filter((feature) => !allowedCoreFeatures.has(feature)) diff --git a/src/apps/cli/AGENTS.md b/src/apps/cli/AGENTS.md index c85ee00d4c..e1280cda31 100644 --- a/src/apps/cli/AGENTS.md +++ b/src/apps/cli/AGENTS.md @@ -112,8 +112,9 @@ restrictions remain enforced. - Assemble CLI through `DeliveryProfile::Cli` and validated product Runtime parts. Hiding a command is not a backend capability restriction. - The CLI selects the reviewed `openbitfun-core` owner-feature closure - (`agent-runtime`, `external-sources`, `plugin-runtime`, `remote-connect`, and - `ssh-remote`) plus the Code Agent atomic tool owners and the independent + (`agent-runtime`, `external-sources`, `plugin-runtime`, `remote-connect`, + `ssh-remote`, and `dispatch-store` for the `dispatch outbound` controller) + plus the Code Agent atomic tool owners and the independent `tools-pages` owner. Pages binds to the executing host's AccountRuntime; controller-local account state must not enable remote tools. It must not register DeepReview, DeepResearch, MiniApp, or Canvas agents/tools. Do not replace the diff --git a/src/apps/cli/Cargo.toml b/src/apps/cli/Cargo.toml index 69c5481202..f42521a3b5 100644 --- a/src/apps/cli/Cargo.toml +++ b/src/apps/cli/Cargo.toml @@ -41,6 +41,7 @@ openbitfun-core = { path = "../../crates/assembly/core", features = [ "plugin-runtime", "opencode-plugin-host", "ssh-remote", + "dispatch-store", "tools-pages", "tools-basic", "tools-git", diff --git a/src/apps/cli/README.md b/src/apps/cli/README.md index 3c3ff7fb63..a1f4f28fd3 100644 --- a/src/apps/cli/README.md +++ b/src/apps/cli/README.md @@ -281,6 +281,23 @@ protocol version before submitting or inspecting jobs. See the [detached task architecture](../../../docs/architecture/detached-task-dispatch.md) for the transport and workspace-snapshot contract. +To dispatch from this machine, use the controller commands. They print JSON and +work from any Git checkout: + +```bash +openbitfun dispatch outbound targets +openbitfun dispatch outbound submit --target "MacBook-Pro-2.local" \ + --approval-policy reject-and-report "Run the integration tests and fix failures" +openbitfun dispatch outbound status --cursor 0 +openbitfun dispatch outbound list +openbitfun dispatch outbound cancel +``` + +`--target` accepts an account device name or alias, `device:`, or +`ssh:`. Device targets need this machine to be signed in to the +same account; an offline device is an error, never a local fallback. The target +works in its own worktree branched from `--base-ref` (default `HEAD`). + The `remote` approval policy lets a controller answer pending tool permissions. Interactive questions from `AskUserQuestion` are unavailable for dispatch jobs; provide required task choices in the prompt instead of waiting for a question. diff --git a/src/apps/cli/src/dispatch/mod.rs b/src/apps/cli/src/dispatch/mod.rs index f3c8b04fac..36927d59f7 100644 --- a/src/apps/cli/src/dispatch/mod.rs +++ b/src/apps/cli/src/dispatch/mod.rs @@ -1,3 +1,4 @@ +pub(crate) mod outbound; mod permissions; pub(crate) mod protocol; mod query_file; diff --git a/src/apps/cli/src/dispatch/outbound.rs b/src/apps/cli/src/dispatch/outbound.rs new file mode 100644 index 0000000000..45abe9fa94 --- /dev/null +++ b/src/apps/cli/src/dispatch/outbound.rs @@ -0,0 +1,479 @@ +//! Controller side of Detached Dispatch for the CLI. +//! +//! `openbitfun dispatch outbound ...` submits jobs to another OpenBitFun host and +//! observes them. The existing `openbitfun dispatch ` commands are the +//! target protocol and stay untouched. The controller semantics (baseline, +//! provisioning, observer records) live in the Core dispatch owner; this module +//! only resolves the CLI's SSH and account-device transports and prints JSON. + +use std::sync::Arc; + +use anyhow::{anyhow, bail, Context, Result}; +use clap::{Subcommand, ValueEnum}; +use openbitfun_core::infrastructure::try_get_path_manager_arc; +use openbitfun_core::service::dispatch::{ + cancel_device_dispatch, cancel_dispatch, get_device_dispatch_status, get_dispatch_status, + list_device_dispatch_jobs, list_dispatch_jobs, list_dispatch_targets, submit_device_dispatch, + submit_dispatch, AccountRuntimeDeviceDispatchRpc, DispatchJobRequest, DispatchListJobsRequest, + DispatchListTargetsRequest, DispatchStatusRequest, DispatchSubmitRequest, DispatchTarget, + DispatchTargetOption, DispatchTargetRequest, OutboundDispatchStore, +}; +use openbitfun_core::service::remote_connect::account_runtime::{AccountDevice, AccountRuntime}; +use openbitfun_core::service::remote_connect::DeviceIdentity; +use openbitfun_core::service::remote_ssh::SSHConnectionManager; +use serde_json::Value; + +const MAX_OUTBOUND_PROMPT_BYTES: u64 = 32 * 1024; + +#[derive(Subcommand, Debug)] +pub(crate) enum OutboundAction { + /// List SSH connections and online account devices this machine can dispatch to + Targets, + /// Submit a detached job to another host and print its acknowledgement + /// + /// The job runs in a worktree on the target, branched from the current Git + /// checkout. This machine may disconnect after submission. + Submit { + /// Target: an account device name, `device:`, or `ssh:` + #[arg(long)] + target: String, + /// How the target answers tool permission prompts while unattended + #[arg(long, value_enum)] + approval_policy: OutboundApprovalPolicy, + /// Task prompt. If omitted, stdin is used when piped. + prompt: Option, + /// Agent type on the target + #[arg(short, long, default_value = "Standard")] + agent: String, + /// Model id on the target; defaults to the target's own default model + #[arg(long)] + model: Option, + /// Revision the target worktree starts from + #[arg(long, default_value = "HEAD")] + base_ref: String, + /// Carry this checkout's uncommitted changes to the target + #[arg(long)] + include_uncommitted: bool, + /// Short job title shown in job lists + #[arg(long)] + title: Option, + /// Git checkout to dispatch from; defaults to the current directory + #[arg(long)] + workspace: Option, + }, + /// Read a job's state and events after a cursor + Status { + job_id: String, + /// Event cursor returned by a previous status call + #[arg(long, default_value_t = 0)] + cursor: u64, + }, + /// List jobs submitted from this machine, or the jobs a target owns + List { + /// Target to ask directly (same forms as `submit --target`) + #[arg(long)] + target: Option, + }, + /// Cancel a submitted job + Cancel { job_id: String }, +} + +#[derive(Clone, Copy, Debug, ValueEnum)] +pub(crate) enum OutboundApprovalPolicy { + /// Approve every tool request that static rules do not deny + Auto, + /// Reject permission prompts and report them in the job result + RejectAndReport, +} + +impl OutboundApprovalPolicy { + fn wire(self) -> &'static str { + match self { + Self::Auto => "auto", + Self::RejectAndReport => "reject-and-report", + } + } +} + +pub(crate) async fn run(action: OutboundAction) -> Result<()> { + let result = run_action(action).await; + match result { + Ok(value) => { + println!("{}", serde_json::to_string_pretty(&value)?); + Ok(()) + } + Err(error) => { + println!( + "{}", + serde_json::to_string_pretty( + &serde_json::json!({ "error": format!("{error:#}") }) + )? + ); + Err(error) + } + } +} + +async fn run_action(action: OutboundAction) -> Result { + openbitfun_core::service::config::initialize_global_config() + .await + .context("initialize global config")?; + let path_manager = try_get_path_manager_arc().context("resolve OpenBitFun data directory")?; + let store = OutboundDispatchStore::new(path_manager.as_ref()); + let context = OutboundContext::default(); + + match action { + OutboundAction::Targets => { + let mut targets = list_dispatch_targets( + &context.ssh_manager().await?, + DispatchListTargetsRequest::default(), + ) + .await?; + targets.retain(|target| target.kind != "local"); + match context.account_devices().await { + Ok(devices) => { + targets.extend(devices.into_iter().map(|device| DispatchTargetOption { + kind: "device".to_string(), + connection_id: None, + display_name: device_label(&device), + device_id: Some(device.device_id), + description: None, + online: Some(device.online), + })) + } + Err(error) => { + tracing::warn!("Account devices are unavailable for dispatch: {error:#}") + } + } + Ok(serde_json::to_value(targets)?) + } + OutboundAction::Submit { + target, + approval_policy, + prompt, + agent, + model, + base_ref, + include_uncommitted, + title, + workspace, + } => { + let prompt = resolve_prompt(prompt)?; + let workspace = match workspace { + Some(path) => path, + None => std::env::current_dir().context("resolve current directory")?, + }; + let workspace = workspace + .canonicalize() + .with_context(|| format!("resolve workspace {}", workspace.display()))?; + let source_workspace_path = workspace + .to_str() + .ok_or_else(|| anyhow!("workspace path is not valid UTF-8"))? + .to_string(); + let resolved = context.resolve_target(&target).await?; + let request = DispatchSubmitRequest { + target: resolved.request(), + base_ref: Some(base_ref), + include_uncommitted, + job_id: uuid::Uuid::new_v4().to_string(), + session_id: uuid::Uuid::new_v4().to_string(), + agent_type: openbitfun_core_types::agent_identity::canonical_agent_id(&agent) + .to_string(), + prompt, + approval_policy: approval_policy.wire().to_string(), + model, + reasoning_preset: None, + title, + source_workspace_path: Some(source_workspace_path), + source_workspace_id: None, + attachments: Vec::new(), + }; + match resolved { + ResolvedTarget::Device { display_name, .. } => { + let rpc = context.device_rpc().await?; + submit_device_dispatch(&rpc, &store, display_name, request).await + } + ResolvedTarget::Ssh { .. } => { + submit_dispatch(&context.ssh_manager().await?, &store, request).await + } + } + } + OutboundAction::Status { job_id, cursor } => { + let request = DispatchStatusRequest { job_id, cursor }; + if job_targets_device(&store, &request.job_id).await? { + get_device_dispatch_status(&context.device_rpc().await?, &store, request).await + } else { + get_dispatch_status(&context.ssh_manager().await?, &store, request).await + } + } + OutboundAction::Cancel { job_id } => { + let request = DispatchJobRequest { job_id }; + if job_targets_device(&store, &request.job_id).await? { + cancel_device_dispatch(&context.device_rpc().await?, &store, request).await + } else { + cancel_dispatch(&context.ssh_manager().await?, &store, request).await + } + } + OutboundAction::List { target } => { + let Some(target) = target else { + return list_dispatch_jobs( + &context.ssh_manager().await?, + &store, + DispatchListJobsRequest { target: None }, + ) + .await; + }; + let resolved = context.resolve_target(&target).await?; + let request = DispatchListJobsRequest { + target: Some(resolved.request()), + }; + match resolved { + ResolvedTarget::Device { display_name, .. } => { + let rpc = context.device_rpc().await?; + list_device_dispatch_jobs(&rpc, &store, display_name, request).await + } + ResolvedTarget::Ssh { .. } => { + list_dispatch_jobs(&context.ssh_manager().await?, &store, request).await + } + } + } + } +} + +async fn job_targets_device(store: &OutboundDispatchStore, job_id: &str) -> Result { + let record = store + .get(job_id) + .await? + .ok_or_else(|| anyhow!("No outbound dispatch job {job_id} was submitted from here"))?; + Ok(matches!(record.target, DispatchTarget::Device { .. })) +} + +fn resolve_prompt(prompt: Option) -> Result { + use std::io::{IsTerminal, Read}; + let prompt = match prompt { + Some(prompt) => prompt, + None => { + let stdin = std::io::stdin(); + if stdin.is_terminal() { + bail!("Provide a prompt argument or pipe it on stdin"); + } + let mut raw = String::new(); + stdin + .take(MAX_OUTBOUND_PROMPT_BYTES + 1) + .read_to_string(&mut raw) + .context("read dispatch prompt from stdin")?; + raw + } + }; + if prompt.trim().is_empty() { + bail!("Dispatch prompt cannot be empty"); + } + Ok(prompt) +} + +#[derive(Debug, Clone, PartialEq, Eq)] +enum ResolvedTarget { + Device { + device_id: String, + display_name: String, + }, + Ssh { + connection_id: String, + }, +} + +impl ResolvedTarget { + fn request(&self) -> DispatchTargetRequest { + // The target chooses where its worktree lands. + match self { + Self::Device { device_id, .. } => DispatchTargetRequest::Device { + device_id: device_id.clone(), + workspace_path: String::new(), + }, + Self::Ssh { connection_id } => DispatchTargetRequest::Ssh { + connection_id: connection_id.clone(), + workspace_path: String::new(), + }, + } + } +} + +fn device_label(device: &AccountDevice) -> String { + device + .device_alias + .as_deref() + .map(str::trim) + .filter(|alias| !alias.is_empty()) + .unwrap_or(&device.device_name) + .to_string() +} + +/// Resolve a `device:` selector or a device name/alias among the account's +/// other devices. Names must match exactly one device. +fn resolve_device_selector(selector: &str, devices: &[AccountDevice]) -> Result { + let selector = selector.trim(); + let matches: Vec<&AccountDevice> = match selector.strip_prefix("device:") { + Some(device_id) => devices + .iter() + .filter(|device| device.device_id == device_id.trim()) + .collect(), + None => { + let by_id: Vec<_> = devices + .iter() + .filter(|device| device.device_id == selector) + .collect(); + if by_id.is_empty() { + devices + .iter() + .filter(|device| { + device.device_name.eq_ignore_ascii_case(selector) + || device + .device_alias + .as_deref() + .is_some_and(|alias| alias.trim().eq_ignore_ascii_case(selector)) + }) + .collect() + } else { + by_id + } + } + }; + match matches.as_slice() { + [device] if device.online => Ok(ResolvedTarget::Device { + device_id: device.device_id.clone(), + display_name: device_label(device), + }), + [_] => bail!("Dispatch device {selector} is offline; no local fallback was attempted"), + [] => bail!( + "No other device on this account matches {selector}; run `dispatch outbound targets`" + ), + _ => bail!("Several account devices match {selector}; use device:"), + } +} + +#[derive(Default)] +struct OutboundContext { + account: tokio::sync::OnceCell>, +} + +impl OutboundContext { + async fn ssh_manager(&self) -> Result { + let state = + openbitfun_core::service::remote_ssh::workspace_state::ensure_saved_connection_services() + .await + .map_err(|error| anyhow!("SSH services are unavailable: {error}"))?; + state + .get_ssh_manager() + .await + .ok_or_else(|| anyhow!("SSH manager is unavailable")) + } + + async fn account(&self) -> Result> { + let account = self + .account + .get_or_init(|| async { + let account = crate::account::build_management_account_runtime(); + account.try_restore_session().await; + account + }) + .await + .clone(); + if !account.is_logged_in().await { + bail!("This machine is not signed in to an account; device dispatch is unavailable"); + } + Ok(account) + } + + async fn device_rpc(&self) -> Result { + Ok(AccountRuntimeDeviceDispatchRpc::new(self.account().await?)) + } + + /// Account devices other than this machine. + async fn account_devices(&self) -> Result> { + let devices = self.account().await?.list_devices().await?; + let local_device_id = DeviceIdentity::from_current_machine() + .map(|identity| identity.device_id) + .ok(); + Ok(devices + .into_iter() + .filter(|device| local_device_id.as_deref() != Some(device.device_id.as_str())) + .collect()) + } + + async fn resolve_target(&self, selector: &str) -> Result { + if let Some(connection_id) = selector.trim().strip_prefix("ssh:") { + let connection_id = connection_id.trim(); + if connection_id.is_empty() { + bail!("ssh: targets need a saved connection id"); + } + return Ok(ResolvedTarget::Ssh { + connection_id: connection_id.to_string(), + }); + } + resolve_device_selector(selector, &self.account_devices().await?) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn device(id: &str, name: &str, alias: Option<&str>, online: bool) -> AccountDevice { + AccountDevice { + device_id: id.to_string(), + device_name: name.to_string(), + device_kind: Some("cli".to_string()), + device_alias: alias.map(str::to_string), + device_model: None, + device_os: None, + device_os_version: None, + device_client_version: None, + device_client_protocol: None, + compatible: None, + online, + } + } + + #[test] + fn device_selector_resolves_ids_names_and_aliases() { + let devices = vec![ + device("mac-id", "MacBook-Pro-2.local", Some("Office Mac"), true), + device("ecs-id", "ecs-box", None, true), + device("old-id", "old-laptop", None, false), + ]; + let expected = ResolvedTarget::Device { + device_id: "mac-id".to_string(), + display_name: "Office Mac".to_string(), + }; + assert_eq!( + resolve_device_selector("device:mac-id", &devices).unwrap(), + expected + ); + assert_eq!( + resolve_device_selector("mac-id", &devices).unwrap(), + expected + ); + assert_eq!( + resolve_device_selector("macbook-pro-2.local", &devices).unwrap(), + expected + ); + assert_eq!( + resolve_device_selector("office mac", &devices).unwrap(), + expected + ); + + let offline = resolve_device_selector("old-laptop", &devices).unwrap_err(); + assert!(offline.to_string().contains("offline"), "{offline}"); + assert!(resolve_device_selector("missing", &devices).is_err()); + } + + #[test] + fn device_selector_rejects_ambiguous_names() { + let devices = vec![ + device("a", "build", None, true), + device("b", "build", None, true), + ]; + let error = resolve_device_selector("build", &devices).unwrap_err(); + assert!(error.to_string().contains("device:"), "{error}"); + } +} diff --git a/src/apps/cli/src/main.rs b/src/apps/cli/src/main.rs index bfc994e99e..37720329f8 100644 --- a/src/apps/cli/src/main.rs +++ b/src/apps/cli/src/main.rs @@ -387,7 +387,11 @@ enum Commands { action: DaemonAction, }, - /// Run and inspect persistent tasks owned by this machine + /// Run persistent tasks owned by this machine, or dispatch them to another host + /// + /// `dispatch outbound ...` is the controller: it submits a job to an SSH + /// connection or account device and observes it. The other verbs are the + /// JSON target protocol that controllers invoke on this machine. Dispatch { #[command(subcommand)] action: DispatchAction, @@ -779,6 +783,11 @@ pub(crate) enum DispatchAction { Continue, /// Read persisted session facts (usage report) without starting a turn Query, + /// Submit jobs to, and observe jobs on, another OpenBitFun host + Outbound { + #[command(subcommand)] + action: dispatch::outbound::OutboundAction, + }, #[command(name = "__workspace_provision", hide = true)] WorkspaceProvision, #[command(name = "__workspace_bundle_begin", hide = true)] @@ -2291,6 +2300,39 @@ mod dispatch_command_tests { .render_long_help() .to_string(); assert!(!dispatch_help.contains("__run")); + assert!(dispatch_help.contains("outbound")); + + let outbound = Cli::try_parse_from([ + "openbitfun", + "dispatch", + "outbound", + "submit", + "--target", + "device:mac-id", + "--approval-policy", + "reject-and-report", + "fix the flaky test", + ]) + .expect("parse outbound submit"); + assert!(matches!( + outbound.command, + Some(Commands::Dispatch { + action: DispatchAction::Outbound { + action: crate::dispatch::outbound::OutboundAction::Submit { ref target, .. } + } + }) if target == "device:mac-id" + )); + // Approval policy is never defaulted for an unattended job. + assert!(Cli::try_parse_from([ + "openbitfun", + "dispatch", + "outbound", + "submit", + "--target", + "mac", + "prompt", + ]) + .is_err()); } } diff --git a/src/apps/cli/src/root_handlers.rs b/src/apps/cli/src/root_handlers.rs index 5855d40688..925bbe7edb 100644 --- a/src/apps/cli/src/root_handlers.rs +++ b/src/apps/cli/src/root_handlers.rs @@ -52,6 +52,7 @@ pub(crate) async fn handle_dispatch_action(action: DispatchAction) -> Result<()> openbitfun_core::product_assembly::DeliveryProfile::Cli, )?; let verb = match action { + DispatchAction::Outbound { action } => return crate::dispatch::outbound::run(action).await, DispatchAction::Run { job } => return crate::dispatch::run_worker(job).await, DispatchAction::WorkspaceProvisionRun { job } => { return crate::dispatch::run_workspace_provision(job) diff --git a/src/apps/desktop/src/api/dispatch_api.rs b/src/apps/desktop/src/api/dispatch_api.rs index 2cfddad774..a1295d0638 100644 --- a/src/apps/desktop/src/api/dispatch_api.rs +++ b/src/apps/desktop/src/api/dispatch_api.rs @@ -11,8 +11,9 @@ use openbitfun_core::infrastructure::PathManager; use openbitfun_core::service::dispatch::{ answer_device_dispatch, answer_dispatch, append_device_dispatch, append_dispatch, cancel_device_dispatch, cancel_dispatch, cancel_dispatch_cli_install, - continue_device_dispatch_job, continue_dispatch_job, get_device_dispatch_status, - get_dispatch_status, list_device_dispatch_jobs, list_dispatch_jobs, list_dispatch_targets, + continue_device_dispatch_job, continue_dispatch_job, decode_device_dispatch_response, + encode_device_dispatch_command, get_device_dispatch_status, get_dispatch_status, + list_device_dispatch_jobs, list_dispatch_jobs, list_dispatch_targets, poll_dispatch_cli_install, probe_device_dispatch_target, probe_dispatch_target, query_device_dispatch_job, query_dispatch_job, start_dispatch_cli_install, submit_device_dispatch, submit_dispatch, sync_device_dispatch_result, @@ -38,17 +39,7 @@ struct AccountDeviceDispatchRpc; #[async_trait] impl DeviceDispatchRpc for AccountDeviceDispatchRpc { async fn invoke(&self, device_id: &str, command: &str, args: Value) -> anyhow::Result { - use openbitfun_core_types::agent_identity_wire::{ - translate_agent_identity_command, translate_agent_identity_response, - AgentIdentityDialect, - }; - let args = translate_agent_identity_command(command, args, AgentIdentityDialect::Legacy) - .map_err(anyhow::Error::msg)?; - let command_json = serde_json::to_string(&serde_json::json!({ - "cmd": "host_invoke", - "command": command, - "args": args, - }))?; + let command_json = encode_device_dispatch_command(command, args)?; let raw = super::remote_connect_api::account_device_rpc( device_id.to_string(), command_json, @@ -56,40 +47,7 @@ impl DeviceDispatchRpc for AccountDeviceDispatchRpc { ) .await .map_err(anyhow::Error::msg)?; - translate_agent_identity_response( - command, - decode_device_dispatch_rpc(&raw)?, - &Value::Null, - AgentIdentityDialect::Canonical, - ) - .map_err(anyhow::Error::msg) - } -} - -fn decode_device_dispatch_rpc(raw: &str) -> anyhow::Result { - let envelope: Value = - serde_json::from_str(raw).map_err(|error| anyhow::anyhow!("decode device RPC: {error}"))?; - match envelope.get("resp").and_then(Value::as_str) { - Some("host_invoke_result") if envelope.get("ok").and_then(Value::as_bool) == Some(true) => { - Ok(envelope.get("value").cloned().unwrap_or(Value::Null)) - } - Some("host_invoke_result") => Err(anyhow::anyhow!( - "{}", - envelope - .get("error") - .and_then(Value::as_str) - .unwrap_or("Device dispatch target rejected the command") - )), - Some("error") => Err(anyhow::anyhow!( - "{}", - envelope - .get("message") - .and_then(Value::as_str) - .unwrap_or("Device dispatch RPC failed") - )), - _ => Err(anyhow::anyhow!( - "Device dispatch target returned an unexpected RPC envelope" - )), + decode_device_dispatch_response(command, &raw) } } @@ -597,9 +555,14 @@ pub async fn dispatch_save_transcript( #[cfg(test)] mod tests { use super::{ - decode_device_dispatch_rpc, DispatchAccountBootstrapStatus, DispatchProvisionTargetResult, + decode_device_dispatch_response, DispatchAccountBootstrapStatus, + DispatchProvisionTargetResult, }; + fn decode_device_dispatch_rpc(raw: &str) -> anyhow::Result { + decode_device_dispatch_response("dispatch_target_submit", raw) + } + #[test] fn device_dispatch_requires_a_correlated_host_invoke_acknowledgement() { let value = decode_device_dispatch_rpc( diff --git a/src/crates/assembly/core/src/service/dispatch/account_device_rpc.rs b/src/crates/assembly/core/src/service/dispatch/account_device_rpc.rs new file mode 100644 index 0000000000..edbc19a827 --- /dev/null +++ b/src/crates/assembly/core/src/service/dispatch/account_device_rpc.rs @@ -0,0 +1,143 @@ +//! Account-device transport for the dispatch controller. +//! +//! Every controller host (Desktop, CLI) routes `dispatch_target_*` verbs to an +//! account device through the same `host_invoke` envelope over the encrypted +//! Relay RPC. The host supplies only the transport call; the envelope and the +//! agent-identity dialect translation live here so they cannot drift. + +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use openbitfun_core_types::agent_identity_wire::{ + translate_agent_identity_command, translate_agent_identity_response, AgentIdentityDialect, +}; +use serde_json::Value; + +use super::DeviceDispatchRpc; +use crate::service::remote_connect::account_runtime::AccountRuntime; +use crate::service::remote_connect::AccountClient; + +const DEVICE_DISPATCH_RPC_TIMEOUT: Duration = Duration::from_secs(120); + +/// Encode one dispatch verb as the plaintext `host_invoke` RPC command. +pub fn encode_device_dispatch_command(command: &str, args: Value) -> anyhow::Result { + let args = translate_agent_identity_command(command, args, AgentIdentityDialect::Legacy) + .map_err(anyhow::Error::msg)?; + Ok(serde_json::to_string(&serde_json::json!({ + "cmd": "host_invoke", + "command": command, + "args": args, + }))?) +} + +/// Decode a target's `host_invoke_result` reply for one dispatch verb. +pub fn decode_device_dispatch_response(command: &str, raw: &str) -> anyhow::Result { + let envelope: Value = + serde_json::from_str(raw).map_err(|error| anyhow::anyhow!("decode device RPC: {error}"))?; + let value = match envelope.get("resp").and_then(Value::as_str) { + Some("host_invoke_result") if envelope.get("ok").and_then(Value::as_bool) == Some(true) => { + envelope.get("value").cloned().unwrap_or(Value::Null) + } + Some("host_invoke_result") => anyhow::bail!( + "{}", + envelope + .get("error") + .and_then(Value::as_str) + .unwrap_or("Device dispatch target rejected the command") + ), + Some("error") => anyhow::bail!( + "{}", + envelope + .get("message") + .and_then(Value::as_str) + .unwrap_or("Device dispatch RPC failed") + ), + _ => anyhow::bail!("Device dispatch target returned an unexpected RPC envelope"), + }; + translate_agent_identity_response( + command, + value, + &Value::Null, + AgentIdentityDialect::Canonical, + ) + .map_err(anyhow::Error::msg) +} + +/// Device dispatch over the account session held by an [`AccountRuntime`]. +pub struct AccountRuntimeDeviceDispatchRpc { + account: Arc, +} + +impl AccountRuntimeDeviceDispatchRpc { + pub fn new(account: Arc) -> Self { + Self { account } + } +} + +#[async_trait] +impl DeviceDispatchRpc for AccountRuntimeDeviceDispatchRpc { + async fn invoke(&self, device_id: &str, command: &str, args: Value) -> anyhow::Result { + let command_json = encode_device_dispatch_command(command, args)?; + let generation = self.account.account_context_generation(); + let (session, relay_url) = self + .account + .read_account_context_for_generation(generation) + .await?; + let raw = tokio::time::timeout( + DEVICE_DISPATCH_RPC_TIMEOUT, + AccountClient::new().device_rpc(&relay_url, &session, device_id, &command_json), + ) + .await + .map_err(|_| { + anyhow::anyhow!( + "device RPC timed out after {}ms", + DEVICE_DISPATCH_RPC_TIMEOUT.as_millis() + ) + })??; + if !self.account.account_context_is_current(generation) { + anyhow::bail!("account context changed"); + } + decode_device_dispatch_response(command, &raw) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn device_dispatch_envelope_round_trips_success_and_errors() { + let encoded = + encode_device_dispatch_command("dispatch_target_probe", serde_json::json!({})).unwrap(); + let encoded: Value = serde_json::from_str(&encoded).unwrap(); + assert_eq!(encoded["cmd"], "host_invoke"); + assert_eq!(encoded["command"], "dispatch_target_probe"); + + let ok = decode_device_dispatch_response( + "dispatch_target_probe", + r#"{"resp":"host_invoke_result","ok":true,"value":{"protocolVersion":6}}"#, + ) + .unwrap(); + assert_eq!(ok["protocolVersion"], 6); + + let rejected = decode_device_dispatch_response( + "dispatch_target_probe", + r#"{"resp":"host_invoke_result","ok":false,"error":"busy"}"#, + ) + .unwrap_err(); + assert_eq!(rejected.to_string(), "busy"); + + let failed = decode_device_dispatch_response( + "dispatch_target_probe", + r#"{"resp":"error","message":"offline"}"#, + ) + .unwrap_err(); + assert_eq!(failed.to_string(), "offline"); + + assert!( + decode_device_dispatch_response("dispatch_target_probe", r#"{"resp":"other"}"#) + .is_err() + ); + } +} diff --git a/src/crates/assembly/core/src/service/dispatch/mod.rs b/src/crates/assembly/core/src/service/dispatch/mod.rs index 804b2449d2..585faae146 100644 --- a/src/crates/assembly/core/src/service/dispatch/mod.rs +++ b/src/crates/assembly/core/src/service/dispatch/mod.rs @@ -1,3 +1,9 @@ +#[cfg(all( + feature = "agent-runtime", + feature = "ssh-remote", + feature = "remote-connect" +))] +mod account_device_rpc; #[cfg(feature = "agent-runtime")] mod baseline; #[cfg(all(feature = "agent-runtime", feature = "ssh-remote"))] @@ -20,6 +26,15 @@ use tokio::fs; use crate::infrastructure::PathManager; +#[cfg(all( + feature = "agent-runtime", + feature = "ssh-remote", + feature = "remote-connect" +))] +pub use account_device_rpc::{ + decode_device_dispatch_response, encode_device_dispatch_command, + AccountRuntimeDeviceDispatchRpc, +}; #[cfg(all(feature = "agent-runtime", feature = "ssh-remote"))] pub use controller::{ answer as answer_dispatch, append as append_dispatch, cancel as cancel_dispatch, From 707a4f69f37e95fe7f99a4a4164f9d40e556fc06 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Sat, 31 Oct 2026 15:35:11 +0800 Subject: [PATCH 4/5] fix(config): follow the current global service in the reload poller Resolve the global config service on every poll instead of capturing the instance created at initialization, so a replaced service is not left stale. --- src/crates/assembly/core/src/service/config/global.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/crates/assembly/core/src/service/config/global.rs b/src/crates/assembly/core/src/service/config/global.rs index e5633a2995..e38466a277 100644 --- a/src/crates/assembly/core/src/service/config/global.rs +++ b/src/crates/assembly/core/src/service/config/global.rs @@ -110,7 +110,7 @@ impl GlobalConfigManager { crate::infrastructure::ai::reasoning_catalog::prepare_model_catalog(); #[cfg(feature = "agent-runtime")] - spawn_config_file_reload_poller(Arc::clone(&config_service)); + spawn_config_file_reload_poller(); info!("Global config service initialized"); @@ -227,7 +227,7 @@ const CONFIG_FILE_RELOAD_POLL_INTERVAL: std::time::Duration = std::time::Duratio /// Polling instead of a file-system watch keeps this working for atomic /// replacement, network home directories, and platforms without inotify. #[cfg(feature = "agent-runtime")] -fn spawn_config_file_reload_poller(service: Arc) { +fn spawn_config_file_reload_poller() { let Ok(runtime) = tokio::runtime::Handle::try_current() else { log::warn!("No async runtime available; config file reload polling is disabled"); return; @@ -239,6 +239,10 @@ fn spawn_config_file_reload_poller(service: Arc) { let mut last_error: Option = None; loop { interval.tick().await; + // Resolve each tick so a replaced global service is followed. + let Ok(service) = GlobalConfigManager::get_service().await else { + continue; + }; match service.reload_if_changed_on_disk().await { Ok(_) => last_error = None, Err(error) => { From ab06afd976ec57bc2af51342e8f584df89233201 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Sat, 31 Oct 2026 15:49:38 +0800 Subject: [PATCH 5/5] test(core-boundaries): admit dispatch-store only into the CLI closure Update the CLI reviewed Core feature fixture and replace the blanket dispatch-store exclusion test with one that allows the CLI controller and still rejects the feature for another reviewed closure. --- scripts/check-core-boundaries.test.mjs | 32 ++++++++++++++++++-------- 1 file changed, 23 insertions(+), 9 deletions(-) diff --git a/scripts/check-core-boundaries.test.mjs b/scripts/check-core-boundaries.test.mjs index 5731b4b88e..6e0ffa78ef 100644 --- a/scripts/check-core-boundaries.test.mjs +++ b/scripts/check-core-boundaries.test.mjs @@ -1916,6 +1916,7 @@ const CLI_REVIEWED_CORE_FEATURES = [ 'remote-connect', 'plugin-runtime', 'opencode-plugin-host', + 'dispatch-store', ]; const APP_SERVER_REVIEWED_CORE_FEATURES = [ @@ -2912,7 +2913,7 @@ test('CLI dependency closure includes build dependencies and excluded capabiliti assert.match(violations[0].message, /openbitfun-core\/product-full/); }); -test('CLI dependency closure excludes the Core dispatch store', () => { +test('only the CLI dependency closure may select the Core dispatch store', () => { const core = { ...packageAt('openbitfun-core', 'src/crates/assembly/core/Cargo.toml'), features: { 'dispatch-store': [] }, @@ -2924,17 +2925,30 @@ test('CLI dependency closure excludes the Core dispatch store', () => { features: ['dispatch-store'], }), ]); - const cli = packageAt('openbitfun-cli', 'src/apps/cli/Cargo.toml', [ - pathDependency('src/crates/assembly/bridge', { name: 'bridge' }), - ]); - - const violations = findProductEntrypointCoreFeatureViolations( - [cli, bridge, core], + const closureViolations = (rootName, manifestPath) => findProductEntrypointCoreFeatureViolations( + [ + packageAt(rootName, manifestPath, [ + pathDependency('src/crates/assembly/bridge', { name: 'bridge' }), + ]), + bridge, + core, + ], { root: TEST_ROOT, crateLayoutRules }, ); - assert.equal(violations.length, 1); - assert.match(violations[0].message, /openbitfun-core\/dispatch-store/); + // The CLI is the reviewed outbound dispatch controller. + assert.deepEqual( + closureViolations('openbitfun-cli', 'src/apps/cli/Cargo.toml') + .filter((violation) => /dispatch-store/.test(violation.message)), + [], + ); + + const acpViolations = closureViolations( + 'openbitfun-acp', + 'src/crates/interfaces/acp/Cargo.toml', + ); + assert.equal(acpViolations.length, 1); + assert.match(acpViolations[0].message, /openbitfun-core\/dispatch-store/); }); test('cargo layer checker rejects reverse edges across dependency kinds', () => {