diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index 4a82cf6306d..5db1e237c70 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -488,6 +488,11 @@ pub struct CliArgs { #[arg(long, env = "BUZZ_ACP_RELAY_OBSERVER", default_value_t = false)] pub relay_observer: bool, + /// Publish channel-visible redacted work status (NIP-AW kind 30181, + /// h = d = channel UUID): turn state, model id, and tool-call titles only. + #[arg(long, env = "BUZZ_ACP_PUBLIC_STATUS", default_value_t = false)] + pub public_status: bool, + /// Exit after this many seconds with no dispatched events and no turn in flight. /// 0 disables inactivity self-termination. #[arg(long, env = "BUZZ_ACP_EXIT_AFTER_INACTIVITY", default_value_t = 0)] @@ -582,6 +587,8 @@ pub struct Config { pub has_generated_codex_config: bool, /// Whether to publish encrypted observer frames through the relay. pub relay_observer: bool, + /// Whether to publish channel-visible redacted work status (NIP-AW kind 30181). + pub public_status: bool, /// Seconds without dispatched events before an idle harness exits. 0 = disabled. pub exit_after_inactivity_secs: u64, /// Whether ACP/LLM subprocess initialization is deferred until accepted work arrives. @@ -1137,6 +1144,7 @@ impl Config { persona_env_vars, has_generated_codex_config, relay_observer: args.relay_observer, + public_status: args.public_status, exit_after_inactivity_secs: args.exit_after_inactivity, lazy_pool: args.lazy_pool, idle_pool_sleep_secs: args.idle_pool_sleep, @@ -1510,6 +1518,7 @@ mod tests { persona_env_vars: vec![], has_generated_codex_config: false, relay_observer: false, + public_status: false, exit_after_inactivity_secs: 0, lazy_pool: false, idle_pool_sleep_secs: 0, diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 146214197a8..1756b55ec1d 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -7,6 +7,7 @@ mod filter; mod observer; mod pool; mod pool_lifecycle; +mod public_status; mod queue; mod relay; mod setup_mode; @@ -1958,9 +1959,8 @@ async fn tokio_main() -> Result<()> { tracing::info!("buzz-acp starting: {}", config.summary()); - let observer = config - .relay_observer - .then(observer::ObserverHandle::in_process); + let observer = + (config.relay_observer || config.public_status).then(observer::ObserverHandle::in_process); if let Some(handle) = &observer { handle.emit( "harness_started", @@ -2162,6 +2162,17 @@ async fn tokio_main() -> Result<()> { )); } + let mut public_status_task = None; + if config.public_status { + if let Some(observer) = observer.clone() { + public_status_task = Some(public_status::spawn_public_status_publisher( + observer, + relay.rest_client(), + )); + tracing::info!("public work status enabled"); + } + } + let runtime_start_nonce = std::env::var("BUZZ_MANAGED_AGENT_START_NONCE").unwrap_or_default(); let dedup_mode = config.dedup_mode; let mut queue = @@ -3525,6 +3536,9 @@ async fn tokio_main() -> Result<()> { if let Some(handle) = relay_observer_publisher_task.take() { handle.abort(); } + if let Some(handle) = public_status_task.take() { + handle.abort(); + } // Graceful relay shutdown — sends WebSocket close frame and waits up to 5s // for the background task to finish, rather than aborting immediately (#40). @@ -6792,6 +6806,7 @@ mod build_mcp_servers_tests { persona_env_vars: vec![], has_generated_codex_config: false, relay_observer: false, + public_status: false, exit_after_inactivity_secs: 0, lazy_pool: false, idle_pool_sleep_secs: 0, @@ -7016,6 +7031,7 @@ mod error_outcome_emission_tests { persona_env_vars: vec![], has_generated_codex_config: false, relay_observer: false, + public_status: false, exit_after_inactivity_secs: 0, lazy_pool: false, idle_pool_sleep_secs: 0, diff --git a/crates/buzz-acp/src/public_status.rs b/crates/buzz-acp/src/public_status.rs new file mode 100644 index 00000000000..56b1e253a12 --- /dev/null +++ b/crates/buzz-acp/src/public_status.rs @@ -0,0 +1,645 @@ +//! Channel-visible agent work status: NIP-AW events (kind 30181) with the +//! channel UUID as both the `h` tag (NIP-29 channel scope — reads and writes +//! inherit channel-membership gating on the relay) and the `d` tag, so each +//! agent stores at most one replaceable status snapshot per channel. +//! +//! This is the public counterpart of the owner-encrypted observer stream +//! (NIP-AO, kind 24200): a whitelist projection of observer frames down to +//! status transitions, the current model id, and redacted tool-call titles. +//! Content bodies, tool arguments, prompts, and thoughts never enter the +//! payload — the projection copies only the fields named in `apply_event`. + +use std::collections::{HashMap, VecDeque}; +use std::time::Duration; + +use serde::Serialize; +use tokio::time::Instant; + +use crate::observer::{ObserverEvent, ObserverHandle}; +use crate::relay::RestClient; + +/// NIP-AW agent work status kind; `h` = `d` = channel UUID, one live +/// snapshot per (agent, channel), channel-membership-gated on the relay. +const PUBLIC_STATUS_KIND: u16 = buzz_core::kind::KIND_AGENT_WORK_STATUS as u16; +/// Newest-last redacted activity entries retained per channel. +const MAX_ACTIVITY_ENTRIES: usize = 20; +/// Hard cap on a single redacted title. +const MAX_TITLE_LEN: usize = 160; +/// Floor between non-urgent republications of one channel's status. +const MIN_PUBLISH_INTERVAL: Duration = Duration::from_secs(5); +/// Cadence of the flush loop that drains throttled updates. +const FLUSH_TICK: Duration = Duration::from_secs(1); +/// Budget for one status publication; failures are logged, never fatal. +const PUBLISH_TIMEOUT: Duration = Duration::from_secs(3); + +/// One redacted activity item: a title and a coarse status, nothing else. +#[derive(Clone, Debug, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ActivityEntry { + pub at: String, + pub kind: String, + pub title: String, + pub status: String, +} + +/// The published content body. Consumers key on `v` + `source` to +/// distinguish harness telemetry from ordinary NIP-38 statuses. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct PublicStatusPayload<'a> { + v: u32, + source: &'static str, + status: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + model: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + session_id: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + turn_id: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + turn_started_at: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + completed_at: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + stop_reason: Option<&'a str>, + updated_at: String, + activity: Vec<&'a ActivityEntry>, +} + +/// Mutable per-channel projection of the observer stream. +#[derive(Debug)] +pub struct ChannelWork { + status: &'static str, + model: Option, + session_id: Option, + turn_id: Option, + turn_started_at: Option, + completed_at: Option, + stop_reason: Option, + activity: VecDeque, + /// Something changed since the last publication. + dirty: bool, + /// A state transition that should bypass the publish floor. + urgent: bool, + last_published: Option, +} + +impl Default for ChannelWork { + fn default() -> Self { + Self { + status: "idle", + model: None, + session_id: None, + turn_id: None, + turn_started_at: None, + completed_at: None, + stop_reason: None, + activity: VecDeque::new(), + dirty: false, + urgent: false, + last_published: None, + } + } +} + +impl ChannelWork { + fn push_activity(&mut self, entry: ActivityEntry) { + if self.activity.len() == MAX_ACTIVITY_ENTRIES { + self.activity.pop_front(); + } + self.activity.push_back(entry); + } + + /// Whether this channel's status should be published now. + pub fn should_publish(&self, now: Instant) -> bool { + if self.urgent { + return true; + } + if !self.dirty { + return false; + } + match self.last_published { + None => true, + Some(at) => now.duration_since(at) >= MIN_PUBLISH_INTERVAL, + } + } + + fn mark_published(&mut self, now: Instant) { + self.dirty = false; + self.urgent = false; + self.last_published = Some(now); + } + + fn payload_json(&self, updated_at: String) -> serde_json::Value { + let payload = PublicStatusPayload { + v: 1, + source: "buzz-acp", + status: self.status, + model: self.model.as_deref(), + session_id: self.session_id.as_deref(), + turn_id: self.turn_id.as_deref(), + turn_started_at: self.turn_started_at.as_deref(), + completed_at: self.completed_at.as_deref(), + stop_reason: self.stop_reason.as_deref(), + updated_at, + activity: self.activity.iter().collect(), + }; + serde_json::to_value(&payload).unwrap_or_else(|_| serde_json::json!({})) + } +} + +fn truncate_title(title: &str) -> String { + if title.chars().count() <= MAX_TITLE_LEN { + return title.to_string(); + } + let mut out: String = title.chars().take(MAX_TITLE_LEN - 1).collect(); + out.push('…'); + out +} + +/// Map an ACP tool-call status string onto the coarse public vocabulary. +fn coarse_tool_status(status: &str) -> &'static str { + match status { + "completed" => "complete", + "failed" => "failed", + _ => "running", + } +} + +/// Fold one observer event into the per-channel projections. Every field +/// copied out of `event.payload` is named here — this function IS the +/// redaction boundary, so additions must stay whitelist-shaped. +pub fn apply_event(states: &mut HashMap, event: &ObserverEvent) { + let Some(channel_id) = event.channel_id.as_deref() else { + return; + }; + let state = states.entry(channel_id.to_string()).or_default(); + + match event.kind.as_str() { + "turn_started" => { + *state = ChannelWork { + status: "working", + model: state.model.take(), + session_id: event.session_id.clone(), + turn_id: event.turn_id.clone(), + turn_started_at: event.started_at.clone().or(Some(event.timestamp.clone())), + last_published: state.last_published, + dirty: true, + urgent: true, + ..ChannelWork::default() + }; + state.push_activity(ActivityEntry { + at: event.timestamp.clone(), + kind: "lifecycle".into(), + title: "Turn started".into(), + status: "complete".into(), + }); + } + "session_resolved" + if event.session_id.is_some() && state.session_id != event.session_id => + { + state.session_id = event.session_id.clone(); + state.dirty = true; + } + "session_config_captured" => { + if let Some(model) = event.payload["models"]["currentModelId"].as_str() { + if state.model.as_deref() != Some(model) { + state.model = Some(model.to_string()); + state.dirty = true; + } + } + } + // Refresh updatedAt for viewers computing elapsed time. + "turn_liveness" if state.status == "working" => { + state.dirty = true; + } + "turn_completed" => { + state.status = "complete"; + state.completed_at = Some(event.timestamp.clone()); + state.dirty = true; + state.urgent = true; + } + "turn_error" => { + state.status = "error"; + state.completed_at = Some(event.timestamp.clone()); + // `outcome` is a closed vocabulary; the error message itself may + // carry payload content and is deliberately not copied. + state.stop_reason = event.payload["outcome"].as_str().map(str::to_string); + state.dirty = true; + state.urgent = true; + } + "acp_read" => { + let update = &event.payload["params"]["update"]; + match update["sessionUpdate"].as_str() { + Some("tool_call") => { + let title = update["title"].as_str().unwrap_or("Tool call"); + state.push_activity(ActivityEntry { + at: event.timestamp.clone(), + kind: "tool".into(), + title: truncate_title(title), + status: coarse_tool_status(update["status"].as_str().unwrap_or("")) + .to_string(), + }); + state.dirty = true; + } + Some("tool_call_update") => { + if let Some(status) = update["status"].as_str() { + if let Some(entry) = state + .activity + .iter_mut() + .rev() + .find(|entry| entry.kind == "tool" && entry.status == "running") + { + entry.status = coarse_tool_status(status).into(); + state.dirty = true; + } + } + } + Some("agent_message_chunk") => { + let already_streaming = state + .activity + .back() + .is_some_and(|entry| entry.kind == "message" && entry.status == "running"); + if !already_streaming { + state.push_activity(ActivityEntry { + at: event.timestamp.clone(), + kind: "message".into(), + title: "Streaming reply".into(), + status: "running".into(), + }); + state.dirty = true; + } + } + _ => {} + } + } + _ => {} + } +} + +/// Build the signed NIP-AW replaceable status event for one channel. +/// +/// The relay requires `h` (channel scope + membership gating) and rejects any +/// event whose `d` tag differs from the `h` channel UUID. +fn build_status_event( + keys: &nostr::Keys, + channel_id: &str, + state: &ChannelWork, +) -> Result { + let updated_at = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + let content = state.payload_json(updated_at).to_string(); + let h_tag = nostr::Tag::parse(["h", channel_id]).map_err(|e| e.to_string())?; + let d_tag = nostr::Tag::parse(["d", channel_id]).map_err(|e| e.to_string())?; + nostr::EventBuilder::new(nostr::Kind::Custom(PUBLIC_STATUS_KIND), content) + .tags([h_tag, d_tag]) + .sign_with_keys(keys) + .map_err(|e| e.to_string()) +} + +async fn publish_channel_status(rest: &RestClient, channel_id: &str, state: &ChannelWork) { + let event = match build_status_event(&rest.keys, channel_id, state) { + Ok(event) => event, + Err(error) => { + tracing::warn!(target: "public_status", channel_id, "sign failed: {error}"); + return; + } + }; + match tokio::time::timeout(PUBLISH_TIMEOUT, rest.submit_event(&event)).await { + Ok(Ok(_)) => {} + Ok(Err(error)) => { + tracing::warn!(target: "public_status", channel_id, "publish failed: {error}"); + } + Err(_) => { + tracing::warn!(target: "public_status", channel_id, "publish timed out"); + } + } +} + +/// Spawn the public-status projection task: a second consumer of the observer +/// bus, independent of the owner-encrypted NIP-AO publisher. +pub fn spawn_public_status_publisher( + observer: ObserverHandle, + rest: RestClient, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + // Subscribe before snapshotting so no event falls between the two — + // the same loss-window closure the NIP-AO publisher uses. Snapshot + // replay is deduped by the monotonic `seq` high-water mark. + let rx = observer.subscribe(); + let snapshot = observer.snapshot(); + run_public_status_publisher(snapshot, rx, rest).await; + }) +} + +async fn run_public_status_publisher( + snapshot: Vec, + mut rx: tokio::sync::broadcast::Receiver, + rest: RestClient, +) { + let mut states: HashMap = HashMap::new(); + let max_snapshot_seq = snapshot.iter().map(|event| event.seq).max().unwrap_or(0); + for event in &snapshot { + apply_event(&mut states, event); + } + + let mut flush_tick = tokio::time::interval(FLUSH_TICK); + flush_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut closed = false; + loop { + tokio::select! { + result = rx.recv(), if !closed => { + match result { + Ok(event) => { + if event.seq <= max_snapshot_seq { + continue; + } + apply_event(&mut states, &event); + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => { + tracing::warn!(target: "public_status", dropped = count, "publisher lagged"); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + closed = true; + } + } + } + _ = flush_tick.tick() => { + let now = Instant::now(); + for (channel_id, state) in states.iter_mut() { + if state.should_publish(now) { + publish_channel_status(&rest, channel_id, state).await; + state.mark_published(now); + } + } + if closed { + break; + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn event(kind: &str, channel: Option<&str>, payload: serde_json::Value) -> ObserverEvent { + ObserverEvent { + seq: 1, + timestamp: "2026-08-22T19:00:00Z".into(), + kind: kind.into(), + agent_index: Some(0), + channel_id: channel.map(str::to_string), + session_id: Some("session-1".into()), + turn_id: Some("turn-1".into()), + started_at: Some("2026-08-22T18:59:59Z".into()), + payload, + } + } + + fn session_update(update: serde_json::Value) -> serde_json::Value { + serde_json::json!({"method": "session/update", "params": {"update": update}}) + } + + const CH: &str = "0b7c0958-3f7f-48c8-af3f-31e549b10e31"; + + #[test] + fn turn_lifecycle_projects_working_then_complete() { + let mut states = HashMap::new(); + apply_event( + &mut states, + &event("turn_started", Some(CH), serde_json::json!({})), + ); + let state = &states[CH]; + assert_eq!(state.status, "working"); + assert!(state.urgent); + assert_eq!( + state.turn_started_at.as_deref(), + Some("2026-08-22T18:59:59Z") + ); + + apply_event( + &mut states, + &event("turn_completed", Some(CH), serde_json::json!({})), + ); + let state = &states[CH]; + assert_eq!(state.status, "complete"); + assert_eq!(state.completed_at.as_deref(), Some("2026-08-22T19:00:00Z")); + assert!(state.urgent); + } + + #[test] + fn turn_error_copies_outcome_but_never_the_error_message() { + let mut states = HashMap::new(); + apply_event( + &mut states, + &event("turn_started", Some(CH), serde_json::json!({})), + ); + apply_event( + &mut states, + &event( + "turn_error", + Some(CH), + serde_json::json!({"outcome": "timeout", "error": "secret detail"}), + ), + ); + let state = &states[CH]; + assert_eq!(state.status, "error"); + assert_eq!(state.stop_reason.as_deref(), Some("timeout")); + let json = state.payload_json("t".into()).to_string(); + assert!(!json.contains("secret detail")); + } + + #[test] + fn tool_calls_are_projected_to_titles_and_updated_in_place() { + let mut states = HashMap::new(); + apply_event( + &mut states, + &event("turn_started", Some(CH), serde_json::json!({})), + ); + apply_event( + &mut states, + &event( + "acp_read", + Some(CH), + session_update(serde_json::json!({ + "sessionUpdate": "tool_call", + "title": "Shell command", + "kind": "execute", + "rawInput": {"command": "cat /etc/secret"}, + })), + ), + ); + let state = &states[CH]; + let tool = state.activity.back().unwrap(); + assert_eq!(tool.title, "Shell command"); + assert_eq!(tool.status, "running"); + assert!(!state + .payload_json("t".into()) + .to_string() + .contains("/etc/secret")); + + apply_event( + &mut states, + &event( + "acp_read", + Some(CH), + session_update(serde_json::json!({ + "sessionUpdate": "tool_call_update", + "toolCallId": "t1", + "status": "completed", + })), + ), + ); + assert_eq!(states[CH].activity.back().unwrap().status, "complete"); + } + + #[test] + fn titles_are_truncated_and_activity_is_capped() { + let mut states = HashMap::new(); + apply_event( + &mut states, + &event("turn_started", Some(CH), serde_json::json!({})), + ); + let long_title = "x".repeat(500); + for _ in 0..(MAX_ACTIVITY_ENTRIES + 10) { + apply_event( + &mut states, + &event( + "acp_read", + Some(CH), + session_update(serde_json::json!({ + "sessionUpdate": "tool_call", + "title": long_title, + })), + ), + ); + } + let state = &states[CH]; + assert_eq!(state.activity.len(), MAX_ACTIVITY_ENTRIES); + assert!(state.activity.back().unwrap().title.chars().count() <= MAX_TITLE_LEN); + } + + #[test] + fn streaming_chunks_collapse_into_one_running_entry() { + let mut states = HashMap::new(); + apply_event( + &mut states, + &event("turn_started", Some(CH), serde_json::json!({})), + ); + for _ in 0..5 { + apply_event( + &mut states, + &event( + "acp_read", + Some(CH), + session_update(serde_json::json!({ + "sessionUpdate": "agent_message_chunk", + "content": {"text": "private streamed text"}, + })), + ), + ); + } + let state = &states[CH]; + let streams = state + .activity + .iter() + .filter(|entry| entry.kind == "message") + .count(); + assert_eq!(streams, 1); + assert!(!state + .payload_json("t".into()) + .to_string() + .contains("private streamed")); + } + + #[test] + fn model_is_captured_from_session_config() { + let mut states = HashMap::new(); + apply_event( + &mut states, + &event( + "session_config_captured", + Some(CH), + serde_json::json!({"models": {"currentModelId": "gpt-5.6-sol"}}), + ), + ); + assert_eq!(states[CH].model.as_deref(), Some("gpt-5.6-sol")); + } + + #[test] + fn unknown_frames_and_channelless_events_are_ignored() { + let mut states = HashMap::new(); + apply_event( + &mut states, + &event("acp_write", Some(CH), serde_json::json!({"x": 1})), + ); + apply_event( + &mut states, + &event("turn_started", None, serde_json::json!({})), + ); + assert!(states.get(CH).is_none_or(|s| s.status == "idle")); + assert_eq!(states.len(), 1); + } + + #[test] + fn publish_gating_is_urgent_or_floored() { + let mut state = ChannelWork { + dirty: true, + urgent: false, + last_published: Some(Instant::now()), + ..ChannelWork::default() + }; + let now = Instant::now(); + assert!(!state.should_publish(now)); + state.urgent = true; + assert!(state.should_publish(now)); + state.urgent = false; + state.last_published = Some(now - MIN_PUBLISH_INTERVAL); + assert!(state.should_publish(now)); + state.dirty = false; + assert!(!state.should_publish(now)); + } + + #[test] + fn status_event_envelope_is_channel_scoped_nip_aw() { + let keys = nostr::Keys::generate(); + let mut states = HashMap::new(); + apply_event( + &mut states, + &event("turn_started", Some(CH), serde_json::json!({})), + ); + let signed = build_status_event(&keys, CH, &states[CH]).unwrap(); + assert_eq!( + signed.kind, + nostr::Kind::Custom(buzz_core::kind::KIND_AGENT_WORK_STATUS as u16) + ); + let tag_value = |name: &str| { + signed.tags.iter().find_map(|tag| { + let parts = tag.as_slice(); + (parts.first().map(|part| part.as_str()) == Some(name)) + .then(|| parts.get(1).map(|part| part.as_str())) + .flatten() + }) + }; + // The relay requires `h` for channel scoping and rejects `d` != `h`. + assert_eq!(tag_value("h"), Some(CH)); + assert_eq!(tag_value("d"), Some(CH)); + } + + #[test] + fn payload_carries_telemetry_markers_and_snake_free_keys() { + let mut states = HashMap::new(); + apply_event( + &mut states, + &event("turn_started", Some(CH), serde_json::json!({})), + ); + let json = states[CH].payload_json("2026-08-22T19:00:10Z".into()); + assert_eq!(json["v"], 1); + assert_eq!(json["source"], "buzz-acp"); + assert_eq!(json["status"], "working"); + assert_eq!(json["turnStartedAt"], "2026-08-22T18:59:59Z"); + assert_eq!(json["updatedAt"], "2026-08-22T19:00:10Z"); + assert!(json.get("completedAt").is_none()); + } +} diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index 3c6f1d5913d..60b9bf6032f 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -117,6 +117,17 @@ pub const KIND_PUSH_LEASE: u32 = 30350; /// plus exact public projection bindings. See `docs/nips/NIP-PMA.md`. pub const KIND_PRIVATE_MANAGED_AGENT: u32 = 30179; +/// NIP-AW: channel-visible agent work status (parameterized replaceable, +/// agent-authored). +/// +/// A redacted, channel-scoped projection of an agent's current work: turn +/// status, model id, and tool-call titles — never tool arguments, prompts, or +/// message content. Addressed by `(pubkey, kind, d_tag)` with `d` = channel +/// UUID and a matching `h` tag, so each agent stores at most one live snapshot +/// per channel and reads inherit NIP-29 channel-membership gating. See +/// `docs/nips/NIP-AW.md`. +pub const KIND_AGENT_WORK_STATUS: u32 = 30181; + /// Kinds whose stored events are readable only by their author. /// /// The relay must never reveal the existence, count, tags, content, schedule, @@ -657,6 +668,7 @@ pub const ALL_KINDS: &[u32] = &[ KIND_MANAGED_AGENT, KIND_TEAM_CATALOG, KIND_PRIVATE_MANAGED_AGENT, + KIND_AGENT_WORK_STATUS, KIND_REPORT, KIND_PRODUCT_FEEDBACK, KIND_NIP29_PUT_USER, @@ -858,6 +870,7 @@ const _: () = assert!(is_parameterized_replaceable(KIND_TEAM)); // 30176 ∈ 300 const _: () = assert!(is_parameterized_replaceable(KIND_MANAGED_AGENT)); // 30177 ∈ 30000–39999 const _: () = assert!(is_parameterized_replaceable(KIND_TEAM_CATALOG)); // 30178 ∈ 30000–39999 const _: () = assert!(is_parameterized_replaceable(KIND_PRIVATE_MANAGED_AGENT)); // 30179 ∈ 30000–39999 +const _: () = assert!(is_parameterized_replaceable(KIND_AGENT_WORK_STATUS)); // 30181 ∈ 30000–39999 const _: () = assert!(is_parameterized_replaceable(KIND_WORKFLOW_DEF)); // 30620 ∈ 30000–39999 const _: () = assert!(is_parameterized_replaceable(KIND_EVENT_REMINDER)); // 30300 ∈ 30000–39999 const _: () = assert!(is_parameterized_replaceable(KIND_DM_VISIBILITY)); // 30622 ∈ 30000–39999 diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index ccba40f3282..db3d8e92d48 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -38,7 +38,7 @@ pub(crate) fn bounded_kind_label(kind: u32) -> String { 8000..=8003 | 9000..=9022 | 9030..=9036 => kind.to_string(), 13534..=13535 => kind.to_string(), 20000..=29999 => kind.to_string(), - 30023 | 30315 | 39000..=39003 => kind.to_string(), + 30023 | 30181 | 30315 | 39000..=39003 => kind.to_string(), 40002..=40100 => kind.to_string(), 41001 | 41010..=41012 => kind.to_string(), 43001..=43006 => kind.to_string(), diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index dd2fa6e93e0..53029de7200 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -13,28 +13,28 @@ use buzz_auth::Scope; use buzz_core::kind::{ event_kind_u32, is_identity_archive_request_kind, is_parameterized_replaceable, is_relay_admin_kind, KIND_AGENT_ENGRAM, KIND_AGENT_PROFILE, KIND_AGENT_TURN_METRIC, - KIND_APPROVAL_DENY, KIND_APPROVAL_GRANT, KIND_AUTH, KIND_BOOKMARK_LIST, KIND_BOOKMARK_SET, - KIND_CANVAS, KIND_CONTACT_LIST, KIND_DELETION, KIND_DM_ADD_MEMBER, KIND_DM_HIDE, KIND_DM_OPEN, - KIND_EMOJI_LIST, KIND_EMOJI_SET, KIND_EVENT_REMINDER, KIND_FOLLOW_SET, KIND_FORUM_COMMENT, - KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_GIFT_WRAP, KIND_GIT_ISSUE, KIND_GIT_PATCH, - KIND_GIT_PR_UPDATE, KIND_GIT_PULL_REQUEST, KIND_GIT_REPO_ANNOUNCEMENT, KIND_GIT_REPO_STATE, - KIND_GIT_STATUS_CLOSED, KIND_GIT_STATUS_DRAFT, KIND_GIT_STATUS_MERGED, KIND_GIT_STATUS_OPEN, - KIND_HUDDLE_ENDED, KIND_HUDDLE_GUIDELINES, KIND_HUDDLE_PARTICIPANT_JOINED, - KIND_HUDDLE_PARTICIPANT_LEFT, KIND_HUDDLE_STARTED, KIND_IA_ARCHIVE_REQUEST, - KIND_IA_UNARCHIVE_REQUEST, KIND_LONG_FORM, KIND_MANAGED_AGENT, KIND_MEMBER_ADDED_NOTIFICATION, - KIND_MEMBER_REMOVED_NOTIFICATION, KIND_MODERATION_BAN, KIND_MODERATION_RESOLVE_REPORT, - KIND_MODERATION_TIMEOUT, KIND_MODERATION_UNBAN, KIND_MODERATION_UNTIMEOUT, KIND_MUTE_LIST, - KIND_NIP29_CREATE_GROUP, KIND_NIP29_DELETE_EVENT, KIND_NIP29_DELETE_GROUP, - KIND_NIP29_EDIT_METADATA, KIND_NIP29_JOIN_REQUEST, KIND_NIP29_LEAVE_REQUEST, - KIND_NIP29_PUT_USER, KIND_NIP29_REMOVE_USER, KIND_NIP43_LEAVE_REQUEST, - KIND_NIP65_RELAY_LIST_METADATA, KIND_PERSONA, KIND_PIN_LIST, KIND_PRESENCE_UPDATE, - KIND_PRIVATE_MANAGED_AGENT, KIND_PRODUCT_FEEDBACK, KIND_PROFILE, KIND_PROJECT, KIND_REACTION, - KIND_READ_STATE, KIND_REPORT, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_BOOKMARKED, - KIND_STREAM_MESSAGE_DIFF, KIND_STREAM_MESSAGE_EDIT, KIND_STREAM_MESSAGE_PINNED, - KIND_STREAM_MESSAGE_SCHEDULED, KIND_STREAM_MESSAGE_V2, KIND_STREAM_REMINDER, KIND_TEAM, - KIND_TEAM_CATALOG, KIND_TEXT_NOTE, KIND_USER_STATUS, KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, - RELAY_ADMIN_ADD_MEMBER, RELAY_ADMIN_CHANGE_ROLE, RELAY_ADMIN_REMOVE_MEMBER, - RELAY_ADMIN_SET_WORKSPACE_PROFILE, + KIND_AGENT_WORK_STATUS, KIND_APPROVAL_DENY, KIND_APPROVAL_GRANT, KIND_AUTH, KIND_BOOKMARK_LIST, + KIND_BOOKMARK_SET, KIND_CANVAS, KIND_CONTACT_LIST, KIND_DELETION, KIND_DM_ADD_MEMBER, + KIND_DM_HIDE, KIND_DM_OPEN, KIND_EMOJI_LIST, KIND_EMOJI_SET, KIND_EVENT_REMINDER, + KIND_FOLLOW_SET, KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_GIFT_WRAP, + KIND_GIT_ISSUE, KIND_GIT_PATCH, KIND_GIT_PR_UPDATE, KIND_GIT_PULL_REQUEST, + KIND_GIT_REPO_ANNOUNCEMENT, KIND_GIT_REPO_STATE, KIND_GIT_STATUS_CLOSED, KIND_GIT_STATUS_DRAFT, + KIND_GIT_STATUS_MERGED, KIND_GIT_STATUS_OPEN, KIND_HUDDLE_ENDED, KIND_HUDDLE_GUIDELINES, + KIND_HUDDLE_PARTICIPANT_JOINED, KIND_HUDDLE_PARTICIPANT_LEFT, KIND_HUDDLE_STARTED, + KIND_IA_ARCHIVE_REQUEST, KIND_IA_UNARCHIVE_REQUEST, KIND_LONG_FORM, KIND_MANAGED_AGENT, + KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_MODERATION_BAN, + KIND_MODERATION_RESOLVE_REPORT, KIND_MODERATION_TIMEOUT, KIND_MODERATION_UNBAN, + KIND_MODERATION_UNTIMEOUT, KIND_MUTE_LIST, KIND_NIP29_CREATE_GROUP, KIND_NIP29_DELETE_EVENT, + KIND_NIP29_DELETE_GROUP, KIND_NIP29_EDIT_METADATA, KIND_NIP29_JOIN_REQUEST, + KIND_NIP29_LEAVE_REQUEST, KIND_NIP29_PUT_USER, KIND_NIP29_REMOVE_USER, + KIND_NIP43_LEAVE_REQUEST, KIND_NIP65_RELAY_LIST_METADATA, KIND_PERSONA, KIND_PIN_LIST, + KIND_PRESENCE_UPDATE, KIND_PRIVATE_MANAGED_AGENT, KIND_PRODUCT_FEEDBACK, KIND_PROFILE, + KIND_PROJECT, KIND_REACTION, KIND_READ_STATE, KIND_REPORT, KIND_STREAM_MESSAGE, + KIND_STREAM_MESSAGE_BOOKMARKED, KIND_STREAM_MESSAGE_DIFF, KIND_STREAM_MESSAGE_EDIT, + KIND_STREAM_MESSAGE_PINNED, KIND_STREAM_MESSAGE_SCHEDULED, KIND_STREAM_MESSAGE_V2, + KIND_STREAM_REMINDER, KIND_TEAM, KIND_TEAM_CATALOG, KIND_TEXT_NOTE, KIND_USER_STATUS, + KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, RELAY_ADMIN_ADD_MEMBER, RELAY_ADMIN_CHANGE_ROLE, + RELAY_ADMIN_REMOVE_MEMBER, RELAY_ADMIN_SET_WORKSPACE_PROFILE, }; use buzz_core::tenant::TenantContext; use buzz_core::verification::verify_event; @@ -445,6 +445,10 @@ fn required_scope_for_kind(kind: u32, event: &Event) -> Result Ok(Scope::MessagesWrite), + // NIP-AW: channel-scoped agent work status — channel content, same + // transport scope as messages; membership is enforced by the generic + // h-tag channel gate. + KIND_AGENT_WORK_STATUS => Ok(Scope::MessagesWrite), // NIP-56 reports are ordinary member writes into the mod-only queue. // Ingest persists them to `moderation_reports` and suppresses public // storage/fanout; reports are signals, never enforcement triggers. @@ -729,6 +733,10 @@ pub(crate) fn requires_h_channel_scope(kind: u32) -> bool { | KIND_HUDDLE_PARTICIPANT_LEFT | KIND_HUDDLE_ENDED | KIND_HUDDLE_GUIDELINES + // NIP-AW: agent work status is meaningless without a channel — a + // missing h tag would store it globally, readable by any relay + // member, defeating the channel-membership privacy contract. + | KIND_AGENT_WORK_STATUS ) } @@ -1881,6 +1889,24 @@ fn validate_engram_nip44_content(content: &str) -> Result<(), String> { /// /// Ownership (`is_agent_owner`) is an async DB check performed separately in /// `ingest_event_inner` after this synchronous envelope check. +/// NIP-AW: the addressable coordinate must be the channel itself. +/// +/// `d` must be exactly one tag whose value is the same UUID as the `h` tag +/// (already resolved to `channel_id` by the caller). This pins one live status +/// per (agent, channel): a divergent `d` would let an author store multiple +/// snapshots for one channel, or collide a coordinate across channels. +fn validate_agent_work_status_envelope(event: &Event, channel_id: Uuid) -> Result<(), String> { + const LABEL: &str = "agent work status"; + let d = single_bounded_d_tag(event, LABEL)?; + let d_uuid = d + .parse::() + .map_err(|_| format!("{LABEL} `d` tag must be a channel UUID"))?; + if d_uuid != channel_id { + return Err(format!("{LABEL} `d` tag must match the `h` channel tag")); + } + Ok(()) +} + fn validate_agent_turn_metric_envelope(event: &nostr::Event) -> Result<(), String> { let event_pubkey_hex = event.pubkey.to_hex(); let mut p_tags: Vec<&str> = Vec::new(); @@ -2787,6 +2813,14 @@ async fn ingest_event_inner( .map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?; } + if kind_u32 == KIND_AGENT_WORK_STATUS { + // requires_h_channel_scope guarantees channel_id is Some here. + if let Some(ch_id) = channel_id { + validate_agent_work_status_envelope(&event, ch_id) + .map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?; + } + } + // Track pre-created channel UUID for compensation on insert failure. let mut pre_created_channel: Option = None; @@ -3756,6 +3790,54 @@ mod tests { assert!(!requires_h_channel_scope(KIND_USER_STATUS)); } + #[test] + fn agent_work_status_is_channel_scoped_messages_write() { + let dummy = make_dummy_event(); + assert_eq!( + required_scope_for_kind(KIND_AGENT_WORK_STATUS, &dummy).unwrap(), + Scope::MessagesWrite, + ); + // Channel-scoped by contract: h required, never stored globally. + assert!(requires_h_channel_scope(KIND_AGENT_WORK_STATUS)); + assert!(!is_global_only_kind(KIND_AGENT_WORK_STATUS)); + } + + #[test] + fn agent_work_status_d_must_match_h_channel() { + let ch = uuid::Uuid::parse_str("0b7c0958-3f7f-48c8-af3f-31e549b10e31").unwrap(); + let other = uuid::Uuid::parse_str("11111111-2222-3333-4444-555555555555").unwrap(); + let ch_str = ch.to_string(); + + let good = make_event_with_tags( + KIND_AGENT_WORK_STATUS, + "{}", + &[&["h", ch_str.as_str()], &["d", ch_str.as_str()]], + ); + assert!(validate_agent_work_status_envelope(&good, ch).is_ok()); + + let mismatched = make_event_with_tags( + KIND_AGENT_WORK_STATUS, + "{}", + &[&["h", ch_str.as_str()], &["d", other.to_string().as_str()]], + ); + assert!(validate_agent_work_status_envelope(&mismatched, ch) + .unwrap_err() + .contains("must match the `h` channel tag")); + + let non_uuid = make_event_with_tags( + KIND_AGENT_WORK_STATUS, + "{}", + &[&["h", ch_str.as_str()], &["d", "general"]], + ); + assert!(validate_agent_work_status_envelope(&non_uuid, ch) + .unwrap_err() + .contains("must be a channel UUID")); + + let missing_d = + make_event_with_tags(KIND_AGENT_WORK_STATUS, "{}", &[&["h", ch_str.as_str()]]); + assert!(validate_agent_work_status_envelope(&missing_d, ch).is_err()); + } + #[test] fn private_sidecars_and_moderation_commands_require_messages_write_scope() { let dummy = make_dummy_event(); diff --git a/crates/buzz-test-client/tests/e2e_relay.rs b/crates/buzz-test-client/tests/e2e_relay.rs index b119d267740..997d5fe2508 100644 --- a/crates/buzz-test-client/tests/e2e_relay.rs +++ b/crates/buzz-test-client/tests/e2e_relay.rs @@ -2517,6 +2517,161 @@ async fn test_private_channel_non_member_cannot_invite() { .expect("disconnect outsider"); } +/// NIP-AW agent work status (kind 30181): writes require channel membership +/// and a `d` tag matching the `h` channel; stored statuses are readable by +/// members and invisible to non-members. +#[tokio::test] +#[ignore] +async fn test_agent_work_status_channel_gating() { + const WORK_STATUS_KIND: u16 = 30181; + let url = relay_url(); + let owner_keys = Keys::generate(); + let agent_keys = Keys::generate(); + let outsider_keys = Keys::generate(); + + // Owner creates a private channel and adds the agent as a member. + let mut owner_client = BuzzTestClient::connect(&url, &owner_keys) + .await + .expect("connect as owner"); + let channel_id = create_private_channel_ws(&mut owner_client, &owner_keys).await; + let (accepted, msg) = add_member_ws( + &mut owner_client, + &channel_id, + &agent_keys.public_key().to_hex(), + &owner_keys, + ) + .await; + assert!(accepted, "owner should add agent as member, got: {msg}"); + + let status_event = |keys: &Keys, h: &str, d: &str| { + EventBuilder::new( + Kind::Custom(WORK_STATUS_KIND), + r#"{"v":1,"source":"buzz-acp","status":"working","activity":[]}"#, + ) + .tags(vec![ + Tag::parse(["h", h]).unwrap(), + Tag::parse(["d", d]).unwrap(), + ]) + .sign_with_keys(keys) + .unwrap() + }; + + // Member agent publishes its status: accepted. + let mut agent_client = BuzzTestClient::connect(&url, &agent_keys) + .await + .expect("connect as agent"); + let ok = agent_client + .send_event(status_event(&agent_keys, &channel_id, &channel_id)) + .await + .expect("send work status"); + assert!(ok.accepted, "member work status rejected: {}", ok.message); + + // `d` diverging from the `h` channel is rejected. + let other_uuid = uuid::Uuid::new_v4().to_string(); + let ok = agent_client + .send_event(status_event(&agent_keys, &channel_id, &other_uuid)) + .await + .expect("send mismatched work status"); + assert!( + !ok.accepted, + "work status with d != h must be rejected, but was accepted" + ); + assert!( + ok.message.contains("must match"), + "rejection should mention the d/h mismatch, got: {}", + ok.message + ); + + // A missing `h` tag is rejected — the kind is channel-scoped by contract. + let no_h = EventBuilder::new(Kind::Custom(WORK_STATUS_KIND), "{}") + .tags(vec![Tag::parse(["d", channel_id.as_str()]).unwrap()]) + .sign_with_keys(&agent_keys) + .unwrap(); + let ok = agent_client + .send_event(no_h) + .await + .expect("send h-less work status"); + assert!( + !ok.accepted, + "work status without an h tag must be rejected, but was accepted" + ); + + // A non-member cannot publish a status into the channel. + let mut outsider_client = BuzzTestClient::connect(&url, &outsider_keys) + .await + .expect("connect as outsider"); + let ok = outsider_client + .send_event(status_event(&outsider_keys, &channel_id, &channel_id)) + .await + .expect("send outsider work status"); + assert!( + !ok.accepted, + "non-member work status must be rejected, but was accepted" + ); + + // A member reads the stored status; a non-member sees nothing. + let filter = Filter::new() + .kind(Kind::Custom(WORK_STATUS_KIND)) + .custom_tags( + SingleLetterTag::lowercase(Alphabet::H), + [channel_id.as_str()], + ); + + let sid = sub_id("work-status-member"); + owner_client + .subscribe(&sid, vec![filter.clone()]) + .await + .expect("owner subscribe"); + let member_view = owner_client + .collect_until_eose(&sid, Duration::from_secs(5)) + .await + .expect("owner EOSE"); + assert_eq!( + member_view.len(), + 1, + "channel member should read exactly the one live status snapshot" + ); + assert_eq!(member_view[0].pubkey, agent_keys.public_key()); + + // The relay rejects a non-member's channel subscription outright: CLOSED + // with a membership restriction, never events, never a silent EOSE. + let sid = sub_id("work-status-outsider"); + outsider_client + .subscribe(&sid, vec![filter]) + .await + .expect("outsider subscribe"); + match outsider_client + .recv_event(Duration::from_secs(5)) + .await + .expect("outsider subscription response") + { + RelayMessage::Closed { + subscription_id, + message, + } => { + assert_eq!(subscription_id, sid); + assert!( + message.contains("not a channel member"), + "closure should cite membership, got: {message}" + ); + } + RelayMessage::Event { event, .. } => { + panic!( + "non-member must not read channel work status, got event from {}", + event.pubkey + ); + } + other => panic!("expected CLOSED for non-member work-status REQ, got: {other:?}"), + } + + owner_client.disconnect().await.expect("disconnect owner"); + agent_client.disconnect().await.expect("disconnect agent"); + outsider_client + .disconnect() + .await + .expect("disconnect outsider"); +} + /// Regular members cannot grant elevated roles (owner/admin) in private channels. #[tokio::test] #[ignore] diff --git a/docs/nips/NIP-AW.md b/docs/nips/NIP-AW.md new file mode 100644 index 00000000000..805674bee64 --- /dev/null +++ b/docs/nips/NIP-AW.md @@ -0,0 +1,115 @@ +NIP-AW +====== + +Agent Work Status +----------------- + +`draft` `optional` + +This NIP defines a channel-scoped, redacted, replaceable event kind through +which an AI agent publishes its current work status to the members of a NIP-29 +channel. + +## Motivation + +Buzz channels host agents that work in long turns: they run tools, stream +replies, and complete or fail. Channel members want a live answer to "what is +this agent doing right now?" without operator-side collectors, SSH access to +the agent's host, or membership-independent broadcast. + +NIP-AO (kind 24200) already streams rich, encrypted telemetry — but only to +the agent's owner, and ephemerally. Kind 30181 is the complementary lane: a +persistent, heavily redacted snapshot whose audience is exactly the channel's +membership. The design bar is that **work status is precisely as private as +the channel it describes** — anyone who can read the channel's messages may +see its status snapshots; nobody else can, and the relay enforces this +server-side with the same membership gate it applies to messages. + +## Event Structure + +```json +{ + "kind": 30181, + "pubkey": "", + "created_at": , + "content": "", + "tags": [ + ["h", ""], + ["d", ""] + ] +} +``` + +Kind 30181 is parameterized replaceable (NIP-33): the relay retains only the +newest event per `(pubkey, kind, d)`. With `d` = channel UUID, each agent +stores at most one live snapshot per channel — a status lane, not a history +log. + +Relays implementing this NIP MUST enforce: + +- **`h` is REQUIRED.** The event is channel content; a missing `h` tag MUST + reject rather than store the event globally. +- **`d` MUST equal `h`.** Exactly one `d` tag, whose value is the same channel + UUID as the `h` tag. A divergent `d` would let an author store multiple + snapshots per channel or collide coordinates across channels. +- **Writes require channel membership** (or open channel visibility), exactly + as for NIP-29 channel messages. +- **Reads are membership-gated.** The stored event is visible only to clients + the relay would allow to read the channel's messages. Non-members MUST NOT + receive it in REQ results, live fan-out, COUNT, or search. + +## Payload + +`content` is plaintext JSON (it is exactly as private as the channel — see +Motivation; channel messages themselves are relay-gated, not E2E): + +```json +{ + "v": 1, + "source": "buzz-acp", + "status": "working" | "complete" | "error" | "idle", + "model": "", + "sessionId": "", + "turnId": "", + "turnStartedAt": "", + "completedAt": "", + "stopReason": "", + "updatedAt": "", + "activity": [ + { "at": "", "kind": "tool" | "message" | "lifecycle", + "title": "", "status": "running" | "complete" } + ] +} +``` + +`v`, `source`, `status`, `updatedAt`, and `activity` are REQUIRED; the rest +are OPTIONAL. Consumers key on `v` + `source` and MUST ignore unknown fields. + +### Redaction contract + +The payload is whitelist-shaped at the publisher: turn status transitions, the +model id, timestamps, and tool-call **titles** only. Tool arguments, file +paths, prompts, streamed message content, and error message text MUST NOT +appear in any field. Streaming output is collapsed to a single `message` +activity entry with a generic title. Publishers SHOULD cap title length and +the number of activity entries (the reference implementation keeps 20 entries +of at most 160 characters). + +## Publisher Behavior + +Publishers SHOULD pace republication (the reference implementation publishes +immediately on turn start/complete/error and otherwise at most once per 5 +seconds per channel) and MUST treat publish failures as non-fatal telemetry +loss, never as turn failure. + +## Relation to Other NIPs + +| Lane | Kind | Audience | Content | +|------|------|----------|---------| +| NIP-AO observer frames | 24200 (ephemeral) | agent owner only | full telemetry, NIP-44 encrypted | +| **NIP-AW work status** | **30181 (replaceable)** | **channel members** | **redacted snapshot, plaintext** | +| NIP-38 user status | 30315 (replaceable) | relay-global | free-form user presence | + +Kind 30315 (NIP-38) is unsuitable for this purpose on Buzz relays: it is a +global, author-owned kind — any authenticated relay member can read it — so it +cannot honor the channel-membership privacy bar this NIP requires.