broker: add storage read/write and live-signal actions to the contract

Justfile

@@ -317,6 +317,9 @@ test-unit:
     ./scripts/test-ensure-local-relay-key.sh
     if command -v cargo-nextest &>/dev/null; then
         cargo nextest run -p buzz-core -p buzz-auth --lib
+        # Broker request/response contracts are shared by every broker client
+        # and host, so their strict-wire and boundary tests must run in CI.
+        cargo nextest run -p buzz-sdk
         # buzz-auth NIP-FI verifier doctests. The sealed-authority
         # `compile_fail` doctests prove the default-feature public API alone
         # cannot forge the issuer→JWKS authority; nextest does not run

crates/buzz-sdk/src/broker/actions/args.rs

@@ -9,9 +9,14 @@ 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,
+    MAX_OBSERVER_BATCH_BYTES, MAX_OBSERVER_FRAMES, MAX_OBSERVER_FRAME_BYTES, MAX_PROMPT_CHARS,
+    MAX_SCALAR_CHARS,
 };
 use crate::SdkError;
+use buzz_core::{
+    engram::{Body, CORE_SLUG, NIP44_PLAINTEXT_MAX},
+    presence::PresenceStatus,
+};
 
 /// Arguments for `channel.read` — the one read action.
 ///
@@ -285,6 +290,271 @@ 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<Self, SdkError> {
+        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`] when the
+    /// complete canonical NIP-AE body exceeds its plaintext limit.
+    pub fn validated(&self) -> Result<Self, SdkError> {
+        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()));
+        }
+        let body = if slug == CORE_SLUG {
+            Body::Core {
+                profile: self.value.clone(),
+            }
+        } else {
+            Body::Memory {
+                slug: slug.clone(),
+                value: Some(self.value.clone()),
+            }
+        };
+        let encoded_len = body.to_json_bytes().len();
+        if encoded_len > NIP44_PLAINTEXT_MAX {
+            return Err(SdkError::ContentTooLarge {
+                max: NIP44_PLAINTEXT_MAX,
+                got: encoded_len,
+            });
+        }
+        Ok(Self {
+            slug,
+            value: self.value.clone(),
+        })
+    }
+}
+
+// ── 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<Self, SdkError> {
+        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<Self, SdkError> {
+        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<Self, SdkError> {
+        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<ObserverFrame>,
+}
+
+impl ObserverEmitArgs {
+    /// Validate and normalize.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`SdkError::InvalidInput`] for an empty batch or one past
+    /// [`super::MAX_OBSERVER_FRAMES`], propagates each frame's own validation
+    /// error, and returns [`SdkError::ContentTooLarge`] when the complete
+    /// serialized argument exceeds [`super::MAX_OBSERVER_BATCH_BYTES`].
+    pub fn validated(&self) -> Result<Self, SdkError> {
+        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()
+            )));
+        }
+        let validated = Self {
+            frames: self
+                .frames
+                .iter()
+                .map(ObserverFrame::validated)
+                .collect::<Result<_, _>>()?,
+        };
+        let encoded_len = serde_json::to_vec(&validated)
+            .map_err(|error| {
+                SdkError::InvalidInput(format!("observer batch is not serializable: {error}"))
+            })?
+            .len();
+        if encoded_len > MAX_OBSERVER_BATCH_BYTES {
+            return Err(SdkError::ContentTooLarge {
+                max: MAX_OBSERVER_BATCH_BYTES,
+                got: encoded_len,
+            });
+        }
+        Ok(validated)
+    }
+}
+
+/// 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<Self, SdkError> {
+        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
@@ -511,6 +781,24 @@ 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),
+    /// 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),
@@ -533,6 +821,12 @@ impl ActionArgs {
             Self::ReactionAdd(_) => Action::ReactionAdd,
             Self::ProfileSet(_) => Action::ProfileSet,
             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,
@@ -555,6 +849,12 @@ 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::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()?),