From 55492ef83425ec21079193ad59ca9227115a9818 Mon Sep 17 00:00:00 2001 From: ymote <151983+ymote@users.noreply.github.com> Date: Mon, 28 Sep 2026 17:37:11 -0700 Subject: [PATCH 1/2] shell: process sandboxes from the manifest; a dead process app keeps its tile [skip ci] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ADR 0004 §3, §11 and §2. tools/native_apps.py validates and generates each app's `sandbox` (network, processes) and `storage` (accounts, external) into native_apps.rs. sandbox/ builds one policy per app and applies it in spawn_client: - macOS: a Seatbelt profile through /usr/bin/sandbox-exec (cargo's runner for a `cargo run` launch, so the build stays outside): the person's data roots closed but for the jail, its secrets and `external`; program and resources read-only; network `none` = the hub on loopback only; `processes: false` = no fork or exec. - Linux: Landlock (paths; TCP ports from ABI 4) and seccomp, installed between fork and exec, best-effort with the layers logged. - Windows: a documented AppContainer TODO; the log says unsandboxed. - No child inherits OCTOS_* / OCTOSENSE_* (scrub_env), and a cargo launch's [env] host variables are unset in its runner. - OCTOSENSE_SANDBOX_NARROW narrows an app for test runs (only removes). A process app that dies unexpectedly now keeps its tile, closed with " stopped" / "Click to restart" (MpRunView::show_stopped), and a press restarts it; the shell and other apps keep running. The processes_available() test now covers the phone targets too. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/shell/src/clients.rs | 104 ++++++++- crates/shell/src/host.rs | 9 +- crates/shell/src/native_apps.rs | 40 ++++ crates/shell/src/run_view.rs | 44 +++- crates/shell/src/sandbox/linux.rs | 296 ++++++++++++++++++++++++ crates/shell/src/sandbox/macos.rs | 111 +++++++++ crates/shell/src/sandbox/mod.rs | 358 ++++++++++++++++++++++++++++++ crates/shell/src/sandbox/tests.rs | 198 +++++++++++++++++ tools/native_apps.py | 60 +++++ tools/test_native_apps.py | 11 + 10 files changed, 1224 insertions(+), 7 deletions(-) create mode 100644 crates/shell/src/sandbox/linux.rs create mode 100644 crates/shell/src/sandbox/macos.rs create mode 100644 crates/shell/src/sandbox/mod.rs create mode 100644 crates/shell/src/sandbox/tests.rs diff --git a/crates/shell/src/clients.rs b/crates/shell/src/clients.rs index 6ca63300..256048b3 100644 --- a/crates/shell/src/clients.rs +++ b/crates/shell/src/clients.rs @@ -550,6 +550,9 @@ pub struct ClientSlot { /// viewport (a map that pans from the edge): the shell's back gesture /// is not recognised over it. Off by default; an app opts in. pub owns_edges: bool, + /// Its process died unexpectedly: the tile stays, closed, with a + /// Restart (ADR 0004 §2); nothing runs behind it any more. + pub stopped: bool, } impl ClientSlot { @@ -582,6 +585,7 @@ impl ClientSlot { linked_at: None, closing: None, pane: false, + stopped: false, } } } @@ -764,6 +768,14 @@ impl ClientSlot { Some(message) } + /// Whether an exit leaves this client's tile in place, closed with a + /// Restart: an unexpected death (`failure`) of a window the person + /// has. A warm instance, the AI pane, a Quick-Look preview and a + /// client being closed go away as before. + pub fn stops_in_place(&self, failure: bool) -> bool { + failure && self.ready && !self.warm && !self.pane && !self.is_preview && self.closing.is_none() && self.child.is_some() + } + pub fn display_title(&self) -> &str { if self.title.is_empty() { &self.app @@ -921,6 +933,60 @@ pub fn launch_argv( Ok((program, args)) } +/// The sandbox of a native app's process launch (`None` for a catalog app +/// that is not in `native-apps.json`). Its jail and secrets folders are +/// created first; its program and resources are the checkout, the target +/// dir and cargo's source cache (crate resources) or, installed, the +/// binary's own directory. +pub fn sandbox_policy(app: &AppDef, root: Option<&Path>, program: &Path, hub_port: u16) -> Option { + let native = crate::native_apps::find(&app.id)?; + let layout = match crate::app_storage::host() { + Some(host) => host.layout().clone(), + None => crate::app_storage::Layout::platform(None).ok()?, + }; + let paths = layout.app(native.id).ok()?; + for (base, dir) in [(layout.apps_root(), &paths.jail), (layout.secrets_root(), &paths.secrets)] { + if let Err(e) = crate::app_storage::ensure_private_dir(base, dir) { + makepad_widgets::log!("sandbox: {}: cannot prepare {}: {e}", native.id, dir.display()); + } + } + let home = crate::sandbox::person_home().unwrap_or_else(|| PathBuf::from("/nonexistent")); + let mut roots: Vec = Vec::new(); + let manifest_root = app.manifest.as_ref().and_then(|m| Path::new(m).parent().map(Path::to_path_buf)); + if let Some(checkout) = manifest_root.or_else(|| root.map(Path::to_path_buf)) { + roots.push(checkout.clone()); + roots.push(app.target_dir.as_ref().map(PathBuf::from).or_else(|| std::env::var_os("CARGO_TARGET_DIR").map(PathBuf::from)).unwrap_or_else(|| checkout.join("target"))); + let cargo_home = std::env::var_os("CARGO_HOME").map(PathBuf::from).unwrap_or_else(|| home.join(".cargo")); + roots.push(cargo_home.join("git")); + roots.push(cargo_home.join("registry")); + } + if let Some(dir) = program.parent() { + roots.push(dir.to_path_buf()); + } + // The shell's own directory: an installed launch's sibling binary. + if let Some(dir) = std::env::current_exe().ok().and_then(|e| e.parent().map(Path::to_path_buf)) { + roots.push(dir); + } + roots.sort(); + roots.dedup(); + let mut policy = crate::sandbox::Policy::for_app(native, paths.jail, paths.secrets, &home, roots, hub_port); + // Cargo reads `.cargo/config.toml` from its working directory up (the + // manifest's directory, else the checkout): every `[env]` host variable + // it would hand the app is taken back out. + let cargo_cwd = app.manifest.as_ref().and_then(|m| Path::new(m).parent().map(Path::to_path_buf)).or_else(|| root.map(Path::to_path_buf)); + if let Some(cwd) = cargo_cwd { + for dir in cwd.ancestors() { + let config = std::fs::read_to_string(dir.join(".cargo/config.toml")).unwrap_or_default(); + for name in crate::sandbox::cargo_env_host_vars(&config) { + if !policy.cargo_env_unset.contains(&name) { + policy.cargo_env_unset.push(name); + } + } + } + } + Some(if crate::sandbox::narrowed_by_env(native.id) { policy.narrowed() } else { policy }) +} + /// Spawn an app as a hub client. pub fn spawn_client( pool: &TaskPool, @@ -944,8 +1010,16 @@ pub fn spawn_client( let root = repo_root(); let via_cargo = !app.package.is_empty() && (app.manifest.is_some() || root.is_some()); let (program, args) = launch_argv(app, root.as_deref(), extra_args)?; - let mut cmd = Command::new(program); - cmd.args(&args); + // A native app runs under the OS sandbox its manifest entry builds + // (ADR 0004 §3, sandbox/); through cargo only the app is sandboxed + // where the platform lets the build stay outside. + let policy = sandbox_policy(app, root.as_deref(), &program, hub_port); + let (mut cmd, applied) = crate::sandbox::command(&program, &args, policy.as_ref(), via_cargo); + match &applied { + Some(crate::sandbox::Applied::Sandboxed(how)) => makepad_widgets::log!("sandbox: {how}"), + Some(crate::sandbox::Applied::Unavailable(why)) => makepad_widgets::log!("sandbox: UNSANDBOXED {why}"), + None => {} + } // Give the process its own group (unix): `cargo run` does not // exec-replace itself, so the compiled app (and, mid-build, rustc) are // further children of the process we hold, not exec'd into it. Sharing @@ -993,6 +1067,9 @@ pub fn spawn_client( if warm { cmd.env(WARM_ENV.0, WARM_ENV.1); } + // No child inherits the kernel's descriptors or the host token: a + // process app reaches its agent only over the peer link (ADR 0004 §3). + crate::sandbox::scrub_env(&mut cmd); let mut child = cmd .spawn() .map_err(|e| format!("spawn {}: {}", app.package, e))?; @@ -1035,6 +1112,7 @@ pub fn spawn_client( linked_at: None, closing: None, pane: false, + stopped: false, }) } @@ -1618,6 +1696,28 @@ mod tests { "exiting successfully before creating a window is also a startup failure"); } + /// ADR 0004 §2: a process app that dies takes only itself down. Its + /// tile stays, closed with a Restart; a warm instance, the AI pane, a + /// preview and a client being closed are removed as before. + #[cfg(unix)] + #[test] + fn an_unexpected_death_keeps_the_tile_closed_with_restart() { + let mut slot = ClientSlot::module(9, "terminal", "Terminal"); + slot.child = Some(std::process::Command::new("/usr/bin/true").spawn().unwrap()); + assert!(slot.stops_in_place(true)); + assert!(!slot.stops_in_place(false), "a clean exit closes the window"); + slot.closing = Some(0.0); + assert!(!slot.stops_in_place(true), "the person closed it"); + slot.closing = None; + for (warm, pane, preview, ready) in [(true, false, false, true), (false, true, false, true), (false, false, true, true), (false, false, false, false)] { + (slot.warm, slot.pane, slot.is_preview, slot.ready) = (warm, pane, preview, ready); + assert!(!slot.stops_in_place(true), "warm {warm} pane {pane} preview {preview} ready {ready}"); + } + let _ = slot.child.take().map(|mut c| c.wait()); + slot.ready = true; + assert!(!slot.stops_in_place(true), "a module slot has no process to stop"); + } + #[cfg(unix)] #[test] fn requested_closes_warm_exits_and_normal_exits_do_not_notify() { diff --git a/crates/shell/src/host.rs b/crates/shell/src/host.rs index 946be539..9393687d 100644 --- a/crates/shell/src/host.rs +++ b/crates/shell/src/host.rs @@ -116,7 +116,14 @@ mod tests { #[test] fn the_desk_knows_where_it_runs() { - assert_eq!(processes_available(), cfg!(not(target_arch = "wasm32"))); + // No processes on the web or on a phone (ADR 0004 §2): Android, + // iOS and OpenHarmony host their native apps in-process. + let mobile = cfg!(any(target_os = "android", target_os = "ios", target_env = "ohos")); + assert_eq!(cfg!(native_mobile), mobile, "native_mobile is exactly the phone targets"); + assert_eq!(processes_available(), !cfg!(target_arch = "wasm32") && !mobile); + if cfg!(any(target_os = "macos", target_os = "windows", target_os = "linux")) && !mobile { + assert!(processes_available(), "a desktop hosts process apps"); + } assert!(now() >= 0.0); } } diff --git a/crates/shell/src/native_apps.rs b/crates/shell/src/native_apps.rs index 1b05f1aa..c4456ae9 100644 --- a/crates/shell/src/native_apps.rs +++ b/crates/shell/src/native_apps.rs @@ -38,6 +38,14 @@ pub struct ToolPolicy { pub auto_approvable: bool, } +/// What a process-hosted instance may reach on the network (ADR 0004 §3). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Network { + /// Nothing but the shell's hub on loopback. + None, + Any, +} + /// One `native-apps.json` entry, as far as the shell reads it. #[derive(Debug)] pub struct NativeApp { @@ -57,6 +65,14 @@ pub struct NativeApp { pub octos: &'static [&'static str], /// Its assistant tools' confirmation rules (`agent.tool_policy`). pub tools: &'static [ToolPolicy], + /// `sandbox.network`: what its OS sandbox lets it reach. + pub network: Network, + /// `sandbox.processes`: whether it may start child processes. + pub processes: bool, + /// `storage.accounts`: one folder (and agent) per account. + pub accounts: bool, + /// `storage.external`: `[/]:ro|rw` outside its jail. + pub external: &'static [&'static str], } pub const APPS: &[NativeApp] = &[ @@ -73,6 +89,10 @@ pub const APPS: &[NativeApp] = &[ wasm: Hosting::Module, octos: &["octos.session.open", "octos.session.history", "octos.turn.start", "octos.turn.interrupt"], tools: &[], + network: Network::Any, + processes: false, + accounts: true, + external: &[], }, NativeApp { id: "reference", @@ -87,6 +107,10 @@ pub const APPS: &[NativeApp] = &[ wasm: Hosting::Module, octos: &[], tools: &[], + network: Network::None, + processes: false, + accounts: false, + external: &[], }, NativeApp { id: "sheets", @@ -101,6 +125,10 @@ pub const APPS: &[NativeApp] = &[ wasm: Hosting::Module, octos: &[], tools: &[], + network: Network::None, + processes: false, + accounts: false, + external: &[], }, NativeApp { id: "terminal", @@ -117,6 +145,10 @@ pub const APPS: &[NativeApp] = &[ tools: &[ ToolPolicy { tool: "run", confirm: Confirm::Host, auto_approvable: false }, ], + network: Network::Any, + processes: true, + accounts: false, + external: &["home:rw"], }, NativeApp { id: "appcard", @@ -131,6 +163,10 @@ pub const APPS: &[NativeApp] = &[ wasm: Hosting::Module, octos: &[], tools: &[], + network: Network::Any, + processes: false, + accounts: false, + external: &[], }, NativeApp { id: "apphub", @@ -145,6 +181,10 @@ pub const APPS: &[NativeApp] = &[ wasm: Hosting::Module, octos: &[], tools: &[], + network: Network::Any, + processes: false, + accounts: false, + external: &[], }, ]; diff --git a/crates/shell/src/run_view.rs b/crates/shell/src/run_view.rs index 4146bb91..94e93c2f 100644 --- a/crates/shell/src/run_view.rs +++ b/crates/shell/src/run_view.rs @@ -160,8 +160,9 @@ pub enum MpRunViewAction { Clicked { client: ClientId, }, - /// The person asked a module tile whose app stopped after a panic to - /// start it again (`module_view.rs`). + /// The person asked a tile whose app stopped (a module after a panic, + /// `module_view.rs`; a process that died, [`MpRunView::show_stopped`]) + /// to start it again. Restart { client: ClientId, }, @@ -203,6 +204,10 @@ pub struct MpRunView { /// Newest stdout/stderr line from the child, shown while it starts. #[rust] status_line: String, + /// The child died unexpectedly: the tile shows it closed, and a press + /// asks for a Restart (ADR 0004 §2). + #[rust] + stopped: Option<(ClientId, String)>, #[rust] startup_initialized: bool, #[rust] startup_glass: bool, @@ -765,8 +770,25 @@ impl MpRunView { } } + /// The child died unexpectedly: show the tile closed (" stopped", + /// "Click to restart") and stop talking to it. + pub fn show_stopped(&mut self, cx: &mut Cx, client: ClientId, label: &str) { + self.stopped = Some((client, format!("{label} stopped"))); + self.status_line = "Click to restart".into(); + self.current_target = None; + self.present_ok_count = 0; + self.first_present_at = None; + self.pending_draw = None; + self.redraw(cx); + } + + pub fn is_stopped(&self) -> bool { + self.stopped.is_some() + } + pub fn set_status_line(&mut self, cx: &mut Cx, line: &str) { - if self.status_line == line { + // A stopped tile keeps saying how to restart it. + if self.stopped.is_some() || self.status_line == line { return; } self.status_line = line.to_string(); @@ -907,7 +929,9 @@ impl Widget for MpRunView { self.startup_initialized = true; self.no_fb_view.set_text(cx, include_str!("../resources/startup.splash")); } - let headline = if self.status_line.starts_with("compiling ") { + let headline = if let Some((_, line)) = &self.stopped { + line.as_str() + } else if self.status_line.starts_with("compiling ") { "Compiling…" } else if self.status_line.starts_with("waiting for another build") { "Waiting to compile…" @@ -937,6 +961,11 @@ impl Widget for MpRunView { }; } } + if self.stopped.is_some() { + // Closed: the panel, never the last frame; presses land on it. + self.area = self.draw_bg.area(); + return DrawStep::done(); + } self.draw_app.draw_abs(cx, rect); self.area = self.draw_app.area(); if target.is_some() && cx.has_key_focus(self.area) { @@ -953,6 +982,13 @@ impl Widget for MpRunView { } fn handle_event(&mut self, cx: &mut Cx, event: &Event, _scope: &mut Scope) { + if let Some((client, _)) = &self.stopped { + let client = *client; + if let Hit::FingerDown(_) = event.hits(cx, self.area) { + cx.widget_action(self.uid, MpRunViewAction::Restart { client }); + } + return; + } let target = self.current_target; if let Event::Timer(timer_event) = event { diff --git a/crates/shell/src/sandbox/linux.rs b/crates/shell/src/sandbox/linux.rs new file mode 100644 index 00000000..3933ca08 --- /dev/null +++ b/crates/shell/src/sandbox/linux.rs @@ -0,0 +1,296 @@ +//! The Linux sandbox: Landlock for paths (and, from ABI 4, TCP ports) and a +//! seccomp filter, both installed in the child between fork and exec. +//! +//! Landlock is an allow-list: everything the ruleset handles is closed +//! except the rules, so the person's home is closed without naming it. +//! The rules: the system's read-only locations (libraries, `/etc`, `/usr`), +//! devices (the GPU), `/proc` and `/sys`, the app's program and resources +//! read-only, and its jail, secrets and `external` grants. Execution is +//! allowed only from the program's roots (and the system's, when +//! `processes` is true). +//! +//! seccomp refuses, with `EPERM`, what no process app needs: `ptrace`, +//! `process_vm_readv/writev`, `perf_event_open`, `bpf`, `userfaultfd`, +//! `kexec_load`, mounts, namespaces and the kernel keyring; with +//! `processes: false` also `fork`, `vfork` and a `clone` that is not a +//! thread (and `clone3`, which libc then retries as `clone`). +//! +//! **Best-effort.** The parent probes the kernel's Landlock ABI before the +//! spawn and says in [`Applied`] which layers took (a kernel before 5.13, or +//! one without Landlock enabled, gets seccomp only; before ABI 4, no port +//! rules). A layer that fails in the child is skipped, never fatal: the app +//! still starts. +//! +//! **A launch through cargo** (a dev run from a checkout) sandboxes the +//! build as well: the build's own paths (the checkout, the target dir, +//! cargo's and rustup's homes) are added and child processes are allowed so +//! cargo can run rustc. An installed launch gets exactly the manifest's +//! sandbox. + +use std::ffi::CString; +use std::os::unix::ffi::OsStrExt; +use std::os::unix::process::CommandExt; +use std::path::PathBuf; +use std::process::Command; + +use super::{Access, Applied, Policy}; +use crate::native_apps::Network; + +const SYS_LANDLOCK_CREATE_RULESET: libc::c_long = 444; +const SYS_LANDLOCK_ADD_RULE: libc::c_long = 445; +const SYS_LANDLOCK_RESTRICT_SELF: libc::c_long = 446; +const LANDLOCK_CREATE_RULESET_VERSION: u32 = 1; +const LANDLOCK_RULE_PATH_BENEATH: u32 = 1; +const LANDLOCK_RULE_NET_PORT: u32 = 2; + +const FS_EXECUTE: u64 = 1 << 0; +const FS_WRITE_FILE: u64 = 1 << 1; +const FS_READ_FILE: u64 = 1 << 2; +const FS_READ_DIR: u64 = 1 << 3; +const FS_TRUNCATE: u64 = 1 << 14; +const FS_IOCTL_DEV: u64 = 1 << 15; +const NET_BIND_TCP: u64 = 1 << 0; +const NET_CONNECT_TCP: u64 = 1 << 1; + +/// The rights a file (not a directory) may carry in a rule. +const FILE_RIGHTS: u64 = FS_EXECUTE | FS_WRITE_FILE | FS_READ_FILE | FS_TRUNCATE | FS_IOCTL_DEV; + +#[repr(C)] +struct RulesetAttr { + handled_access_fs: u64, + handled_access_net: u64, +} + +#[repr(C, packed)] +struct PathBeneathAttr { + allowed_access: u64, + parent_fd: i32, +} + +#[repr(C)] +struct NetPortAttr { + allowed_access: u64, + port: u64, +} + +/// The kernel's Landlock ABI version, 0 when it has none. +pub fn landlock_abi() -> u32 { + let r = unsafe { libc::syscall(SYS_LANDLOCK_CREATE_RULESET, std::ptr::null::(), 0usize, LANDLOCK_CREATE_RULESET_VERSION) }; + if r < 0 { 0 } else { r as u32 } +} + +/// Every filesystem right the ruleset handles at `abi`. +pub fn handled_fs(abi: u32) -> u64 { + let mut rights = (1u64 << 13) - 1; // ABI 1: EXECUTE ..= MAKE_SYM + if abi >= 2 { + rights |= 1 << 13; // REFER + } + if abi >= 3 { + rights |= FS_TRUNCATE; + } + if abi >= 5 { + rights |= FS_IOCTL_DEV; + } + rights +} + +/// One path rule, prepared in the parent. +#[derive(Clone, Debug, PartialEq)] +pub struct Rule { + pub path: PathBuf, + pub access: u64, +} + +fn read() -> u64 { + FS_READ_FILE | FS_READ_DIR +} + +/// The path rules for `policy` at `abi` (`via_cargo`: the build's paths too). +pub fn rules(policy: &Policy, abi: u32, via_cargo: bool) -> Vec { + let all = handled_fs(abi); + let rx = read() | FS_EXECUTE; + let processes = policy.processes || via_cargo; + let system_exec = if processes { rx } else { read() }; + let mut out = Vec::new(); + let mut add = |path: PathBuf, access: u64| { + if path.as_os_str().is_empty() { + return; + } + let access = access & all; + let access = if path.is_dir() { access } else { access & FILE_RIGHTS }; + out.push(Rule { path, access }); + }; + // Shared libraries are mapped, not executed: read is enough unless the + // app may start other programs. + for dir in ["/usr", "/lib", "/lib64", "/lib32", "/bin", "/sbin", "/opt", "/nix/store"] { + add(PathBuf::from(dir), system_exec); + } + // The dynamic loader is exec'd with the program. + for loader in ["/lib64/ld-linux-x86-64.so.2", "/lib/ld-linux-aarch64.so.1"] { + add(PathBuf::from(loader), rx); + } + add(PathBuf::from("/etc"), read()); + add(PathBuf::from("/sys"), read()); + add(PathBuf::from("/proc"), read() | FS_WRITE_FILE); + add(PathBuf::from("/dev"), read() | FS_WRITE_FILE | FS_IOCTL_DEV); + add(PathBuf::from("/run"), read()); + if let Some(home) = super::person_home() { + for fonts in [".local/share/fonts", ".fonts", ".cache/fontconfig", ".config/fontconfig"] { + add(home.join(fonts), read()); + } + } + for program in &policy.program { + add(program.clone(), rx); + } + add(policy.jail.clone(), all); + add(policy.secrets.clone(), all); + for (path, access) in &policy.external { + add(path.clone(), if *access == Access::Read { read() } else { all }); + } + if via_cargo { + let home = super::person_home().unwrap_or_default(); + let cargo_home = std::env::var_os("CARGO_HOME").map(PathBuf::from).unwrap_or_else(|| home.join(".cargo")); + let rustup_home = std::env::var_os("RUSTUP_HOME").map(PathBuf::from).unwrap_or_else(|| home.join(".rustup")); + add(cargo_home, all); + add(rustup_home, rx); + add(PathBuf::from("/tmp"), all); + for program in &policy.program { + add(program.clone(), all); + } + } + out +} + +/// The seccomp program for `processes` on this architecture, `None` where +/// the table below has no numbers for it. +pub fn seccomp_filter(processes: bool) -> Option> { + #[cfg(target_arch = "x86_64")] + let (arch, denied, forks, clone, clone3): (u32, &[u32], &[u32], u32, u32) = ( + 0xC000_003E, + &[101, 310, 311, 298, 246, 321, 323, 165, 166, 155, 250, 248, 249, 272, 308], + &[57, 58], + 56, + 435, + ); + #[cfg(target_arch = "aarch64")] + let (arch, denied, forks, clone, clone3): (u32, &[u32], &[u32], u32, u32) = ( + 0xC000_00B7, + &[117, 270, 271, 241, 104, 280, 282, 40, 39, 41, 219, 217, 218, 97, 268], + &[], + 220, + 435, + ); + #[cfg(not(any(target_arch = "x86_64", target_arch = "aarch64")))] + { + let _ = processes; + return None; + } + #[cfg(any(target_arch = "x86_64", target_arch = "aarch64"))] + { + const LD_W_ABS: u16 = 0x20; + const JEQ_K: u16 = 0x15; + const JSET_K: u16 = 0x45; + const RET_K: u16 = 0x06; + const ALLOW: u32 = 0x7fff_0000; + const ERRNO: u32 = 0x0005_0000; + const CLONE_THREAD: u32 = 0x0001_0000; + let st = |code: u16, jt: u8, jf: u8, k: u32| libc::sock_filter { code, jt, jf, k }; + let eperm = ERRNO | libc::EPERM as u32; + let enosys = ERRNO | libc::ENOSYS as u32; + let mut f = vec![ + st(LD_W_ABS, 0, 0, 4), // arch + st(JEQ_K, 1, 0, arch), + st(RET_K, 0, 0, ALLOW), // an unexpected arch: not ours to judge + st(LD_W_ABS, 0, 0, 0), // nr + ]; + let deny = |f: &mut Vec, nr: u32, ret: u32| { + f.push(st(JEQ_K, 0, 1, nr)); + f.push(st(RET_K, 0, 0, ret)); + }; + for &nr in denied { + deny(&mut f, nr, eperm); + } + if !processes { + for &nr in forks { + deny(&mut f, nr, eperm); + } + deny(&mut f, clone3, enosys); + // clone: a thread (CLONE_THREAD) passes, a new process does not. + f.push(st(JEQ_K, 0, 4, clone)); + f.push(st(LD_W_ABS, 0, 0, 16)); // args[0], low word + f.push(st(JSET_K, 1, 0, CLONE_THREAD)); + f.push(st(RET_K, 0, 0, eperm)); + f.push(st(RET_K, 0, 0, ALLOW)); + } + f.push(st(RET_K, 0, 0, ALLOW)); + Some(f) + } +} + +/// Put Landlock and seccomp on `cmd` (installed in the child before exec). +pub fn apply(cmd: &mut Command, policy: &Policy, via_cargo: bool) -> Applied { + let abi = landlock_abi(); + let processes = policy.processes || via_cargo; + // Everything the child does is prepared here: between fork and exec it + // only opens paths and makes system calls. + let prepared: Vec<(CString, u64)> = if abi == 0 { + Vec::new() + } else { + rules(policy, abi, via_cargo) + .into_iter() + .filter(|r| r.path.exists()) + .filter_map(|r| CString::new(r.path.as_os_str().as_bytes()).ok().map(|c| (c, r.access))) + .collect() + }; + let net = abi >= 4 && policy.network == Network::None && !via_cargo; + let hub_port = policy.hub_port; + let filter = seccomp_filter(processes); + let handled = handled_fs(abi); + let mut layers = Vec::new(); + if abi > 0 { + layers.push(format!("landlock abi {abi}{}", if net { " + ports" } else { "" })); + } else { + layers.push("no landlock (kernel lacks it): paths are not restricted".to_string()); + } + if abi > 0 && abi < 4 && policy.network == Network::None { + layers.push("network not restricted (landlock < 4)".into()); + } + layers.push(if filter.is_some() { "seccomp".into() } else { "no seccomp on this architecture".to_string() }); + if via_cargo { + layers.push("launched through cargo: the build's paths and processes added".into()); + } + unsafe { + cmd.pre_exec(move || { + // Landlock and seccomp both need no_new_privs. + libc::prctl(libc::PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0); + if abi > 0 { + let attr = RulesetAttr { handled_access_fs: handled, handled_access_net: if net { NET_BIND_TCP | NET_CONNECT_TCP } else { 0 } }; + let size = if abi >= 4 { std::mem::size_of::() } else { 8 }; + let ruleset = libc::syscall(SYS_LANDLOCK_CREATE_RULESET, &attr as *const RulesetAttr, size, 0u32) as i32; + if ruleset >= 0 { + for (path, access) in &prepared { + let fd = libc::open(path.as_ptr(), libc::O_PATH | libc::O_CLOEXEC); + if fd < 0 { + continue; + } + let rule = PathBeneathAttr { allowed_access: *access, parent_fd: fd }; + libc::syscall(SYS_LANDLOCK_ADD_RULE, ruleset, LANDLOCK_RULE_PATH_BENEATH, &rule as *const PathBeneathAttr, 0u32); + libc::close(fd); + } + if net { + let rule = NetPortAttr { allowed_access: NET_CONNECT_TCP, port: hub_port as u64 }; + libc::syscall(SYS_LANDLOCK_ADD_RULE, ruleset, LANDLOCK_RULE_NET_PORT, &rule as *const NetPortAttr, 0u32); + } + libc::syscall(SYS_LANDLOCK_RESTRICT_SELF, ruleset, 0u32); + libc::close(ruleset); + } + } + if let Some(filter) = &filter { + let prog = libc::sock_fprog { len: filter.len() as u16, filter: filter.as_ptr() as *mut libc::sock_filter }; + libc::prctl(libc::PR_SET_SECCOMP, libc::SECCOMP_MODE_FILTER, &prog as *const libc::sock_fprog); + } + Ok(()) + }); + } + Applied::Sandboxed(format!("{} ({})", policy.summary(), layers.join(", "))) +} diff --git a/crates/shell/src/sandbox/macos.rs b/crates/shell/src/sandbox/macos.rs new file mode 100644 index 00000000..d2663110 --- /dev/null +++ b/crates/shell/src/sandbox/macos.rs @@ -0,0 +1,111 @@ +//! The macOS sandbox: a Seatbelt (SBPL) profile run through +//! `/usr/bin/sandbox-exec` (deprecated in Apple's headers, still the +//! mechanism; see the module above). +//! +//! The profile allows by default what a GPU app needs from the system +//! (libraries, fonts, the window server, Metal's shader cache) and closes +//! the person's data roots except the grants. In SBPL the last matching +//! rule wins, so the order below is the policy: close the roots, reopen the +//! ancestors' metadata (paths must resolve), the program read-only, then +//! the jail, secrets and `external` grants. + +use std::path::{Path, PathBuf}; + +use super::{ancestors, resolved, Access, Policy}; +use crate::native_apps::Network; + +/// The one binary the mechanism needs. +pub const SANDBOX_EXEC: &str = "/usr/bin/sandbox-exec"; + +fn quote(path: &Path) -> String { + let s = path.to_string_lossy(); + let mut out = String::with_capacity(s.len() + 2); + out.push('"'); + for c in s.chars() { + if c == '"' || c == '\\' { + out.push('\\'); + } + out.push(c); + } + out.push('"'); + out +} + +fn subpaths(paths: &[PathBuf]) -> String { + paths.iter().map(|p| format!(" (subpath {})", quote(p))).collect() +} + +/// The profile text for `policy` (paths resolved, as the kernel sees them). +pub fn profile(policy: &Policy) -> String { + let protected: Vec = policy.protected.iter().map(|p| resolved(p)).collect(); + let program: Vec = policy.program.iter().map(|p| resolved(p)).collect(); + let mut rw = vec![resolved(&policy.jail), resolved(&policy.secrets)]; + let mut ro = Vec::new(); + for (path, access) in &policy.external { + match access { + Access::ReadWrite => rw.push(resolved(path)), + Access::Read => ro.push(resolved(path)), + } + } + // Every ancestor of an opened path inside a closed root: metadata only. + let mut metadata: Vec = Vec::new(); + for path in rw.iter().chain(&ro).chain(&program) { + for a in ancestors(path) { + if protected.iter().any(|root| a.starts_with(root)) && !metadata.contains(&a) { + metadata.push(a); + } + } + } + let mut out = String::new(); + out.push_str(&format!(";; OctoSense process sandbox for {} (ADR 0004 §3), generated\n", policy.app)); + out.push_str("(version 1)\n(allow default)\n"); + if !protected.is_empty() { + out.push_str(&format!(";; the person's data roots are closed\n(deny file-read* file-write*{})\n", subpaths(&protected))); + } + if !metadata.is_empty() { + let lits: String = metadata.iter().map(|p| format!(" (literal {})", quote(p))).collect(); + out.push_str(&format!("(allow file-read-metadata{lits})\n")); + } + if !program.is_empty() { + out.push_str(&format!(";; its program and resources, read-only\n(allow file-read*{})\n", subpaths(&program))); + } + if !ro.is_empty() { + out.push_str(&format!("(allow file-read*{})\n", subpaths(&ro))); + } + out.push_str(&format!(";; its jail, its secrets and external grants\n(allow file-read* file-write*{})\n", subpaths(&rw))); + if policy.network == Network::None { + out.push_str(&format!( + ";; network: the shell's hub only\n(deny network*)\n(allow network-outbound (remote ip \"localhost:{}\"))\n", + policy.hub_port + )); + } + if !policy.processes { + out.push_str(";; no child processes\n(deny process-fork)\n"); + let execs: String = program.iter().map(|p| format!(" (subpath {})", quote(p))).collect(); + out.push_str(&format!("(deny process-exec)\n(allow process-exec{execs})\n")); + } + out +} + +/// Write the profile where the child can be pointed at it (no whitespace: +/// cargo splits its `runner` on spaces). +pub fn write_profile(policy: &Policy) -> Result { + static NEXT: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(1); + let n = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let dir = resolved(&std::env::temp_dir()); + let path = dir.join(format!("octosense-sandbox-{}-{}-{n}.sb", std::process::id(), policy.app)); + if path.to_string_lossy().chars().any(char::is_whitespace) { + return Err(format!("the temp dir {} has whitespace", dir.display())); + } + std::fs::write(&path, profile(policy)).map_err(|e| format!("write {}: {e}", path.display()))?; + Ok(path) +} + +/// The cargo `runner` variable for this host's target triple. +pub fn runner_var() -> &'static str { + if cfg!(target_arch = "aarch64") { + "CARGO_TARGET_AARCH64_APPLE_DARWIN_RUNNER" + } else { + "CARGO_TARGET_X86_64_APPLE_DARWIN_RUNNER" + } +} diff --git a/crates/shell/src/sandbox/mod.rs b/crates/shell/src/sandbox/mod.rs new file mode 100644 index 00000000..0efc15ee --- /dev/null +++ b/crates/shell/src/sandbox/mod.rs @@ -0,0 +1,358 @@ +//! Process sandboxes for native apps hosted in their own process (ADR 0004 +//! §3 and the §11 enforcement table: "the OS sandbox allows the jail, its +//! secrets and `external` only"). +//! +//! [`Policy::for_app`] turns a `native-apps.json` entry (its `sandbox` and +//! `storage` blocks) into one [`Policy`]; [`command`] builds the child's +//! [`Command`] where `clients::spawn_client` starts it. The same policy is +//! built for every app; only its manifest entry differs (the Terminal's is +//! necessarily broad: `home:rw`, processes and any network). +//! +//! | OS | Mechanism | Status | +//! | --- | --- | --- | +//! | macOS | a Seatbelt profile ([`macos`]) run through `/usr/bin/sandbox-exec`; a `cargo run` launch gets it as cargo's `runner`, so only the app is sandboxed, never the build | built and tested | +//! | Linux | Landlock for paths and TCP ports, seccomp for ptrace and friends ([`linux`]), installed between fork and exec; best-effort, logged per layer when the kernel lacks it | built, compile-checked; not run here | +//! | Windows | AppContainer (design below) | **TODO**: not built; a process app runs unsandboxed and the shell says so in its log | +//! +//! **Files.** The person's data roots ([`Policy::protected`]: the home +//! directory and mounted volumes) are closed except the app's jail +//! (`/apps//`), its secrets (`secrets//`) and its +//! reviewed `external` grants. The system's own read-only locations stay +//! readable: a GPU app needs its libraries, fonts, shader caches and the +//! window server. The app's program and resources (its binary, the +//! checkout it was built from, cargo's source cache for crate resources) +//! are readable, never writable. +//! +//! **Network.** `none`: nothing but the shell's hub on loopback (the socket +//! the app is hosted over). `any`: unrestricted. +//! +//! **Child processes.** `processes: false`: no fork and no exec after the +//! app's own start. +//! +//! **Environment.** Whatever the policy, a process app never inherits the +//! kernel's descriptors or tokens: [`scrub_env`] removes every `OCTOS_*` +//! and `OCTOSENSE_*` variable (host and external tokens, the core and +//! kernel directories, the secrets vault) before the child starts, and on +//! macOS a `cargo run` launch drops the ones the checkout's +//! `.cargo/config.toml` `[env]` would hand back (build paths) in its +//! runner. A process app reaches its agent only over the peer link +//! (`peer_link`). +//! +//! **macOS deprecation.** `sandbox-exec` and `sandbox_init` are marked +//! deprecated in Apple's headers (since 10.8) but remain the mechanism the +//! system and major browsers use for helper processes; the supported +//! replacement, App Sandbox entitlements, applies to a whole signed bundle +//! and cannot express a per-app jail under the person's home. When Apple +//! removes it, a process app falls back to running unsandboxed with a +//! logged warning ([`Applied::Unavailable`]), never to failing to start. +//! +//! **Windows design (TODO).** Create one AppContainer profile per app +//! (`CreateAppContainerProfile("OctoSense.")`), grant its SID +//! `FILE_ALL_ACCESS` on the jail and secrets folders and the `external` +//! grants (ACLs, set once when the folders are created), add the +//! `internetClient` capability only for `network: any` (loopback to the hub +//! needs a loopback exemption for the container, `NetworkIsolation...`), +//! start the child with `PROC_THREAD_ATTRIBUTE_SECURITY_CAPABILITIES` +//! through `CreateProcessW` (std's `Command` cannot pass attribute lists, so +//! this needs its own spawn path), and put it in a job object with +//! `JOB_OBJECT_LIMIT_ACTIVE_PROCESS = 1` for `processes: false`. D3D11 +//! shared handles work from an AppContainer. Until then [`command`] reports +//! [`Applied::Unavailable`] on Windows. + +use std::path::{Path, PathBuf}; +use std::process::Command; + +use crate::native_apps::{NativeApp, Network}; + +#[cfg(target_os = "linux")] +pub mod linux; +#[cfg(any(target_os = "macos", test))] +pub mod macos; + +/// Read-only or read-write. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Access { + Read, + ReadWrite, +} + +/// One app's sandbox, built from its manifest entry. +#[derive(Clone, Debug, PartialEq)] +pub struct Policy { + pub app: String, + /// The app's jail, read-write. + pub jail: PathBuf, + /// Its secrets folder, read-write (never inside the jail). + pub secrets: PathBuf, + /// Its reviewed `external` grants. + pub external: Vec<(PathBuf, Access)>, + /// Its program and resources: readable, never writable. + pub program: Vec, + /// The person's data roots: closed except for the grants above. + pub protected: Vec, + pub network: Network, + /// The shell's hub port, reachable on loopback whatever `network` is. + pub hub_port: u16, + pub processes: bool, + /// Host variables a `cargo run` launch would put back from the + /// checkout's `.cargo/config.toml` `[env]` (build paths such as + /// `OCTOSENSE_WORKSPACE`): removed again before the app starts where the + /// platform runs it through a runner ([`cargo_env_host_vars`]). + pub cargo_env_unset: Vec, +} + +/// What [`command`] did. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum Applied { + /// The command now starts the app inside the sandbox. + Sandboxed(String), + /// No sandbox on this platform or kernel; the app runs without one. + Unavailable(String), +} + +/// Where a manifest `external` root is, for the person's home `home`. +fn external_root(root: &str, home: &Path) -> Option { + Some(match root { + "home" => home.to_path_buf(), + "documents" => home.join("Documents"), + "downloads" => home.join("Downloads"), + "desktop" => home.join("Desktop"), + "pictures" => home.join("Pictures"), + "music" => home.join("Music"), + "movies" => home.join("Movies"), + "tmp" => std::env::temp_dir(), + _ => return None, + }) +} + +/// `[/]:ro|rw` (checked by tools/native_apps.py) as a path. +pub fn parse_external(grant: &str, home: &Path) -> Option<(PathBuf, Access)> { + let (place, access) = grant.rsplit_once(':')?; + let access = match access { + "ro" => Access::Read, + "rw" => Access::ReadWrite, + _ => return None, + }; + let (root, rest) = place.split_once('/').unwrap_or((place, "")); + if rest.split('/').any(|part| part == "..") { + return None; + } + let mut path = external_root(root, home)?; + if !rest.is_empty() { + path = path.join(rest); + } + Some((path, access)) +} + +/// The person's home directory. +pub fn person_home() -> Option { + let var = if cfg!(windows) { "USERPROFILE" } else { "HOME" }; + std::env::var_os(var).filter(|h| !h.is_empty()).map(PathBuf::from) +} + +impl Policy { + /// The policy of `app` for a launch whose program and resources live + /// under `program`, hosted on `hub_port`. `home` is the person's home + /// (the protected root external grants are relative to); the jail and + /// secrets come from the host's storage layout. + pub fn for_app(app: &NativeApp, jail: PathBuf, secrets: PathBuf, home: &Path, program: Vec, hub_port: u16) -> Policy { + let external = app.external.iter().filter_map(|grant| parse_external(grant, home)).collect(); + let mut protected = vec![home.to_path_buf()]; + if cfg!(target_os = "macos") { + protected.push(PathBuf::from("/Volumes")); + } else if cfg!(target_os = "linux") { + protected.push(PathBuf::from("/media")); + protected.push(PathBuf::from("/mnt")); + } + Policy { + app: app.id.to_string(), + jail, + secrets, + external, + program, + protected, + network: app.network, + hub_port, + processes: app.processes, + cargo_env_unset: Vec::new(), + } + } + + /// The same policy with the dev narrowing of `OCTOSENSE_SANDBOX_NARROW` + /// (a comma list of app ids): its jail only (no `external`) and no + /// network; child processes as the manifest says. It can only take + /// rights away; a test run uses it to see a broad app (the Terminal, + /// whose shell must still start) refused outside its jail. + pub fn narrowed(mut self) -> Policy { + self.external.clear(); + self.network = Network::None; + self + } + + /// One line for the shell's log. + pub fn summary(&self) -> String { + let external: Vec = self + .external + .iter() + .map(|(p, a)| format!("{}:{}", p.display(), if *a == Access::Read { "ro" } else { "rw" })) + .collect(); + format!( + "{}: files = jail + secrets{}{}, network = {}, processes = {}", + self.app, + if external.is_empty() { "" } else { " + " }, + external.join(" "), + match self.network { + Network::None => format!("hub only (127.0.0.1:{})", self.hub_port), + Network::Any => "any".into(), + }, + if self.processes { "allowed" } else { "none" }, + ) + } +} + +/// Whether `OCTOSENSE_SANDBOX_NARROW` names `app` (see [`Policy::narrowed`]). +pub fn narrowed_by_env(app: &str) -> bool { + std::env::var("OCTOSENSE_SANDBOX_NARROW").map(|v| v.split(',').any(|a| a.trim() == app)).unwrap_or(false) +} + +/// The environment variables no process app may inherit. +pub fn is_host_secret_var(name: &str) -> bool { + let upper = name.to_ascii_uppercase(); + upper.starts_with("OCTOS_") || upper.starts_with("OCTOSENSE_") +} + +/// The host variables (`OCTOS_*`, `OCTOSENSE_*`) a checkout's +/// `.cargo/config.toml` `[env]` sets, which `cargo run` hands the app. +pub fn cargo_env_host_vars(config: &str) -> Vec { + let mut in_env = false; + let mut out = Vec::new(); + for line in config.lines() { + let line = line.trim(); + if line.starts_with('[') { + in_env = line == "[env]"; + continue; + } + if !in_env { + continue; + } + if let Some((name, _)) = line.split_once('=') { + let name = name.trim(); + if is_host_secret_var(name) && name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') { + out.push(name.to_string()); + } + } + } + out +} + +/// Remove every host variable from what `cmd` passes on (ADR 0004 §3: a +/// process app never connects to the kernel and never sees the host token). +pub fn scrub_env(cmd: &mut Command) { + for (name, _) in std::env::vars_os() { + if name.to_str().is_some_and(is_host_secret_var) { + cmd.env_remove(&name); + } + } + let explicit: Vec = + cmd.get_envs().filter(|(k, v)| v.is_some() && k.to_str().is_some_and(is_host_secret_var)).map(|(k, _)| k.to_os_string()).collect(); + for name in explicit { + cmd.env_remove(name); + } +} + +/// The command that starts `program args` under `policy` (`None`: no +/// sandbox, the plain command). `via_cargo`: `program` is cargo and the app +/// is what `cargo run` runs at the end, so only the app is sandboxed where +/// the platform allows (macOS, through cargo's `runner`). +pub fn command(program: &Path, args: &[String], policy: Option<&Policy>, via_cargo: bool) -> (Command, Option) { + let Some(policy) = policy else { + let mut cmd = Command::new(program); + cmd.args(args); + return (cmd, None); + }; + platform_command(program, args, policy, via_cargo) +} + +#[cfg(target_os = "macos")] +fn platform_command(program: &Path, args: &[String], policy: &Policy, via_cargo: bool) -> (Command, Option) { + let plain = || { + let mut cmd = Command::new(program); + cmd.args(args); + cmd + }; + if !Path::new(macos::SANDBOX_EXEC).is_file() { + return (plain(), Some(Applied::Unavailable(format!("{}: {} is missing", policy.app, macos::SANDBOX_EXEC)))); + } + let profile = match macos::write_profile(policy) { + Ok(p) => p, + Err(e) => return (plain(), Some(Applied::Unavailable(format!("{}: {e}", policy.app)))), + }; + let how = if via_cargo { + let mut cmd = plain(); + // cargo re-adds its `[env]` to the app: `env -u` takes the host's + // back out, outside the sandbox, before sandbox-exec starts the app. + let unset: String = policy.cargo_env_unset.iter().map(|v| format!("-u {v} ")).collect(); + let runner = if unset.is_empty() { + format!("{} -f {}", macos::SANDBOX_EXEC, profile.display()) + } else { + format!("/usr/bin/env {unset}{} -f {}", macos::SANDBOX_EXEC, profile.display()) + }; + cmd.env(macos::runner_var(), runner); + (cmd, "as cargo's runner") + } else { + let mut cmd = Command::new(macos::SANDBOX_EXEC); + cmd.arg("-f").arg(&profile).arg(program).args(args); + (cmd, "via sandbox-exec") + }; + (how.0, Some(Applied::Sandboxed(format!("{} ({}, profile {})", policy.summary(), how.1, profile.display())))) +} + +#[cfg(target_os = "linux")] +fn platform_command(program: &Path, args: &[String], policy: &Policy, via_cargo: bool) -> (Command, Option) { + let mut cmd = Command::new(program); + cmd.args(args); + let applied = linux::apply(&mut cmd, policy, via_cargo); + (cmd, Some(applied)) +} + +#[cfg(not(any(target_os = "macos", target_os = "linux")))] +fn platform_command(program: &Path, args: &[String], policy: &Policy, _via_cargo: bool) -> (Command, Option) { + let mut cmd = Command::new(program); + cmd.args(args); + let why = format!("{}: no process sandbox on this platform yet (Windows AppContainer is a TODO, sandbox/mod.rs); running unsandboxed", policy.app); + (cmd, Some(Applied::Unavailable(why))) +} + +/// Every ancestor of `path`, root first (their metadata is readable so the +/// path itself resolves). +pub fn ancestors(path: &Path) -> Vec { + let mut out: Vec = path.ancestors().skip(1).map(Path::to_path_buf).collect(); + out.reverse(); + out +} + +/// A path with symlinks resolved as far as it exists (macOS: `/tmp` is +/// `/private/tmp`, and a sandbox profile matches resolved paths). +pub fn resolved(path: &Path) -> PathBuf { + let mut existing = path.to_path_buf(); + let mut rest = Vec::new(); + loop { + if let Ok(real) = existing.canonicalize() { + let mut out = real; + for part in rest.iter().rev() { + out.push(part); + } + return out; + } + match (existing.parent(), existing.file_name()) { + (Some(parent), Some(name)) => { + rest.push(name.to_os_string()); + existing = parent.to_path_buf(); + } + _ => return path.to_path_buf(), + } + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/shell/src/sandbox/tests.rs b/crates/shell/src/sandbox/tests.rs new file mode 100644 index 00000000..a6383389 --- /dev/null +++ b/crates/shell/src/sandbox/tests.rs @@ -0,0 +1,198 @@ +//! The sandbox, proved on this machine: a child started through +//! [`command`] under a jail-only policy reads its jail and is refused a +//! file beside it, a network connection other than the hub, and a child +//! process; and no child inherits a host token. + +#![cfg_attr(not(unix), allow(unused))] + +use super::*; +use std::io::Read; +use std::net::TcpListener; + +struct Scratch(PathBuf); + +impl Scratch { + fn new(tag: &str) -> Scratch { + let dir = std::env::temp_dir().join(format!("octosense-sandbox-test-{}-{tag}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(dir.join("apps/probe")).unwrap(); + std::fs::create_dir_all(dir.join("secrets/probe")).unwrap(); + Scratch(dir) + } +} + +impl Drop for Scratch { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +fn jail_only(root: &Path, hub_port: u16) -> Policy { + let app = crate::native_apps::NativeApp { + id: "probe", + feature: "app-probe", + bin: Some("probe"), + macos: crate::native_apps::Hosting::Process, + windows: crate::native_apps::Hosting::Process, + linux: crate::native_apps::Hosting::Process, + android: crate::native_apps::Hosting::Module, + ios: crate::native_apps::Hosting::Module, + ohos: crate::native_apps::Hosting::Module, + wasm: crate::native_apps::Hosting::Module, + octos: &[], + tools: &[], + network: Network::None, + processes: false, + accounts: false, + external: &[], + }; + // The scratch root stands in for the person's home: closed but for the + // jail. The probe's tools are its program. + Policy::for_app(&app, root.join("apps/probe"), root.join("secrets/probe"), root, vec!["/bin".into(), "/usr/bin".into()], hub_port) +} + +fn run(program: &str, args: &[&str], policy: &Policy) -> (bool, String) { + let args: Vec = args.iter().map(|s| s.to_string()).collect(); + let (mut cmd, applied) = command(Path::new(program), &args, Some(policy), false); + assert!(matches!(applied, Some(Applied::Sandboxed(_))), "{applied:?}"); + let out = cmd.output().expect("the probe starts"); + (out.status.success(), format!("{}{}", String::from_utf8_lossy(&out.stdout), String::from_utf8_lossy(&out.stderr))) +} + +fn sandbox_works_here() -> bool { + if cfg!(target_os = "macos") { + return Path::new(macos::SANDBOX_EXEC).is_file(); + } + #[cfg(target_os = "linux")] + return linux::landlock_abi() > 0; + #[allow(unreachable_code)] + false +} + +#[cfg(unix)] +#[test] +fn a_jail_only_app_reads_its_jail_and_is_refused_everything_else() { + if !sandbox_works_here() { + eprintln!("no process sandbox on this machine; skipped"); + return; + } + let scratch = Scratch::new("jail"); + let root = &scratch.0; + std::fs::write(root.join("apps/probe/mine.txt"), "mine").unwrap(); + std::fs::write(root.join("secrets/probe/key"), "its own secret").unwrap(); + std::fs::write(root.join("outside.txt"), "not yours").unwrap(); + std::fs::create_dir_all(root.join("apps/other")).unwrap(); + std::fs::write(root.join("apps/other/theirs.txt"), "another app's").unwrap(); + let hub = TcpListener::bind("127.0.0.1:0").unwrap(); + let other = TcpListener::bind("127.0.0.1:0").unwrap(); + let hub_port = hub.local_addr().unwrap().port(); + let other_port = other.local_addr().unwrap().port(); + let policy = jail_only(root, hub_port); + + let cat = |p: PathBuf| run("/bin/cat", &[p.to_str().unwrap()], &policy); + let (ok, out) = cat(root.join("apps/probe/mine.txt")); + assert!(ok && out.contains("mine"), "its jail: {out}"); + let (ok, out) = cat(root.join("secrets/probe/key")); + assert!(ok, "its secrets: {out}"); + let (ok, out) = cat(root.join("outside.txt")); + assert!(!ok && !out.contains("not yours"), "a file outside the jail is refused: {out}"); + let (ok, out) = cat(root.join("apps/other/theirs.txt")); + assert!(!ok && !out.contains("another app's"), "another app's jail is refused: {out}"); + let (ok, _) = run("/bin/sh", &["-c", &format!("echo x > {}/apps/probe/written", root.display())], &Policy { processes: true, ..policy.clone() }); + assert!(ok, "it writes in its jail"); + let (ok, out) = run("/bin/sh", &["-c", &format!("echo x > {}/written", root.display())], &Policy { processes: true, ..policy.clone() }); + assert!(!ok, "it writes nothing outside: {out}"); + + // Network: the hub on loopback, nothing else. + let nc = |port: u16| run("/usr/bin/nc", &["-z", "-w", "2", "127.0.0.1", &port.to_string()], &policy); + let (ok, out) = nc(hub_port); + assert!(ok, "the hub is reachable: {out}"); + let (ok, out) = nc(other_port); + assert!(!ok, "any other connection is refused: {out}"); + + // Processes: none. + let (ok, out) = run("/bin/sh", &["-c", "/bin/echo first; /bin/echo spawned"], &policy); + assert!(!ok || !out.contains("spawned"), "no child process: {out}"); + drop((hub, other)); +} + +#[test] +fn the_terminals_manifest_gives_a_broad_sandbox_and_narrowing_only_takes_away() { + let terminal = crate::native_apps::find("terminal").unwrap(); + let home = Path::new("/home/person"); + let p = Policy::for_app(terminal, "/h/apps/terminal".into(), "/h/secrets/terminal".into(), home, vec![], 8765); + assert_eq!(p.external, vec![(home.to_path_buf(), Access::ReadWrite)]); + assert_eq!((p.network, p.processes), (Network::Any, true)); + let n = p.clone().narrowed(); + assert!(n.external.is_empty()); + assert_eq!((n.network, n.processes), (Network::None, true), "narrowing keeps the shell's processes"); + assert_eq!((n.jail, n.secrets), (p.jail, p.secrets), "narrowing never moves the jail"); + let reference = crate::native_apps::find("reference").unwrap(); + let r = Policy::for_app(reference, "/h/apps/reference".into(), "/h/secrets/reference".into(), home, vec![], 8765); + assert!(r.external.is_empty()); + assert_eq!((r.network, r.processes), (Network::None, false)); +} + +#[test] +fn external_grants_parse_under_their_roots_only() { + let home = Path::new("/home/person"); + assert_eq!(parse_external("home:rw", home), Some((home.to_path_buf(), Access::ReadWrite))); + assert_eq!(parse_external("documents/Notes:ro", home), Some((home.join("Documents/Notes"), Access::Read))); + assert_eq!(parse_external("home/../etc:ro", home), None); + assert_eq!(parse_external("/etc:ro", home), None); + assert_eq!(parse_external("home:rwx", home), None); +} + +#[test] +fn the_macos_profile_closes_the_roots_then_opens_the_grants() { + let p = jail_only(Path::new("/nonexistent/home"), 8765); + let text = macos::profile(&p); + let deny = text.find("(deny file-read* file-write* (subpath \"/nonexistent/home\")").expect(&text); + let jail = text.find("(allow file-read* file-write* (subpath \"/nonexistent/home/apps/probe\")").expect(&text); + assert!(deny < jail, "the last matching rule wins: grants come after the close\n{text}"); + assert!(text.contains("(allow file-read-metadata (literal \"/nonexistent/home\") (literal \"/nonexistent/home/apps\")"), "{text}"); + assert!(text.contains("(allow network-outbound (remote ip \"localhost:8765\"))"), "{text}"); + assert!(text.contains("(deny process-fork)"), "{text}"); + let broad = Policy { network: Network::Any, processes: true, ..p }; + let text = macos::profile(&broad); + assert!(!text.contains("network") && !text.contains("process-"), "{text}"); +} + +#[test] +fn cargos_env_block_host_variables_are_taken_back_out() { + let config = "[env]\nMAKEPAD_BUNDLE_NAME = { value = \"OctoSense\" }\nOCTOSENSE_WORKSPACE = { value = \".sources\", relative = true }\nOCTOS_X = \"1\"\n[target.x]\nOCTOSENSE_NOT_ENV = 1\n"; + assert_eq!(cargo_env_host_vars(config), vec!["OCTOSENSE_WORKSPACE".to_string(), "OCTOS_X".to_string()]); + // The repository's own config is read the same way. + let repo = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/../../.cargo/config.toml")).unwrap_or_default(); + for name in cargo_env_host_vars(&repo) { + assert!(is_host_secret_var(&name)); + } +} + +#[cfg(unix)] +#[test] +fn no_child_inherits_a_host_token_or_a_kernel_path() { + assert!(is_host_secret_var("OCTOS_AUTH_TOKEN")); + assert!(is_host_secret_var("OCTOS_HOST_EXTERNAL_TOKEN")); + assert!(is_host_secret_var("OCTOS_APP_CORE_DIR")); + assert!(is_host_secret_var("OCTOSENSE_SECRETS")); + assert!(!is_host_secret_var("STUDIO_HOST")); + assert!(!is_host_secret_var("HOME")); + let mut cmd = Command::new("/usr/bin/env"); + // As if the shell's own environment held them (the kernel's), plus one + // set on the command itself. + cmd.env("OCTOS_AUTH_TOKEN", "host-token") + .env("OCTOS_APP_CORE_DIR", "/core") + .env("OCTOSENSE_SECRETS", "/vault") + .env("STUDIO_HOST", "http://127.0.0.1:8765"); + scrub_env(&mut cmd); + let mut out = String::new(); + let mut child = cmd.stdout(std::process::Stdio::piped()).spawn().unwrap(); + child.stdout.take().unwrap().read_to_string(&mut out).unwrap(); + child.wait().unwrap(); + assert!(out.contains("STUDIO_HOST="), "{out}"); + for line in out.lines() { + let name = line.split('=').next().unwrap_or(""); + assert!(!is_host_secret_var(name), "the child sees {line}"); + } +} diff --git a/tools/native_apps.py b/tools/native_apps.py index cafaefb3..473cbe82 100644 --- a/tools/native_apps.py +++ b/tools/native_apps.py @@ -161,6 +161,10 @@ def validate(data): for key in ("sandbox", "storage", "agent"): if not isinstance(app[key], dict): problems.append(f"{where}: {key} must be an object") + if isinstance(app["sandbox"], dict): + problems += [f"{where}: {p}" for p in sandbox_problems(app["sandbox"])] + if isinstance(app["storage"], dict): + problems += [f"{where}: {p}" for p in storage_problems(app["storage"])] if isinstance(app["agent"], dict) and not isinstance(app["agent"].get("octos"), list): problems.append(f"{where}: agent.octos must be a list") if isinstance(app["agent"], dict): @@ -170,6 +174,40 @@ def validate(data): return data["apps"] +NETWORKS = ("none", "any") +EXTERNAL_ROOTS = ("home", "documents", "downloads", "desktop", "pictures", "music", "movies", "tmp") + + +def sandbox_problems(sandbox): + """`sandbox`: {"network": "none"|"any", "processes": bool} (ADR 0004 §3): + what a process-hosted instance's OS sandbox allows beyond its files.""" + problems = [] + if set(sandbox) != {"network", "processes"}: + problems.append("sandbox needs exactly network and processes") + if sandbox.get("network") not in NETWORKS: + problems.append(f"sandbox.network must be one of {', '.join(NETWORKS)}") + if not isinstance(sandbox.get("processes"), bool): + problems.append("sandbox.processes must be true or false") + return problems + + +def storage_problems(storage): + """`storage` (ADR 0004 §11): `accounts` a bool, `external` a list of + `[/]:ro|rw` grants outside the jail (part of the sandbox, + never of an agent's workspace).""" + problems = [] + if not isinstance(storage.get("accounts"), bool): + problems.append("storage.accounts must be true or false") + external = storage.get("external") + if not isinstance(external, list): + return problems + ["storage.external must be a list"] + for grant in external: + match = isinstance(grant, str) and re.fullmatch(r"([a-z]+)(/[A-Za-z0-9._/-]+)?:(ro|rw)", grant) + if not match or match.group(1) not in EXTERNAL_ROOTS or ".." in grant: + problems.append(f"storage.external: {grant!r} is not [/]:ro|rw with root one of {', '.join(EXTERNAL_ROOTS)}") + return problems + + def tool_policy_problems(policy): """`agent.tool_policy`: {tool: {"confirm": "host"|"app", "auto_approvable": bool}} (ADR 0004 §8). A tool the shell confirms live and no standing rule may @@ -355,6 +393,14 @@ def render_rust(apps): " pub auto_approvable: bool,", "}", "", + "/// What a process-hosted instance may reach on the network (ADR 0004 §3).", + "#[derive(Clone, Copy, Debug, PartialEq, Eq)]", + "pub enum Network {", + " /// Nothing but the shell's hub on loopback.", + " None,", + " Any,", + "}", + "", "/// One `native-apps.json` entry, as far as the shell reads it.", "#[derive(Debug)]", "pub struct NativeApp {", @@ -374,6 +420,14 @@ def render_rust(apps): " pub octos: &'static [&'static str],", " /// Its assistant tools' confirmation rules (`agent.tool_policy`).", " pub tools: &'static [ToolPolicy],", + " /// `sandbox.network`: what its OS sandbox lets it reach.", + " pub network: Network,", + " /// `sandbox.processes`: whether it may start child processes.", + " pub processes: bool,", + " /// `storage.accounts`: one folder (and agent) per account.", + " pub accounts: bool,", + " /// `storage.external`: `[/]:ro|rw` outside its jail.", + " pub external: &'static [&'static str],", "}", "", "pub const APPS: &[NativeApp] = &[", @@ -400,6 +454,12 @@ def render_rust(apps): out.append(" ],") else: out.append(" tools: &[],") + network = "Network::Any" if app["sandbox"]["network"] == "any" else "Network::None" + out.append(f" network: {network},") + out.append(f" processes: {'true' if app['sandbox']['processes'] else 'false'},") + out.append(f" accounts: {'true' if app['storage']['accounts'] else 'false'},") + external = ", ".join(s(x) for x in app["storage"]["external"]) + out.append(f" external: &[{external}],") out.append(" },") out += [ "];", diff --git a/tools/test_native_apps.py b/tools/test_native_apps.py index e0853b6a..2f25103c 100644 --- a/tools/test_native_apps.py +++ b/tools/test_native_apps.py @@ -103,6 +103,17 @@ def test_tool_policy_is_checked(self): self.app("terminal")["agent"]["tool_policy"] = {"run": {"confirm": "host"}} self.assertRefused(r"tool_policy\.run: needs exactly confirm and auto_approvable") + def test_sandbox_and_storage_are_checked(self): + self.app("terminal")["sandbox"] = {"network": "some", "processes": "yes"} + self.assertRefused(r"sandbox\.network must be one of none, any") + self.app("terminal")["sandbox"] = {"network": "any", "processes": True, "gpu": True} + self.assertRefused(r"sandbox needs exactly network and processes") + self.app("terminal")["sandbox"] = {"network": "any", "processes": True} + self.app("terminal")["storage"]["external"] = ["home:rw", "/etc:ro"] + self.assertRefused(r"storage\.external: '/etc:ro'") + self.app("terminal")["storage"]["external"] = ["home/../x:rw"] + self.assertRefused(r"storage\.external") + def test_the_terminals_commands_are_host_confirmed_and_never_auto_approved(self): self.assertEqual(self.app("terminal")["agent"]["tool_policy"], {"run": {"confirm": "host", "auto_approvable": False}}) From 5424151d3e80067382a0cc99905e22cd07285d3e Mon Sep 17 00:00:00 2001 From: ymote <151983+ymote@users.noreply.github.com> Date: Mon, 28 Sep 2026 17:37:11 -0700 Subject: [PATCH 2/2] shell: the peer link for process-hosted native apps [skip ci] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ADR 0004 §5. peer_link/ gives a process app its own channel to its octos app agent on its hub socket ("octos_peer" frames, never the AI bus): requests on its one peer through the app-peers service in-process modules get (an adapter until octos#2567's relay), and the tool-call half behind seams for the relay (set_tool_relay, tool_call, tool_cancel, context_owner, has_link, close_account). Identity is the socket's slot; a call's account, context and client come from the shell's own context records. Links open only for apps whose native-apps.json entry grants agent.octos, and requests need consent. The shell keeps the host obligations (once per call, nothing after cancel, acknowledge before any sheet; confirmations through the #120 router, whose decisions get their own peerlink: route), and a dead process fails its calls (outcome_unknown unless read-only), closes its contexts and keeps the peer. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/shell/src/approvals/mod.rs | 17 +- crates/shell/src/lib.rs | 60 ++- crates/shell/src/peer_link/link.rs | 595 ++++++++++++++++++++++++++++ crates/shell/src/peer_link/mod.rs | 221 +++++++++++ crates/shell/src/peer_link/tests.rs | 454 +++++++++++++++++++++ crates/shell/src/peer_link/wire.rs | 196 +++++++++ 6 files changed, 1537 insertions(+), 6 deletions(-) create mode 100644 crates/shell/src/peer_link/link.rs create mode 100644 crates/shell/src/peer_link/mod.rs create mode 100644 crates/shell/src/peer_link/tests.rs create mode 100644 crates/shell/src/peer_link/wire.rs diff --git a/crates/shell/src/approvals/mod.rs b/crates/shell/src/approvals/mod.rs index 037ed2e3..f43cc450 100644 --- a/crates/shell/src/approvals/mod.rs +++ b/crates/shell/src/approvals/mod.rs @@ -63,6 +63,9 @@ pub struct Approvals { /// Decisions for the AI bus's held calls (`bus:` ids), drained by the /// shell ([`take_bus_decisions`]). bus: RecordingRelay, + /// Decisions for the peer link's calls (`peerlink:` ids), drained by + /// `peer_link` ([`take_peer_decisions`]). + peer: RecordingRelay, external: Arc>>>, } @@ -70,6 +73,7 @@ pub struct Approvals { /// shell, the rest to the octos#2567 relay (queued until it is installed). struct Dispatch { bus: RecordingRelay, + peer: RecordingRelay, queue: RecordingRelay, external: Arc>>>, } @@ -79,6 +83,9 @@ impl ApprovalRelay for Dispatch { if id.0.starts_with(crate::ai_bus::HELD_PREFIX) { return self.bus.approval_decided(id, decision, reason); } + if id.0.starts_with(crate::peer_link::link::HELD_PREFIX) { + return self.peer.approval_decided(id, decision, reason); + } match self.external.lock().unwrap_or_else(|e| e.into_inner()).as_mut() { Some(relay) => relay.approval_decided(id, decision, reason), None => self.queue.approval_decided(id, decision, reason), @@ -97,10 +104,11 @@ impl Approvals { fn with_parts(rules: rules::RuleStore, audit: audit::AuditLog, consent: consent::ConsentStore, contacts: contacts::ContactsGate) -> Approvals { let queue = RecordingRelay::default(); let bus = RecordingRelay::default(); + let peer = RecordingRelay::default(); let external = Arc::new(Mutex::new(None)); - let dispatch = Dispatch { bus: bus.clone(), queue: queue.clone(), external: external.clone() }; + let dispatch = Dispatch { bus: bus.clone(), peer: peer.clone(), queue: queue.clone(), external: external.clone() }; let router = Router::new(rules, audit, Box::new(dev_hooks::ShellDevMode), contacts, Box::new(dispatch)); - Approvals { router, consent, settings_open: false, queue, bus, external } + Approvals { router, consent, settings_open: false, queue, bus, peer, external } } /// Sheets, rules, consent and the page: one number for "redraw". pub fn generation(&self) -> u64 { @@ -174,6 +182,11 @@ pub fn take_bus_decisions() -> Vec<(RequestId, Decision, String)> { with(|a| a.bus.take()).unwrap_or_default() } +/// The peer link's decisions (`peer_link::link::HELD_PREFIX` ids). +pub fn take_peer_decisions() -> Vec<(RequestId, Decision, String)> { + with(|a| a.peer.take()).unwrap_or_default() +} + /// An app module registers its own confirmation sheet (`confirm: app`). pub fn register_app_confirm(app: &str, handler: Box) { with(|a| a.router.register_app_confirm(app, handler)); diff --git a/crates/shell/src/lib.rs b/crates/shell/src/lib.rs index 1036c491..770a6f55 100644 --- a/crates/shell/src/lib.rs +++ b/crates/shell/src/lib.rs @@ -50,7 +50,9 @@ pub mod module_host; mod module_panic_tests; pub mod module_view; pub mod native_apps; +pub mod sandbox; pub mod pane_links; +pub mod peer_link; pub mod preview; pub mod run_view; pub mod shell; @@ -1541,8 +1543,9 @@ impl App { fn request_close(&mut self, cx: &mut Cx, client: ClientId) { // A module instance has no process to ask politely and nothing to // reap later: it ends now, through the same removal as a death. - // So does one of its extra windows. - if self.module_host.is_module(client) || self.module_windows.contains_key(&client) { + // So does one of its extra windows, and a stopped process's tile. + let stopped = self.state.as_ref().and_then(|s| s.clients.get(&client)).is_some_and(|s| s.stopped); + if stopped || self.module_host.is_module(client) || self.module_windows.contains_key(&client) { self.remove_client(cx, client); self.update_bar(cx); return; @@ -1595,8 +1598,33 @@ impl App { .unwrap_or(false) } + /// A process app died unexpectedly: everything behind its tile goes + /// (the bus's registrations, its peer link's calls and contexts, the + /// process handle) and the tile stays, closed with a Restart. + fn process_stopped(&mut self, cx: &mut Cx, client: ClientId) { + let Some(slot) = self.state_mut().clients.get_mut(&client) else { return }; + slot.stopped = true; + slot.sender = None; + slot.socket = None; + if let Some(mut child) = slot.child.take() { + let _ = child.wait(); + } + let app = slot.app.clone(); + let label = clients::find_app(&app).map(|a| a.label).unwrap_or_else(|| app.clone()); + log!("wm: {app} (client {client}) stopped; its tile stays closed with Restart"); + peer_link::process_gone(client); + if let Some(bye) = self.ai_bus.client_died(client) { + self.send_to_pane(bye); + } + self.pane_links.close_instance(client); + self.desk(cx).borrow_mut::().map(|mut d| d.with_run_view(cx, client, |cx, v| v.show_stopped(cx, client, &label))); + self.update_bar(cx); + self.redraw_all(cx); + } + fn remove_client(&mut self, cx: &mut Cx, client: ClientId) { log!("wm: removing client {}", client); + peer_link::process_gone(client); // An instance's extra window: the tile lets go of the root (the // instance keeps the widget), and the instance hears it was closed. if let Some((owner, key)) = self.module_windows.remove(&client) { @@ -1764,6 +1792,7 @@ impl App { // Warm instances excluded: nobody is watching a tile that // does not exist, and asking for one would build it. !s.warm + && !s.stopped && s.linked && s.sender.is_none() && s.linked_at.map(|t| host::now() - t > GRACE).unwrap_or(false) @@ -1821,7 +1850,14 @@ impl App { }) .collect(); for (id, failure) in dead { - self.remove_client(cx, id); + // ADR 0004 §2: a process app that dies takes only itself down; + // its tile stays, closed with a Restart. + let in_place = self.state_mut().clients.get(&id).is_some_and(|s| s.stops_in_place(failure.is_some())); + if in_place { + self.process_stopped(cx, id); + } else { + self.remove_client(cx, id); + } if let Some(message) = failure { log!("octosense: {message}"); self.notify(cx, "App stopped", &message); @@ -2210,7 +2246,8 @@ impl App { /// Restart on a failed module's tile: that window closes and the app /// opens afresh, as a launch from the menu would open it. fn restart_failed_module(&mut self, cx: &mut Cx, client: ClientId) { - if !self.module_host.is_failed(client) { + let stopped = self.state.as_ref().and_then(|s| s.clients.get(&client)).is_some_and(|s| s.stopped); + if !self.module_host.is_failed(client) && !stopped { return; } let Some(app) = self.state.as_ref().and_then(|s| s.clients.get(&client)).map(|slot| slot.app.clone()) else { @@ -2536,6 +2573,9 @@ impl App { .to_string_lossy() .to_string(); if let Some(slot) = self.state_mut().clients.get_mut(&client) { + // The peer link (ADR 0004 §5): the socket is this + // slot's app; granted apps get a link on it. + peer_link::connected(client, &slot.app, sender.clone()); slot.sender = Some(sender); slot.socket = Some(socket); if let Some(sender) = &slot.sender { @@ -2567,6 +2607,7 @@ impl App { .find(|(_, s)| s.socket == Some(socket)) .map(|(id, _)| *id); if let Some(client) = client { + peer_link::process_gone(client); if let Some(slot) = self.state_mut().clients.get_mut(&client) { slot.sender = None; slot.socket = None; @@ -2694,6 +2735,16 @@ impl App { self.sync_phone_keyboard(cx); return; } + // The peer link's own channel (ADR 0004 §5): never the + // bus's. Its app is the socket's slot, never the frame's. + if peer_link::wire::is_peer_frame(&json) { + let (app, sender) = match self.state_mut().clients.get(&client) { + Some(slot) => (slot.app.clone(), slot.sender.clone()), + None => return, + }; + peer_link::on_frame(client, &app, &json, sender); + return; + } // The typed app<->WM vocabulary (libs/wm_api) first; what // is not the WM's own envelope is the AI bus's (or noise). if let Some(req) = WmRequest::parse(&json) { @@ -3121,6 +3172,7 @@ impl App { /// must be told (every automatic approval among it). fn approvals_tick(&mut self, cx: &mut Cx) { approvals::tick(); + peer_link::tick(); self.approvals_changed(cx); } diff --git a/crates/shell/src/peer_link/link.rs b/crates/shell/src/peer_link/link.rs new file mode 100644 index 00000000..96feb711 --- /dev/null +++ b/crates/shell/src/peer_link/link.rs @@ -0,0 +1,595 @@ +//! The peer links of every process app: identity, grants and consent, +//! request contexts, the tool-call half and its host obligations, and what +//! a process's death does. No I/O of its own: frames go out through each +//! link's [`FrameOut`], the shell's decisions come from a [`PeerHost`], and +//! tool results go to the [`ToolRelay`] (octos#2567's relay, home-96's). + +use std::collections::{BTreeSet, HashMap}; +use std::sync::Arc; + +use serde_json::{json, Map, Value}; + +use super::wire::{self, Down, Outcome, Risk, ToolCallDown, Up}; +use crate::ai_host::app_peers::{ContextEvent, ContextOp, ContextSpec, OctosAppService, OctosContext, OCTOS_SERVICES}; +use crate::approvals::{Caller, Decision, RequestContext, RequestId, Route, ToolSpec, Trigger}; +use crate::hub::ClientId; +use crate::native_apps::Confirm; + +/// Sends one peer frame (JSON) to one app process. Called from provider +/// threads too (streamed turn events), so it is `Send + Sync`. +pub type FrameOut = Arc; + +/// The account of an app that keeps no accounts (ADR 0004 §11: "device"). +pub const DEVICE: &str = "device"; +/// The approval ids the peer link asks the router with. +pub const HELD_PREFIX: &str = "peerlink:"; + +/// What the link needs from the rest of the shell. [`super::ShellHost`] is +/// the real one; the tests have their own. +pub trait PeerHost: Send { + /// The `octos.*` services the app's reviewed `native-apps.json` entry + /// grants it (every one in developer mode); empty: no peer link. + fn granted(&self, app: &str) -> BTreeSet; + /// `storage.accounts`: the app names the account a context acts for. + fn keeps_accounts(&self, app: &str) -> bool; + /// `consent::granted(app)`; when undecided, the first-use sheet is + /// shown (once) and this answers false until the person allows it. + fn consent(&mut self, app: &str) -> bool; + /// The app's assistant service (its one peer), made once and kept by + /// the link across the process's restarts. + fn service(&mut self, app: &str, services: &BTreeSet) -> Option>; + /// `agent.tool_policy` for one tool. + fn tool_rule(&self, app: &str, tool: &str) -> Option<(Confirm, bool)>; + /// The #120 router: a call needs the person (or a rule, or developer mode). + fn request_approval(&mut self, app: &str, tool: ToolSpec, args: Value, caller: Caller, context: RequestContext) -> Route; + /// The router's decisions on this link's requests since the last call. + fn take_decisions(&mut self) -> Vec<(RequestId, Decision, String)>; + /// The owning app's own sheet answered a `confirm: app` request. + fn app_confirm_answered(&mut self, id: &RequestId, approved: bool, reason: &str); + /// A link opened (the app's own sheet can take `confirm: app` requests) or closed. + fn link_opened(&mut self, app: &str); + fn link_closed(&mut self, app: &str); +} + +/// A call's final answer, as the relay hands it to the kernel. +#[derive(Clone, Debug, PartialEq)] +pub enum ToolCallResult { + Ok(Value), + Error(String), + /// The call may or may not have happened (the process died with it in + /// flight): octos marks it `outcome_unknown`, never retried without the person. + OutcomeUnknown, +} + +/// The seam home-96's octos#2567 relay implements: where tool-call +/// outcomes go. Each call gets at most one `acknowledged` and exactly one +/// `finished`. +pub trait ToolRelay: Send { + /// The host took the call and a confirmation may follow (sent before + /// any confirmation sheet is shown). + fn acknowledged(&mut self, app: &str, call_id: &str); + fn finished(&mut self, app: &str, call_id: &str, result: ToolCallResult); +} + +/// The test double, and the queue until a relay is installed. +#[derive(Clone, Default)] +pub struct RecordingToolRelay { + pub events: Arc)>>>, +} + +impl RecordingToolRelay { + pub fn take(&self) -> Vec<(String, String, Option)> { + std::mem::take(&mut *self.events.lock().unwrap_or_else(|e| e.into_inner())) + } +} + +impl ToolRelay for RecordingToolRelay { + fn acknowledged(&mut self, app: &str, call_id: &str) { + self.events.lock().unwrap_or_else(|e| e.into_inner()).push((app.into(), call_id.into(), None)); + } + fn finished(&mut self, app: &str, call_id: &str, result: ToolCallResult) { + self.events.lock().unwrap_or_else(|e| e.into_inner()).push((app.into(), call_id.into(), Some(result))); + } +} + +/// A kernel `peer/tool/call` for an app, as the relay hands it over. It +/// carries the kernel's `context_id`, never an account or client: the +/// shell stamps those from its own records. +#[derive(Clone, Debug, PartialEq)] +pub struct KernelToolCall { + pub call_id: String, + pub name: String, + pub args: Value, + pub risk: Risk, + pub timeout_ms: u64, + /// The kernel's request context (`None`: the agent's own session). + pub context_id: Option, + /// Who is calling, from the relay's own connection records. + pub caller: Caller, + pub trigger: Trigger, + pub outcome_unknown: bool, +} + +/// Why a call was not taken. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum Refused { + /// No process of the app holds a peer link now. + NotConnected, + /// The same call id again: once per call. + Duplicate, + /// A context this app never opened (or one already closed). + UnknownContext, + /// The router refused it outright. + Declined(String), +} + +/// Who owns a request context, as the shell recorded it at open. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ContextOwner { + pub app: String, + pub client_id: ClientId, + pub account: String, + pub client: Option, +} + +struct Link { + app: String, + out: FrameOut, + /// Contexts this process opened (its handles). + contexts: Vec, +} + +struct Ctx { + owner: ContextOwner, + ctx: Arc, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum CallState { + /// Waiting for the shell's sheet (nothing sent to the app yet). + Held, + /// Sent to the app. + Sent, +} + +struct Call { + client_id: ClientId, + risk: Risk, + state: CallState, + confirm_required: bool, + acked: bool, + /// The approval id, when the router was asked. + approval: Option, + deadline: f64, + down: ToolCallDown, +} + +/// Every process app's peer link. +pub struct PeerLinks { + host: Box, + relay: Box, + links: HashMap, + /// One service (peer) per app, kept when its process dies. + services: HashMap>, + contexts: HashMap, + calls: HashMap<(String, String), Call>, + next_context: u64, + /// Lines for the shell's log (drained by the caller). + pub log: Vec, +} + +fn caller_wire(caller: &Caller) -> String { + match caller { + Caller::OwnAgent { .. } => "own_agent".into(), + Caller::AppAgent { app } => format!("app:{app}"), + Caller::SystemAgent => "system_agent".into(), + } +} + +impl PeerLinks { + pub fn new(host: Box, relay: Box) -> PeerLinks { + PeerLinks { + host, + relay, + links: HashMap::new(), + services: HashMap::new(), + contexts: HashMap::new(), + calls: HashMap::new(), + next_context: 1, + log: Vec::new(), + } + } + + pub fn set_relay(&mut self, relay: Box) { + self.relay = relay; + } + + /// A process connected its hub socket. `app` is the app the shell + /// launched on that socket (its client slot), never a claim of the + /// process. A link opens only for an app granted assistant services. + pub fn connected(&mut self, client_id: ClientId, app: &str, out: FrameOut) -> bool { + if self.host.granted(app).is_empty() { + return false; + } + self.links.insert(client_id, Link { app: app.to_string(), out, contexts: Vec::new() }); + self.host.link_opened(app); + self.log.push(format!("peer link: {app} (client {client_id}) opened")); + true + } + + pub fn has_link(&self, app: &str) -> bool { + self.links.values().any(|l| l.app == app) + } + + /// A peer frame from `client_id`. `app` is the app of that socket. The + /// frame is consumed (true) whenever it carries the peer envelope, so it + /// never reaches the AI bus. + pub fn on_frame(&mut self, client_id: ClientId, app: &str, frame: &str, reply_out: Option) -> bool { + if !wire::is_peer_frame(frame) { + return false; + } + let Some(up) = Up::parse(frame) else { + self.log.push(format!("peer link: {app} (client {client_id}): malformed frame dropped")); + return true; + }; + match up { + Up::Request { req_id, method, args } => self.on_request(client_id, app, req_id, &method, args, reply_out), + Up::ToolResult { call_id, outcome } => self.on_tool_result(client_id, app, &call_id, outcome), + } + true + } + + fn reply(out: &FrameOut, req_id: u64, result: Result) { + out(Down::Reply { req_id, result }.to_json()); + } + + fn on_request(&mut self, client_id: ClientId, app: &str, req_id: u64, method: &str, args: Map, reply_out: Option) { + let Some(link) = self.links.get(&client_id) else { + // No link: the app has no granted agent. Answer so it can say so. + if let Some(out) = reply_out { + Self::reply(&out, req_id, Err("no_agent: this app is not granted an agent".into())); + } + return; + }; + // The socket's app, never the frame's. + debug_assert_eq!(link.app, app); + let out = link.out.clone(); + let app = link.app.clone(); + let granted = self.host.granted(&app); + if method != "octos.context.close" && !granted.contains(method) { + return Self::reply(&out, req_id, Err(format!("not_granted: {method}"))); + } + if !self.host.consent(&app) { + return Self::reply(&out, req_id, Err("consent_pending: the person has not allowed this app's agent".into())); + } + let services: BTreeSet = granted.iter().filter(|s| OCTOS_SERVICES.contains(&s.as_str())).cloned().collect(); + let service = match self.services.get(&app) { + Some(s) => s.clone(), + None => match self.host.service(&app, &services) { + Some(s) => { + self.services.insert(app.clone(), s.clone()); + s + } + None => return Self::reply(&out, req_id, Err("unavailable: no assistant on this shell".into())), + }, + }; + match method { + "octos.session.open" => self.open_session(client_id, &app, req_id, &args, service, services, out), + _ => { + let Some(handle) = args.get("context").and_then(Value::as_str).map(str::to_string) else { + return Self::reply(&out, req_id, Err("bad_args: context is required".into())); + }; + // A process may use only the contexts it opened. + let owned = self.contexts.get(&handle).is_some_and(|c| c.owner.client_id == client_id && c.owner.app == app); + if !owned { + return Self::reply(&out, req_id, Err("unknown_context".into())); + } + if method == "octos.context.close" { + self.close_context(&handle); + return Self::reply(&out, req_id, Ok(json!({"closed": handle}))); + } + let op = match method { + "octos.session.history" => ContextOp::History, + "octos.turn.interrupt" => ContextOp::Interrupt, + "octos.turn.start" => match args.get("text").and_then(Value::as_str) { + Some(text) => ContextOp::Turn { text: text.to_string() }, + None => return Self::reply(&out, req_id, Err("bad_args: text is required".into())), + }, + _ => return Self::reply(&out, req_id, Err(format!("unsupported: {method}"))), + }; + let ctx = self.contexts[&handle].ctx.clone(); + let sink_out = out.clone(); + let sink = Arc::new(move |event: ContextEvent| match event { + ContextEvent::Data(event) => sink_out(Down::Event { req_id, event }.to_json()), + ContextEvent::Complete(result) => sink_out(Down::Reply { req_id, result }.to_json()), + }); + if let Err(e) = ctx.call(op, sink) { + Self::reply(&out, req_id, Err(e)); + } + } + } + } + + #[allow(clippy::too_many_arguments)] + fn open_session(&mut self, client_id: ClientId, app: &str, req_id: u64, args: &Map, service: Arc, services: BTreeSet, out: FrameOut) { + let client = match args.get("client") { + None | Some(Value::Null) => None, + Some(Value::String(c)) if wire::id_ok(c) => Some(c.clone()), + Some(_) => return Self::reply(&out, req_id, Err("bad_args: client".into())), + }; + let account = if self.host.keeps_accounts(app) { + match args.get("account").and_then(Value::as_str) { + Some(a) if wire::id_ok(a) => a.to_string(), + _ => return Self::reply(&out, req_id, Err("bad_args: account is required".into())), + } + } else { + DEVICE.to_string() + }; + let handle = format!("pl{client_id}-{}", self.next_context); + self.next_context += 1; + service.set_account(Some(&account)); + let spec = ContextSpec { account: account.clone(), instance: handle.clone(), services }; + let ctx = match service.open_context(spec) { + Ok(ctx) => ctx, + Err(e) => return Self::reply(&out, req_id, Err(e)), + }; + let owner = ContextOwner { app: app.to_string(), client_id, account, client }; + self.contexts.insert(handle.clone(), Ctx { owner, ctx: ctx.clone() }); + if let Some(link) = self.links.get_mut(&client_id) { + link.contexts.push(handle.clone()); + } + let sink_out = out.clone(); + let reply_handle = handle.clone(); + let sink = Arc::new(move |event: ContextEvent| match event { + ContextEvent::Data(event) => sink_out(Down::Event { req_id, event }.to_json()), + ContextEvent::Complete(result) => { + let result = result.map(|session| json!({"context": reply_handle, "session": session})); + sink_out(Down::Reply { req_id, result }.to_json()) + } + }); + if let Err(e) = ctx.call(ContextOp::Open, sink) { + self.close_context(&handle); + Self::reply(&out, req_id, Err(e)); + } + } + + fn close_context(&mut self, handle: &str) { + if let Some(c) = self.contexts.remove(handle) { + c.ctx.close(); + if let Some(link) = self.links.get_mut(&c.owner.client_id) { + link.contexts.retain(|h| h != handle); + } + } + } + + /// Who owns the kernel's `context_id` of `app` (octos derives it from + /// the handle the shell chose: `-`). + pub fn context_owner(&self, app: &str, kernel_context_id: &str) -> Option { + self.contexts + .iter() + .find(|(handle, c)| { + c.owner.app == app && (kernel_context_id == handle.as_str() || kernel_context_id.ends_with(&format!("-{handle}"))) + }) + .map(|(_, c)| c.owner.clone()) + } + + /// The relay hands over a kernel `peer/tool/call` for `app`. + pub fn tool_call(&mut self, app: &str, call: KernelToolCall, now: f64) -> Result<(), Refused> { + let key = (app.to_string(), call.call_id.clone()); + if self.calls.contains_key(&key) { + return Err(Refused::Duplicate); + } + // The newest process of the app with a link takes the call. + let Some((&client_id, link)) = self.links.iter().filter(|(_, l)| l.app == app).max_by_key(|(id, _)| **id) else { + return Err(Refused::NotConnected); + }; + let out = link.out.clone(); + // Identity stamped from the shell's records, never the app's. + let (account, context_id, client) = match &call.context_id { + Some(kernel) => match self.context_owner(app, kernel) { + Some(owner) => (Some(owner.account), Some(kernel.clone()), owner.client), + None => return Err(Refused::UnknownContext), + }, + None => (Some(self.contexts.values().find(|c| c.owner.app == app).map(|c| c.owner.account.clone()).unwrap_or_else(|| DEVICE.into())), None, None), + }; + let caller = match &call.caller { + Caller::OwnAgent { .. } => Caller::OwnAgent { client: client.clone() }, + other => other.clone(), + }; + let rule = self.host.tool_rule(app, &call.name); + let needs_approval = rule.is_some() || call.risk == Risk::Destructive || call.outcome_unknown; + let mut down = ToolCallDown { + call_id: call.call_id.clone(), + name: call.name.clone(), + args: call.args.clone(), + risk: call.risk, + confirm_required: false, + timeout_ms: call.timeout_ms, + account: account.clone(), + context_id: context_id.clone(), + client, + caller: caller_wire(&caller), + }; + let deadline = now + call.timeout_ms as f64 / 1000.0; + let mut record = Call { client_id, risk: call.risk, state: CallState::Sent, confirm_required: false, acked: false, approval: None, deadline, down: down.clone() }; + if needs_approval { + // The obligation: acknowledge before any confirmation sheet. + self.relay.acknowledged(app, &call.call_id); + let (confirm, auto) = rule.unwrap_or((Confirm::Host, true)); + let mut spec = match confirm { + Confirm::Host => ToolSpec::host(&call.name), + Confirm::App => ToolSpec::app(&call.name), + }; + spec.auto_approvable = auto; + let id = RequestId(format!("{HELD_PREFIX}{app}:{}", call.call_id)); + let context = RequestContext { + call_id: id.0.clone(), + trigger: call.trigger.clone(), + context_id: context_id.clone(), + account: account.clone(), + outcome_unknown: call.outcome_unknown, + ..RequestContext::default() + }; + record.approval = Some(id); + match self.host.request_approval(app, spec, call.args.clone(), caller, context) { + Route::Approved(_) => {} + Route::HandedToApp => { + down.confirm_required = true; + record.confirm_required = true; + record.down = down.clone(); + } + Route::Sheet(_) | Route::WaitingForApp { .. } => record.state = CallState::Held, + Route::Refused(why) => { + self.relay.finished(app, &call.call_id, ToolCallResult::Error(format!("declined: {why}"))); + return Err(Refused::Declined(why)); + } + } + } + if record.state == CallState::Sent { + out(Down::ToolCall(down).to_json()); + } + self.calls.insert(key, record); + Ok(()) + } + + /// The relay (the kernel) cancels a call: nothing of it reaches the + /// kernel afterwards, and the app is told to stop. + pub fn tool_cancel(&mut self, app: &str, call_id: &str) { + if let Some(call) = self.calls.remove(&(app.to_string(), call_id.to_string())) { + if call.state == CallState::Sent { + if let Some(link) = self.links.get(&call.client_id) { + (link.out)(Down::ToolCancel { call_id: call_id.to_string() }.to_json()); + } + } + self.log.push(format!("peer link: {app}: call {call_id} cancelled")); + } + } + + fn on_tool_result(&mut self, client_id: ClientId, app: &str, call_id: &str, outcome: Outcome) { + let key = (app.to_string(), call_id.to_string()); + let Some(call) = self.calls.get_mut(&key) else { + self.log.push(format!("peer link: {app}: result for {call_id} ignored (answered, cancelled or never made)")); + return; + }; + if call.client_id != client_id || call.state != CallState::Sent { + self.log.push(format!("peer link: {app}: result for {call_id} ignored (not this process's, or not sent)")); + return; + } + match outcome { + Outcome::AwaitingConfirmation => { + if call.confirm_required && !call.acked { + call.acked = true; + } + } + Outcome::Ok(_) | Outcome::Error(_) if call.confirm_required && !call.acked => { + // The app ran a confirm-required call without its sheet. + let call = self.calls.remove(&key).expect("present"); + if let Some(id) = &call.approval { + self.host.app_confirm_answered(id, false, "the app skipped its confirmation"); + } + self.relay.finished(app, call_id, ToolCallResult::Error("refused: the app answered without acknowledging its confirmation".into())); + if let Some(link) = self.links.get(&client_id) { + (link.out)(Down::ToolCancel { call_id: call_id.to_string() }.to_json()); + } + self.log.push(format!("peer link: {app}: call {call_id} refused (no acknowledgement before its sheet)")); + } + Outcome::Ok(data) => { + let call = self.calls.remove(&key).expect("present"); + if call.confirm_required { + if let Some(id) = &call.approval { + self.host.app_confirm_answered(id, true, "approved on the app's sheet"); + } + } + self.relay.finished(app, call_id, ToolCallResult::Ok(data)); + } + Outcome::Error(error) => { + let call = self.calls.remove(&key).expect("present"); + if call.confirm_required { + if let Some(id) = &call.approval { + self.host.app_confirm_answered(id, false, &error); + } + } + self.relay.finished(app, call_id, ToolCallResult::Error(error)); + } + } + } + + /// The router's decisions and the deadlines; call on the shell's tick. + pub fn tick(&mut self, now: f64) { + for (id, decision, reason) in self.host.take_decisions() { + let Some(key) = self.calls.iter().find(|(_, c)| c.approval.as_ref() == Some(&id) && c.state == CallState::Held).map(|(k, _)| k.clone()) else { + continue; + }; + if decision.approved() { + let call = self.calls.get_mut(&key).expect("present"); + call.state = CallState::Sent; + if let Some(link) = self.links.get(&call.client_id) { + (link.out)(Down::ToolCall(call.down.clone()).to_json()); + } + } else { + self.calls.remove(&key); + self.relay.finished(&key.0, &key.1, ToolCallResult::Error(format!("declined: {reason}"))); + } + } + let late: Vec<(String, String)> = self.calls.iter().filter(|(_, c)| now >= c.deadline).map(|(k, _)| k.clone()).collect(); + for key in late { + let call = self.calls.remove(&key).expect("present"); + if call.state == CallState::Sent { + if let Some(link) = self.links.get(&call.client_id) { + (link.out)(Down::ToolCancel { call_id: key.1.clone() }.to_json()); + } + } + self.relay.finished(&key.0, &key.1, ToolCallResult::Error("timed_out".into())); + } + } + + /// The process on `client_id` is gone (exited, killed, or its socket + /// closed): its outstanding calls fail (`outcome_unknown` unless they + /// only read), its request contexts close, and its app's peer stays. + pub fn process_gone(&mut self, client_id: ClientId) { + let Some(link) = self.links.remove(&client_id) else { return }; + let keys: Vec<(String, String)> = self.calls.iter().filter(|(_, c)| c.client_id == client_id).map(|(k, _)| k.clone()).collect(); + for key in keys { + let call = self.calls.remove(&key).expect("present"); + let result = if call.risk == Risk::Read || call.state == CallState::Held { + ToolCallResult::Error("app_exited".into()) + } else { + ToolCallResult::OutcomeUnknown + }; + self.relay.finished(&key.0, &key.1, result); + } + for handle in link.contexts.clone() { + self.close_context(&handle); + } + if !self.has_link(&link.app) { + self.host.link_closed(&link.app); + } + self.log.push(format!("peer link: {} (client {client_id}) closed; its peer is kept", link.app)); + } + + /// Close every context of `app`'s `account` (signing out, §11): the + /// app is told, and later calls stamped for it are refused. + pub fn close_account(&mut self, app: &str, account: &str, reason: &str) { + let handles: Vec = self.contexts.iter().filter(|(_, c)| c.owner.app == app && c.owner.account == account).map(|(h, _)| h.clone()).collect(); + for handle in handles { + let client_id = self.contexts[&handle].owner.client_id; + self.close_context(&handle); + if let Some(link) = self.links.get(&client_id) { + (link.out)(Down::ContextClosed { context: handle.clone(), reason: reason.to_string() }.to_json()); + } + } + } + + /// Whether the shell keeps a peer (service) for `app`. + pub fn keeps_peer(&self, app: &str) -> bool { + self.services.contains_key(app) + } + + pub fn open_calls(&self) -> usize { + self.calls.len() + } + + pub fn open_contexts(&self, app: &str) -> usize { + self.contexts.values().filter(|c| c.owner.app == app).count() + } +} diff --git a/crates/shell/src/peer_link/mod.rs b/crates/shell/src/peer_link/mod.rs new file mode 100644 index 00000000..5b8619d3 --- /dev/null +++ b/crates/shell/src/peer_link/mod.rs @@ -0,0 +1,221 @@ +//! The peer link (ADR 0004 §5): a process-hosted native app's own channel +//! to ITS octos app agent, over its hub socket, under the `"octos_peer"` +//! envelope ([`wire`]); never registered with the AI services bus. +//! +//! - **Identity** is the socket's: the shell launched that process for one +//! app (its client slot), and every frame is that app's. A tool call's +//! account, context and client come from the shell's own records of the +//! contexts the process opened ([`link::ContextOwner`]); `caller` comes +//! from the relay. A process may use only contexts it opened. +//! - **Grants and consent**: only an app whose `native-apps.json` entry +//! grants `agent.octos` services gets a link, and each request needs +//! `consent::granted(app)` (the first-use sheet asks, once). +//! - **Requests** (`octos.session.open|history`, `octos.turn.start|interrupt`, +//! `octos.context.close`) run on the app's one peer through the same +//! service in-process modules get (`crates/app-peers`' broker), by an +//! adapter ([`ShellHost::service`]), until octos#2567's relay replaces it. +//! - **Tool calls** (kernel → app) enter at [`tool_call`] and their outcome +//! leaves through the installed [`ToolRelay`]; confirmations go through +//! the #120 router. The host obligations are kept here: once per call, +//! nothing after cancel, and an acknowledgement before any confirmation +//! sheet (the kernel's, and the app's own `awaiting_confirmation` before a +//! `confirm: app` result). +//! - **A process that dies** fails its outstanding calls (`outcome_unknown` +//! unless they only read), closes its request contexts and keeps the peer. +//! +//! The seams for octos#2567's relay (home-96): [`set_tool_relay`], +//! [`tool_call`], [`tool_cancel`], [`context_owner`], [`has_link`], +//! [`close_account`]. + +pub mod link; +pub mod wire; + +#[cfg(test)] +mod tests; + +use std::collections::BTreeSet; +use std::sync::{Arc, Mutex}; + +use serde_json::Value; + +pub use link::{ContextOwner, FrameOut, KernelToolCall, PeerHost, PeerLinks, RecordingToolRelay, Refused, ToolCallResult, ToolRelay}; +pub use wire::Risk; + +use crate::ai_host::app_peers::OctosAppService; +use crate::approvals::{Caller, Decision, RequestContext, RequestId, Route, ToolSpec}; +use crate::hub::ClientId; +use crate::native_apps::Confirm; + +/// The shell's decisions for the link: the manifest, consent, the router +/// and the app's peer. +pub struct ShellHost; + +/// The `confirm: app` hand-off: the router tells the app's sheet through +/// the link, which sends the call with `confirm_required` itself. +struct LinkConfirm; + +impl crate::approvals::AppConfirm for LinkConfirm { + fn confirm(&mut self, _request: &crate::approvals::AppConfirmRequest) {} +} + +impl PeerHost for ShellHost { + fn granted(&self, app: &str) -> BTreeSet { + let Some(entry) = crate::native_apps::find(app) else { return BTreeSet::new() }; + if entry.octos.is_empty() && !crate::dev_mode::grants_all(app) { + return BTreeSet::new(); + } + if crate::dev_mode::grants_all(app) { + return crate::ai_host::app_peers::OCTOS_SERVICES.iter().map(|s| s.to_string()).collect(); + } + entry.octos.iter().map(|s| s.to_string()).collect() + } + fn keeps_accounts(&self, app: &str) -> bool { + crate::native_apps::find(app).is_some_and(|a| a.accounts) + } + fn consent(&mut self, app: &str) -> bool { + let caps: Vec = self.granted(app).into_iter().collect(); + let caps: Vec<&str> = caps.iter().map(String::as_str).collect(); + let label = crate::approvals::sheet::app_label(app); + crate::approvals::with(|a| crate::approvals::module_gate(a, app, &label, &caps)).unwrap_or(false) + } + #[allow(unused_variables)] + fn service(&mut self, app: &str, services: &BTreeSet) -> Option> { + #[cfg(feature = "octos-core")] + { + let policy = crate::ai_host::app_peers::hosted::HostPolicy::default(); + policy.allow(app, services.iter().map(String::as_str)); + let label = crate::approvals::sheet::app_label(app); + let broker = crate::ai_host::app_peers::hosted::launch(app, &label, services.iter().map(String::as_str), &policy)?; + Some(Arc::new(broker) as Arc) + } + #[cfg(not(feature = "octos-core"))] + None + } + fn tool_rule(&self, app: &str, tool: &str) -> Option<(Confirm, bool)> { + crate::native_apps::find(app).and_then(|a| a.tool(tool)).map(|r| (r.confirm, r.auto_approvable)) + } + fn request_approval(&mut self, app: &str, tool: ToolSpec, args: Value, caller: Caller, context: RequestContext) -> Route { + crate::approvals::approval_requested(app, tool, args, caller, context) + } + fn take_decisions(&mut self) -> Vec<(RequestId, Decision, String)> { + crate::approvals::take_peer_decisions() + } + fn app_confirm_answered(&mut self, id: &RequestId, approved: bool, reason: &str) { + let _ = crate::approvals::app_confirm_answered(id, approved, reason); + } + fn link_opened(&mut self, app: &str) { + crate::approvals::register_app_confirm(app, Box::new(LinkConfirm)); + } + fn link_closed(&mut self, app: &str) { + crate::approvals::unregister_app_confirm(app); + } +} + +/// Until the relay is installed, outcomes queue here and are handed over. +struct Queued(RecordingToolRelay); + +impl ToolRelay for Queued { + fn acknowledged(&mut self, app: &str, call_id: &str) { + self.0.acknowledged(app, call_id) + } + fn finished(&mut self, app: &str, call_id: &str, result: ToolCallResult) { + self.0.finished(app, call_id, result) + } +} + +static LINKS: Mutex> = Mutex::new(None); + +fn with(f: impl FnOnce(&mut PeerLinks) -> R) -> R { + let mut guard = LINKS.lock().unwrap_or_else(|e| e.into_inner()); + let (links, _) = guard.get_or_insert_with(|| { + let queue = RecordingToolRelay::default(); + (PeerLinks::new(Box::new(ShellHost), Box::new(Queued(queue.clone()))), queue) + }); + let out = f(links); + for line in links.log.drain(..) { + makepad_widgets::log!("{line}"); + } + out +} + +/// A frame sender for a client socket. +pub fn frame_out(sender: std::sync::mpsc::Sender>) -> FrameOut { + let sender = Mutex::new(sender); + Arc::new(move |json: String| { + let sender = sender.lock().unwrap_or_else(|e| e.into_inner()); + crate::hub::send_to_app(&sender, vec![makepad_studio_protocol::StudioToApp::Custom(json)]); + }) +} + +// ------------------------------------------------------ the shell's hooks + +/// A client connected its hub socket (`lib.rs`, `HubEvent::Connected`). +pub fn connected(client: ClientId, app: &str, sender: std::sync::mpsc::Sender>) { + with(|l| l.connected(client, app, frame_out(sender))); +} + +/// A `Custom` frame from a client: true when it was the peer link's. +pub fn on_frame(client: ClientId, app: &str, frame: &str, sender: Option>>) -> bool { + if !wire::is_peer_frame(frame) { + return false; + } + with(|l| l.on_frame(client, app, frame, sender.map(frame_out))) +} + +/// The client's process died or its socket closed. +pub fn process_gone(client: ClientId) { + with(|l| l.process_gone(client)); +} + +/// Once a second, with the approvals tick. +pub fn tick() { + let now = crate::host::now(); + with(|l| l.tick(now)); +} + +// ------------------------------------------------ the seams for home-96 + +/// Install octos#2567's relay; outcomes queued before are handed over. +pub fn set_tool_relay(mut relay: Box) { + let mut guard = LINKS.lock().unwrap_or_else(|e| e.into_inner()); + if guard.is_none() { + drop(guard); + with(|_| ()); + guard = LINKS.lock().unwrap_or_else(|e| e.into_inner()); + } + if let Some((links, queue)) = guard.as_mut() { + for (app, call, result) in queue.take() { + match result { + None => relay.acknowledged(&app, &call), + Some(r) => relay.finished(&app, &call, r), + } + } + links.set_relay(relay); + } +} + +/// A kernel `peer/tool/call` for a process app. +pub fn tool_call(app: &str, call: KernelToolCall) -> Result<(), Refused> { + let now = crate::host::now(); + with(|l| l.tool_call(app, call, now)) +} + +/// The kernel cancelled a call. +pub fn tool_cancel(app: &str, call_id: &str) { + with(|l| l.tool_cancel(app, call_id)); +} + +/// Who owns a kernel request context of `app` (account and client). +pub fn context_owner(app: &str, kernel_context_id: &str) -> Option { + with(|l| l.context_owner(app, kernel_context_id)) +} + +/// Whether a process of `app` holds a peer link now. +pub fn has_link(app: &str) -> bool { + with(|l| l.has_link(app)) +} + +/// Signing out (§11): close `account`'s contexts of `app`. +pub fn close_account(app: &str, account: &str) { + with(|l| l.close_account(app, account, "signed_out")); +} diff --git a/crates/shell/src/peer_link/tests.rs b/crates/shell/src/peer_link/tests.rs new file mode 100644 index 00000000..491c0a33 --- /dev/null +++ b/crates/shell/src/peer_link/tests.rs @@ -0,0 +1,454 @@ +//! The peer link without a kernel: framing, identity stamping, grants and +//! consent, the host obligations (once per call, nothing after cancel, an +//! acknowledgement before a sheet) and what a process's death does. + +use super::link::*; +use super::wire::{self, Down, Outcome, Up}; +use crate::ai_host::app_peers::{Availability, ContextEvent, ContextOp, ContextSpec, Deployment, EventSink, ModelInfo, OctosAppService, OctosContext, SettingsEntry}; +use crate::approvals::{Caller, Decision, RequestContext, RequestId, Route, ToolSpec, Trigger}; +use crate::native_apps::Confirm; +use serde_json::{json, Value}; +use std::collections::BTreeSet; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; + +// ------------------------------------------------------------ doubles + +#[derive(Default)] +struct Shared { + /// Everything that happened, in order ("ack c1", "approval c1", …). + order: Vec, + consent: bool, + route: Option, + decisions: Vec<(RequestId, Decision, String)>, + confirmed: Vec<(RequestId, bool)>, + services_made: usize, + contexts: Vec>, + specs: Vec, +} + +#[derive(Clone, Default)] +struct World(Arc>); + +impl World { + fn with(&self, f: impl FnOnce(&mut Shared) -> R) -> R { + f(&mut self.0.lock().unwrap()) + } +} + +struct FakeHost(World); + +impl PeerHost for FakeHost { + fn granted(&self, app: &str) -> BTreeSet { + match app { + "notes" => ["octos.session.open", "octos.turn.start", "octos.session.history"].iter().map(|s| s.to_string()).collect(), + "mail" => ["octos.session.open"].iter().map(|s| s.to_string()).collect(), + _ => BTreeSet::new(), + } + } + fn keeps_accounts(&self, app: &str) -> bool { + app == "mail" + } + fn consent(&mut self, _app: &str) -> bool { + self.0.with(|s| s.consent) + } + fn service(&mut self, _app: &str, _services: &BTreeSet) -> Option> { + self.0.with(|s| s.services_made += 1); + Some(Arc::new(FakeService(self.0.clone()))) + } + fn tool_rule(&self, _app: &str, tool: &str) -> Option<(Confirm, bool)> { + match tool { + "send" => Some((Confirm::App, true)), + "run" => Some((Confirm::Host, false)), + _ => None, + } + } + fn request_approval(&mut self, _app: &str, _tool: ToolSpec, _args: Value, caller: Caller, context: RequestContext) -> Route { + self.0.with(|s| { + s.order.push(format!("approval {} {:?} {:?}", context.call_id, caller, context.account)); + s.route.clone().unwrap_or(Route::Sheet(1)) + }) + } + fn take_decisions(&mut self) -> Vec<(RequestId, Decision, String)> { + self.0.with(|s| std::mem::take(&mut s.decisions)) + } + fn app_confirm_answered(&mut self, id: &RequestId, approved: bool, _reason: &str) { + self.0.with(|s| s.confirmed.push((id.clone(), approved))); + } + fn link_opened(&mut self, app: &str) { + self.0.with(|s| s.order.push(format!("opened {app}"))); + } + fn link_closed(&mut self, app: &str) { + self.0.with(|s| s.order.push(format!("closed {app}"))); + } +} + +struct FakeService(World); + +impl OctosAppService for FakeService { + fn deployment(&self) -> Deployment { + Deployment::Hosted + } + fn availability(&self) -> Availability { + Availability::Ready + } + fn services(&self) -> BTreeSet { + BTreeSet::new() + } + fn model(&self) -> Option { + None + } + fn settings_entry(&self) -> SettingsEntry { + SettingsEntry::Host + } + fn set_account(&self, _account: Option<&str>) {} + fn open_context(&self, spec: ContextSpec) -> Result, String> { + let ctx = Arc::new(FakeContext { closed: AtomicBool::new(false), calls: AtomicUsize::new(0) }); + self.0.with(|s| { + s.contexts.push(ctx.clone()); + s.specs.push(spec); + }); + Ok(ctx) + } + fn release(&self) {} + fn shutdown(&self) {} +} + +struct FakeContext { + closed: AtomicBool, + calls: AtomicUsize, +} + +impl OctosContext for FakeContext { + fn call(&self, op: ContextOp, sink: EventSink) -> Result<(), String> { + if self.closed.load(Ordering::SeqCst) { + return Err("closed".into()); + } + self.calls.fetch_add(1, Ordering::SeqCst); + match op { + ContextOp::Open => sink(ContextEvent::Complete(Ok(json!({"session_id": "s"})))), + ContextOp::Turn { text } => { + sink(ContextEvent::Data(json!({"method": "message/delta", "text": text}))); + sink(ContextEvent::Complete(Ok(json!({"text": "done"})))); + } + _ => sink(ContextEvent::Complete(Ok(Value::Null))), + } + Ok(()) + } + fn close(&self) { + self.closed.store(true, Ordering::SeqCst); + } + fn is_open(&self) -> bool { + !self.closed.load(Ordering::SeqCst) + } +} + +type Frames = Arc>>; + +fn out() -> (FrameOut, Frames) { + let frames: Frames = Arc::default(); + let f = frames.clone(); + (Arc::new(move |json: String| f.lock().unwrap().push(json)), frames) +} + +fn downs(frames: &Frames) -> Vec { + std::mem::take(&mut *frames.lock().unwrap()).iter().map(|f| Down::parse(f).expect(f)).collect() +} + +fn setup() -> (PeerLinks, World, RecordingToolRelay) { + let world = World::default(); + world.with(|s| s.consent = true); + let relay = RecordingToolRelay::default(); + (PeerLinks::new(Box::new(FakeHost(world.clone())), Box::new(relay.clone())), world, relay) +} + +fn request(req_id: u64, method: &str, args: Value) -> String { + json!({"octos_peer": {"up": "request", "req_id": req_id, "method": method, "args": args}}).to_string() +} + +fn result(call_id: &str, outcome: Value) -> String { + let mut inner = json!({"up": "tool_result", "call_id": call_id}); + for (k, v) in outcome.as_object().unwrap() { + inner[k] = v.clone(); + } + json!({ "octos_peer": inner }).to_string() +} + +/// Open a context on `client` for `notes`; its handle. +fn open(links: &mut PeerLinks, client: u64, frames: &Frames, label: Option<&str>) -> String { + assert!(links.on_frame(client, "notes", &request(1, "octos.session.open", json!({"client": label})), None)); + match downs(frames).pop() { + Some(Down::Reply { req_id: 1, result: Ok(v) }) => v["context"].as_str().unwrap().to_string(), + other => panic!("{other:?}"), + } +} + +fn call(id: &str, name: &str, risk: wire::Risk, context: Option<&str>) -> KernelToolCall { + KernelToolCall { + call_id: id.into(), + name: name.into(), + args: json!({"to": "#room", "text": "hi"}), + risk, + timeout_ms: 30_000, + context_id: context.map(str::to_string), + caller: Caller::OwnAgent { client: None }, + trigger: Trigger::Person, + outcome_unknown: false, + } +} + +// ------------------------------------------------------------ framing + +/// The frames Makepad's client (`makepad_ai_services::peer`) writes, byte +/// for byte, parse here; and the shell's frames carry its field names. +#[test] +fn the_shell_reads_the_makepad_clients_frames() { + let from_makepad = [ + r#"{"octos_peer":{"up":"request","req_id":7,"method":"octos.turn.start","args":{"context":"c","text":"hi"}}}"#, + r#"{"octos_peer":{"up":"tool_result","call_id":"k1","ok":true,"data":{"n":3}}}"#, + r#"{"octos_peer":{"up":"tool_result","call_id":"k1","ok":false,"error":"no"}}"#, + r#"{"octos_peer":{"up":"tool_result","call_id":"k1","ok":false,"awaiting_confirmation":true}}"#, + ]; + let parsed: Vec = from_makepad.iter().map(|f| Up::parse(f).expect(f)).collect(); + assert!(matches!(&parsed[0], Up::Request { req_id: 7, method, args } if method == "octos.turn.start" && args["text"] == "hi")); + assert_eq!(parsed[1], Up::ToolResult { call_id: "k1".into(), outcome: Outcome::Ok(json!({"n": 3})) }); + assert_eq!(parsed[2], Up::ToolResult { call_id: "k1".into(), outcome: Outcome::Error("no".into()) }); + assert_eq!(parsed[3], Up::ToolResult { call_id: "k1".into(), outcome: Outcome::AwaitingConfirmation }); + // Not the link's: the AI bus's and the WM's envelopes, unknown methods, + // non-object args, a second top-level key, an oversized frame. + assert!(!wire::is_peer_frame(r#"{"wm_ai":{"from":null,"msg":"Unregister"}}"#)); + assert_eq!(Up::parse(&request(1, "octos.admin", json!({}))), None); + assert_eq!(Up::parse(r#"{"octos_peer":{"up":"request","req_id":1,"method":"octos.turn.start","args":[1]}}"#), None); + assert_eq!(Up::parse(r#"{"octos_peer":{"up":"request","req_id":1,"method":"octos.turn.start"},"from":"rinx"}"#), None); + let big = request(1, "octos.turn.start", json!({"text": "a".repeat(wire::MAX_FRAME_BYTES)})); + assert_eq!(Up::parse(&big), None); + let call = Down::ToolCall(wire::ToolCallDown { + call_id: "k2".into(), + name: "send".into(), + args: json!({}), + risk: wire::Risk::Destructive, + confirm_required: true, + timeout_ms: 1, + account: Some("a".into()), + context_id: None, + client: None, + caller: "system_agent".into(), + }) + .to_json(); + for key in ["\"down\":\"tool_call\"", "\"confirm_required\":true", "\"caller\":\"system_agent\"", "\"context_id\":null"] { + assert!(call.contains(key), "{call}"); + } +} + +#[test] +fn bus_frames_are_not_consumed_and_malformed_peer_frames_are() { + let (mut links, _, _) = setup(); + let (o, frames) = out(); + links.connected(1, "notes", o); + assert!(!links.on_frame(1, "notes", r#"{"wm_ai":{"from":null,"msg":"Unregister"}}"#, None)); + assert!(links.on_frame(1, "notes", r#"{"octos_peer":{"up":"nonsense"}}"#, None)); + assert!(downs(&frames).is_empty()); +} + +// --------------------------------------------- grants, consent, identity + +#[test] +fn only_granted_and_consented_apps_get_a_link_and_only_granted_methods() { + let (mut links, world, _) = setup(); + let (o, frames) = out(); + assert!(!links.connected(1, "sheets", o.clone()), "no agent.octos grant: no link"); + assert!(links.on_frame(1, "sheets", &request(1, "octos.session.open", json!({})), Some(o.clone()))); + assert!(matches!(downs(&frames).as_slice(), [Down::Reply { result: Err(e), .. }] if e.starts_with("no_agent"))); + assert!(links.connected(2, "notes", o.clone())); + world.with(|s| s.consent = false); + links.on_frame(2, "notes", &request(2, "octos.session.open", json!({})), None); + assert!(matches!(downs(&frames).as_slice(), [Down::Reply { result: Err(e), .. }] if e.starts_with("consent_pending"))); + assert_eq!(world.with(|s| s.services_made), 0, "no peer before consent"); + world.with(|s| s.consent = true); + let ctx = open(&mut links, 2, &frames, None); + links.on_frame(2, "notes", &request(3, "octos.turn.interrupt", json!({"context": ctx})), None); + assert!(matches!(downs(&frames).as_slice(), [Down::Reply { req_id: 3, result: Err(e) }] if e == "not_granted: octos.turn.interrupt")); +} + +#[test] +fn identity_is_the_sockets_and_a_process_uses_only_its_own_contexts() { + let (mut links, world, _) = setup(); + let (a, frames_a) = out(); + let (b, frames_b) = out(); + links.connected(1, "notes", a); + links.connected(2, "notes", b); + // A frame that claims another app changes nothing: the socket's app runs. + let claimed = r#"{"octos_peer":{"up":"request","req_id":1,"method":"octos.session.open","args":{"client":"mini","app":"mail","account":"someone"}}}"#; + links.on_frame(1, "notes", claimed, None); + let ctx = match downs(&frames_a).pop() { + Some(Down::Reply { result: Ok(v), .. }) => v["context"].as_str().unwrap().to_string(), + other => panic!("{other:?}"), + }; + let spec = world.with(|s| s.specs[0].clone()); + assert_eq!(spec.account, DEVICE, "notes keeps no accounts: the frame's account is ignored"); + assert_eq!(spec.instance, ctx); + let owner = links.context_owner("notes", &format!("ab12cd34-{ctx}")).expect("the kernel id resolves"); + assert_eq!((owner.client_id, owner.client.as_deref()), (1, Some("mini"))); + assert!(links.context_owner("mail", &ctx).is_none(), "another app's records"); + // Process 2 of the same app cannot drive process 1's context. + links.on_frame(2, "notes", &request(5, "octos.turn.start", json!({"context": ctx, "text": "x"})), None); + assert!(matches!(downs(&frames_b).as_slice(), [Down::Reply { req_id: 5, result: Err(e) }] if e == "unknown_context")); + // Process 1 can, and its turn streams then replies. + links.on_frame(1, "notes", &request(6, "octos.turn.start", json!({"context": ctx, "text": "x"})), None); + let got = downs(&frames_a); + assert!(matches!(got.as_slice(), [Down::Event { req_id: 6, .. }, Down::Reply { req_id: 6, result: Ok(_) }]), "{got:?}"); +} + +#[test] +fn an_app_that_keeps_accounts_names_one() { + let (mut links, world, _) = setup(); + let (o, frames) = out(); + links.connected(1, "mail", o); + links.on_frame(1, "mail", &request(1, "octos.session.open", json!({})), None); + assert!(matches!(downs(&frames).as_slice(), [Down::Reply { result: Err(e), .. }] if e.contains("account"))); + links.on_frame(1, "mail", &request(2, "octos.session.open", json!({"account": "me@example.org"})), None); + assert!(matches!(downs(&frames).as_slice(), [Down::Reply { result: Ok(_), .. }])); + assert_eq!(world.with(|s| s.specs[0].account.clone()), "me@example.org"); +} + +// ------------------------------------------------------------ tool calls + +#[test] +fn a_tool_call_is_stamped_from_the_shells_records() { + let (mut links, _, relay) = setup(); + let (o, frames) = out(); + links.connected(1, "notes", o); + let ctx = open(&mut links, 1, &frames, Some("mini")); + links.tool_call("notes", call("c1", "lookup", wire::Risk::Read, Some(&format!("n0nce-{ctx}"))), 0.0).unwrap(); + match downs(&frames).as_slice() { + [Down::ToolCall(c)] => { + assert_eq!((c.account.as_deref(), c.client.as_deref(), c.caller.as_str()), (Some(DEVICE), Some("mini"), "own_agent")); + assert!(!c.confirm_required); + } + other => panic!("{other:?}"), + } + assert_eq!(links.tool_call("notes", call("c1", "lookup", wire::Risk::Read, None), 0.0), Err(Refused::Duplicate), "once per call"); + assert_eq!(links.tool_call("notes", call("c2", "lookup", wire::Risk::Read, Some("n0nce-pl9-9")), 0.0), Err(Refused::UnknownContext)); + assert_eq!(links.tool_call("mail", call("c3", "lookup", wire::Risk::Read, None), 0.0), Err(Refused::NotConnected)); + // One result reaches the relay; a second is ignored. + links.on_frame(1, "notes", &result("c1", json!({"ok": true, "data": {"hits": 2}})), None); + links.on_frame(1, "notes", &result("c1", json!({"ok": true, "data": {"hits": 3}})), None); + assert_eq!(relay.take(), vec![("notes".into(), "c1".into(), Some(ToolCallResult::Ok(json!({"hits": 2}))))]); +} + +#[test] +fn nothing_after_cancel() { + let (mut links, _, relay) = setup(); + let (o, frames) = out(); + links.connected(1, "notes", o); + links.tool_call("notes", call("c1", "lookup", wire::Risk::Act, None), 0.0).unwrap(); + links.tool_cancel("notes", "c1"); + assert!(matches!(downs(&frames).as_slice(), [Down::ToolCall(_), Down::ToolCancel { call_id }] if call_id == "c1")); + links.on_frame(1, "notes", &result("c1", json!({"ok": true, "data": null})), None); + assert!(relay.take().is_empty(), "a result after cancel never reaches the kernel"); +} + +#[test] +fn a_host_confirmed_call_is_acknowledged_before_its_sheet_and_waits_for_the_person() { + let (mut links, world, relay) = setup(); + let (o, frames) = out(); + links.connected(1, "notes", o); + world.with(|s| s.route = Some(Route::Sheet(4))); + links.tool_call("notes", call("c1", "run", wire::Risk::Destructive, None), 0.0).unwrap(); + assert!(downs(&frames).is_empty(), "nothing reaches the app while the sheet is up"); + let events = relay.take(); + assert_eq!(events, vec![("notes".into(), "c1".into(), None)], "acknowledged first"); + let order = world.with(|s| s.order.clone()); + assert!(order.iter().any(|l| l.starts_with("approval peerlink:notes:c1")), "{order:?}"); + // The person approves: the call goes to the app, once. + world.with(|s| s.decisions.push((RequestId("peerlink:notes:c1".into()), Decision::ApproveOnce, "sheet".into()))); + links.tick(1.0); + links.tick(2.0); + assert!(matches!(downs(&frames).as_slice(), [Down::ToolCall(c)] if c.call_id == "c1" && !c.confirm_required)); + // A denial of another call finishes it with the reason. + links.tool_call("notes", call("c2", "run", wire::Risk::Destructive, None), 0.0).unwrap(); + world.with(|s| s.decisions.push((RequestId("peerlink:notes:c2".into()), Decision::Deny, "denied on the sheet".into()))); + links.tick(1.0); + let events = relay.take(); + assert!(events.contains(&("notes".into(), "c2".into(), Some(ToolCallResult::Error("declined: denied on the sheet".into())))), "{events:?}"); +} + +#[test] +fn an_app_confirmed_call_needs_the_apps_acknowledgement_before_its_result() { + let (mut links, world, relay) = setup(); + let (o, frames) = out(); + links.connected(1, "notes", o); + world.with(|s| s.route = Some(Route::HandedToApp)); + links.tool_call("notes", call("c1", "send", wire::Risk::Destructive, None), 0.0).unwrap(); + assert!(matches!(downs(&frames).as_slice(), [Down::ToolCall(c)] if c.confirm_required)); + // Answering without the acknowledgement is refused and the call stops. + links.on_frame(1, "notes", &result("c1", json!({"ok": true, "data": null})), None); + let events = relay.take(); + assert!(matches!(events.as_slice(), [_, (_, _, Some(ToolCallResult::Error(e)))] if e.contains("without acknowledging")), "{events:?}"); + assert!(matches!(downs(&frames).as_slice(), [Down::ToolCancel { .. }])); + assert_eq!(world.with(|s| s.confirmed.clone()), vec![(RequestId("peerlink:notes:c1".into()), false)]); + // The right order: acknowledge, show the sheet, answer. + links.tool_call("notes", call("c2", "send", wire::Risk::Destructive, None), 0.0).unwrap(); + links.on_frame(1, "notes", &result("c2", json!({"ok": false, "awaiting_confirmation": true})), None); + links.on_frame(1, "notes", &result("c2", json!({"ok": true, "data": {"sent": true}})), None); + let events = relay.take(); + assert_eq!(events.last(), Some(&("notes".into(), "c2".into(), Some(ToolCallResult::Ok(json!({"sent": true})))))); + assert_eq!(world.with(|s| s.confirmed.last().cloned()), Some((RequestId("peerlink:notes:c2".into()), true))); +} + +#[test] +fn a_call_past_its_deadline_times_out_and_the_app_is_told() { + let (mut links, _, relay) = setup(); + let (o, frames) = out(); + links.connected(1, "notes", o); + links.tool_call("notes", call("c1", "lookup", wire::Risk::Read, None), 10.0).unwrap(); + links.tick(39.0); + assert!(relay.take().is_empty()); + links.tick(40.5); + assert_eq!(relay.take(), vec![("notes".into(), "c1".into(), Some(ToolCallResult::Error("timed_out".into())))]); + assert!(matches!(downs(&frames).as_slice(), [Down::ToolCall(_), Down::ToolCancel { .. }])); +} + +// ------------------------------------------------------------ death + +#[test] +fn a_dead_process_fails_its_calls_closes_its_contexts_and_keeps_the_peer() { + let (mut links, world, relay) = setup(); + let (o, frames) = out(); + links.connected(1, "notes", o.clone()); + let _ctx = open(&mut links, 1, &frames, None); + links.tool_call("notes", call("read", "lookup", wire::Risk::Read, None), 0.0).unwrap(); + links.tool_call("notes", call("act", "save", wire::Risk::Act, None), 0.0).unwrap(); + links.process_gone(1); + let mut events = relay.take(); + events.sort_by(|a, b| a.1.cmp(&b.1)); + assert_eq!( + events, + vec![ + ("notes".into(), "act".into(), Some(ToolCallResult::OutcomeUnknown)), + ("notes".into(), "read".into(), Some(ToolCallResult::Error("app_exited".into()))), + ] + ); + assert!(world.with(|s| s.contexts.iter().all(|c| !c.is_open())), "its request contexts are closed"); + assert_eq!(links.open_contexts("notes"), 0); + assert!(!links.has_link("notes")); + assert!(links.keeps_peer("notes"), "the peer stays"); + assert!(world.with(|s| s.order.contains(&"closed notes".to_string()))); + // A result from the dead process's calls can no longer arrive; a + // restarted process gets the same peer, not a new one. + assert_eq!(links.tool_call("notes", call("later", "lookup", wire::Risk::Read, None), 0.0), Err(Refused::NotConnected)); + links.connected(3, "notes", o); + let _ = open(&mut links, 3, &frames, None); + assert_eq!(world.with(|s| s.services_made), 1, "the same peer after a restart"); +} + +#[test] +fn signing_out_closes_the_accounts_contexts_and_tells_the_app() { + let (mut links, _, _) = setup(); + let (o, frames) = out(); + links.connected(1, "notes", o); + let ctx = open(&mut links, 1, &frames, None); + links.close_account("notes", DEVICE, "signed_out"); + assert!(matches!(downs(&frames).as_slice(), [Down::ContextClosed { context, reason }] if *context == ctx && reason == "signed_out")); + assert_eq!(links.tool_call("notes", call("c1", "lookup", wire::Risk::Read, Some(&ctx)), 0.0), Err(Refused::UnknownContext)); +} diff --git a/crates/shell/src/peer_link/wire.rs b/crates/shell/src/peer_link/wire.rs new file mode 100644 index 00000000..9d0905a4 --- /dev/null +++ b/crates/shell/src/peer_link/wire.rs @@ -0,0 +1,196 @@ +//! The peer link's frames, as the shell reads and writes them: the same +//! JSON as Makepad's `makepad_ai_services::peer` (the app's side), inside a +//! studio `Custom` frame under the `"octos_peer"` key. The shell parses +//! with serde_json and depends on no Makepad type for it, so the link works +//! with any Makepad revision that carries the client; the fixtures in +//! `tests.rs` are the frames that client writes. + +use serde_json::{json, Map, Value}; + +/// The envelope key (Makepad's `PEER_KEY`). +pub const PEER_KEY: &str = "octos_peer"; +/// What an app may ask, by exact name (Makepad's `PEER_METHODS`). +pub const PEER_METHODS: [&str; 5] = [ + "octos.session.open", + "octos.session.history", + "octos.turn.start", + "octos.turn.interrupt", + "octos.context.close", +]; +/// Bytes of one frame; larger frames are dropped unread. +pub const MAX_FRAME_BYTES: usize = 1024 * 1024; +/// Longest call id, context id or client label. +pub const MAX_ID_BYTES: usize = 128; + +/// How the app answers a tool call. +#[derive(Clone, Debug, PartialEq)] +pub enum Outcome { + Ok(Value), + Error(String), + AwaitingConfirmation, +} + +/// App → shell. +#[derive(Clone, Debug, PartialEq)] +pub enum Up { + Request { req_id: u64, method: String, args: Map }, + ToolResult { call_id: String, outcome: Outcome }, +} + +/// How much a tool can break (the wire's `risk`). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Risk { + Read, + Act, + Destructive, +} + +impl Risk { + pub fn as_str(self) -> &'static str { + match self { + Risk::Read => "read", + Risk::Act => "act", + Risk::Destructive => "destructive", + } + } +} + +/// A tool call as the app gets it: identity fields stamped by the shell. +#[derive(Clone, Debug, PartialEq)] +pub struct ToolCallDown { + pub call_id: String, + pub name: String, + pub args: Value, + pub risk: Risk, + pub confirm_required: bool, + pub timeout_ms: u64, + pub account: Option, + pub context_id: Option, + pub client: Option, + /// `own_agent`, `app:` or `system_agent`. + pub caller: String, +} + +/// Shell → app. +#[derive(Clone, Debug, PartialEq)] +pub enum Down { + Reply { req_id: u64, result: Result }, + Event { req_id: u64, event: Value }, + ToolCall(ToolCallDown), + ToolCancel { call_id: String }, + ContextClosed { context: String, reason: String }, +} + +pub fn id_ok(id: &str) -> bool { + !id.is_empty() && id.len() <= MAX_ID_BYTES && !id.chars().any(char::is_control) +} + +/// Whether a studio `Custom` frame is the peer link's (then the shell +/// never hands it to the AI bus or the WM, parsed or not). +pub fn is_peer_frame(frame: &str) -> bool { + frame.trim_start().starts_with("{\"octos_peer\"") +} + +impl Up { + /// `None` for anything not a well-formed up frame: not ours, over the + /// cap, an unknown method, args that are not an object, a bad id. + pub fn parse(frame: &str) -> Option { + if frame.len() > MAX_FRAME_BYTES || !is_peer_frame(frame) { + return None; + } + let value: Value = serde_json::from_str(frame).ok()?; + let outer = value.as_object()?; + if outer.len() != 1 { + return None; + } + let v = outer.get(PEER_KEY)?.as_object()?; + match v.get("up")?.as_str()? { + "request" => { + let req_id = v.get("req_id")?.as_u64()?; + let method = v.get("method")?.as_str()?; + if !PEER_METHODS.contains(&method) { + return None; + } + let args = match v.get("args") { + None => Map::new(), + Some(Value::Object(m)) => m.clone(), + Some(_) => return None, + }; + Some(Up::Request { req_id, method: method.to_string(), args }) + } + "tool_result" => { + let call_id = v.get("call_id")?.as_str().filter(|c| id_ok(c))?.to_string(); + let outcome = if v.get("awaiting_confirmation").and_then(Value::as_bool) == Some(true) { + Outcome::AwaitingConfirmation + } else if v.get("ok")?.as_bool()? { + Outcome::Ok(v.get("data").cloned().unwrap_or(Value::Null)) + } else { + Outcome::Error(v.get("error").and_then(Value::as_str).unwrap_or("failed").to_string()) + }; + Some(Up::ToolResult { call_id, outcome }) + } + _ => None, + } + } +} + +impl Down { + pub fn to_json(&self) -> String { + let inner = match self { + Down::Reply { req_id, result } => match result { + Ok(data) => json!({"down": "reply", "req_id": req_id, "ok": true, "data": data}), + Err(error) => json!({"down": "reply", "req_id": req_id, "ok": false, "error": error}), + }, + Down::Event { req_id, event } => json!({"down": "event", "req_id": req_id, "event": event}), + Down::ToolCall(c) => json!({ + "down": "tool_call", + "call_id": c.call_id, + "name": c.name, + "args": c.args, + "risk": c.risk.as_str(), + "confirm_required": c.confirm_required, + "timeout_ms": c.timeout_ms, + "account": c.account, + "context_id": c.context_id, + "client": c.client, + "caller": c.caller, + }), + Down::ToolCancel { call_id } => json!({"down": "tool_cancel", "call_id": call_id}), + Down::ContextClosed { context, reason } => json!({"down": "context_closed", "context": context, "reason": reason}), + }; + json!({ PEER_KEY: inner }).to_string() + } + + /// For tests and the recording link: the frame read back. + pub fn parse(frame: &str) -> Option { + let value: Value = serde_json::from_str(frame).ok()?; + let v = value.get(PEER_KEY)?.as_object()?; + let opt = |k: &str| v.get(k).and_then(Value::as_str).map(str::to_string); + Some(match v.get("down")?.as_str()? { + "reply" => Down::Reply { + req_id: v.get("req_id")?.as_u64()?, + result: if v.get("ok")?.as_bool()? { Ok(v.get("data").cloned().unwrap_or(Value::Null)) } else { Err(opt("error").unwrap_or_default()) }, + }, + "event" => Down::Event { req_id: v.get("req_id")?.as_u64()?, event: v.get("event")?.clone() }, + "tool_call" => Down::ToolCall(ToolCallDown { + call_id: opt("call_id")?, + name: opt("name")?, + args: v.get("args")?.clone(), + risk: match v.get("risk")?.as_str()? { + "read" => Risk::Read, + "act" => Risk::Act, + _ => Risk::Destructive, + }, + confirm_required: v.get("confirm_required")?.as_bool()?, + timeout_ms: v.get("timeout_ms")?.as_u64()?, + account: opt("account"), + context_id: opt("context_id"), + client: opt("client"), + caller: opt("caller")?, + }), + "tool_cancel" => Down::ToolCancel { call_id: opt("call_id")? }, + "context_closed" => Down::ContextClosed { context: opt("context")?, reason: opt("reason").unwrap_or_default() }, + _ => return None, + }) + } +}