jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-29 21:41:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

broker: add presence, typing, observer, and liveness signals to the contract

Browse Source 
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 <noreply@anthropic.com>
Signed-off-by: Joel Robotham <jrobotham@squareup.com>
This commit is contained in:
Joel Robotham
2026-08-27 17:21:42 +10:00
parent a2f0e11a9f
commit b25a20206d
6 changed files with 391 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

198
crates/buzz-sdk/src/broker/actions/args.rs
View File

@@ -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<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`], and propagates each frame's own
/// validation error.
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()
)));
}
Ok(Self {
frames: self
.frames
.iter()
.map(ObserverFrame::validated)
.collect::<Result<_, _>>()?,
})
}
}
/// 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
@@ -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()?),

48
crates/buzz-sdk/src/broker/actions/mod.rs
View File

@@ -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.

38
crates/buzz-sdk/src/broker/actions/outcomes.rs
View File

@@ -182,6 +182,18 @@ pub struct StorageRecord {
pub value: Option<String>,
}
/// 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)?;

4
crates/buzz-sdk/src/broker/correlate.rs
View File

@@ -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(()),

33
crates/buzz-sdk/src/broker/mod.rs
View File

@@ -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,

90
crates/buzz-sdk/src/broker/tests.rs
View File

@@ -86,6 +86,22 @@ fn action_fixtures() -> Vec<ActionArgs> {
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> {
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::<PresenceSetArgs>(
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<ObserverFrame>| 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,