diff --git a/docs/agents.md b/docs/agents.md index 61780362b3..a5e9cd2c02 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -79,10 +79,11 @@ servers do not support these operations. In communities the build flag `BUZZ_OWNER_MCP_COMMUNITIES` enables, an agent's runs in this app on a bundled harness list `owner_channel_list`, `owner_channel_get`, -`owner_member_list`, `owner_read`, `owner_canvas_get` and `owner_canvas_history`. +`owner_member_list`, `owner_read`, `owner_canvas_get`, `owner_canvas_history` +and `owner_media_get`. They read the owner's channels, one channel's details and the owner's rights in it, its roster, its messages, whole threads and searches, -and its Canvas and earlier revisions, through the +its Canvas and earlier revisions, and the community's media, through the owner's own signed-in session, including private and archived channels, DMs and Solo chats the agent is not in. Message reads and Canvas history page with a `next` cursor, as `read` and the app's history do. A search covers only the @@ -91,10 +92,16 @@ channels' matches still fill its page, so `limited` with few results means narrowing it by channel or time. A message shows its newest edit by its author or, as in the app, by the author's verified owner. One read fetches a page's edits; when they fill it, the answer says `edits_limited` and some messages may show earlier text. They write -nothing: no membership, subscription, read state, post, agent start or Canvas +nothing to the community: no membership, subscription, read state, post, agent start or Canvas change. Content comes from the relay's own answer, not the owner's open views, -so it changes no unread counts. They do not save the owner's attachments: a -saved file would be readable before the call's final checks. The app +so it changes no unread counts. `owner_media_get` gets one media blob of the +agent's community, by its URL or `sha256[.ext]`, as `media_get` does but signed +as the owner: the app holds it in memory, up to 500 MiB, until protection has +reviewed the answer, and saves it to a local file only after the call's final +checks, while the owner is still signed in and the agent was not deleted. A +refused, cancelled or failed call saves nothing. Agents running on a remote +execution target do not list it and are refused it, since its file would be +on this device. The app decides the agent, owner and community; answers name them in `executedAs`. Each call needs the agent's owner signed in to that community now and fails, returning nothing, when the account, community diff --git a/docs/buzz-help.md b/docs/buzz-help.md index 71781afbe0..cb6ff03ddf 100644 --- a/docs/buzz-help.md +++ b/docs/buzz-help.md @@ -836,10 +836,11 @@ available rather than silently discarding it. - In communities where your build enables it, an agent running on this device with a built-in harness can list your channels, including private and - archived ones, DMs and Solo chats it is not in, and see a channel's details - and members, read with your access. It cannot change anything this way, and - it stops working when you sign out, switch accounts, disconnect or stop its - turn. + archived ones, DMs and Solo chats it is not in, and read a channel's details, + members, messages and Canvas, and the community's files, with your access. + It cannot change anything this way, and it stops working when you sign out, + switch accounts, disconnect or stop its turn. An agent on a remote execution + target cannot get files this way. Saved teams from **Your teams** appear in the inline `@` suggestions and the mention picker. Select a team to insert its members as individual mentions. diff --git a/src-tauri/build.rs b/src-tauri/build.rs index 07008803b5..748edbc8ea 100644 --- a/src-tauri/build.rs +++ b/src-tauri/build.rs @@ -208,6 +208,9 @@ fn main() { "app_agent_search_profiles", "app_agent_upload", "app_agent_media", + "app_agent_owner_media_stage", + "app_agent_owner_media_publish", + "app_agent_owner_media_discard", "app_agent_git_auth", "app_agent_remember", "app_agent_claim", diff --git a/src-tauri/capabilities/default.json b/src-tauri/capabilities/default.json index b0399c052d..e0d48c711e 100644 --- a/src-tauri/capabilities/default.json +++ b/src-tauri/capabilities/default.json @@ -154,6 +154,9 @@ "allow-app-agent-search-profiles", "allow-app-agent-upload", "allow-app-agent-media", + "allow-app-agent-owner-media-stage", + "allow-app-agent-owner-media-publish", + "allow-app-agent-owner-media-discard", "allow-app-agent-git-auth", "allow-app-agent-remember", "allow-app-agent-claim", diff --git a/src-tauri/src/app_agents.rs b/src-tauri/src/app_agents.rs index dc8c1c7bec..90674ada37 100644 --- a/src-tauri/src/app_agents.rs +++ b/src-tauri/src/app_agents.rs @@ -20,6 +20,8 @@ use crate::federated_identity::{ assertion, }; +pub(crate) mod owner_media; + /// A NIP-98 `Authorization` header for one signed HTTP auth event. fn authorization(event: Value) -> Result { use base64::Engine; @@ -56,6 +58,8 @@ pub(crate) struct AppAgentHost { tokens: AgentTokens, /// Serializes owner prompts, without holding cross-process identity/custody locks. protection_upgrade: Arc>, + /// Owner media read for review and not yet published. + owner_media: Arc, } impl AppAgentHost { /// `packaged` is false for a dev build, which steps aside for a packaged app. @@ -103,6 +107,7 @@ impl AppAgentHost { }), tokens, protection_upgrade: Arc::default(), + owner_media: Arc::default(), } } /// The saved identities, for readers outside Agents such as the inventory. @@ -119,6 +124,12 @@ impl AppAgentHost { let mut removals = self.removals(); removals.page += 1; removals.held.clear(); + drop(removals); + self.owner_media.clear(); + } + /// Drops held owner media, as the window closing publishes none. + pub(crate) fn owner_media_clear(&self) { + self.owner_media.clear(); } fn keys(&self) -> std::sync::MutexGuard<'_, HashMap>> { self.keys @@ -1412,19 +1423,7 @@ async fn save_media_admitted( } } agent.verify_media_hash(url, &format!("{:x}", hash.finalize()))?; - // The blob's own name, so the same media is saved once; with an extension - // from its type when the URL gives none. - let mut name = url.rsplit('/').next().unwrap_or_default().to_owned(); - if !name.contains('.') { - if let Some(ext) = mime - .split_once('/') - .map(|(_, subtype)| subtype) - .filter(|ext| ext.len() <= 8 && ext.bytes().all(|b| b.is_ascii_alphanumeric())) - { - name = format!("{name}.{ext}"); - } - } - let path = dir.join(name); + let path = dir.join(media_name(url, &mime)); let persist = || { file.persist(&path) .map(drop) @@ -1444,6 +1443,22 @@ async fn save_media_admitted( }) } +/// The blob's own name, so the same media is saved once; with an extension +/// from its type when the URL gives none. +fn media_name(url: &str, mime: &str) -> String { + let mut name = url.rsplit('/').next().unwrap_or_default().to_owned(); + if !name.contains('.') { + if let Some(ext) = mime + .split_once('/') + .map(|(_, subtype)| subtype) + .filter(|ext| ext.len() <= 8 && ext.bytes().all(|b| b.is_ascii_alphanumeric())) + { + name = format!("{name}.{ext}"); + } + } + name +} + /// Writes one memory entry, encrypted to the agent's owner; a null body /// deletes it. `after` is the `created_at` of the entry it replaces, or 0. #[tauri::command] @@ -1743,6 +1758,7 @@ mod tests { hosting: Arc::new(hosting(Err("unused".into()), true)), tokens: AgentTokens::unconfigured(), protection_upgrade: Arc::default(), + owner_media: Arc::default(), }; let agent = AppAgent { pubkey: key.pubkey().into(), @@ -1932,6 +1948,7 @@ mod tests { hosting: Arc::new(hosting(Err("unused".into()), true)), tokens: f.tokens.clone(), protection_upgrade: Arc::default(), + owner_media: Arc::default(), }; let import = |key: &Secret| { let key = Secret::parse(&key.hex(), key.pubkey()).unwrap(); @@ -2028,6 +2045,7 @@ mod tests { hosting: Arc::new(hosting(Err("unused".into()), true)), tokens: f.tokens.clone(), protection_upgrade: Arc::default(), + owner_media: Arc::default(), }; let agents = host.agents().unwrap(); @@ -2073,6 +2091,7 @@ mod tests { hosting: Arc::new(hosting(Err("unused".into()), true)), tokens: f.tokens.clone(), protection_upgrade: Arc::default(), + owner_media: Arc::default(), }; let import = || { let key = Secret::parse(&key.hex(), key.pubkey()).unwrap(); @@ -2523,6 +2542,7 @@ mod tests { hosting: Arc::new(hosting(Err("unused".into()), true)), tokens: AgentTokens::unconfigured(), protection_upgrade: Arc::default(), + owner_media: Arc::default(), }; let import = || { let prepared = AppAgents::adopt( diff --git a/src-tauri/src/app_agents/owner_media.rs b/src-tauri/src/app_agents/owner_media.rs new file mode 100644 index 0000000000..da434b0f78 --- /dev/null +++ b/src-tauri/src/app_agents/owner_media.rs @@ -0,0 +1,458 @@ +//! An agent's owner media (`owner_media_get`): one of its community's blobs, +//! read with its owner's own access into memory only. The WebView reviews the +//! reply, then publishes it: only then, and only while the owner is still +//! signed in and the agent was not deleted, is any byte written to a file. +//! Until then a discard, a reload or the window closing drops it. +use std::collections::HashMap; +use std::io::Write; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +use serde::Serialize; +use tokio::sync::oneshot; + +use super::{media_name, mime_type, AppAgentHost, MEDIA_FILE_LIMIT, MEDIA_LIMIT, SHOWN_TYPES}; +use crate::federated_identity::{agent::Stamp, FederatedIdentity}; +use crate::identity::IdentityHost; + +/// Media held at once, and the bytes they may reserve: one of the largest. +const OPS: usize = 16; +const BUDGET: u64 = 512 * 1024 * 1024; +/// Result protection waits up to 310 seconds for its checker; then a staged +/// blob, or a discard that overtook its stage, is dropped. +const LIFETIME: Duration = Duration::from_secs(330); +const TOMBSTONES: usize = 64; +/// Written while the owner and agent are held; the rest after. +const FIRST_WRITE: usize = 64 * 1024; + +const CANCELLED: &str = "The owner media request was cancelled; nothing was saved"; +const BUSY: &str = "Owner media is busy; try again shortly"; +const NOTHING: &str = "This owner media request saved nothing; get it again"; + +struct Staged { + agent: String, + relay: String, + owner: String, + stamp: Stamp, + retirements: u64, + name: String, + body: Vec, + at: Instant, +} +enum Op { + Fetching { + cancel: Option>, + reserved: u64, + }, + Staged(Staged), + /// Taken by its publication, whose body still counts until it is dropped. + Publishing { + reserved: u64, + }, +} +impl Op { + fn reserved(&self) -> u64 { + match self { + Op::Fetching { reserved, .. } | Op::Publishing { reserved } => *reserved, + Op::Staged(staged) => staged.body.capacity() as u64, + } + } +} +#[derive(Default)] +struct State { + ops: HashMap, + /// Discards that found nothing: one may have overtaken its stage. + cancelled: HashMap, +} +impl State { + fn prune(&mut self, now: Instant) { + self.ops.retain(|_, op| match op { + Op::Staged(staged) => now.duration_since(staged.at) < LIFETIME, + Op::Fetching { .. } | Op::Publishing { .. } => true, + }); + self.cancelled + .retain(|_, at| now.duration_since(*at) < LIFETIME); + } +} + +/// Where a test holds a publication: past its peek, or inside the fence. +#[cfg(test)] +#[derive(Clone, Copy, PartialEq)] +enum Pause { + Peeked, + Fenced, +} +#[cfg(test)] +type Gate = Option<( + Pause, + std::sync::mpsc::Sender<()>, + std::sync::mpsc::Receiver<()>, +)>; + +/// Owner media between its read and its publication, in memory only. +#[derive(Default)] +pub(crate) struct OwnerMedia(Mutex, #[cfg(test)] Mutex); +impl OwnerMedia { + fn lock(&self) -> std::sync::MutexGuard<'_, State> { + self.0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + /// Registers `id`'s read, unless a discard already came for it. + fn begin(&self, id: &str) -> Result, String> { + let mut state = self.lock(); + state.prune(Instant::now()); + if state.cancelled.remove(id).is_some() { + return Err(CANCELLED.into()); + } + if state.ops.contains_key(id) { + return Err("This owner media request is already in progress".into()); + } + if state.ops.len() >= OPS { + return Err(BUSY.into()); + } + let (cancel, cancelled) = oneshot::channel(); + state.ops.insert( + id.into(), + Op::Fetching { + cancel: Some(cancel), + reserved: 0, + }, + ); + Ok(cancelled) + } + /// Reserves `bytes` for `id`'s body before any of it is read. + fn reserve(&self, id: &str, bytes: u64) -> Result<(), String> { + let mut state = self.lock(); + state.prune(Instant::now()); + let others: u64 = state + .ops + .iter() + .filter(|(key, _)| key.as_str() != id) + .map(|(_, op)| op.reserved()) + .sum(); + let Some(Op::Fetching { reserved, .. }) = state.ops.get_mut(id) else { + return Err(CANCELLED.into()); + }; + if others + bytes > BUDGET { + return Err(BUSY.into()); + } + *reserved = bytes; + Ok(()) + } + /// Keeps `id`'s body until it is published or discarded. + fn stage(&self, id: &str, staged: Staged) -> Result<(), String> { + let mut state = self.lock(); + match state.ops.get_mut(id) { + Some(op @ Op::Fetching { .. }) => { + *op = Op::Staged(staged); + Ok(()) + } + _ => Err(CANCELLED.into()), + } + } + /// Forgets a read that failed. + fn end(&self, id: &str) { + let mut state = self.lock(); + if matches!(state.ops.get(id), Some(Op::Fetching { .. })) { + state.ops.remove(id); + } + } + /// Drops `id`'s read or body; one not seen yet is refused when it comes. + pub(crate) fn discard(&self, id: &str) { + let mut state = self.lock(); + let now = Instant::now(); + state.prune(now); + if matches!(state.ops.get(id), Some(Op::Publishing { .. })) { + return; + } + if let Some(op) = state.ops.remove(id) { + if let Op::Fetching { + cancel: Some(cancel), + .. + } = op + { + let _ = cancel.send(()); + } + return; + } + if state.cancelled.len() >= TOMBSTONES && !state.cancelled.contains_key(id) { + // As `Uploads`: retire the oldest so a new discard still wins. + if let Some(old) = state + .cancelled + .iter() + .min_by_key(|(_, at)| **at) + .map(|(key, _)| key.clone()) + { + state.cancelled.remove(&old); + } + } + state.cancelled.insert(id.into(), now); + } + /// Who `id`'s staged body was read for, without taking it. + fn peek(&self, id: &str, agent: &str, relay: &str) -> Result<(String, Stamp, u64), String> { + let mut state = self.lock(); + state.prune(Instant::now()); + match state.ops.get(id) { + Some(Op::Staged(staged)) if staged.agent == agent && staged.relay == relay => { + Ok((staged.owner.clone(), staged.stamp, staged.retirements)) + } + _ => Err(NOTHING.into()), + } + } + /// Takes `id`'s staged body for `agent` in `relay`; it can be taken once. + /// It stays counted until [`OwnerMedia::published`]. + fn take(&self, id: &str, agent: &str, relay: &str) -> Result { + let mut state = self.lock(); + state.prune(Instant::now()); + let Some(op) = state.ops.get_mut(id) else { + return Err(NOTHING.into()); + }; + match op { + Op::Staged(staged) if staged.agent == agent && staged.relay == relay => { + let reserved = op.reserved(); + match std::mem::replace(op, Op::Publishing { reserved }) { + Op::Staged(staged) => Ok(staged), + _ => unreachable!(), + } + } + _ => Err(NOTHING.into()), + } + } + /// Releases what `id`'s publication held, once its body is dropped. + fn published(&self, id: &str) { + let mut state = self.lock(); + if matches!(state.ops.get(id), Some(Op::Publishing { .. })) { + state.ops.remove(id); + } + } + /// Drops every read and body, as a new page or a closed window starts none. + /// A publication already writing keeps its body counted until it drops it. + pub(crate) fn clear(&self) { + let mut state = self.lock(); + state.ops.retain(|_, op| match op { + Op::Fetching { cancel, .. } => { + if let Some(cancel) = cancel.take() { + let _ = cancel.send(()); + } + false + } + Op::Staged(_) => false, + Op::Publishing { .. } => true, + }); + state.cancelled.clear(); + } + /// Holds the next publication at `at`: returns its arrival and its release. + #[cfg(test)] + fn hold_at(&self, at: Pause) -> (std::sync::mpsc::Receiver<()>, std::sync::mpsc::Sender<()>) { + let (reached, arrived) = std::sync::mpsc::channel(); + let (release, released) = std::sync::mpsc::channel(); + *self.1.lock().unwrap() = Some((at, reached, released)); + (arrived, release) + } + #[cfg(test)] + fn pause(&self, at: Pause) { + let gate = self.1.lock().unwrap().take_if(|(point, ..)| *point == at); + if let Some((_, reached, release)) = gate { + let _ = reached.send(()); + let _ = release.recv_timeout(Duration::from_secs(10)); + } + } + #[cfg(test)] + fn held(&self) -> (usize, u64) { + let state = self.lock(); + (state.ops.len(), state.ops.values().map(Op::reserved).sum()) + } +} + +/// Owner media read and held for review; nothing is at `path` yet. +#[derive(Debug, Serialize, PartialEq)] +pub(crate) struct OwnerMediaStaged { + path: String, + #[serde(rename = "type")] + mime: String, + size: u64, + #[serde(skip_serializing_if = "Option::is_none")] + data: Option, +} + +fn dir() -> PathBuf { + std::env::temp_dir().join("buzz-agent-media") +} + +fn valid_id(id: &str) -> Result<(), String> { + if id.is_empty() || id.len() > 64 || !id.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-') + { + return Err("Invalid owner media request".into()); + } + Ok(()) +} + +/// Reads one of the agent's community media blobs (its URL or `sha256[.ext]`) +/// with its owner's access, into memory, as request `id`. +#[tauri::command] +pub(crate) async fn app_agent_owner_media_stage( + state: tauri::State<'_, AppAgentHost>, + identity: tauri::State<'_, IdentityHost>, + fi: tauri::State<'_, FederatedIdentity>, + pubkey: String, + relay: String, + id: String, + url: String, +) -> Result { + stage(&state, &identity, Some(&fi), &dir(), pubkey, relay, id, url).await +} + +#[allow(clippy::too_many_arguments)] +async fn stage( + host: &AppAgentHost, + identity: &IdentityHost, + fi: Option<&FederatedIdentity>, + dir: &Path, + pubkey: String, + relay: String, + id: String, + url: String, +) -> Result { + use base64::Engine; + use sha2::{Digest, Sha256}; + valid_id(&id)?; + let media = host.owner_media.clone(); + let cancelled = media.begin(&id)?; + let read = async { + // As a local caller's token: retirements, then the stamp, then the row. + let retirements = host.caller_retirements(); + let (agent, stamp) = host.caller_stamp(pubkey.clone(), relay.clone()).await?; + if host.caller_closed() { + return Err("Your owner signed out of Buzz here".to_owned()); + } + let url = agent.media_url(&url)?; + let target = url::Url::parse(&url).map_err(|_| "Invalid media URL")?; + let (kind, body) = crate::relay::owner_media( + identity, + fi, + &agent.owner, + target, + MEDIA_FILE_LIMIT, + |bytes| media.reserve(&id, bytes), + ) + .await?; + agent.verify_media_hash(&url, &format!("{:x}", Sha256::digest(&body)))?; + let mime = kind + .as_deref() + .and_then(mime_type) + .unwrap_or_else(|| "application/octet-stream".into()); + let shown = SHOWN_TYPES.contains(&mime.as_str()) && body.len() as u64 <= MEDIA_LIMIT; + let name = media_name(&url, &mime); + let reply = OwnerMediaStaged { + path: dir.join(&name).to_string_lossy().into_owned(), + data: shown.then(|| base64::engine::general_purpose::STANDARD.encode(&body)), + size: body.len() as u64, + mime, + }; + media.stage( + &id, + Staged { + agent: agent.pubkey, + relay: relay.clone(), + owner: agent.owner, + stamp, + retirements, + name, + body, + at: Instant::now(), + }, + )?; + Ok(reply) + }; + let result = tokio::select! { + biased; + _ = cancelled => Err(CANCELLED.to_owned()), + result = read => result, + }; + if result.is_err() { + media.end(&id); + } + result +} + +/// Writes request `id`'s reviewed media to its path. Refused, with nothing +/// written, unless it is still held and its owner and agent still act. +#[tauri::command] +pub(crate) async fn app_agent_owner_media_publish( + state: tauri::State<'_, AppAgentHost>, + identity: tauri::State<'_, IdentityHost>, + pubkey: String, + relay: String, + id: String, +) -> Result<(), String> { + let (host, identity) = (state.inner().clone(), identity.inner().clone()); + super::blocking(move || publish(&host, &identity, &dir(), &pubkey, &relay, &id)).await +} + +fn publish( + host: &AppAgentHost, + identity: &IdentityHost, + dir: &Path, + pubkey: &str, + relay: &str, + id: &str, +) -> Result<(), String> { + valid_id(id)?; + let (owner, stamp, retirements) = host.owner_media.peek(id, pubkey, relay)?; + #[cfg(test)] + host.owner_media.pause(Pause::Peeked); + // Declared first, so dropped after the body it keeps counted. + let _published = Published(&host.owner_media, id); + // The order every credential owner takes: identity, then the agent tokens' + // locks, then this request's. Sign-out, deletion and a change of account + // either refuse this or wait for its first bytes. + let held = identity.hold_owner(&owner); + let (staged, mut file) = match held { + Err(error) => { + host.owner_media.discard(id); + return Err(format!("{error}; nothing was saved")); + } + Ok(_held) => host + .tokens + .commit_caller(pubkey, relay, stamp, retirements, || { + #[cfg(test)] + host.owner_media.pause(Pause::Fenced); + let staged = host.owner_media.take(id, pubkey, relay)?; + std::fs::create_dir_all(dir).map_err(|_| "Media storage unavailable")?; + let mut file = tempfile::NamedTempFile::new_in(dir) + .map_err(|_| "Media storage unavailable")?; + let first = staged.body.len().min(FIRST_WRITE); + file.write_all(&staged.body[..first]) + .map_err(|_| "Media storage unavailable")?; + Ok((staged, file)) + }) + .inspect_err(|_| host.owner_media.discard(id))?, + }; + let path = dir.join(&staged.name); + let first = staged.body.len().min(FIRST_WRITE); + file.write_all(&staged.body[first..]) + .and_then(|()| file.flush()) + .map_err(drop) + .and_then(|()| file.persist(&path).map(drop).map_err(drop)) + .map_err(|()| format!("Saving {} failed; nothing was saved", path.display())) +} + +struct Published<'a>(&'a OwnerMedia, &'a str); +impl Drop for Published<'_> { + fn drop(&mut self) { + self.0.published(self.1); + } +} + +/// Drops request `id`'s read or held media; one still to come is refused. +#[tauri::command] +pub(crate) fn app_agent_owner_media_discard(state: tauri::State<'_, AppAgentHost>, id: String) { + if valid_id(&id).is_ok() { + state.owner_media.discard(&id); + } +} + +#[cfg(test)] +mod tests; diff --git a/src-tauri/src/app_agents/owner_media/tests.rs b/src-tauri/src/app_agents/owner_media/tests.rs new file mode 100644 index 0000000000..2e3427b60f --- /dev/null +++ b/src-tauri/src/app_agents/owner_media/tests.rs @@ -0,0 +1,538 @@ +use super::*; +use crate::federated_identity::agent::{tests::response_fixture, AgentTokens}; +use crate::federated_identity::testing::{community, Community}; +use buzz_agent_controller::{AppAgent, Secret}; +use sha2::{Digest, Sha256}; +use std::sync::Arc; + +const PNG: &str = "\u{89}PNG fixture body"; + +fn reply(headers: &str, body: &str) -> String { + format!("HTTP/1.1 200 OK\r\n{headers}Connection: close\r\n\r\n{body}") +} +fn sha(body: &str) -> String { + format!("{:x}", Sha256::digest(body.as_bytes())) +} + +/// An agent of the community's signed-in viewer (or of `owner`), held by a host. +async fn agent_of( + community: &Community, + owner: Option, + tokens: AgentTokens, +) -> (AppAgentHost, AppAgent, tempfile::TempDir) { + let key = Secret::generate().unwrap(); + let owner = match owner { + Some(owner) => owner, + None => community.host.viewer().await.unwrap(), + }; + let agent: AppAgent = serde_json::from_value(serde_json::json!({ + "pubkey": key.pubkey(), + "relay": format!("wss://{}", community.origin.authority()), + "owner": owner, + "auth": r#"["auth","owner","conditions","sig"]"#, "type": "p/t", "name": "Ada" + })) + .unwrap(); + let dir = tempfile::tempdir().unwrap(); + let host = AppAgentHost::holding(dir.path().join("identities.json"), &agent, key, tokens); + (host, agent, dir) +} + +async fn stage_at( + host: &AppAgentHost, + community: &Community, + agent: &AppAgent, + dir: &Path, + id: &str, + url: &str, +) -> Result { + stage( + host, + &community.host, + Some(&community.fi), + dir, + agent.pubkey.clone(), + agent.relay.clone(), + id.into(), + url.into(), + ) + .await +} + +#[tokio::test] +async fn owner_media_is_read_as_the_owner_and_written_only_when_published() { + use base64::Engine; + for required in [false, true] { + let community = community(required).await; + let (host, agent, _store) = agent_of(&community, None, AgentTokens::unconfigured()).await; + let served = community.serve(vec![reply( + &format!( + "Content-Type: image/png\r\nContent-Length: {}\r\n", + PNG.len() + ), + PNG, + )]); + let dir = tempfile::tempdir().unwrap(); + let name = format!("{}.png", sha(PNG)); + let staged = stage_at(&host, &community, &agent, dir.path(), "one", &name) + .await + .unwrap(); + let head = &served.heads().await[0]; + community.assert_carried(head); + let auth = head + .lines() + .find_map(|line| line.strip_prefix("authorization: Nostr ")) + .expect(head); + let event: serde_json::Value = serde_json::from_slice( + &base64::engine::general_purpose::STANDARD + .decode(auth.trim()) + .unwrap(), + ) + .unwrap(); + assert_eq!( + event["pubkey"], agent.owner, + "signed as the owner, not the agent" + ); + assert_eq!(event["kind"], 24242); + assert_eq!(staged.mime, "image/png"); + assert_eq!(staged.size, PNG.len() as u64); + assert!(staged.data.is_some(), "a small image is shown"); + assert_eq!(staged.path, dir.path().join(&name).to_string_lossy()); + assert_eq!( + std::fs::read_dir(dir.path()).map_or(0, Iterator::count), + 0, + "nothing is on disk before publication" + ); + publish( + &host, + &community.host, + dir.path(), + &agent.pubkey, + &agent.relay, + "one", + ) + .unwrap(); + assert_eq!( + std::fs::read(dir.path().join(&name)).unwrap(), + PNG.as_bytes() + ); + assert_eq!( + publish( + &host, + &community.host, + dir.path(), + &agent.pubkey, + &agent.relay, + "one" + ) + .unwrap_err(), + NOTHING, + "published once" + ); + assert_eq!(host.owner_media.held(), (0, 0)); + } +} + +#[tokio::test] +async fn owner_media_is_never_signed_or_sent_for_another_owner() { + let community = community(true).await; + let (host, agent, _store) = agent_of( + &community, + Some("b".repeat(64)), + AgentTokens::unconfigured(), + ) + .await; + // Nothing answers: any request, discovery included, would hang here. + let dir = tempfile::tempdir().unwrap(); + let refused = tokio::time::timeout( + Duration::from_secs(5), + stage_at(&host, &community, &agent, dir.path(), "one", &sha(PNG)), + ) + .await + .expect("refused before any request"); + assert_eq!( + refused.unwrap_err(), + "Your owner is not signed in to Buzz here now" + ); + assert_eq!(host.owner_media.held(), (0, 0)); +} + +#[tokio::test] +async fn owner_media_is_only_the_agents_community_blob() { + let community = community(false).await; + let (host, agent, _store) = agent_of(&community, None, AgentTokens::unconfigured()).await; + let dir = tempfile::tempdir().unwrap(); + for url in [ + "https://elsewhere.example.org/media/".to_owned() + &sha(PNG), + format!("{}upload", community.origin), + "../secret".into(), + ] { + assert_eq!( + stage_at(&host, &community, &agent, dir.path(), "one", &url) + .await + .unwrap_err(), + "Agents fetch only their community's media" + ); + } + // A body that is not the blob named is dropped. + let served = community.serve(vec![reply("Content-Length: 5\r\n", "other")]); + assert_eq!( + stage_at(&host, &community, &agent, dir.path(), "two", &sha(PNG)) + .await + .unwrap_err(), + "Media did not match its SHA-256" + ); + served.heads().await; + assert_eq!(host.owner_media.held(), (0, 0)); + for id in ["", "a/b", &"x".repeat(65)] { + assert!( + stage_at(&host, &community, &agent, dir.path(), id, &sha(PNG)) + .await + .is_err() + ); + } +} + +#[tokio::test] +async fn owner_media_bodies_are_bounded_by_what_was_reserved() { + let community = community(false).await; + let owner = community.host.viewer().await.unwrap(); + let url = community + .origin + .join(&format!("/media/{}", sha(PNG))) + .unwrap(); + let read = |limit, reserved: Arc>>| { + crate::relay::owner_media( + &community.host, + Some(&community.fi), + &owner, + url.clone(), + limit, + move |bytes| { + reserved.lock().unwrap().push(bytes); + Ok(()) + }, + ) + }; + // No length: the limit is reserved and a longer body is refused. + let reserved = Arc::default(); + let served = community.serve(vec![reply("", "0123456789")]); + assert!(read(8, Arc::clone(&reserved)).await.is_err()); + served.heads().await; + assert_eq!(*reserved.lock().unwrap(), [8]); + // Nor does one within it grow its buffer past what was reserved. + let served = community.serve(vec![reply("", "01234")]); + let (_, body) = read(5, Arc::default()).await.unwrap(); + served.heads().await; + assert_eq!((body.as_slice(), body.capacity()), (&b"01234"[..], 5)); + // A declared length over the limit reserves nothing. + let reserved = Arc::default(); + let served = community.serve(vec![reply("Content-Length: 10\r\n", "0123456789")]); + assert!(read(8, Arc::clone(&reserved)).await.is_err()); + served.heads().await; + assert!(reserved.lock().unwrap().is_empty()); + // A declared length within it is what is reserved. + let reserved = Arc::default(); + let served = community.serve(vec![reply("Content-Length: 4\r\n", "0123")]); + let (_, body) = read(8, Arc::clone(&reserved)).await.unwrap(); + served.heads().await; + assert_eq!( + (body.as_slice(), reserved.lock().unwrap()[0]), + (&b"0123"[..], 4) + ); + // A refused reservation reads no body. + let served = community.serve(vec![reply("Content-Length: 4\r\n", "0123")]); + let refused = crate::relay::owner_media( + &community.host, + Some(&community.fi), + &owner, + url.clone(), + 8, + |_| Err(BUSY.into()), + ) + .await; + served.heads().await; + assert_eq!(refused.unwrap_err(), BUSY); +} + +fn staged(agent: &str, relay: &str, owner: &str, stamp: Stamp, retirements: u64) -> Staged { + Staged { + agent: agent.into(), + relay: relay.into(), + owner: owner.into(), + stamp, + retirements, + name: "blob.png".into(), + body: PNG.as_bytes().to_vec(), + at: Instant::now(), + } +} + +#[test] +fn reads_share_one_bounded_budget() { + let media = OwnerMedia::default(); + let _a = media.begin("a").unwrap(); + let _b = media.begin("b").unwrap(); + media.reserve("a", BUDGET).unwrap(); + assert_eq!(media.reserve("b", 1).unwrap_err(), BUSY); + media.end("a"); + media.reserve("b", 1).unwrap(); + assert_eq!(media.held(), (1, 1)); + let receivers: Vec<_> = (0..OPS - 1) + .map(|n| media.begin(&format!("n{n}")).unwrap()) + .collect(); + assert_eq!(media.begin("over").unwrap_err(), BUSY); + drop(receivers); + media.clear(); + assert_eq!(media.held(), (0, 0)); +} + +#[tokio::test] +async fn a_discard_cancels_its_read_even_when_it_overtakes_the_stage() { + let media = OwnerMedia::default(); + // During the read: it is told to stop, and cannot reserve or stage. + let cancelled = media.begin("a").unwrap(); + media.discard("a"); + cancelled.await.unwrap(); + assert_eq!(media.reserve("a", 1).unwrap_err(), CANCELLED); + let staged = || staged("agent", "wss://relay.example", "owner", Stamp::default(), 0); + assert_eq!(media.stage("a", staged()).unwrap_err(), CANCELLED); + // Before the read began: the read is refused once. + media.discard("b"); + assert_eq!(media.begin("b").unwrap_err(), CANCELLED); + let _b = media.begin("b").unwrap(); + // After staging: the body is gone. + media.stage("b", staged()).unwrap(); + media.discard("b"); + assert_eq!(media.held(), (0, 0)); + // Tombstones are bounded, keeping the newest. + for n in 0..TOMBSTONES + 8 { + media.discard(&format!("t{n}")); + } + assert_eq!(media.lock().cancelled.len(), TOMBSTONES); + assert!(media.begin(&format!("t{}", TOMBSTONES + 7)).is_err()); + // A reload drops them, and every read and body. + let cancelled = media.begin("c").unwrap(); + media.clear(); + cancelled.await.unwrap(); + assert!(media.lock().cancelled.is_empty()); + assert_eq!(media.held(), (0, 0)); +} + +#[test] +fn a_staged_body_is_taken_once_by_its_own_agent_before_it_expires() { + let media = OwnerMedia::default(); + let relay = "wss://relay.example"; + let _a = media.begin("a").unwrap(); + media + .stage("a", staged("agent", relay, "owner", Stamp::default(), 0)) + .unwrap(); + assert_eq!(media.take("a", "other", relay).err().unwrap(), NOTHING); + assert_eq!( + media + .take("a", "agent", "wss://other.example") + .err() + .unwrap(), + NOTHING + ); + assert_eq!(media.held().0, 1, "a foreign call leaves it held"); + assert_eq!(media.peek("a", "agent", relay).unwrap().0, "owner"); + let taken = media.take("a", "agent", relay).unwrap(); + assert_eq!(taken.body, PNG.as_bytes()); + assert_eq!(media.take("a", "agent", relay).err().unwrap(), NOTHING); + // Until its publication drops it, the body still counts, even past a + // discard or a reload. + media.discard("a"); + media.clear(); + let counted = (1, taken.body.capacity() as u64); + assert_eq!(media.held(), counted); + drop(taken); + media.published("a"); + let _b = media.begin("b").unwrap(); + let mut old = staged("agent", relay, "owner", Stamp::default(), 0); + old.at = Instant::now().checked_sub(LIFETIME).unwrap(); + media.stage("b", old).unwrap(); + assert_eq!( + media.peek("b", "agent", relay).err().as_deref(), + Some(NOTHING) + ); + assert_eq!(media.held(), (0, 0)); +} + +/// The agent of a fixture's tokens, owned by `IdentityHost::fixture`'s viewer, +/// with one body staged. +async fn staged_for() -> ( + crate::federated_identity::agent::tests::Fixture, + AppAgentHost, + AppAgent, + tempfile::TempDir, +) { + let f = response_fixture().await; + let agent = AppAgent { + owner: IdentityHost::fixture().viewer().await.unwrap(), + ..f.agent.clone() + }; + let store = tempfile::tempdir().unwrap(); + let host = AppAgentHost::holding( + store.path().join("identities.json"), + &agent, + Secret::parse(&f.key.hex(), f.key.pubkey()).unwrap(), + f.tokens.clone(), + ); + let retirements = host.caller_retirements(); + let (_, stamp) = host + .caller_stamp(agent.pubkey.clone(), agent.relay.clone()) + .await + .unwrap(); + let _read = host.owner_media.begin("one").unwrap(); + host.owner_media + .stage( + "one", + staged( + &agent.pubkey, + &agent.relay, + &agent.owner, + stamp, + retirements, + ), + ) + .unwrap(); + (f, host, agent, store) +} + +#[tokio::test] +async fn nothing_is_written_once_the_owner_or_agent_stops_acting() { + for change in [ + "delete", + "sign-out", + "retirement-pending", + "identity-closed", + "none", + ] { + let identity = IdentityHost::fixture(); + let (f, host, agent, _store) = staged_for().await; + match change { + "delete" => f.tokens.forget(&agent.owner, &agent.pubkey).await.unwrap(), + "sign-out" => drop(f.tokens.sign_out(None).await.unwrap()), + "retirement-pending" => f.tokens.mark_retirement_pending(), + "identity-closed" => identity.close(), + _ => {} + } + let dir = tempfile::tempdir().unwrap(); + let published = publish( + &host, + &identity, + dir.path(), + &agent.pubkey, + &agent.relay, + "one", + ); + let files = std::fs::read_dir(dir.path()).map_or(0, Iterator::count); + if change == "none" { + published.unwrap(); + assert_eq!(files, 1); + } else { + assert!(published.is_err(), "{change}"); + assert_eq!(files, 0, "{change}: no file, not even a temporary one"); + } + assert_eq!( + host.owner_media.held(), + (0, 0), + "{change}: the body is dropped" + ); + } + // Another owner signed in here now. + let other = IdentityHost::fixture_signed_in_as(2); + let (_f, host, agent, _store) = staged_for().await; + let dir = tempfile::tempdir().unwrap(); + assert!(publish( + &host, + &other, + dir.path(), + &agent.pubkey, + &agent.relay, + "one" + ) + .is_err()); + assert_eq!(std::fs::read_dir(dir.path()).map_or(0, Iterator::count), 0); +} + +/// Publishes `agent`'s staged body on a blocking thread, as its command runs. +fn publishing( + host: &AppAgentHost, + identity: &IdentityHost, + dir: &Path, + agent: &AppAgent, +) -> tokio::task::JoinHandle> { + let (host, identity, dir) = (host.clone(), identity.clone(), dir.to_owned()); + let (pubkey, relay) = (agent.pubkey.clone(), agent.relay.clone()); + tokio::task::spawn_blocking(move || publish(&host, &identity, &dir, &pubkey, &relay, "one")) +} +async fn signalled(signal: std::sync::mpsc::Receiver<()>) { + tokio::task::spawn_blocking(move || signal.recv_timeout(Duration::from_secs(10))) + .await + .unwrap() + .expect("signalled"); +} +async fn finished(task: tokio::task::JoinHandle) -> T { + tokio::time::timeout(Duration::from_secs(10), task) + .await + .expect("finished") + .unwrap() +} + +/// Deletion, sign-out or a closed identity that comes after the review, while +/// publication is on its way to the fence, refuses it: nothing is written. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_retirement_before_the_fence_refuses_publication() { + for change in ["delete", "sign-out", "identity-closed"] { + let identity = IdentityHost::fixture(); + let (f, host, agent, _store) = staged_for().await; + let dir = tempfile::tempdir().unwrap(); + let (arrived, release) = host.owner_media.hold_at(Pause::Peeked); + let published = publishing(&host, &identity, dir.path(), &agent); + signalled(arrived).await; + match change { + "delete" => f.tokens.forget(&agent.owner, &agent.pubkey).await.unwrap(), + "sign-out" => drop(f.tokens.sign_out(None).await.unwrap()), + _ => identity.close(), + } + release.send(()).unwrap(); + assert!(finished(published).await.is_err(), "{change}"); + assert_eq!( + std::fs::read_dir(dir.path()).map_or(0, Iterator::count), + 0, + "{change}" + ); + assert_eq!( + host.owner_media.held(), + (0, 0), + "{change}: the body is dropped" + ); + } +} + +/// A sign-out that comes while publication holds the fence waits for it, so +/// the whole file is written, and then completes. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_sign_out_waits_for_publication_holding_the_fence() { + let identity = IdentityHost::fixture(); + let (f, host, agent, _store) = staged_for().await; + let dir = tempfile::tempdir().unwrap(); + let (arrived, release) = host.owner_media.hold_at(Pause::Fenced); + let published = publishing(&host, &identity, dir.path(), &agent); + signalled(arrived).await; + let contended = f.tokens.on_sign_out_contended(); + let tokens = f.tokens.clone(); + let signing_out = tokio::spawn(async move { tokens.sign_out(None).await.map(drop) }); + // The sign-out found the retirement lock held by the publication. + signalled(contended).await; + assert!(!signing_out.is_finished()); + release.send(()).unwrap(); + finished(published).await.unwrap(); + assert_eq!( + std::fs::read(dir.path().join("blob.png")).unwrap(), + PNG.as_bytes() + ); + finished(signing_out).await.unwrap(); + assert_eq!(host.owner_media.held(), (0, 0)); +} diff --git a/src-tauri/src/browser_permissions_tests.rs b/src-tauri/src/browser_permissions_tests.rs index fb43f6bc7d..3aa7f30a02 100644 --- a/src-tauri/src/browser_permissions_tests.rs +++ b/src-tauri/src/browser_permissions_tests.rs @@ -191,6 +191,9 @@ fn native_command_permissions_isolate_webviews() { "app_agent_search_profiles", "app_agent_upload", "app_agent_media", + "app_agent_owner_media_stage", + "app_agent_owner_media_publish", + "app_agent_owner_media_discard", "app_agent_git_auth", "app_agent_remember", "app_agent_claim", diff --git a/src-tauri/src/federated_identity/agent/mod.rs b/src-tauri/src/federated_identity/agent/mod.rs index b30f932d0d..7fbfd980f0 100644 --- a/src-tauri/src/federated_identity/agent/mod.rs +++ b/src-tauri/src/federated_identity/agent/mod.rs @@ -436,6 +436,51 @@ impl AgentTokens { && !row.closed && row.generation == incarnation.row } + /// Runs `write` only while a local caller's work stamped `incarnation` at + /// `retirements` still acts, holding the locks [`Admission::commit`] holds in + /// the same order, so deletion, sign-out and a change of account either + /// precede it or wait for it. `write` must not reenter the credential owners. + pub(crate) fn commit_caller( + &self, + agent: &str, + relay: &str, + incarnation: Stamp, + retirements: u64, + write: impl FnOnce() -> Result, + ) -> Result { + let row = row_key(agent, relay)?; + let retired = lock(&self.0.retired); + let incarnations = lock(&self.0.incarnations); + let (at, row) = ( + incarnations.get(agent).copied().unwrap_or_default(), + incarnations.get(&row).copied().unwrap_or_default(), + ); + let state = self.0.identity.state(); + if at.closed + || at.generation != incarnation.key + || row.closed + || row.generation != incarnation.row + { + return Err(DELETED.into()); + } + // An organization sign-out marks its retirement pending before it counts it. + if retired.pending == Some(true) || state.closed || state.retirements != retirements { + return Err(CHANGED.into()); + } + write() + } + /// As an organization sign-out does before it counts its retirement. + #[cfg(test)] + pub(crate) fn mark_retirement_pending(&self) { + lock(&self.0.retired).pending = Some(true); + } + /// Signals once the next sign-out finds the retirement lock taken. + #[cfg(test)] + pub(crate) fn on_sign_out_contended(&self) -> std::sync::mpsc::Receiver<()> { + let (contended, waiting) = std::sync::mpsc::channel(); + *lock(&self.0.contended) = Some(contended); + waiting + } /// The owner's retirements so far: sign-out, another account, or a /// retired grant each add one. pub(crate) fn retirements(&self) -> u64 { diff --git a/src-tauri/src/identity.rs b/src-tauri/src/identity.rs index d524b34ed1..738aff99d8 100644 --- a/src-tauri/src/identity.rs +++ b/src-tauri/src/identity.rs @@ -781,7 +781,12 @@ pub struct IdentityHost(Arc>); impl IdentityHost { #[cfg(test)] pub(crate) fn fixture() -> Self { - Self::fixture_in(State::Ready(Key::new(Zeroizing::new([1; 32])))) + Self::fixture_signed_in_as(1) + } + /// Signed in with the key of 32 `byte`s. + #[cfg(test)] + pub(crate) fn fixture_signed_in_as(byte: u8) -> Self { + Self::fixture_in(State::Ready(Key::new(Zeroizing::new([byte; 32])))) } /// No saved key, or a key that cannot be read (a signing-out host refuses every read). #[cfg(test)] @@ -1127,7 +1132,51 @@ impl IdentityHost { .await } } +/// The identity lock, held while `owner` is the signed-in identity, for an +/// agent's owner media to be written; signing out waits for it. It holds no key. +pub(crate) struct OwnerHeld<'a>(#[allow(dead_code)] std::sync::MutexGuard<'a, Identity>); + +const NOT_OWNER: &str = "Your owner is not signed in to Buzz here now"; + impl IdentityHost { + /// Signs a Blossom media read (kind 24242, `t` `get` only) for an agent's + /// owner media, only while `owner` is the signed-in identity. + pub(crate) async fn sign_owner_media_get( + &self, + owner: String, + event: EventTemplate, + ) -> Result { + let verbs: Vec<_> = event + .tags + .iter() + .filter(|tag| tag.first().is_some_and(|name| name == "t")) + .collect(); + if event.kind != 24242 || verbs.is_empty() || verbs.iter().any(|tag| tag[1..] != ["get"]) { + return Err("Only a media read is signed for owner media".into()); + } + with_identity(self.clone(), move |identity| { + if identity.restore()?.as_deref() != Some(owner.as_str()) { + return Err(NOT_OWNER.into()); + } + match &identity.state { + State::Ready(key) => key.sign(event), + _ => Err(NOT_OWNER.into()), + } + }) + .await + } + /// Blocks on the identity lock: call off the async runtime. Refuses unless + /// `owner` is signed in. + pub(crate) fn hold_owner(&self, owner: &str) -> Result> { + let mut identity = self + .0 + .lock() + .map_err(|_| "Identity is unavailable; restart the app")?; + if identity.restore()?.as_deref() != Some(owner) { + return Err(NOT_OWNER.into()); + } + Ok(OwnerHeld(identity)) + } /// Close signing for the rest of this process. Taken under the same lock /// every operation holds while it runs, so jobs already queued refuse too. pub(crate) fn close(&self) { diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index e25db7c83f..718e116cdc 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -803,6 +803,9 @@ fn commands() -> impl Fn(tauri::ipc::Invoke) -> bool + Sen app_agents::app_agent_search_profiles, app_agents::app_agent_upload, app_agents::app_agent_media, + app_agents::owner_media::app_agent_owner_media_stage, + app_agents::owner_media::app_agent_owner_media_publish, + app_agents::owner_media::app_agent_owner_media_discard, app_agents::app_agent_git_auth, app_agents::app_agent_remember, app_agents::app_agent_claim, @@ -1195,7 +1198,7 @@ pub fn run() { }) .on_window_event(|window, event| { if window.label() == "main" && matches!(event, tauri::WindowEvent::Destroyed | tauri::WindowEvent::CloseRequested { .. }) { window.state::().cancel_all(); } - if window.label() == "main" && matches!(event, tauri::WindowEvent::Destroyed) { window.state::().cancel_all(&window.state::()); } + if window.label() == "main" && matches!(event, tauri::WindowEvent::Destroyed) { window.state::().cancel_all(&window.state::()); window.state::().owner_media_clear(); } #[cfg(target_os = "macos")] if window.label() == "main" { if let tauri::WindowEvent::CloseRequested { api, .. } = event { diff --git a/src-tauri/src/relay.rs b/src-tauri/src/relay.rs index 5a451d1113..4edb0010d1 100644 --- a/src-tauri/src/relay.rs +++ b/src-tauri/src/relay.rs @@ -1137,8 +1137,19 @@ async fn blossom_auth( url: &Url, verb: &str, content: &str, - mut tags: Vec>, + tags: Vec>, ) -> Result { + blossom_header( + host.sign(blossom_template(url, verb, content, tags)?) + .await?, + ) +} +fn blossom_template( + url: &Url, + verb: &str, + content: &str, + mut tags: Vec>, +) -> Result { let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map_err(|_| "System clock is unavailable")? @@ -1151,14 +1162,14 @@ async fn blossom_auth( // The relay's strict NIP-FI window is 60 seconds. vec!["expiration".into(), (now + 60).to_string()], ]); - let event = host - .sign(EventTemplate { - kind: 24242, - created_at: now, - content: content.into(), - tags, - }) - .await?; + Ok(EventTemplate { + kind: 24242, + created_at: now, + content: content.into(), + tags, + }) +} +fn blossom_header(event: serde_json::Value) -> Result { Ok(format!( "Nostr {}", STANDARD.encode( @@ -2209,6 +2220,16 @@ async fn send_media( let auth = blossom_auth(host, &url, "get", "Get buzz-media", Vec::new()) .await .map_err(|_| 401u16)?; + send_signed(host, fi, assertion, url, range, auth).await +} +async fn send_signed( + host: &IdentityHost, + fi: Option<&FederatedIdentity>, + assertion: Option<&Authorization>, + url: Url, + range: Option<&str>, + auth: String, +) -> std::result::Result { let mut request = client() .map_err(|_| 502u16)? .get(url) @@ -2238,6 +2259,87 @@ async fn send_media( Ok(upstream) } +/// One of an agent's community media blobs, read into memory with its owner's +/// own access: the request is signed only while `owner` is signed in. `reserve` +/// takes the bytes the body may use, its `Content-Length` or else `limit`, +/// before any is read; a body past that is refused. Returns its `Content-Type`. +pub(crate) async fn owner_media( + host: &IdentityHost, + fi: Option<&FederatedIdentity>, + owner: &str, + url: Url, + limit: u64, + reserve: impl FnOnce(u64) -> Result<()>, +) -> Result<(Option, Vec)> { + // Signed first, so nothing, not even discovery, is sent for another owner; + // then signed again once the community credential is known, so a change of + // account while it was found cannot pair it with this owner's signature. + let template = || blossom_template(&url, "get", "Get buzz-media", Vec::new()); + host.sign_owner_media_get(owner.into(), template()?).await?; + let assertion = evidence(host, fi, &url).await?; + let auth = blossom_header(host.sign_owner_media_get(owner.into(), template()?).await?)?; + let operation = async { + let mut upstream = send_signed(host, fi, assertion.as_ref(), url, None, auth) + .await + .map_err(|status| match status { + 401 | 403 => "Your owner may not read this media".to_owned(), + 404 => "The community has no such media".to_owned(), + 502 => "The media request failed".to_owned(), + status => format!("The community refused the media: status {status}"), + })?; + if upstream.status().as_u16() != 200 { + return Err("The community sent part of the media".into()); + } + let declared = upstream.content_length(); + let reserved = declared.unwrap_or(limit); + let too_large = || format!("Media over {} MiB cannot be saved", limit / 1024 / 1024); + if reserved > limit { + return Err(too_large()); + } + reserve(reserved)?; + let kind = upstream + .headers() + .get("content-type") + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let mut body = Vec::new(); + let no_memory = |_| "Not enough memory for the media"; + if declared.is_some() { + // One buffer of the declared size, so the body is never copied to grow it. + body.try_reserve_exact(reserved as usize) + .map_err(no_memory)?; + } + while let Some(chunk) = upstream + .chunk() + .await + .map_err(|_| "The media request failed")? + { + let needed = body.len() + chunk.len(); + if needed as u64 > reserved { + return Err(match declared { + Some(_) => "The media was larger than the community said".into(), + None => too_large(), + }); + } + if needed > body.capacity() { + // Grown as `Vec` would, but never past what was reserved. + let grown = needed.max(body.capacity() * 2).min(reserved as usize); + body.try_reserve_exact(grown - body.len()) + .map_err(no_memory)?; + } + body.extend_from_slice(&chunk); + } + Ok((kind, body)) + }; + match (fi, assertion.as_ref()) { + (Some(fi), Some(assertion)) => { + fi.during_admitted_operation(host, assertion, false, operation) + .await? + } + _ => operation.await, + } +} + /// The response type of an upstream media response, as `media_type` decides it. fn upstream_type(upstream: &reqwest::Response) -> (String, bool) { media_type( diff --git a/src/bundled/claude-code/execution.test.ts b/src/bundled/claude-code/execution.test.ts index f1e7067330..0c2bae0046 100644 --- a/src/bundled/claude-code/execution.test.ts +++ b/src/bundled/claude-code/execution.test.ts @@ -102,7 +102,7 @@ createInterface({input:process.stdin}).on('line', (line) => { writeFileSync(session+'.saved','saved'); send({type:'system',subtype:'init',session_id:'session-1',model:'haiku'}); if (wire.message.content === 'hold' || wire.message.content.includes('hold-request') || (wire.message.content === '/compact' && existsSync('hold-compact'))) return; - send({type:'control_request',request_id:'tool',request:{subtype:'mcp_message',server_name:'buzz',message:{jsonrpc:'2.0',id:1,method:'tools/call',params:{name:wire.message.content.includes('read-path-request')?'send':'read',arguments:{path:'proof.bin',audience:'everyone'}}}}}); + send({type:'control_request',request_id:'tool',request:{subtype:'mcp_message',server_name:'buzz',message:{jsonrpc:'2.0',id:1,method:'tools/call',params:wire.message.content.includes('owner-media-request')?{name:'owner_media_get',arguments:{url:'a.pdf'}}:{name:wire.message.content.includes('read-path-request')?'send':'read',arguments:{path:'proof.bin',audience:'everyone'}}}}}); } else if (wire.type === 'control_response') { if (wire.response.request_id === 'tool') { process.stderr.write(JSON.stringify(wire.response.response)); @@ -387,6 +387,31 @@ it("resumes a saved target before its first post-restart compaction and gives Ne } }); +it("neither lists nor runs owner media for a remote Claude, whose files are elsewhere", async () => { + const f = await remoteFixture(); + const media = vi.fn(); + Object.assign(f.handle, { + ownerAccess: { + channels: vi.fn(), + channel: vi.fn(), + members: vi.fn(), + messages: vi.fn(), + canvas: vi.fn(), + canvasHistory: vi.fn(), + held: vi.fn(), + media, + }, + }); + try { + await f.deliver("owner-media-request"); + await vi.waitFor(() => expect(f.output.join("")).toContain("mcp_response")); + expect(f.output.join("")).toContain("owner_media_get is not available"); + expect(media).not.toHaveBeenCalled(); + } finally { + await f.cleanup(); + } +}); + it("interrupts remote work with a stale binding cache and no relay scope", async () => { const f = await remoteFixture(); const pending = f.deliver("hold-request"); diff --git a/src/bundled/claude-code/runtime.ts b/src/bundled/claude-code/runtime.ts index 945371b2a7..7d31c80f9a 100644 --- a/src/bundled/claude-code/runtime.ts +++ b/src/bundled/claude-code/runtime.ts @@ -8,7 +8,8 @@ import { executionSpawn } from "./execution"; // so one agent works in several conversations at once. Claude acts in Buzz with // the in-process Buzz tools, signed by the agent's native-held key; the runtime // only reports a turn that fails. -import type { Context } from "../../buzz-mcp/client"; +import type { Context, OwnerAccess } from "../../buzz-mcp/client"; +import { withoutOwnerMedia } from "../../buzz-mcp/owner"; import { RESUMED_PROMPT } from "../../buzz-mcp/prompt"; import { type Listing, respond } from "../../buzz-mcp/rpc"; import { @@ -162,7 +163,10 @@ const where = ({ workspace, model, effort }: Config) => ({ /** How Claude lists the session tools: in full up front only those marked * `alwaysLoad`, and the rest by name until its tool search loads them, with * the server's instructions saying what they are. */ -const listing = (added: readonly AddedTool[], owner = false): Listing => ({ +const listing = ( + added: readonly AddedTool[], + owner?: Pick, +): Listing => ({ instructions: sessionHint(added, owner), tool: ({ load, ...tool }) => load === "always" @@ -1379,7 +1383,8 @@ export class ClaudeRuntime { const handle = entry.handle; const client = handle && - appClient(handle, { + // A remote target's files are not here, so neither is owner media. + appClient(this.target ? withoutOwnerMedia(handle) : handle, { // Its PreToolUse hook reviews every tool, Buzz's included. nativeToolGate: true, protectionContext: async () => { @@ -1412,7 +1417,7 @@ export class ClaudeRuntime { context, message, this.added(), - listing(this.added(), !!client?.ownerAccess), + listing(this.added(), client?.ownerAccess), ); }; const saved = this.savedSessions(agent); diff --git a/src/bundled/codex/runtime.test.ts b/src/bundled/codex/runtime.test.ts index a75b0a06b9..b09e85e85a 100644 --- a/src/bundled/codex/runtime.test.ts +++ b/src/bundled/codex/runtime.test.ts @@ -6389,6 +6389,102 @@ it("runs bound Codex tools in the guest workspace, resumes, and stops on provide } }); +it.each([false, true])( + "lists and runs owner_media_get only where its files are, not on a remote target: remote %s", + async (remote) => { + let f!: ReturnType; + f = remote + ? fixture([], undefined, false, {}, undefined, undefined, async () => + guestTarget(f), + ) + : fixture(); + const executedAs = { agent: "agent", owner: "owner", community: "c" }; + const media = vi.fn(async () => ({ + execution: executedAs, + path: "/tmp/buzz-agent-media/a.pdf", + type: "application/pdf", + size: 3, + release: vi.fn(async () => {}), + discard: vi.fn(), + })); + const ownerAccess = { + channels: vi.fn(), + channel: vi.fn(), + members: vi.fn(), + messages: vi.fn(), + canvas: vi.fn(), + canvasHistory: vi.fn(), + held: vi.fn(), + media, + }; + const delivery = f.delivery("@Codex get that file"); + await f.runtime.run({ + ...delivery, + agent: { ...delivery.agent, ownerAccess } as typeof delivery.agent, + }); + await vi.waitFor(() => expect(f.starts()).toHaveLength(1)); + const thread = f.sent.find((wire) => wire.method === "thread/start") + ?.params as { + dynamicTools: { description: string; tools: { name: string }[] }[]; + }; + const names = thread.dynamicTools[0]?.tools.map((tool) => tool.name); + expect(names).toContain("owner_read"); + expect(names?.includes("owner_media_get")).toBe(!remote); + expect( + thread.dynamicTools[0]?.description.includes("owner_media_get"), + ).toBe(!remote); + // A remembered or guessed call is refused there before anything is read. + const reply = (await f.tool("owner_media_get", { url: "a.pdf" })) + ?.result as { success: boolean; contentItems: { text: string }[] }; + expect(reply.success).toBe(!remote); + expect(reply.contentItems[0]?.text).toContain( + remote ? "nothing was read" : '"executedAs"', + ); + expect(media).toHaveBeenCalledTimes(remote ? 0 : 1); + await f.complete(); + }, +); + +it("starts fresh instead of restoring an owner binding saved before owner_media_get was listed", async () => { + const f = fixture(); + f.storage.set( + `buzz.codex.sessions.v17:${scope}:${pubkey}`, + JSON.stringify({ + [JSON.stringify(["channel", root])]: { + threadId: "owner-thread", + workspace: "/tmp/codex-test", + tools: ":owner", + }, + }), + ); + const media = vi.fn(); + const ownerAccess = { + channels: vi.fn(), + channel: vi.fn(), + members: vi.fn(), + messages: vi.fn(), + canvas: vi.fn(), + canvasHistory: vi.fn(), + held: vi.fn(), + media, + }; + const delivery = f.delivery("new work"); + await f.runtime.run({ + ...delivery, + agent: { ...delivery.agent, ownerAccess } as typeof delivery.agent, + }); + await vi.waitFor(() => expect(f.starts()).toHaveLength(1)); + expect(f.sent.filter((wire) => wire.method === "thread/resume")).toHaveLength( + 0, + ); + const thread = f.sent.find((wire) => wire.method === "thread/start") + ?.params as { dynamicTools: { tools: { name: string }[] }[] }; + expect(thread.dynamicTools[0]?.tools.map((tool) => tool.name)).toContain( + "owner_media_get", + ); + await f.complete(); +}); + function guestTarget(guest: ReturnType): AgentExecution { const lifetime = new AbortController(); return { diff --git a/src/bundled/codex/runtime.ts b/src/bundled/codex/runtime.ts index 8c8f52e06d..2e2a9811d0 100644 --- a/src/bundled/codex/runtime.ts +++ b/src/bundled/codex/runtime.ts @@ -49,7 +49,8 @@ import type { SessionControlAction, SessionState, } from "../../features/agents/service"; -import type { Context } from "../../buzz-mcp/client"; +import type { Context, OwnerAccess } from "../../buzz-mcp/client"; +import { isOwnerTool, withoutOwnerMedia } from "../../buzz-mcp/owner"; import { type AddedTool, addedKey, @@ -488,6 +489,11 @@ export class CodexRuntime { private readonly controls = new SessionControls(() => { for (const listener of this.listeners) listener(); }); + /** The agent's owner access as its threads get it: a remote target's + * files are not here, so neither is owner media. */ + private ownerAccess(agent: Pick) { + return (this.target ? withoutOwnerMedia(agent) : agent).ownerAccess; + } private controlKey(agent: Agent, place: SessionPlace) { return JSON.stringify([ place.channelId, @@ -1916,7 +1922,7 @@ export class CodexRuntime { { ...params, ephemeral: true, - dynamicTools: dynamicTools(this.added(), !!agent.ownerAccess), + dynamicTools: dynamicTools(this.added(), this.ownerAccess(agent)), }, ); once.threadId = started.thread.id; @@ -2166,7 +2172,10 @@ export class CodexRuntime { if ( !turn || call.namespace !== "buzz" || - !sessionTools(this.added(), true).some((tool) => tool.name === call.tool) + !( + isOwnerTool(call.tool) || + sessionTools(this.added()).some((tool) => tool.name === call.tool) + ) ) throw new Error("No active Buzz tool for this thread and turn"); const check = () => { @@ -2240,7 +2249,7 @@ export class CodexRuntime { const final = isFinalMessage(call.tool, args); const reply = await callSessionTool( this.target - ? { + ? withoutOwnerMedia({ ...client, media: async () => { this.target?.signal.throwIfAborted(); @@ -2248,7 +2257,7 @@ export class CodexRuntime { "Media download is unavailable for remote Codex agents.", ); }, - } + }) : client, context, call.tool, @@ -2569,8 +2578,10 @@ export class CodexRuntime { // Resume restores the thread's tools, so one started with other added // tools starts fresh. const added = this.added(); - const owner = !!agent.ownerAccess; - const tools = addedKey(added) + (owner ? ":owner" : ""); + const owner = this.ownerAccess(agent); + const tools = + addedKey(added) + + (owner ? (owner.media ? ":owner+media" : ":owner") : ""); const resuming = existing?.workspace === workspace && (existing.tools ?? "") === tools; type Started = { thread: { id: string; cwd: string } }; @@ -2990,7 +3001,10 @@ function follow( * keeps a deferred tool from the model until its tool search finds it. Every * model in Codex's catalog has tool search (supports_search_tool); one * without it would not see deferred tools. */ -const dynamicTools = (added: readonly AddedTool[], owner: boolean) => [ +const dynamicTools = ( + added: readonly AddedTool[], + owner: Pick | undefined, +) => [ { type: "namespace", name: "buzz", diff --git a/src/bundled/pi/runtime.ts b/src/bundled/pi/runtime.ts index e5c2ac347d..d5328909b5 100644 --- a/src/bundled/pi/runtime.ts +++ b/src/bundled/pi/runtime.ts @@ -1142,8 +1142,8 @@ export class PiRuntime { agent, // Whichever host runs the agent fills in its tools' credential. ...(handle && { as: { pubkey: handle.pubkey, relay: handle.relay } }), - tools: sessionTools(this.added(), !!handle?.ownerAccess), - hint: sessionHint(this.added(), !!handle?.ownerAccess), + tools: sessionTools(this.added(), handle?.ownerAccess), + hint: sessionHint(this.added(), handle?.ownerAccess), }); } diff --git a/src/buzz-mcp/AGENTS.md b/src/buzz-mcp/AGENTS.md index add3e298e6..6ba22ad641 100644 --- a/src/buzz-mcp/AGENTS.md +++ b/src/buzz-mcp/AGENTS.md @@ -20,10 +20,13 @@ deliberately. `settings_set`). The settings themselves are the host's catalog (`src/app/device-settings.ts`); the tools never name one. - `owner.ts`: the owner tools (`owner_channel_list`, `owner_channel_get`, - `owner_member_list`, `owner_read`, `owner_canvas_get`, `owner_canvas_history`), - which read with the owner's access where the host gives + `owner_member_list`, `owner_read`, `owner_canvas_get`, `owner_canvas_history`, + `owner_media_get`), which read with the owner's access where the host gives the client `ownerAccess`; `src/features/agents/owner-access.ts` owns the - reads and their binding. Plugins' clients never have it. + reads and their binding. Plugins' clients never have it. `owner_media_get` + is listed only where `ownerAccess.media` exists: native + (`src-tauri/src/app_agents/owner_media.rs`) holds the bytes until the + reviewed answer's release saves them; remote-target runtimes strip it. - `session-tools.ts`: what a session lists, each tool's `load` (`always` or `search`), and the server instructions that cover the whole tool set. - `prompt.md`: the posting conventions every agent gets after the shared base diff --git a/src/buzz-mcp/app-client.ts b/src/buzz-mcp/app-client.ts index e400bd6938..c0a082afb5 100644 --- a/src/buzz-mcp/app-client.ts +++ b/src/buzz-mcp/app-client.ts @@ -48,6 +48,7 @@ export function appClient( const handoff = agent.handoff; const plugins = agent.plugins; const { settings, ownerAccess } = agent; + const ownerMedia = ownerAccess?.media; /** An owner read, only while the turn still holds before and after it. */ const owned = async (read: (signal?: AbortSignal) => Promise) => { check(); @@ -106,6 +107,29 @@ export function appClient( check(); ownerAccess.held(execution); }, + ...(ownerMedia && { + media: async (url) => { + check(); + const media = await ownerMedia(url, options.signal); + /** Checks the turn, dropping the media if it ended. */ + const live = () => { + try { + check(); + } catch (error) { + media.discard(); + throw error; + } + }; + live(); + return { + ...media, + release: () => { + live(); + return media.release(); + }, + }; + }, + }), }, }), relayAuthor: options.relayAuthor, diff --git a/src/buzz-mcp/client.ts b/src/buzz-mcp/client.ts index 1e892a9569..22a277bd96 100644 --- a/src/buzz-mcp/client.ts +++ b/src/buzz-mcp/client.ts @@ -222,7 +222,20 @@ export type OwnerAccess = Readonly<{ /** Throws unless the binding that answered with `execution` still holds, * as checked just before its answer is released; reads nothing. */ held(execution: OwnerExecution): void; + /** One of the community's media blobs, by its URL or `sha256[.ext]`, read + * with the owner's access into memory; nothing is at its `path` until + * `release`. Only where the host can hold media for review. */ + media?(url: string, signal?: AbortSignal): Promise; }>; +/** Owner media held for review. `release` saves it to `path` while the + * binding it was read with still holds, checked and started synchronously; + * `discard` drops it. Either settles it. */ +export type OwnerMedia = Media & + Readonly<{ + execution: OwnerExecution; + release(): Promise; + discard(): void; + }>; export interface BuzzClient { protect?(action: ProtectionAction): Promise; diff --git a/src/buzz-mcp/owner.test.ts b/src/buzz-mcp/owner.test.ts index 74383cebf0..4126cf1965 100644 --- a/src/buzz-mcp/owner.test.ts +++ b/src/buzz-mcp/owner.test.ts @@ -5,9 +5,15 @@ import type { BuzzEvent, OwnerAccess, OwnerChannel, + OwnerMedia, } from "./client"; import { CANVAS_HISTORY_PAGE } from "../features/channel-templates/canvas-limits"; -import { OWNER_TOOLS, callOwnerTool, isOwnerTool } from "./owner"; +import { + OWNER_TOOLS, + callOwnerTool, + isOwnerTool, + withoutOwnerMedia, +} from "./owner"; import { respond } from "./rpc"; import { type AddedTool, @@ -86,6 +92,25 @@ function owned() { held: vi.fn(), }; } +const PNG = "iVBORw0KGgoAAAANSUhEUg=="; +const release = vi.fn(async () => {}); +const discard = vi.fn(); +const media = (patch: Partial = {}): OwnerMedia => ({ + execution: executedAs, + path: "/tmp/a.png", + type: "image/png", + size: 16, + data: PNG, + release, + discard, + ...patch, +}); +/** Owner access that also holds media for review. */ +function withMedia() { + release.mockClear(); + discard.mockClear(); + return { ...owned(), media: vi.fn(async () => media()) }; +} const clientWith = (fields: Partial = {}) => ({ pubkey: executedAs.agent, @@ -104,7 +129,7 @@ describe("callOwnerTool", () => { q: "PLAN", limit: 1, }) - ).text, + ).reply as string, ), ).toEqual({ executedAs, @@ -121,7 +146,10 @@ describe("callOwnerTool", () => { }); expect( JSON.parse( - String((await callOwnerTool(client, "owner_channel_list", {})).text), + String( + (await callOwnerTool(client, "owner_channel_list", {})) + .reply as string, + ), ), ).toEqual({ executedAs, @@ -138,7 +166,7 @@ describe("callOwnerTool", () => { expect( JSON.parse( (await callOwnerTool(client, "owner_channel_get", { channel: id })) - .text, + .reply as string, ), ).toMatchObject({ executedAs, channel: { name: "Plans" } }); expect(ownerAccess.channel).toHaveBeenCalledWith(id); @@ -150,7 +178,7 @@ describe("callOwnerTool", () => { offset: 1, limit: 1, }) - ).text, + ).reply as string, ), ).toEqual({ executedAs, @@ -222,7 +250,9 @@ describe("callOwnerTool", () => { const ownerAccess = owned(); const client = clientWith({ ownerAccess }); const read = async (args: Record) => - JSON.parse((await callOwnerTool(client, "owner_read", args)).text); + JSON.parse( + (await callOwnerTool(client, "owner_read", args)).reply as string, + ); ownerAccess.messages.mockResolvedValueOnce({ execution: executedAs, channel: id, @@ -385,7 +415,7 @@ describe("callOwnerTool", () => { const ownerAccess = owned(); const client = clientWith({ ownerAccess }); const call = async (name: string, args: Record) => - JSON.parse((await callOwnerTool(client, name, args)).text); + JSON.parse((await callOwnerTool(client, name, args)).reply as string); expect(await call("owner_canvas_get", { channel: id })).toEqual({ executedAs, channel: id, @@ -455,6 +485,7 @@ describe("callOwnerTool", () => { "owner_read", "owner_canvas_get", "owner_canvas_history", + "owner_media_get", ]); expect( OWNER_TOOLS.every( @@ -475,11 +506,11 @@ describe("owner tools in the session", () => { it("lists them only to a client with owner access, on search", async () => { expect(names(sessionTools()).some(isOwnerTool)).toBe(false); expect(sessionHint()).not.toContain("owner_channel_list"); - const tools = sessionTools([], true); + const tools = sessionTools([], withMedia()); expect( tools.filter((tool) => isOwnerTool(tool.name)).map((tool) => tool.load), ).toEqual(OWNER_TOOLS.map(() => "search")); - expect(sessionHint([], true)).toContain("owner_channel_list"); + expect(sessionHint([], withMedia())).toContain("owner_channel_list"); const list = async (client?: BuzzClient) => ( (await respond(client, {}, { id: 1, method: "tools/list" })) as { @@ -488,11 +519,90 @@ describe("owner tools in the session", () => { ).result.tools; expect(names(await list()).some(isOwnerTool)).toBe(false); expect(names(await list(clientWith())).some(isOwnerTool)).toBe(false); - expect(names(await list(clientWith({ ownerAccess: owned() })))).toEqual( + expect(names(await list(clientWith({ ownerAccess: withMedia() })))).toEqual( expect.arrayContaining(names(OWNER_TOOLS)), ); }); + it("lists and runs owner_media_get only where owner media is held for review", async () => { + const listed = async (client: BuzzClient) => + names( + ( + (await respond(client, {}, { id: 1, method: "tools/list" })) as { + result: { tools: { name: string }[] }; + } + ).result.tools, + ); + const plain = clientWith({ ownerAccess: owned() }); + const media = clientWith({ ownerAccess: withMedia() }); + // Without it (another host, or a remote target's runtime) it is neither + // listed, hinted nor run, directly or from a remembered list. + for (const client of [plain, withoutOwnerMedia(media)]) { + expect(await listed(client)).toContain("owner_read"); + expect(await listed(client)).not.toContain("owner_media_get"); + expect(sessionHint([], client.ownerAccess)).not.toContain( + "owner_media_get", + ); + await expect( + callSessionTool(client, {}, "owner_media_get", { url: "a.png" }), + ).rejects.toThrow("nothing was read"); + } + expect(media.ownerAccess?.media).not.toHaveBeenCalled(); + expect(await listed(media)).toContain("owner_media_get"); + expect(sessionHint([], media.ownerAccess)).toContain("owner_media_get"); + expect(withoutOwnerMedia(plain)).toBe(plain); + }); + + it("answers owner media as media_get does, saving it only once released", async () => { + const ownerAccess = withMedia(); + const client = clientWith({ ownerAccess }); + const answer = await callOwnerTool(client, "owner_media_get", { + url: " a.png ", + }); + expect(ownerAccess.media).toHaveBeenCalledWith("a.png"); + expect(answer.reply).toEqual({ + data: PNG, + type: "image/png", + text: JSON.stringify({ + executedAs, + path: "/tmp/a.png", + type: "image/png", + size: 16, + }), + }); + expect(release).not.toHaveBeenCalled(); + await answer.held(); + expect(release).toHaveBeenCalledOnce(); + answer.discard?.(); + expect(discard).toHaveBeenCalledOnce(); + // Not an image: its path only. + const { data: _, ...pdf } = media({ type: "application/pdf" }); + ownerAccess.media.mockResolvedValueOnce(pdf); + expect( + JSON.parse( + (await callOwnerTool(client, "owner_media_get", { url: "b.pdf" })) + .reply as string, + ), + ).toMatchObject({ type: "application/pdf" }); + await expect(callOwnerTool(client, "owner_media_get", {})).rejects.toThrow( + "url is required", + ); + await expect( + callOwnerTool(client, "owner_media_get", { url: "a.png", agent: "x" }), + ).rejects.toThrow("agent cannot be used"); + }); + + it("drops owner media whose release fails", async () => { + const client = clientWith({ ownerAccess: withMedia() }); + release.mockImplementationOnce(() => { + throw new Error("changed during the call"); + }); + await expect( + callSessionTool(client, {}, "owner_media_get", { url: "a.png" }), + ).rejects.toThrow("changed during the call"); + expect(discard).toHaveBeenCalled(); + }); + it("refuses a direct or remembered call from a client without owner access", async () => { await expect( callSessionTool(clientWith(), {}, "owner_channel_list", {}), diff --git a/src/buzz-mcp/owner.ts b/src/buzz-mcp/owner.ts index 7035c3caa0..07df396f7c 100644 --- a/src/buzz-mcp/owner.ts +++ b/src/buzz-mcp/owner.ts @@ -15,6 +15,8 @@ import { HEX, messageLine, pubkey, + type Reply, + shownImage, type Tool, time, unixTime, @@ -128,18 +130,58 @@ export const OWNER_TOOLS: readonly Tool[] = [ { channel, before, before_id: beforeId }, ["channel"], ), + tool( + "owner_media_get", + "Get a file from a Buzz media URL in this community with your owner's access, as media_get does with yours: saves it locally and gives its path; an image is also shown.", + { + url: { + type: "string", + description: "A media URL in this community, or its sha256[.ext].", + }, + }, + ["url"], + ), ]; export const isOwnerTool = (name: string) => OWNER_TOOLS.some((tool) => tool.name === name); +/** The owner tools a client with `ownerAccess` lists: owner_media_get only + * where it can hold media for review. */ +export const ownerTools = ( + ownerAccess: Pick | undefined, +): readonly Tool[] => + ownerAccess + ? OWNER_TOOLS.filter( + (tool) => tool.name !== "owner_media_get" || !!ownerAccess.media, + ) + : []; + +/** `holder` without owner media: for an agent whose files are on a remote + * target, where a path saved here means nothing. */ +export function withoutOwnerMedia< + T extends Readonly<{ ownerAccess?: OwnerAccess | undefined }>, +>(holder: T): T { + if (!holder.ownerAccess?.media) return holder; + const { media: _, ...ownerAccess } = holder.ownerAccess; + return { ...holder, ownerAccess }; +} + +/** An owner tool's reply, and its release: it is sent only once `held` + * settles, and `discard` drops what it holds when it is not. */ +export type OwnerAnswer = Readonly<{ + reply: Reply; + held(): void | Promise; + discard?(): void; +}>; + /** Runs owner tool `name`. Throws, before reading anything, when this * client has no owner access, as a plugin's or another host's has not. */ export async function callOwnerTool( client: BuzzClient, name: string, args: Readonly>, -): Promise> { +): Promise { const tool = OWNER_TOOLS.find((tool) => tool.name === name); if (!tool) throw new Error(`No tool ${name}`); const { ownerAccess } = client; @@ -170,6 +212,27 @@ export async function callOwnerTool( }); } if (name === "owner_read") return read(ownerAccess, args); + if (name === "owner_media_get") { + if (!ownerAccess.media) + throw new Error( + "owner_media_get is not available to this agent here; nothing was read.", + ); + const url = optionalText(args.url, "url"); + if (!url) throw new Error("url is required."); + const media = await ownerAccess.media(url); + const text = JSON.stringify({ + executedAs: media.execution, + path: media.path, + type: media.type, + size: media.size, + }); + const type = shownImage(media); + return { + reply: type ? { data: media.data as string, type, text } : text, + held: () => media.release(), + discard: () => media.discard(), + }; + } const id = optionalText(args.channel, "channel"); if (!id) throw new Error("channel is required."); if (name === "owner_canvas_get") { @@ -346,7 +409,7 @@ function answer( fields: object, ) { return { - text: JSON.stringify({ executedAs: execution, ...fields }), + reply: JSON.stringify({ executedAs: execution, ...fields }), held: () => ownerAccess.held(execution), }; } diff --git a/src/buzz-mcp/rpc.ts b/src/buzz-mcp/rpc.ts index 34b18c5387..8bd6503ccc 100644 --- a/src/buzz-mcp/rpc.ts +++ b/src/buzz-mcp/rpc.ts @@ -55,7 +55,7 @@ export async function respond( return result({ // A connection keeps its client, so its list stays the same. One // whose client can't change settings, such as a plugin's, lacks them. - tools: sessionTools(added, !!client?.ownerAccess) + tools: sessionTools(added, client?.ownerAccess) .filter( ({ name }) => !client || client.settings || !isSettingsTool(name), ) diff --git a/src/buzz-mcp/session-tools.ts b/src/buzz-mcp/session-tools.ts index 0f25cb9512..6fe47d6bff 100644 --- a/src/buzz-mcp/session-tools.ts +++ b/src/buzz-mcp/session-tools.ts @@ -13,8 +13,13 @@ import { callAttentionTool, isAttentionTool, } from "./attention"; -import type { BuzzClient, Context } from "./client"; -import { callOwnerTool, isOwnerTool, OWNER_TOOLS } from "./owner"; +import type { BuzzClient, Context, OwnerAccess } from "./client"; +import { + callOwnerTool, + isOwnerTool, + type OwnerAnswer, + ownerTools, +} from "./owner"; import { callPluginTool, isPluginTool, PLUGIN_TOOLS } from "./plugins"; import { callSettingsTool, isSettingsTool, SETTINGS_TOOLS } from "./settings"; import { @@ -86,31 +91,34 @@ const listed = ({ name, description, inputSchema }: AddedTool): Listed => ({ export type SessionTool = Listed & Readonly<{ load: "always" | "search" }>; /** The tools a session lists: the Buzz tools, then those plugins add. The - * owner tools only with `owner`: a client with `ownerAccess`. */ + * owner tools only with `owner`, a client's `ownerAccess`. */ export const sessionTools = ( added: readonly AddedTool[] = [], - owner = false, + owner?: Pick, ): readonly SessionTool[] => [ ...TOOLS.map((tool) => ({ ...tool, load: "always" as const })), ...ATTENTION_TOOLS.map((tool) => ({ ...tool, load: "search" as const })), ...SOLO_WORKER_TOOLS.map((tool) => ({ ...tool, load: "search" as const })), ...PLUGIN_TOOLS.map((tool) => ({ ...tool, load: "search" as const })), ...SETTINGS_TOOLS.map((tool) => ({ ...tool, load: "search" as const })), - ...(owner - ? OWNER_TOOLS.map((tool) => ({ ...tool, load: "search" as const })) - : []), + ...ownerTools(owner).map((tool) => ({ ...tool, load: "search" as const })), ...added.map((tool) => ({ ...listed(tool), load: "search" as const })), ]; /** What the server (Claude) or namespace (Codex) with the tools says up * front, so the model knows what it can search for. */ -export function sessionHint(added: readonly AddedTool[] = [], owner = false) { +export function sessionHint( + added: readonly AddedTool[] = [], + owner?: Pick, +) { const plugins = added.length ? ` and tools plugins add (${added.map((tool) => tool.name).join(", ")})` : ""; - const owned = owner - ? ", your owner's channels, members, messages and canvases read with their access (owner_channel_list, owner_read)" - : ""; + const owned = !owner + ? "" + : owner.media + ? ", your owner's channels, members, messages, canvases and media read with their access (owner_channel_list, owner_read, owner_media_get)" + : ", your owner's channels, members, messages and canvases read with their access (owner_channel_list, owner_read)"; return `Read and act in Buzz as this agent. Search for the rest when you need them: your attention (Interests, event watches and timers; interest_set and watch_add say how they work), Nexus workers an owner lets you run (solo_worker_connections), this device's plugins (plugin_list; install, update, enable and remove are requests your owner confirms)${owned}${plugins}.`; } @@ -173,17 +181,23 @@ export async function callSessionTool( ...(typeof args.command === "string" ? { command: args.command } : {}), }); let reply: Reply; - let held: (() => void) | undefined; + let owned: OwnerAnswer | undefined; try { if (isOwnerTool(name) && !added.some((tool) => tool.name === name)) - ({ text: reply, held } = await callOwnerTool(client, name, args)); + ({ reply } = owned = await callOwnerTool(client, name, args)); else reply = await dispatchSessionTool(client, context, name, args, added); } catch (error) { return checkedToolError(client, session, error); } - await client.protect?.({ kind: "tool-result", session, content: reply }); - // An owner read is released only while the binding it was read with holds. - held?.(); + try { + await client.protect?.({ kind: "tool-result", session, content: reply }); + // An owner read is released only while the binding it was read with + // holds; owner media is saved only then. + await owned?.held(); + } catch (error) { + owned?.discard?.(); + throw error; + } return reply; } diff --git a/src/buzz-mcp/tools.ts b/src/buzz-mcp/tools.ts index 171cdf5550..cef4184490 100644 --- a/src/buzz-mcp/tools.ts +++ b/src/buzz-mcp/tools.ts @@ -51,6 +51,7 @@ import type { ChannelTemplate, Context, Filter, + Media, Memory, Upload, } from "./client"; @@ -1497,14 +1498,10 @@ const HANDLERS: Readonly> = { async media_get(tools, args) { const media = await tools.client.media(required(args, "url")); const saved = `Saved ${media.path} (${media.type}, ${media.size} bytes).`; - if (media.data) { - // The first 12 bytes name the type. - const head = atob(media.data.slice(0, 16)); - const type = sniff(Uint8Array.from(head, (c) => c.charCodeAt(0))); - if (type?.startsWith("image/")) - return { data: media.data, type, text: saved }; - } - return `${saved} Open it with your own tools.`; + const type = shownImage(media); + return type + ? { data: media.data as string, type, text: saved } + : `${saved} Open it with your own tools.`; }, async tabs(tools, args) { @@ -2296,6 +2293,15 @@ function proposedPicture(client: BuzzClient, path: unknown) { }; } +/** The type of `media`'s shown image, by its own leading bytes (the first + * 12 name it), when it has one. */ +export function shownImage(media: Media) { + if (!media.data) return undefined; + const head = atob(media.data.slice(0, 16)); + const type = sniff(Uint8Array.from(head, (c) => c.charCodeAt(0))); + return type?.startsWith("image/") ? type : undefined; +} + /** An image or MP4's type, by the file's leading bytes. */ function sniff(data: Uint8Array) { const at = (offset: number, ...bytes: number[]) => diff --git a/src/features/agent-harness/kit.ts b/src/features/agent-harness/kit.ts index 346b24384b..86bf5a65a6 100644 --- a/src/features/agent-harness/kit.ts +++ b/src/features/agent-harness/kit.ts @@ -454,7 +454,7 @@ export function harnessKit(sources: Sources): HarnessKit { }); const added = sources.added(); return respond(client, context, message, added, { - instructions: sessionHint(added, !!client.ownerAccess), + instructions: sessionHint(added, client.ownerAccess), }); }; }, diff --git a/src/features/agents/native.ts b/src/features/agents/native.ts index 567d09e2f0..cb5ffe8d47 100644 --- a/src/features/agents/native.ts +++ b/src/features/agents/native.ts @@ -115,6 +115,16 @@ export type AgentMedia = Readonly<{ size: number; data?: string; }>; +/** The agent's owner media, in native memory between its read and its + * publication. `id` names one request; a discard for it may come first. */ +export type OwnerMediaNative = Readonly<{ + /** Reads a media blob of the agent's community with its signed-in owner's + * access; nothing is at the `path` it gives yet. */ + stage(member: Membership, id: string, url: string): Promise; + /** Saves it to its path, only while the owner and agent still act. */ + publish(member: Membership, id: string): Promise; + discard(id: string): Promise; +}>; /** A repository's URL and the `Authorization` header Git sends there. */ export type GitAuth = Readonly<{ url: string; header: string }>; export type AgentsNative = { @@ -205,6 +215,7 @@ export type AgentsNative = { /** A repository's URL in the agent's community and the Git header that * authorizes it there as the agent for a minute. */ gitAuth(member: Membership, owner: string, repo: string): Promise; + ownerMedia?: OwnerMediaNative; /** Writes memory entry `slug`, encrypted to the owner, newer than `after`; * a `null` body deletes it. */ remember( @@ -324,6 +335,13 @@ export const nativeAgents = (): AgentsNative | undefined => invoke("app_agent_media", { pubkey, relay, url }), gitAuth: ({ pubkey, relay }, owner, repo) => invoke("app_agent_git_auth", { pubkey, relay, owner, repo }), + ownerMedia: { + stage: ({ pubkey, relay }, id, url) => + invoke("app_agent_owner_media_stage", { pubkey, relay, id, url }), + publish: ({ pubkey, relay }, id) => + invoke("app_agent_owner_media_publish", { pubkey, relay, id }), + discard: (id) => invoke("app_agent_owner_media_discard", { id }), + }, remember: ({ pubkey, relay }, slug, body, after) => invoke("app_agent_remember", { pubkey, relay, slug, body, after }), publishProfile: ({ pubkey, relay }) => diff --git a/src/features/agents/owner-access.test.ts b/src/features/agents/owner-access.test.ts index 223b1761b5..b2c6d4fd40 100644 --- a/src/features/agents/owner-access.test.ts +++ b/src/features/agents/owner-access.test.ts @@ -23,6 +23,7 @@ const person = "c".repeat(64); const otherOwner = "d".repeat(64); const relayAuthor = "e".repeat(64); const community = "https://relay.example.test"; +const relay = "wss://relay.example.test"; const joined = "11111111-1111-4111-8111-111111111111"; const archived = "22222222-2222-4222-8222-222222222222"; const system = "44444444-4444-4444-8444-444444444444"; @@ -276,7 +277,7 @@ describe("owner access", () => { it("lists the owner's channels, including ones the agent is not in, as the owner", async () => { const fake = fakeSession({ channels }); const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(fake.session).bind, ); await expect(access.channels()).resolves.toEqual({ @@ -314,7 +315,7 @@ describe("owner access", () => { ])("marks a list that is not complete partial: %j", async (state, asks) => { const fake = fakeSession({ channels: [], ...state }); const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(fake.session).bind, ); const list = await access.channels(); @@ -329,7 +330,7 @@ describe("owner access", () => { signed(joined, { [owner]: "admin", [person]: "member", [agent]: "bot" }), ); const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(fake.session).bind, ); const details = await access.channel(joined); @@ -377,7 +378,7 @@ describe("owner access", () => { signed(joined, { [owner]: "owner", [person]: "member" }, metadata), ); const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(fake.session).bind, ); await expect(access.channel(joined)).resolves.toMatchObject({ @@ -405,7 +406,7 @@ describe("owner access", () => { }), ); const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(fake.session).bind, ); await expect(access.channel(joined)).resolves.toMatchObject({ @@ -438,7 +439,7 @@ describe("owner access", () => { signed(joined, { [person]: "owner" }), ); const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(fake.session).bind, ); await expect(access.channel(joined)).resolves.toMatchObject({ @@ -471,7 +472,7 @@ describe("owner access", () => { ])("fails for %s", async (_what, id, events) => { const fake = fakeSession({ channels }, events); const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(fake.session).bind, ); await expect(access.channel(id)).rejects.toThrow(); @@ -487,7 +488,7 @@ describe("owner access", () => { () => signed(untyped, { [owner]: "owner" }, [["t", "system"]]), ); const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(fake.session).bind, ); await expect(access.channel(untyped)).rejects.toThrow( @@ -505,7 +506,7 @@ describe("owner access", () => { signed(joined, { [owner]: "owner" }), ); const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(fake.session).bind, ); await expect( @@ -524,7 +525,7 @@ describe("owner access", () => { }), ); const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(fake.session).bind, ); await expect( @@ -552,7 +553,7 @@ describe("owner access", () => { signed(joined, { [owner]: "owner" }), ); const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(fake.session).bind, ); await access.channels(); @@ -595,7 +596,10 @@ describe("owner access", () => { else if (state === "another account's binding") binding.set({ ...binding.first, viewer: otherOwner }); else binding.controller.abort(); - const access = createOwnerAccess({ pubkey: agent, owner }, binding.bind); + const access = createOwnerAccess( + { pubkey: agent, relay, owner }, + binding.bind, + ); await expect(access.channels()).rejects.toThrow("not connected"); await expect(access.channel(joined)).rejects.toThrow("not connected"); expect(fake.readRemote).not.toHaveBeenCalled(); @@ -617,7 +621,10 @@ describe("owner access", () => { await held; return signed(joined, { [owner]: "owner" }); }); - const access = createOwnerAccess({ pubkey: agent, owner }, binding.bind); + const access = createOwnerAccess( + { pubkey: agent, relay, owner }, + binding.bind, + ); const call = access.channel(joined); await reading.promise; if (change === "signs in as someone else") @@ -640,7 +647,10 @@ describe("owner access", () => { signed(joined, { [owner]: "owner" }), ); const binding = bound(fake.session); - const access = createOwnerAccess({ pubkey: agent, owner }, binding.bind); + const access = createOwnerAccess( + { pubkey: agent, relay, owner }, + binding.bind, + ); const reviewing = started(); const approve = started(); const protect = vi.fn(async (action: ProtectionAction) => { @@ -674,7 +684,7 @@ describe("owner access", () => { it("answers only reads it made itself", () => { const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(fakeSession({ channels }).session).bind, ); expect(() => access.held({ agent, owner, community })).toThrow( @@ -695,7 +705,10 @@ describe("owner access", () => { ); }), ); - const access = createOwnerAccess({ pubkey: agent, owner }, binding.bind); + const access = createOwnerAccess( + { pubkey: agent, relay, owner }, + binding.bind, + ); const call = access.channel(joined); await reading.promise; binding.controller.abort(); @@ -707,7 +720,7 @@ describe("owner access", () => { signed(joined, { [owner]: "owner" }), ); const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(fake.session).bind, ); const before = new AbortController(); @@ -730,9 +743,9 @@ describe("owner access", () => { }); const mine = bound(first.session); const theirs = bound(second.session, otherOwner); - const a = createOwnerAccess({ pubkey: agent, owner }, mine.bind); + const a = createOwnerAccess({ pubkey: agent, relay, owner }, mine.bind); const b = createOwnerAccess( - { pubkey: person, owner: otherOwner }, + { pubkey: person, relay, owner: otherOwner }, theirs.bind, ); await a.channels(); @@ -766,7 +779,7 @@ describe("owner reads", () => { return { fake, binding, - access: createOwnerAccess({ pubkey: agent, owner }, binding.bind), + access: createOwnerAccess({ pubkey: agent, relay, owner }, binding.bind), }; }; const contentReads = (fake: ReturnType) => @@ -914,7 +927,7 @@ describe("owner reads", () => { post("1", joined, 10, "secret"), ]); const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(fake.session).bind, ); await expect( @@ -945,7 +958,7 @@ describe("owner reads", () => { key === "snapshotRead" ? undefined : Reflect.get(object, key), }); const access = createOwnerAccess( - { pubkey: agent, owner }, + { pubkey: agent, relay, owner }, bound(session).bind, ); await expect( @@ -1104,3 +1117,206 @@ describe("owner reads", () => { expect(fake.snapshotRead).toHaveBeenCalled(); }); }); + +describe("owner media", () => { + const png = "iVBORw0KGgoAAAANSUhEUg=="; + const staged = { + path: "/tmp/buzz-agent-media/a.png", + type: "image/png", + size: 16, + data: png, + }; + function media() { + const order: string[] = []; + const native = { + stage: vi.fn(async (_member: unknown, _id: string, _url: string) => { + order.push("stage"); + return staged; + }), + publish: vi.fn(async () => { + order.push("publish"); + }), + discard: vi.fn(async () => { + order.push("discard"); + }), + }; + return { native, order }; + } + const access = ( + binding: ReturnType, + native?: ReturnType["native"], + ) => createOwnerAccess({ pubkey: agent, relay, owner }, binding.bind, native); + const client = ( + ownerAccess: ReturnType, + fields: Partial = {}, + ) => + ({ + pubkey: agent, + key: agent, + ownerAccess, + ...fields, + }) as unknown as BuzzClient; + + it("is offered only where native can hold it", () => { + const binding = bound(fakeSession({ channels }).session); + expect(access(binding).media).toBeUndefined(); + expect(access(binding, media().native).media).toBeTypeOf("function"); + }); + + it("stages as the agent's own membership, then saves it in the same step as its check", async () => { + const { native, order } = media(); + const binding = bound(fakeSession({ channels }).session); + const read = await access(binding, native).media?.("a.png"); + expect(native.stage).toHaveBeenCalledWith( + { pubkey: agent, relay }, + expect.any(String), + "a.png", + ); + expect(read).toMatchObject({ + ...staged, + execution: { agent, owner, community }, + }); + expect(native.publish).not.toHaveBeenCalled(); + const saving = read?.release(); + // Invoked before anything else can run: no await between check and save. + expect(native.publish).toHaveBeenCalledWith( + { pubkey: agent, relay }, + native.stage.mock.calls[0]?.[1], + ); + await saving; + expect(order).toEqual(["stage", "publish"]); + expect(() => read?.release()).toThrow("changed during the call"); + expect(native.publish).toHaveBeenCalledOnce(); + }); + + it.each(["signs in as someone else", "reconnects", "disconnects"])( + "drops staged media, saving nothing, when the owner's connection %s before its release", + async (change) => { + const { native } = media(); + const binding = bound(fakeSession({ channels }).session); + const read = await access(binding, native).media?.("a.png"); + if (change === "signs in as someone else") + binding.set({ ...binding.first, viewer: otherOwner }); + else if (change === "reconnects") + binding.set({ + ...binding.first, + session: fakeSession({ channels }).session, + }); + else binding.set(undefined); + expect(() => read?.release()).toThrow(); + expect(native.publish).not.toHaveBeenCalled(); + expect(native.discard).toHaveBeenCalledOnce(); + }, + ); + + it("drops media whose binding is replaced or whose caller cancels, even before it arrives", async () => { + for (const cause of ["rebind", "cancel"]) { + const { native } = media(); + const binding = bound(fakeSession({ channels }).session); + const arriving = started(); + let arrive = () => {}; + native.stage.mockImplementationOnce( + () => + new Promise((resolve) => { + arriving.resolve(); + arrive = () => resolve(staged); + }), + ); + const caller = new AbortController(); + const read = access(binding, native).media?.("a.png", caller.signal); + await arriving.promise; + if (cause === "rebind") binding.controller.abort(); + else caller.abort(new Error("Turn ended")); + // Told at once, before the read's own reply. + expect(native.discard).toHaveBeenCalledOnce(); + arrive(); + await expect(read).rejects.toThrow(); + expect(native.discard).toHaveBeenCalledOnce(); + expect(native.publish).not.toHaveBeenCalled(); + } + // After staging, a rebind drops it too. + const { native } = media(); + const binding = bound(fakeSession({ channels }).session); + const read = await access(binding, native).media?.("a.png"); + binding.controller.abort(); + expect(native.discard).toHaveBeenCalledOnce(); + expect(() => read?.release()).toThrow("changed during the call"); + expect(native.publish).not.toHaveBeenCalled(); + }); + + it("is saved only after its reviewed result, and dropped when review refuses it", async () => { + for (const verdict of ["allow", "deny"]) { + const { native, order } = media(); + const binding = bound(fakeSession({ channels }).session); + const protect = vi.fn(async (action: ProtectionAction) => { + order.push(action.kind); + if (verdict === "deny" && action.kind === "tool-result") + throw new Error("denied result"); + }); + const call = callSessionTool( + client(access(binding, native), { protect }), + {}, + "owner_media_get", + { url: "a.png" }, + ); + if (verdict === "deny") { + await expect(call).rejects.toThrow("denied result"); + expect(order).toEqual(["tool", "stage", "tool-result", "discard"]); + } else { + await expect(call).resolves.toMatchObject({ + data: png, + type: "image/png", + }); + expect(order).toEqual(["tool", "stage", "tool-result", "publish"]); + // The reviewed reply named the media and who read it. + const content = protect.mock.calls.at(-1)?.[0]?.content as + | { text: string } + | undefined; + expect(JSON.parse(String(content?.text))).toEqual({ + executedAs: { agent, owner, community }, + path: staged.path, + type: staged.type, + size: staged.size, + }); + } + } + }); + + it("drops media when the turn ends during its review", async () => { + const { native } = media(); + const binding = bound(fakeSession({ channels }).session); + let live = true; + const reviewing = started(); + const approve = started(); + const turn = appClient( + { + pubkey: agent, + key: agent, + owner, + ownerAccess: access(binding, native), + protection: { + check: async (action: ProtectionAction) => { + if (action.kind !== "tool-result") return; + reviewing.resolve(); + await approve.promise; + }, + }, + } as unknown as AgentHandle, + { + read: vi.fn(), + memories: async () => ({ entries: [], partial: false }), + protectionContext: async () => ({}), + check: () => { + if (!live) throw new Error("The turn has ended"); + }, + }, + ); + const call = callSessionTool(turn, {}, "owner_media_get", { url: "a.png" }); + await reviewing.promise; + live = false; + approve.resolve(); + await expect(call).rejects.toThrow("The turn has ended"); + expect(native.publish).not.toHaveBeenCalled(); + expect(native.discard).toHaveBeenCalled(); + }); +}); diff --git a/src/features/agents/owner-access.ts b/src/features/agents/owner-access.ts index a5d2304f5c..49a279a0c9 100644 --- a/src/features/agents/owner-access.ts +++ b/src/features/agents/owner-access.ts @@ -8,6 +8,7 @@ import type { OwnerAccess, OwnerChannel, OwnerExecution, + OwnerMedia, OwnerReadRequest, OwnerRole, } from "../../buzz-mcp/client"; @@ -26,6 +27,7 @@ import { } from "../relay/channel-lifecycle-protocol"; import { soloLabels } from "../solo/labels"; import type { RelaySession } from "../relay/session"; +import type { OwnerMediaNative } from "./native"; /** Whether this build lets agents read with their owner's access in * `community`: its exact canonical origin is in the build's list. */ @@ -76,10 +78,12 @@ const changed = () => ); /** `agent`'s owner reads. `bind` answers the owner's current binding for the - * agent's community, or undefined when it may not be used now. */ + * agent's community, or undefined when it may not be used now. With `media`, + * native holds owner media for review. */ export function createOwnerAccess( - agent: Readonly<{ pubkey: string; owner: string }>, + agent: Readonly<{ pubkey: string; relay: string; owner: string }>, bind: () => OwnerBinding | undefined, + media?: OwnerMediaNative, ): OwnerAccess { /** The binding each answer was read with. */ const answered = new WeakMap(); @@ -281,6 +285,55 @@ export function createOwnerAccess( }; }; + /** Media read into native memory and held there for review; a cancel, + * unbind or rebind drops it, even one that comes before the bytes do. */ + async function ownerMedia( + native: OwnerMediaNative, + url: string, + signal: AbortSignal | undefined, + ): Promise { + const id = crypto.randomUUID(); + const member = { pubkey: agent.pubkey, relay: agent.relay }; + let bound: AbortSignal | undefined; + let settled = false; + const discard = () => { + if (settled) return; + settled = true; + bound?.removeEventListener("abort", discard); + void native.discard(id).catch(() => {}); + }; + try { + return await run(signal, async (binding, signal) => { + bound = signal; + signal.addEventListener("abort", discard, { once: true }); + const staged = await native.stage(member, id, url); + signal.throwIfAborted(); + return { + ...staged, + execution: execution(binding), + // Checked and started in one synchronous step, as a text answer is + // released right after its own check. + release: () => { + if (settled) throw changed(); + try { + holds(binding); + } catch (error) { + discard(); + throw error; + } + settled = true; + signal.removeEventListener("abort", discard); + return native.publish(member, id); + }, + discard, + }; + }); + } catch (error) { + discard(); + throw error; + } + } + return Object.freeze({ channels: (signal?: AbortSignal) => run(signal, (binding) => { @@ -539,6 +592,10 @@ export function createOwnerAccess( if (!binding) throw changed(); holds(binding); }, + ...(media && { + media: (url: string, signal?: AbortSignal) => + ownerMedia(media, url, signal), + }), }); } diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index dd5ebed4ee..5a3898d131 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -2453,19 +2453,23 @@ export class AgentsService extends Service implements Agents { settings: () => this.deviceSettings, ownerAccess: (agent) => ownerMcpEnabled(agent.relay) - ? createOwnerAccess(agent, () => { - const binding = this.bindingOf(agent); - return binding && - ownerMcpEnabled(binding.origin) && - this.communities.snapshot().viewer === agent.owner - ? { - viewer: binding.viewer, - origin: binding.origin, - session: binding.session, - signal: binding.controller.signal, - } - : undefined; - }) + ? createOwnerAccess( + agent, + () => { + const binding = this.bindingOf(agent); + return binding && + ownerMcpEnabled(binding.origin) && + this.communities.snapshot().viewer === agent.owner + ? { + viewer: binding.viewer, + origin: binding.origin, + session: binding.session, + signal: binding.controller.signal, + } + : undefined; + }, + native?.ownerMedia, + ) : undefined, classifier: () => this.classifier(), decisions: () => this.decisions,