diff --git a/crates/sprout-acp/src/config.rs b/crates/sprout-acp/src/config.rs index 28e5d0c53..17ba3408c 100644 --- a/crates/sprout-acp/src/config.rs +++ b/crates/sprout-acp/src/config.rs @@ -153,6 +153,14 @@ pub struct CliArgs { #[arg(long, env = "SPROUT_ACP_CONTEXT_MESSAGE_LIMIT", default_value_t = 12, value_parser = clap::value_parser!(u32).range(0..=100))] pub context_message_limit: u32, + + /// Disable automatic presence (online/offline) status. + #[arg(long, env = "SPROUT_ACP_NO_PRESENCE")] + pub no_presence: bool, + + /// Disable typing indicators while agent is processing. + #[arg(long, env = "SPROUT_ACP_NO_TYPING")] + pub no_typing: bool, } // ── Merged NIP-01 filter ────────────────────────────────────────────────────── @@ -190,6 +198,8 @@ pub struct Config { pub no_mention_filter: bool, pub config_path: PathBuf, pub context_message_limit: u32, + pub presence_enabled: bool, + pub typing_enabled: bool, } impl Config { @@ -269,13 +279,15 @@ impl Config { no_mention_filter: args.no_mention_filter, config_path: args.config, context_message_limit: args.context_message_limit, + presence_enabled: !args.no_presence, + typing_enabled: !args.no_typing, }) } /// Human-readable summary (no secrets). pub fn summary(&self) -> String { format!( - "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} timeout={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} ignore_self={} context_limit={}", + "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} timeout={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} ignore_self={} context_limit={} presence={} typing={}", self.relay_url, self.keys.public_key().to_hex(), self.agent_command, @@ -288,6 +300,8 @@ impl Config { self.dedup_mode, self.ignore_self, self.context_message_limit, + self.presence_enabled, + self.typing_enabled, ) } } @@ -582,6 +596,8 @@ mod tests { no_mention_filter: false, config_path: PathBuf::from("./sprout-acp.toml"), context_message_limit: 12, + presence_enabled: true, + typing_enabled: true, } } diff --git a/crates/sprout-acp/src/main.rs b/crates/sprout-acp/src/main.rs index 26531f30c..64c6eadeb 100644 --- a/crates/sprout-acp/src/main.rs +++ b/crates/sprout-acp/src/main.rs @@ -74,6 +74,18 @@ async fn main() -> Result<()> { .map_err(|e| anyhow::anyhow!("membership notification subscribe error: {e}"))?; tracing::info!("subscribed to membership notifications"); + // ── Step 2c: Set initial presence ───────────────────────────────────────── + let rest_client_for_presence = relay.rest_client(); + if config.presence_enabled { + match rest_client_for_presence + .put_json("/api/presence", &serde_json::json!({"status": "online"})) + .await + { + Ok(_) => tracing::info!("presence set to online"), + Err(e) => tracing::warn!("failed to set initial presence: {e}"), + } + } + // ── Step 3: Discover channels and build subscription rules ──────────────── let channel_info_map = relay .discover_channels() @@ -167,6 +179,30 @@ async fn main() -> Result<()> { }; let mut heartbeat_in_flight = false; + // ── Step 6b: Presence heartbeat timer (refreshes 90s TTL every 60s) ─────── + let mut presence_heartbeat = if config.presence_enabled { + let interval = Duration::from_secs(60); + Some(tokio::time::interval_at( + tokio::time::Instant::now() + interval, + interval, + )) + } else { + None + }; + + // ── Step 6c: Typing refresh timer (re-publishes kind:20002 every 3s) ────── + let mut typing_refresh = if config.typing_enabled { + let interval = Duration::from_secs(3); + Some(tokio::time::interval_at( + tokio::time::Instant::now() + interval, + interval, + )) + } else { + None + }; + let mut typing_channels: HashSet = HashSet::new(); + let mut presence_task: Option> = None; + // ── Step 7: Shutdown signal ─────────────────────────────────────────────── let (shutdown_tx, mut shutdown_rx) = watch::channel(()); @@ -326,6 +362,7 @@ async fn main() -> Result<()> { // Track removed channels so checked-out agents get // their sessions stripped when they return to the pool. removed_channels.insert(ch); + typing_channels.remove(&ch); if drained > 0 || invalidated > 0 { tracing::info!( channel_id = %ch, @@ -357,7 +394,7 @@ async fn main() -> Result<()> { received_at: std::time::Instant::now(), prompt_tag, }); - dispatch_pending(&mut pool, &mut queue, &ctx); + typing_channels.extend(dispatch_pending(&mut pool, &mut queue, &ctx)); } None => { tracing::warn!("relay event stream ended — requesting reconnect"); @@ -379,7 +416,7 @@ async fn main() -> Result<()> { let _ = result_rx; if queue.has_flushable_work() { tracing::debug!("heartbeat_skipped_events"); - dispatch_pending(&mut pool, &mut queue, &ctx); + typing_channels.extend(dispatch_pending(&mut pool, &mut queue, &ctx)); } else if pool.any_idle() { dispatch_heartbeat(&mut pool, &ctx, &mut heartbeat_in_flight); } else { @@ -387,6 +424,41 @@ async fn main() -> Result<()> { } None } + _ = async { + match presence_heartbeat.as_mut() { + Some(t) => t.tick().await, + None => std::future::pending().await, + } + } => { + let _ = result_rx; + // Abort previous heartbeat if still in flight (prevents race on shutdown). + if let Some(h) = presence_task.take() { + h.abort(); + } + let rc = rest_client_for_presence.clone(); + presence_task = Some(tokio::spawn(async move { + if let Err(e) = rc.put_json("/api/presence", &serde_json::json!({"status": "online"})).await { + tracing::warn!("presence heartbeat failed: {e}"); + } + })); + None + } + _ = async { + match typing_refresh.as_mut() { + Some(t) => t.tick().await, + None => std::future::pending().await, + } + } => { + let _ = result_rx; + for &ch in &typing_channels { + if let Ok(event) = relay.build_typing_event(ch) { + if let Err(e) = relay.publish_event(event).await { + tracing::debug!("typing indicator failed for {ch}: {e}"); + } + } + } + None + } _ = shutdown_rx.changed() => { tracing::info!("shutting down"); break; @@ -396,6 +468,10 @@ async fn main() -> Result<()> { match pool_event { Some(PoolEvent::Result(result)) => { + // Stop typing indicator for the completed channel. + if let PromptSource::Channel(ch) = &result.source { + typing_channels.remove(ch); + } if handle_prompt_result( &mut pool, &mut queue, @@ -415,13 +491,14 @@ async fn main() -> Result<()> { &config, &mut heartbeat_in_flight, &removed_channels, + &mut typing_channels, ) .await == LoopAction::Exit { break; } - dispatch_pending(&mut pool, &mut queue, &ctx); + typing_channels.extend(dispatch_pending(&mut pool, &mut queue, &ctx)); } Some(PoolEvent::Panic(join_error)) => { tracing::error!("agent task panicked: {join_error}"); @@ -432,13 +509,14 @@ async fn main() -> Result<()> { join_error, &mut heartbeat_in_flight, &removed_channels, + &mut typing_channels, ) .await; if pool.live_count() == 0 { tracing::error!("all agents dead — exiting"); break; } - dispatch_pending(&mut pool, &mut queue, &ctx); + typing_channels.extend(dispatch_pending(&mut pool, &mut queue, &ctx)); } None => {} // relay/heartbeat/shutdown branches handled inline above } @@ -460,6 +538,27 @@ async fn main() -> Result<()> { pool.join_set.shutdown().await; } drop(pool); + + // Cancel any in-flight presence heartbeat before sending offline. + if let Some(h) = presence_task.take() { + h.abort(); + } + + // Best-effort: set presence to offline before exiting. + if config.presence_enabled { + match tokio::time::timeout( + Duration::from_secs(2), + rest_client_for_presence + .put_json("/api/presence", &serde_json::json!({"status": "offline"})), + ) + .await + { + Ok(Ok(_)) => tracing::info!("presence set to offline"), + Ok(Err(e)) => tracing::warn!("failed to set offline presence: {e}"), + Err(_) => tracing::warn!("offline presence timed out"), + } + } + tracing::info!("sprout-acp stopped"); Ok(()) } @@ -475,8 +574,12 @@ enum LoopAction { // ── dispatch_pending ────────────────────────────────────────────────────────── /// Flush queued work to available agents. -fn dispatch_pending(pool: &mut AgentPool, queue: &mut EventQueue, ctx: &Arc) { - let mut dispatched: usize = 0; +fn dispatch_pending( + pool: &mut AgentPool, + queue: &mut EventQueue, + ctx: &Arc, +) -> Vec { + let mut dispatched_channels = Vec::new(); loop { let batch = match queue.flush_next() { Some(b) => b, @@ -519,13 +622,14 @@ fn dispatch_pending(pool: &mut AgentPool, queue: &mut EventQueue, ctx: &Arc, + typing_channels: &mut HashSet, ) { let task_id = join_error.id(); let Some(meta) = pool.task_map_mut().remove(&task_id) else { @@ -631,6 +736,7 @@ async fn recover_panicked_agent( if let Some(ch) = meta.channel_id { queue.mark_complete(ch); + typing_channels.remove(&ch); tracing::warn!("cleared wedged in-flight channel {ch} from panicked agent {i}"); } else { *heartbeat_in_flight = false; @@ -674,6 +780,7 @@ async fn drain_ready_join_results( config: &Config, heartbeat_in_flight: &mut bool, removed_channels: &HashSet, + typing_channels: &mut HashSet, ) -> LoopAction { while let Some(Some(join_result)) = pool.join_set.join_next().now_or_never() { if let Err(join_error) = join_result { @@ -685,6 +792,7 @@ async fn drain_ready_join_results( join_error, heartbeat_in_flight, removed_channels, + typing_channels, ) .await; if pool.live_count() == 0 { diff --git a/crates/sprout-acp/src/relay.rs b/crates/sprout-acp/src/relay.rs index da69a51f9..c2aa0cc23 100644 --- a/crates/sprout-acp/src/relay.rs +++ b/crates/sprout-acp/src/relay.rs @@ -1,12 +1,9 @@ //! Harness-side Sprout relay client. //! //! Connects to the Sprout relay via NIP-01 WebSocket, authenticates via NIP-42, -//! discovers channels via REST API, and streams matching events back to the -//! harness main loop. -//! -//! This is a simplified receive-only client adapted from `sprout-mcp`'s -//! `relay_client.rs`. It does not publish events or perform queries — it only -//! subscribes and receives. +//! discovers channels via REST API, and streams events back to the harness main +//! loop. Also publishes ephemeral events (typing indicators) via the same +//! WebSocket connection. //! //! ## Architecture //! @@ -15,6 +12,7 @@ //! - Forwards `SproutEvent`s through an `mpsc` channel //! - Handles reconnection with `since` filters to avoid event loss //! - Responds to mid-session AUTH challenges +//! - Publishes ephemeral events (typing indicators) via `PublishEvent` commands //! //! `HarnessRelay` communicates with the background task via a `RelayCommand` //! channel. `next_event()` reads from the event receiver. @@ -51,7 +49,9 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); use futures_util::{SinkExt, StreamExt}; use nostr::{Event, EventBuilder, Keys, Kind, Tag, Url as NostrUrl}; use serde_json::{json, Value}; -use sprout_core::kind::{KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION}; +use sprout_core::kind::{ + KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_TYPING_INDICATOR, +}; use tokio::sync::mpsc; use tokio::time::timeout; use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream}; @@ -105,6 +105,37 @@ impl RestClient { .await .map_err(|e| RelayError::Http(e.to_string())) } + + /// PUT a JSON body to an endpoint, returning the parsed response. + /// + /// Returns `Value::Null` for empty response bodies (e.g. 204 No Content). + pub async fn put_json(&self, path: &str, body: &Value) -> Result { + let url = format!("{}{}", self.base_url, path); + let builder = self.http.put(&url).json(body); + let builder = apply_auth(builder, &self.api_token, &self.keys); + + let resp = builder + .send() + .await + .map_err(|e| RelayError::Http(e.to_string()))?; + + if !resp.status().is_success() { + return Err(RelayError::Http(format!( + "PUT {} returned HTTP {}", + path, + resp.status() + ))); + } + + let text = resp + .text() + .await + .map_err(|e| RelayError::Http(e.to_string()))?; + if text.is_empty() { + return Ok(Value::Null); + } + serde_json::from_str(&text).map_err(|e| RelayError::Http(e.to_string())) + } } /// Events the harness cares about. @@ -200,6 +231,8 @@ enum RelayCommand { Shutdown, /// Subscribe to global membership notifications. SubscribeMembership, + /// Publish a signed event to the relay (for typing indicators, etc.). + PublishEvent { event: Box }, } // ── WebSocket stream type alias ─────────────────────────────────────────────── @@ -401,6 +434,25 @@ impl HarnessRelay { self.event_rx.recv().await.flatten() } + /// Publish a signed event to the relay via the background WebSocket task. + pub async fn publish_event(&self, event: Event) -> Result<(), RelayError> { + self.cmd_tx + .send(RelayCommand::PublishEvent { + event: Box::new(event), + }) + .await + .map_err(|_| RelayError::ConnectionClosed) + } + + /// Build a typing indicator event (kind:20002) for a channel. + pub fn build_typing_event(&self, channel_id: Uuid) -> Result { + let h_tag = Tag::parse(&["h", &channel_id.to_string()]) + .map_err(|e| RelayError::AuthFailed(e.to_string()))?; + let event = EventBuilder::new(Kind::Custom(KIND_TYPING_INDICATOR as u16), "", [h_tag]) + .sign_with_keys(&self.keys)?; + Ok(event) + } + /// Reconnect after connection loss. Instructs the background task to /// re-authenticate and resubscribe to all previously active channels. pub async fn reconnect(&mut self) -> Result<(), RelayError> { @@ -578,6 +630,12 @@ async fn run_background_task( let _ = send_membership_subscribe(&mut ws, &agent_pubkey_hex, None).await; state.membership_sub_active = true; } + RelayCommand::PublishEvent { event } => { + let msg = json!(["EVENT", event]); + if let Ok(text) = serde_json::to_string(&msg) { + let _ = ws.send(Message::Text(text.into())).await; + } + } } } } else { @@ -629,6 +687,14 @@ async fn run_background_task( ); } } + Some(RelayCommand::PublishEvent { event }) => { + let msg = json!(["EVENT", event]); + if let Ok(text) = serde_json::to_string(&msg) { + if let Err(e) = ws.send(Message::Text(text.into())).await { + warn!("failed to publish event: {e}"); + } + } + } Some(RelayCommand::Reconnect) => { // Reconnect command already consumed — skip the drain loop. wait_for_reconnect( @@ -949,6 +1015,8 @@ async fn wait_for_reconnect( Some(RelayCommand::SubscribeMembership) => { state.membership_sub_active = true; } + // Ephemeral events are meaningless while disconnected — drop silently. + Some(RelayCommand::PublishEvent { .. }) => {} } } }