Skip to content

Repository navigation

feat(agents): wake remote agents from scheduled workflows #119

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tee8z merged 1 commit into from
Nov 4, 2026
Merged

feat(agents): wake remote agents from scheduled workflows #119

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 6 additions & 61 deletions crates/buzz-acp/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -449,64 +449,9 @@ fn verified_workflow_owner(
relay_self: Option<&str>,
agent_pubkey_hex: &str,
) -> Option<String> {
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.
Expand Down Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions crates/buzz-acp/src/pool.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand Down
39 changes: 34 additions & 5 deletions crates/buzz-acp/src/sharing.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)]
Expand Down Expand Up @@ -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<Option<String>, Error> {
pub(crate) async fn authorize(&self, event: &Event) -> Result<Admission, Error> {
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<Option<String>, Error> {
async fn exchange(&self, trigger: &Event) -> Result<Admission, Error> {
let auth = buzz_sdk::nip_oa::parse_auth_tag(&self.auth).map_err(|_| Error)?;
let request = Request::new(
Op::Admit,
Expand Down Expand Up @@ -148,15 +157,15 @@ 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);
}
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),
_ => {}
Expand Down Expand Up @@ -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<PublicKey, Error> {
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) {
Expand Down
62 changes: 53 additions & 9 deletions crates/buzz-acp/src/sharing/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ async fn encrypted_admission_uses_fresh_ids_and_request_scoped_reply_filters() {
let requests = Arc::new(Mutex::new(Vec::<String>::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();
Expand Down Expand Up @@ -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);
Expand All @@ -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;
Expand Down Expand Up @@ -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();
Expand All @@ -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();
Expand Down
15 changes: 11 additions & 4 deletions crates/buzz-sandbox-reaper/src/mentions/tests/shared_live.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down Expand Up @@ -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::<uuid::Uuid>()
.unwrap();
let mut command = Command::new(acp);
for (name, _) in std::env::vars()
.filter(|(name, _)| name.starts_with("BUZZ_") || name.starts_with("SHARED_FIXTURE_"))
Expand Down
Loading
Loading