feat(agents): admit gateway-owned workflow prompts

crates/buzz-acp/src/gateway.rs

@@ -37,16 +37,15 @@ const LOOKUP_TIMEOUT: Duration = Duration::from_secs(5);
 #[error("gateway delegation lookup failed: {0}")]
 pub(crate) struct LookupError(String);
 
+struct Authority {
+    gateways: Vec<buzz_sdk::gateway::Live>,
+    relay: Option<PublicKey>,
+}
+
 enum Cached {
     Empty,
-    Fresh {
-        at: Instant,
-        gateways: Vec<buzz_sdk::gateway::Live>,
-    },
-    Failed {
-        at: Instant,
-        error: LookupError,
-    },
+    Fresh { at: Instant, authority: Authority },
+    Failed { at: Instant, error: LookupError },
 }
 
 /// The owner's live gateways for one agent, from the relay, with a bounded
@@ -82,6 +81,10 @@ impl GatewayAccess {
     /// The live gateways now, newest delegation first, at most
     /// [`buzz_sdk::gateway::MAX_LIVE_GATEWAYS`].
     pub(crate) async fn live(&self) -> Result<Vec<PublicKey>, LookupError> {
+        Ok(self.authority().await?.0)
+    }
+
+    async fn authority(&self) -> Result<(Vec<PublicKey>, Option<PublicKey>), LookupError> {
         let mut cached = self.cached.lock().await;
         let reuse = match &*cached {
             Cached::Fresh { at, .. } => at.elapsed() < VERDICT_TTL,
@@ -93,9 +96,9 @@ impl GatewayAccess {
             // cannot extend the authority cache beyond the documented TTL.
             let lookup_started = Instant::now();
             *cached = match tokio::time::timeout(LOOKUP_TIMEOUT, self.lookup()).await {
-                Ok(Ok(gateways)) => Cached::Fresh {
+                Ok(Ok(authority)) => Cached::Fresh {
                     at: lookup_started,
-                    gateways,
+                    authority,
                 },
                 Ok(Err(error)) => Cached::Failed {
                     at: Instant::now(),
@@ -110,11 +113,15 @@ impl GatewayAccess {
         let now = unix_now();
         match &*cached {
             // A cached delegation never outlives its own expiration.
-            Cached::Fresh { gateways, .. } => Ok(gateways
-                .iter()
-                .filter(|live| live.is_live_at(now))
-                .map(|live| live.gateway())
-                .collect()),
+            Cached::Fresh { authority, .. } => Ok((
+                authority
+                    .gateways
+                    .iter()
+                    .filter(|live| live.is_live_at(now))
+                    .map(|live| live.gateway())
+                    .collect(),
+                authority.relay,
+            )),
             Cached::Failed { error, .. } => Err(error.clone()),
             Cached::Empty => Err(LookupError("no lookup".into())),
         }
@@ -128,7 +135,33 @@ impl GatewayAccess {
         Ok(self.live().await?.contains(key))
     }
 
-    async fn lookup(&self) -> Result<Vec<buzz_sdk::gateway::Live>, LookupError> {
+    /// Admit a direct gateway prompt, or a workflow prompt whose gateway owner
+    /// is proven by the current relay's signature and explicit agent mention.
+    /// Workflow routing tags alone never confer gateway authority.
+    pub(crate) async fn prompt_author(
+        &self,
+        event: &Event,
+    ) -> Result<Option<PublicKey>, LookupError> {
+        let (gateways, relay) = self.authority().await?;
+        if gateways.contains(&event.pubkey) {
+            return Ok(Some(event.pubkey));
+        }
+        let Some(relay) = relay else {
+            return Ok(None);
+        };
+        let Some(owner) = buzz_sdk::workflow::owner(event, relay, self.agent) else {
+            return Ok(None);
+        };
+        if !gateways.contains(&owner)
+            || !buzz_sdk::sharing::task::scope(event, self.agent, unix_now())
+                .is_ok_and(|scope| scope.channel == self.channel)
+        {
+            return Ok(None);
+        }
+        Ok(Some(owner))
+    }
+
+    async fn lookup(&self) -> Result<Authority, LookupError> {
         let now = unix_now();
         let heads = self
             .query(json!({
@@ -142,7 +175,10 @@ impl GatewayAccess {
             .map(|live| live.gateway().to_hex())
             .collect();
         if candidates.is_empty() {
-            return Ok(Vec::new());
+            return Ok(Authority {
+                gateways: Vec::new(),
+                relay: None,
+            });
         }
         let profiles = self
             .query(json!({
@@ -166,22 +202,31 @@ impl GatewayAccess {
             return Err(LookupError("ambiguous channel membership".into()));
         }
         let Some(roster) = rosters.first() else {
-            return Ok(Vec::new());
+            return Ok(Authority {
+                gateways: Vec::new(),
+                relay: Some(relay),
+            });
         };
         let member = |key| buzz_sdk::gateway::channel_member(roster, relay, self.channel, key);
         if !member(self.owner) || !member(self.agent) {
-            return Ok(Vec::new());
+            return Ok(Authority {
+                gateways: Vec::new(),
+                relay: Some(relay),
+            });
         }
-        Ok(buzz_sdk::gateway::resolve(
-            self.owner,
-            &heads,
-            &profiles,
-            &[self.owner, self.agent],
-            now,
-        )
-        .into_iter()
-        .filter(|live| member(live.gateway()))
-        .collect())
+        Ok(Authority {
+            gateways: buzz_sdk::gateway::resolve(
+                self.owner,
+                &heads,
+                &profiles,
+                &[self.owner, self.agent],
+                now,
+            )
+            .into_iter()
+            .filter(|live| member(live.gateway()))
+            .collect(),
+            relay: Some(relay),
+        })
     }
 
     /// One `/query`, refusing any response that is not an array of events.

crates/buzz-acp/src/gateway/tests.rs

@@ -9,6 +9,8 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
 use super::*;
 
+mod workflow;
+
 /// Fake relay state shared with the serving task.
 pub(crate) struct FakeRelay {
     pub(crate) heads: Arc<Mutex<Vec<Event>>>,
@@ -259,7 +261,10 @@ async fn gateway_access_is_cached_bounded_and_fails_closed() {
         .unwrap();
     *access.cached.lock().await = Cached::Fresh {
         at: tokio::time::Instant::now(),
-        gateways: buzz_sdk::gateway::live_heads(owner.public_key(), &[expired], now - 1),
+        authority: Authority {
+            gateways: buzz_sdk::gateway::live_heads(owner.public_key(), &[expired], now - 1),
+            relay: None,
+        },
     };
     assert!(
         !access.is_live(&gw).await.unwrap(),

crates/buzz-acp/src/gateway/tests/workflow.rs

@@ -0,0 +1,152 @@
+use super::*;
+
+fn workflow(signer: &Keys, gateway: PublicKey, agent: PublicKey, channel: uuid::Uuid) -> Event {
+    EventBuilder::new(Kind::Custom(9), "@agent run the gateway workflow")
+        .tags([
+            nostr::Tag::parse(["h", &channel.to_string()]).unwrap(),
+            nostr::Tag::public_key(agent),
+            nostr::Tag::parse(["buzz:workflow", "true"]).unwrap(),
+            nostr::Tag::parse(["buzz:workflow-owner", &gateway.to_hex()]).unwrap(),
+            nostr::Tag::parse(["buzz:workflow-mention", &agent.to_hex()]).unwrap(),
+        ])
+        .sign_with_keys(signer)
+        .unwrap()
+}
+
+/// Exercise both production admission boundaries. Removing provenance, scope,
+/// current delegation or membership checks admits one of the negative cases.
+#[tokio::test]
+async fn gateway_workflows_require_provenance_scope_and_live_authority() {
+    let relay = fake_relay().await;
+    let (owner, agent, gateway) = (Keys::generate(), Keys::generate(), Keys::generate());
+    let relay_key = Keys::parse(&"11".repeat(32)).unwrap();
+    relay
+        .heads
+        .lock()
+        .unwrap()
+        .push(grant(&owner, gateway.public_key(), 10));
+    relay.admit(&[owner.public_key(), agent.public_key(), gateway.public_key()]);
+    let access = Arc::new(GatewayAccess::new(
+        owner.public_key(),
+        agent.public_key(),
+        relay.rest.clone(),
+        channel(),
+    ));
+    let mut gate = crate::InboundAuthorGate::connect(
+        &relay.rest,
+        &agent.public_key().to_hex(),
+        "workflow test",
+    )
+    .await;
+    gate.enable_gateways(Some(access.clone()));
+    let owner_cache = crate::OwnerCache::new(Some(owner.public_key().to_hex()));
+    let allowlist = std::collections::HashSet::new();
+    let mut ctx = crate::pool::tests::make_prompt_context_no_owner();
+    ctx.agent_owner_pubkey = Some(owner.public_key());
+    ctx.gateways = Some(access.clone());
+    let check = |event: Event, expected: bool| {
+        let (gate, cache, allowlist, ctx, rest) =
+            (&gate, &owner_cache, &allowlist, &ctx, &relay.rest);
+        let gateway = gateway.public_key();
+        async move {
+            let decision = gate
+                .evaluate_for_test(
+                    &event,
+                    &crate::RespondTo::Owner,
+                    allowlist,
+                    false,
+                    cache,
+                    rest,
+                )
+                .await;
+            assert_eq!(decision.allowed, expected, "ingress {}", event.id);
+            if expected {
+                assert_eq!(decision.effective_author, gateway.to_hex());
+            }
+            let batch = crate::queue::FlushBatch {
+                channel_id: channel(),
+                scope: crate::scope::SessionScope::Thread {
+                    channel_id: channel(),
+                    root_event_id: event.id.to_hex(),
+                },
+                events: vec![crate::queue::BatchEvent {
+                    event,
+                    prompt_tag: "@mention".into(),
+                    received_at: std::time::Instant::now(),
+                    edit: None,
+                }],
+                cancelled_events: vec![],
+                cancel_reason: None,
+            };
+            assert_eq!(
+                crate::pool::dispatch_refusal(ctx, &batch).await.is_none(),
+                expected,
+                "dispatch"
+            );
+        }
+    };
+    let valid = workflow(
+        &relay_key,
+        gateway.public_key(),
+        agent.public_key(),
+        channel(),
+    );
+    check(valid.clone(), true).await;
+    check(
+        workflow(
+            &Keys::generate(),
+            gateway.public_key(),
+            agent.public_key(),
+            channel(),
+        ),
+        false,
+    )
+    .await;
+    check(
+        workflow(
+            &relay_key,
+            gateway.public_key(),
+            agent.public_key(),
+            uuid::Uuid::new_v4(),
+        ),
+        false,
+    )
+    .await;
+    check(
+        workflow(
+            &relay_key,
+            gateway.public_key(),
+            Keys::generate().public_key(),
+            channel(),
+        ),
+        false,
+    )
+    .await;
+    let mut tampered = valid.clone();
+    tampered.content.push_str(" forged");
+    check(tampered, false).await;
+
+    relay.admit(&[owner.public_key(), agent.public_key()]);
+    skip(VERDICT_TTL).await;
+    check(valid.clone(), false).await;
+    relay.admit(&[owner.public_key(), agent.public_key(), gateway.public_key()]);
+    skip(VERDICT_TTL).await;
+    check(valid.clone(), true).await;
+
+    let now = unix_now();
+    *relay.heads.lock().unwrap() =
+        vec![
+            buzz_sdk::gateway::Delegation::grant(gateway.public_key(), now, now - 10)
+                .unwrap()
+                .sign_with_keys(&owner)
+                .unwrap(),
+        ];
+    skip(VERDICT_TTL).await;
+    check(valid.clone(), false).await;
+    *relay.heads.lock().unwrap() = vec![
+        grant(&owner, gateway.public_key(), 10),
+        revoke(&owner, gateway.public_key(), 5),
+    ];
+    skip(VERDICT_TTL).await;
+    check(valid, false).await;
+}

crates/buzz-acp/src/lib.rs

@@ -658,7 +658,7 @@ mod inbound_author_gate {
             owner_cache: &OwnerCache,
             rest_client: &relay::RestClient,
         ) -> InboundAuthorGateDecision {
-            let effective_author = if matches!(respond_to, RespondTo::Owner) {
+            let mut effective_author = if matches!(respond_to, RespondTo::Owner) {
                 event.pubkey.to_hex()
             } else {
                 effective_prompt_author(event, self.relay_self.as_deref(), &self.agent_pubkey_hex)
@@ -673,12 +673,16 @@ mod inbound_author_gate {
             )
             .await;
             // Strict owner mode also admits a live owner-delegated gateway, in
-            // channels only. The raw signer is the effective author here, and
-            // the owner's DMs are never delegated.
+            // channels only. A relay workflow must cryptographically prove
+            // its gateway owner and signed agent/channel scope.
             if !allowed && !is_dm && matches!(respond_to, RespondTo::Owner) {
                 if let Some(gateways) = &self.gateways {
-                    allowed = match gateways.is_live(&event.pubkey).await {
-                        Ok(live) => live,
+                    allowed = match gateways.prompt_author(event).await {
+                        Ok(Some(author)) => {
+                            effective_author = author.to_hex();
+                            true
+                        }
+                        Ok(None) => false,
                         Err(error) => {
                             tracing::warn!(
                                 author = %event.pubkey,

crates/buzz-acp/src/pool.rs

@@ -4943,9 +4943,9 @@ pub(crate) async fn dispatch_refusal(ctx: &PromptContext, pending: &FlushBatch)
         }
         let mut gateway_error = None;
         if let Some(gateways) = &ctx.gateways {
-            match gateways.is_live(&item.event.pubkey).await {
-                Ok(true) => continue,
-                Ok(false) => {}
+            match gateways.prompt_author(&item.event).await {
+                Ok(Some(_)) => continue,
+                Ok(None) => {}
                 Err(error) => gateway_error = Some(error.to_string()),
             }
         }