From a2f0e11a9f336dea51f4687b9c2e67b72df8b83c Mon Sep 17 00:00:00 2001 From: Joel Robotham Date: Thu, 27 Aug 2026 16:43:31 +1000 Subject: [PATCH 1/2] broker: add storage.get and storage.put to the action contract Memory read/write graduates from a deferred operation into v1. A keyless agent could address an encrypted-memory record (storage.address) but not read or write one, so a keyless runtime stays amnesiac across wakes. Both actions are slug-addressed and mirror the existing nine: - storage.get { slug } -> { value? } (value absent = no record, not an error) - storage.put { slug, value } -> EventPublished The host derives the address, encrypts on put, and decrypts on get, so the secret never leaves the key holder -- the same reason storage.address routes through the interface. value is non-empty and bounded by MAX_CONTENT_BYTES. No patch/rm: a client composes those from read-modify-write, as profile.set already does for partial updates. storage.address is left in place; whether get/put make it redundant is an open question for review. Contract crate only. buzz-cli's exhaustive matches gain the two variants when the keyless-client branch rebases onto this. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Joel Robotham --- crates/buzz-sdk/src/broker/actions/args.rs | 81 ++++++++++++++++++- crates/buzz-sdk/src/broker/actions/mod.rs | 13 ++- .../buzz-sdk/src/broker/actions/outcomes.rs | 30 ++++++- crates/buzz-sdk/src/broker/correlate.rs | 2 + crates/buzz-sdk/src/broker/mod.rs | 13 +-- crates/buzz-sdk/src/broker/tests.rs | 47 ++++++++++- 6 files changed, 174 insertions(+), 12 deletions(-) diff --git a/crates/buzz-sdk/src/broker/actions/args.rs b/crates/buzz-sdk/src/broker/actions/args.rs index 1be4546ee17..b8ec727309d 100644 --- a/crates/buzz-sdk/src/broker/actions/args.rs +++ b/crates/buzz-sdk/src/broker/actions/args.rs @@ -8,8 +8,8 @@ use serde::{Deserialize, Serialize}; use super::{ absent_or_valued, absent_or_valued_hex64, channel, channel_id, content, cursor, event_id, hex64_field, is_false, limit, mentions, optional, required, respond_to, validate_slug, Action, - PubkeyHex, DEFAULT_PAGE_LIMIT, MAX_ABOUT_CHARS, MAX_EMOJI_CHARS, MAX_NAME_CHARS, - MAX_PROMPT_CHARS, MAX_SCALAR_CHARS, + PubkeyHex, DEFAULT_PAGE_LIMIT, MAX_ABOUT_CHARS, MAX_CONTENT_BYTES, MAX_EMOJI_CHARS, + MAX_NAME_CHARS, MAX_PROMPT_CHARS, MAX_SCALAR_CHARS, }; use crate::SdkError; @@ -285,6 +285,73 @@ impl StorageAddressArgs { } } +/// Arguments for `storage.get`. +/// +/// Slug-addressed like [`StorageAddressArgs`]: the host derives the record's +/// address from the slug and the secret, fetches it, and decrypts. The keyless +/// caller sends a name and receives plaintext — never a key or a relay filter. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct StorageGetArgs { + /// Memory slug — `core` or `mem/…`, per NIP-AE. + pub slug: String, +} + +impl StorageGetArgs { + /// Validate and normalize. + /// + /// # Errors + /// + /// Returns [`SdkError::InvalidInput`] when the slug fails the NIP-AE + /// grammar. + pub fn validated(&self) -> Result { + let slug = required(&self.slug, "slug", 255)?; + validate_slug(&slug).map_err(|e| SdkError::InvalidInput(e.to_string()))?; + Ok(Self { slug }) + } +} + +/// Arguments for `storage.put`. +/// +/// The caller supplies plaintext; the host encrypts it under the record's key +/// and publishes it. Encryption stays host-side for the same reason addressing +/// does — the secret is the thing this contract exists to avoid handing over. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct StoragePutArgs { + /// Memory slug — `core` or `mem/…`, per NIP-AE. + pub slug: String, + /// Plaintext record body; the host encrypts it before publishing. + pub value: String, +} + +impl StoragePutArgs { + /// Validate and normalize. + /// + /// # Errors + /// + /// Returns [`SdkError::InvalidInput`] when the slug fails the NIP-AE grammar + /// or the value is empty, and [`SdkError::ContentTooLarge`] past + /// [`super::MAX_CONTENT_BYTES`]. + pub fn validated(&self) -> Result { + let slug = required(&self.slug, "slug", 255)?; + validate_slug(&slug).map_err(|e| SdkError::InvalidInput(e.to_string()))?; + if self.value.trim().is_empty() { + return Err(SdkError::InvalidInput("value must not be empty".into())); + } + if self.value.len() > MAX_CONTENT_BYTES { + return Err(SdkError::ContentTooLarge { + max: MAX_CONTENT_BYTES, + got: self.value.len(), + }); + } + Ok(Self { + slug, + value: self.value.clone(), + }) + } +} + // ── Agent arguments ───────────────────────────────────────────────────────── /// Which agent an update or delete targets — exactly one selector, so a host @@ -511,6 +578,12 @@ pub enum ActionArgs { /// Derive an encrypted-memory address. #[serde(rename = "storage.address")] StorageAddress(StorageAddressArgs), + /// Read an encrypted-memory record. + #[serde(rename = "storage.get")] + StorageGet(StorageGetArgs), + /// Write an encrypted-memory record. + #[serde(rename = "storage.put")] + StoragePut(StoragePutArgs), /// Mint a managed agent. #[serde(rename = "agents.create")] AgentsCreate(AgentsCreateArgs), @@ -533,6 +606,8 @@ impl ActionArgs { Self::ReactionAdd(_) => Action::ReactionAdd, Self::ProfileSet(_) => Action::ProfileSet, Self::StorageAddress(_) => Action::StorageAddress, + Self::StorageGet(_) => Action::StorageGet, + Self::StoragePut(_) => Action::StoragePut, Self::AgentsCreate(_) => Action::AgentsCreate, Self::AgentsUpdate(_) => Action::AgentsUpdate, Self::AgentsDelete(_) => Action::AgentsDelete, @@ -555,6 +630,8 @@ impl ActionArgs { Self::ReactionAdd(args) => Self::ReactionAdd(args.validated()?), Self::ProfileSet(args) => Self::ProfileSet(args.validated()?), Self::StorageAddress(args) => Self::StorageAddress(args.validated()?), + Self::StorageGet(args) => Self::StorageGet(args.validated()?), + Self::StoragePut(args) => Self::StoragePut(args.validated()?), Self::AgentsCreate(args) => Self::AgentsCreate(args.validated()?), Self::AgentsUpdate(args) => Self::AgentsUpdate(args.validated()?), Self::AgentsDelete(args) => Self::AgentsDelete(args.validated()?), diff --git a/crates/buzz-sdk/src/broker/actions/mod.rs b/crates/buzz-sdk/src/broker/actions/mod.rs index c6f15cf42eb..c14eec8cd84 100644 --- a/crates/buzz-sdk/src/broker/actions/mod.rs +++ b/crates/buzz-sdk/src/broker/actions/mod.rs @@ -14,10 +14,11 @@ pub mod outcomes; pub use args::{ ActionArgs, AgentTarget, AgentsCreateArgs, AgentsDeleteArgs, AgentsUpdateArgs, ChannelReadArgs, MessagePostArgs, MessageReplyArgs, ProfileSetArgs, ReactionAddArgs, StorageAddressArgs, + StorageGetArgs, StoragePutArgs, }; pub use outcomes::{ ActionOutcome, AgentsCreateOutcome, AgentsDeleteOutcome, AgentsUpdateOutcome, BrokerMessage, - EventPublished, MessagePage, StorageAddress, + EventPublished, MessagePage, StorageAddress, StorageRecord, }; /// Maximum characters in a display name or agent name. @@ -144,6 +145,10 @@ pub enum Action { ProfileSet, /// Derive the address of one encrypted-memory record. StorageAddress, + /// Read one encrypted-memory record by slug. + StorageGet, + /// Write one encrypted-memory record by slug. + StoragePut, /// Mint a managed agent owned by the requester. AgentsCreate, /// Patch a managed agent the requester owns. @@ -154,7 +159,7 @@ pub enum Action { impl Action { /// Every action in this protocol version, in wire-name order. - pub const ALL: [Self; 9] = [ + pub const ALL: [Self; 11] = [ Self::AgentsCreate, Self::AgentsDelete, Self::AgentsUpdate, @@ -164,6 +169,8 @@ impl Action { Self::ProfileSet, Self::ReactionAdd, Self::StorageAddress, + Self::StorageGet, + Self::StoragePut, ]; /// Stable wire name. @@ -176,6 +183,8 @@ impl Action { Self::ReactionAdd => "reaction.add", Self::ProfileSet => "profile.set", Self::StorageAddress => "storage.address", + Self::StorageGet => "storage.get", + Self::StoragePut => "storage.put", Self::AgentsCreate => "agents.create", Self::AgentsUpdate => "agents.update", Self::AgentsDelete => "agents.delete", diff --git a/crates/buzz-sdk/src/broker/actions/outcomes.rs b/crates/buzz-sdk/src/broker/actions/outcomes.rs index 064f3ed3aa4..5e1b676c095 100644 --- a/crates/buzz-sdk/src/broker/actions/outcomes.rs +++ b/crates/buzz-sdk/src/broker/actions/outcomes.rs @@ -165,6 +165,23 @@ pub struct StorageAddress { pub d_tag: String, } +/// Outcome of `storage.get`. +/// +/// Carries the record's plaintext, decrypted host-side. `value` is absent when +/// no record exists at the slug — a normal first-read state, not a failure, so +/// an agent tells "empty memory" from "call failed" without reading an error. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct StorageRecord { + /// Decrypted record body; absent when the slug holds no record. + #[serde( + default, + deserialize_with = "absent_or_valued", + skip_serializing_if = "Option::is_none" + )] + pub value: Option, +} + /// Outcome of a successful `agents.create`. /// /// Carries the new agent's **public** identity only — there is no field for @@ -226,6 +243,12 @@ pub enum ActionOutcome { /// `storage.address` succeeded. #[serde(rename = "storage.address")] StorageAddress(StorageAddress), + /// `storage.get` succeeded. + #[serde(rename = "storage.get")] + StorageGet(StorageRecord), + /// `storage.put` succeeded. + #[serde(rename = "storage.put")] + StoragePut(EventPublished), /// `agents.create` succeeded. #[serde(rename = "agents.create")] AgentsCreate(AgentsCreateOutcome), @@ -248,6 +271,8 @@ impl ActionOutcome { Self::ReactionAdd(_) => Action::ReactionAdd, Self::ProfileSet(_) => Action::ProfileSet, Self::StorageAddress(_) => Action::StorageAddress, + Self::StorageGet(_) => Action::StorageGet, + Self::StoragePut(_) => Action::StoragePut, Self::AgentsCreate(_) => Action::AgentsCreate, Self::AgentsUpdate(_) => Action::AgentsUpdate, Self::AgentsDelete(_) => Action::AgentsDelete, @@ -278,12 +303,15 @@ impl ActionOutcome { Self::MessagePost(published) | Self::MessageReply(published) | Self::ReactionAdd(published) - | Self::ProfileSet(published) => { + | Self::ProfileSet(published) + | Self::StoragePut(published) => { event_id(&published.event_id, "eventId")?; } Self::StorageAddress(address) => { event_id(&address.d_tag, "dTag")?; } + // A record body has no identifier or cursor to check. + Self::StorageGet(_) => {} Self::AgentsCreate(outcome) => { channel(&outcome.channel_id)?; required(&outcome.display_name, "display name", MAX_NAME_CHARS)?; diff --git a/crates/buzz-sdk/src/broker/correlate.rs b/crates/buzz-sdk/src/broker/correlate.rs index b6d4baf6eff..2138dfaf536 100644 --- a/crates/buzz-sdk/src/broker/correlate.rs +++ b/crates/buzz-sdk/src/broker/correlate.rs @@ -85,6 +85,8 @@ pub(super) fn correlate_identities( | (ActionArgs::ReactionAdd(_), _) | (ActionArgs::ProfileSet(_), _) | (ActionArgs::StorageAddress(_), _) + | (ActionArgs::StorageGet(_), _) + | (ActionArgs::StoragePut(_), _) | (ActionArgs::AgentsCreate(_), _) | (ActionArgs::AgentsUpdate(_), _) | (ActionArgs::AgentsDelete(_), _) => Ok(()), diff --git a/crates/buzz-sdk/src/broker/mod.rs b/crates/buzz-sdk/src/broker/mod.rs index e1117cc8104..4951ed5202a 100644 --- a/crates/buzz-sdk/src/broker/mod.rs +++ b/crates/buzz-sdk/src/broker/mod.rs @@ -45,11 +45,11 @@ //! //! # Deferred operations //! -//! Not in v1, all purely additive later: memory read/write (intent-level -//! operations over the encrypted store — until then [`Action::StorageAddress`] -//! only addresses a record, and the key holder remains the only reader/writer), -//! `presence.set`, `typing.set`, and streaming reads (waking on a mention is -//! `channel.read` with `mentionsOnly`, polled). +//! Not in v1, all purely additive later: `presence.set`, `typing.set`, and +//! streaming reads (waking on a mention is `channel.read` with `mentionsOnly`, +//! polled). Memory read/write graduated into v1 as `storage.get`/`storage.put`, +//! both slug-addressed; the host encrypts and decrypts, so the key holder stays +//! the only party that ever sees the record's key. //! //! # Non-goals //! @@ -72,7 +72,8 @@ pub use actions::{ Action, ActionArgs, ActionOutcome, AgentTarget, AgentsCreateArgs, AgentsCreateOutcome, AgentsDeleteArgs, AgentsDeleteOutcome, AgentsUpdateArgs, AgentsUpdateOutcome, BrokerMessage, ChannelReadArgs, EventPublished, MessagePage, MessagePostArgs, MessageReplyArgs, - ProfileSetArgs, PubkeyHex, ReactionAddArgs, StorageAddress, StorageAddressArgs, + ProfileSetArgs, PubkeyHex, ReactionAddArgs, StorageAddress, StorageAddressArgs, StorageGetArgs, + StoragePutArgs, StorageRecord, }; pub use client::{ BrokerClient, BrokerClientExt, BrokerFuture, BrokerTransportError, Dispatch, ValidatedFuture, diff --git a/crates/buzz-sdk/src/broker/tests.rs b/crates/buzz-sdk/src/broker/tests.rs index 14a55005ae0..e8bb5bb5299 100644 --- a/crates/buzz-sdk/src/broker/tests.rs +++ b/crates/buzz-sdk/src/broker/tests.rs @@ -79,6 +79,13 @@ fn action_fixtures() -> Vec { ActionArgs::StorageAddress(StorageAddressArgs { slug: "mem/broker-foundation".into(), }), + ActionArgs::StorageGet(StorageGetArgs { + slug: "core".into(), + }), + ActionArgs::StoragePut(StoragePutArgs { + slug: "mem/broker-foundation".into(), + value: "a remembered fact".into(), + }), ActionArgs::AgentsCreate(AgentsCreateArgs { channel_id: CHANNEL.into(), display_name: "Research helper".into(), @@ -119,12 +126,16 @@ fn outcome_fixtures(keys: &Keys) -> Vec { ActionOutcome::MessagePost(published.clone()), ActionOutcome::MessageReply(published.clone()), ActionOutcome::ReactionAdd(published.clone()), - ActionOutcome::ProfileSet(published), + ActionOutcome::ProfileSet(published.clone()), ActionOutcome::StorageAddress(StorageAddress { author_pubkey: pubkey(), kind: 30174, d_tag: EVENT.into(), }), + ActionOutcome::StorageGet(StorageRecord { + value: Some("a remembered fact".into()), + }), + ActionOutcome::StoragePut(published), ActionOutcome::AgentsCreate(AgentsCreateOutcome { agent_pubkey: pubkey(), display_name: "Research helper".into(), @@ -250,6 +261,20 @@ fn every_outcome_round_trips_through_a_response_envelope() { } } +/// A `storage.get` outcome omits `value` when the slug holds no record, and the +/// absence survives the round trip as `None` — never a stored empty string. +#[test] +fn storage_get_outcome_absent_value_round_trips() { + let empty = ActionOutcome::StorageGet(StorageRecord { value: None }); + let json = serde_json::to_value(&empty).expect("outcome serializes"); + assert!( + json["outcome"].get("value").is_none(), + "an absent record must not put a value key on the wire" + ); + let parsed: ActionOutcome = serde_json::from_value(json).expect("outcome deserializes"); + assert_eq!(parsed, empty); +} + /// Args and outcome share the `action` discriminator, so a payload can never /// pair one action's name with another's shape. #[test] @@ -936,6 +961,8 @@ fn every_payload_has_an_exact_and_secret_free_wire_schema() { ), ("profile.set/args", vec!["about", "displayName", "picture"]), ("storage.address/args", vec!["slug"]), + ("storage.get/args", vec!["slug"]), + ("storage.put/args", vec!["slug", "value"]), ( "agents.create/args", vec![ @@ -974,6 +1001,8 @@ fn every_payload_has_an_exact_and_secret_free_wire_schema() { "storage.address/outcome", vec!["authorPubkey", "dTag", "kind"], ), + ("storage.get/outcome", vec!["value"]), + ("storage.put/outcome", vec!["createdAt", "eventId", "kind"]), ( "agents.create/outcome", vec!["agentPubkey", "channelId", "displayName"], @@ -1354,6 +1383,22 @@ fn validators_accept_and_reject_at_their_boundaries() { assert!(!slug("secrets")); assert!(!slug("mem/Bad Slug")); + // storage.put carries a body: non-empty, bounded, and still slug-gated. + let put = |slug: &str, value: String| { + StoragePutArgs { + slug: slug.into(), + value, + } + .validated() + }; + assert!(put("core", "remember this".into()).is_ok()); + assert!(put("core", " ".into()).is_err()); + assert!(matches!( + put("core", "x".repeat(actions::MAX_CONTENT_BYTES + 1)).unwrap_err(), + SdkError::ContentTooLarge { .. } + )); + assert!(put("Core", "v".into()).is_err()); + // Patch-shaped writes must change something, and reject unknown modes. let profile_error = ProfileSetArgs { display_name: None, From b25a20206d68e66ca0e90d5bf3e21607fef7c2a4 Mon Sep 17 00:00:00 2001 From: Joel Robotham Date: Thu, 27 Aug 2026 17:21:42 +1000 Subject: [PATCH 2/2] broker: add presence, typing, observer, and liveness signals to the contract Give a keyless agent parity with a local one for the ephemeral signals a running agent emits so an owner and channel can see it work. presence.set and typing.set were named as deferred by #6742; observer.emit and liveness.ping are net-new -- the trajectory and keepalive planes the contract never enumerated but a keyless agent needs just as much once it holds no relay connection. Four best-effort actions, all following the existing contract shape (strict wire, one spelling of every identity, no member names its own subject): - presence.set -> status only (reuses buzz_core PresenceStatus) - typing.set -> channelId only; ephemeral, no stop counterpart - observer.emit -> a batch of frames, each { kind, payload }; payload is opaque and encrypted host-side, and the outcome is a batch receipt since re-batched frames have no stable per-frame id - liveness.ping -> { channelId, turnId }; distinct from an observer frame so a host can attach meaning (reset a stall watchdog), not just forward it The host still derives owner, key, encryption, and all Nostr metadata; the agent supplies only content. observer.emit and liveness.ping overlap on the wire -- flagged in the module docs so a reviewer can collapse liveness.ping if preferred. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Joel Robotham --- crates/buzz-sdk/src/broker/actions/args.rs | 198 +++++++++++++++++- crates/buzz-sdk/src/broker/actions/mod.rs | 48 ++++- .../buzz-sdk/src/broker/actions/outcomes.rs | 38 +++- crates/buzz-sdk/src/broker/correlate.rs | 4 + crates/buzz-sdk/src/broker/mod.rs | 33 ++- crates/buzz-sdk/src/broker/tests.rs | 90 +++++++- 6 files changed, 391 insertions(+), 20 deletions(-) diff --git a/crates/buzz-sdk/src/broker/actions/args.rs b/crates/buzz-sdk/src/broker/actions/args.rs index b8ec727309d..9525bba62bb 100644 --- a/crates/buzz-sdk/src/broker/actions/args.rs +++ b/crates/buzz-sdk/src/broker/actions/args.rs @@ -9,9 +9,11 @@ use super::{ absent_or_valued, absent_or_valued_hex64, channel, channel_id, content, cursor, event_id, hex64_field, is_false, limit, mentions, optional, required, respond_to, validate_slug, Action, PubkeyHex, DEFAULT_PAGE_LIMIT, MAX_ABOUT_CHARS, MAX_CONTENT_BYTES, MAX_EMOJI_CHARS, - MAX_NAME_CHARS, MAX_PROMPT_CHARS, MAX_SCALAR_CHARS, + MAX_NAME_CHARS, MAX_OBSERVER_FRAMES, MAX_OBSERVER_FRAME_BYTES, MAX_PROMPT_CHARS, + MAX_SCALAR_CHARS, }; use crate::SdkError; +use buzz_core::presence::PresenceStatus; /// Arguments for `channel.read` — the one read action. /// @@ -352,6 +354,180 @@ impl StoragePutArgs { } } +// ── Live-signal arguments ─────────────────────────────────────────────────── + +/// Arguments for `presence.set`. +/// +/// The status is the whole payload; there is no subject, since presence is +/// always the requester's own. The host publishes it as the ephemeral presence +/// kind on the requester's behalf. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct PresenceSetArgs { + /// Presence to publish. + pub status: PresenceStatus, +} + +impl PresenceSetArgs { + /// Validate and normalize. + /// + /// The status enum is closed, so an unknown value is already rejected at + /// deserialization; there is nothing further to normalize. + /// + /// # Errors + /// + /// Never fails today; the signature matches its siblings so a future + /// invariant has a place to live. + pub fn validated(&self) -> Result { + Ok(Self { + status: self.status, + }) + } +} + +/// Arguments for `typing.set`. +/// +/// A momentary "composing" signal scoped to one channel. There is no stop +/// counterpart: the indicator is ephemeral and lapses on its own, so a client +/// signals by re-sending and stops by falling silent. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct TypingSetArgs { + /// Channel the requester is composing in. + #[serde(deserialize_with = "channel_id")] + pub channel_id: String, +} + +impl TypingSetArgs { + /// Validate and normalize. + /// + /// # Errors + /// + /// Returns [`SdkError::InvalidInput`] for a malformed channel UUID. + pub fn validated(&self) -> Result { + Ok(Self { + channel_id: channel(&self.channel_id)?, + }) + } +} + +/// One observer frame the host publishes on the requester's behalf. +/// +/// The payload is **opaque to this contract**: the host encrypts it to the +/// owner and wraps it in the observer kind without parsing it, so its structure +/// is the runtime's concern, not the wire's. `kind` stays a top-level field so a +/// host can apply per-kind policy (pacing, dropping liveness under load) without +/// decrypting anything. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct ObserverFrame { + /// Frame kind — the runtime's telemetry discriminator (`acp_write`, + /// `turn_started`, …), not a Nostr kind. + pub kind: String, + /// Opaque serialized frame body the host encrypts verbatim. + pub payload: String, +} + +impl ObserverFrame { + /// Validate and normalize one frame. + /// + /// # Errors + /// + /// Returns [`SdkError::InvalidInput`] for an empty `kind` or `payload`, and + /// [`SdkError::ContentTooLarge`] past [`super::MAX_OBSERVER_FRAME_BYTES`]. + pub fn validated(&self) -> Result { + let kind = required(&self.kind, "frame kind", MAX_SCALAR_CHARS)?; + if self.payload.trim().is_empty() { + return Err(SdkError::InvalidInput( + "frame payload must not be empty".into(), + )); + } + if self.payload.len() > MAX_OBSERVER_FRAME_BYTES { + return Err(SdkError::ContentTooLarge { + max: MAX_OBSERVER_FRAME_BYTES, + got: self.payload.len(), + }); + } + Ok(Self { + kind, + payload: self.payload.clone(), + }) + } +} + +/// Arguments for `observer.emit`. +/// +/// A batch of frames, since trajectory is high-volume; the host re-batches and +/// paces publication. Frames carry no owner, key, or Nostr metadata — those are +/// host-derived, the same separation the rest of this contract keeps. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct ObserverEmitArgs { + /// Frames to publish, in order. + pub frames: Vec, +} + +impl ObserverEmitArgs { + /// Validate and normalize. + /// + /// # Errors + /// + /// Returns [`SdkError::InvalidInput`] for an empty batch or one past + /// [`super::MAX_OBSERVER_FRAMES`], and propagates each frame's own + /// validation error. + pub fn validated(&self) -> Result { + if self.frames.is_empty() { + return Err(SdkError::InvalidInput( + "observer.emit requires at least one frame".into(), + )); + } + if self.frames.len() > MAX_OBSERVER_FRAMES { + return Err(SdkError::InvalidInput(format!( + "observer.emit carries {} frames, over the {MAX_OBSERVER_FRAMES} cap", + self.frames.len() + ))); + } + Ok(Self { + frames: self + .frames + .iter() + .map(ObserverFrame::validated) + .collect::>()?, + }) + } +} + +/// Arguments for `liveness.ping`. +/// +/// A turn-scoped keepalive: it tells the host a turn is still running, which a +/// host can act on (resetting a stall watchdog) rather than merely forward. +/// That host-side meaning is what distinguishes it from an `observer.emit` frame +/// carrying the same context. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct LivenessPingArgs { + /// Channel the turn belongs to. + #[serde(deserialize_with = "channel_id")] + pub channel_id: String, + /// Opaque, process-local turn identifier the keepalive refers to. + pub turn_id: String, +} + +impl LivenessPingArgs { + /// Validate and normalize. + /// + /// # Errors + /// + /// Returns [`SdkError::InvalidInput`] for a malformed channel UUID or an + /// empty or over-long turn id. + pub fn validated(&self) -> Result { + Ok(Self { + channel_id: channel(&self.channel_id)?, + turn_id: required(&self.turn_id, "turnId", MAX_SCALAR_CHARS)?, + }) + } +} + // ── Agent arguments ───────────────────────────────────────────────────────── /// Which agent an update or delete targets — exactly one selector, so a host @@ -584,6 +760,18 @@ pub enum ActionArgs { /// Write an encrypted-memory record. #[serde(rename = "storage.put")] StoragePut(StoragePutArgs), + /// Set the requester's presence. + #[serde(rename = "presence.set")] + PresenceSet(PresenceSetArgs), + /// Signal the requester is composing. + #[serde(rename = "typing.set")] + TypingSet(TypingSetArgs), + /// Publish a batch of observer frames. + #[serde(rename = "observer.emit")] + ObserverEmit(ObserverEmitArgs), + /// Signal a turn is still alive. + #[serde(rename = "liveness.ping")] + LivenessPing(LivenessPingArgs), /// Mint a managed agent. #[serde(rename = "agents.create")] AgentsCreate(AgentsCreateArgs), @@ -608,6 +796,10 @@ impl ActionArgs { Self::StorageAddress(_) => Action::StorageAddress, Self::StorageGet(_) => Action::StorageGet, Self::StoragePut(_) => Action::StoragePut, + Self::PresenceSet(_) => Action::PresenceSet, + Self::TypingSet(_) => Action::TypingSet, + Self::ObserverEmit(_) => Action::ObserverEmit, + Self::LivenessPing(_) => Action::LivenessPing, Self::AgentsCreate(_) => Action::AgentsCreate, Self::AgentsUpdate(_) => Action::AgentsUpdate, Self::AgentsDelete(_) => Action::AgentsDelete, @@ -632,6 +824,10 @@ impl ActionArgs { Self::StorageAddress(args) => Self::StorageAddress(args.validated()?), Self::StorageGet(args) => Self::StorageGet(args.validated()?), Self::StoragePut(args) => Self::StoragePut(args.validated()?), + Self::PresenceSet(args) => Self::PresenceSet(args.validated()?), + Self::TypingSet(args) => Self::TypingSet(args.validated()?), + Self::ObserverEmit(args) => Self::ObserverEmit(args.validated()?), + Self::LivenessPing(args) => Self::LivenessPing(args.validated()?), Self::AgentsCreate(args) => Self::AgentsCreate(args.validated()?), Self::AgentsUpdate(args) => Self::AgentsUpdate(args.validated()?), Self::AgentsDelete(args) => Self::AgentsDelete(args.validated()?), diff --git a/crates/buzz-sdk/src/broker/actions/mod.rs b/crates/buzz-sdk/src/broker/actions/mod.rs index c14eec8cd84..4d29ec80081 100644 --- a/crates/buzz-sdk/src/broker/actions/mod.rs +++ b/crates/buzz-sdk/src/broker/actions/mod.rs @@ -13,12 +13,14 @@ pub mod outcomes; pub use args::{ ActionArgs, AgentTarget, AgentsCreateArgs, AgentsDeleteArgs, AgentsUpdateArgs, ChannelReadArgs, - MessagePostArgs, MessageReplyArgs, ProfileSetArgs, ReactionAddArgs, StorageAddressArgs, - StorageGetArgs, StoragePutArgs, + LivenessPingArgs, MessagePostArgs, MessageReplyArgs, ObserverEmitArgs, ObserverFrame, + PresenceSetArgs, ProfileSetArgs, ReactionAddArgs, StorageAddressArgs, StorageGetArgs, + StoragePutArgs, TypingSetArgs, }; +pub use buzz_core::presence::PresenceStatus; pub use outcomes::{ ActionOutcome, AgentsCreateOutcome, AgentsDeleteOutcome, AgentsUpdateOutcome, BrokerMessage, - EventPublished, MessagePage, StorageAddress, StorageRecord, + EventPublished, MessagePage, ObserverReceipt, StorageAddress, StorageRecord, }; /// Maximum characters in a display name or agent name. @@ -57,6 +59,19 @@ pub const DEFAULT_PAGE_LIMIT: u32 = 100; /// Maximum accepted length of a read cursor, in bytes. pub const MAX_CURSOR_LEN: usize = 256; +/// Maximum observer frames a single `observer.emit` may carry. +/// +/// Trajectory is high-volume, so a client batches frames per call; the host +/// re-batches and paces publication. This bounds one call, not the stream. +pub const MAX_OBSERVER_FRAMES: usize = 256; + +/// Maximum bytes of one observer frame's opaque payload. +/// +/// The payload is passed through and encrypted host-side without being parsed, +/// so it is bounded but not interpreted. Matches the runtime's per-frame +/// plaintext budget. +pub const MAX_OBSERVER_FRAME_BYTES: usize = 64 * 1024; + /// Inbound author gate modes a requester may ask for. /// /// `allowlist` is deliberately absent: it needs a pubkey list this request @@ -149,6 +164,14 @@ pub enum Action { StorageGet, /// Write one encrypted-memory record by slug. StoragePut, + /// Publish the requester's presence status. + PresenceSet, + /// Signal the requester is composing in a channel. + TypingSet, + /// Publish a batch of the requester's owner-scoped observer frames. + ObserverEmit, + /// Signal a turn is still alive. + LivenessPing, /// Mint a managed agent owned by the requester. AgentsCreate, /// Patch a managed agent the requester owns. @@ -159,18 +182,22 @@ pub enum Action { impl Action { /// Every action in this protocol version, in wire-name order. - pub const ALL: [Self; 11] = [ + pub const ALL: [Self; 15] = [ Self::AgentsCreate, Self::AgentsDelete, Self::AgentsUpdate, Self::ChannelRead, + Self::LivenessPing, Self::MessagePost, Self::MessageReply, + Self::ObserverEmit, + Self::PresenceSet, Self::ProfileSet, Self::ReactionAdd, Self::StorageAddress, Self::StorageGet, Self::StoragePut, + Self::TypingSet, ]; /// Stable wire name. @@ -185,6 +212,10 @@ impl Action { Self::StorageAddress => "storage.address", Self::StorageGet => "storage.get", Self::StoragePut => "storage.put", + Self::PresenceSet => "presence.set", + Self::TypingSet => "typing.set", + Self::ObserverEmit => "observer.emit", + Self::LivenessPing => "liveness.ping", Self::AgentsCreate => "agents.create", Self::AgentsUpdate => "agents.update", Self::AgentsDelete => "agents.delete", @@ -204,7 +235,14 @@ impl Action { /// [`super::BrokerErrorCode::Unsupported`] for how a caller reacts. #[must_use] pub fn is_best_effort(self) -> bool { - matches!(self, Self::ReactionAdd) + matches!( + self, + Self::ReactionAdd + | Self::PresenceSet + | Self::TypingSet + | Self::ObserverEmit + | Self::LivenessPing + ) } /// Resolve a wire name. diff --git a/crates/buzz-sdk/src/broker/actions/outcomes.rs b/crates/buzz-sdk/src/broker/actions/outcomes.rs index 5e1b676c095..8db27f4296a 100644 --- a/crates/buzz-sdk/src/broker/actions/outcomes.rs +++ b/crates/buzz-sdk/src/broker/actions/outcomes.rs @@ -182,6 +182,18 @@ pub struct StorageRecord { pub value: Option, } +/// Outcome of `observer.emit`. +/// +/// A batch acknowledgement, not per-frame receipts: the host re-batches and +/// paces publication, so individual frames have no stable published id to echo. +/// `accepted` is how many frames the host took for publication. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct ObserverReceipt { + /// Frames accepted for publication from this batch. + pub accepted: u32, +} + /// Outcome of a successful `agents.create`. /// /// Carries the new agent's **public** identity only — there is no field for @@ -249,6 +261,18 @@ pub enum ActionOutcome { /// `storage.put` succeeded. #[serde(rename = "storage.put")] StoragePut(EventPublished), + /// `presence.set` succeeded. + #[serde(rename = "presence.set")] + PresenceSet(EventPublished), + /// `typing.set` succeeded. + #[serde(rename = "typing.set")] + TypingSet(EventPublished), + /// `observer.emit` succeeded. + #[serde(rename = "observer.emit")] + ObserverEmit(ObserverReceipt), + /// `liveness.ping` succeeded. + #[serde(rename = "liveness.ping")] + LivenessPing(EventPublished), /// `agents.create` succeeded. #[serde(rename = "agents.create")] AgentsCreate(AgentsCreateOutcome), @@ -273,6 +297,10 @@ impl ActionOutcome { Self::StorageAddress(_) => Action::StorageAddress, Self::StorageGet(_) => Action::StorageGet, Self::StoragePut(_) => Action::StoragePut, + Self::PresenceSet(_) => Action::PresenceSet, + Self::TypingSet(_) => Action::TypingSet, + Self::ObserverEmit(_) => Action::ObserverEmit, + Self::LivenessPing(_) => Action::LivenessPing, Self::AgentsCreate(_) => Action::AgentsCreate, Self::AgentsUpdate(_) => Action::AgentsUpdate, Self::AgentsDelete(_) => Action::AgentsDelete, @@ -304,14 +332,18 @@ impl ActionOutcome { | Self::MessageReply(published) | Self::ReactionAdd(published) | Self::ProfileSet(published) - | Self::StoragePut(published) => { + | Self::StoragePut(published) + | Self::PresenceSet(published) + | Self::TypingSet(published) + | Self::LivenessPing(published) => { event_id(&published.event_id, "eventId")?; } Self::StorageAddress(address) => { event_id(&address.d_tag, "dTag")?; } - // A record body has no identifier or cursor to check. - Self::StorageGet(_) => {} + // Neither a record body nor a batch receipt carries an identifier or + // cursor to check. + Self::StorageGet(_) | Self::ObserverEmit(_) => {} Self::AgentsCreate(outcome) => { channel(&outcome.channel_id)?; required(&outcome.display_name, "display name", MAX_NAME_CHARS)?; diff --git a/crates/buzz-sdk/src/broker/correlate.rs b/crates/buzz-sdk/src/broker/correlate.rs index 2138dfaf536..c86c161138d 100644 --- a/crates/buzz-sdk/src/broker/correlate.rs +++ b/crates/buzz-sdk/src/broker/correlate.rs @@ -87,6 +87,10 @@ pub(super) fn correlate_identities( | (ActionArgs::StorageAddress(_), _) | (ActionArgs::StorageGet(_), _) | (ActionArgs::StoragePut(_), _) + | (ActionArgs::PresenceSet(_), _) + | (ActionArgs::TypingSet(_), _) + | (ActionArgs::ObserverEmit(_), _) + | (ActionArgs::LivenessPing(_), _) | (ActionArgs::AgentsCreate(_), _) | (ActionArgs::AgentsUpdate(_), _) | (ActionArgs::AgentsDelete(_), _) => Ok(()), diff --git a/crates/buzz-sdk/src/broker/mod.rs b/crates/buzz-sdk/src/broker/mod.rs index 4951ed5202a..44705cb2bc0 100644 --- a/crates/buzz-sdk/src/broker/mod.rs +++ b/crates/buzz-sdk/src/broker/mod.rs @@ -43,13 +43,29 @@ //! nothing stops a host from *holding* keys — that is the point. It stops one //! from handing them over. //! +//! # Live signals +//! +//! The ephemeral signals a running agent emits so an owner and channel can see +//! it work are actions too, so a keyless agent keeps parity with a local one: +//! `presence.set` (status), `typing.set` (composing in a channel), +//! `observer.emit` (a batch of owner-scoped trajectory frames), and +//! `liveness.ping` (a turn keepalive). The host derives owner, key, encryption, +//! and Nostr metadata; the agent supplies only content. All four are +//! [best-effort][Action::is_best_effort] — a host that does not offer one +//! refuses it and the agent carries on. +//! +//! `observer.emit`'s payload is opaque to this contract: the host encrypts it +//! and never parses it. `liveness.ping` overlaps `observer.emit` on the wire (a +//! keepalive could be one more frame) but is its own action so a host can attach +//! meaning to it — resetting a stall watchdog — rather than only forward it. +//! //! # Deferred operations //! -//! Not in v1, all purely additive later: `presence.set`, `typing.set`, and -//! streaming reads (waking on a mention is `channel.read` with `mentionsOnly`, -//! polled). Memory read/write graduated into v1 as `storage.get`/`storage.put`, -//! both slug-addressed; the host encrypts and decrypts, so the key holder stays -//! the only party that ever sees the record's key. +//! Not in v1, purely additive later: streaming reads (waking on a mention is +//! `channel.read` with `mentionsOnly`, polled). Memory read/write graduated into +//! v1 as `storage.get`/`storage.put`, both slug-addressed; the host encrypts and +//! decrypts, so the key holder stays the only party that ever sees the record's +//! key. //! //! # Non-goals //! @@ -71,9 +87,10 @@ use actions::absent_or_valued; pub use actions::{ Action, ActionArgs, ActionOutcome, AgentTarget, AgentsCreateArgs, AgentsCreateOutcome, AgentsDeleteArgs, AgentsDeleteOutcome, AgentsUpdateArgs, AgentsUpdateOutcome, BrokerMessage, - ChannelReadArgs, EventPublished, MessagePage, MessagePostArgs, MessageReplyArgs, - ProfileSetArgs, PubkeyHex, ReactionAddArgs, StorageAddress, StorageAddressArgs, StorageGetArgs, - StoragePutArgs, StorageRecord, + ChannelReadArgs, EventPublished, LivenessPingArgs, MessagePage, MessagePostArgs, + MessageReplyArgs, ObserverEmitArgs, ObserverFrame, ObserverReceipt, PresenceSetArgs, + PresenceStatus, ProfileSetArgs, PubkeyHex, ReactionAddArgs, StorageAddress, StorageAddressArgs, + StorageGetArgs, StoragePutArgs, StorageRecord, TypingSetArgs, }; pub use client::{ BrokerClient, BrokerClientExt, BrokerFuture, BrokerTransportError, Dispatch, ValidatedFuture, diff --git a/crates/buzz-sdk/src/broker/tests.rs b/crates/buzz-sdk/src/broker/tests.rs index e8bb5bb5299..5f8df198598 100644 --- a/crates/buzz-sdk/src/broker/tests.rs +++ b/crates/buzz-sdk/src/broker/tests.rs @@ -86,6 +86,22 @@ fn action_fixtures() -> Vec { slug: "mem/broker-foundation".into(), value: "a remembered fact".into(), }), + ActionArgs::PresenceSet(PresenceSetArgs { + status: PresenceStatus::Online, + }), + ActionArgs::TypingSet(TypingSetArgs { + channel_id: CHANNEL.into(), + }), + ActionArgs::ObserverEmit(ObserverEmitArgs { + frames: vec![ObserverFrame { + kind: "acp_write".into(), + payload: r#"{"jsonrpc":"2.0","method":"session/update"}"#.into(), + }], + }), + ActionArgs::LivenessPing(LivenessPingArgs { + channel_id: CHANNEL.into(), + turn_id: "turn-7f3a".into(), + }), ActionArgs::AgentsCreate(AgentsCreateArgs { channel_id: CHANNEL.into(), display_name: "Research helper".into(), @@ -135,7 +151,11 @@ fn outcome_fixtures(keys: &Keys) -> Vec { ActionOutcome::StorageGet(StorageRecord { value: Some("a remembered fact".into()), }), - ActionOutcome::StoragePut(published), + ActionOutcome::StoragePut(published.clone()), + ActionOutcome::PresenceSet(published.clone()), + ActionOutcome::TypingSet(published.clone()), + ActionOutcome::ObserverEmit(ObserverReceipt { accepted: 1 }), + ActionOutcome::LivenessPing(published), ActionOutcome::AgentsCreate(AgentsCreateOutcome { agent_pubkey: pubkey(), display_name: "Research helper".into(), @@ -717,8 +737,6 @@ fn only_declared_action_names_resolve() { "nip98.auth", "keys.export", "identity.nsec", - "presence.set", - "typing.set", ] { assert!( Action::parse(rejected).is_err(), @@ -963,6 +981,10 @@ fn every_payload_has_an_exact_and_secret_free_wire_schema() { ("storage.address/args", vec!["slug"]), ("storage.get/args", vec!["slug"]), ("storage.put/args", vec!["slug", "value"]), + ("presence.set/args", vec!["status"]), + ("typing.set/args", vec!["channelId"]), + ("observer.emit/args", vec!["frames"]), + ("liveness.ping/args", vec!["channelId", "turnId"]), ( "agents.create/args", vec![ @@ -1003,6 +1025,13 @@ fn every_payload_has_an_exact_and_secret_free_wire_schema() { ), ("storage.get/outcome", vec!["value"]), ("storage.put/outcome", vec!["createdAt", "eventId", "kind"]), + ("presence.set/outcome", vec!["createdAt", "eventId", "kind"]), + ("typing.set/outcome", vec!["createdAt", "eventId", "kind"]), + ("observer.emit/outcome", vec!["accepted"]), + ( + "liveness.ping/outcome", + vec!["createdAt", "eventId", "kind"], + ), ( "agents.create/outcome", vec!["agentPubkey", "channelId", "displayName"], @@ -1399,6 +1428,61 @@ fn validators_accept_and_reject_at_their_boundaries() { )); assert!(put("Core", "v".into()).is_err()); + // Live signals: presence needs a known status, typing and liveness a real + // channel, and observer.emit a non-empty, bounded batch of non-empty frames. + assert!(PresenceSetArgs { + status: PresenceStatus::Away + } + .validated() + .is_ok()); + assert!(serde_json::from_value::( + serde_json::json!({ "status": "invisible" }) + ) + .is_err()); + assert!(TypingSetArgs { + channel_id: CHANNEL.into() + } + .validated() + .is_ok()); + assert!(TypingSetArgs { + channel_id: "not-a-uuid".into() + } + .validated() + .is_err()); + let ping = |channel: &str, turn: &str| { + LivenessPingArgs { + channel_id: channel.into(), + turn_id: turn.into(), + } + .validated() + }; + assert!(ping(CHANNEL, "turn-1").is_ok()); + assert!(ping(CHANNEL, " ").is_err()); + assert!(ping("not-a-uuid", "turn-1").is_err()); + + let frame = |payload: String| ObserverFrame { + kind: "acp_write".into(), + payload, + }; + let emit = |frames: Vec| ObserverEmitArgs { frames }.validated(); + assert!(emit(vec![frame("{}".into())]).is_ok()); + assert!(emit(vec![]).is_err()); + assert!(emit(vec![ObserverFrame { + kind: String::new(), + payload: "{}".into() + }]) + .is_err()); + assert!(emit(vec![frame(" ".into())]).is_err()); + assert!(matches!( + emit(vec![frame( + "x".repeat(actions::MAX_OBSERVER_FRAME_BYTES + 1) + )]) + .unwrap_err(), + SdkError::ContentTooLarge { .. } + )); + assert!(emit(vec![frame("{}".into()); actions::MAX_OBSERVER_FRAMES]).is_ok()); + assert!(emit(vec![frame("{}".into()); actions::MAX_OBSERVER_FRAMES + 1]).is_err()); + // Patch-shaped writes must change something, and reject unknown modes. let profile_error = ProfileSetArgs { display_name: None,