feat(relay): atomically consume sandbox attachment approvals

crates/buzz-cli/src/commands/sandbox.rs

@@ -8,7 +8,7 @@ use clap::Subcommand;
 use nostr::{Event, EventId, PublicKey, Timestamp};
 use serde::Deserialize;
 use serde_json::{json, Value};
-use tokio::time::{sleep, timeout, Instant};
+use tokio::time::{sleep, timeout};
 use uuid::Uuid;
 
 use crate::{client::BuzzClient, error::CliError};

crates/buzz-cli/src/commands/sandbox/terminal.rs

@@ -7,6 +7,7 @@ use std::os::unix::fs::OpenOptionsExt;
 use buzz_sdk::sandbox_attach::{Flow, Frame, Packet, Rate};
 use buzz_ws_client::{NostrWsConnection, RelayMessage, WsClientError};
 use tokio::io::unix::AsyncFd;
+use tokio::time::Instant;
 
 use super::*;
 

crates/buzz-core/src/lib.rs

@@ -86,3 +86,5 @@ pub mod test_helpers {
 pub mod artifact;
 /// NIP-AR sync filter parsing and limits.
 pub mod artifact_sync;
+/// Canonical single-use sandbox attachment acceptance coordinates.
+pub mod sandbox_attach;

crates/buzz-core/src/sandbox_attach.rs

@@ -0,0 +1,64 @@
+//! Coordinates whose first accepted event permanently consumes an attachment.
+
+use nostr::Event;
+use uuid::Uuid;
+
+/// Parse a canonical terminal or snapshot acceptance coordinate. Requests,
+/// approvals and end records retain ordinary replacement/deletion semantics.
+pub fn parse_acceptance_coordinate(value: &str) -> Option<(&str, Uuid)> {
+    let mut parts = value.split(':');
+    let mode = parts.next()?;
+    if !matches!(mode, "terminal" | "snapshot") {
+        return None;
+    }
+    let text = parts.next()?;
+    let session = Uuid::parse_str(text).ok()?;
+    if session.to_string() != text || parts.next()? != "accepted" || parts.next().is_some() {
+        return None;
+    }
+    Some((mode, session))
+}
+
+/// Identify an unambiguous kind30624 acceptance coordinate. This classifies a
+/// stored envelope; the relay separately verifies its signed audit metadata.
+pub fn acceptance_coordinate(event: &Event) -> Option<(&str, Uuid)> {
+    if u32::from(event.kind.as_u16()) != crate::kind::KIND_SANDBOX_ATTACH {
+        return None;
+    }
+    let mut tags = event
+        .tags
+        .iter()
+        .map(|tag| tag.as_slice())
+        .filter(|tag| tag.first().is_some_and(|name| name == "d"));
+    let [_, value] = tags.next()? else {
+        return None;
+    };
+    if tags.next().is_some() {
+        return None;
+    }
+    parse_acceptance_coordinate(value)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn only_canonical_acceptance_coordinates_are_burns() {
+        let session: Uuid = "abcdefab-1234-4321-8765-abcdefabcdef".parse().unwrap();
+        for mode in ["terminal", "snapshot"] {
+            assert_eq!(
+                parse_acceptance_coordinate(&format!("{mode}:{session}:accepted")),
+                Some((mode, session))
+            );
+            for value in [
+                format!("{mode}:{session}:grant"),
+                format!("{mode}:{session}:request"),
+                format!("{mode}:{session}:ended"),
+                format!("{mode}:{session}:accepted:extra"),
+                format!("{mode}:{}:accepted", session.to_string().to_uppercase()),
+            ] {
+                assert!(parse_acceptance_coordinate(&value).is_none(), "{value}");
+            }
+        }
+    }
+}

crates/buzz-db/src/store/replaceable.rs

@@ -31,6 +31,9 @@ pub enum ParameterizedReplaceStatus {
 pub enum ParameterizedReplacePrecondition<'a> {
     /// Apply normal NIP-33 ordering without a revision precondition.
     Unconditional,
+    /// Insert only when this coordinate has never been accepted, including
+    /// after deletion of its event body. A watermark durably retains the burn.
+    ExpectNoHead,
     /// Require the live head to match this validated event ID.
     ExpectedRevision(&'a [u8]),
     /// Accept only an exact live-head replay and perform no mutation otherwise.
@@ -111,6 +114,22 @@ async fn replace_parameterized_event_in_transaction_impl(
 ) -> Result<ParameterizedReplaceResult> {
     let community_id = tx.community();
     let kind_i32 = buzz_core::kind::event_kind_i32(event);
+    let acceptance = buzz_core::sandbox_attach::acceptance_coordinate(event);
+    if acceptance.is_some_and(|(mode, session)| d_tag != format!("{mode}:{session}:accepted")) {
+        return Err(DbError::InvalidData(
+            "attachment acceptance coordinate mismatch".into(),
+        ));
+    }
+    let acceptance = acceptance.is_some();
+    // This kind's acceptance cannot opt out through another caller or an
+    // omitted client tag. Requests and approvals use ordinary NIP-33 ordering.
+    let precondition = if acceptance {
+        ParameterizedReplacePrecondition::ExpectNoHead
+    } else {
+        precondition
+    };
+    let retain_watermark =
+        acceptance || precondition == ParameterizedReplacePrecondition::ExpectNoHead;
     let pubkey_bytes = event.pubkey.to_bytes();
     let created_at_secs = event.created_at.as_secs() as i64;
     let created_at = DateTime::from_timestamp(created_at_secs, 0)
@@ -177,7 +196,7 @@ async fn replace_parameterized_event_in_transaction_impl(
     .bind(d_tag)
     .fetch_optional(tx.conn())
     .await?;
-    let watermark: Option<(DateTime<Utc>, Vec<u8>)> = if is_nip_rs {
+    let watermark: Option<(DateTime<Utc>, Vec<u8>)> = if is_nip_rs || retain_watermark {
         sqlx::query_as(
             "SELECT created_at, event_id FROM parameterized_event_watermarks \
              WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4",
@@ -217,6 +236,17 @@ async fn replace_parameterized_event_in_transaction_impl(
         ));
     }
 
+    if precondition == ParameterizedReplacePrecondition::ExpectNoHead
+        && (existing.is_some() || watermark.is_some())
+    {
+        return Ok(ParameterizedReplaceResult::new(
+            event,
+            received_at,
+            channel_id,
+            ParameterizedReplaceStatus::RevisionMismatch,
+        ));
+    }
+
     if let ParameterizedReplacePrecondition::ExpectedRevision(expected_revision) = precondition {
         let status = match existing.as_ref() {
             None => Some(ParameterizedReplaceStatus::RevisionMissing),
@@ -330,7 +360,7 @@ async fn replace_parameterized_event_in_transaction_impl(
             return Ok(false);
         }
 
-        if is_nip_rs {
+        if is_nip_rs || retain_watermark {
             sqlx::query(
                 "INSERT INTO parameterized_event_watermarks \
                      (community_id, kind, pubkey, d_tag, created_at, event_id) \

crates/buzz-relay/src/handlers/attach_claim.rs

@@ -0,0 +1,184 @@
+//! Agent-authored first acceptance of an owner-approved attachment.
+
+use buzz_core::tenant::TenantContext;
+use buzz_sdk::sandbox_attach::Scope;
+use nostr::{Event, PublicKey};
+use serde::Deserialize;
+use uuid::Uuid;
+
+use super::ingest::IngestError;
+use crate::state::AppState;
+
+#[derive(Deserialize)]
+#[serde(tag = "mode", rename_all = "lowercase", deny_unknown_fields)]
+enum Audit {
+    Terminal {
+        version: u8,
+        session: Uuid,
+        status: String,
+        grant: Event,
+        bytes_in: u64,
+        bytes_out: u64,
+    },
+    Snapshot {
+        version: u8,
+        session: Uuid,
+        status: String,
+        grant: Event,
+        bytes: u64,
+        reason: Option<serde_json::Value>,
+    },
+}
+
+fn one<'a>(event: &'a Event, name: &str) -> Result<&'a str, String> {
+    let mut tags = event
+        .tags
+        .iter()
+        .map(|tag| tag.as_slice())
+        .filter(|tag| tag.first().is_some_and(|key| key == name));
+    let Some([_, value]) = tags.next() else {
+        return Err("missing or malformed acceptance tag".into());
+    };
+    if tags.next().is_some() {
+        return Err("ambiguous acceptance tag".into());
+    }
+    Ok(value)
+}
+
+fn metadata(event: &Event) -> Result<Option<PublicKey>, String> {
+    if u32::from(event.kind.as_u16()) != buzz_core::kind::KIND_SANDBOX_ATTACH {
+        return Ok(None);
+    }
+    let candidate = event.tags.iter().any(|tag| {
+        tag.as_slice().first().is_some_and(|key| key == "d")
+            && tag
+                .as_slice()
+                .get(1)
+                .is_some_and(|value| value.ends_with(":accepted"))
+    }) || serde_json::from_str::<serde_json::Value>(&event.content)
+        .ok()
+        .is_some_and(|value| value["status"] == "accepted");
+    if !candidate {
+        return Ok(None);
+    }
+    let invalid = || "invalid attachment acceptance metadata".to_owned();
+    let (mode, session) =
+        buzz_core::sandbox_attach::acceptance_coordinate(event).ok_or_else(invalid)?;
+    if event.content.len() > 8192
+        || event.tags.iter().any(|tag| {
+            tag.as_slice()
+                .first()
+                .is_some_and(|key| key == "expiration")
+        })
+    {
+        return Err("attachment acceptance must be bounded and have no expiration".into());
+    }
+    let audit: Audit = serde_json::from_str(&event.content).map_err(|_| invalid())?;
+    let (version, body_mode, body_session, status, grant, empty) = match audit {
+        Audit::Terminal {
+            version,
+            session,
+            status,
+            grant,
+            bytes_in,
+            bytes_out,
+        } => (
+            version,
+            "terminal",
+            session,
+            status,
+            grant,
+            bytes_in == 0 && bytes_out == 0,
+        ),
+        Audit::Snapshot {
+            version,
+            session,
+            status,
+            grant,
+            bytes,
+            reason,
+        } => (
+            version,
+            "snapshot",
+            session,
+            status,
+            grant,
+            bytes == 0 && reason.is_none(),
+        ),
+    };
+    if version != 1
+        || body_mode != mode
+        || body_session != session
+        || status != "accepted"
+        || !empty
+    {
+        return Err(invalid());
+    }
+    let nonce = one(event, "buzz:attach-claim")?;
+    if Uuid::parse_str(nonce).map_err(|_| invalid())?.to_string() != nonce {
+        return Err(invalid());
+    }
+    let body: serde_json::Value = serde_json::from_str(&grant.content).map_err(|_| invalid())?;
+    let scope: Scope = serde_json::from_value(body["scope"].clone()).map_err(|_| invalid())?;
+    let (approved_session, requester) = if mode == "terminal" {
+        let approved =
+            buzz_sdk::sandbox_attach::Grant::parse(&grant, scope, event.created_at.as_secs())
+                .map_err(|_| invalid())?;
+        (approved.session(), approved.requester())
+    } else {
+        let approved = buzz_sdk::sandbox_attach::snapshot::Grant::parse(
+            &grant,
+            scope,
+            event.created_at.as_secs(),
+        )
+        .map_err(|_| invalid())?;
+        (approved.session(), approved.requester())
+    };
+    if approved_session != session
+        || event.pubkey != scope.agent
+        || grant.pubkey == scope.agent
+        || one(&grant, "h")? != scope.channel.to_string()
+        || one(&grant, "e")? != scope.thread.to_hex()
+        || one(&grant, "p")? != scope.agent.to_hex()
+        || one(event, "h")? != scope.channel.to_string()
+        || one(event, "e")? != scope.thread.to_hex()
+        || one(event, "p")? != requester.to_hex()
+    {
+        return Err(invalid());
+    }
+    Ok(Some(grant.pubkey))
+}
+
+/// Ordinary requests/grants remain mutable and revocable. An acceptance is
+/// signed only by the recorded agent and retains the owner's signed approval.
+pub(super) async fn validate(
+    tenant: &TenantContext,
+    event: &Event,
+    state: &AppState,
+) -> Result<(), IngestError> {
+    let Some(owner) =
+        metadata(event).map_err(|error| IngestError::Rejected(format!("invalid: {error}")))?
+    else {
+        return Ok(());
+    };
+    let owned = state
+        .db
+        .is_agent_owner(
+            tenant.community(),
+            event.pubkey.as_bytes(),
+            owner.as_bytes(),
+        )
+        .await
+        .map_err(|error| {
+            IngestError::Internal(format!("attachment owner lookup failed: {}", error.kind()))
+        })?;
+    if !owned {
+        return Err(IngestError::Rejected(
+            "restricted: only the approved agent may claim attachment acceptance".into(),
+        ));
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod postgres_tests;