diff --git a/crates/sprout-acp/src/pool.rs b/crates/sprout-acp/src/pool.rs index 09313f620..64bb6404f 100644 --- a/crates/sprout-acp/src/pool.rs +++ b/crates/sprout-acp/src/pool.rs @@ -1903,7 +1903,11 @@ 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); +// Reactions ("👀"/"💬" working cues) are best-effort. 500ms was tuned for the +// local HTTP bridge; the serverless pool does a real WebSocket publish + OK +// round-trip across public relays, which needs more headroom or the cue never +// lands. 4s covers a flaky-relay round-trip while staying well under a turn. +const REACTION_TIMEOUT: Duration = Duration::from_secs(4); /// Percent-encode a string for use in a URL path segment (used in tests only). #[cfg(test)] diff --git a/crates/sprout-acp/src/relay.rs b/crates/sprout-acp/src/relay.rs index b7326bb56..bfffeb933 100644 --- a/crates/sprout-acp/src/relay.rs +++ b/crates/sprout-acp/src/relay.rs @@ -1273,7 +1273,9 @@ impl HarnessRelay { // block; dropping on error is acceptable for typing indicators. let pool = backend.pool(); tokio::spawn(async move { - let _ = pool.send_event(&event).await; + if let Err(e) = pool.send_event(&event).await { + tracing::debug!("serverless ephemeral publish failed: {e}"); + } }); return Ok(()); } diff --git a/crates/sprout-acp/tests/e2e_agent_responds.rs b/crates/sprout-acp/tests/e2e_agent_responds.rs index dc6451114..70573691e 100644 --- a/crates/sprout-acp/tests/e2e_agent_responds.rs +++ b/crates/sprout-acp/tests/e2e_agent_responds.rs @@ -124,6 +124,93 @@ async fn query(relay: &str, filter: serde_json::Value) -> Vec { out } +/// Open a LIVE subscription (no `until`, `since=now`) on one relay and watch for +/// an event matching `pred` for up to `secs`. Used for ephemeral events (typing +/// indicators, kind 20002) that relays do NOT store, so a normal `query` after +/// the fact returns nothing — they must be caught live as they stream. +async fn listen_live( + relay: &str, + filter: serde_json::Value, + secs: u64, + pred: impl Fn(&nostr::Event) -> bool, +) -> bool { + let ws = match connect_async(relay).await { + Ok((ws, _)) => ws, + Err(e) => { + eprintln!(" (listen connect to {relay} failed: {e})"); + return false; + } + }; + let (mut write, mut read) = ws.split(); + let sub = "live1"; + let req = serde_json::json!(["REQ", sub, filter]).to_string(); + if write.send(Message::Text(req.into())).await.is_err() { + return false; + } + let found = tokio::time::timeout(Duration::from_secs(secs), async { + while let Some(Ok(m)) = read.next().await { + if let Message::Text(t) = m { + let v: serde_json::Value = match serde_json::from_str(&t) { + Ok(v) => v, + Err(_) => continue, + }; + let arr = v.as_array().cloned().unwrap_or_default(); + if arr.first().and_then(|x| x.as_str()) == Some("EVENT") + && arr.get(1).and_then(|x| x.as_str()) == Some(sub) + { + if let Some(ev) = arr.get(2) { + if let Ok(e) = serde_json::from_value::(ev.clone()) { + if pred(&e) { + return true; + } + } + } + } + } + } + false + }) + .await + .unwrap_or(false); + let _ = write.close().await; + found +} + +/// Poll all relays (merged) for a kind-9 reply from `agent_pk` whose content +/// contains `marker`. Returns true if found within `attempts` × 3s. +async fn await_reply( + relay_list: &str, + channel: &str, + agent_pk: nostr::PublicKey, + marker: &str, + attempts: u32, +) -> bool { + for attempt in 0..attempts { + tokio::time::sleep(Duration::from_secs(3)).await; + let mut events: Vec = Vec::new(); + for r in relay_list.split(',') { + let mut got = query( + r.trim(), + serde_json::json!({"kinds":[9],"#h":[channel],"limit":50}), + ) + .await; + events.append(&mut got); + } + if events + .iter() + .any(|e| e.content.contains(marker) && e.pubkey == agent_pk) + { + eprintln!("✅ reply '{marker}' found after {}s", (attempt + 1) * 3); + return true; + } + eprintln!( + " …waiting for '{marker}': attempt {attempt}, {} msgs", + events.len() + ); + } + false +} + #[tokio::test] #[ignore = "network: hits live public relays; spawns sprout-acp + sprout binaries"] async fn agent_responds_in_channel_e2e() { @@ -206,6 +293,9 @@ async fn agent_responds_in_channel_e2e() { .env("STUB_AGENT_REPLY", &reply_marker) .env("STUB_AGENT_SPROUT_BIN", &sprout_bin) .env("STUB_AGENT_LOG", &log_path) + // Hold each turn for 5s so the harness typing-indicator loop (3s tick) + // fires while a turn is in flight — proves the "agent is typing…" cue. + .env("STUB_AGENT_PROMPT_DELAY", "5") .env("RUST_LOG", "sprout_acp=debug") .stdout(Stdio::from(harness_log_out)) .stderr(Stdio::from(harness_log)) @@ -216,53 +306,127 @@ async fn agent_responds_in_channel_e2e() { // Give the harness time to connect to all relays + discover the channel. tokio::time::sleep(Duration::from_secs(8)).await; - // 3. Publish a message into the channel (the human talking to the agent). - let msg = EventBuilder::new(Kind::Custom(9), "@agent hello, please reply") - .tags(vec![ - nostr::Tag::parse(["h", &channel]).unwrap(), - nostr::Tag::parse(["p", &agent_pk]).unwrap(), - ]) - .sign_with_keys(&human) - .unwrap(); - let msg_ok = publish_all(&relay_list, &msg).await; - assert!(msg_ok, "no relay accepted the @mention message"); - eprintln!("published @mention message; waiting for agent reply…"); - - // 4. Poll the relays for the agent's reply (kind 9 from agent, content - // marker). Query EVERY relay and merge — the reply may land on only one - // relay (the agent publishes to whichever accepts first), and a single - // relay may be 503-ing, so polling just one relay can miss it entirely. - let mut found = false; - for attempt in 0..20 { - tokio::time::sleep(Duration::from_secs(3)).await; - let mut events: Vec = Vec::new(); - for r in relay_list.split(',') { - let mut got = query( - r.trim(), - serde_json::json!({"kinds":[9],"#h":[channel],"limit":50}), - ) - .await; - events.append(&mut got); + // Helper: build + publish a kind-9 @mention from the human. + let send_mention = |body: &str| { + let body = body.to_string(); + let channel = channel.clone(); + let agent_pk = agent_pk.clone(); + let human = human.clone(); + let relay_list = relay_list.clone(); + async move { + let msg = EventBuilder::new(Kind::Custom(9), body) + .tags(vec![ + nostr::Tag::parse(["h", &channel]).unwrap(), + nostr::Tag::parse(["p", &agent_pk]).unwrap(), + ]) + .sign_with_keys(&human) + .unwrap(); + publish_all(&relay_list, &msg).await } - if events - .iter() - .any(|e| e.content.contains(&reply_marker) && e.pubkey == agent.public_key()) - { - found = true; - eprintln!("✅ agent reply found on relay after {}s", (attempt + 1) * 3); - break; - } - eprintln!(" …attempt {attempt}: {} msgs, no reply yet", events.len()); - } + }; - let _ = child.kill().await; - if !found { + let dump_logs = || { if let Ok(h) = std::fs::read_to_string(&harness_log_path) { eprintln!("--- harness (sprout-acp) log ---\n{h}\n-----------------------------------"); } if let Ok(log) = std::fs::read_to_string(&log_path) { eprintln!("--- stub agent log ---\n{log}\n----------------------"); } - panic!("agent never posted a reply to the channel (see logs above)"); + }; + + // ── Message 1: prove the agent receives + replies, AND emits a typing + // indicator (kind 20002) while the turn is in flight. Typing events are + // ephemeral, so we listen live (concurrently) as we send the message. The + // agent publishes to whichever relay accepts (one may be 503-ing), so we + // must listen on ALL relays — listening on just one can miss it entirely. ── + let typing_channel = channel.clone(); + let typing_agent = agent.public_key(); + let typing_relays: Vec = relay_list + .split(',') + .map(|s| s.trim().to_string()) + .collect(); + let typing_fut = tokio::spawn(async move { + let mut handles = Vec::new(); + for relay in typing_relays { + let ch = typing_channel.clone(); + handles.push(tokio::spawn(async move { + listen_live( + &relay, + serde_json::json!({ + "kinds": [20002], + "#h": [ch], + "since": (chrono::Utc::now().timestamp() - 2), + }), + 30, + move |e| e.pubkey == typing_agent, + ) + .await + })); + } + // True if ANY relay's listener caught the typing indicator. + let mut seen = false; + for h in handles { + if h.await.unwrap_or(false) { + seen = true; + } + } + seen + }); + + assert!( + send_mention("@agent hello, please reply").await, + "no relay accepted the first @mention" + ); + eprintln!("published @mention #1; waiting for reply-1…"); + + let reply1 = await_reply( + &relay_list, + &channel, + agent.public_key(), + &format!("{reply_marker}-1"), + 20, + ) + .await; + if !reply1 { + let _ = child.kill().await; + dump_logs(); + panic!("agent never posted reply #1 (see logs above)"); } + + // With a 5s turn delay, the harness's 3s typing tick MUST fire while the + // turn is in flight. This proves the "agent is typing…" cue reaches the + // relays in serverless (it's published via the nostr-relay-pool backend). + let typing_seen = typing_fut.await.unwrap_or(false); + if !typing_seen { + let _ = child.kill().await; + dump_logs(); + panic!("no typing indicator (kind 20002) observed during a 5s turn — the 'agent is typing…' cue is broken in serverless"); + } + eprintln!("✅ typing indicator (kind 20002) observed during turn"); + + // ── Message 2: prove the cancel/redispatch path the live GUI hits. A + // second @mention from the owner triggers OwnerInterrupt (cancel any + // in-flight turn) then a fresh dispatch. The stub replies per-prompt, so a + // distinct `-2` reply proves the new turn actually ran. ── + assert!( + send_mention("@agent and one more thing").await, + "no relay accepted the second @mention" + ); + eprintln!("published @mention #2; waiting for reply-2 (cancel/redispatch path)…"); + + let reply2 = await_reply( + &relay_list, + &channel, + agent.public_key(), + &format!("{reply_marker}-2"), + 20, + ) + .await; + + let _ = child.kill().await; + if !reply2 { + dump_logs(); + panic!("agent never posted reply #2 — cancel/redispatch path is broken (see logs above)"); + } + eprintln!("✅ both replies landed; cancel/redispatch path works"); } diff --git a/crates/sprout-acp/tests/stub_agent.sh b/crates/sprout-acp/tests/stub_agent.sh index 2075223b3..9a29da870 100755 --- a/crates/sprout-acp/tests/stub_agent.sh +++ b/crates/sprout-acp/tests/stub_agent.sh @@ -16,7 +16,10 @@ set -euo pipefail -reply_sent=0 +# Reply to EVERY prompt with a distinct marker (`-1`, `-2`, …) so the +# test can prove the harness dispatches a fresh turn per message — including the +# cancel/redispatch path when a second message arrives (OwnerInterrupt). +prompt_count=0 send_response() { # $1 = id, $2 = result json @@ -36,15 +39,19 @@ while IFS= read -r line; do send_response "$id" '{"sessionId":"stub-session-1"}' ;; session/prompt) - if [ "$reply_sent" -eq 0 ]; then - reply_sent=1 - # Post the reply to the channel using the sprout CLI — the real reply - # mechanism per base_prompt.md. Auth env is inherited from the harness. - "${STUB_AGENT_SPROUT_BIN}" messages send \ - --channel "${STUB_AGENT_CHANNEL}" \ - --content "${STUB_AGENT_REPLY}" >/dev/null 2>>"${STUB_AGENT_LOG:-/dev/stderr}" || \ - echo "stub: sprout messages send failed" >>"${STUB_AGENT_LOG:-/dev/stderr}" + prompt_count=$((prompt_count + 1)) + # Optional delay BEFORE replying so the turn stays in-flight long enough + # for the harness typing-indicator loop (3s tick) to fire — proves the + # "agent is typing…" cue works in serverless. + if [ -n "${STUB_AGENT_PROMPT_DELAY:-}" ]; then + sleep "${STUB_AGENT_PROMPT_DELAY}" fi + # Post a distinct reply per prompt using the sprout CLI — the real reply + # mechanism per base_prompt.md. Auth env is inherited from the harness. + "${STUB_AGENT_SPROUT_BIN}" messages send \ + --channel "${STUB_AGENT_CHANNEL}" \ + --content "${STUB_AGENT_REPLY}-${prompt_count}" >/dev/null 2>>"${STUB_AGENT_LOG:-/dev/stderr}" || \ + echo "stub: sprout messages send failed" >>"${STUB_AGENT_LOG:-/dev/stderr}" send_response "$id" '{"stopReason":"end_turn"}' ;; "")