diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 531b1ceee7a..bab06526ed7 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -189,6 +189,7 @@ The `kind` integer is the only dispatch switch. The relay routes, stores, and fa | 45003 | KIND_FORUM_COMMENT | Forum thread reply | | 46001–46012 | KIND_WORKFLOW_* | Workflow execution events | | 20001 | KIND_PRESENCE_UPDATE | Ephemeral presence heartbeat | +| 20003 | KIND_AGENT_PRESENCE | Ephemeral presence heartbeat an owner sends for its agents | `buzz-core` defines each event kind as a `pub const u32` and exports the full registry as `ALL_KINDS: &[u32]` (127 kinds at the time of writing); `crates/buzz-core/src/kind.rs` is the source of truth for the current list. Kinds are `u32` (NIP-01 specifies unsigned integer; `u32` covers the full range). Buzz uses both standard Nostr kinds (e.g., kind 7 for reactions) and custom ranges (40000+). @@ -305,6 +306,8 @@ Ephemeral events bypass DB storage, audit, and search. Two sub-paths: ``` Presence events skip membership checks and use local-only fan-out. Multi-node presence fan-out would require Redis pub/sub (documented as future work). +**Agent presence (kind 20003):** an owner's one heartbeat for all its agents, so an app that runs agents without a socket per agent can still show them online. Content is the status (`online`, `away` or `offline`) and each `p` tag names an agent, at most 256. WebSocket only. The relay keeps the agents the signer owns (`users.agent_owner_pubkey`, through the owner cache), skips the rest, and sets or clears their presence exactly as their own kind 20001 would, so the presence snapshot reports them unchanged. It is not fanned out, since live presence is matched by author; viewers see it on their next snapshot read. + **Other ephemeral events (e.g., typing indicators):** ``` 1. VERIFY — spawn_blocking(verify_event) diff --git a/NOSTR.md b/NOSTR.md index cce70f2f77f..cb1f20e84fb 100644 --- a/NOSTR.md +++ b/NOSTR.md @@ -62,6 +62,7 @@ PGPASSWORD=buzz_dev psql -h localhost -U buzz -d buzz -c \ | **Group members (kind:39002)** | ✅ | Relay-signed; `d` tag + `p` tags for all members | | **Membership notifications** | ✅ | kind:44100 (added) / kind:44101 (removed); relay-signed, community-global scope (`channel_id=None` inside the connected community) | | **Presence (kind:20001)** | ✅ | Ephemeral; arbitrary status string (truncated to 128 chars); writes to Redis (`set_presence`/`clear_presence` on `"offline"`), then fan-out to local subscribers. In multi-community mode presence is scoped to the connected community. | +| **Agent presence (kind:20003)** | ✅ | Ephemeral; an owner's presence for the agents it owns, named by `p` tags (at most 256); `online`/`away` set and `offline` clears each agent's presence as its own kind:20001 would. Not fanned out. | | **Typing indicators (kind:20002)** | ✅ | Ephemeral, not stored; published via Redis pub/sub (multi-node capable unlike presence fan-out) | | **NIP-42 authentication** | ✅ | Proactive challenge; optional pubkey allowlist | | **NIP-11 relay info** | ✅ | `GET /` with `Accept: application/nostr+json` | diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index 32d8959fb69..332e97d9589 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -467,6 +467,9 @@ pub const KIND_PRESENCE_UPDATE: u32 = 20001; pub const KIND_PAIRING: u32 = 24134; /// Ephemeral: typing indicator for a channel. pub const KIND_TYPING_INDICATOR: u32 = 20002; +/// Ephemeral: an owner's presence for its agents. Content is the status, each +/// `p` tag names one agent the signer owns. Never the signer's own presence. +pub const KIND_AGENT_PRESENCE: u32 = 20003; /// Ephemeral: owner-scoped encrypted agent observer telemetry and control frame. pub const KIND_AGENT_OBSERVER_FRAME: u32 = 24200; /// Ephemeral: huddle emoji reaction burst. Channel-scoped to the ephemeral @@ -703,6 +706,7 @@ pub const ALL_KINDS: &[u32] = &[ KIND_THREAD_WINDOW_BOUNDS, KIND_PRESENCE_UPDATE, KIND_TYPING_INDICATOR, + KIND_AGENT_PRESENCE, KIND_HUDDLE_REACTION, KIND_BLOSSOM_AUTH, KIND_PAIRING, diff --git a/crates/buzz-db/src/store/user.rs b/crates/buzz-db/src/store/user.rs index e6f9ad4cd51..ec4b897a617 100644 --- a/crates/buzz-db/src/store/user.rs +++ b/crates/buzz-db/src/store/user.rs @@ -404,6 +404,24 @@ pub async fn list_agents_for_owner( .await?) } +/// Of `candidates`, the agents whose `agent_owner_pubkey` is `owner_pubkey` in +/// one community. +pub async fn owned_agents_among( + pool: &PgPool, + community_id: CommunityId, + owner_pubkey: &[u8], + candidates: &[Vec], +) -> Result>> { + Ok(sqlx::query_scalar::<_, Vec>( + "SELECT pubkey FROM users WHERE community_id = $1 AND agent_owner_pubkey = $2 AND pubkey = ANY($3)", + ) + .bind(community_id.as_uuid()) + .bind(owner_pubkey) + .bind(candidates) + .fetch_all(pool) + .await?) +} + /// Check whether `actor_pubkey` is the `agent_owner_pubkey` of `target_pubkey`. /// Queries `agent_owner_pubkey` directly rather than going through /// `get_agent_channel_policy`, which would fetch unrelated fields. @@ -593,6 +611,17 @@ impl Db { crate::user::list_agents_for_owner(&self.pool, community_id, owner_pubkey).await } + /// Of `candidates`, the agents owned by `owner_pubkey` in one community. + #[datastore_span(name = "owned_agents_among", system = "postgresql")] + pub async fn owned_agents_among( + &self, + community_id: CommunityId, + owner_pubkey: &[u8], + candidates: &[Vec], + ) -> Result>> { + crate::user::owned_agents_among(&self.pool, community_id, owner_pubkey, candidates).await + } + /// Check whether `actor_pubkey` is the agent owner of `target_pubkey`. #[datastore_span(name = "is_agent_owner", system = "postgresql")] pub async fn is_agent_owner( diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index 93417e11c38..487ea9b6302 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/crates/buzz-pubsub/src/lib.rs @@ -398,6 +398,25 @@ impl PubSubManager { presence::clear_presence(&self.pool, ctx, pubkey).await } + /// Set one status with the 180s TTL for each of `pubkeys`, in one round trip. + pub async fn set_presence_many( + &self, + ctx: &TenantContext, + pubkeys: &[PublicKey], + status: &str, + ) -> Result<(), PubSubError> { + presence::set_presence_many(&self.pool, ctx, pubkeys, status).await + } + + /// Remove presence for each of `pubkeys`. + pub async fn clear_presence_many( + &self, + ctx: &TenantContext, + pubkeys: &[PublicKey], + ) -> Result<(), PubSubError> { + presence::clear_presence_many(&self.pool, ctx, pubkeys).await + } + /// Returns the current presence status for `pubkey`, or `None` if not set. pub async fn get_presence( &self, diff --git a/crates/buzz-pubsub/src/presence.rs b/crates/buzz-pubsub/src/presence.rs index a3490222715..4de5fae8a3d 100644 --- a/crates/buzz-pubsub/src/presence.rs +++ b/crates/buzz-pubsub/src/presence.rs @@ -58,6 +58,53 @@ pub async fn clear_presence( Ok(()) } +/// Sets one status for every pubkey in `pubkeys`, each with the +/// [`PRESENCE_TTL_SECS`] TTL, in one atomic round trip. +pub async fn set_presence_many( + pool: &Pool, + ctx: &TenantContext, + pubkeys: &[PublicKey], + status: &str, +) -> Result<(), PubSubError> { + if pubkeys.is_empty() { + return Ok(()); + } + let mut conn = pool.get().await?; + let mut pipe = redis::pipe(); + pipe.atomic(); + for pubkey in pubkeys { + pipe.cmd("SET") + .arg(presence_key(ctx, pubkey)) + .arg(status) + .arg("EX") + .arg(PRESENCE_TTL_SECS) + .ignore(); + } + pipe.query_async::<()>(&mut conn).await?; + Ok(()) +} + +/// Removes the presence entries for every pubkey in `pubkeys`. +pub async fn clear_presence_many( + pool: &Pool, + ctx: &TenantContext, + pubkeys: &[PublicKey], +) -> Result<(), PubSubError> { + if pubkeys.is_empty() { + return Ok(()); + } + let mut conn = pool.get().await?; + let keys: Vec = pubkeys + .iter() + .map(|pubkey| presence_key(ctx, pubkey)) + .collect(); + redis::cmd("DEL") + .arg(&keys) + .query_async::<()>(&mut conn) + .await?; + Ok(()) +} + /// Returns the current presence status for `pubkey`, or `None` if not set or expired. pub async fn get_presence( pool: &Pool, @@ -209,6 +256,50 @@ mod tests { clear_presence(&pool, &ctx, &pk2).await.unwrap(); } + #[tokio::test] + #[ignore = "requires Redis"] + async fn test_presence_many_sets_with_ttl_and_clears() { + let pool = make_test_pool(); + let pk1 = make_pubkey(); + let pk2 = make_pubkey(); + let untouched = make_pubkey(); + let ctx = ctx(0xaaaa, "a.example"); + set_presence(&pool, &ctx, &untouched, "away").await.unwrap(); + + set_presence_many(&pool, &ctx, &[pk1, pk2], "online") + .await + .unwrap(); + let result = get_presence_bulk(&pool, &ctx, &[pk1, pk2, untouched]) + .await + .unwrap(); + assert_eq!( + result.get(&pk1.to_hex()).map(String::as_str), + Some("online") + ); + assert_eq!( + result.get(&pk2.to_hex()).map(String::as_str), + Some("online") + ); + assert_eq!( + result.get(&untouched.to_hex()).map(String::as_str), + Some("away") + ); + let mut conn = pool.get().await.unwrap(); + let ttl: i64 = redis::cmd("TTL") + .arg(presence_key(&ctx, &pk2)) + .query_async(&mut conn) + .await + .unwrap(); + assert!(ttl > 0 && ttl <= PRESENCE_TTL_SECS as i64, "got TTL {ttl}"); + + clear_presence_many(&pool, &ctx, &[pk1, pk2]).await.unwrap(); + let result = get_presence_bulk(&pool, &ctx, &[pk1, pk2, untouched]) + .await + .unwrap(); + assert_eq!(result.len(), 1, "only the untouched key remains"); + clear_presence(&pool, &ctx, &untouched).await.unwrap(); + } + #[tokio::test] #[ignore = "requires Redis"] async fn test_presence_ttl() { diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index b4df5bec17f..a937ff0e4f2 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -8,7 +8,8 @@ use tracing::{debug, error, info, warn}; use buzz_core::event::StoredEvent; use buzz_core::kind::{ event_kind_u32, is_ephemeral, is_unshared_gated_event, AUTHOR_ONLY_KINDS, - KIND_AGENT_OBSERVER_FRAME, KIND_GIFT_WRAP, KIND_PRESENCE_UPDATE, KIND_TYPING_INDICATOR, + KIND_AGENT_OBSERVER_FRAME, KIND_AGENT_PRESENCE, KIND_GIFT_WRAP, KIND_PRESENCE_UPDATE, + KIND_TYPING_INDICATOR, }; use buzz_core::observer::{ content_looks_like_nip44, OBSERVER_AGENT_TAG, OBSERVER_FRAME_CONTROL, OBSERVER_FRAME_TAG, @@ -926,6 +927,10 @@ async fn handle_ephemeral_event( Err(_) => return Err(IngestError::Internal("error: internal error".to_string())), } + if event_kind_u32(&event) == KIND_AGENT_PRESENCE { + return set_agent_presence(&state, &conn.tenant, &event, &auth_pubkey).await; + } + // Special handling for presence events (kind:20001). if event_kind_u32(&event) == KIND_PRESENCE_UPDATE { // Accept both bare strings ("online") and legacy JSON ({"status":"online"}). @@ -1026,6 +1031,89 @@ async fn handle_ephemeral_event( Ok(()) } +/// Most agents one agent presence event (kind:20003) may name. +const MAX_AGENT_PRESENCE: usize = 256; + +/// The status and distinct agents an agent presence event names. +fn agent_presence_request(event: &Event) -> Result<(&str, Vec), IngestError> { + let status = event.content.as_str(); + if !matches!(status, "online" | "away" | "offline") { + return Err(IngestError::Rejected( + "invalid: agent presence must be online, away or offline".into(), + )); + } + let mut agents = Vec::new(); + for tag in event.tags.iter() { + let values = tag.as_slice(); + if values.first().map(String::as_str) != Some("p") { + continue; + } + let agent = values + .get(1) + .and_then(|hex| PublicKey::from_hex(hex).ok()) + .ok_or_else(|| IngestError::Rejected("invalid: agent presence p tag".into()))?; + if !agents.contains(&agent) { + agents.push(agent); + } + } + if agents.is_empty() || agents.len() > MAX_AGENT_PRESENCE { + return Err(IngestError::Rejected(format!( + "invalid: agent presence must name 1 to {MAX_AGENT_PRESENCE} agents" + ))); + } + Ok((status, agents)) +} + +/// Stores an owner's presence for its agents (kind:20003), the way each +/// agent's own kind:20001 would be stored, so the presence snapshot reads it +/// back unchanged. Agents the signer does not own are skipped rather than +/// failing the rest, as an owner's list can be briefly stale. Not fanned out: +/// live presence is matched by author, and the author here is the owner. +async fn set_agent_presence( + state: &AppState, + tenant: &TenantContext, + event: &Event, + owner: &PublicKey, +) -> Result<(), IngestError> { + let (status, agents) = agent_presence_request(event)?; + let community = tenant.community(); + let owner_bytes = owner.to_bytes().to_vec(); + let key = |agent: &PublicKey| (community, agent.to_bytes().to_vec(), owner_bytes.clone()); + // An owner link never changes once made, so a cached `true` is final. A + // missing one may just not be made yet (an agent that has not signed in), + // so only `true` is cached here and the rest are asked again, in one query + // bounded by the event's own agents. + let (mut owned, unknown): (Vec<_>, Vec<_>) = agents + .into_iter() + .partition(|agent| state.observer_owner_cache.get(&key(agent)) == Some(true)); + if !unknown.is_empty() { + let candidates: Vec> = unknown.iter().map(|a| a.to_bytes().to_vec()).collect(); + let linked = state + .db + .owned_agents_among(community, &owner_bytes, &candidates) + .await + .map_err(|e| { + warn!(event_id = %event.id.to_hex(), "agent presence owner lookup failed: {e}"); + IngestError::Internal("error: agent ownership unavailable".into()) + })?; + for agent in unknown { + if linked.iter().any(|pubkey| pubkey[..] == agent.to_bytes()) { + state.observer_owner_cache.insert(key(&agent), true); + owned.push(agent); + } + } + } + let stored = if status == "offline" { + state.pubsub.clear_presence_many(tenant, &owned).await + } else { + state.pubsub.set_presence_many(tenant, &owned, status).await + }; + stored.map_err(|e| { + warn!(event_id = %event.id.to_hex(), "agent presence storage failed: {e}"); + IngestError::Internal("error: presence storage unavailable".into()) + }) +} + /// Publishes a verified channel-scoped ephemeral event to the channel's live /// subscribers, on this node and through Redis to the others. The sender must /// be a channel member. @@ -1460,6 +1548,48 @@ mod tests { ); } + #[test] + fn agent_presence_request_names_distinct_agents_with_a_known_status() { + let owner = Keys::generate(); + let agent = Keys::generate().public_key().to_hex(); + let event = |status: &str, tags: Vec>| { + EventBuilder::new(Kind::Custom(super::KIND_AGENT_PRESENCE as u16), status) + .tags(tags.into_iter().map(|tag| Tag::parse(tag).unwrap())) + .sign_with_keys(&owner) + .unwrap() + }; + let p = |hex: &str| vec!["p".to_string(), hex.to_string()]; + + let ok = event( + "away", + vec![p(&agent), p(&agent), vec!["h".into(), "x".into()]], + ); + let (status, agents) = super::agent_presence_request(&ok).unwrap(); + assert_eq!(status, "away"); + assert_eq!(agents.len(), 1, "duplicates collapse, other tags ignored"); + + for bad in [ + event("busy", vec![p(&agent)]), + event("online", vec![]), + event("online", vec![p("not-a-pubkey")]), + event( + "online", + (0..=super::MAX_AGENT_PRESENCE) + .map(|_| p(&Keys::generate().public_key().to_hex())) + .collect(), + ), + ] { + assert!( + matches!( + super::agent_presence_request(&bad), + Err(super::IngestError::Rejected(_)) + ), + "{:?} must be rejected", + bad.content + ); + } + } + #[test] fn agent_observer_route_accepts_agent_to_owner_telemetry() { let agent = Keys::generate(); @@ -1814,6 +1944,128 @@ mod tests { exercise(true, &["online", "offline"]).await; } + /// An owner's batch (kind:20003) stores presence for the agents it + /// owns, skips any it does not, and never touches its own presence. + #[tokio::test] + #[ignore = "requires PostgreSQL and Redis"] + async fn agent_presence_stores_only_owned_agents() { + let redis_url = std::env::var("REDIS_URL").expect("test Redis URL required"); + let state = fanout_access::test_state_with_redis_url(&redis_url).await; + let pool = sqlx::PgPool::connect(&state.config.database_url) + .await + .unwrap(); + let community_uuid = Uuid::new_v4(); + let host = format!("agent-presence-{}.example", community_uuid.simple()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_uuid) + .bind(&host) + .execute(&pool) + .await + .unwrap(); + let tenant = TenantContext::resolved(CommunityId::from_uuid(community_uuid), host); + let owner = Keys::generate(); + let other = Keys::generate(); + let (mine, theirs) = (Keys::generate(), Keys::generate()); + for (user, user_owner) in [ + (&owner, None), + (&other, None), + (&mine, Some(&owner)), + (&theirs, Some(&other)), + ] { + sqlx::query( + "INSERT INTO users (community_id, pubkey, agent_owner_pubkey) VALUES ($1, $2, $3)", + ) + .bind(community_uuid) + .bind(user.public_key().to_bytes().to_vec()) + .bind(user_owner.map(|key| key.public_key().to_bytes().to_vec())) + .execute(&pool) + .await + .unwrap(); + } + let (send_tx, _send_rx) = mpsc::channel(10); + let conn = Arc::new(crate::connection::ConnectionState { + conn_id: Uuid::new_v4(), + tenant: tenant.clone(), + remote_addr: "127.0.0.1:1234".parse().unwrap(), + auth_state: std::sync::Mutex::new(crate::connection::AuthState::Authenticated( + buzz_auth::AuthContext { + pubkey: owner.public_key(), + scopes: vec![], + channel_ids: None, + auth_method: buzz_auth::AuthMethod::Nip42, + agent_owner_pubkey: None, + }, + )), + subscriptions: Arc::new(Mutex::new(HashMap::new())), + send_tx, + ctrl_tx: mpsc::channel(1).0, + terminal_ctrl_tx: mpsc::channel(1).0, + cancel: CancellationToken::new(), + backpressure_count: Arc::new(AtomicU8::new(0)), + grace_limit: 3, + nip_fi_assertion: None, + session_deadline: None, + nip_fi_gate: crate::nip_fi_gate::SessionAdmissionGate::off_mode( + CancellationToken::new(), + ), + community_control: crate::state::CommunityConnectionControl::new( + CancellationToken::new(), + ), + }); + let send = |status: &str| { + let event = EventBuilder::new( + Kind::Custom(super::super::KIND_AGENT_PRESENCE as u16), + status, + ) + .tags([ + Tag::parse(["p", &mine.public_key().to_hex()]).unwrap(), + Tag::parse(["p", &theirs.public_key().to_hex()]).unwrap(), + ]) + .sign_with_keys(&owner) + .unwrap(); + super::super::handle_ephemeral_event( + event, + conn.conn_id, + owner.public_key().to_bytes().to_vec(), + owner.public_key(), + Arc::clone(&conn), + Arc::clone(&state), + ) + }; + let everyone = [mine.public_key(), theirs.public_key(), owner.public_key()]; + // A `false` cached before the link existed must not hide it now. + for agent in [&mine, &theirs] { + state.observer_owner_cache.insert( + ( + tenant.community(), + agent.public_key().to_bytes().to_vec(), + owner.public_key().to_bytes().to_vec(), + ), + false, + ); + } + + send("online").await.expect("owner batch accepted"); + let stored = state + .pubsub + .get_presence_bulk(&tenant, &everyone) + .await + .unwrap(); + assert_eq!( + stored.get(&mine.public_key().to_hex()).map(String::as_str), + Some("online") + ); + assert_eq!(stored.len(), 1, "only the owned agent: {stored:?}"); + + send("offline").await.expect("owner clear accepted"); + let stored = state + .pubsub + .get_presence_bulk(&tenant, &everyone) + .await + .unwrap(); + assert!(stored.is_empty(), "offline clears it: {stored:?}"); + } + /// Production-seam regression for the rejection classification: a /// Redis presence-storage outage must count as a server `error`, /// never as client `invalid`, while a genuinely invalid event diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 49b906ee3d8..564ad3d00dd 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -2413,7 +2413,11 @@ async fn ingest_event_inner( // Typing indicators (kind:20002) never reach this gate: the HTTP bridge // routes them to `publish_http_typing` before ingest. - if auth.is_http() && (kind_u32 == KIND_GIFT_WRAP || kind_u32 == KIND_PRESENCE_UPDATE) { + if auth.is_http() + && (kind_u32 == KIND_GIFT_WRAP + || kind_u32 == KIND_PRESENCE_UPDATE + || kind_u32 == buzz_core::kind::KIND_AGENT_PRESENCE) + { return Err(IngestError::Rejected(format!( "invalid: kind {kind_u32} is only accepted via WebSocket" )));