From 0e26e06bcd0a18109af3b26992c66e7fe8d742e8 Mon Sep 17 00:00:00 2001 From: Tee8z Date: Wed, 4 Nov 2026 12:45:27 -0500 Subject: [PATCH] feat(agents): wake remote agents from scheduled workflows Signed-off-by: Tee8z --- crates/buzz-acp/src/lib.rs | 67 +----- crates/buzz-acp/src/pool.rs | 4 +- crates/buzz-acp/src/sharing.rs | 39 +++- crates/buzz-acp/src/sharing/tests.rs | 62 ++++- .../src/mentions/tests/shared_live.rs | 15 +- .../src/mentions/tests/workflow_live.rs | 220 ++++++++++++++++++ .../src/shared_budget/tests.rs | 1 + crates/buzz-sdk/src/lib.rs | 1 + crates/buzz-sdk/src/sharing/budget.rs | 4 +- crates/buzz-sdk/src/sharing/budget/tests.rs | 1 + crates/buzz-sdk/src/sharing/policy.rs | 29 ++- crates/buzz-sdk/src/sharing/policy/tests.rs | 3 + .../src/sharing/policy/workflow_tests.rs | 172 ++++++++++++++ crates/buzz-sdk/src/sharing/task.rs | 2 +- crates/buzz-sdk/src/workflow.rs | 51 ++++ docs/shared-channel-agents.md | 39 ++++ 16 files changed, 615 insertions(+), 95 deletions(-) create mode 100644 crates/buzz-sandbox-reaper/src/mentions/tests/workflow_live.rs create mode 100644 crates/buzz-sdk/src/sharing/policy/workflow_tests.rs create mode 100644 crates/buzz-sdk/src/workflow.rs diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index fabcbe0803c..30ad6faa1c4 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -449,64 +449,9 @@ fn verified_workflow_owner( relay_self: Option<&str>, agent_pubkey_hex: &str, ) -> Option { - if event.kind.as_u16() as u32 != KIND_STREAM_MESSAGE { - return None; - } - - let relay_self = nostr::PublicKey::from_hex(relay_self?).ok()?; - if event.pubkey != relay_self || event.verify().is_err() { - return None; - } - - let markers: Vec<&[String]> = event - .tags - .iter() - .map(|tag| tag.as_slice()) - .filter(|values| values.first().map(String::as_str) == Some("buzz:workflow")) - .collect(); - if markers.as_slice() != [["buzz:workflow", "true"]] { - return None; - } - - let owners: Vec<&[String]> = event - .tags - .iter() - .map(|tag| tag.as_slice()) - .filter(|values| values.first().map(String::as_str) == Some("buzz:workflow-owner")) - .collect(); - let [owner_tag] = owners.as_slice() else { - return None; - }; - let [_, owner_value] = owner_tag else { - return None; - }; - let owner = nostr::PublicKey::from_hex(owner_value).ok()?.to_hex(); - if owner_value.as_str() != owner { - return None; - } - - let agent_pubkey = nostr::PublicKey::from_hex(agent_pubkey_hex).ok()?.to_hex(); - let workflow_mentions: Vec<&[String]> = event - .tags - .iter() - .map(|tag| tag.as_slice()) - .filter(|values| values.first().map(String::as_str) == Some("buzz:workflow-mention")) - .collect(); - let mut mentioned_pubkeys = HashSet::with_capacity(workflow_mentions.len()); - for mention_tag in workflow_mentions { - let [_, mention_value] = mention_tag else { - return None; - }; - let mention = nostr::PublicKey::from_hex(mention_value).ok()?.to_hex(); - if mention_value.as_str() != mention || !mentioned_pubkeys.insert(mention) { - return None; - } - } - if !mentioned_pubkeys.contains(&agent_pubkey) { - return None; - } - - Some(owner) + let relay = nostr::PublicKey::from_hex(relay_self?).ok()?; + let agent = nostr::PublicKey::from_hex(agent_pubkey_hex).ok()?; + buzz_sdk::workflow::owner(event, relay, agent).map(|owner| owner.to_hex()) } /// Resolve the author principal used by the inbound author gate. @@ -779,13 +724,13 @@ mod inbound_author_gate { if let Some(sharing) = &self.sharing { if sharing.accepts(&buzz_event.event) { let refusal = match sharing.authorize(&buzz_event.event).await { - Ok(None) => { + Ok(crate::sharing::Admission::Allowed(author)) => { return Ok(Some(AuthorizedListenerEvent { buzz_event, - effective_author: decision.effective_author, + effective_author: author.to_hex(), })) } - Ok(Some(reason)) => reason, + Ok(crate::sharing::Admission::Refused(reason)) => reason, Err(error) => error.to_string(), }; // Persist a visible refusal before handling a transient diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index ec13fe652d7..0a86526415b 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -2521,8 +2521,8 @@ pub async fn run_prompt_task( continue; } match sharing.authorize(&item.event).await { - Ok(None) => {} - Ok(Some(reason)) => { + Ok(crate::sharing::Admission::Allowed(_)) => {} + Ok(crate::sharing::Admission::Refused(reason)) => { denied = Some(reason); break; } diff --git a/crates/buzz-acp/src/sharing.rs b/crates/buzz-acp/src/sharing.rs index 594336c775c..90be312711e 100644 --- a/crates/buzz-acp/src/sharing.rs +++ b/crates/buzz-acp/src/sharing.rs @@ -18,6 +18,13 @@ mod tests; #[error("shared task admission unavailable; retry or recover this task")] pub struct Error; +/// A verified manager admission, or a permanent refusal safe to show the user. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum Admission { + Allowed(PublicKey), + Refused(String), +} + /// Agent credentials used only for encrypted requests and signed channel replies. /// Deliberately has no Debug implementation. #[derive(Clone)] @@ -72,16 +79,18 @@ impl Client { } /// Return a permanent refusal separately from a retryable transport failure. - pub(crate) async fn authorize(&self, event: &Event) -> Result, Error> { + pub(crate) async fn authorize(&self, event: &Event) -> Result { if !self.accepts(event) { - return Ok(Some("task is outside this bound conversation".into())); + return Ok(Admission::Refused( + "task is outside this bound conversation".into(), + )); } tokio::time::timeout(Duration::from_secs(50), self.exchange(event)) .await .map_err(|_| Error)? } - async fn exchange(&self, trigger: &Event) -> Result, Error> { + async fn exchange(&self, trigger: &Event) -> Result { let auth = buzz_sdk::nip_oa::parse_auth_tag(&self.auth).map_err(|_| Error)?; let request = Request::new( Op::Admit, @@ -148,7 +157,7 @@ impl Client { } let response = reply.response.ok_or(Error)?; if response["ok"] == true { - return Ok(None); + return Ok(Admission::Allowed(admitted_author(&response, trigger)?)); } if response["code"] == "shared_task_retry" { return Err(Error); @@ -156,7 +165,7 @@ impl Client { let reason = response["error"] .as_str() .unwrap_or("task permission was refused"); - return Ok(Some(reason.chars().take(500).collect())); + return Ok(Admission::Refused(reason.chars().take(500).collect())); } RelayMessage::Closed { .. } | RelayMessage::Auth { .. } => return Err(Error), _ => {} @@ -268,6 +277,26 @@ impl Client { } } +// Called only after authenticating the manager's encrypted, request-bound reply. +// Root tasks adopt the authorizing human; handoffs retain the sending agent. +fn admitted_author(response: &serde_json::Value, trigger: &Event) -> Result { + if buzz_sdk::sharing::task::origin(trigger) + .map_err(|_| Error)? + .is_some() + { + return Ok(trigger.pubkey); + } + let Some(value) = response.get("human") else { + return Ok(trigger.pubkey); + }; + let text = value.as_str().ok_or(Error)?; + let key = PublicKey::from_hex(text).map_err(|_| Error)?; + if key.to_hex() != text { + return Err(Error); + } + Ok(key) +} + pub(crate) struct NoticeWorker(tokio::task::JoinHandle<()>); impl Drop for NoticeWorker { fn drop(&mut self) { diff --git a/crates/buzz-acp/src/sharing/tests.rs b/crates/buzz-acp/src/sharing/tests.rs index 2d756d5244a..f2e2aca1cd2 100644 --- a/crates/buzz-acp/src/sharing/tests.rs +++ b/crates/buzz-acp/src/sharing/tests.rs @@ -41,6 +41,7 @@ async fn encrypted_admission_uses_fresh_ids_and_request_scoped_reply_filters() { let requests = Arc::new(Mutex::new(Vec::::new())); let captured = requests.clone(); let manager_pub = manager.public_key(); + let human = Keys::generate().public_key(); let server = tokio::spawn(async move { for _ in 0..5 { let (stream, _) = listener.accept().await.unwrap(); @@ -105,10 +106,12 @@ async fn encrypted_admission_uses_fresh_ids_and_request_scoped_reply_filters() { )) .await .unwrap(); - let response = if attempt == 0 || attempt == 3 { - json!({"ok":true}) - } else { - json!({"ok":false,"code":"shared_task_denied","error":"sharing was revoked"}) + let response = match attempt { + 0 => json!({"ok":true}), + 3 => json!({"ok":true,"human":human}), + _ => { + json!({"ok":false,"code":"shared_task_denied","error":"sharing was revoked"}) + } }; let reply = buzz_session_control::Reply::done(opened.request.request_id, response); @@ -134,10 +137,13 @@ async fn encrypted_admission_uses_fresh_ids_and_request_scoped_reply_filters() { } }); let (client, event) = fixture(url, manager_pub, directory.path().into()); - assert!(client.authorize(&event).await.unwrap().is_none()); assert_eq!( - client.authorize(&event).await.unwrap().as_deref(), - Some("sharing was revoked") + client.authorize(&event).await.unwrap(), + Admission::Allowed(event.pubkey) + ); + assert_eq!( + client.authorize(&event).await.unwrap(), + Admission::Refused("sharing was revoked".into()) ); let (rest, rest_server) = crate::author_gate_tests::nip11_server(json!({"self":manager_pub})).await; @@ -189,8 +195,8 @@ async fn encrypted_admission_uses_fresh_ids_and_request_scoped_reply_filters() { .unwrap(); assert_eq!( allowed.into_parts().1, - event.pubkey.to_hex(), - "shared requests retain the actual signer" + human.to_hex(), + "a verified manager response supplies the workflow's authorizing human" ); rest_server.abort(); server.await.unwrap(); @@ -202,6 +208,44 @@ async fn encrypted_admission_uses_fresh_ids_and_request_scoped_reply_filters() { ); } +#[test] +fn delegated_author_metadata_cannot_replace_a_handoff_signer_or_accept_malformed_humans() { + let directory = tempfile::tempdir().unwrap(); + let (client, root) = fixture( + "ws://127.0.0.1:1".into(), + Keys::generate().public_key(), + directory.path().into(), + ); + let human = Keys::generate().public_key(); + assert_eq!( + admitted_author(&json!({"human":human}), &root).unwrap(), + human + ); + for value in [ + json!(null), + json!(42), + json!("invalid"), + json!(human.to_hex().to_uppercase()), + ] { + assert!(admitted_author(&json!({"human":value}), &root).is_err()); + } + let handoff = buzz_sdk::sharing::task::handoff( + &root, + client.keys.public_key(), + Keys::generate().public_key(), + "continue", + Timestamp::now().as_secs(), + ) + .unwrap() + .sign_with_keys(&client.keys) + .unwrap(); + assert_eq!( + admitted_author(&json!({"human":human}), &handoff).unwrap(), + client.keys.public_key(), + "a handoff must retain its actual agent signer" + ); +} + #[tokio::test] async fn failed_refusal_delivery_is_durable_and_does_not_generate_agent_mentions() { let directory = tempfile::tempdir().unwrap(); diff --git a/crates/buzz-sandbox-reaper/src/mentions/tests/shared_live.rs b/crates/buzz-sandbox-reaper/src/mentions/tests/shared_live.rs index 12fea19dd78..1853f66c42e 100644 --- a/crates/buzz-sandbox-reaper/src/mentions/tests/shared_live.rs +++ b/crates/buzz-sandbox-reaper/src/mentions/tests/shared_live.rs @@ -7,6 +7,9 @@ use buzz_ws_client::RelayMessage; use std::path::{Path, PathBuf}; use tokio::process::{Child, Command}; +#[path = "workflow_live.rs"] +mod workflow_live; + #[tokio::test] #[ignore = "requires isolated local relay and built buzz-acp/buzz binaries"] async fn shared_mentions_wake_offline_owners_agents_and_handoff_in_the_original_thread() { @@ -101,10 +104,14 @@ fn harness( ) .unwrap(); let log = std::fs::File::create(directory.join("harness.log")).unwrap(); - let channel = - buzz_sdk::sharing::task::scope(root, keys("55").public_key(), Timestamp::now().as_secs()) - .unwrap() - .channel; + let channel = root + .tags + .iter() + .find_map(|tag| tag.as_slice().strip_prefix(&["h".to_owned()])) + .and_then(|values| values.first()) + .unwrap() + .parse::() + .unwrap(); let mut command = Command::new(acp); for (name, _) in std::env::vars() .filter(|(name, _)| name.starts_with("BUZZ_") || name.starts_with("SHARED_FIXTURE_")) diff --git a/crates/buzz-sandbox-reaper/src/mentions/tests/workflow_live.rs b/crates/buzz-sandbox-reaper/src/mentions/tests/workflow_live.rs new file mode 100644 index 00000000000..a1e8844676f --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/mentions/tests/workflow_live.rs @@ -0,0 +1,220 @@ +//! The real relay scheduler signs the trigger; neither owner nor desktop sends +//! a wake request. The manager and shipped harness must both authorize it. + +use super::*; + +#[tokio::test] +#[ignore = "requires isolated local relay and built buzz-acp/buzz binaries"] +async fn scheduled_workflow_wakes_private_remote_agent_with_owner_offline() { + let relay = std::env::var("BUZZ_MENTION_TEST_RELAY").unwrap(); + let url = url::Url::parse(&relay).unwrap(); + assert_eq!(url.scheme(), "ws"); + assert!(matches!(url.host_str(), Some("127.0.0.1" | "localhost"))); + let acp = PathBuf::from(std::env::var("BUZZ_SHARING_TEST_ACP").unwrap()); + let cli = PathBuf::from(std::env::var("BUZZ_SHARING_TEST_CLI").unwrap()); + assert!(acp.is_file() && cli.is_file()); + tokio::time::timeout(Duration::from_secs(150), exercise(&relay, &acp, &cli)) + .await + .expect("scheduled remote-agent workflow timed out"); +} + +async fn exercise(relay: &str, acp: &Path, cli: &Path) { + let (owner, observer, agent) = (keys("33"), keys("44"), keys("22")); + let channel = uuid::Uuid::new_v4(); + let mut phone = NostrWsConnection::connect_authenticated(relay, &owner, None) + .await + .unwrap(); + for member in [observer.public_key(), keys("11").public_key()] { + publish( + &mut phone, + EventBuilder::new(Kind::Custom(9030), "").tag(Tag::public_key(member)), + &owner, + ) + .await; + } + publish(&mut phone, EventBuilder::new(Kind::Metadata, "{}"), &owner).await; + publish( + &mut phone, + buzz_sdk::build_create_channel( + channel, + &format!("scheduled-{channel}"), + Some(buzz_sdk::Visibility::Private), + Some(buzz_sdk::ChannelKind::Stream), + None, + None, + ) + .unwrap(), + &owner, + ) + .await; + for (member, role) in [ + (observer.public_key(), None), + (agent.public_key(), Some(buzz_sdk::MemberRole::Bot)), + ] { + publish( + &mut phone, + buzz_sdk::build_add_member(channel, &member.to_hex(), role).unwrap(), + &owner, + ) + .await; + } + let auth = buzz_sdk::nip_oa::compute_auth_tag(&owner, &agent.public_key(), "").unwrap(); + let tag = buzz_sdk::nip_oa::parse_auth_tag(&auth).unwrap(); + let mut agent_socket = NostrWsConnection::connect_authenticated(relay, &agent, Some(&tag)) + .await + .unwrap(); + publish( + &mut agent_socket, + EventBuilder::new(Kind::Metadata, r#"{"name":"ScheduledBrokkr"}"#).tag(tag), + &agent, + ) + .await; + drop(agent_socket); + publish( + &mut phone, + EventBuilder::new(Kind::Custom(30177), "{}").tags([ + Tag::parse(["d", &agent.public_key().to_hex()]).unwrap(), + Tag::parse(["runtime", "voltage.remote-workspace/sandbox"]).unwrap(), + ]), + &owner, + ) + .await; + // Deliberately no sharing grant: an owner's workflow can use their private + // agent. Teammate workflows still require the existing current grant. + let cluster = sandbox_fake::Shared::default(); + register(&cluster, relay).await; + let config = ManagerConfig::parse(&json!({"version":1,"shared_tasks":true, + "checkpoint_bucket":"fixture-checkpoints","developers":{"dev":{ + "owner_pubkey":owner.public_key(),"namespace":NS, + "profile_secret_arn":"arn:aws:secretsmanager:us-west-2:000000000000:secret:voltage/staging/agents/dev/profile-AbCdEf" + }}}).to_string()).unwrap(); + let manager = Arc::new( + Manager::new( + keys("11"), + config, + sandbox_fake::client(&cluster), + crate::enroll::tests::offline(Default::default()), + ) + .with_sandbox_defaults( + json!({"image":format!("example.invalid/agent@sha256:{}","a".repeat(64)), + "pod_options":{"active_deadline_seconds":3600}}) + .as_object() + .unwrap() + .clone(), + ), + ); + let cancel = CancellationToken::new(); + let server = { + let cancel = cancel.clone(); + let relay = relay.to_owned(); + tokio::spawn(async move { + crate::serve::run_services(manager, &relay, None, true, cancel.cancelled()).await; + }) + }; + wait_for(|| !inbox(&cluster).is_null()).await; + let workflow_id = uuid::Uuid::new_v4(); + let yaml = "name: Scheduled remote agent\ntrigger:\n on: schedule\n cron: '* * * * *'\nsteps:\n - id: work\n action: send_message\n text: '@ScheduledBrokkr scheduled fixture task'\n"; + let definition = publish( + &mut phone, + buzz_sdk::build_workflow_def(channel, workflow_id, yaml).unwrap(), + &owner, + ) + .await; + drop(phone); + let mut observer_socket = NostrWsConnection::connect_authenticated(relay, &observer, None) + .await + .unwrap(); + let prompt = tokio::time::timeout(Duration::from_secs(75), async { + loop { + if let Some(event) = channel_events(&mut observer_socket, channel) + .await + .into_iter() + .find(|event| event.content.contains("scheduled fixture task")) + { + return event; + } + tokio::time::sleep(Duration::from_secs(1)).await; + } + }) + .await + .expect("relay scheduler did not fire"); + assert_ne!(prompt.pubkey, owner.public_key()); + assert_eq!( + buzz_sdk::workflow::owner(&prompt, prompt.pubkey, agent.public_key()), + Some(owner.public_key()) + ); + // Stop future fires before checking the harness, including on later failure. + let mut phone = NostrWsConnection::connect_authenticated(relay, &owner, None) + .await + .unwrap(); + publish( + &mut phone, + buzz_sdk::build_workflow_update( + channel, + workflow_id, + &format!("enabled: false\n{yaml}"), + &definition.id.to_hex(), + ) + .unwrap(), + &owner, + ) + .await; + drop(phone); + wait_for(|| completed(&cluster, &prompt)).await; + assert_eq!(cluster.lock().unwrap().sandboxes.len(), 1); + { + let cluster = cluster.lock().unwrap(); + let sandbox = cluster.sandboxes.values().next().unwrap(); + let annotations = &sandbox["metadata"]["annotations"]; + assert_eq!( + annotations[buzz_backend_kubernetes::lifecycle::CHANNEL], + channel.to_string() + ); + assert_eq!( + annotations[buzz_backend_kubernetes::lifecycle::THREAD], + prompt.id.to_hex() + ); + } + let directory = tempfile::Builder::new() + .prefix("buzz-scheduled-agent-") + .tempdir() + .unwrap() + .keep(); + eprintln!("Scheduled harness evidence: {}", directory.display()); + let process = harness( + (acp, cli), + &directory, + relay, + (&owner, &agent), + &prompt, + None, + ); + let reply = wait_message( + &mut observer_socket, + channel, + agent.public_key(), + "shared fixture completed", + ) + .await; + let ancestry: Vec<_> = reply + .tags + .iter() + .map(Tag::as_slice) + .filter(|tag| { + tag.first().map(String::as_str) == Some("e") + && matches!(tag.get(3).map(String::as_str), Some("root" | "reply")) + }) + .collect(); + assert!(!ancestry.is_empty()); + assert!(ancestry + .iter() + .all(|tag| tag.get(1) == Some(&prompt.id.to_hex()))); + assert_eq!( + std::fs::read_to_string(directory.join("model-turns")).unwrap(), + "1" + ); + stop(process).await; + std::fs::remove_dir_all(directory).unwrap(); + cancel.cancel(); + tokio::time::timeout(BOUND, server).await.unwrap().unwrap(); +} diff --git a/crates/buzz-sandbox-reaper/src/shared_budget/tests.rs b/crates/buzz-sandbox-reaper/src/shared_budget/tests.rs index 297a9ebb7fb..81d543588bf 100644 --- a/crates/buzz-sandbox-reaper/src/shared_budget/tests.rs +++ b/crates/buzz-sandbox-reaper/src/shared_budget/tests.rs @@ -13,6 +13,7 @@ fn task() -> Task { root: id, }, requester: human, + origin_requester: human, agent, owner: human, origin: Origin { diff --git a/crates/buzz-sdk/src/lib.rs b/crates/buzz-sdk/src/lib.rs index 295a67d2a95..42af4c7e733 100644 --- a/crates/buzz-sdk/src/lib.rs +++ b/crates/buzz-sdk/src/lib.rs @@ -17,6 +17,7 @@ pub mod builders; pub mod mentions; pub mod nip_oa; pub mod sharing; +pub mod workflow; pub use builders::*; diff --git a/crates/buzz-sdk/src/sharing/budget.rs b/crates/buzz-sdk/src/sharing/budget.rs index ed4394387dd..c5178966df2 100644 --- a/crates/buzz-sdk/src/sharing/budget.rs +++ b/crates/buzz-sdk/src/sharing/budget.rs @@ -68,7 +68,7 @@ impl Budget { receipts: vec![Receipt { event: task.origin.request, agent: task.origin.agent, - requester: task.human, + requester: task.origin_requester, handoff: false, }], }; @@ -106,7 +106,7 @@ impl Budget { let first = &self.receipts[0]; if first.event != self.request || first.agent != self.agent - || first.requester != self.human + || first.requester != task.origin_requester || first.handoff { return Err(Error::Journal); diff --git a/crates/buzz-sdk/src/sharing/budget/tests.rs b/crates/buzz-sdk/src/sharing/budget/tests.rs index ac4629e0eb8..d5363906c28 100644 --- a/crates/buzz-sdk/src/sharing/budget/tests.rs +++ b/crates/buzz-sdk/src/sharing/budget/tests.rs @@ -12,6 +12,7 @@ pub fn task() -> Task { root: id, }, requester: human, + origin_requester: human, agent, owner: human, origin: Origin { diff --git a/crates/buzz-sdk/src/sharing/policy.rs b/crates/buzz-sdk/src/sharing/policy.rs index b1a455f2864..74a5233836a 100644 --- a/crates/buzz-sdk/src/sharing/policy.rs +++ b/crates/buzz-sdk/src/sharing/policy.rs @@ -29,13 +29,15 @@ pub struct Task { pub scope: Scope, /// The actual signed requester, never replaced with the authorizing owner. pub requester: PublicKey, + /// Actual signer of the original task, including a verified relay delegation. + pub origin_requester: PublicKey, /// Addressed agent. pub agent: PublicKey, /// Target agent's current owner. pub owner: PublicKey, /// Original human request and preceding admission. pub origin: Origin, - /// Original human identity. + /// Original authorizing human, including a verified workflow owner. pub human: PublicKey, /// Initial agent's owner; chooses its existing namespace for the journal. pub origin_owner: PublicKey, @@ -62,6 +64,7 @@ where Fut: Future, ()>>, { let scope = task::scope(event, agent, now).map_err(|_| Error::Denied("invalid mention"))?; + let requester = crate::workflow::owner(event, relay, agent).unwrap_or(event.pubkey); let stored = one( &mut query, json!({"kinds":[9,40002], "ids":[event.id], "#h":[scope.channel], "limit":1}), @@ -102,14 +105,16 @@ where }) { return Err(Error::Denied("conversation cannot be shared")); } - for member in [owner, agent, event.pubkey] { + for member in [owner, agent, requester] { member_role(&roster, member)?; } let target = profile(&mut query, agent).await?; if profile_owner(&target, now).ok() != Some(owner) { return Err(Error::Denied("agent ownership changed")); } - grant(&mut query, owner, agent, &scope, event, now).await?; + if requester != owner { + grant(&mut query, owner, agent, &scope, event, now).await?; + } // Channel roles and self-authored profiles can change. Only a direct // workspace member may originate a human budget; an agent cannot gain a // fresh budget by stripping its profile credential or rejoining as member. @@ -125,13 +130,13 @@ where { return Err(Error::Unavailable); } - let requester = profile(&mut query, event.pubkey).await?; - let automated = is_agent(&requester, member_role(&roster, event.pubkey)?, now)?; + let requester_profile = profile(&mut query, requester).await?; + let automated = is_agent(&requester_profile, member_role(&roster, requester)?, now)?; let claimed = task::origin(event).map_err(|_| Error::Denied("invalid task ancestry"))?; let origin = if automated { claimed.ok_or(Error::Denied("agent handoff requires task ancestry"))? } else { - workspace_human(&humans, event.pubkey)?; + workspace_human(&humans, requester)?; if claimed.is_some() { return Err(Error::Denied("human tasks cannot claim agent ancestry")); } @@ -159,11 +164,12 @@ where { return Err(Error::Denied("request is outside this task")); } - let human = profile(&mut query, root.pubkey).await?; - if is_agent(&human, member_role(&roster, root.pubkey)?, now)? { + let human_key = crate::workflow::owner(&root, relay, origin.agent).unwrap_or(root.pubkey); + let human = profile(&mut query, human_key).await?; + if is_agent(&human, member_role(&roster, human_key)?, now)? { return Err(Error::Denied("an agent cannot reset the human budget")); } - workspace_human(&humans, root.pubkey)?; + workspace_human(&humans, human_key)?; if automated { ancestry(&mut query, event, &root, origin, scope, now).await?; } @@ -172,16 +178,17 @@ where let origin_owner = profile_owner(&initial, now).map_err(|_| Error::Denied("invalid initial agent owner"))?; member_role(&roster, origin_owner)?; - if root.pubkey != origin_owner { + if human_key != origin_owner { grant(&mut query, origin_owner, origin.agent, &scope, &root, now).await?; } Ok(Task { scope, requester: event.pubkey, + origin_requester: root.pubkey, agent, owner, origin, - human: root.pubkey, + human: human_key, origin_owner, trigger: event.id, handoff: automated, diff --git a/crates/buzz-sdk/src/sharing/policy/tests.rs b/crates/buzz-sdk/src/sharing/policy/tests.rs index b61bb0ace53..86fc23a1a99 100644 --- a/crates/buzz-sdk/src/sharing/policy/tests.rs +++ b/crates/buzz-sdk/src/sharing/policy/tests.rs @@ -7,6 +7,9 @@ use std::{ }; use uuid::Uuid; +#[path = "workflow_tests.rs"] +mod workflows; + struct Fixture { relay: Keys, owner: Keys, diff --git a/crates/buzz-sdk/src/sharing/policy/workflow_tests.rs b/crates/buzz-sdk/src/sharing/policy/workflow_tests.rs new file mode 100644 index 00000000000..20b0a0a0fbf --- /dev/null +++ b/crates/buzz-sdk/src/sharing/policy/workflow_tests.rs @@ -0,0 +1,172 @@ +use super::*; + +fn prompt(f: &Fixture, signer: &Keys, owner: PublicKey, literal: bool, at: u64) -> Event { + let mut tags = vec![ + vec!["h".into(), f.channel.to_string()], + vec!["p".into(), owner.to_hex()], + vec!["p".into(), f.agent.public_key().to_hex()], + vec!["buzz:workflow".into(), "true".into()], + vec!["buzz:workflow-owner".into(), owner.to_hex()], + ]; + if literal { + tags.push(vec![ + "buzz:workflow-mention".into(), + f.agent.public_key().to_hex(), + ]); + } + signed(signer, 9, tags, "scheduled task", at) +} + +fn without_grant(f: &mut Fixture) { + f.events + .retain(|event| !(event.kind.as_u16() == 30180 && event.pubkey == f.owner.public_key())); +} + +#[test] +fn owner_workflow_wakes_private_agent_and_retains_relay_signature_provenance() { + let mut f = Fixture::new(); + without_grant(&mut f); + for at in [101, 102] { + let event = prompt(&f, &f.relay, f.owner.public_key(), true, at); + f.events.push(event.clone()); + let task = f + .check(&event, f.owner.public_key(), f.agent.public_key()) + .unwrap(); + assert_eq!(task.requester, f.relay.public_key()); + assert_eq!(task.origin_requester, f.relay.public_key()); + assert_eq!(task.human, f.owner.public_key()); + assert_eq!(task.origin.request, event.id); + assert_eq!(task.scope.root, event.id); + assert!(!task.handoff); + let mut budget = crate::sharing::budget::Budget::new(&task, 103).unwrap(); + assert!(!budget.admit(&task, 103).unwrap()); + let bytes = serde_json::to_vec(&budget).unwrap(); + let mut recovered: crate::sharing::budget::Budget = serde_json::from_slice(&bytes).unwrap(); + assert!(!recovered.admit(&task, 104).unwrap()); + } +} + +#[test] +fn workflow_handoffs_keep_the_authorizing_human_and_the_original_budget() { + let mut f = Fixture::new(); + without_grant(&mut f); + let event = prompt(&f, &f.relay, f.owner.public_key(), true, 101); + f.events.push(event.clone()); + let initial = f + .check(&event, f.owner.public_key(), f.agent.public_key()) + .unwrap(); + let mut budget = crate::sharing::budget::Budget::new(&initial, 103).unwrap(); + let handoff = task::handoff( + &event, + f.agent.public_key(), + f.peer.public_key(), + "review this", + 103, + ) + .unwrap() + .custom_created_at(Timestamp::from(102)) + .sign_with_keys(&f.agent) + .unwrap(); + f.events.push(handoff.clone()); + let next = f + .check(&handoff, f.other.public_key(), f.peer.public_key()) + .unwrap(); + assert_eq!(next.requester, f.agent.public_key()); + assert_eq!(next.origin_requester, f.relay.public_key()); + assert_eq!(next.human, f.owner.public_key()); + assert_eq!(next.origin.request, event.id); + assert!(next.handoff); + assert!(budget.admit(&next, 103).unwrap()); +} + +#[test] +fn another_owners_workflow_requires_a_current_sharing_grant() { + let mut f = Fixture::new(); + let event = prompt(&f, &f.relay, f.other.public_key(), true, 101); + f.events.push(event.clone()); + let task = f + .check(&event, f.owner.public_key(), f.agent.public_key()) + .unwrap(); + assert_eq!(task.human, f.other.public_key()); + assert_eq!(task.requester, f.relay.public_key()); + without_grant(&mut f); + assert!(matches!( + f.check(&event, f.owner.public_key(), f.agent.public_key()), + Err(Error::Denied(_)) + )); +} + +#[test] +fn forged_delegation_or_rendered_only_mentions_cannot_wake_a_private_agent() { + let mut f = Fixture::new(); + without_grant(&mut f); + for event in [ + prompt(&f, &f.other, f.owner.public_key(), true, 101), + prompt(&f, &f.relay, f.owner.public_key(), false, 101), + prompt(&f, &f.relay, f.peer.public_key(), true, 101), + ] { + f.events.push(event.clone()); + assert!(matches!( + f.check(&event, f.owner.public_key(), f.agent.public_key()), + Err(Error::Denied(_)) + )); + } +} + +#[test] +fn a_queued_workflow_rechecks_membership_archive_and_workspace_access() { + for revoke in ["owner", "agent", "archive", "workspace"] { + let mut f = Fixture::new(); + without_grant(&mut f); + let event = prompt(&f, &f.relay, f.owner.public_key(), true, 101); + f.events.push(event.clone()); + assert!(f + .check(&event, f.owner.public_key(), f.agent.public_key()) + .is_ok()); + match revoke { + "archive" => { + f.events.retain(|e| e.kind.as_u16() != 39000); + f.events.push(signed( + &f.relay, + 39000, + vec![ + vec!["d".into(), f.channel.to_string()], + vec!["archived".into(), "true".into()], + ], + "", + 102, + )); + } + "workspace" => { + f.events.retain(|e| e.kind.as_u16() != 13534); + f.events.push(signed(&f.relay, 13534, vec![], "", 102)); + } + _ => { + let removed = if revoke == "owner" { + f.owner.public_key() + } else { + f.agent.public_key() + }; + let index = f + .events + .iter() + .position(|e| e.kind.as_u16() == 39002) + .unwrap(); + let tags = f.events[index] + .tags + .iter() + .filter(|t| t.as_slice().get(1) != Some(&removed.to_hex())) + .map(|t| t.as_slice().to_vec()) + .collect(); + f.events[index] = signed(&f.relay, 39002, tags, "", 102); + } + } + assert!( + matches!( + f.check(&event, f.owner.public_key(), f.agent.public_key()), + Err(Error::Denied(_)) + ), + "{revoke}" + ); + } +} diff --git a/crates/buzz-sdk/src/sharing/task.rs b/crates/buzz-sdk/src/sharing/task.rs index 9ceb53b26bb..d9ffd5d047f 100644 --- a/crates/buzz-sdk/src/sharing/task.rs +++ b/crates/buzz-sdk/src/sharing/task.rs @@ -19,7 +19,7 @@ pub struct Scope { #[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)] #[serde(deny_unknown_fields)] pub struct Origin { - /// The original human's signed task message. + /// The original human task, signed directly or delegated by the verified relay. pub request: EventId, /// The agent that human initially addressed. pub agent: PublicKey, diff --git a/crates/buzz-sdk/src/workflow.rs b/crates/buzz-sdk/src/workflow.rs new file mode 100644 index 00000000000..ce72e314478 --- /dev/null +++ b/crates/buzz-sdk/src/workflow.rs @@ -0,0 +1,51 @@ +//! Verified owner delegation carried by relay-authored workflow prompts. + +use std::collections::HashSet; + +use nostr::{Event, PublicKey, Tag}; + +/// Identify the workflow owner only when the active relay signed a kind:9 +/// prompt explicitly targeting this agent in the owner's stored template. +/// Routing `p` tags alone never convey delegated authority. Callers obtain +/// `relay` from the authenticated community's NIP-11 `self` key. +pub fn owner(event: &Event, relay: PublicKey, agent: PublicKey) -> Option { + if event.kind.as_u16() as u32 != crate::kind::KIND_STREAM_MESSAGE + || event.pubkey != relay + || event.verify().is_err() + { + return None; + } + let tags = |name: &str| { + event + .tags + .iter() + .map(Tag::as_slice) + .filter(|tag| tag.first().map(String::as_str) == Some(name)) + .collect::>() + }; + if tags("buzz:workflow").as_slice() != [["buzz:workflow", "true"]] { + return None; + } + let owners = tags("buzz:workflow-owner"); + let [owner] = owners.as_slice() else { + return None; + }; + let [_, owner] = *owner else { + return None; + }; + let parsed = PublicKey::from_hex(owner).ok()?; + if parsed.to_hex() != *owner { + return None; + } + let mut mentioned = HashSet::new(); + for tag in tags("buzz:workflow-mention") { + let [_, value] = tag else { + return None; + }; + let key = PublicKey::from_hex(value).ok()?; + if key.to_hex() != *value || !mentioned.insert(key) { + return None; + } + } + mentioned.contains(&agent).then_some(parsed) +} diff --git a/docs/shared-channel-agents.md b/docs/shared-channel-agents.md index 7849be18020..cb0a84bb233 100644 --- a/docs/shared-channel-agents.md +++ b/docs/shared-channel-agents.md @@ -43,6 +43,39 @@ parent event. A durable manager journal permits four handoffs total and at most two admissions per agent under one human-to-initial-agent task. A duplicate admission consumes no additional budget. The task expires after seven days. +## Schedule remote work + +In desktop, open **Workflows**, choose a channel, and create a workflow with a +**Schedule** trigger and a **Send message** step. Put an explicit `@AgentName` +mention and the task instructions in the saved step text. The schedule editor +supports hourly, daily, weekly, and custom cron schedules. Schedule times use UTC. + +The relay runs the schedule while the owner's desktop is closed. Leave the step's +reply target unset so each fire starts a new channel thread and wakes the remote +agent through the ordinary mention listener. Register the agent from desktop once. +Keep its runtime set to Remote workspace. Both owner and agent must remain explicit +channel members. This path requires the matching manager and harness versions and the +existing `shared_tasks: true` configuration described below. + +An owner's workflow can invoke their private agent without a channel sharing +grant. A teammate's workflow requires the target agent's current channel grant. +Every admission checks current ownership, channel and workspace membership, and +archive status. Existing task budgets also apply to workflow handoffs. + +Only the active relay's signed workflow proof delegates the workflow owner's +authority. A mention inserted through trigger-controlled template substitutions +does not delegate it. The signed event and budget receipt keep the relay as the +actual signer; the verified workflow owner supplies the human authorization. + +Disable the workflow to stop future scheduled messages. This does not cancel an +already emitted task. An unthreaded fire has a separate root and budget. Schedules +use the existing server scheduler and durable fire claims. Remote sessions use +their normal lifecycle and the configured compute and model budget. + +To test after rollout, schedule a short task that replies in the channel, close +desktop, and wait for the scheduled time. Confirm a remote session starts and +replies beneath that scheduled message. Disable the workflow after the test. + ## Rollout The implementation is off by default. Before enabling it: @@ -85,3 +118,9 @@ This runs real relay admission, wake listeners, harness processes, and CLI commands for two owners and two agents, followed by revocation. A deterministic ACP adapter replaces the model and a simulated Kubernetes API replaces the cluster. It does not validate a real image pull, cloud Pod, or external model. + +With the same environment variables, run +`cargo test -p buzz-sandbox-reaper scheduled_workflow_wakes -- --ignored --nocapture` +to exercise an actual relay cron fire, private-agent wake, and a reply from the +shipped harness while the owner's connection is closed. The local test uses a +one-minute cron schedule and can take approximately two minutes.