jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(acp): wake agents from workflow messages (#6953)

Browse Source 
> Pinky, an AI agent, is opening this PR on Wes's behalf.

## Summary

Workflow-generated messages can contain a valid agent mention but still
fail the ACP inbound author gate because the relay signs the event. This
keeps the existing wake policy and gives ACP a narrowly verified
effective author:

- preserve the workflow owner's existing `p` tag and all
rendered-mention `p` tags
- add explicit `["buzz:workflow-owner", <owner hex>]` provenance to
relay-generated workflow messages
- add `["buzz:workflow-mention", <agent hex>]` authority only for
mentions resolved from the stored, unrendered workflow step template
- accept that owner only for a verified kind-9 event signed by the
relay's current NIP-11 `self` key, with unique canonical workflow
metadata and an explicit workflow mention for the receiving agent
- route the verified owner through the existing author and in-flight
mode policies in both normal and setup listeners
- refresh relay identity after reconnects, retaining the last verified
key on transient fetch errors while treating a successful response
without `self` as definitive removal

Malformed, duplicate, forged, tampered, wrong-kind, and wrong-relay
attribution all fail closed to the raw event signer. `respond-to=nobody`
remains absolute. Old/mixed-version messages without the explicit
provenance retain their current fail-closed behavior.

## Trust boundary

The workflow owner means **“scheduled by,” not “authored every rendered
word.”** Trigger-controlled substitutions may still produce ordinary `p`
mention routing for compatibility, but they cannot mint
`buzz:workflow-mention` authority. Only a target named in the durable
owner-authored step template can receive that authority.

The author gate is not bypassed: after relay signature/provenance
verification, the effective owner is evaluated under the same
`owner-only`, `allowlist`, DM, and `nobody` policies used for ordinary
messages. Owner control commands continue to use the raw event signer.

## Why this PR

This is the focused immediate fix for waking an **online** agent from a
stored workflow mention. Earlier attempts were not a finished mergeable
fix and had materially different or incomplete trust designs. Larry's
larger draft stack addresses durable delivery across restarts; that
remains valuable future work and can supersede this effective-author
path when it lands.

## Validation

At exact clean commit `fe5b55619fe44176343eefb4cb7fe180df45a7d8`:

- `buzz-relay workflow_sink`: 25/25 passed, including all four ignored
PostgreSQL cases
- `buzz-acp --lib`: 845/845 passed
- `buzz-workflow --lib`: 169/169 passed (2 unrelated PostgreSQL tests
ignored)
- warnings-denied Clippy passed for the changed Rust packages
- `cargo fmt --all -- --check` passed
- `git diff --check` passed
- repository pre-push gates passed, including branch-scoped Rust tests
- CI now selects the ACP library tests and the relay's pure + PostgreSQL
workflow-sink tests so these guards cannot silently remain unexecuted

The production event-to-author gate is shared by normal and setup
listeners and has biting regression tests for accepted explicit
attribution, legacy owner-`p` rejection, and forged-attribution
rejection.

## Exact-head local relay + ACP proof

Following the release-binary/local-relay shape in `TESTING.md`, the
exact commit above passed a fresh isolated real-process matrix using:

- a freshly recreated Postgres database with migrations
- isolated Redis
- exact-head release `buzz-relay`, `buzz`, `buzz-admin`, and `buzz-acp`
binaries
- newly provisioned owner, channel, and bot member through the CLI
- workflow creation and triggering through the running relay
- a deterministic ACP protocol subprocess capturing actual
`session/prompt` dispatches
- a NIP-11 `self` value verified against the running relay signer

Cases:

1. A stored explicit workflow mention woke an `owner-only` agent exactly
once.
2. A workflow message without an agent mention did not wake it.
3. A non-relay signer forging every workflow authority tag did not wake
it.
4. Trigger-controlled `{{trigger.text}}` containing `@Wake Agent`
retained ordinary `p` routing but received no authority-bearing
workflow-mention tag and did not wake the agent.
5. `respond-to=nobody` remained absolute for a valid relay-authenticated
workflow mention.

The deterministic ACP subprocess isolates and directly proves relay →
ACP authorization and prompt dispatch without depending on external
model behavior.

## Deployment and residual risk

Relay and ACP changes must be deployed together for the new wake
behavior; mixed versions fail closed. Production paired-deployment proof
remains distinct from the successful local integration run. Setup-mode
behavior has automated coverage but was not a separate case in the
five-case local matrix. Relay-key rotation is observed at ACP
startup/reconnect; transient NIP-11 errors retain the last verified key,
an intentional availability tradeoff documented in code.

---------

Signed-off-by: Pinky <5f5ab050ec58ae208332edd544ebf705221e24c1b86d82a6ca07038a7a8f6ac9@buzz.block.builderlab.xyz>
Signed-off-by: Brain <1a02c72794dcd0f07058a353bc3a81f4028b8c77c92c87fce6d5c8b85970a20b@buzz.block.builderlab.xyz>
Signed-off-by: Wes <wesbillman@users.noreply.github.com>
Co-authored-by: Pinky <5f5ab050ec58ae208332edd544ebf705221e24c1b86d82a6ca07038a7a8f6ac9@buzz.block.builderlab.xyz>
Co-authored-by: LioLionel <62820906+LioLionel@users.noreply.github.com>
Co-authored-by: Brain <1a02c72794dcd0f07058a353bc3a81f4028b8c77c92c87fce6d5c8b85970a20b@buzz.block.builderlab.xyz>
Co-authored-by: Carl <9d00794d3df50972eb8b615511783cab12a77a8fd5dd5edd58073ec73b54bd8b@buzz.block.builderlab.xyz>
This commit is contained in:
Wes
2026-08-31 14:24:09 -06:00
committed by GitHub
parent bc006f6708
commit 93237b4a74
10 changed files with 2593 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
crates/buzz-acp/README.md
View File

@@ -147,6 +147,15 @@ Controls which authors' events the harness forwards to the agent. Events from di
| `anyone` | Forward all events (no author filtering). |
| `nobody` | Drop all inbound events. Agent only acts on heartbeat prompts. |
Relay-signed workflow messages delegate to their recorded owner only when they
explicitly target this agent with authenticated workflow-mention provenance.
The owner tag means that owner scheduled the workflow; it does not claim that
the owner authored every word after template rendering. ACP verifies the
provenance against the relay's NIP-11 `self` key, then evaluates the owner under
the same author policy as ordinary messages. Legacy workflow messages and
workflow output without an explicit agent mention remain attributed to the relay
signer. `nobody` remains absolute.
The gate applies to **all** inbound events — @mentions, DMs, thread replies, and any event delivered by the relay. Owner control commands are checked **before** the gate, so the owner can still manage the harness regardless of mode:
| Command | Effect |

1966
crates/buzz-acp/src/lib.rs
View File

File diff suppressed because it is too large Load Diff

223
crates/buzz-acp/src/relay.rs
View File

@@ -277,6 +277,75 @@ fn unix_now_secs() -> u64 {
}
impl RestClient {
/// Fetch the relay's stable signing identity from its NIP-11 document.
///
/// Relay-authored workflow attribution is trusted only when the event signer
/// matches this key. Missing, malformed, or unavailable identity data fails
/// closed by returning an error/`None` to the caller. NIP-11 is standardized
/// at the relay root; `/info` remains a compatibility fallback for relays
/// that expose the document through Buzz's explicit alias.
pub async fn relay_self(&self) -> Result<Option<String>, RelayError> {
let mut failures = Vec::new();
let mut saw_document_without_self = false;
for path in ["/", "/info"] {
let url = format!("{}{path}", self.base_url);
let response = match self
.http
.get(&url)
.header(reqwest::header::ACCEPT, "application/nostr+json")
.send()
.await
{
Ok(response) => response,
Err(error) => {
failures.push(format!("GET {path} failed: {error}"));
continue;
}
};
if !response.status().is_success() {
failures.push(format!("GET {path} returned HTTP {}", response.status()));
continue;
}
let document: serde_json::Value = match response.json().await {
Ok(document) => document,
Err(error) => {
failures.push(format!("GET {path} returned invalid NIP-11 JSON: {error}"));
continue;
}
};
let Some(relay_self) = document.get("self") else {
saw_document_without_self = true;
continue;
};
let Some(relay_self) = relay_self.as_str() else {
failures.push(format!("GET {path} returned a non-string NIP-11 self key"));
continue;
};
let relay_self = match nostr::PublicKey::from_hex(relay_self) {
Ok(pubkey) => pubkey.to_hex(),
Err(error) => {
failures.push(format!(
"GET {path} returned an invalid NIP-11 self key: {error}"
));
continue;
}
};
return Ok(Some(relay_self));
}
if saw_document_without_self {
Ok(None)
} else {
Err(RelayError::Http(format!(
"failed to fetch a usable NIP-11 document: {}",
failures.join("; ")
)))
}
}
/// Sign a NIP-98 HTTP Auth event (kind:27235) for the given method/URL/body.
///
/// Returns the `Authorization: Nostr <base64>` header value (without the
@@ -515,6 +584,10 @@ impl RestClient {
/// Events the harness cares about.
#[derive(Debug, Clone)]
pub struct BuzzEvent {
/// Which authenticated relay connection delivered this event. Generation 0
/// is the initial connection; each successful reconnect increments it
/// before any buffered or live event from that connection is forwarded.
pub connection_generation: u64,
/// Which channel this event belongs to.
pub channel_id: Uuid,
/// The underlying Nostr event.
@@ -1140,6 +1213,10 @@ struct BgState {
/// A single failed channel REQ is parked here instead of aborting the whole
/// reconnect. Drained by the main loop. Flushed on each reconnect attempt.
resubscribe_retry: HashSet<Uuid>,
/// Current authenticated WebSocket generation. Incremented immediately
/// after each successful reconnect handshake, before buffered or live
/// events from the new connection are forwarded.
connection_generation: u64,
/// Current position in the exponential backoff ladder.
///
/// Persisted across calls to `wait_for_reconnect` so a flapping link stays at
@@ -1171,6 +1248,7 @@ impl BgState {
observer_in_flight: VecDeque::new(),
gated_observer_dropped: 0,
resubscribe_retry: HashSet::new(),
connection_generation: 0,
backoff_step: 0,
}
}
@@ -2189,6 +2267,7 @@ async fn handle_ws_message(
}
let ts = event.created_at.as_secs();
let buzz_event = BuzzEvent {
connection_generation: state.connection_generation,
channel_id: channel_uuid,
event: *event,
};
@@ -2230,6 +2309,7 @@ async fn handle_ws_message(
let event_id_hex = event.id.to_hex();
if state.record_event(channel_id, &event) {
let buzz_event = BuzzEvent {
connection_generation: state.connection_generation,
channel_id,
event: *event,
};
@@ -3013,6 +3093,7 @@ async fn try_autonomous_reconnect(
match do_connect(relay_url, keys, auth_tag).await {
Ok((new_ws, handshake_buffer)) => {
*ws = new_ws;
state.connection_generation = state.connection_generation.saturating_add(1);
info!("autonomous reconnect succeeded (attempt {})", attempt + 1);
let handshake_ok = process_handshake_buffer(
ws,
@@ -3151,6 +3232,7 @@ async fn wait_for_reconnect(
match do_connect(relay_url, keys, auth_tag).await {
Ok((new_ws, handshake_buffer)) => {
*ws = new_ws;
state.connection_generation = state.connection_generation.saturating_add(1);
info!("relay reconnected to {relay_url}");
let handshake_ok = process_handshake_buffer(
ws,
@@ -4084,6 +4166,147 @@ async fn wait_for_any_ok(
mod tests {
use super::*;
async fn nip11_test_client(
responses: HashMap<String, (u16, String)>,
) -> (
RestClient,
std::sync::Arc<std::sync::Mutex<Vec<(String, bool)>>>,
tokio::task::JoinHandle<()>,
) {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind NIP-11 test server");
let base_url = format!(
"http://{}",
listener.local_addr().expect("test server address")
);
let requests = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let server_requests = requests.clone();
let server = tokio::spawn(async move {
while let Ok((mut socket, _)) = listener.accept().await {
let mut request = vec![0; 8192];
let bytes_read = socket.read(&mut request).await.unwrap_or_default();
let request = String::from_utf8_lossy(&request[..bytes_read]);
let path = request
.lines()
.next()
.and_then(|line| line.split_whitespace().nth(1))
.unwrap_or("/")
.to_string();
let has_nip11_accept = request
.lines()
.any(|line| line.eq_ignore_ascii_case("accept: application/nostr+json"));
server_requests
.lock()
.expect("lock recorded NIP-11 requests")
.push((path.clone(), has_nip11_accept));
let (status, body) = responses
.get(&path)
.cloned()
.unwrap_or_else(|| (404, "not found".into()));
let reason = if status == 200 { "OK" } else { "Not Found" };
let response = format!(
"HTTP/1.1 {status} {reason}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
body.len(),
body
);
let _ = socket.write_all(response.as_bytes()).await;
}
});
let client = RestClient {
http: reqwest::Client::new(),
base_url,
keys: Keys::generate(),
auth_tag_json: None,
};
(client, requests, server)
}
#[tokio::test]
async fn relay_self_reads_and_normalizes_standard_root_document() {
let uppercase = "AB".repeat(32);
let responses = HashMap::from([
(
"/".to_string(),
(200, serde_json::json!({ "self": uppercase }).to_string()),
),
(
"/info".to_string(),
(
200,
serde_json::json!({ "self": "cd".repeat(32) }).to_string(),
),
),
]);
let (client, requests, server) = nip11_test_client(responses).await;
assert_eq!(
client.relay_self().await.expect("fetch relay self"),
Some("ab".repeat(32))
);
assert_eq!(
*requests.lock().expect("lock recorded requests"),
vec![("/".to_string(), true)],
"the standard root document should be preferred and request NIP-11 JSON"
);
server.abort();
}
#[tokio::test]
async fn relay_self_falls_back_to_info_alias() {
let responses = HashMap::from([
("/".to_string(), (404, "not found".into())),
(
"/info".to_string(),
(
200,
serde_json::json!({ "self": "cd".repeat(32) }).to_string(),
),
),
]);
let (client, requests, server) = nip11_test_client(responses).await;
assert_eq!(
client.relay_self().await.expect("fetch relay self"),
Some("cd".repeat(32))
);
assert_eq!(
*requests.lock().expect("lock recorded requests"),
vec![("/".to_string(), true), ("/info".to_string(), true)]
);
server.abort();
}
#[tokio::test]
async fn relay_self_rejects_malformed_identity_at_both_endpoints() {
let responses = HashMap::from([
(
"/".to_string(),
(
200,
serde_json::json!({ "self": "not-a-pubkey" }).to_string(),
),
),
(
"/info".to_string(),
(200, serde_json::json!({ "self": 42 }).to_string()),
),
]);
let (client, _requests, server) = nip11_test_client(responses).await;
let error = client
.relay_self()
.await
.expect_err("malformed relay identities must fail closed");
assert!(error
.to_string()
.contains("failed to fetch a usable NIP-11 document"));
server.abort();
}
#[test]
fn relay_ws_to_http_plain() {
assert_eq!(

255
crates/buzz-acp/src/setup_mode.rs
View File

@@ -71,10 +71,11 @@ pub(crate) enum AcpAvailabilityStatus {
}
use crate::{
author_allowed,
config::Config,
event_mentions_agent, filter,
relay::{HarnessRelay, RelayEventPublisher},
inbound_author_gate::AuthorizedListenerEvent,
relay::{self, HarnessRelay, RelayEventPublisher},
InboundAuthorGate, OwnerCache,
};
// ── Payload ───────────────────────────────────────────────────────────────────
@@ -342,6 +343,10 @@ pub(crate) async fn run_setup_listener(config: Config, payload: SetupPayload) ->
tracing::info!("setup-mode: connected and subscribed to membership notifications");
let rest_client = relay.rest_client();
let mut author_gate_ctx =
crate::InboundAuthorGate::connect(&rest_client, &pubkey_hex, "setup startup").await;
// Resolve owner for author-gate (same priority as normal mode).
let startup_owner = crate::resolve_agent_owner(&config);
let owner_cache = crate::OwnerCache::new(startup_owner);
@@ -381,7 +386,6 @@ pub(crate) async fn run_setup_listener(config: Config, payload: SetupPayload) ->
}
let publisher = relay.event_publisher();
let rest_client = relay.rest_client();
let channel_info = crate::pool::ChannelInfoResolver::new(channel_info_map, rest_client.clone());
@@ -428,80 +432,115 @@ pub(crate) async fn run_setup_listener(config: Config, payload: SetupPayload) ->
// Apply the same author gate as normal mode so the nudge only goes
// to authors the real agent would have answered. Same DM hardening:
// in DMs only owner/siblings get a nudge (fail-closed on unknown type).
let author_hex = buzz_event.event.pubkey.to_hex();
let is_dm = crate::is_dm_channel(buzz_event.channel_id, &channel_info).await;
let allowed = author_allowed(
let Some(authorized_event) = authorize_setup_listener_event(
&mut author_gate_ctx,
buzz_event,
&config.respond_to,
&config.respond_to_allowlist,
&author_hex,
is_dm,
&owner_cache,
&channel_info,
&rest_client,
)
.await;
.await
else {
continue;
};
// Apply channel/kind filter rules.
let filter_matched = filter::match_event(
&buzz_event.event,
buzz_event.channel_id,
if !nudge_authorized_event(
authorized_event,
&rules,
&pubkey_hex,
)
.await
.is_some();
// Pure gate: author gate verdict + event-id dedup.
if !should_nudge_for_event(
buzz_event.event.id,
allowed,
filter_matched,
&mut nudged_event_ids,
) {
continue;
}
// Build and publish the setup nudge.
if let Err(e) = publish_setup_nudge(
&publisher,
&config.keys,
buzz_event.channel_id,
&buzz_event.event,
&payload,
)
.await
{
tracing::warn!("setup-mode: failed to publish nudge: {e}");
} else {
tracing::info!(
channel_id = %buzz_event.channel_id,
event_id = %buzz_event.event.id,
"setup-mode: nudge published"
);
continue;
}
}
Ok(())
}
/// Outcome of the pure per-event gate checks in setup mode.
async fn nudge_authorized_event(
authorized_event: AuthorizedListenerEvent,
rules: &[filter::SubscriptionRule],
pubkey_hex: &str,
nudged_event_ids: &mut HashSet<EventId>,
publisher: &RelayEventPublisher,
keys: &nostr::Keys,
payload: &SetupPayload,
) -> bool {
let (buzz_event, effective_author) = authorized_event.into_parts();
// Apply channel/kind filter rules.
let filter_matched =
filter::match_event(&buzz_event.event, buzz_event.channel_id, rules, pubkey_hex)
.await
.is_some();
if !should_nudge_for_event(buzz_event.event.id, filter_matched, nudged_event_ids) {
return false;
}
// Build and publish the setup nudge.
if let Err(e) = publish_setup_nudge(
publisher,
keys,
buzz_event.channel_id,
&buzz_event.event,
&effective_author,
payload,
)
.await
{
tracing::warn!("setup-mode: failed to publish nudge: {e}");
} else {
tracing::info!(
channel_id = %buzz_event.channel_id,
event_id = %buzz_event.event.id,
"setup-mode: nudge published"
);
}
true
}
pub(super) async fn authorize_setup_listener_event(
author_gate: &mut InboundAuthorGate,
buzz_event: relay::BuzzEvent,
respond_to: &crate::config::RespondTo,
allowlist: &HashSet<String>,
owner_cache: &OwnerCache,
channel_info: &crate::pool::ChannelInfoResolver,
rest_client: &relay::RestClient,
) -> Option<AuthorizedListenerEvent> {
author_gate
.authorize_listener_event(
buzz_event,
respond_to,
allowlist,
owner_cache,
channel_info,
rest_client,
)
.await
}
/// Outcome of the synchronous per-event setup checks.
///
/// Callers compute the async gates (`author_allowed`, `filter::match_event`)
/// up-front, then pass the boolean results here. This helper handles
/// everything that is synchronous and stateful: the author gate verdict
/// and event-id dedup.
/// This helper owns only filter matching and event-id deduplication; the
/// production path can call it only through `nudge_authorized_event`, whose
/// input is the gate's private authorized capability.
///
/// Returns `true` when the event should produce a nudge.
#[must_use]
pub(crate) fn should_nudge_for_event(
event_id: EventId,
author_allowed: bool,
filter_matched: bool,
nudged_event_ids: &mut HashSet<EventId>,
) -> bool {
if !author_allowed {
tracing::debug!("setup-mode: event filtered by author gate");
return false;
}
if !filter_matched {
return false;
}
@@ -591,12 +630,13 @@ async fn handle_setup_membership(
/// Build and publish a setup nudge reply to the triggering event.
///
/// Threading: flat reply to the thread root if one exists; otherwise reply
/// to the triggering event itself. P-tags the asker.
/// to the triggering event itself. P-tags the verified effective asker.
async fn publish_setup_nudge(
publisher: &RelayEventPublisher,
keys: &nostr::Keys,
channel_id: Uuid,
triggering_event: &nostr::Event,
recipient_hex: &str,
payload: &SetupPayload,
) -> Result<()> {
use buzz_sdk::ThreadRef;
@@ -621,13 +661,12 @@ async fn publish_setup_nudge(
};
let body = payload.nudge_body();
let author_hex = triggering_event.pubkey.to_hex();
let event_builder = buzz_sdk::build_message(
channel_id,
&body,
thread_ref.as_ref(),
&[&author_hex], // p-tag the asker
&[recipient_hex], // p-tag the verified effective asker
false,
&[],
)
@@ -699,6 +738,89 @@ mod tests {
));
}
#[tokio::test]
async fn authorized_workflow_nudge_mentions_effective_owner_not_relay_signer() {
let agent_keys = nostr::Keys::generate();
let relay_keys = nostr::Keys::generate();
let workflow_owner = nostr::Keys::generate().public_key().to_hex();
let agent = nostr::Keys::generate().public_key().to_hex();
let channel_id = Uuid::new_v4();
let event = relay::BuzzEvent {
connection_generation: 0,
channel_id,
event: crate::author_gate_tests::relay_signed_workflow_dispatch(
&relay_keys,
&workflow_owner,
&agent,
),
};
let relay_hex = relay_keys.public_key().to_hex();
let (rest_client, server) =
crate::author_gate_tests::nip11_server(serde_json::json!({ "self": relay_hex })).await;
let mut gate = InboundAuthorGate::connect(&rest_client, &agent, "setup nudge test").await;
let owner_cache = OwnerCache::new(Some(workflow_owner.clone()));
let channel_info = crate::pool::ChannelInfoResolver::new(
std::collections::HashMap::from([(
channel_id,
relay::ChannelInfo {
name: "workflow".into(),
channel_type: "stream".into(),
description: None,
},
)]),
rest_client.clone(),
);
let authorized = authorize_setup_listener_event(
&mut gate,
event,
&crate::config::RespondTo::OwnerOnly,
&HashSet::new(),
&owner_cache,
&channel_info,
&rest_client,
)
.await
.expect("workflow owner should pass the setup author gate");
let rules = vec![filter::SubscriptionRule {
name: "workflow".into(),
channels: filter::ChannelScope::All("all".into()),
..Default::default()
}];
let (publisher, mut published) = RelayEventPublisher::test_pair();
let payload = SetupPayload {
agent_name: "Fizz".into(),
agent_pubkey: agent.clone(),
requirements: vec![],
};
assert!(
nudge_authorized_event(
authorized,
&rules,
&agent,
&mut HashSet::new(),
&publisher,
&agent_keys,
&payload,
)
.await
);
let nudge = published.recv().await.expect("setup nudge published");
let recipients: Vec<&str> = nudge
.tags
.iter()
.filter_map(|tag| {
let values = tag.as_slice();
(values.first().map(String::as_str) == Some("p"))
.then(|| values.get(1).map(String::as_str))
.flatten()
})
.collect();
assert!(recipients.contains(&workflow_owner.as_str()));
assert!(!recipients.contains(&relay_hex.as_str()));
server.abort();
}
#[test]
fn nudge_body_names_all_requirements() {
let payload = SetupPayload {
@@ -988,32 +1110,25 @@ mod tests {
// ── should_nudge_for_event gate tests ─────────────────────────────────────
//
// These tests exercise the loop-wiring for the two safety-critical guards:
// (a) non-allowlisted author → no nudge, (b) same event-id → exactly one
// nudge. They use the extracted `should_nudge_for_event` helper, which is
// the exact code the live loop calls.
// These tests exercise the loop-adjacent synchronous guards after an event
// has passed the structurally mandatory author capability: (a) unmatched
// filter → no nudge, (b) same event-id → exactly one nudge.
fn fake_event_id(byte: u8) -> EventId {
EventId::from_byte_array([byte; 32])
}
#[test]
fn test_non_allowlisted_author_returns_no_nudge() {
// author_allowed = false → should return false regardless of other args.
fn test_unmatched_filter_returns_no_nudge() {
let mut dedup: HashSet<EventId> = HashSet::new();
let event_id = fake_event_id(0xAA);
let result = should_nudge_for_event(
event_id, false, // author NOT allowed
true, // filter matched — would otherwise nudge
&mut dedup,
);
let result = should_nudge_for_event(event_id, false, &mut dedup);
assert!(!result, "non-allowlisted author must not produce a nudge");
// Dedup set must remain empty — no phantom insertion for blocked author.
assert!(!result, "unmatched event must not produce a nudge");
assert!(
dedup.is_empty(),
"dedup set must not record event for blocked author"
"dedup set must not record an unmatched event"
);
}
@@ -1024,19 +1139,11 @@ mod tests {
let mut dedup: HashSet<EventId> = HashSet::new();
let event_id = fake_event_id(0xBB);
let first = should_nudge_for_event(
event_id, true, // allowed
true, // matched
&mut dedup,
);
let first = should_nudge_for_event(event_id, true, &mut dedup);
assert!(first, "first occurrence must be accepted");
// Simulate reconnect replay: same event arrives again.
let second = should_nudge_for_event(
event_id, true, // allowed
true, // matched
&mut dedup,
);
let second = should_nudge_for_event(event_id, true, &mut dedup);
assert!(
!second,
"replay of the same event-id must be rejected (dedup)"

372
crates/buzz-relay/src/workflow_sink.rs
View File

@@ -148,6 +148,39 @@ fn resolve_mention_pubkeys(text: &str, members: &[(String, String)]) -> Vec<Stri
out
}
/// Append legacy routing tags from rendered output and authority-bearing tags
/// only for targets also named in the workflow owner's stored step template.
fn append_workflow_mention_tags(
tags: &mut Vec<Tag>,
rendered_text: &str,
authored_text: &str,
members: &[(String, String)],
author_pubkey_hex: &str,
) -> Result<(), ActionSinkError> {
let rendered_mentions = resolve_mention_pubkeys(rendered_text, members);
let authored_mentions: std::collections::HashSet<String> =
resolve_mention_pubkeys(authored_text, members)
.into_iter()
.collect();
for mentioned in rendered_mentions {
if mentioned != author_pubkey_hex {
tags.push(
Tag::parse(["p", &mentioned])
.map_err(|e| ActionSinkError::EventBuild(format!("mention p tag: {e}")))?,
);
}
if authored_mentions.contains(&mentioned) {
tags.push(
Tag::parse(["buzz:workflow-mention", &mentioned]).map_err(|e| {
ActionSinkError::EventBuild(format!("workflow mention tag: {e}"))
})?,
);
}
}
Ok(())
}
/// Relay-side action sink — executes workflow side-effects directly.
///
/// Holds a **weak** reference to `AppState` to avoid an `Arc` reference cycle:
@@ -175,11 +208,13 @@ impl ActionSink for RelayActionSink {
community_id: CommunityId,
channel_id: &str,
text: &str,
authored_text: &str,
author_pubkey: &str,
reply_to: Option<&str>,
) -> Pin<Box<dyn Future<Output = Result<String, ActionSinkError>> + Send + '_>> {
let channel_id = channel_id.to_owned();
let text = text.to_owned();
let authored_text = authored_text.to_owned();
let author_pubkey = author_pubkey.to_owned();
let reply_to = reply_to.map(str::to_owned);
@@ -257,8 +292,14 @@ impl ActionSink for RelayActionSink {
// - `p` tag attributes the message to the workflow owner
// - `h` tag scopes to the channel (NIP-29, canonical UUID)
// - `buzz:workflow` tag prevents recursive workflow triggering
// - one `p` tag per `@Name` that resolves to a channel member,
// so mentioned agents are woken (wake is `p`-tag gated)
// - `buzz:workflow-owner` lets harnesses apply the owner's
// inbound-author policy after verifying the relay signature
// - one `p` tag for every resolved mention in the rendered output,
// preserving legacy wake/feed behavior
// - one `buzz:workflow-mention` tag only when the same target was
// named in the workflow owner's stored step template. This is the
// authority-bearing provenance used by ACP; trigger-controlled
// template substitutions cannot create it.
let mut tags = vec![
Tag::parse(["p", &author_pubkey_hex])
.map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?,
@@ -266,6 +307,8 @@ impl ActionSink for RelayActionSink {
.map_err(|e| ActionSinkError::EventBuild(format!("h tag: {e}")))?,
Tag::parse(["buzz:workflow", "true"])
.map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?,
Tag::parse(["buzz:workflow-owner", &author_pubkey_hex])
.map_err(|e| ActionSinkError::EventBuild(format!("workflow owner tag: {e}")))?,
];
// Resolve thread ancestry when this is a threaded reply, so the
@@ -312,10 +355,13 @@ impl ActionSink for RelayActionSink {
}
}
// Resolve `@Name` mentions to channel-member pubkeys and append a
// `p` tag for each (skipping the author, already tagged above). A
// resolution failure must not drop the message, so log and proceed
// with the base tags.
// Resolve `@Name` mentions to channel-member pubkeys. The rendered
// text supplies the legacy `p` tags used by subscriptions and feeds.
// The stored author-written template independently supplies the
// authority-bearing workflow-mention tags. A trigger may therefore
// render an `@Name` into visible output, but it cannot borrow the
// workflow owner's authority to wake that agent. A resolution failure
// must not drop the message, so log and proceed with the base tags.
let members = state
.db
.get_members(tenant.community(), channel_uuid)
@@ -334,15 +380,13 @@ impl ActionSink for RelayActionSink {
Some((name, nostr::PublicKey::from_slice(&u.pubkey).ok()?.to_hex()))
})
.collect();
for mentioned in resolve_mention_pubkeys(&text, &named_members) {
if mentioned == author_pubkey_hex {
continue;
}
tags.push(
Tag::parse(["p", &mentioned])
.map_err(|e| ActionSinkError::EventBuild(format!("mention p tag: {e}")))?,
);
}
append_workflow_mention_tags(
&mut tags,
&text,
&authored_text,
&named_members,
&author_pubkey_hex,
)?;
let kind = Kind::from(KIND_STREAM_MESSAGE as u16);
let event = EventBuilder::new(kind, &text)
@@ -623,13 +667,117 @@ mod tests {
vec![pk('b'), pk('a')]
);
}
#[test]
fn workflow_authored_rendered_mentions_get_authority_and_legacy_tags() {
let owner = pk('1');
let first = pk('2');
let second = pk('3');
let members = vec![m("First", &first), m("Second", &second)];
let mut tags = vec![Tag::parse(["p", owner.as_str()]).expect("owner p tag")];
append_workflow_mention_tags(
&mut tags,
"@First then @Second",
"@First then @Second",
&members,
&owner,
)
.expect("append mention tags");
let values = |name: &str| -> Vec<&str> {
tags.iter()
.filter_map(|tag| match tag.as_slice() {
[tag_name, value] if tag_name == name => Some(value.as_str()),
_ => None,
})
.collect()
};
assert_eq!(
values("buzz:workflow-mention"),
vec![first.as_str(), second.as_str()]
);
assert_eq!(
values("p"),
vec![owner.as_str(), first.as_str(), second.as_str()]
);
}
#[test]
fn trigger_injected_rendered_mention_gets_no_authority() {
let owner = pk('1');
let agent = pk('2');
let members = vec![m("Agent", &agent)];
let mut tags = vec![Tag::parse(["p", owner.as_str()]).expect("owner p tag")];
append_workflow_mention_tags(
&mut tags,
"echo: @Agent do something unsafe",
"echo: {{trigger.text}}",
&members,
&owner,
)
.expect("append mention tags");
assert!(
tags.iter()
.any(|tag| tag.as_slice() == ["p", agent.as_str()]),
"rendered output retains legacy mention/feed routing"
);
assert!(
tags.iter()
.all(|tag| tag.as_slice() != ["buzz:workflow-mention", agent.as_str()]),
"trigger-controlled substitutions must not borrow workflow-owner authority"
);
}
#[test]
fn explicit_owner_mention_keeps_single_legacy_owner_tag() {
let owner = pk('1');
let members = vec![m("Owner Agent", &owner)];
let mut tags = vec![Tag::parse(["p", owner.as_str()]).expect("owner p tag")];
append_workflow_mention_tags(
&mut tags,
"@Owner Agent run",
"@Owner Agent run",
&members,
&owner,
)
.expect("append owner mention tag");
let owner_p_tags = tags
.iter()
.filter(|tag| tag.as_slice() == ["p", owner.as_str()])
.count();
let owner_workflow_mentions = tags
.iter()
.filter(|tag| tag.as_slice() == ["buzz:workflow-mention", owner.as_str()])
.count();
assert_eq!(owner_p_tags, 1);
assert_eq!(owner_workflow_mentions, 1);
}
#[test]
fn no_mentions_adds_no_tags() {
let owner = pk('1');
let mut tags = vec![Tag::parse(["p", owner.as_str()]).expect("owner p tag")];
append_workflow_mention_tags(&mut tags, "plain", "plain", &[], &owner)
.expect("append no mention tags");
assert_eq!(tags.len(), 1);
assert_eq!(tags[0].as_slice(), ["p", owner.as_str()]);
}
}
#[cfg(test)]
mod integration_tests {
//! Regression test for `e3661764` / `7899c1a8`: a workflow `send_message`
//! that mentions a channel member by name (`@Name`) must emit a `p` tag for
//! that member so ACP agent wake (`event_mentions_agent`, p-tag gated) fires.
//! that mentions a channel member by name (`@Name`) in its author-written
//! step template must emit both the legacy `p` tag and authenticated
//! workflow-mention provenance for that member. Rendered trigger data may
//! still create a legacy `p` tag, but never authority-bearing provenance.
//!
//! Postgres-gated like the other DB-backed relay tests. Run with:
//! `cargo test -p buzz-relay --lib workflow_sink -- --ignored`
@@ -676,9 +824,79 @@ mod integration_tests {
Arc::new(state)
}
async fn execute_send_message_workflow(
state: &Arc<AppState>,
community: CommunityId,
channel_id: Uuid,
owner_pubkey: &[u8],
name: &str,
authored_text: &str,
trigger_text: &str,
) -> String {
let definition = serde_json::json!({
"name": name,
"trigger": {"on": "message_posted"},
"steps": [{
"id": "send",
"action": "send_message",
"text": authored_text,
}],
"enabled": true,
});
let definition_hash_byte = name.as_bytes().first().copied().unwrap_or_default();
let workflow_id = state
.db
.create_workflow(
community,
Some(channel_id),
owner_pubkey,
name,
&definition.to_string(),
&[definition_hash_byte; 32],
)
.await
.expect("create workflow");
let trigger_ctx = buzz_workflow::executor::TriggerContext {
text: trigger_text.to_owned(),
channel_id: channel_id.to_string(),
..Default::default()
};
let trigger_ctx_json = serde_json::to_value(&trigger_ctx).expect("serialize trigger");
let run_id = state
.db
.create_workflow_run(community, workflow_id, None, Some(&trigger_ctx_json))
.await
.expect("create workflow run");
// Load the definition back from Postgres before execution. This pins the
// authority source to the durable owner-authored template rather than a
// second test-only string passed directly to RelayActionSink.
let stored_workflow = state
.db
.get_workflow(community, workflow_id)
.await
.expect("load stored workflow");
let stored_definition: buzz_workflow::WorkflowDef =
serde_json::from_value(stored_workflow.definition).expect("parse stored definition");
let result = buzz_workflow::executor::execute_run(
&state.workflow_engine,
community,
run_id,
&stored_definition,
&trigger_ctx,
)
.await
.expect("execute workflow");
result.step_outputs["send"]["event_id"]
.as_str()
.expect("send_message event id")
.to_owned()
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn workflow_send_message_p_tags_mentioned_member() {
async fn workflow_send_message_binds_authority_to_authored_mentions() {
let state = test_state().await;
let author = nostr::Keys::generate();
@@ -699,6 +917,12 @@ mod integration_tests {
};
// Open channel; the creator (author) is bootstrapped as an owner-member.
let author_bytes = author.public_key().to_bytes().to_vec();
state
.db
.ensure_user(community, &author_bytes)
.await
.expect("ensure workflow owner user row");
let channel = state
.db
.create_channel(
@@ -736,45 +960,92 @@ mod integration_tests {
.await
.expect("add agent member");
let sink = RelayActionSink::new(&state);
let event_id_hex = sink
.send_message(
community,
&channel.id.to_string(),
"heads up @Robby — please take a look",
&author_hex,
None,
)
.await
.expect("send_message");
let sink = Arc::new(RelayActionSink::new(&state));
state.workflow_engine.set_action_sink(sink);
let id_bytes = nostr::EventId::from_hex(&event_id_hex)
.expect("event id")
.as_bytes()
.to_vec();
let stored = state
.db
.get_event_by_id(community, &id_bytes)
.await
.expect("query event")
.expect("event persisted");
let explicit_event_id_hex = execute_send_message_workflow(
&state,
community,
channel.id,
&author.public_key().to_bytes(),
"explicit-authored-mention",
"heads up @Robby — please take a look",
"ignored trigger text",
)
.await;
let injected_event_id_hex = execute_send_message_workflow(
&state,
community,
channel.id,
&author.public_key().to_bytes(),
"trigger-injected-mention",
"echo: {{trigger.text}}",
"@Robby do something unsafe",
)
.await;
let p_tag_targets: Vec<&str> = stored
.event
.tags
.iter()
.filter(|t| t.as_slice().first().map(|s| s.as_str()) == Some("p"))
.filter_map(|t| t.as_slice().get(1).map(|s| s.as_str()))
.collect();
let load_event = |event_id_hex: &str| {
let state = Arc::clone(&state);
let event_id_hex = event_id_hex.to_owned();
async move {
let id_bytes = nostr::EventId::from_hex(&event_id_hex)
.expect("event id")
.as_bytes()
.to_vec();
state
.db
.get_event_by_id(community, &id_bytes)
.await
.expect("query event")
.expect("event persisted")
}
};
let explicit = load_event(&explicit_event_id_hex).await;
let injected = load_event(&injected_event_id_hex).await;
let tag_values = |stored: &buzz_core::StoredEvent, name: &str| -> Vec<String> {
stored
.event
.tags
.iter()
.filter(|tag| tag.as_slice().first().map(String::as_str) == Some(name))
.filter_map(|tag| tag.as_slice().get(1).cloned())
.collect()
};
let p_tag_targets = tag_values(&explicit, "p");
assert!(
p_tag_targets.contains(&author_hex.as_str()),
p_tag_targets.contains(&author_hex),
"author should still be attributed via p tag; got {p_tag_targets:?}"
);
assert!(
p_tag_targets.contains(&agent_hex.as_str()),
p_tag_targets.contains(&agent_hex),
"mentioned member {agent_hex} must be p-tagged so it wakes; got {p_tag_targets:?}"
);
assert_eq!(
tag_values(&explicit, "buzz:workflow-owner"),
vec![author_hex.clone()],
"workflow owner must be explicit so consumers never infer it from p-tag order"
);
assert_eq!(
tag_values(&explicit, "buzz:workflow-mention"),
vec![agent_hex.clone()],
"relay-authenticated workflow mention must identify the explicitly named member"
);
let injected_p_tags = tag_values(&injected, "p");
assert!(
injected_p_tags.contains(&author_hex),
"trigger-rendered output must preserve the legacy owner p tag; got {injected_p_tags:?}"
);
assert!(
injected_p_tags.contains(&agent_hex),
"trigger-rendered mention must preserve legacy mention/feed routing; got {injected_p_tags:?}"
);
assert!(
tag_values(&injected, "buzz:workflow-mention").is_empty(),
"a mention introduced solely by trigger data must not receive owner-delegated authority"
);
}
#[tokio::test]
@@ -818,6 +1089,7 @@ mod integration_tests {
community,
&channel.id.to_string(),
"root message",
"root message",
&author_hex,
None,
)
@@ -830,6 +1102,7 @@ mod integration_tests {
community,
&channel.id.to_string(),
"threaded reply",
"threaded reply",
&author_hex,
Some(&root_hex),
)
@@ -972,6 +1245,7 @@ mod integration_tests {
community,
&channel_hex,
"workflow reply",
"workflow reply",
&author_hex,
Some(&parent_hex),
)
@@ -1053,6 +1327,7 @@ mod integration_tests {
community,
&channel_hex,
"workflow reply to root-only parent",
"workflow reply to root-only parent",
&author_hex,
Some(&root_only_parent_hex),
)
@@ -1115,6 +1390,7 @@ mod integration_tests {
community,
&channel.id.to_string(),
"orphan reply",
"orphan reply",
&author_hex,
Some(&unknown),
)

6
crates/buzz-workflow/src/action_sink.rs
View File

@@ -54,7 +54,10 @@ pub trait ActionSink: Send + Sync {
/// carries its owning community so a workflow in community B posts into B
/// even though the side effect has no inbound connection to bind.
/// - `channel_id`: UUID string of the target channel
/// - `text`: message body (must not be empty/whitespace-only)
/// - `text`: rendered message body (must not be empty/whitespace-only)
/// - `authored_text`: the workflow owner's stored, unrendered step template;
/// consumers must use this rather than trigger-controlled rendered output
/// when attaching authority-bearing metadata
/// - `author_pubkey`: hex-encoded pubkey of the workflow owner (used for
/// the `p` attribution tag; the relay keypair signs the event)
/// - `reply_to`: when `Some(event_id_hex)`, the message is posted as a
@@ -67,6 +70,7 @@ pub trait ActionSink: Send + Sync {
community_id: CommunityId,
channel_id: &str,
text: &str,
authored_text: &str,
author_pubkey: &str,
reply_to: Option<&str>,
) -> Pin<Box<dyn Future<Output = Result<String, ActionSinkError>> + Send + '_>>;

16
crates/buzz-workflow/src/executor.rs
View File

@@ -535,7 +535,7 @@ fn resolve_send_message_channel(
/// `RequestApproval` returns `StepResult::Suspended` — the caller must
/// persist state and stop the execution loop.
pub async fn dispatch_action(
step_id: &str,
step: &Step,
action: &ActionDef,
engine: &WorkflowEngine,
community_id: CommunityId,
@@ -544,6 +544,8 @@ pub async fn dispatch_action(
) -> Result<StepResult, WorkflowError> {
use ActionDef::*;
let step_id = &step.id;
// The workflow engine can outlive the serving request that spawned it.
// Revalidate the durable community fence immediately before every external
// side effect (message publish, webhook, delay/resume). A storage failure is
@@ -622,12 +624,22 @@ pub async fn dispatch_action(
"SendMessage → {channel_id}: {text}"
);
let authored_text = match &step.action {
SendMessage { text, .. } => text.as_str(),
_ => {
return Err(WorkflowError::InvalidDefinition(
"SendMessage: resolved action does not match its authored step"
.into(),
));
}
};
let event_id = engine
.action_sink()?
.send_message(
community_id,
&channel_id,
text,
authored_text,
&owner_pubkey_hex,
reply_to,
)
@@ -1220,7 +1232,7 @@ async fn execute_steps(
let dispatch_result = tokio::time::timeout(
std::time::Duration::from_secs(timeout_secs),
dispatch_action(
&step.id,
step,
&resolved_action,
engine,
community_id,