From 57303061b5335bb0aac158933d610752336ed687 Mon Sep 17 00:00:00 2001 From: Zach Marley Date: Tue, 3 Nov 2026 16:24:15 -0500 Subject: [PATCH 1/4] feat(host): release opted-in attachments after confirmed sends Record bounded process-local same-owner consent at managed save, pass snapshot-matching attachment hints outside signed events, and release exact verified sources only after native publication confirmation. Keep cleanup best-effort and preserve confirmed send success. Co-authored-by: goose Signed-off-by: Zach Marley --- docs/buzz-help.md | 6 +- docs/plugin-architecture.md | 76 +- src-tauri/src/app_agents.rs | 138 ++- src-tauri/src/host_files.rs | 505 ++++++++++- src-tauri/src/host_files/managed_tests.rs | 985 +++++++++++++++++++++- src-tauri/src/identity.rs | 17 + src-tauri/src/identity/tests.rs | 26 + src-tauri/src/lib.rs | 1 + src/buzz-mcp/app-client.test.ts | 50 ++ src/buzz-mcp/app-client.ts | 6 +- src/buzz-mcp/client.ts | 13 +- src/buzz-mcp/tools.test.ts | 186 ++++ src/buzz-mcp/tools.ts | 59 +- src/features/agents/native.test.ts | 24 + src/features/agents/native.ts | 12 +- src/features/agents/runner.ts | 13 +- src/features/agents/service.test.ts | 56 ++ src/features/agents/service.ts | 7 +- src/features/host/service.test.ts | 92 ++ src/features/host/service.ts | 27 +- src/plugins/author.test.mjs | 9 +- 21 files changed, 2207 insertions(+), 101 deletions(-) diff --git a/docs/buzz-help.md b/docs/buzz-help.md index d72c24d86b..8667796249 100644 --- a/docs/buzz-help.md +++ b/docs/buzz-help.md @@ -182,8 +182,10 @@ that agent in Buzz; they do not notify the external app through this connection. uploads are available. Files and media are shared with the conversation's members. If an upload fails, its attachment shows the reason, **Retry**, and **Remove**. Follow the error's guidance before retrying, or remove the attachment - to unblock Send; your typed message stays in the composer. Do not promise - support for every file type or size. + to unblock Send; your typed message stays in the composer. Some plugins supply + temporary attachment sources that may be released after a confirmed post; keep + the posted link for reuse, and do not assume local cleanup will happen + automatically. Do not promise support for every file type or size. - **Emoji suggestions** appear when you type `:` and an emoji name. If a name is shortened with an ellipsis, hover over it or use the arrow keys to select it and see the full name in a tooltip. diff --git a/docs/plugin-architecture.md b/docs/plugin-architecture.md index 34c26c180f..fb1b4a6b69 100644 --- a/docs/plugin-architecture.md +++ b/docs/plugin-architecture.md @@ -1500,8 +1500,80 @@ refuse. Only the verified fixed ownership file is excluded. The shared root permits at most 64 MiB and 16 artifact files. Any plugin can fill it; there is no per-plugin quota or fairness promise. A full root refuses before another file is created, without evicting unexpired files. Failures remove output -best-effort; residue remains counted. There is no index, sidecar, recovery adoption, -content deduplication or temporary-to-final file service. +best-effort; residue remains counted. There is no persistent index, sidecar, +recovery adoption, content deduplication or temporary-to-final file service. +The optional after-send policy below uses only a bounded in-memory consent map. + +#### Opt-in temporary attachment release + +A compatible desktop frontend/native pair accepts +`await host.saveBytes(bytes, {sha256, releaseAfterSend: true})`. The flag defaults +to off; false/omission preserves ordinary saves. Require the returned +`releaseAfterSend === true` acknowledgement. An older native host may save while +ignoring the option: missing/false acknowledgement rejects as unsupported and +**possibly saved**, never permission to retry. Establish a compatible released +pair before enabling a consumer; method detection alone is not evidence. +No new grant, caller path/account, deletion callback or release command exists. + +Native captures the saving **human Nostr owner**, not a plugin ID or federated +login account. Opt-in saves may refuse if identity is unavailable, closed or busy +at the final commit. Consent insertion and identity close share a short native +fence; ordinary saves remain identity-independent. Native command timeout/drop +before commit prevents consent. Renderer disposal/page retirement/lost IPC replies +are not native cancellation signals: they can leave a possibly saved file and +committed consent even before the renderer learns the result. Never automatically +save again to recover an unknown outcome. + +The app's shared attachment tools collect bounded local hints after read/upload, +using the bytes' exact SHA256/size and a matching upload descriptor. These hints +travel outside signed event tags and tool arguments. They do not create deletion +authority. Only after the **full native evidence-wrapped publication** succeeds +with the relay accepting that exact signed kind-9 event ID does native attempt +release. Upload, local event insertion, renderer claims and an unconfirmed receipt +are insufficient. This is not recipient delivery/read or permanent remote storage. +The trusted app supplies read/upload mapping; no independent cryptographic native +provenance or malicious-plugin sandbox is claimed. Standalone/browser clients do +not gain a local deletion implementation. + +A different runtime plugin may send the saving plugin's source for the same human +owner. Native requires its process-local save consent, exact direct-child path, +saver digest, private no-link file identity, content hash/size, and the confirmed +event's URL/hash/size together in one imeta. Saver update/disable and later UI or +federated-session changes do not retarget previously committed consent. Unknown, +legacy, replaced, aliased, unmarked or other-owner files never acquire consent +from filenames or tool hints. Platforms without safely verifiable file identity +retain the source. Public explicit deletion retains its existing grant/owner rules. + +One bounded best-effort attempt per distinct source uses the shared byte permit, +root writer and reader exclusion, with **one shared 30-second cleanup-entry budget +for the whole message**, including queue wait. Failure/refusal never changes a +confirmed send into an error or repost. Active app-wide readers refuse cleanup; +while cleanup verifies content, new direct/process reads instead promptly refuse +with `File reads are busy; try again later`, without spawning/reading or silently +retrying. A stalled hash can outlive the response budget; dispatched work retains +admission/exclusion through real completion and its unlink outcome can be unknown. + +| Outcome | Source and caller behavior | +| --- | --- | +| Read/upload/body/mention/emoji failure, or native-unconfirmed publish | Keep sources; no cleanup or automatic resend. Earlier uploads may exist remotely. | +| Confirmed native publish, cleanup succeeds | Local source removed and quota reclaimed; posted message/media unchanged. | +| Confirmed publish, busy reader/root/queue, mismatch or cleanup error | Send still succeeds; source may remain indefinitely, without a cleanup retry. | +| Dispatched cleanup exceeds the shared deadline | Send succeeds; local cleanup outcome unknown and actual work may finish late. | +| Second sequential send of a released path | Normal missing-file failure before upload; reuse the posted link, not the old source. | +| Concurrent send with bytes already read | It may finish from its snapshot even after the first confirmed send releases the source. Future reads are not reserved. | +| Restart/crash/different process | Consent is lost, never adopted from a filename; files may remain indefinitely. | + +**Cleanup may never happen automatically; expiry only makes a file eligible for removal during a later save.** +There is no durable/cross-process consent, background sweep, completion callback +or eventual-erasure guarantee. Only this new after-send path requires confirmed +publish; owner deletion, lazy expiry sweeping, erasure and disk loss remain +independent. Save acknowledgement is not a publication/deletion receipt; +`sent ` confirms publication, not local deletion. + +When returning an opted-in path, explain: “Temporary attachment source: send this +file once. After a confirmed post, the host may release the local copy; keep the +posted link for reuse. Failed or uncertain sends are not safe to repeat +automatically.” Also disclose that cleanup refusal may retain it indefinitely. Optional `host.deleteManagedFile(path)` requires `host.managedFiles` and the current enabled plugin revision, not `host.files`. It accepts only a direct, host-generated diff --git a/src-tauri/src/app_agents.rs b/src-tauri/src/app_agents.rs index 0dad1ce0f2..a7c77940c5 100644 --- a/src-tauri/src/app_agents.rs +++ b/src-tauri/src/app_agents.rs @@ -878,23 +878,65 @@ pub(crate) struct AppAgentEvent { /// Signs one event as the agent and posts it to the agent's own community. #[tauri::command] -pub(crate) async fn app_agent_publish( +pub(crate) async fn app_agent_publish( + app: tauri::AppHandle, state: tauri::State<'_, AppAgentHost>, pubkey: String, relay: String, event: AppAgentEvent, + managed_attachments: Option, ) -> Result { + use tauri::Manager as _; let (agent, key, evidence) = state.signer(pubkey, relay).await?; let signed = agent.sign(&key, event.kind, event.content, event.tags)?; + let root = app + .path() + .app_data_dir() + .ok() + .map(|root| root.join("plugin-artifacts")); + confirmed_post( + &agent, + &key, + evidence, + signed, + app.state::() + .inner() + .clone(), + root, + managed_attachments, + ) + .await +} + +async fn confirmed_post( + agent: &AppAgent, + key: &Secret, + evidence: Admission, + signed: Value, + registry: crate::host_files::ManagedFiles, + root: Option, + managed_attachments: Option, +) -> Result { let event_id = signed .get("id") .and_then(Value::as_str) .ok_or("Invalid agent event")? .to_owned(); - evidence - .run(true, post(&agent, &key, evidence.header(), signed)) + let confirmed = evidence + .run(true, post(agent, key, evidence.header(), signed)) .await - .map_err(|error| publish_error(&event_id, error)) + .map_err(|error| publish_error(&event_id, error))?; + if let Some(root) = root { + crate::host_files::release_after_send( + registry, + root, + agent.owner.clone(), + confirmed.clone(), + managed_attachments, + ) + .await; + } + Ok(confirmed) } fn publish_error(event_id: &str, error: String) -> String { @@ -2010,7 +2052,7 @@ mod tests { (origin, served) } - fn agent_at(relay: String, key: &Secret) -> AppAgent { + pub(super) fn agent_at(relay: String, key: &Secret) -> AppAgent { AppAgent { pubkey: key.pubkey().into(), relay, @@ -2574,3 +2616,89 @@ mod tests { } } } + +#[cfg(test)] +mod managed_publish_tests { + use super::*; + use crate::host_files::{ + managed_tests::{commands, release_fixture_data, Fixture}, + ManagedFiles, + }; + use tauri::Manager as _; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + /// The actual HTTP post/evidence wrapper + real managed save/unlink. Loopback + /// fixture only: not Wry/trusted-HTTPS or native signing-custody acceptance. + #[test] + fn host_release_only_after_full_native_matching_receipt_and_evidence_admission() { + let _serial = commands(); + for outcome in [ + "accepted", + "refused", + "wrong-id", + "bad-json", + "lost-reply", + "http-error", + "retired", + "cleanup-refused", + ] { + let f = Fixture::new(true, false); + let (hints, event, owner) = release_fixture_data(&f, b"native confirmed source"); + let path = PathBuf::from(hints[0]["path"].as_str().unwrap()); + let registry = f.app.state::().inner().clone(); + let root = f.root(); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let tokens = AgentTokens::unconfigured(); + let key = Secret::generate().unwrap(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let mut agent = tests::agent_at("wss://relay.example".into(), &key); + agent.owner = owner; + let stamp = tokens.stamp(&agent.pubkey, &agent.relay).unwrap(); + let evidence = tokens.authorize(&agent, &key, stamp).await.unwrap(); + let retire = tokens.clone(); + let pubkey = agent.pubkey.clone(); + let relay = agent.relay.clone(); + agent.relay = format!("http://{}", listener.local_addr().unwrap()); + let tags = serde_json::from_value(event["tags"].clone()).unwrap(); + let signed = agent.sign(&key, 9, "attachment".into(), tags).unwrap(); + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut head = vec![]; + while !head.ends_with(b"\r\n\r\n") { + let mut byte = [0]; socket.read_exact(&mut byte).await.unwrap(); head.push(byte[0]); + } + let head = String::from_utf8(head).unwrap(); + let size: usize = head.lines().find_map(|line| line.to_ascii_lowercase().strip_prefix("content-length: ").map(str::to_owned)).unwrap().parse().unwrap(); + let mut body = vec![0; size]; socket.read_exact(&mut body).await.unwrap(); + let received: Value = serde_json::from_slice(&body).unwrap(); + if outcome == "retired" { retire.forget_row(&pubkey, &relay).unwrap(); } + if outcome == "lost-reply" { return received; } + let receipt = match outcome { + "refused" => serde_json::json!({"accepted":false,"event_id":received["id"]}).to_string(), + "wrong-id" => serde_json::json!({"accepted":true,"event_id":"b".repeat(64)}).to_string(), + "bad-json" => "not JSON".into(), + _ => serde_json::json!({"accepted":true,"event_id":received["id"]}).to_string(), + }; + let status = if outcome == "http-error" { "403 Forbidden" } else { "200 OK" }; + socket.write_all(format!("HTTP/1.1 {status}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{receipt}", receipt.len()).as_bytes()).await.unwrap(); + received + }); + let lease = if outcome == "cleanup-refused" { Some(crate::host_files::FileReadLease::new().unwrap()) } else { None }; + let result = confirmed_post(&agent, &key, evidence, signed, registry, Some(root), Some(hints)).await; + let received = server.await.unwrap(); // Exactly one connection/request; no repost path. + if matches!(outcome, "accepted" | "cleanup-refused") { + assert_eq!(result.unwrap()["id"], received["id"]); + assert_eq!(path.exists(), outcome == "cleanup-refused"); + } else { + assert!(result.is_err(), "{outcome} must be unconfirmed/refused"); + assert!(path.exists(), "{outcome} must retain source"); + } + drop(lease); + }); + } + } +} diff --git a/src-tauri/src/host_files.rs b/src-tauri/src/host_files.rs index 3f16b2e70d..1c59b90499 100644 --- a/src-tauri/src/host_files.rs +++ b/src-tauri/src/host_files.rs @@ -182,22 +182,53 @@ pub(crate) async fn byte_admission( // App-wide, conservative deletion exclusion: any plugin's host.readFile call or // process named read/read-protected blocks every plugin's managed deletion. // Leases outlive async waiters until the actual blocking/process work completes. -static FILE_READS: std::sync::Mutex = std::sync::Mutex::new(0); -pub(crate) struct FileReadLease; +static FILE_READS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); +const READ_EXCLUSIVE: usize = usize::MAX; +pub(crate) struct FileReadLease(()); impl FileReadLease { pub(crate) fn new() -> Result { - let mut reads = FILE_READS - .lock() - .map_err(|_| "File read admission unavailable")?; - *reads = reads.checked_add(1).ok_or("Too many file reads")?; - Ok(Self) + FILE_READS + .fetch_update( + std::sync::atomic::Ordering::SeqCst, + std::sync::atomic::Ordering::SeqCst, + |reads| (reads < READ_EXCLUSIVE - 1).then(|| reads + 1), + ) + .map_err(|reads| { + if reads == READ_EXCLUSIVE { + "File reads are busy; try again later" + } else { + "Too many file reads" + } + })?; + Ok(Self(())) } } impl Drop for FileReadLease { fn drop(&mut self) { - if let Ok(mut reads) = FILE_READS.lock() { - *reads -= 1; - } + FILE_READS.fetch_sub(1, std::sync::atomic::Ordering::SeqCst); + } +} +// A single nonblocking transition excludes new readers only when no real work +// holds a lease. Pinned rustc E0277 confirms std::sync::RwLockReadGuard is !Send; +// these leases must move into actual blocking-read/process-monitor work and outlive +// timed-out waiters. No mutex in async admission. +struct FileReadExclusion(()); +impl FileReadExclusion { + fn new() -> Result { + FILE_READS + .compare_exchange( + 0, + READ_EXCLUSIVE, + std::sync::atomic::Ordering::SeqCst, + std::sync::atomic::Ordering::SeqCst, + ) + .map_err(|_| "Managed file is in use by a file read")?; + Ok(Self(())) + } +} +impl Drop for FileReadExclusion { + fn drop(&mut self) { + FILE_READS.store(0, std::sync::atomic::Ordering::SeqCst); } } const MAX_SAVE_BYTES: usize = 16 * 1024 * 1024; @@ -214,6 +245,8 @@ pub(crate) struct SavedBytes { size: usize, sha256: String, expires_at: u64, + #[serde(skip_serializing_if = "Option::is_none")] + release_after_send: Option, } fn managed_granted( @@ -249,29 +282,370 @@ pub(crate) async fn plugin_host_save_bytes( ) -> Result { use tauri::Manager as _; let deadline = std::time::Instant::now() + SAVE_WINDOW; - // Refuse JSON fallback before any body clone or filesystem operation. let tauri::ipc::InvokeBody::Raw(body) = request.body() else { return Err("Managed save body must be raw bytes; outcome may be unknown".into()); }; let id = save_header(&request, "x-buzz-plugin-id", 256)?; let revision = save_header(&request, "x-buzz-plugin-revision", 128)?; let hash = save_header(&request, "x-buzz-sha256", 64)?; + let release = match request.headers().get("x-buzz-release-after-send") { + None => false, + Some(value) if value.as_bytes() == b"1" => true, + _ => return Err("Invalid managed save release flag".into()), + }; validate_save(body, &hash)?; - let place = byte_admission(tokio::time::Instant::from_std(deadline)).await?; - let manager = manager.0.clone()?; + let registry = app.state::().inner().clone(); + let identity = release.then(|| app.state::().inner().clone()); let root = app .path() .app_data_dir() .map_err(|e| e.to_string())? .join("plugin-artifacts"); - let bytes = body.clone(); - let writer = managed_writer(place, move || { - save_managed(&root, &bytes, &hash, deadline, &manager, &id, &revision) - }); - tokio::time::timeout_at(tokio::time::Instant::from_std(deadline), writer) + // The outer future owns this guard before its first await. Renderer disposal + // is not a native cancellation signal; timeout/drop here is. + let eligibility = SaveEligibility::default(); + let operation = async { + let owner = match identity.as_ref() { + Some(identity) => Some(identity.viewer().await?), + None => None, + }; + #[cfg(test)] + if let Some(entered) = registry.1.lock().unwrap().captured.take() { + let _ = entered.send(()); + } + let place = byte_admission(tokio::time::Instant::from_std(deadline)).await?; + let manager = manager.0.clone()?; + let bytes = body.clone(); + let consent = identity.zip(owner).map(|(identity, owner)| SaveConsent { + identity, + owner, + registry: registry.clone(), + eligibility: eligibility.0.clone(), + }); + managed_writer(place, move || { + save_managed(ManagedSave { + root: &root, + bytes: &bytes, + hash: &hash, + deadline, + manager: &manager, + id: &id, + revision: &revision, + registry: Some(®istry), + consent: consent.as_ref(), + }) + }) .await - .map_err(|_| "Managed save timed out; outcome unknown")? .map_err(|e| e.to_string())? + }; + tokio::time::timeout_at(tokio::time::Instant::from_std(deadline), operation) + .await + .map_err(|_| "Managed save timed out; outcome unknown")? +} + +// Command-owned cancellation races atomically with the short identity/map commit. +// Neither a lost reply nor JS-local retirement is evidence this guard was dropped. +#[derive(Default)] +struct SaveEligibility(std::sync::Arc); +impl Drop for SaveEligibility { + fn drop(&mut self) { + let _ = self.0.compare_exchange( + 0, + 2, + std::sync::atomic::Ordering::SeqCst, + std::sync::atomic::Ordering::SeqCst, + ); + } +} +struct SaveConsent { + identity: crate::identity::IdentityHost, + owner: String, + registry: ManagedFiles, + eligibility: std::sync::Arc, +} + +// One actual app owns one root and at most ROOT_FILES compact process-local records. +// Never adopt a filename after restart. No bytes, sidecars, timers or public release command. +#[derive(Clone, Default)] +pub(crate) struct ManagedFiles( + std::sync::Arc>>, + #[cfg(test)] std::sync::Arc>, +); +#[cfg(test)] +#[derive(Default)] +struct ManagedGates { + captured: Option>, + save: Option<( + tokio::sync::oneshot::Sender<()>, + std::sync::mpsc::Receiver<()>, + )>, + hash: Option<( + tokio::sync::oneshot::Sender<()>, + std::sync::mpsc::Receiver<()>, + )>, +} +#[cfg(test)] +fn gated_work( + gate: Option<( + tokio::sync::oneshot::Sender<()>, + std::sync::mpsc::Receiver<()>, + )>, +) { + if let Some((entered, release)) = gate { + let _ = entered.send(()); + release + .recv_timeout(std::time::Duration::from_secs(10)) + .expect("managed work barrier watchdog"); + } +} +#[derive(Clone)] +struct Consent { + root: PathBuf, + saver: String, + owner: String, + hash: String, + size: u64, + identity: Option<(u64, u64)>, +} +fn file_identity(metadata: &std::fs::Metadata) -> Option<(u64, u64)> { + #[cfg(unix)] + { + use std::os::unix::fs::MetadataExt as _; + Some((metadata.dev(), metadata.ino())) + } + #[cfg(not(unix))] + { + let _ = metadata; + None + } +} +impl ManagedFiles { + // Only called inside admitted work with the root lock held. No map -> root/identity order. + fn removed(&self, path: &Path) { + if let Ok(mut records) = self.0.try_lock() { + records.remove(path); + } + } + fn prune(&self, root: &Path) -> Result<(), String> { + let stale: Vec<_> = { + let records = self.0.try_lock().map_err(|_| "Managed consent is busy")?; + records + .keys() + .filter(|path| path.parent() == Some(root)) + .cloned() + .collect() + }; + // Filesystem I/O never holds the map mutex. + for path in stale { + if std::fs::symlink_metadata(&path) + .is_err_and(|e| e.kind() == std::io::ErrorKind::NotFound) + { + self.removed(&path); + } + } + Ok(()) + } +} + +#[derive(Clone, serde::Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct ManagedAttachment { + path: String, + sha256: String, + size: u64, + url: String, +} +fn confirmed_candidates( + signed: &serde_json::Value, + hints: Option, +) -> Vec { + if signed.get("kind").and_then(serde_json::Value::as_u64) != Some(9) { + return vec![]; + } + let Some(serde_json::Value::Array(hints)) = hints else { + return vec![]; + }; + if hints.len() > ROOT_FILES { + return vec![]; + } + let Some(tags) = signed.get("tags").and_then(serde_json::Value::as_array) else { + return vec![]; + }; + let mut candidates: Vec> = vec![]; + let mut paths = std::collections::HashMap::new(); + for hint in hints { + let Ok(candidate) = serde_json::from_value::(hint) else { + return vec![]; + }; + if candidate.path.is_empty() + || candidate.path.len() > 4096 + || candidate.path.contains('\0') + || candidate.url.is_empty() + || candidate.url.len() > 8192 + || candidate.size == 0 + || candidate.size > MAX_SAVE_BYTES as u64 + || validate_save(&[1], &candidate.sha256).is_err() + { + return vec![]; + } + let matches = tags.iter().any(|tag| { + let Some(tag) = tag.as_array() else { + return false; + }; + tag.first().and_then(serde_json::Value::as_str) == Some("imeta") + && [ + format!("url {}", candidate.url), + format!("x {}", candidate.sha256), + format!("size {}", candidate.size), + ] + .iter() + .all(|field| tag.iter().any(|value| value.as_str() == Some(field))) + }); + if let Some(&index) = paths.get(&candidate.path) { + let previous: &mut Option = &mut candidates[index]; + if !matches + || previous.as_ref().is_none_or(|previous| { + previous.sha256 != candidate.sha256 || previous.size != candidate.size + }) + { + *previous = None; + } + } else { + paths.insert(candidate.path.clone(), candidates.len()); + candidates.push(matches.then_some(candidate)); + } + } + candidates.into_iter().flatten().collect() +} + +/// Called only after the full evidence-wrapped exact-ID post has succeeded. +/// Invalid hints and every cleanup failure preserve that original send success. +pub(crate) async fn release_after_send( + registry: ManagedFiles, + root: PathBuf, + owner: String, + signed: serde_json::Value, + hints: Option, +) { + let deadline = tokio::time::Instant::now() + SAVE_WINDOW; + let candidates = confirmed_candidates(&signed, hints); + if candidates.is_empty() { + return; + } + let budget = SaveEligibility::default(); + let active = budget.0.clone(); + let _ = delete_with_admission(deadline, move || { + release_batch( + ®istry, + &root, + &owner, + &candidates, + deadline.into_std(), + &active, + ) + }) + .await; +} + +fn release_batch( + registry: &ManagedFiles, + root: &Path, + owner: &str, + candidates: &[ManagedAttachment], + deadline: std::time::Instant, + active: &std::sync::atomic::AtomicU8, +) -> Result<(), String> { + use sha2::{Digest as _, Sha256}; + let _lock = managed_lock(root)?; + let _reads = FileReadExclusion::new()?; + for candidate in candidates { + if std::time::Instant::now() >= deadline + || active.load(std::sync::atomic::Ordering::SeqCst) != 0 + { + break; + } + // One refusal never authorizes another path or prevents later safe candidates. + let _ = (|| -> Result<(), String> { + let path = Path::new(&candidate.path); + if path.parent() != Some(root) + || path + .file_name() + .is_none_or(|name| root.join(name).as_os_str() != path.as_os_str()) + { + return Err("Invalid managed attachment path".into()); + } + let record = registry + .0 + .try_lock() + .map_err(|_| "Managed consent is busy")? + .get(path) + .cloned(); + let Some(record) = record else { + return Ok(()); + }; + let name = path + .file_name() + .and_then(|name| name.to_str()) + .ok_or("Invalid managed attachment name")?; + if record.root != root + || record.owner != owner + || record.hash != candidate.sha256 + || record.size != candidate.size + || artifact_expiry(name).is_none() + || !name.ends_with(&format!("-{}", managed_owner(&record.saver))) + { + return Err("Managed attachment consent mismatch".into()); + } + let metadata = private_metadata(path, false)?; + let mut file = private_options().open(path).map_err(|e| e.to_string())?; + let opened = file.metadata().map_err(|e| e.to_string())?; + if record.identity.is_none() + || record.identity != file_identity(&metadata) + || record.identity != file_identity(&opened) + || opened.len() != record.size + { + return Err("Managed attachment changed".into()); + } + #[cfg(test)] + { + let gate = registry.1.lock().unwrap().hash.take(); + gated_work(gate); + } + let mut hash = Sha256::new(); + let mut buffer = [0; 64 * 1024]; + let mut size = 0u64; + loop { + let count = file.read(&mut buffer).map_err(|e| e.to_string())?; + if count == 0 { + break; + } + size += count as u64; + if size > record.size { + return Err("Managed attachment grew".into()); + } + hash.update(&buffer[..count]); + } + if size != record.size || format!("{:x}", hash.finalize()) != record.hash { + return Err("Managed attachment content changed".into()); + } + // Keep the descriptor, root and reader exclusion through final identity/unlink. + let final_metadata = private_metadata(path, false)?; + let final_opened = file.metadata().map_err(|e| e.to_string())?; + if file_identity(&final_metadata) != record.identity + || file_identity(&final_opened) != record.identity + || final_metadata.len() != size + || final_opened.len() != size + { + return Err("Managed attachment changed".into()); + } + // This is already-dispatched work. Timeout may have answered the sender, + // but the writer retains all admission until actual completion. + std::fs::remove_file(path).map_err(|e| e.to_string())?; + registry.removed(path); + Ok(()) + })(); + } + Ok(()) } /// The writer owns admission until actual completion, not merely until await ends. @@ -437,6 +811,7 @@ fn delete_managed( manager: &buzzodz_plugins::Manager, id: &str, revision: &str, + registry: Option<&ManagedFiles>, ) -> Result<(), String> { managed_granted(manager, id, revision)?; let path = Path::new(path); @@ -451,15 +826,14 @@ fn delete_managed( return Err("Managed file is not owned by this plugin".into()); } let _lock = managed_lock(root)?; - let reads = FILE_READS - .try_lock() - .map_err(|_| "Managed file is in use by a file read")?; - if *reads != 0 { - return Err("Managed file is in use by a file read".into()); - } + let _reads = FileReadExclusion::new()?; private_metadata(path, false)?; managed_granted(manager, id, revision)?; - std::fs::remove_file(path).map_err(|e| e.to_string()) + std::fs::remove_file(path).map_err(|e| e.to_string())?; + if let Some(registry) = registry { + registry.removed(path); + } + Ok(()) } #[tauri::command] @@ -478,8 +852,9 @@ pub(crate) async fn plugin_host_delete_managed_file( .map_err(|e| e.to_string())? .join("plugin-artifacts"); let manager = manager.0.clone()?; + let registry = app.state::().inner().clone(); delete_with_admission(deadline, move || { - delete_managed(&root, &path, &manager, &id, &revision) + delete_managed(&root, &path, &manager, &id, &revision, Some(®istry)) }) .await } @@ -541,26 +916,10 @@ fn managed_capacity(root: &Path, now: u64, incoming: usize) -> Result<(), String Ok(()) } -fn save_managed( - root: &Path, - bytes: &[u8], - hash: &str, - deadline: std::time::Instant, - manager: &buzzodz_plugins::Manager, - id: &str, - revision: &str, -) -> Result { +fn save_managed(input: ManagedSave<'_>) -> Result { use std::io::Write as _; save_managed_with_io( - ManagedSave { - root, - bytes, - hash, - deadline, - manager, - id, - revision, - }, + input, |file, bytes| file.write_all(bytes), std::fs::File::sync_all, |root| { @@ -581,6 +940,8 @@ struct ManagedSave<'a> { manager: &'a buzzodz_plugins::Manager, id: &'a str, revision: &'a str, + registry: Option<&'a ManagedFiles>, + consent: Option<&'a SaveConsent>, } // Keep OS operations private and local; admission, result and cleanup stay in one sequence. @@ -599,6 +960,8 @@ fn save_managed_with_io( manager, id, revision, + registry, + consent, } = input; validate_save(bytes, hash)?; managed_granted(manager, id, revision)?; @@ -608,7 +971,11 @@ fn save_managed_with_io( .checked_add(RETENTION_MS) .filter(|n| *n <= 9_007_199_254_740_991) .ok_or("Invalid managed file expiry")?; - managed_capacity(root, now, bytes.len())?; + let capacity = managed_capacity(root, now, bytes.len()); + if let Some(registry) = registry { + registry.prune(root)?; + } + capacity?; let path = root.join(format!( "{expiry}-{}-{}", uuid::Uuid::new_v4(), @@ -637,11 +1004,57 @@ fn save_managed_with_io( if std::time::Instant::now() >= deadline { return Err("Managed save expired".into()); } + #[cfg(test)] + if let Some(registry) = registry { + let gate = registry.1.lock().unwrap().save.take(); + gated_work(gate); + } + if let Some(consent) = consent { + let metadata = private_metadata(&path, false)?; + let opened = file.metadata().map_err(|e| e.to_string())?; + if file_identity(&metadata) != file_identity(&opened) + || metadata.len() != bytes.len() as u64 + { + return Err("Managed file changed before consent".into()); + } + let record = Consent { + root: root.to_owned(), + saver: id.into(), + owner: consent.owner.clone(), + hash: hash.into(), + size: bytes.len() as u64, + identity: file_identity(&opened), + }; + consent + .identity + .try_commit_managed_consent(&consent.owner, || { + let mut records = consent + .registry + .0 + .try_lock() + .map_err(|_| "Managed consent is busy")?; + if records.len() >= ROOT_FILES || std::time::Instant::now() >= deadline { + return Err("Managed consent expired or full".into()); + } + consent + .eligibility + .compare_exchange( + 0, + 1, + std::sync::atomic::Ordering::SeqCst, + std::sync::atomic::Ordering::SeqCst, + ) + .map_err(|_| "Managed save cancelled")?; + records.insert(path.clone(), record); + Ok(()) + })?; + } Ok(SavedBytes { path: path_text, size: bytes.len(), sha256: hash.into(), expires_at: expiry, + release_after_send: consent.map(|_| true), }) })(); drop(file); diff --git a/src-tauri/src/host_files/managed_tests.rs b/src-tauri/src/host_files/managed_tests.rs index f707eeef63..8bc9ae631a 100644 --- a/src-tauri/src/host_files/managed_tests.rs +++ b/src-tauri/src/host_files/managed_tests.rs @@ -6,6 +6,13 @@ use tauri::{ Manager as _, }; +// Capture the actual command future on a test-owned runtime for abort/paused-time +// coverage, without changing the production command or inventing a cancel IPC. +type CommandControl = ( + tokio::runtime::Handle, + std::sync::Arc>>>, +); + pub(crate) struct Fixture { pub(crate) temp: tempfile::TempDir, pub(crate) app: tauri::App, @@ -15,6 +22,21 @@ pub(crate) struct Fixture { } impl Fixture { pub(crate) fn new(managed: bool, files: bool) -> Self { + Self::configured( + managed, + files, + crate::identity::IdentityHost::fixture(), + ManagedFiles::default(), + None, + ) + } + fn configured( + managed: bool, + files: bool, + identity: crate::identity::IdentityHost, + registry: ManagedFiles, + control: Option, + ) -> Self { let temp = tempfile::tempdir().unwrap(); let base = dunce::canonicalize(temp.path()).unwrap(); let manager = @@ -43,7 +65,39 @@ impl Fixture { let app = mock_builder() .manage(PluginManager(Ok(manager.clone()))) .manage(crate::host_process::HostProcesses::default()) - .invoke_handler(crate::commands()) + .manage(registry) + .manage(identity) + .invoke_handler(move |invoke| { + if let Some((runtime, task)) = &control { + if invoke.message.command() == "plugin_host_save_bytes" { + use tauri::ipc::CommandArg as _; + let job = runtime.spawn(async move { + let request = + tauri::ipc::Request::from_command(tauri::ipc::CommandItem { + plugin: None, + name: "plugin_host_save_bytes", + key: "request", + message: &invoke.message, + acl: &invoke.acl, + }) + .unwrap(); + let app = invoke.message.webview().app_handle().clone(); + let result = plugin_host_save_bytes( + app.clone(), + app.state::(), + request, + ) + .await; + invoke + .resolver + .respond(result.map_err(tauri::ipc::InvokeError::from)); + }); + *task.lock().unwrap() = Some(job); + return true; + } + } + crate::commands()(invoke) + }) .build(context) .unwrap(); let view = tauri::WebviewWindowBuilder::new(&app, "main", Default::default()) @@ -64,12 +118,12 @@ impl Fixture { .unwrap() .join("plugin-artifacts") } - pub(crate) fn ipc( + fn start_ipc( &self, command: &str, body: tauri::ipc::InvokeBody, headers: reqwest::header::HeaderMap, - ) -> Result { + ) -> std::sync::mpsc::Receiver { let (reply, result) = std::sync::mpsc::sync_channel(1); self.view.as_ref().clone().on_message( tauri::webview::InvokeRequest { @@ -85,6 +139,15 @@ impl Fixture { let _ = reply.send(response); }), ); + result + } + pub(crate) fn ipc( + &self, + command: &str, + body: tauri::ipc::InvokeBody, + headers: reqwest::header::HeaderMap, + ) -> Result { + let result = self.start_ipc(command, body, headers); match result .recv_timeout(std::time::Duration::from_secs(5)) .expect("native fixture IPC watchdog") @@ -94,7 +157,15 @@ impl Fixture { } } pub(crate) fn save(&self, bytes: &[u8], hash: &str) -> Result { - let headers = reqwest::header::HeaderMap::from_iter([ + self.save_policy(bytes, hash, false) + } + pub(crate) fn save_policy( + &self, + bytes: &[u8], + hash: &str, + release: bool, + ) -> Result { + let mut headers = reqwest::header::HeaderMap::from_iter([ ( reqwest::header::HeaderName::from_static("x-buzz-plugin-id"), "example.bytes".parse().unwrap(), @@ -108,6 +179,9 @@ impl Fixture { hash.parse().unwrap(), ), ]); + if release { + headers.insert("x-buzz-release-after-send", "1".parse().unwrap()); + } self.ipc( "plugin_host_save_bytes", tauri::ipc::InvokeBody::Raw(bytes.to_vec()), @@ -236,26 +310,30 @@ fn invalid_input_hash_mismatch_and_deadline_leave_no_published_file() { .unwrap_err() .contains("mismatch")); assert_eq!(std::fs::read_dir(f.root()).unwrap().count(), 1); // Only ownership lock. - let result = save_managed( - &f.root(), - b"data", - &hash, - std::time::Instant::now(), - &f.manager, - "example.bytes", - &f.revision, - ); + let result = save_managed(ManagedSave { + root: &f.root(), + bytes: b"data", + hash: &hash, + deadline: std::time::Instant::now(), + manager: &f.manager, + id: "example.bytes", + revision: &f.revision, + registry: None, + consent: None, + }); assert!(result.err().unwrap().contains("expired")); assert_eq!(std::fs::read_dir(f.root()).unwrap().count(), 1); - assert!(save_managed( - &f.root(), - b"data", - &hash, - std::time::Instant::now() + SAVE_WINDOW, - &f.manager, - "example.bytes", - "stale" - ) + assert!(save_managed(ManagedSave { + root: &f.root(), + bytes: b"data", + hash: &hash, + deadline: std::time::Instant::now() + SAVE_WINDOW, + manager: &f.manager, + id: "example.bytes", + revision: "stale", + registry: None, + consent: None, + }) .is_err()); } @@ -526,6 +604,8 @@ fn save_failure_at(boundary: SaveBoundary) { manager: &manager, id: "example.bytes", revision: &revision, + registry: None, + consent: None, }, |file, bytes| { if boundary == SaveBoundary::BeforeWrite { @@ -685,7 +765,7 @@ fn own_managed_delete_refuses_other_plugins_legacy_paths_and_active_reads() { ) .unwrap_err(); assert!(error.contains("not owned")); - assert!(delete_managed(&f.root(), path, &f.manager, "example.bytes", "stale").is_err()); + assert!(delete_managed(&f.root(), path, &f.manager, "example.bytes", "stale", None).is_err()); assert!(remove( f.temp.path().join("outside").to_str().unwrap(), @@ -840,11 +920,16 @@ fn process_attachment_reader_blocks_delete_until_output_and_killed_child_complet .await .unwrap(); // cat is held on stdin; its actual monitor owns the admitted reader lease. - assert!( - delete_managed(&f.root(), path, &f.manager, "example.bytes", &f.revision) - .unwrap_err() - .contains("in use") - ); + assert!(delete_managed( + &f.root(), + path, + &f.manager, + "example.bytes", + &f.revision, + None + ) + .unwrap_err() + .contains("in use")); crate::host_process::plugin_host_process_write( f.app.state(), "example.bytes".into(), @@ -871,7 +956,15 @@ fn process_attachment_reader_blocks_delete_until_output_and_killed_child_complet break; } } - delete_managed(&f.root(), path, &f.manager, "example.bytes", &f.revision).unwrap(); + delete_managed( + &f.root(), + path, + &f.manager, + "example.bytes", + &f.revision, + None, + ) + .unwrap(); }); } @@ -915,6 +1008,8 @@ fn managed_deletes_queue_behind_a_held_save_and_each_other() { manager: &manager, id: "example.bytes", revision: &revision, + registry: None, + consent: None, }, |file, bytes| { file.write_all(bytes)?; @@ -999,3 +1094,835 @@ fn queued_managed_delete_timeout_never_dispatches_or_reports_unknown_outcome() { assert_eq!(BYTE_WAITERS.available_permits(), 16); }); } + +#[test] +fn host_release_reader_admission_refuses_while_exclusion_is_held() { + let _serial = commands(); + let exclusion = FileReadExclusion::new().unwrap(); + let (send, result) = std::sync::mpsc::channel(); + let worker = std::thread::spawn(move || { + send.send(FileReadLease::new().map(|lease| { + drop(lease); + })) + .unwrap(); + }); + let refused = result.recv_timeout(std::time::Duration::from_secs(1)); + drop(exclusion); + worker.join().unwrap(); + assert!(refused.is_ok(), "read admission blocked on exclusion"); + assert_eq!( + refused.unwrap().unwrap_err(), + "File reads are busy; try again later" + ); +} + +#[test] +fn host_release_raw_save_requires_strict_flag_and_native_ack() { + let _serial = commands(); + let f = Fixture::new(true, false); + let bytes = b"temporary"; + let headers = |value: &str| { + reqwest::header::HeaderMap::from_iter([ + ( + reqwest::header::HeaderName::from_static("x-buzz-plugin-id"), + "example.bytes".parse().unwrap(), + ), + ( + reqwest::header::HeaderName::from_static("x-buzz-plugin-revision"), + f.revision.parse().unwrap(), + ), + ( + reqwest::header::HeaderName::from_static("x-buzz-sha256"), + digest(bytes).parse().unwrap(), + ), + ( + reqwest::header::HeaderName::from_static("x-buzz-release-after-send"), + value.parse().unwrap(), + ), + ]) + }; + assert!(f + .ipc( + "plugin_host_save_bytes", + tauri::ipc::InvokeBody::Raw(bytes.to_vec()), + headers("true") + ) + .is_err()); + let saved: serde_json::Value = f + .ipc( + "plugin_host_save_bytes", + tauri::ipc::InvokeBody::Raw(bytes.to_vec()), + headers("1"), + ) + .unwrap() + .deserialize() + .unwrap(); + assert_eq!(saved["releaseAfterSend"], true); +} + +#[test] +fn host_release_confirmed_attachment_reclaims_only_native_consented_exact_file() { + let _serial = commands(); + let f = Fixture::new(true, false); + let bytes = b"temporary"; + let hash = digest(bytes); + let headers = reqwest::header::HeaderMap::from_iter([ + ( + reqwest::header::HeaderName::from_static("x-buzz-plugin-id"), + "example.bytes".parse().unwrap(), + ), + ( + reqwest::header::HeaderName::from_static("x-buzz-plugin-revision"), + f.revision.parse().unwrap(), + ), + ( + reqwest::header::HeaderName::from_static("x-buzz-sha256"), + hash.parse().unwrap(), + ), + ( + reqwest::header::HeaderName::from_static("x-buzz-release-after-send"), + "1".parse().unwrap(), + ), + ]); + let saved: serde_json::Value = f + .ipc( + "plugin_host_save_bytes", + tauri::ipc::InvokeBody::Raw(bytes.to_vec()), + headers, + ) + .unwrap() + .deserialize() + .unwrap(); + let sibling = f.save(bytes, &hash).unwrap(); + let path = saved["path"].as_str().unwrap(); + let owner = + tauri::async_runtime::block_on(f.app.state::().viewer()) + .unwrap(); + let candidate = serde_json::json!({"path":path, "sha256":hash, "size":bytes.len(), "url":"https://example.test/media"}); + let signed = serde_json::json!({"kind":9,"tags":[["imeta", "url https://example.test/media", format!("x {hash}"),format!("size {}", bytes.len())]]}); + tauri::async_runtime::block_on(release_after_send( + f.app.state::().inner().clone(), + f.root(), + owner, + signed, + Some(serde_json::json!([candidate])), + )); + assert!( + !Path::new(path).exists(), + "confirmed matching source must release" + ); + assert!( + Path::new(sibling["path"].as_str().unwrap()).exists(), + "ordinary saves never adopt consent" + ); +} + +struct ReleaseGate(Option>); +impl Drop for ReleaseGate { + fn drop(&mut self) { + if let Some(release) = self.0.take() { + let _ = release.send(()); + } + } +} +fn managed_gate() -> ( + tokio::sync::oneshot::Sender<()>, + tokio::sync::oneshot::Receiver<()>, + ReleaseGate, + std::sync::mpsc::Receiver<()>, +) { + let (entered, ready) = tokio::sync::oneshot::channel(); + let (release, allowed) = std::sync::mpsc::channel(); + (entered, ready, ReleaseGate(Some(release)), allowed) +} +fn start_release_save(f: &Fixture) -> std::sync::mpsc::Receiver { + let headers = reqwest::header::HeaderMap::from_iter([ + ( + reqwest::header::HeaderName::from_static("x-buzz-plugin-id"), + "example.bytes".parse().unwrap(), + ), + ( + reqwest::header::HeaderName::from_static("x-buzz-plugin-revision"), + f.revision.parse().unwrap(), + ), + ( + reqwest::header::HeaderName::from_static("x-buzz-sha256"), + digest(b"fenced").parse().unwrap(), + ), + ( + reqwest::header::HeaderName::from_static("x-buzz-release-after-send"), + "1".parse().unwrap(), + ), + ]); + f.start_ipc( + "plugin_host_save_bytes", + tauri::ipc::InvokeBody::Raw(b"fenced".to_vec()), + headers, + ) +} + +#[test] +fn host_release_actual_command_timeout_and_abort_fence_dispatched_save() { + let _serial = commands(); + for abort in [false, true] { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let registry = ManagedFiles::default(); + let task = std::sync::Arc::new(std::sync::Mutex::new(None)); + let identity = crate::identity::IdentityHost::fixture(); + let f = Fixture::configured( + true, + false, + identity, + registry.clone(), + Some((tokio::runtime::Handle::current(), task.clone())), + ); + let (entered, ready, release, allowed) = managed_gate(); + registry.1.lock().unwrap().save = Some((entered, allowed)); + let response = start_release_save(&f); + ready.await.unwrap(); + let job = task.lock().unwrap().take().unwrap(); + assert_eq!(BYTE_WORK.available_permits(), 0); + if abort { + job.abort(); + assert!(job.await.unwrap_err().is_cancelled()); + } else { + tokio::time::pause(); + tokio::time::advance(SAVE_WINDOW + std::time::Duration::from_secs(1)).await; + job.await.unwrap(); + tokio::time::resume(); + assert!(matches!( + response.try_recv().unwrap(), + tauri::ipc::InvokeResponse::Err(_) + )); + } + assert!(registry.0.lock().unwrap().is_empty()); + assert_eq!( + BYTE_WORK.available_permits(), + 0, + "cancelled waiter must not release actual writer admission" + ); + drop(release); + let permit = byte_admission(tokio::time::Instant::now() + SAVE_WINDOW) + .await + .unwrap(); + assert!(registry.0.lock().unwrap().is_empty()); + assert_eq!( + std::fs::read_dir(f.root()).unwrap().count(), + 1, + "precommit cancellation rolls back file" + ); + drop(permit); + }); + } +} + +#[test] +fn host_release_identity_close_while_queued_or_at_commit_refuses_without_reassignment() { + let _serial = commands(); + for queued in [true, false] { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let registry = ManagedFiles::default(); + let identity = crate::identity::IdentityHost::fixture(); + let task = std::sync::Arc::new(std::sync::Mutex::new(None)); + let f = Fixture::configured( + true, + false, + identity.clone(), + registry.clone(), + Some((tokio::runtime::Handle::current(), task.clone())), + ); + let held = if queued { + Some(BYTE_WORK.acquire().await.unwrap()) + } else { + None + }; + let (entered, ready, release, allowed) = managed_gate(); + if queued { + registry.1.lock().unwrap().captured = Some(entered); + } else { + registry.1.lock().unwrap().save = Some((entered, allowed)); + } + let response = start_release_save(&f); + ready.await.unwrap(); + identity.close(); + drop(held); + drop(release); + let job = task.lock().unwrap().take().unwrap(); + job.await.unwrap(); + assert!(matches!( + response.try_recv().unwrap(), + tauri::ipc::InvokeResponse::Err(_) + )); + let permit = byte_admission(tokio::time::Instant::now() + SAVE_WINDOW) + .await + .unwrap(); + assert!(registry.0.lock().unwrap().is_empty()); + assert_eq!(std::fs::read_dir(f.root()).unwrap().count(), 1); + drop(permit); + let ordinary = f.start_ipc( + "plugin_host_save_bytes", + tauri::ipc::InvokeBody::Raw(b"ordinary".to_vec()), + reqwest::header::HeaderMap::from_iter([ + ( + reqwest::header::HeaderName::from_static("x-buzz-plugin-id"), + "example.bytes".parse().unwrap(), + ), + ( + reqwest::header::HeaderName::from_static("x-buzz-plugin-revision"), + f.revision.parse().unwrap(), + ), + ( + reqwest::header::HeaderName::from_static("x-buzz-sha256"), + digest(b"ordinary").parse().unwrap(), + ), + ]), + ); + let job = task.lock().unwrap().take().unwrap(); + job.await.unwrap(); + assert!( + matches!( + ordinary.try_recv().unwrap(), + tauri::ipc::InvokeResponse::Ok(_) + ), + "ordinary saves remain identity independent" + ); + }); + } +} + +pub(crate) fn release_fixture_data( + f: &Fixture, + bytes: &[u8], +) -> (serde_json::Value, serde_json::Value, String) { + let hash = digest(bytes); + let saved = f.save_policy(bytes, &hash, true).unwrap(); + let hints = serde_json::json!([{"path":saved["path"],"sha256":hash,"size":bytes.len(),"url":"https://example.test/media"}]); + let event = serde_json::json!({"kind":9,"tags":[["imeta","url https://example.test/media",format!("x {hash}"),format!("size {}",bytes.len())]]}); + let owner = + tauri::async_runtime::block_on(f.app.state::().viewer()) + .unwrap(); + (hints, event, owner) +} + +#[test] +fn host_release_reclaims_more_than_sixteen_sequential_saves_and_byte_capacity() { + let _serial = commands(); + let f = Fixture::new(true, false); + let registry = f.app.state::().inner().clone(); + for _ in 0..20 { + let (hints, event, owner) = release_fixture_data(&f, b"one send"); + let path = hints[0]["path"].as_str().unwrap().to_owned(); + tauri::async_runtime::block_on(release_after_send( + registry.clone(), + f.root(), + owner, + event, + Some(hints), + )); + assert!(!Path::new(&path).exists()); + assert!(registry.0.lock().unwrap().is_empty()); + } + let bytes = vec![7; MAX_SAVE_BYTES]; + let mut batch = vec![]; + let mut event = serde_json::json!({"kind":9,"tags":[]}); + let mut owner = String::new(); + for _ in 0..4 { + let (hints, signed, key) = release_fixture_data(&f, &bytes); + batch.push(hints[0].clone()); + event["tags"] + .as_array_mut() + .unwrap() + .push(signed["tags"][0].clone()); + owner = key; + } + assert!(f.save(b"over capacity", &digest(b"over capacity")).is_err()); + tauri::async_runtime::block_on(release_after_send( + registry.clone(), + f.root(), + owner, + event, + Some(serde_json::json!(batch)), + )); + assert!(registry.0.lock().unwrap().is_empty()); + f.save(&bytes, &digest(&bytes)).unwrap(); +} + +#[test] +fn host_release_refusal_is_not_an_expiry_timer_and_restart_never_adopts_consent() { + let _serial = commands(); + let f = Fixture::new(true, false); + let registry = f.app.state::().inner().clone(); + let (hints, event, owner) = release_fixture_data(&f, b"retained"); + let path = PathBuf::from(hints[0]["path"].as_str().unwrap()); + let lease = FileReadLease::new().unwrap(); + tauri::async_runtime::block_on(release_after_send( + registry.clone(), + f.root(), + owner.clone(), + event.clone(), + Some(hints.clone()), + )); + assert!(path.exists()); + drop(lease); + tauri::async_runtime::block_on(release_after_send( + ManagedFiles::default(), + f.root(), + owner, + event, + Some(hints), + )); + assert!(path.exists(), "restart map must not adopt consent"); + let expiry = artifact_expiry(path.file_name().unwrap().to_str().unwrap()).unwrap(); + let _future = expiry + 1; // No work is scheduled when this clock boundary passes. + assert!(path.exists()); + assert_eq!(registry.0.lock().unwrap().len(), 1); + let lock = managed_lock(&f.root()).unwrap(); + managed_capacity(&f.root(), expiry + 1, 1).unwrap(); + registry.prune(&f.root()).unwrap(); + drop(lock); + assert!(!path.exists()); + assert!(registry.0.lock().unwrap().is_empty()); +} + +#[test] +fn host_release_stalled_hash_has_one_budget_and_prompt_direct_and_process_read_refusal() { + let _serial = commands(); + for count in [1, ROOT_FILES] { + let f = Fixture::new(true, true); + let registry = f.app.state::().inner().clone(); + let mut candidates = vec![]; + let mut signed = serde_json::json!({"kind":9,"tags":[]}); + let mut owner = String::new(); + for _ in 0..count { + let (hints, event, key) = release_fixture_data(&f, b"hash gate"); + candidates.push(hints[0].clone()); + signed["tags"] + .as_array_mut() + .unwrap() + .push(event["tags"][0].clone()); + owner = key; + } + let path = candidates[0]["path"].as_str().unwrap().to_owned(); + let root = f.root(); + let (entered, ready, release, allowed) = managed_gate(); + registry.1.lock().unwrap().hash = Some((entered, allowed)); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let cleanup = tokio::spawn(release_after_send(registry.clone(), root, owner, signed, Some(serde_json::json!(candidates)))); + ready.await.unwrap(); + let direct = plugin_host_read_file(f.app.state::(), "example.bytes".into(), f.revision.clone(), path.clone(), None); + assert!(matches!(futures_util::poll!(Box::pin(direct)), std::task::Poll::Ready(Err(error)) if error == "File reads are busy; try again later")); + for process in ["read", "read-protected"] { + let reader = crate::host_process::plugin_host_process_spawn( + f.app.handle().clone(), f.app.state::(), f.app.state::(), + "example.bytes".into(), f.revision.clone(), process.into(), vec![path.clone()], None, None, + tauri::ipc::Channel::new(|_| panic!("refused process emitted output")), None, None, + ); + assert!(matches!(futures_util::poll!(Box::pin(reader)), std::task::Poll::Ready(Err(error)) if error == "File reads are busy; try again later")); + } + tokio::time::pause(); + let timer = tokio::spawn(async { tokio::time::sleep(std::time::Duration::from_secs(1)).await; }); + tokio::time::advance(SAVE_WINDOW + std::time::Duration::from_secs(1)).await; + timer.await.unwrap(); + cleanup.await.unwrap(); + tokio::time::resume(); + assert!(Path::new(&path).exists()); + assert_eq!(BYTE_WORK.available_permits(), 0); + assert!(FileReadExclusion::new().is_err(), "actual hash still owns exclusion after waiter timeout"); + let mut writer = Box::pin(byte_admission(tokio::time::Instant::now() + SAVE_WINDOW)); + assert!(matches!(futures_util::poll!(&mut writer), std::task::Poll::Pending)); + drop(release); + let permit = writer.await.unwrap(); + assert!(!Path::new(&path).exists()); + // Undispatched later candidates stop at the same elapsed budget. + if count > 1 { assert_eq!(registry.0.lock().unwrap().len(), count - 1); } + drop(permit); + assert!(FileReadLease::new().is_ok()); + }); + } +} + +#[test] +fn host_release_rejects_forged_owner_content_paths_metadata_and_replaced_files() { + let _serial = commands(); + let cases = [ + "owner", + "hash", + "size", + "url", + "split-imeta", + "kind", + "outside", + "alias", + "relative", + "unknown-field", + "oversized", + "replaced", + "content", + "hardlink", + "symlink", + "nonprivate", + ]; + for case in cases { + let f = Fixture::new(true, false); + let registry = f.app.state::().inner().clone(); + let (mut hints, mut signed, mut owner) = release_fixture_data(&f, b"private source"); + let path = PathBuf::from(hints[0]["path"].as_str().unwrap()); + let mut held_original = None; + match case { + "owner" => owner = "b".repeat(64), + "hash" => hints[0]["sha256"] = "0".repeat(64).into(), + "size" => hints[0]["size"] = 1.into(), + "url" => hints[0]["url"] = "https://different.test/media".into(), + "split-imeta" => { + signed["tags"] = serde_json::json!([ + ["imeta", "url https://example.test/media"], + [ + "imeta", + format!("x {}", digest(b"private source")), + "size 14" + ] + ]) + } + "kind" => signed["kind"] = 1.into(), + "outside" => hints[0]["path"] = f.temp.path().join("outside").to_str().unwrap().into(), + "alias" => { + hints[0]["path"] = format!( + "{}/../plugin-artifacts/{}", + f.root().display(), + path.file_name().unwrap().to_str().unwrap() + ) + .into() + } + "relative" => hints[0]["path"] = path.file_name().unwrap().to_str().unwrap().into(), + "unknown-field" => hints[0]["owner"] = owner.clone().into(), + "oversized" => hints = serde_json::json!(vec![hints[0].clone(); ROOT_FILES + 1]), + "replaced" => { + held_original = Some(std::fs::File::open(&path).unwrap()); + std::fs::remove_file(&path).unwrap(); + use std::io::Write as _; + private_options() + .create_new(true) + .open(&path) + .unwrap() + .write_all(b"private source") + .unwrap(); + } + "content" => std::fs::write(&path, b"changed source").unwrap(), + #[cfg(unix)] + "hardlink" => std::fs::hard_link(&path, f.temp.path().join("alias")).unwrap(), + #[cfg(unix)] + "symlink" => { + std::fs::rename(&path, f.temp.path().join("original")).unwrap(); + std::os::unix::fs::symlink(f.temp.path().join("original"), &path).unwrap(); + } + #[cfg(unix)] + "nonprivate" => { + use std::os::unix::fs::PermissionsExt as _; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap(); + } + _ => continue, + } + tauri::async_runtime::block_on(release_after_send( + registry.clone(), + f.root(), + owner, + signed, + Some(hints), + )); + assert!(path.exists(), "{case} must not authorize deletion"); + assert_eq!( + registry.0.lock().unwrap().len(), + 1, + "{case} must not discard consent" + ); + drop(held_original); + } +} + +#[test] +fn host_release_committed_consent_survives_saver_disable_and_later_identity_close() { + let _serial = commands(); + let f = Fixture::new(true, false); + let registry = f.app.state::().inner().clone(); + let (hints, event, owner) = release_fixture_data(&f, b"committed"); + let path = hints[0]["path"].as_str().unwrap().to_owned(); + f.manager.change("disable", "example.bytes").unwrap(); + f.app.state::().close(); + tauri::async_runtime::block_on(release_after_send( + registry, + f.root(), + owner, + event, + Some(hints), + )); + assert!( + !Path::new(&path).exists(), + "committed owner is not current UI identity or runtime plugin grant" + ); +} + +#[test] +fn host_release_parallel_reader_counter_never_refuses_without_cleanup_exclusion() { + let _serial = commands(); + let start = std::sync::Arc::new(std::sync::Barrier::new(8)); + let mut jobs = vec![]; + for _ in 0..8 { + let start = start.clone(); + jobs.push(std::thread::spawn(move || { + start.wait(); + for _ in 0..10_000 { + let first = FileReadLease::new() + .expect("ordinary read contention is not cleanup exclusion"); + let second = FileReadLease::new().expect("parallel readers must coexist"); + drop(first); + drop(second); + } + })); + } + let results: Vec<_> = jobs.into_iter().map(|job| job.join()).collect(); + assert!(results.iter().all(Result::is_ok)); +} + +#[test] +fn host_release_reader_and_exclusion_races_are_mutually_exclusive_and_unwind_safe() { + let _serial = commands(); + for reader_wins in [true, false] { + let (entered, ready) = std::sync::mpsc::channel(); + let (release, allowed) = std::sync::mpsc::channel(); + let worker = std::thread::spawn(move || { + if reader_wins { + let _lease = FileReadLease::new().unwrap(); + entered.send(()).unwrap(); + allowed.recv().unwrap(); + } else { + let _exclusion = FileReadExclusion::new().unwrap(); + entered.send(()).unwrap(); + allowed.recv().unwrap(); + } + }); + ready.recv().unwrap(); + if reader_wins { + assert!(FileReadExclusion::new().is_err()); + } else { + assert!(FileReadLease::new().is_err()); + } + release.send(()).unwrap(); + worker.join().unwrap(); + assert_eq!(FILE_READS.load(std::sync::atomic::Ordering::SeqCst), 0); + } + for reader in [true, false] { + let panic = std::panic::catch_unwind(|| { + if reader { + let _lease = FileReadLease::new().unwrap(); + panic!("reader unwind"); + } else { + let _exclusion = FileReadExclusion::new().unwrap(); + panic!("writer unwind"); + } + }); + assert!(panic.is_err()); + assert_eq!(FILE_READS.load(std::sync::atomic::Ordering::SeqCst), 0); + assert!(FileReadExclusion::new().is_ok()); + assert!(FileReadLease::new().is_ok()); + } +} + +#[test] +fn host_release_duplicate_parallel_and_public_delete_paths_are_single_use() { + let _serial = commands(); + let f = Fixture::new(true, false); + let registry = f.app.state::().inner().clone(); + let (hints, signed, owner) = release_fixture_data(&f, b"once"); + let path = hints[0]["path"].as_str().unwrap().to_owned(); + let duplicates = serde_json::json!([hints[0].clone(), hints[0].clone()]); + assert_eq!( + confirmed_candidates(&signed, Some(duplicates.clone())).len(), + 1 + ); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let first = release_after_send( + registry.clone(), + f.root(), + owner.clone(), + signed.clone(), + Some(duplicates), + ); + let second = release_after_send( + registry.clone(), + f.root(), + owner.clone(), + signed.clone(), + Some(hints.clone()), + ); + tokio::join!(first, second); + assert!(!Path::new(&path).exists()); + assert!(read(&path, None).unwrap_err().contains("No such file")); + assert!(registry.0.lock().unwrap().is_empty()); + assert!( + delete_managed( + &f.root(), + &path, + &f.manager, + "example.bytes", + &f.revision, + Some(®istry) + ) + .is_err(), + "public missing-file behavior unchanged" + ); + }); + let (hints, signed, owner) = release_fixture_data(&f, b"public first"); + let path = hints[0]["path"].as_str().unwrap(); + delete_managed( + &f.root(), + path, + &f.manager, + "example.bytes", + &f.revision, + Some(®istry), + ) + .unwrap(); + assert!(registry.0.lock().unwrap().is_empty()); + tauri::async_runtime::block_on(release_after_send( + registry, + f.root(), + owner, + signed, + Some(hints), + )); +} + +#[test] +fn host_release_missing_consent_prunes_on_save_and_one_bad_candidate_keeps_later_safe_work() { + let _serial = commands(); + let f = Fixture::new(true, false); + let registry = f.app.state::().inner().clone(); + let (mut first, signed, owner) = release_fixture_data(&f, b"matching"); + let first_path = first[0]["path"].as_str().unwrap().to_owned(); + std::fs::remove_file(&first_path).unwrap(); + let (second, _, _) = release_fixture_data(&f, b"matching"); + assert_eq!( + registry.0.lock().unwrap().len(), + 1, + "new save prunes missing record" + ); + first[0]["path"] = first_path.into(); + let batch = serde_json::json!([first[0].clone(), second[0].clone()]); + tauri::async_runtime::block_on(release_after_send( + registry.clone(), + f.root(), + owner, + signed, + Some(batch), + )); + assert!(!Path::new(second[0]["path"].as_str().unwrap()).exists()); + assert!(registry.0.lock().unwrap().is_empty()); +} + +#[test] +fn host_release_identity_and_consent_busy_refuse_without_true_ack_or_adoption() { + let _serial = commands(); + let f = Fixture::new(true, false); + let registry = f.app.state::().inner().clone(); + let identity = f + .app + .state::() + .inner() + .clone(); + let owner = tauri::async_runtime::block_on(identity.viewer()).unwrap(); + let eligibility = SaveEligibility::default(); + let consent = SaveConsent { + identity, + owner, + registry: registry.clone(), + eligibility: eligibility.0.clone(), + }; + let held_map = registry.0.lock().unwrap(); + let result = save_managed(ManagedSave { + root: &f.root(), + bytes: b"busy", + hash: &digest(b"busy"), + deadline: std::time::Instant::now() + SAVE_WINDOW, + manager: &f.manager, + id: "example.bytes", + revision: &f.revision, + registry: None, + consent: Some(&consent), + }); + assert!(result.is_err()); + assert!(held_map.is_empty()); + drop(held_map); + assert_eq!(std::fs::read_dir(f.root()).unwrap().count(), 1); + let closed = Fixture::configured( + true, + false, + crate::identity::IdentityHost::fixture_without_key(false), + ManagedFiles::default(), + None, + ); + assert!(closed + .save_policy(b"closed", &digest(b"closed"), true) + .is_err()); + assert!( + !closed.root().exists(), + "unavailable initial identity refuses before creation" + ); + assert!(closed.save(b"ordinary", &digest(b"ordinary")).is_ok()); +} + +#[test] +fn host_release_lost_renderer_waiter_is_not_native_command_cancellation() { + let _serial = commands(); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let registry = ManagedFiles::default(); + let task = std::sync::Arc::new(std::sync::Mutex::new(None)); + let f = Fixture::configured( + true, + false, + crate::identity::IdentityHost::fixture(), + registry.clone(), + Some((tokio::runtime::Handle::current(), task.clone())), + ); + let (entered, ready, release, allowed) = managed_gate(); + registry.1.lock().unwrap().save = Some((entered, allowed)); + let reply = start_release_save(&f); + ready.await.unwrap(); + drop(reply); // The native future remains alive: no cancel IPC exists. + drop(release); + let job = task.lock().unwrap().take().unwrap(); + job.await.unwrap(); + assert_eq!( + registry.0.lock().unwrap().len(), + 1, + "lost renderer reply can still commit native consent" + ); + let path = registry.0.lock().unwrap().keys().next().unwrap().clone(); + assert!(path.exists()); + f.app.state::().close(); + assert_eq!( + registry.0.lock().unwrap().len(), + 1, + "later identity close does not erase or reassign committed consent" + ); + }); +} diff --git a/src-tauri/src/identity.rs b/src-tauri/src/identity.rs index 06ec81f34b..2b95c94b46 100644 --- a/src-tauri/src/identity.rs +++ b/src-tauri/src/identity.rs @@ -774,6 +774,23 @@ impl IdentityHost { .ok_or_else(|| "Set up your identity first".into()) } + /// Short creation-time consent fence. Never restores secure storage or waits. + /// Close uses this same mutex; callers must not perform I/O in `commit`. + pub(crate) fn try_commit_managed_consent( + &self, + expected: &str, + commit: impl FnOnce() -> Result, + ) -> Result { + let identity = self + .0 + .try_lock() + .map_err(|_| "Managed save identity is busy")?; + match &identity.state { + State::Ready(key) if key.viewer()? == expected => commit(), + _ => Err("Managed save identity is unavailable or changed".into()), + } + } + pub(crate) async fn require_missing(&self) -> Result<()> { with_identity(self.clone(), |identity| identity.require_missing()).await } diff --git a/src-tauri/src/identity/tests.rs b/src-tauri/src/identity/tests.rs index 1f693e8348..9e7306639f 100644 --- a/src-tauri/src/identity/tests.rs +++ b/src-tauri/src/identity/tests.rs @@ -770,3 +770,29 @@ async fn main_agent_signer_admits_one_agent_or_a_cleared_choice() { .is_err()); } } + +#[tokio::test] +async fn managed_consent_commit_is_fenced_by_identity_close_and_never_restores() { + let host = IdentityHost::fixture(); + let owner = host.viewer().await.unwrap(); + assert_eq!( + host.try_commit_managed_consent(&owner, || Ok(7)).unwrap(), + 7 + ); + assert!(host + .try_commit_managed_consent(&"b".repeat(64), || Ok(())) + .is_err()); + let guard = host.0.lock().unwrap(); + assert!(host.try_commit_managed_consent(&owner, || Ok(())).is_err()); + drop(guard); + host.close(); + assert!(host + .try_commit_managed_consent::<()>(&owner, || panic!("closed identity committed")) + .is_err()); + for readable in [true, false] { + let host = IdentityHost::fixture_without_key(readable); + assert!(host + .try_commit_managed_consent::<()>(&owner, || panic!("missing identity committed")) + .is_err()); + } +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index dc6a17b91c..000ba2b588 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -1063,6 +1063,7 @@ pub fn run() { .manage(Terminals::default()) .manage(HostCalls::default()) .manage(HostProcesses::default()) + .manage(host_files::ManagedFiles::default()) .manage(HostMcpServers::default()) .manage(OAuthCallbackHost::default()) .manage(NsecBackupHost::default().with_federated_identity(federated_identity)) diff --git a/src/buzz-mcp/app-client.test.ts b/src/buzz-mcp/app-client.test.ts index 9a1f7e17b3..50f189f1ab 100644 --- a/src/buzz-mcp/app-client.test.ts +++ b/src/buzz-mcp/app-client.test.ts @@ -308,3 +308,53 @@ it("proposes no picture for a turn stopped during its upload", async () => { release(); await expect(picture).rejects.toThrow("Turn ended"); }); + +it.each(["direct", "process"])( + "forwards %s reader attachment hints through the shared handle without losing its check", + async (reader) => { + const publish = vi.fn().mockResolvedValue({ id: "confirmed" }); + const spawn = vi.fn>(async (_id, options) => { + options?.onStdout?.(btoa("bytes")); + return { + write: vi.fn(), + end: vi.fn(), + kill: vi.fn(), + exited: Promise.resolve(0), + }; + }); + const client = appClient( + { pubkey: "agent", publish } as unknown as AgentHandle, + { + read: + reader === "direct" + ? async () => new TextEncoder().encode("bytes") + : (path) => readFile(spawn, "/work", path), + memories: async () => ({ entries: [], partial: false }), + }, + ); + await expect(client.read("file")).resolves.toEqual( + new TextEncoder().encode("bytes"), + ); + const event = { kind: 9, content: "attachment", tags: [] }; + const options = { + managedAttachments: [ + { + path: "/saved/file", + sha256: "a".repeat(64), + size: 5, + url: "https://example.test/media", + }, + ], + }; + await expect(client.publish(event, options)).resolves.toEqual({ + id: "confirmed", + }); + expect(publish).toHaveBeenCalledExactlyOnceWith( + event, + expect.any(Function), + options, + ); + await client.publish(event); + expect(publish).toHaveBeenLastCalledWith(event, expect.any(Function)); + }, +); diff --git a/src/buzz-mcp/app-client.ts b/src/buzz-mcp/app-client.ts index 63664ba8e5..7f40107874 100644 --- a/src/buzz-mcp/app-client.ts +++ b/src/buzz-mcp/app-client.ts @@ -123,9 +123,11 @@ export function appClient( check(); return agent.searchProfiles(query, limit); }, - publish: (event) => { + publish: (event, options) => { check(); - return agent.publish(event, check); + return options + ? agent.publish(event, check, options) + : agent.publish(event, check); }, upload: (data, mime) => { check(); diff --git a/src/buzz-mcp/client.ts b/src/buzz-mcp/client.ts index 22065ce1a5..f62492fd88 100644 --- a/src/buzz-mcp/client.ts +++ b/src/buzz-mcp/client.ts @@ -19,6 +19,17 @@ export type Template = Readonly<{ content: string; tags: readonly (readonly string[])[]; }>; +/** Trusted app read/upload hints, never independent deletion authority. + * Native requires creation-time consent and exact confirmed event/file matches. */ +export type ManagedAttachment = Readonly<{ + path: string; + sha256: string; + size: number; + url: string; +}>; +export type PublishOptions = Readonly<{ + managedAttachments?: readonly ManagedAttachment[]; +}>; export type Filter = Readonly>; /** The community's description of an uploaded blob. */ export type Upload = Readonly<{ @@ -129,7 +140,7 @@ export interface BuzzClient { limit: number, ): Promise; /** Signs `event` as the agent and posts it. */ - publish(event: Template): Promise; + publish(event: Template, options?: PublishOptions): Promise; /** Uploads a file; `mime` is a hint, and the community decides the type. */ upload(data: Uint8Array, mime: string): Promise; /** Uploads an image for an avatar, without its metadata, as the owner's diff --git a/src/buzz-mcp/tools.test.ts b/src/buzz-mcp/tools.test.ts index aae7ad69cb..348a650e47 100644 --- a/src/buzz-mcp/tools.test.ts +++ b/src/buzz-mcp/tools.test.ts @@ -4300,3 +4300,189 @@ it("saves a canvas of exactly the live limit in UTF-8 bytes, and refuses one byt ).rejects.toThrow(`over its limit of ${LIVE_CANVAS_BYTES} bytes`); expect(published).toHaveLength(1); }); + +it.each(["send", "reply_in_subthread"])( + "%s carries exact snapshot-matching attachment hints only out of band", + async (name) => { + const { client, published } = fake(community); + const hash = createHash("sha256").update(PNG).digest("hex"); + vi.mocked(client.upload).mockResolvedValue({ + url: "https://relay/media/x.png", + sha256: hash, + size: PNG.length, + type: "image/png", + }); + await callTool(client, { channel: "chan", root }, name, { + text: "attachment", + audience: "everyone", + ...(name === "send" ? { reply: nested } : { parent: nested }), + files: ["shot.png", "shot.png"], + }); + expect(client.publish).toHaveBeenLastCalledWith( + expect.objectContaining({ kind: 9 }), + { + managedAttachments: [ + { + path: "shot.png", + sha256: hash, + size: PNG.length, + url: "https://relay/media/x.png", + }, + ], + }, + ); + expect(published[0]?.tags).toContainEqual( + expect.arrayContaining(["imeta", `x ${hash}`, `size ${PNG.length}`]), + ); + }, +); + +it.each(["read", "upload", "body", "mentions", "emoji"])( + "never publishes release hints after %s fails, including a later attachment failure", + async (boundary) => { + const { client } = fake(community); + const hash = createHash("sha256").update(PNG).digest("hex"); + vi.mocked(client.upload).mockResolvedValue({ + url: "https://relay/media/x.png", + sha256: hash, + size: PNG.length, + type: "image/png", + }); + if (boundary === "read") + vi.mocked(client.read) + .mockResolvedValueOnce(PNG) + .mockRejectedValueOnce(new Error("read failed")); + if (boundary === "upload") + vi.mocked(client.upload) + .mockResolvedValueOnce({ + url: "https://relay/media/x.png", + sha256: hash, + size: PNG.length, + type: "image/png", + }) + .mockRejectedValueOnce(new Error("upload failed")); + if (boundary === "body") + vi.mocked(client.read).mockImplementation(async (path) => { + if (path === "body.txt") throw new Error("body failed"); + return PNG; + }); + if (boundary === "mentions" || boundary === "emoji") + vi.mocked(client.query).mockImplementation(async () => { + throw new Error(`${boundary} failed`); + }); + await expect( + callTool(client, { channel: "chan" }, "send", { + audience: "everyone", + files: ["one.png", "two.png"], + ...(boundary === "body" + ? { path: "body.txt" } + : { text: boundary === "emoji" ? ":sample:" : "attachment" }), + ...(boundary === "mentions" && { mentions: [bob] }), + }), + ).rejects.toThrow("failed"); + expect(client.publish).not.toHaveBeenCalled(); + }, +); + +it.each(["mismatch", "conflict", "overflow", "text-path"])( + "omits managed hints for %s without changing ordinary attachment publication", + async (kind) => { + const { client } = fake(community); + const hash = createHash("sha256").update(PNG).digest("hex"); + vi.mocked(client.upload).mockResolvedValue({ + url: "https://relay/media/x.png", + sha256: hash, + size: PNG.length, + type: "image/png", + }); + if (kind === "mismatch") + vi.mocked(client.upload).mockResolvedValue({ + url: "https://relay/media/x.png", + sha256: "forged", + size: PNG.length, + type: "image/png", + }); + if (kind === "conflict") { + vi.mocked(client.read) + .mockResolvedValueOnce(PNG) + .mockResolvedValueOnce(new Uint8Array([1, 2, 3])); + vi.mocked(client.upload).mockImplementation(async (data, type) => ({ + url: "https://relay/media/x.png", + sha256: createHash("sha256").update(data).digest("hex"), + size: data.length, + type, + })); + } + await callTool(client, { channel: "chan" }, "send", { + audience: "everyone", + ...(kind === "text-path" + ? { path: "notes.md" } + : { + text: "attachment", + files: + kind === "overflow" + ? Array.from({ length: 17 }, (_, i) => `${i}.png`) + : ["shot.png", "shot.png"], + }), + }); + expect(client.publish).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ kind: 9 }), + ); + }, +); + +it("holds publication until the actual attachment upload settles", async () => { + const { client } = fake(community); + const read = gate(); + const upload = gate>>(); + const uploaded = gate(); + vi.mocked(client.read).mockReturnValue(read.promise); + vi.mocked(client.upload).mockImplementation(() => { + uploaded.resolve(); + return upload.promise; + }); + const pending = callTool(client, { channel: "chan" }, "send", { + audience: "everyone", + text: "attachment", + files: ["file.png"], + }); + try { + expect(client.upload).not.toHaveBeenCalled(); + expect(client.publish).not.toHaveBeenCalled(); + read.resolve(PNG); + await uploaded.promise; + expect(client.publish).not.toHaveBeenCalled(); + upload.resolve({ + url: "https://relay/media/x.png", + sha256: createHash("sha256").update(PNG).digest("hex"), + size: PNG.length, + type: "image/png", + }); + expect(await pending).toContain("sent "); + expect(client.publish).toHaveBeenCalledOnce(); + } finally { + read.resolve(PNG); + upload.resolve({ url: "x", sha256: "", size: 0, type: "" }); + await pending.catch(() => undefined); + } +}); + +it("preserves ordinary attachment sends when snapshot hashing is unavailable", async () => { + const { client } = fake(community); + const hash = vi + .spyOn(crypto.subtle, "digest") + .mockRejectedValueOnce(new Error("hash unavailable")); + try { + await callTool(client, { channel: "chan" }, "send", { + audience: "everyone", + text: "attachment", + files: ["shot.png"], + }); + expect(client.upload).toHaveBeenCalledOnce(); + expect(client.publish).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ kind: 9 }), + ); + } finally { + hash.mockRestore(); + } +}); diff --git a/src/buzz-mcp/tools.ts b/src/buzz-mcp/tools.ts index 774e75e92a..c786130692 100644 --- a/src/buzz-mcp/tools.ts +++ b/src/buzz-mcp/tools.ts @@ -1,6 +1,7 @@ // The Buzz tools: what an agent does in Buzz, in the wire formats the `buzz` // CLI writes. Names, arguments and descriptions are kept short because every // session pays for them in context. Each tool answers in plain text. +import { bytesToHex } from "nostr-tools/utils"; import { decode, npubEncode } from "nostr-tools/nip19"; import buzzHelp from "../../docs/buzz-help.md?raw"; import { CANVAS_HISTORY_PAGE } from "../features/channel-templates/canvas-limits"; @@ -44,7 +45,9 @@ import type { Context, Filter, Memory, + ManagedAttachment, Upload, + Template, } from "./client"; const ROLES = ["owner", "admin", "member", "guest", "bot"]; @@ -1507,7 +1510,7 @@ async function post( const files = await tools.attach(list(args.files)); const content = `${(await tools.body(args, !files.links)) ?? ""}${files.links}`; const mentions = await tools.mentions(channel, content, list(args.mentions)); - const event = await tools.client.publish({ + const template: Template = { kind: 9, content, tags: [ @@ -1518,7 +1521,12 @@ async function post( ...(await tools.emoji(content)), ...extra, ], - }); + }; + const event = await (files.managedAttachments.length + ? tools.client.publish(template, { + managedAttachments: files.managedAttachments, + }) + : tools.client.publish(template)); return `sent ${event.id}`; } @@ -2055,17 +2063,62 @@ class Tools { async attach(paths: readonly string[]) { const tags: string[][] = []; let links = ""; + // Bound bookkeeping, not ordinary attachment count. On overflow omit all hints. + const candidates = new Map(); + let overflow = false; for (const path of paths) { const data = await this.client.read(path); + const size = data.byteLength; + // Hints are optional: hash support/failure must not break an ordinary send. + let sha256: string | undefined; + if (size <= 16 * 1024 * 1024) { + try { + sha256 = bytesToHex( + new Uint8Array( + await crypto.subtle.digest( + "SHA-256", + data as Uint8Array, + ), + ), + ); + } catch { + /* No cleanup hint; preserve the existing upload/send behavior. */ + } + } const upload = await this.client.upload( data, sniff(data) ?? "application/octet-stream", ); + if (!overflow) { + if (!candidates.has(path) && candidates.size === 16) { + overflow = true; + candidates.clear(); + } else { + const previous = candidates.get(path); + candidates.set( + path, + sha256 !== undefined && + upload.sha256 === sha256 && + upload.size === size && + previous !== null && + (!previous || + (previous.sha256 === sha256 && previous.size === size)) + ? { path, sha256, size, url: upload.url } + : null, + ); + } + } const name = attachmentName(path.split(/[\\/]/).pop() ?? ""); tags.push([...imeta(upload), `filename ${name}`]); links += `\n${attachmentMarkdown(name, upload)}`; } - return { tags, links }; + return { + tags, + links, + managedAttachments: [...candidates.values()].filter( + (candidate): candidate is ManagedAttachment => candidate !== null, + ), + }; } } diff --git a/src/features/agents/native.test.ts b/src/features/agents/native.test.ts index 2bf99518d8..02525aeed2 100644 --- a/src/features/agents/native.test.ts +++ b/src/features/agents/native.test.ts @@ -79,3 +79,27 @@ it("searches profile names natively and reports a community without the search", /malformed or invalidly signed/, ); }); + +it("keeps managed attachment hints out of the signed event and preserves absent-context IPC", async () => { + vi.mocked(invoke).mockReset().mockResolvedValue(message); + const template = { kind: 9, content: "attachment", tags: [["h", "c"]] }; + const managedAttachments = [ + { + path: "/saved/file", + sha256: "a".repeat(64), + size: 5, + url: "https://example.test/media", + }, + ]; + await native.publish(member, template, { managedAttachments }); + expect(invoke).toHaveBeenLastCalledWith("app_agent_publish", { + ...member, + event: template, + managedAttachments, + }); + await native.publish(member, template); + expect(invoke).toHaveBeenLastCalledWith("app_agent_publish", { + ...member, + event: template, + }); +}); diff --git a/src/features/agents/native.ts b/src/features/agents/native.ts index c9f548f040..26dbff0ca4 100644 --- a/src/features/agents/native.ts +++ b/src/features/agents/native.ts @@ -1,3 +1,4 @@ +import type { PublishOptions } from "../../buzz-mcp/client"; import type { ProtectionBinding, ProtectionDefaults, @@ -178,7 +179,11 @@ export type AgentsNative = { /** Signs one event as the agent (deletions, reactions, messages, edits, * canvases and DM opens only) with its owner attestation and posts it to the * membership's community. */ - publish(member: Membership, event: AgentEventTemplate): Promise; + publish( + member: Membership, + event: AgentEventTemplate, + options?: PublishOptions, + ): Promise; /** Reads the membership's community as the agent: one to eight filters. */ query(member: Membership, filters: readonly object[]): Promise; /** Finds profiles by `name` or `display_name` in the agent's community, as @@ -272,10 +277,13 @@ export const nativeAgents = (): AgentsNative | undefined => invoke("app_agent_delete", { pubkey, relay }), forget: ({ pubkey, relay }) => invoke("app_agent_forget", { pubkey, relay }), - publish: ({ pubkey, relay }, event) => + publish: ({ pubkey, relay }, event, options) => invoke("app_agent_publish", { pubkey, relay, + ...(options?.managedAttachments && { + managedAttachments: options.managedAttachments, + }), event: { kind: event.kind, content: event.content, diff --git a/src/features/agents/runner.ts b/src/features/agents/runner.ts index 9156ae38aa..60fc7901d5 100644 --- a/src/features/agents/runner.ts +++ b/src/features/agents/runner.ts @@ -3,6 +3,7 @@ // host run each agent. It knows nothing of where it runs: everything it needs // of the device, community and storage comes through its RunnerHost, so the // desktop app and a standalone host run agents with the same code. +import type { PublishOptions } from "../../buzz-mcp/client"; import { bytesToHex } from "nostr-tools/utils"; import type { ProtectionAction } from "../agent-protection/service"; import type { @@ -2684,7 +2685,11 @@ export class AgentRunners { owner: agent.owner, // What acts as the agent stops with it: a turn still working once it // is deleted, removed from this community or retyped can't. - publish: async (event: AgentEventTemplate, check?: () => void) => { + publish: async ( + event: AgentEventTemplate, + check?: () => void, + options?: PublishOptions, + ) => { lifetime.throwIfAborted(); const destination = event.tags?.find((tag) => tag[0] === "h")?.[1]; const ownReply = @@ -2696,6 +2701,7 @@ export class AgentRunners { event, lifetime, check, + options, ); if (ownReply) { const channel = this.host.channel(agent, published); @@ -3116,6 +3122,7 @@ export class AgentRunners { event: AgentEventTemplate, lifetime: AbortSignal, check?: () => void, + options?: PublishOptions, ) { const channel = event.tags?.find((tag) => tag[0] === "h")?.[1]; const agent = this.host.agent(runner.key); @@ -3126,7 +3133,9 @@ export class AgentRunners { lifetime.throwIfAborted(); } check?.(); - return this.require().publish(runner.member, event); + return options + ? this.require().publish(runner.member, event, options) + : this.require().publish(runner.member, event); } } diff --git a/src/features/agents/service.test.ts b/src/features/agents/service.test.ts index 1786804f27..825e75d0d3 100644 --- a/src/features/agents/service.test.ts +++ b/src/features/agents/service.test.ts @@ -12364,6 +12364,20 @@ it("sends actual private-file bytes through the app-owned sourceCommand reader a ([, template]) => template.kind === 9, ); expect(published?.[0]).toEqual(member(bot)); + expect(native.publish).toHaveBeenCalledWith( + member(bot), + expect.objectContaining({ kind: 9 }), + { + managedAttachments: [ + { + path, + sha256: hash, + size: bytes.length, + url: "https://files.example.test/bytes", + }, + ], + }, + ); expect(published?.[1].tags).toContainEqual(["h", channel]); expect(published?.[1].tags).toContainEqual( expect.arrayContaining(["imeta", `x ${hash}`, "filename private-file"]), @@ -14554,3 +14568,45 @@ it("keeps owned protection settings visible while the selected community reconne await settle(); expect(f.run).not.toHaveBeenCalled(); }); + +it("threads local attachment context through the runner's existing custody owner", async () => { + const { service, run, emit, native } = await setup(); + await service.create({ type: "example/echo", name: "Echo" }); + const options = { + managedAttachments: [ + { + path: "/saved/file", + sha256: "a".repeat(64), + size: 5, + url: "https://example.test/media", + }, + ], + }; + const check = vi.fn(); + const template = { kind: 9, content: "attachment", tags: [["h", channel]] }; + let finish!: () => void; + const completed = new Promise((resolve) => { + finish = resolve; + }); + run.mockImplementationOnce(async ({ agent }) => { + try { + await agent.publish(template, check, options); + } finally { + finish(); + } + }); + emit({ + events: [ + event("managed-attachment", { + tags: [ + ["h", channel], + ["p", bot], + ], + }), + ], + channelId: channel, + }); + await completed; + expect(check).toHaveBeenCalledOnce(); + expect(native.publish).toHaveBeenCalledWith(member(bot), template, options); +}); diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index 24869f3a97..5ffbbb8f4c 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -1,3 +1,4 @@ +import type { PublishOptions } from "../../buzz-mcp/client"; import { followUpQueue } from "../messages/follow-up-queue"; import type {} from "../navigation/service"; import { removeSessionPrompts } from "./session-prompts"; @@ -431,7 +432,11 @@ export type AgentHandle = Readonly<{ * conversations, as `Agents.wake` would, telling it the wake is its own. */ handoff?(handoff: AgentHandoff): void; /** Rechecks an operation lifetime after channel startup, before signing. */ - publish(event: AgentEventTemplate, check?: () => void): Promise; + publish( + event: AgentEventTemplate, + check?: () => void, + options?: PublishOptions, + ): Promise; /** Reads the community as the agent, seeing only what it may see. */ query(filters: readonly object[]): Promise; /** Finds profiles by `name` or `display_name`, as the agent, best match diff --git a/src/features/host/service.test.ts b/src/features/host/service.test.ts index 58df464fee..6a658e41d3 100644 --- a/src/features/host/service.test.ts +++ b/src/features/host/service.test.ts @@ -1593,3 +1593,95 @@ it("preserves requestOnce owner-first errors and matches upload owner wording", ).rejects.toThrow(ownerError); expect(invoke).not.toHaveBeenCalled(); }); + +it("requires native acknowledgement for release-after-send saves without retry", async () => { + const { fiber, host } = await oncePlugin(); + try { + const input = { sha256: byteHash, releaseAfterSend: true }; + vi.mocked(invoke).mockResolvedValueOnce(byteSaved); + await expect(host.saveBytes?.(new Uint8Array(3), input)).rejects.toThrow( + "possibly saved", + ); + expect(invoke).toHaveBeenCalledTimes(1); + expect(vi.mocked(invoke).mock.calls[0]?.[2]?.headers).toMatchObject({ + "x-buzz-release-after-send": "1", + }); + vi.mocked(invoke).mockResolvedValueOnce({ + ...byteSaved, + releaseAfterSend: true, + }); + await expect( + host.saveBytes?.(new Uint8Array(3), input), + ).resolves.toMatchObject({ releaseAfterSend: true }); + } finally { + await fiber.dispose(); + } +}); + +it("validates the release flag before IPC and omits false", async () => { + const { fiber, host } = await oncePlugin(); + try { + for (const flag of [null, 1, "true", {}]) { + const input = { sha256: byteHash, releaseAfterSend: flag as boolean }; + await expect(host.saveBytes?.(new Uint8Array(3), input)).rejects.toThrow( + "Invalid", + ); + } + expect(invoke).not.toHaveBeenCalled(); + vi.mocked(invoke).mockResolvedValueOnce(byteSaved); + await host.saveBytes?.(new Uint8Array(3), { + sha256: byteHash, + releaseAfterSend: false, + }); + expect(vi.mocked(invoke).mock.calls[0]?.[2]?.headers).not.toHaveProperty( + "x-buzz-release-after-send", + ); + } finally { + await fiber.dispose(); + } +}); + +it.each([undefined, false, "true", 1])( + "rejects %s release acknowledgement without fallback or retry", + async (ack) => { + const { host, fiber } = await oncePlugin(); + try { + vi.mocked(invoke).mockResolvedValueOnce({ + ...byteSaved, + releaseAfterSend: ack, + }); + await expect( + host.saveBytes?.(new Uint8Array(3), { + sha256: byteHash, + releaseAfterSend: true, + }), + ).rejects.toThrow("possibly saved"); + expect(invoke).toHaveBeenCalledOnce(); + } finally { + await fiber.dispose(); + } + }, +); + +it("renderer disposal suppresses an opted-in acknowledgement without claiming native cancellation", async () => { + const { host, fiber } = await oncePlugin(); + let finish!: (value: unknown) => void; + vi.mocked(invoke).mockImplementation( + () => + new Promise((resolve) => { + finish = resolve; + }), + ); + const pending = host.saveBytes?.(new Uint8Array(3), { + sha256: byteHash, + releaseAfterSend: true, + }); + await fiber.dispose(); + finish({ ...byteSaved, releaseAfterSend: true }); + await expect(pending).rejects.toThrow("inactive"); + expect(invoke).toHaveBeenCalledExactlyOnceWith( + "plugin_host_save_bytes", + expect.any(Uint8Array), + expect.anything(), + ); +}); diff --git a/src/features/host/service.ts b/src/features/host/service.ts index c2fa58af31..70cefe4ddb 100644 --- a/src/features/host/service.ts +++ b/src/features/host/service.ts @@ -68,6 +68,8 @@ export type HostSavedBytes = Readonly<{ sha256: string; /** Host wall-clock Unix milliseconds, not a deletion timer. */ expiresAt: number; + /** Native recorded process-local, same-human-owner consent; not a deletion receipt. */ + releaseAfterSend?: true; }>; /** Reusable NIP-OA proof for one agent key; preparing it does not submit it. */ export type NipOaAuthorization = readonly [ @@ -180,10 +182,16 @@ export interface Host { /** No automatic replay protection or text/array fallback. */ requestBytes?: (input: HostByteRequest) => Promise; /** Requires host.managedFiles, not host.files. Keep bytes immutable through - * hashing/save. An ambiguous failure must not be automatically retried. */ + * hashing/save. An ambiguous failure must not be automatically retried. + * releaseAfterSend defaults off. True requires compatible native acknowledgement + * and may refuse if native identity is unavailable/busy at commit. After the + * same human Nostr owner's confirmed attachment post, native makes one bounded + * best-effort release attempt. Consent is not durable or cross-process. + * Cleanup may never happen automatically; expiry only makes a file eligible for removal during a later save. + * Owner deletion, expiry sweeping, erasure and disk loss remain independent. */ saveBytes?: ( bytes: Uint8Array, - input: Readonly<{ sha256: string }>, + input: Readonly<{ sha256: string; releaseAfterSend?: boolean }>, ) => Promise; /** Removes this plugin's own managed file. Await attachment sends first; * Conservatively refuses app-wide while any plugin's host.readFile call or @@ -600,7 +608,10 @@ export class HostService extends Service implements Host { async saveBytes( bytes: Uint8Array, - { sha256 }: Readonly<{ sha256: string }>, + { + sha256, + releaseAfterSend, + }: Readonly<{ sha256: string; releaseAfterSend?: boolean }>, ): Promise { const ctx = this.ctx; const owner = ctx.pluginOwner; @@ -610,7 +621,8 @@ export class HostService extends Service implements Host { !(bytes instanceof Uint8Array) || bytes.byteLength < 1 || bytes.byteLength > BYTE_LIMIT || - !/^[0-9a-f]{64}$/.test(sha256) + !/^[0-9a-f]{64}$/.test(sha256) || + (releaseAfterSend !== undefined && typeof releaseAfterSend !== "boolean") ) throw new Error("Invalid managed save bytes or SHA256"); const size = bytes.byteLength; @@ -624,6 +636,9 @@ export class HostService extends Service implements Host { "x-buzz-plugin-id": owner.id, "x-buzz-plugin-revision": owner.revision, "x-buzz-sha256": sha256, + ...(releaseAfterSend === true && { + "x-buzz-release-after-send": "1", + }), }, }, ).catch(rethrow); @@ -639,6 +654,10 @@ export class HostService extends Service implements Host { saved.expiresAt <= 0 ) throw new Error("Invalid managed save result; outcome unknown"); + if (releaseAfterSend === true && saved.releaseAfterSend !== true) + throw new Error( + "Release after send is unsupported; possibly saved, do not retry", + ); return saved; } finally { lifetime.release(); diff --git a/src/plugins/author.test.mjs b/src/plugins/author.test.mjs index 489c6b8cee..a118b23ff4 100644 --- a/src/plugins/author.test.mjs +++ b/src/plugins/author.test.mjs @@ -128,12 +128,17 @@ export function apply(ctx: Context) { const bytes: HostByteRequest = { url: "https://api.example.com/bytes", maxBytes: 1024 }; void host.requestBytes?.(bytes).then((response: HostByteResponse) => { const body: Uint8Array = response.body; - void host.saveBytes?.(body, { sha256: "0".repeat(64) }).then((saved: HostSavedBytes) => { - void [saved.path, saved.size, saved.sha256, saved.expiresAt]; + void host.saveBytes?.(body, { sha256: "0".repeat(64), releaseAfterSend: true }).then((saved: HostSavedBytes) => { + const acknowledged: true | undefined = saved.releaseAfterSend; + void [saved.path, saved.size, saved.sha256, saved.expiresAt, acknowledged]; }); }); // @ts-expect-error Byte GET has no caller headers. host.requestBytes?.({ ...bytes, headers: {} }); + // @ts-expect-error Release policy must be a boolean, not a callback. + host.saveBytes?.(new Uint8Array([1]), { sha256: "0".repeat(64), releaseAfterSend: () => true }); + // @ts-expect-error Managed save has no caller account. + host.saveBytes?.(new Uint8Array([1]), { sha256: "0".repeat(64), owner: "caller supplied" }); // @ts-expect-error Managed save has no caller path. host.saveBytes?.(new Uint8Array([1]), { sha256: "0".repeat(64), path: "/chosen" }); ctx.pages.register({ From 9cbe9ef0bd688897f383bd083154b97c645fbb68 Mon Sep 17 00:00:00 2001 From: Zach Marley Date: Tue, 3 Nov 2026 18:19:25 -0500 Subject: [PATCH 2/4] test(host): respect platform identity in attachment release coverage Gate verified-unlink success coverage to Unix, exercise confirmed-send retention on non-Unix, and bound native fixture gate entry with a red-first diagnostic regression. Narrow the explicit-sweep test wording without changing production identity checks. Co-authored-by: goose Signed-off-by: Zach Marley --- src-tauri/src/app_agents.rs | 13 ++++++- src-tauri/src/host_files/managed_tests.rs | 43 +++++++++++++++++++---- 2 files changed, 49 insertions(+), 7 deletions(-) diff --git a/src-tauri/src/app_agents.rs b/src-tauri/src/app_agents.rs index a7c77940c5..33f1b8763f 100644 --- a/src-tauri/src/app_agents.rs +++ b/src-tauri/src/app_agents.rs @@ -2629,8 +2629,19 @@ mod managed_publish_tests { /// The actual HTTP post/evidence wrapper + real managed save/unlink. Loopback /// fixture only: not Wry/trusted-HTTPS or native signing-custody acceptance. + #[cfg(unix)] #[test] fn host_release_only_after_full_native_matching_receipt_and_evidence_admission() { + check_managed_publish_outcomes(true); + } + + #[cfg(not(unix))] + #[test] + fn host_release_confirmed_send_retains_source_without_platform_file_identity() { + check_managed_publish_outcomes(false); + } + + fn check_managed_publish_outcomes(releases_accepted_source: bool) { let _serial = commands(); for outcome in [ "accepted", @@ -2692,7 +2703,7 @@ mod managed_publish_tests { let received = server.await.unwrap(); // Exactly one connection/request; no repost path. if matches!(outcome, "accepted" | "cleanup-refused") { assert_eq!(result.unwrap()["id"], received["id"]); - assert_eq!(path.exists(), outcome == "cleanup-refused"); + assert_eq!(path.exists(), !releases_accepted_source || outcome == "cleanup-refused"); } else { assert!(result.is_err(), "{outcome} must be unconfirmed/refused"); assert!(path.exists(), "{outcome} must retain source"); diff --git a/src-tauri/src/host_files/managed_tests.rs b/src-tauri/src/host_files/managed_tests.rs index 8bc9ae631a..e883c23f51 100644 --- a/src-tauri/src/host_files/managed_tests.rs +++ b/src-tauri/src/host_files/managed_tests.rs @@ -1160,6 +1160,7 @@ fn host_release_raw_save_requires_strict_flag_and_native_ack() { assert_eq!(saved["releaseAfterSend"], true); } +#[cfg(unix)] #[test] fn host_release_confirmed_attachment_reclaims_only_native_consented_exact_file() { let _serial = commands(); @@ -1235,6 +1236,31 @@ fn managed_gate() -> ( let (release, allowed) = std::sync::mpsc::channel(); (entered, ready, ReleaseGate(Some(release)), allowed) } +async fn wait_for_managed_gate( + ready: tokio::sync::oneshot::Receiver<()>, +) -> Result<(), &'static str> { + tokio::time::timeout(std::time::Duration::from_secs(10), ready) + .await + .map_err(|_| "managed gate was not entered before watchdog")? + .map_err(|_| "managed gate closed before entry") +} + +#[tokio::test(start_paused = true)] +async fn managed_gate_entry_has_a_diagnostic_bound_even_if_sender_stays_alive() { + let (entered, ready) = tokio::sync::oneshot::channel(); + let mut entry = Box::pin(wait_for_managed_gate(ready)); + assert!(matches!( + futures_util::poll!(&mut entry), + std::task::Poll::Pending + )); + tokio::time::advance(std::time::Duration::from_secs(10)).await; + assert!(matches!( + futures_util::poll!(&mut entry), + std::task::Poll::Ready(Err("managed gate was not entered before watchdog")) + )); + drop(entered); +} + fn start_release_save(f: &Fixture) -> std::sync::mpsc::Receiver { let headers = reqwest::header::HeaderMap::from_iter([ ( @@ -1283,7 +1309,7 @@ fn host_release_actual_command_timeout_and_abort_fence_dispatched_save() { let (entered, ready, release, allowed) = managed_gate(); registry.1.lock().unwrap().save = Some((entered, allowed)); let response = start_release_save(&f); - ready.await.unwrap(); + wait_for_managed_gate(ready).await.unwrap(); let job = task.lock().unwrap().take().unwrap(); assert_eq!(BYTE_WORK.available_permits(), 0); if abort { @@ -1351,7 +1377,7 @@ fn host_release_identity_close_while_queued_or_at_commit_refuses_without_reassig registry.1.lock().unwrap().save = Some((entered, allowed)); } let response = start_release_save(&f); - ready.await.unwrap(); + wait_for_managed_gate(ready).await.unwrap(); identity.close(); drop(held); drop(release); @@ -1412,6 +1438,7 @@ pub(crate) fn release_fixture_data( (hints, event, owner) } +#[cfg(unix)] #[test] fn host_release_reclaims_more_than_sixteen_sequential_saves_and_byte_capacity() { let _serial = commands(); @@ -1456,7 +1483,7 @@ fn host_release_reclaims_more_than_sixteen_sequential_saves_and_byte_capacity() } #[test] -fn host_release_refusal_is_not_an_expiry_timer_and_restart_never_adopts_consent() { +fn host_release_refusal_and_restart_retain_until_explicit_sweep_and_prune() { let _serial = commands(); let f = Fixture::new(true, false); let registry = f.app.state::().inner().clone(); @@ -1481,7 +1508,7 @@ fn host_release_refusal_is_not_an_expiry_timer_and_restart_never_adopts_consent( )); assert!(path.exists(), "restart map must not adopt consent"); let expiry = artifact_expiry(path.file_name().unwrap().to_str().unwrap()).unwrap(); - let _future = expiry + 1; // No work is scheduled when this clock boundary passes. + // Immediate retention above and explicit sweep/prune below; no clock is advanced. assert!(path.exists()); assert_eq!(registry.0.lock().unwrap().len(), 1); let lock = managed_lock(&f.root()).unwrap(); @@ -1492,6 +1519,7 @@ fn host_release_refusal_is_not_an_expiry_timer_and_restart_never_adopts_consent( assert!(registry.0.lock().unwrap().is_empty()); } +#[cfg(unix)] #[test] fn host_release_stalled_hash_has_one_budget_and_prompt_direct_and_process_read_refusal() { let _serial = commands(); @@ -1520,7 +1548,7 @@ fn host_release_stalled_hash_has_one_budget_and_prompt_direct_and_process_read_r .unwrap(); runtime.block_on(async { let cleanup = tokio::spawn(release_after_send(registry.clone(), root, owner, signed, Some(serde_json::json!(candidates)))); - ready.await.unwrap(); + wait_for_managed_gate(ready).await.unwrap(); let direct = plugin_host_read_file(f.app.state::(), "example.bytes".into(), f.revision.clone(), path.clone(), None); assert!(matches!(futures_util::poll!(Box::pin(direct)), std::task::Poll::Ready(Err(error)) if error == "File reads are busy; try again later")); for process in ["read", "read-protected"] { @@ -1651,6 +1679,7 @@ fn host_release_rejects_forged_owner_content_paths_metadata_and_replaced_files() } } +#[cfg(unix)] #[test] fn host_release_committed_consent_survives_saver_disable_and_later_identity_close() { let _serial = commands(); @@ -1739,6 +1768,7 @@ fn host_release_reader_and_exclusion_races_are_mutually_exclusive_and_unwind_saf } } +#[cfg(unix)] #[test] fn host_release_duplicate_parallel_and_public_delete_paths_are_single_use() { let _serial = commands(); @@ -1808,6 +1838,7 @@ fn host_release_duplicate_parallel_and_public_delete_paths_are_single_use() { )); } +#[cfg(unix)] #[test] fn host_release_missing_consent_prunes_on_save_and_one_bad_candidate_keeps_later_safe_work() { let _serial = commands(); @@ -1906,7 +1937,7 @@ fn host_release_lost_renderer_waiter_is_not_native_command_cancellation() { let (entered, ready, release, allowed) = managed_gate(); registry.1.lock().unwrap().save = Some((entered, allowed)); let reply = start_release_save(&f); - ready.await.unwrap(); + wait_for_managed_gate(ready).await.unwrap(); drop(reply); // The native future remains alive: no cancel IPC exists. drop(release); let job = task.lock().unwrap().take().unwrap(); From 5745b7ab955a5e25c96031f41c772ba5cecd3d89 Mon Sep 17 00:00:00 2001 From: Zach Marley Date: Tue, 3 Nov 2026 23:05:03 -0500 Subject: [PATCH 3/4] fix(host): detach managed cleanup from confirmed sends Return the evidence-confirmed exact event before cleanup byte admission. Capture the owned cleanup deadline before spawning on the app runtime, retaining existing actual-work admission and exclusion. Cover held byte admission, runner sending completion, and expired or cancelled cleanup; remove the harness-only watchdog test. Co-authored-by: goose Signed-off-by: Zach Marley --- docs/plugin-architecture.md | 11 +++- src-tauri/src/app_agents.rs | 37 +++++++++--- src-tauri/src/host_files.rs | 49 ++++++++++------ src-tauri/src/host_files/managed_tests.rs | 66 +++++++++++++++------ src/features/agents/service.test.ts | 70 +++++++++++++++++++++++ 5 files changed, 188 insertions(+), 45 deletions(-) diff --git a/docs/plugin-architecture.md b/docs/plugin-architecture.md index fb1b4a6b69..884e62da11 100644 --- a/docs/plugin-architecture.md +++ b/docs/plugin-architecture.md @@ -1546,8 +1546,15 @@ retain the source. Public explicit deletion retains its existing grant/owner rul One bounded best-effort attempt per distinct source uses the shared byte permit, root writer and reader exclusion, with **one shared 30-second cleanup-entry budget -for the whole message**, including queue wait. Failure/refusal never changes a -confirmed send into an error or repost. Active app-wide readers refuse cleanup; +for the whole message**, including scheduling and queue wait. The confirmed send +returns immediately without awaiting cleanup; cleanup runs detached on the app's +native runtime, not the runner lifetime. Runner disposal does not cancel it. +The deadline is captured before spawning, and actual blocking work keeps its +byte/root/reader guards until completion. App shutdown need not drain cleanup: +undispatched work can be lost and sources retained; dispatched unlink may finish +with an unknown outcome. No work is durable or required to outlive the app. +Failure/refusal never changes a confirmed send into an error or repost. +Active app-wide readers refuse cleanup; while cleanup verifies content, new direct/process reads instead promptly refuse with `File reads are busy; try again later`, without spawning/reading or silently retrying. A stalled hash can outlive the response budget; dispatched work retains diff --git a/src-tauri/src/app_agents.rs b/src-tauri/src/app_agents.rs index 33f1b8763f..03eae8ad76 100644 --- a/src-tauri/src/app_agents.rs +++ b/src-tauri/src/app_agents.rs @@ -926,15 +926,17 @@ async fn confirmed_post( .run(true, post(agent, key, evidence.header(), signed)) .await .map_err(|error| publish_error(&event_id, error))?; - if let Some(root) = root { - crate::host_files::release_after_send( + if let (Some(root), Some(hints)) = (root, managed_attachments) { + // App-runtime-owned, not runner-owned: confirmation never waits for byte + // admission or cleanup. The future captures its deadline before spawning. + // App exit need not drain it; undispatched work may leave the source behind. + tauri::async_runtime::spawn(crate::host_files::release_after_send( registry, root, agent.owner.clone(), confirmed.clone(), - managed_attachments, - ) - .await; + Some(hints), + )); } Ok(confirmed) } @@ -2621,7 +2623,7 @@ mod tests { mod managed_publish_tests { use super::*; use crate::host_files::{ - managed_tests::{commands, release_fixture_data, Fixture}, + managed_tests::{commands, release_completion, release_fixture_data, Fixture}, ManagedFiles, }; use tauri::Manager as _; @@ -2631,7 +2633,7 @@ mod managed_publish_tests { /// fixture only: not Wry/trusted-HTTPS or native signing-custody acceptance. #[cfg(unix)] #[test] - fn host_release_only_after_full_native_matching_receipt_and_evidence_admission() { + fn host_release_confirmed_send_returns_before_cleanup_byte_admission_and_preserves_evidence() { check_managed_publish_outcomes(true); } @@ -2645,6 +2647,7 @@ mod managed_publish_tests { let _serial = commands(); for outcome in [ "accepted", + "byte-busy", "refused", "wrong-id", "bad-json", @@ -2657,6 +2660,7 @@ mod managed_publish_tests { let (hints, event, owner) = release_fixture_data(&f, b"native confirmed source"); let path = PathBuf::from(hints[0]["path"].as_str().unwrap()); let registry = f.app.state::().inner().clone(); + let complete = release_completion(®istry); let root = f.root(); let runtime = tokio::runtime::Builder::new_current_thread() .enable_all() @@ -2699,10 +2703,25 @@ mod managed_publish_tests { received }); let lease = if outcome == "cleanup-refused" { Some(crate::host_files::FileReadLease::new().unwrap()) } else { None }; - let result = confirmed_post(&agent, &key, evidence, signed, registry, Some(root), Some(hints)).await; + let held = if outcome == "byte-busy" { + Some(crate::host_files::BYTE_WORK.acquire().await.unwrap()) + } else { + None + }; + let publish = tokio::time::timeout( + std::time::Duration::from_secs(2), + confirmed_post(&agent, &key, evidence, signed, registry, Some(root), Some(hints)), + ).await; let received = server.await.unwrap(); // Exactly one connection/request; no repost path. - if matches!(outcome, "accepted" | "cleanup-refused") { + if held.is_some() { + assert!(path.exists(), "busy byte permit must prevent early cleanup"); + } + let result = publish.expect("confirmed send waited for cleanup byte admission"); + drop(held); + if matches!(outcome, "accepted" | "byte-busy" | "cleanup-refused") { assert_eq!(result.unwrap()["id"], received["id"]); + tokio::time::timeout(std::time::Duration::from_secs(10), complete) + .await.expect("detached cleanup did not finish").unwrap(); assert_eq!(path.exists(), !releases_accepted_source || outcome == "cleanup-refused"); } else { assert!(result.is_err(), "{outcome} must be unconfirmed/refused"); diff --git a/src-tauri/src/host_files.rs b/src-tauri/src/host_files.rs index 1c59b90499..a368aae81e 100644 --- a/src-tauri/src/host_files.rs +++ b/src-tauri/src/host_files.rs @@ -374,6 +374,7 @@ pub(crate) struct ManagedFiles( #[cfg(test)] #[derive(Default)] struct ManagedGates { + released: Option>, captured: Option>, save: Option<( tokio::sync::oneshot::Sender<()>, @@ -520,31 +521,43 @@ fn confirmed_candidates( /// Called only after the full evidence-wrapped exact-ID post has succeeded. /// Invalid hints and every cleanup failure preserve that original send success. -pub(crate) async fn release_after_send( +pub(crate) fn release_after_send( registry: ManagedFiles, root: PathBuf, owner: String, signed: serde_json::Value, hints: Option, -) { +) -> impl std::future::Future + Send + 'static { + // Capture before detaching: scheduling never grants a fresh 30-second budget. let deadline = tokio::time::Instant::now() + SAVE_WINDOW; - let candidates = confirmed_candidates(&signed, hints); - if candidates.is_empty() { - return; + async move { + let candidates = confirmed_candidates(&signed, hints); + if candidates.is_empty() { + return; + } + let budget = SaveEligibility::default(); + let active = budget.0.clone(); + #[cfg(test)] + let completion = registry.clone(); + let _ = delete_with_admission(deadline, move || { + release_batch( + ®istry, + &root, + &owner, + &candidates, + deadline.into_std(), + &active, + ) + }) + .await; + #[cfg(test)] + { + let released = completion.1.lock().unwrap().released.take(); + if let Some(released) = released { + let _ = released.send(()); + } + } } - let budget = SaveEligibility::default(); - let active = budget.0.clone(); - let _ = delete_with_admission(deadline, move || { - release_batch( - ®istry, - &root, - &owner, - &candidates, - deadline.into_std(), - &active, - ) - }) - .await; } fn release_batch( diff --git a/src-tauri/src/host_files/managed_tests.rs b/src-tauri/src/host_files/managed_tests.rs index e883c23f51..b98c0b07c9 100644 --- a/src-tauri/src/host_files/managed_tests.rs +++ b/src-tauri/src/host_files/managed_tests.rs @@ -1245,22 +1245,6 @@ async fn wait_for_managed_gate( .map_err(|_| "managed gate closed before entry") } -#[tokio::test(start_paused = true)] -async fn managed_gate_entry_has_a_diagnostic_bound_even_if_sender_stays_alive() { - let (entered, ready) = tokio::sync::oneshot::channel(); - let mut entry = Box::pin(wait_for_managed_gate(ready)); - assert!(matches!( - futures_util::poll!(&mut entry), - std::task::Poll::Pending - )); - tokio::time::advance(std::time::Duration::from_secs(10)).await; - assert!(matches!( - futures_util::poll!(&mut entry), - std::task::Poll::Ready(Err("managed gate was not entered before watchdog")) - )); - drop(entered); -} - fn start_release_save(f: &Fixture) -> std::sync::mpsc::Receiver { let headers = reqwest::header::HeaderMap::from_iter([ ( @@ -1424,6 +1408,12 @@ fn host_release_identity_close_while_queued_or_at_commit_refuses_without_reassig } } +pub(crate) fn release_completion(registry: &ManagedFiles) -> tokio::sync::oneshot::Receiver<()> { + let (released, complete) = tokio::sync::oneshot::channel(); + registry.1.lock().unwrap().released = Some(released); + complete +} + pub(crate) fn release_fixture_data( f: &Fixture, bytes: &[u8], @@ -1438,6 +1428,50 @@ pub(crate) fn release_fixture_data( (hints, event, owner) } +#[test] +fn host_release_deadline_starts_before_first_poll_and_cancelled_queue_never_unlinks() { + let _serial = commands(); + for cancel in [false, true] { + let f = Fixture::new(true, false); + let registry = f.app.state::().inner().clone(); + let (hints, event, owner) = release_fixture_data(&f, b"queued source"); + let path = PathBuf::from(hints[0]["path"].as_str().unwrap()); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let held = BYTE_WORK.acquire().await.unwrap(); + let cleanup = release_after_send(registry, f.root(), owner, event, Some(hints)); + if cancel { + let task = tokio::spawn(cleanup); + // On this current-thread runtime, let cleanup reach its blocked + // admission await, then observe its bounded waiter reservation. + tokio::task::yield_now().await; + assert_eq!(BYTE_WAITERS.available_permits(), 15); + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + } else { + tokio::time::pause(); + tokio::time::advance(SAVE_WINDOW + std::time::Duration::from_secs(1)).await; + cleanup.await; // No first-poll reset of the captured budget. + tokio::time::resume(); + } + assert_eq!(BYTE_WAITERS.available_permits(), 16); + assert!(path.exists()); + drop(held); + let settled = byte_admission(tokio::time::Instant::now() + SAVE_WINDOW) + .await + .unwrap(); + assert!( + path.exists(), + "expired/cancelled queued work must not dispatch later" + ); + drop(settled); + }); + } +} + #[cfg(unix)] #[test] fn host_release_reclaims_more_than_sixteen_sequential_saves_and_byte_capacity() { diff --git a/src/features/agents/service.test.ts b/src/features/agents/service.test.ts index 825e75d0d3..2cd6713dd0 100644 --- a/src/features/agents/service.test.ts +++ b/src/features/agents/service.test.ts @@ -14569,6 +14569,76 @@ it("keeps owned protection settings visible while the selected community reconne expect(f.run).not.toHaveBeenCalled(); }); +it("returns sent and leaves sending on native confirmation without waiting for detached cleanup", async () => { + const { service, run, emit, native, activity } = await setup(); + await service.create({ type: "example/echo", name: "Echo" }); + const bytes = new Uint8Array([1, 2, 3]); + const hash = createHash("sha256").update(bytes).digest("hex"); + native.upload.mockResolvedValue({ + url: "https://files.example.test/source", + sha256: hash, + size: bytes.length, + type: "application/octet-stream", + }); + let confirm!: (value: RelayEvent) => void; + const confirmation = new Promise((resolve) => { + confirm = resolve; + }); + let finishCleanup!: () => void; + const cleanup = new Promise((resolve) => { + finishCleanup = resolve; + }); + let cleaned = false; + native.publish.mockImplementationOnce(() => confirmation); + let reply: unknown; + run.mockImplementationOnce(async ({ agent }) => { + const client = appClient(agent, { + read: async () => bytes, + memories: async () => ({ entries: [], partial: false }), + }); + reply = await callTool(client, { channel }, "send", { + audience: "everyone", + text: "attachment", + files: ["source"], + }); + }); + emit({ + events: [ + event("confirmed-attachment", { + tags: [ + ["h", channel], + ["p", bot], + ], + }), + ], + channelId: channel, + }); + try { + await vi.waitFor(() => + expect(activity.reader.work()[0]).toMatchObject({ publishing: 1 }), + ); + const confirmed = event("confirmed-source", { + pubkey: bot, + tags: [["h", channel]], + }); + // The native regression holds the real BYTE_WORK permit. This JS boundary + // models its confirmed response while detached cleanup still has work queued. + const background = cleanup.then(() => { + cleaned = true; + }); + confirm(confirmed); + await vi.waitFor(() => expect(reply).toBe(`sent ${confirmed.id}`)); + expect(activity.reader.work()[0]).not.toHaveProperty("publishing"); + expect(activity.reader.work()[0]).toMatchObject({ state: "done" }); + expect(cleaned).toBe(false); + finishCleanup(); + await background; + expect(cleaned).toBe(true); + } finally { + finishCleanup(); + } +}); + it("threads local attachment context through the runner's existing custody owner", async () => { const { service, run, emit, native } = await setup(); await service.create({ type: "example/echo", name: "Echo" }); From 1c5a5f46a475447d0402890a87bd3158eed617ef Mon Sep 17 00:00:00 2001 From: Zach Marley Date: Wed, 4 Nov 2026 08:12:14 -0500 Subject: [PATCH 4/4] test(host): discriminate cleanup deadline reset Require immediate first-poll completion after cleanup deadline expiry, preserving real permit, file, capacity and cancellation assertions. Remove only the redundant test-local JS cleanup model; existing publication and native held-byte-permit regressions remain. Co-authored-by: goose Signed-off-by: Zach Marley --- src-tauri/src/host_files/managed_tests.rs | 11 +++- src/features/agents/service.test.ts | 70 ----------------------- 2 files changed, 10 insertions(+), 71 deletions(-) diff --git a/src-tauri/src/host_files/managed_tests.rs b/src-tauri/src/host_files/managed_tests.rs index b98c0b07c9..6ef10e8f9e 100644 --- a/src-tauri/src/host_files/managed_tests.rs +++ b/src-tauri/src/host_files/managed_tests.rs @@ -1454,7 +1454,16 @@ fn host_release_deadline_starts_before_first_poll_and_cancelled_queue_never_unli } else { tokio::time::pause(); tokio::time::advance(SAVE_WINDOW + std::time::Duration::from_secs(1)).await; - cleanup.await; // No first-poll reset of the captured budget. + // An expired attempt must finish on its first poll, without + // letting paused time auto-advance through a reset deadline. + let mut cleanup = Box::pin(cleanup); + assert!( + matches!( + futures_util::poll!(&mut cleanup), + std::task::Poll::Ready(()) + ), + "expired cleanup must return immediately without a fresh budget" + ); tokio::time::resume(); } assert_eq!(BYTE_WAITERS.available_permits(), 16); diff --git a/src/features/agents/service.test.ts b/src/features/agents/service.test.ts index 2cd6713dd0..825e75d0d3 100644 --- a/src/features/agents/service.test.ts +++ b/src/features/agents/service.test.ts @@ -14569,76 +14569,6 @@ it("keeps owned protection settings visible while the selected community reconne expect(f.run).not.toHaveBeenCalled(); }); -it("returns sent and leaves sending on native confirmation without waiting for detached cleanup", async () => { - const { service, run, emit, native, activity } = await setup(); - await service.create({ type: "example/echo", name: "Echo" }); - const bytes = new Uint8Array([1, 2, 3]); - const hash = createHash("sha256").update(bytes).digest("hex"); - native.upload.mockResolvedValue({ - url: "https://files.example.test/source", - sha256: hash, - size: bytes.length, - type: "application/octet-stream", - }); - let confirm!: (value: RelayEvent) => void; - const confirmation = new Promise((resolve) => { - confirm = resolve; - }); - let finishCleanup!: () => void; - const cleanup = new Promise((resolve) => { - finishCleanup = resolve; - }); - let cleaned = false; - native.publish.mockImplementationOnce(() => confirmation); - let reply: unknown; - run.mockImplementationOnce(async ({ agent }) => { - const client = appClient(agent, { - read: async () => bytes, - memories: async () => ({ entries: [], partial: false }), - }); - reply = await callTool(client, { channel }, "send", { - audience: "everyone", - text: "attachment", - files: ["source"], - }); - }); - emit({ - events: [ - event("confirmed-attachment", { - tags: [ - ["h", channel], - ["p", bot], - ], - }), - ], - channelId: channel, - }); - try { - await vi.waitFor(() => - expect(activity.reader.work()[0]).toMatchObject({ publishing: 1 }), - ); - const confirmed = event("confirmed-source", { - pubkey: bot, - tags: [["h", channel]], - }); - // The native regression holds the real BYTE_WORK permit. This JS boundary - // models its confirmed response while detached cleanup still has work queued. - const background = cleanup.then(() => { - cleaned = true; - }); - confirm(confirmed); - await vi.waitFor(() => expect(reply).toBe(`sent ${confirmed.id}`)); - expect(activity.reader.work()[0]).not.toHaveProperty("publishing"); - expect(activity.reader.work()[0]).toMatchObject({ state: "done" }); - expect(cleaned).toBe(false); - finishCleanup(); - await background; - expect(cleaned).toBe(true); - } finally { - finishCleanup(); - } -}); - it("threads local attachment context through the runner's existing custody owner", async () => { const { service, run, emit, native } = await setup(); await service.create({ type: "example/echo", name: "Echo" });