From ae3c93af982ad6a9317d7769d35cf2b7bc366448 Mon Sep 17 00:00:00 2001 From: Kyle Zemek Date: Fri, 6 Nov 2026 22:47:51 -0800 Subject: [PATCH 1/4] feat(agents): let a bundled agent get its community's media as its owner owner_media_get reads one media blob of the agent's community signed as its owner, holds it in native memory, and saves it to the agent's media folder only once protection has reviewed the answer and the binding, owner sign-in and agent still hold. Remote Codex and Claude targets do not list it and are refused it before any read. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Kyle Zemek --- docs/agents.md | 17 +- docs/buzz-help.md | 9 +- src-tauri/build.rs | 3 + src-tauri/capabilities/default.json | 3 + src-tauri/src/app_agents.rs | 46 +- src-tauri/src/app_agents/owner_media.rs | 394 ++++++++++++++ src-tauri/src/app_agents/owner_media/tests.rs | 484 ++++++++++++++++++ src-tauri/src/browser_permissions_tests.rs | 3 + src-tauri/src/federated_identity/agent/mod.rs | 32 ++ src-tauri/src/identity.rs | 51 +- src-tauri/src/lib.rs | 5 +- src-tauri/src/relay.rs | 109 +++- src/bundled/claude-code/execution.test.ts | 27 +- src/bundled/claude-code/runtime.ts | 13 +- src/bundled/codex/runtime.test.ts | 56 ++ src/bundled/codex/runtime.ts | 26 +- src/bundled/pi/runtime.ts | 4 +- src/buzz-mcp/AGENTS.md | 9 +- src/buzz-mcp/app-client.ts | 24 + src/buzz-mcp/client.ts | 13 + src/buzz-mcp/owner.test.ts | 130 ++++- src/buzz-mcp/owner.ts | 67 ++- src/buzz-mcp/rpc.ts | 2 +- src/buzz-mcp/session-tools.ts | 46 +- src/buzz-mcp/tools.ts | 22 +- src/features/agent-harness/kit.ts | 2 +- src/features/agents/native.ts | 18 + src/features/agents/owner-access.test.ts | 260 +++++++++- src/features/agents/owner-access.ts | 61 ++- src/features/agents/service.ts | 30 +- 30 files changed, 1841 insertions(+), 125 deletions(-) create mode 100644 src-tauri/src/app_agents/owner_media.rs create mode 100644 src-tauri/src/app_agents/owner_media/tests.rs 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..0863748cd7 --- /dev/null +++ b/src-tauri/src/app_agents/owner_media.rs @@ -0,0 +1,394 @@ +//! 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), +} +impl Op { + fn reserved(&self) -> u64 { + match self { + Op::Fetching { reserved, .. } => *reserved, + Op::Staged(staged) => staged.body.len() 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 { .. } => true, + }); + self.cancelled + .retain(|_, at| now.duration_since(*at) < LIFETIME); + } +} + +/// Owner media between its read and its publication, in memory only. +#[derive(Default)] +pub(crate) struct OwnerMedia(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(); + 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 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. + fn take(&self, id: &str, agent: &str, relay: &str) -> Result { + 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 => {} + _ => return Err(NOTHING.into()), + } + match state.ops.remove(id) { + Some(Op::Staged(staged)) => Ok(staged), + _ => Err(NOTHING.into()), + } + } + /// Drops every read and body, as a new page or a closed window starts none. + pub(crate) fn clear(&self) { + let mut state = self.lock(); + for (_, op) in state.ops.drain() { + if let Op::Fetching { + cancel: Some(cancel), + .. + } = op + { + let _ = cancel.send(()); + } + } + state.cancelled.clear(); + } + #[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> { + let (owner, stamp, retirements) = host.owner_media.peek(id, pubkey, relay)?; + // 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, || { + 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 {} started but did not finish; part of it may have been written", + path.display() + ) + }) +} + +/// 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) { + 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..85a114b7da --- /dev/null +++ b/src-tauri/src/app_agents/owner_media/tests.rs @@ -0,0 +1,484 @@ +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]); + // 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"); + assert_eq!( + media.take("a", "agent", relay).unwrap().body, + PNG.as_bytes() + ); + assert_eq!(media.take("a", "agent", relay).err().unwrap(), NOTHING); + 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", "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()), + "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); +} + +/// Publication takes the identity, then the agent tokens' locks; sign-out +/// and deletion take them in that order or unnested. Racing them all must +/// finish, each publication writing all of its file or none of it. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn publication_keeps_the_credential_lock_order_under_sign_out_and_deletion() { + for round in 0..8 { + let identity = IdentityHost::fixture(); + let (f, host, agent, _store) = staged_for().await; + let dir = tempfile::tempdir().unwrap(); + let publishing = { + let (host, identity, path) = (host.clone(), identity.clone(), dir.path().to_owned()); + let (pubkey, relay) = (agent.pubkey.clone(), agent.relay.clone()); + tokio::task::spawn_blocking(move || { + publish(&host, &identity, &path, &pubkey, &relay, "one") + }) + }; + let stopping = { + let (tokens, identity) = (f.tokens.clone(), identity.clone()); + let (owner, pubkey) = (agent.owner.clone(), agent.pubkey.clone()); + tokio::spawn(async move { + if round % 2 == 0 { + tokens.forget(&owner, &pubkey).await.unwrap(); + } else { + tokens.sign_out(None).await.unwrap(); + identity.close(); + } + }) + }; + let (published, stopped) = tokio::time::timeout(Duration::from_secs(10), async { + (publishing.await.unwrap(), stopping.await) + }) + .await + .expect("no lock-order deadlock"); + stopped.unwrap(); + let written = std::fs::read_dir(dir.path()).unwrap().count(); + match published { + Ok(()) => assert_eq!( + std::fs::read(dir.path().join("blob.png")).unwrap(), + PNG.as_bytes() + ), + Err(_) => assert_eq!(written, 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..d2ce72866d 100644 --- a/src-tauri/src/federated_identity/agent/mod.rs +++ b/src-tauri/src/federated_identity/agent/mod.rs @@ -436,6 +436,38 @@ 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()); + } + if state.closed || state.retirements != retirements { + return Err(CHANGED.into()); + } + write() + } /// 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..ad250d8d32 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,76 @@ 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: nothing, not even discovery, is sent for another owner. + let template = blossom_template(&url, "get", "Get buzz-media", Vec::new())?; + let auth = blossom_header(host.sign_owner_media_get(owner.into(), template).await?)?; + let assertion = evidence(host, fi, &url).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(); + 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(|_| "Not enough memory for the media")?; + } + while let Some(chunk) = upstream + .chunk() + .await + .map_err(|_| "The media request failed")? + { + if (body.len() + chunk.len()) as u64 > reserved { + return Err(match declared { + Some(_) => "The media was larger than the community said".into(), + None => too_large(), + }); + } + 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..ffd5687f9b 100644 --- a/src/bundled/codex/runtime.test.ts +++ b/src/bundled/codex/runtime.test.ts @@ -6389,6 +6389,62 @@ 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(); + }, +); + 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..91a0ec49f6 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,7 +2578,7 @@ 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 owner = this.ownerAccess(agent); const tools = addedKey(added) + (owner ? ":owner" : ""); const resuming = existing?.workspace === workspace && (existing.tools ?? "") === tools; @@ -2990,7 +2999,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, From 14a3ae356a09876b785c4f6a41a5a120575a8679 Mon Sep 17 00:00:00 2001 From: Kyle Zemek Date: Fri, 6 Nov 2026 23:11:13 -0800 Subject: [PATCH 2/4] fix(agents): keep owner media bounded and refused through sign-out - Refuse publication while an organization sign-out's retirement is pending. - Count a staged or publishing body's capacity until it is dropped, and never grow an unknown-length body past its reservation. - Sign the media read again once the community credential is known. - Start Codex threads fresh when owner_media_get joins their tools. - Validate request ids on discard and publish; say nothing was saved when saving fails. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Kyle Zemek --- src-tauri/src/app_agents/owner_media.rs | 60 ++++++++++++++----- src-tauri/src/app_agents/owner_media/tests.rs | 26 ++++++-- src-tauri/src/federated_identity/agent/mod.rs | 10 +++- src-tauri/src/relay.rs | 21 +++++-- src/bundled/codex/runtime.test.ts | 40 +++++++++++++ src/bundled/codex/runtime.ts | 4 +- 6 files changed, 132 insertions(+), 29 deletions(-) diff --git a/src-tauri/src/app_agents/owner_media.rs b/src-tauri/src/app_agents/owner_media.rs index 0863748cd7..7b65fc5e13 100644 --- a/src-tauri/src/app_agents/owner_media.rs +++ b/src-tauri/src/app_agents/owner_media.rs @@ -46,12 +46,16 @@ enum Op { 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, .. } => *reserved, - Op::Staged(staged) => staged.body.len() as u64, + Op::Fetching { reserved, .. } | Op::Publishing { reserved } => *reserved, + Op::Staged(staged) => staged.body.capacity() as u64, } } } @@ -65,7 +69,7 @@ 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 { .. } => true, + Op::Fetching { .. } | Op::Publishing { .. } => true, }); self.cancelled .retain(|_, at| now.duration_since(*at) < LIFETIME); @@ -107,6 +111,7 @@ impl OwnerMedia { /// 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() @@ -145,6 +150,9 @@ impl OwnerMedia { 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), @@ -180,18 +188,31 @@ impl OwnerMedia { } } /// 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()); - match state.ops.get(id) { - Some(Op::Staged(staged)) if staged.agent == agent && staged.relay == relay => {} - _ => return Err(NOTHING.into()), - } - match state.ops.remove(id) { - Some(Op::Staged(staged)) => Ok(staged), + 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. pub(crate) fn clear(&self) { let mut state = self.lock(); @@ -346,7 +367,10 @@ fn publish( relay: &str, id: &str, ) -> Result<(), String> { + valid_id(id)?; let (owner, stamp, retirements) = host.owner_media.peek(id, pubkey, relay)?; + // 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. @@ -376,18 +400,22 @@ fn publish( .and_then(|()| file.flush()) .map_err(drop) .and_then(|()| file.persist(&path).map(drop).map_err(drop)) - .map_err(|()| { - format!( - "Saving {} started but did not finish; part of it may have been written", - path.display() - ) - }) + .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) { - state.owner_media.discard(&id); + if valid_id(&id).is_ok() { + state.owner_media.discard(&id); + } } #[cfg(test)] diff --git a/src-tauri/src/app_agents/owner_media/tests.rs b/src-tauri/src/app_agents/owner_media/tests.rs index 85a114b7da..1d891369dd 100644 --- a/src-tauri/src/app_agents/owner_media/tests.rs +++ b/src-tauri/src/app_agents/owner_media/tests.rs @@ -219,6 +219,11 @@ async fn owner_media_bodies_are_bounded_by_what_was_reserved() { 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")]); @@ -331,11 +336,15 @@ fn a_staged_body_is_taken_once_by_its_own_agent_before_it_expires() { ); assert_eq!(media.held().0, 1, "a foreign call leaves it held"); assert_eq!(media.peek("a", "agent", relay).unwrap().0, "owner"); - assert_eq!( - media.take("a", "agent", relay).unwrap().body, - PNG.as_bytes() - ); + 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. + media.discard("a"); + 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(); @@ -390,12 +399,19 @@ async fn staged_for() -> ( #[tokio::test] async fn nothing_is_written_once_the_owner_or_agent_stops_acting() { - for change in ["delete", "sign-out", "identity-closed", "none"] { + 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(), _ => {} } diff --git a/src-tauri/src/federated_identity/agent/mod.rs b/src-tauri/src/federated_identity/agent/mod.rs index d2ce72866d..8e06155f84 100644 --- a/src-tauri/src/federated_identity/agent/mod.rs +++ b/src-tauri/src/federated_identity/agent/mod.rs @@ -449,7 +449,7 @@ impl AgentTokens { write: impl FnOnce() -> Result, ) -> Result { let row = row_key(agent, relay)?; - let _retired = lock(&self.0.retired); + let retired = lock(&self.0.retired); let incarnations = lock(&self.0.incarnations); let (at, row) = ( incarnations.get(agent).copied().unwrap_or_default(), @@ -463,11 +463,17 @@ impl AgentTokens { { return Err(DELETED.into()); } - if state.closed || state.retirements != retirements { + // 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); + } /// 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/relay.rs b/src-tauri/src/relay.rs index ad250d8d32..4edb0010d1 100644 --- a/src-tauri/src/relay.rs +++ b/src-tauri/src/relay.rs @@ -2271,10 +2271,13 @@ pub(crate) async fn owner_media( limit: u64, reserve: impl FnOnce(u64) -> Result<()>, ) -> Result<(Option, Vec)> { - // Signed first: nothing, not even discovery, is sent for another owner. - let template = blossom_template(&url, "get", "Get buzz-media", Vec::new())?; - let auth = blossom_header(host.sign_owner_media_get(owner.into(), template).await?)?; + // 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 @@ -2300,22 +2303,30 @@ pub(crate) async fn owner_media( .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(|_| "Not enough memory for the media")?; + .map_err(no_memory)?; } while let Some(chunk) = upstream .chunk() .await .map_err(|_| "The media request failed")? { - if (body.len() + chunk.len()) as u64 > reserved { + 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)) diff --git a/src/bundled/codex/runtime.test.ts b/src/bundled/codex/runtime.test.ts index ffd5687f9b..b09e85e85a 100644 --- a/src/bundled/codex/runtime.test.ts +++ b/src/bundled/codex/runtime.test.ts @@ -6445,6 +6445,46 @@ it.each([false, true])( }, ); +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 91a0ec49f6..2e2a9811d0 100644 --- a/src/bundled/codex/runtime.ts +++ b/src/bundled/codex/runtime.ts @@ -2579,7 +2579,9 @@ export class CodexRuntime { // tools starts fresh. const added = this.added(); const owner = this.ownerAccess(agent); - const tools = addedKey(added) + (owner ? ":owner" : ""); + 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 } }; From 9572f5f584b03daf8ad3d8a5fd827c8c9891954e Mon Sep 17 00:00:00 2001 From: Kyle Zemek Date: Fri, 6 Nov 2026 23:49:49 -0800 Subject: [PATCH 3/4] test(agents): prove owner media's fence order with held publications Replace the ungated lock-order race with two controlled interleavings: a publication held after review refuses a deletion, sign-out or closed identity that came first, and a sign-out that arrives while a publication holds the fence waits for its whole file. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Kyle Zemek --- src-tauri/src/app_agents/owner_media.rs | 36 +++++- src-tauri/src/app_agents/owner_media/tests.rs | 112 ++++++++++++------ src-tauri/src/federated_identity/agent/mod.rs | 7 ++ 3 files changed, 116 insertions(+), 39 deletions(-) diff --git a/src-tauri/src/app_agents/owner_media.rs b/src-tauri/src/app_agents/owner_media.rs index 7b65fc5e13..39ac67ecaf 100644 --- a/src-tauri/src/app_agents/owner_media.rs +++ b/src-tauri/src/app_agents/owner_media.rs @@ -76,9 +76,23 @@ impl State { } } +/// 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); +pub(crate) struct OwnerMedia(Mutex, #[cfg(test)] Mutex); impl OwnerMedia { fn lock(&self) -> std::sync::MutexGuard<'_, State> { self.0 @@ -227,6 +241,22 @@ impl OwnerMedia { } 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(); @@ -369,6 +399,8 @@ fn publish( ) -> 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' @@ -383,6 +415,8 @@ fn publish( 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) diff --git a/src-tauri/src/app_agents/owner_media/tests.rs b/src-tauri/src/app_agents/owner_media/tests.rs index 1d891369dd..401cd88e25 100644 --- a/src-tauri/src/app_agents/owner_media/tests.rs +++ b/src-tauri/src/app_agents/owner_media/tests.rs @@ -454,47 +454,83 @@ async fn nothing_is_written_once_the_owner_or_agent_stops_acting() { assert_eq!(std::fs::read_dir(dir.path()).map_or(0, Iterator::count), 0); } -/// Publication takes the identity, then the agent tokens' locks; sign-out -/// and deletion take them in that order or unnested. Racing them all must -/// finish, each publication writing all of its file or none of it. -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn publication_keeps_the_credential_lock_order_under_sign_out_and_deletion() { - for round in 0..8 { +/// 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 publishing = { - let (host, identity, path) = (host.clone(), identity.clone(), dir.path().to_owned()); - let (pubkey, relay) = (agent.pubkey.clone(), agent.relay.clone()); - tokio::task::spawn_blocking(move || { - publish(&host, &identity, &path, &pubkey, &relay, "one") - }) - }; - let stopping = { - let (tokens, identity) = (f.tokens.clone(), identity.clone()); - let (owner, pubkey) = (agent.owner.clone(), agent.pubkey.clone()); - tokio::spawn(async move { - if round % 2 == 0 { - tokens.forget(&owner, &pubkey).await.unwrap(); - } else { - tokens.sign_out(None).await.unwrap(); - identity.close(); - } - }) - }; - let (published, stopped) = tokio::time::timeout(Duration::from_secs(10), async { - (publishing.await.unwrap(), stopping.await) - }) - .await - .expect("no lock-order deadlock"); - stopped.unwrap(); - let written = std::fs::read_dir(dir.path()).unwrap().count(); - match published { - Ok(()) => assert_eq!( - std::fs::read(dir.path().join("blob.png")).unwrap(), - PNG.as_bytes() - ), - Err(_) => assert_eq!(written, 0), + 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/federated_identity/agent/mod.rs b/src-tauri/src/federated_identity/agent/mod.rs index 8e06155f84..7fbfd980f0 100644 --- a/src-tauri/src/federated_identity/agent/mod.rs +++ b/src-tauri/src/federated_identity/agent/mod.rs @@ -474,6 +474,13 @@ impl AgentTokens { 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 { From 94b5b17ea37e641a4d8c2c78a0b18f813b22f758 Mon Sep 17 00:00:00 2001 From: Kyle Zemek Date: Sat, 7 Nov 2026 00:07:59 -0800 Subject: [PATCH 4/4] fix(agents): keep a writing publication counted across reload A reload cleared every owner media operation, including a publication still writing its body, so the next page could reserve a full budget while that body was live. Keep publishing operations until their publication drops them. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Kyle Zemek --- src-tauri/src/app_agents/owner_media.rs | 18 ++++++++++-------- src-tauri/src/app_agents/owner_media/tests.rs | 4 +++- 2 files changed, 13 insertions(+), 9 deletions(-) diff --git a/src-tauri/src/app_agents/owner_media.rs b/src-tauri/src/app_agents/owner_media.rs index 39ac67ecaf..da434b0f78 100644 --- a/src-tauri/src/app_agents/owner_media.rs +++ b/src-tauri/src/app_agents/owner_media.rs @@ -228,17 +228,19 @@ impl OwnerMedia { } } /// 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(); - for (_, op) in state.ops.drain() { - if let Op::Fetching { - cancel: Some(cancel), - .. - } = op - { - let _ = cancel.send(()); + 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. diff --git a/src-tauri/src/app_agents/owner_media/tests.rs b/src-tauri/src/app_agents/owner_media/tests.rs index 401cd88e25..2e3427b60f 100644 --- a/src-tauri/src/app_agents/owner_media/tests.rs +++ b/src-tauri/src/app_agents/owner_media/tests.rs @@ -339,8 +339,10 @@ fn a_staged_body_is_taken_once_by_its_own_agent_before_it_expires() { 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. + // 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);