test(serverless): prove agent reply + typing cue + cancel/redispatch e2e

Extend e2e_agent_responds to exercise the paths the live GUI hits:
- second @mention proves the OwnerInterrupt cancel/redispatch path (a
  follow-up message gets a fresh reply) — previously untested
- live listener asserts the typing indicator (kind 20002) fires during a
  turn so the 'agent is typing…' cue reaches relays in serverless
- stub now replies per-prompt (-1, -2) instead of once

Fixes found by the test:
- reaction working-cue timeout 500ms -> 4s: the pool does a real
  WebSocket publish + OK round-trip across public relays, too slow for
  500ms, so the 💬 cue never landed
- typing listener now watches ALL relays (agent publishes to whichever
  accepts; one relay may be 503-ing)

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)]

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(());
         }

crates/sprout-acp/tests/e2e_agent_responds.rs

@@ -124,6 +124,93 @@ async fn query(relay: &str, filter: serde_json::Value) -> Vec<nostr::Event> {
     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::<nostr::Event>(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<nostr::Event> = 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<nostr::Event> = 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<String> = 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");
 }

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 (`<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"}'
       ;;
     "")