From 661c004f5af81101df896e22063733a76b495c27 Mon Sep 17 00:00:00 2001 From: tlongwell-block <109685178+tlongwell-block@users.noreply.github.com> Date: Tue, 17 Mar 2026 11:44:00 -0400 Subject: [PATCH] =?UTF-8?q?feat(acp):=20Reaction-based=20lifecycle=20indic?= =?UTF-8?q?ators=20(=F0=9F=91=80=20seen=20=E2=86=92=20=F0=9F=92=AC=20worki?= =?UTF-8?q?ng=20=E2=86=92=20cleared)=20(#85)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/sprout-acp/src/main.rs | 35 ++++++- crates/sprout-acp/src/pool.rs | 186 +++++++++++++++++++++++++++++++++ crates/sprout-acp/src/queue.rs | 41 +++++--- crates/sprout-acp/src/relay.rs | 52 +++++++++ 4 files changed, 293 insertions(+), 21 deletions(-) diff --git a/crates/sprout-acp/src/main.rs b/crates/sprout-acp/src/main.rs index 64c6eadeb..4d9bf7704 100644 --- a/crates/sprout-acp/src/main.rs +++ b/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 => { diff --git a/crates/sprout-acp/src/pool.rs b/crates/sprout-acp/src/pool.rs index 05a3e01dc..ffe4c5915 100644 --- a/crates/sprout-acp/src/pool.rs +++ b/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 = 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, + ids: Vec, +} + +impl ReactionGuard { + fn new(rest: crate::relay::RestClient, ids: Vec) -> 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) { + 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"); + } } diff --git a/crates/sprout-acp/src/queue.rs b/crates/sprout-acp/src/queue.rs index 27916dcba..77bea242a 100644 --- a/crates/sprout-acp/src/queue.rs +++ b/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 { + 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 } } diff --git a/crates/sprout-acp/src/relay.rs b/crates/sprout-acp/src/relay.rs index c2aa0cc23..7ad29af39 100644 --- a/crates/sprout-acp/src/relay.rs +++ b/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 { + 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.