test(agent): await asynchronously collected recovery diagnostics

Wait on the harness stderr collector rather than treating the ACP stdout response as a drain barrier. Preserve budget and floor discrimination and corroborate exactly four provider requests. Exercise the same budget scenario with collection gated until after stdout, and prove absent diagnostics time out.

Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
Signed-off-by: Logan Johnson <loganj@squareup.com>

crates/buzz-agent/tests/common/mod.rs

@@ -19,7 +19,7 @@ use std::time::Duration;
 use serde_json::{json, Value};
 use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 use tokio::net::TcpListener;
-use tokio::sync::Mutex;
+use tokio::sync::{oneshot, Mutex, Notify};
 
 pub struct CapturingLlm {
     pub url: String,
@@ -107,11 +107,22 @@ pub struct Harness {
     stdin: tokio::process::ChildStdin,
     stdout: BufReader<tokio::process::ChildStdout>,
     stderr: Arc<StdMutex<String>>,
+    stderr_changed: Arc<Notify>,
     next_id: i64,
 }
 
 impl Harness {
     pub async fn spawn_with_env(base_url: &str, extra: &[(&str, &str)]) -> Self {
+        Self::spawn_with_stderr_gate(base_url, extra, None).await
+    }
+
+    /// Delay stderr collection until released, to exercise stdout/stderr ordering
+    /// without changing the child or relying on scheduler timing.
+    pub async fn spawn_with_stderr_gate(
+        base_url: &str,
+        extra: &[(&str, &str)],
+        stderr_gate: Option<oneshot::Receiver<()>>,
+    ) -> Self {
         let bin = env!("CARGO_BIN_EXE_buzz-agent");
         let mut cmd = tokio::process::Command::new(bin);
         cmd.env("BUZZ_AGENT_PROVIDER", "openai")
@@ -135,7 +146,14 @@ impl Harness {
         let stderr = child.stderr.take().unwrap();
         let stderr_buf = Arc::new(StdMutex::new(String::new()));
         let stderr_out = Arc::clone(&stderr_buf);
+        let stderr_changed = Arc::new(Notify::new());
+        let changed = Arc::clone(&stderr_changed);
         tokio::spawn(async move {
+            if let Some(gate) = stderr_gate {
+                // Dropping the sender (e.g. on assertion failure) also unblocks
+                // collection, rather than leaving a detached reader waiting.
+                let _ = gate.await;
+            }
             let mut reader = BufReader::new(stderr);
             let mut line = String::new();
             loop {
@@ -150,6 +168,7 @@ impl Harness {
                 if let Ok(mut out) = stderr_out.lock() {
                     out.push_str(&line);
                 }
+                changed.notify_waiters();
             }
         });
         Self {
@@ -157,6 +176,7 @@ impl Harness {
             stdin,
             stdout,
             stderr: stderr_buf,
+            stderr_changed,
             next_id: 1,
         }
     }
@@ -228,9 +248,36 @@ impl Harness {
         let _ = self.child.start_kill();
     }
 
+    /// Snapshot only: receiving a response on stdout does not drain stderr.
     pub fn stderr_text(&self) -> String {
         self.stderr.lock().map(|s| s.clone()).unwrap_or_default()
     }
+
+    /// Wait for a diagnostic in the independently collected stderr stream.
+    /// Returns the matching snapshot so subsequent assertions see its prefix.
+    pub async fn wait_for_stderr(&self, needle: &str, timeout: Duration) -> String {
+        tokio::time::timeout(timeout, async {
+            loop {
+                let changed = self.stderr_changed.notified();
+                tokio::pin!(changed);
+                // Register before inspecting the buffer: a line collected between
+                // the snapshot and await must not become a lost wakeup.
+                changed.as_mut().enable();
+                let stderr = self.stderr_text();
+                if stderr.contains(needle) {
+                    return stderr;
+                }
+                changed.await;
+            }
+        })
+        .await
+        .unwrap_or_else(|_| {
+            panic!(
+                "timed out waiting for stderr diagnostic {needle:?}; stderr={}",
+                self.stderr_text()
+            )
+        })
+    }
 }
 
 pub fn openai_text(content: &str) -> Value {

crates/buzz-agent/tests/regressions.rs

@@ -2706,6 +2706,30 @@ async fn ordinary_400_stays_terminal_and_triggers_no_recovery() {
 /// part of the assertion.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn context_recovery_budget_exhaustion_surfaces_the_error() {
+    assert_context_recovery_budget_exhaustion(false).await;
+}
+
+/// The same real provider/ACP scenario with stderr collection held until after
+/// the stdout response. The old immediate snapshot cannot observe the budget.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn context_recovery_budget_exhaustion_waits_for_delayed_stderr() {
+    assert_context_recovery_budget_exhaustion(true).await;
+}
+
+#[tokio::test]
+#[should_panic(expected = "timed out waiting for stderr diagnostic")]
+async fn stderr_diagnostic_wait_is_bounded_when_absent() {
+    let llm = spawn_capturing_llm(vec![]).await;
+    let h = Harness::spawn(&llm.url).await;
+    h.wait_for_stderr(
+        "diagnostic that is never emitted",
+        Duration::from_millis(20),
+    )
+    .await;
+}
+
+async fn assert_context_recovery_budget_exhaustion(delay_stderr: bool) {
+    let (release_stderr, stderr_gate) = tokio::sync::oneshot::channel();
     // Enough canned 400s that the queue is never the thing that stops the loop;
     // the fallback response is also a 400-shaped body under this helper only if
     // queued, so keep the queue generously long.
@@ -2713,7 +2737,7 @@ async fn context_recovery_budget_exhaustion_surfaces_the_error() {
         .map(|_| (400, openai_context_length_error()))
         .collect();
     let llm = spawn_capturing_llm_with_status(responses).await;
-    let mut h = Harness::spawn_with_env(
+    let mut h = Harness::spawn_with_stderr_gate(
         &llm.url,
         &[
             ("BUZZ_AGENT_MAX_CONTEXT_TOKENS", "200000"),
@@ -2723,6 +2747,7 @@ async fn context_recovery_budget_exhaustion_surfaces_the_error() {
             ),
             ("BUZZ_AGENT_MAX_HANDOFFS", "0"),
         ],
+        delay_stderr.then_some(stderr_gate),
     )
     .await;
     let sid = init_session(&mut h, json!([])).await;
@@ -2751,8 +2776,27 @@ async fn context_recovery_budget_exhaustion_surfaces_the_error() {
     // floor produce a surfaced error, so the assertion above passes either way
     // — and the floor can fire on the first rung without the budget ever being
     // consumed, which would make this test silently exercise a different
-    // mechanism than its name claims. Pin the budget explicitly.
+    // mechanism than its name claims. Pin the budget explicitly. Stdout is not
-    let stderr = h.stderr_text();
+    // a barrier for the independent stderr collector.
+    let stderr = {
+        let wait = h.wait_for_stderr("context recovery budget spent", Duration::from_secs(5));
+        tokio::pin!(wait);
+        if delay_stderr {
+            assert!(
+                !h.stderr_text().contains("context recovery budget spent"),
+                "the old immediate snapshot must miss the held diagnostic"
+            );
+            // Prove the actual wait stays pending before releasing the collector,
+            // without a sleep or depending on how quickly either task runs.
+            std::future::poll_fn(|cx| {
+                assert!(std::future::Future::poll(wait.as_mut(), cx).is_pending());
+                std::task::Poll::Ready(())
+            })
+            .await;
+            release_stderr.send(()).expect("release stderr collection");
+        }
+        wait.await
+    };
     assert!(
         stderr.contains("context recovery budget spent"),
         "the per-run recovery BUDGET must be what stops the loop here, not the prompt floor; \
@@ -2767,6 +2811,15 @@ async fn context_recovery_budget_exhaustion_surfaces_the_error() {
         "expected all 3 recovery rungs to be attempted before giving up, saw {rungs} — \
          stderr={stderr}"
     );
+    assert!(
+        !stderr.contains("context recovery would shrink"),
+        "the prompt floor must not stop this fixture: {stderr}"
+    );
+    assert_eq!(
+        llm.captured.lock().await.len(),
+        4,
+        "expected the rejected completion plus exactly three failed summaries"
+    );
     h.shutdown().await;
 }
 
@@ -2816,7 +2869,9 @@ async fn small_history_context_400_refuses_rescue_at_the_prompt_floor() {
         r0.get("error").is_some(),
         "a context 400 with no shrinkable history must surface the error, got: {r0}"
     );
-    let stderr = h.stderr_text();
+    let stderr = h
+        .wait_for_stderr("context recovery would shrink", Duration::from_secs(5))
+        .await;
     assert!(
         stderr.contains("below the") && stderr.contains("floor"),
         "the prompt-budget FLOOR must be what stops this, not the recovery budget; got: {stderr}"