feat(acp): Reaction-based lifecycle indicators (👀 seen → 💬 working → cleared) (#85)

crates/sprout-acp/src/main.rs

@@ -357,16 +357,31 @@ async fn main() -> Result<()> {
                                     // removed channel. Events already in-flight will
                                     // complete normally (the relay may reject actions if
                                     // the agent lost access).
-                                    let drained = queue.drain_channel(ch);
+                                    let drained_ids = queue.drain_channel(ch);
                                     let invalidated = pool.invalidate_channel_sessions(ch);
                                     // 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 {
+                                    // Best-effort: clean up 👀 on drained events.
+                                    // Note: the relay revokes membership before
+                                    // emitting the notification, so this DELETE may
+                                    // 403 on non-open channels. Stale 👀 in that
+                                    // case is a known limitation — fix belongs in
+                                    // the relay (clean up bot reactions on removal).
+                                    if !drained_ids.is_empty() {
+                                        let rc = ctx.rest_client.clone();
+                                        let ids = drained_ids.clone();
+                                        tokio::spawn(async move {
+                                            for eid in &ids {
+                                                pool::reaction_remove(&rc, eid, "👀").await;
+                                            }
+                                        });
+                                    }
+                                    if !drained_ids.is_empty() || invalidated > 0 {
                                         tracing::info!(
                                             channel_id = %ch,
-                                            drained,
+                                            drained = drained_ids.len(),
                                             invalidated,
                                             "cleaned up after membership removal"
                                         );
@@ -388,12 +403,24 @@ async fn main() -> Result<()> {
                                     continue;
                                 }
                             };
-                            queue.push(QueuedEvent {
+                            let event_id_hex = sprout_event.event.id.to_hex();
+                            let accepted = queue.push(QueuedEvent {
                                 channel_id: sprout_event.channel_id,
                                 event: sprout_event.event,
                                 received_at: std::time::Instant::now(),
                                 prompt_tag,
                             });
+                            // 👀 — immediate "seen" reaction, only if the event
+                            // was actually queued (not dropped by DedupMode::Drop).
+                            // Fire-and-forget: on rare fast-failure paths the
+                            // guard's cleanup may race with this add, leaving a
+                            // cosmetic stale 👀. Acceptable — see ReactionGuard docs.
+                            if accepted {
+                                let rc = ctx.rest_client.clone();
+                                tokio::spawn(async move {
+                                    pool::reaction_add(&rc, &event_id_hex, "👀").await;
+                                });
+                            }
                             typing_channels.extend(dispatch_pending(&mut pool, &mut queue, &ctx));
                         }
                         None => {

crates/sprout-acp/src/pool.rs

@@ -270,6 +270,16 @@ pub async fn run_prompt_task(
         None => PromptSource::Heartbeat,
     };
 
+    // ── Reaction cleanup guard ────────────────────────────────────────────
+    // Collects event IDs up front. On drop (any exit path — normal, early
+    // return, or panic), spawns best-effort cleanup of both 👀 and 💬.
+    // See `ReactionGuard` docs for ordering guarantees and known edge cases.
+    let reaction_ids: Vec<String> = batch
+        .as_ref()
+        .map(|b| b.events.iter().map(|be| be.event.id.to_hex()).collect())
+        .unwrap_or_default();
+    let _reaction_guard = ReactionGuard::new(ctx.rest_client.clone(), reaction_ids.clone());
+
     let (session_id, is_new_session) = match &source {
         PromptSource::Channel(cid) => {
             if let Some(sid) = agent.sessions.get(cid) {
@@ -483,6 +493,14 @@ pub async fn run_prompt_task(
         return;
     };
 
+    // 💬 — awaited inline so it completes before the prompt fires.
+    // This guarantees add-before-remove ordering for 💬: the guard's
+    // cleanup (spawned on drop) always runs after this returns.
+    // 👀 is fire-and-forget from main.rs (see race note in guard docs).
+    if !reaction_ids.is_empty() {
+        react_working(&ctx.rest_client, &reaction_ids).await;
+    }
+
     // ── Send the actual prompt ────────────────────────────────────────────
 
     let prompt_result = timeout(
@@ -585,6 +603,7 @@ pub async fn run_prompt_task(
             }
         }
     }
+    // _reaction_guard drops here → spawns clear_reactions for all exit paths.
 }
 
 // ── Context fetching ──────────────────────────────────────────────────────────
@@ -883,6 +902,136 @@ fn log_stop_reason(source: &PromptSource, stop_reason: &StopReason) {
     }
 }
 
+// ── Reaction indicators ───────────────────────────────────────────────────────
+//
+// Two-phase lifecycle visible to users:
+//   👀  "seen"    — event was queued and an agent will handle it
+//   💬  "working" — agent is actively prompting
+//
+// 💬 is awaited inline in `run_prompt_task` before the prompt fires, so
+// add-before-remove ordering is structural. 👀 is fire-and-forget from
+// `main.rs` at queue-push time for immediate responsiveness; on rare
+// fast-failure paths the guard's cleanup may race with the 👀 add,
+// leaving a cosmetic stale 👀 (see `ReactionGuard` docs).
+//
+// Cleanup is fire-and-forget via `ReactionGuard` (spawned on drop).
+// Failures are debug-logged and ignored — reactions are cosmetic.
+
+/// Drop guard that spawns reaction cleanup on any exit path.
+///
+/// Created at the top of `run_prompt_task`. On drop — normal return, early
+/// return, or panic — spawns fire-and-forget removal of both 👀 and 💬.
+///
+/// ## Ordering
+///
+/// 💬 (`react_working`) is awaited inline before the prompt fires, so it is
+/// guaranteed to precede cleanup. No race possible.
+///
+/// 👀 (`react_seen`) is fire-and-forget from `main.rs` at queue-push time.
+/// On rare fast-failure paths (e.g., `session_new` error on an idle agent),
+/// the cleanup spawn may race with the 👀 add, leaving a stale 👀. This is
+/// accepted as a cosmetic edge case — the message will be retried and the
+/// stale 👀 is harmless.
+struct ReactionGuard {
+    rest: Option<crate::relay::RestClient>,
+    ids: Vec<String>,
+}
+
+impl ReactionGuard {
+    fn new(rest: crate::relay::RestClient, ids: Vec<String>) -> Self {
+        Self {
+            rest: if ids.is_empty() { None } else { Some(rest) },
+            ids,
+        }
+    }
+}
+
+impl Drop for ReactionGuard {
+    fn drop(&mut self) {
+        // Safety: always called from within a tokio task (run_prompt_task is
+        // spawned via JoinSet::spawn), so a runtime context is guaranteed.
+        // During shutdown the spawned cleanup may be dropped before completing
+        // — acceptable for a cosmetic indicator.
+        if let Some(rest) = self.rest.take() {
+            let ids = std::mem::take(&mut self.ids);
+            tokio::spawn(clear_reactions(rest, ids));
+        }
+    }
+}
+
+const REACTION_SEEN: &str = "👀";
+const REACTION_WORKING: &str = "💬";
+
+/// Best-effort timeout for a single reaction REST call.
+const REACTION_TIMEOUT: Duration = Duration::from_millis(500);
+
+/// Percent-encode a string for use in a URL path segment.
+/// Emoji bytes are not URL-safe; event IDs (hex) pass through unchanged.
+fn pct_encode(s: &str) -> String {
+    let mut out = String::with_capacity(s.len() * 3);
+    for byte in s.bytes() {
+        match byte {
+            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
+                out.push(byte as char);
+            }
+            _ => {
+                use std::fmt::Write;
+                let _ = write!(out, "%{byte:02X}");
+            }
+        }
+    }
+    out
+}
+
+/// Best-effort: add a reaction. Returns immediately on timeout or error.
+pub(crate) async fn reaction_add(rest: &crate::relay::RestClient, event_id: &str, emoji: &str) {
+    let path = format!("/api/messages/{}/reactions", pct_encode(event_id));
+    let body = serde_json::json!({ "emoji": emoji });
+    match tokio::time::timeout(REACTION_TIMEOUT, rest.post_json(&path, &body)).await {
+        Ok(Ok(_)) => {}
+        Ok(Err(e)) => tracing::debug!(event_id, emoji, "reaction add failed: {e}"),
+        Err(_) => tracing::debug!(event_id, emoji, "reaction add timed out"),
+    }
+}
+
+/// Best-effort: remove a reaction. Returns immediately on timeout or error.
+pub(crate) async fn reaction_remove(rest: &crate::relay::RestClient, event_id: &str, emoji: &str) {
+    let path = format!(
+        "/api/messages/{}/reactions/{}",
+        pct_encode(event_id),
+        pct_encode(emoji),
+    );
+    match tokio::time::timeout(REACTION_TIMEOUT, rest.delete(&path)).await {
+        Ok(Ok(_)) => {}
+        Ok(Err(e)) => tracing::debug!(event_id, emoji, "reaction remove failed: {e}"),
+        Err(_) => tracing::debug!(event_id, emoji, "reaction remove timed out"),
+    }
+}
+
+/// Add 💬 to all events concurrently. Awaited inline before the prompt fires.
+/// Bounded by `REACTION_TIMEOUT` per call — worst case is a single 500ms wait
+/// regardless of batch size.
+async fn react_working(rest: &crate::relay::RestClient, event_ids: &[String]) {
+    futures_util::future::join_all(
+        event_ids
+            .iter()
+            .map(|eid| reaction_add(rest, eid, REACTION_WORKING)),
+    )
+    .await;
+}
+
+/// Fire-and-forget: remove both 👀 and 💬 from all events. Spawned on turn complete.
+/// All removals run concurrently — bounded by `REACTION_TIMEOUT` per call.
+async fn clear_reactions(rest: crate::relay::RestClient, event_ids: Vec<String>) {
+    futures_util::future::join_all(event_ids.iter().flat_map(|eid| {
+        [
+            reaction_remove(&rest, eid, REACTION_SEEN),
+            reaction_remove(&rest, eid, REACTION_WORKING),
+        ]
+    }))
+    .await;
+}
+
 // ─── Unit Tests ──────────────────────────────────────────────────────────────
 
 #[cfg(test)]
@@ -1131,4 +1280,41 @@ mod tests {
         let msg = json_to_context_message(&obj).expect("should parse");
         assert_eq!(msg.pubkey, "unknown");
     }
+
+    // ── pct_encode tests ─────────────────────────────────────────────────
+
+    #[test]
+    fn test_pct_encode_hex_passthrough() {
+        let hex = "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2";
+        assert_eq!(pct_encode(hex), hex);
+    }
+
+    #[test]
+    fn test_pct_encode_emoji() {
+        // 👀 = U+1F440 = F0 9F 91 80 in UTF-8
+        assert_eq!(pct_encode("👀"), "%F0%9F%91%80");
+    }
+
+    #[test]
+    fn test_pct_encode_emoji_speech_balloon() {
+        // 💬 = U+1F4AC = F0 9F 92 AC in UTF-8
+        assert_eq!(pct_encode("💬"), "%F0%9F%92%AC");
+    }
+
+    #[test]
+    fn test_pct_encode_empty() {
+        assert_eq!(pct_encode(""), "");
+    }
+
+    #[test]
+    fn test_pct_encode_unreserved_passthrough() {
+        assert_eq!(pct_encode("AZaz09-_.~"), "AZaz09-_.~");
+    }
+
+    #[test]
+    fn test_pct_encode_reserved_chars() {
+        assert_eq!(pct_encode("/"), "%2F");
+        assert_eq!(pct_encode("+"), "%2B");
+        assert_eq!(pct_encode(" "), "%20");
+    }
 }

crates/sprout-acp/src/queue.rs

@@ -103,7 +103,9 @@ impl EventQueue {
     ///
     /// In [`DedupMode::Drop`], events for any currently in-flight channel are
     /// silently discarded (debug-logged).
-    pub fn push(&mut self, event: QueuedEvent) {
+    ///
+    /// Returns `true` if the event was accepted, `false` if dropped.
+    pub fn push(&mut self, event: QueuedEvent) -> bool {
         if matches!(self.dedup_mode, DedupMode::Drop)
             && self.in_flight_channels.contains(&event.channel_id)
         {
@@ -111,12 +113,13 @@ impl EventQueue {
                 channel_id = %event.channel_id,
                 "dropping event for in-flight channel (drop mode)"
             );
-            return;
+            return false;
         }
         self.queues
             .entry(event.channel_id)
             .or_default()
             .push_back(event);
+        true
     }
 
     /// Try to flush the next batch.
@@ -252,14 +255,18 @@ impl EventQueue {
     /// Also clears any `retry_after` throttle for the channel.
     ///
     /// Returns the number of events dropped.
-    pub fn drain_channel(&mut self, channel_id: Uuid) -> usize {
-        let dropped = self
+    /// Drop all queued (non-in-flight) events for a channel.
+    ///
+    /// Returns the event IDs of dropped events so the caller can clean up
+    /// any reactions (👀) that were added at queue-push time.
+    pub fn drain_channel(&mut self, channel_id: Uuid) -> Vec<String> {
+        let ids = self
             .queues
             .remove(&channel_id)
-            .map(|q| q.len())
-            .unwrap_or(0);
+            .map(|q| q.into_iter().map(|e| e.event.id.to_hex()).collect())
+            .unwrap_or_default();
         self.retry_after.remove(&channel_id);
-        dropped
+        ids
     }
 
     /// Whether a prompt is currently in-flight for the given channel.
@@ -1656,8 +1663,8 @@ mod tests {
         q.push(make_queued(ch, "msg2"));
         assert_eq!(q.pending_count(), 2);
 
-        let dropped = q.drain_channel(ch);
-        assert_eq!(dropped, 2);
+        let drained = q.drain_channel(ch);
+        assert_eq!(drained.len(), 2);
         assert_eq!(q.pending_count(), 0);
     }
 
@@ -1670,8 +1677,8 @@ mod tests {
         q.push(make_queued(ch_a, "A"));
         q.push(make_queued(ch_b, "B"));
 
-        let dropped = q.drain_channel(ch_a);
-        assert_eq!(dropped, 1);
+        let drained = q.drain_channel(ch_a);
+        assert_eq!(drained.len(), 1);
         assert_eq!(q.pending_count(), 1); // ch_b still has 1
     }
 
@@ -1687,16 +1694,16 @@ mod tests {
 
         // Channel is throttled — verify drain clears it.
         assert!(!q.has_flushable_work());
-        let dropped = q.drain_channel(ch);
-        assert_eq!(dropped, 1);
+        let drained = q.drain_channel(ch);
+        assert_eq!(drained.len(), 1);
         assert_eq!(q.pending_count(), 0);
     }
 
     #[test]
-    fn test_drain_channel_empty_returns_zero() {
+    fn test_drain_channel_empty_returns_empty() {
         let mut q = EventQueue::new(DedupMode::Queue);
         let ch = Uuid::new_v4();
-        assert_eq!(q.drain_channel(ch), 0);
+        assert!(q.drain_channel(ch).is_empty());
     }
 
     #[test]
@@ -1712,8 +1719,8 @@ mod tests {
         q.push(make_queued(ch, "msg2"));
 
         // drain_channel should only remove the queued event, not the in-flight one.
-        let dropped = q.drain_channel(ch);
-        assert_eq!(dropped, 1);
+        let drained = q.drain_channel(ch);
+        assert_eq!(drained.len(), 1);
         assert!(q.is_in_flight()); // in-flight unaffected
     }
 }

crates/sprout-acp/src/relay.rs

@@ -136,6 +136,58 @@ impl RestClient {
         }
         serde_json::from_str(&text).map_err(|e| RelayError::Http(e.to_string()))
     }
+
+    /// POST 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 post_json(&self, path: &str, body: &Value) -> Result<Value, RelayError> {
+        let url = format!("{}{}", self.base_url, path);
+        let builder = self.http.post(&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!(
+                "POST {} 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()))
+    }
+
+    /// DELETE an endpoint. Returns `Ok(())` on 2xx.
+    pub async fn delete(&self, path: &str) -> Result<(), RelayError> {
+        let url = format!("{}{}", self.base_url, path);
+        let builder = self.http.delete(&url);
+        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!(
+                "DELETE {} returned HTTP {}",
+                path,
+                resp.status()
+            )));
+        }
+        Ok(())
+    }
 }
 
 /// Events the harness cares about.