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/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', () => { 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/config/global.rs b/src/crates/assembly/core/src/service/config/global.rs index 412a4c6cd3..e38466a277 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(); + info!("Global config service initialized"); Ok(()) @@ -214,6 +217,50 @@ 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() { + 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; + // 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) => { + // 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; 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, 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());