feat(sprout-acp): auto-presence and typing indicators (#84)

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

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<Uuid> = HashSet::new();
+    let mut presence_task: Option<tokio::task::JoinHandle<()>> = 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<PromptContext>) {
-    let mut dispatched: usize = 0;
+fn dispatch_pending(
+    pool: &mut AgentPool,
+    queue: &mut EventQueue,
+    ctx: &Arc<PromptContext>,
+) -> Vec<Uuid> {
+    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<Prom
                 recoverable_batch,
             },
         );
-        dispatched += 1;
+        dispatched_channels.push(channel_id);
     }
     tracing::debug!(
-        dispatched,
+        dispatched = dispatched_channels.len(),
         queue_depth = queue.pending_channels(),
         "dispatch_pending"
     );
+    dispatched_channels
 }
 
 // ── handle_prompt_result ──────────────────────────────────────────────────────
@@ -621,6 +725,7 @@ async fn recover_panicked_agent(
     join_error: tokio::task::JoinError,
     heartbeat_in_flight: &mut bool,
     removed_channels: &HashSet<Uuid>,
+    typing_channels: &mut HashSet<Uuid>,
 ) {
     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<Uuid>,
+    typing_channels: &mut HashSet<Uuid>,
 ) -> 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 {

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<Value, RelayError> {
+        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<Event> },
 }
 
 // ── 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<Event, RelayError> {
+        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 { .. }) => {}
             }
         }
     }