fix(desktop): derive agent availability from relay presence

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,13 +168,15 @@ impl Harness {
                 if let Ok(mut out) = stderr_out.lock() {
                     out.push_str(&line);
                 }
+                changed.notify_waiters();
             }
         });
         Self {
             child,
             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,14 +2706,38 @@ 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.
     let responses: Vec<(u16, Value)> = (0..40)
         .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.
-    let stderr = h.stderr_text();
+    // mechanism than its name claims. Pin the budget explicitly. Stdout is not
+    // 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}"

desktop/playwright.config.ts

@@ -177,6 +177,7 @@ export default defineConfig({
       name: "integration",
       testMatch: [
         "**/agents.spec.ts",
+        "**/agent-availability.spec.ts",
         "**/agent-snapshot-recipient.spec.ts",
         "**/onboarding.spec.ts",
         "**/stream.spec.ts",

desktop/src-tauri/src/commands/profile.rs

@@ -344,17 +344,16 @@ pub async fn get_presence(
     }
 
     // Presence is published as kind:20001 ephemeral events. Query the most
-    // recent per author. Some relays don't retain ephemeral events — we
-    // best-effort and return what we get.
+    // recent per author. Only a successful empty snapshot establishes absence;
+    // transport/auth/storage failures must reject so consumers remain unknown.
     let events = query_relay(
         &state,
         &[serde_json::json!({
             "kinds": [20001],
             "authors": pubkeys,
         })],
     )
-    .await
-    .unwrap_or_default();
+    .await?;
 
     let mut latest: HashMap<String, (u64, PresenceStatus)> = HashMap::new();
     for ev in &events {
@@ -482,3 +481,7 @@ mod tests {
         assert_eq!(filter["page"], serde_json::json!(1));
     }
 }
+
+#[cfg(test)]
+#[path = "profile_presence_tests.rs"]
+mod presence_tests;

desktop/src-tauri/src/commands/profile_presence_tests.rs

@@ -0,0 +1,103 @@
+//! Drive the actual get_presence command through its authenticated HTTP query.
+//! In particular, an error must not become a successful empty IPC snapshot.
+use super::get_presence;
+use crate::app_state::build_app_state;
+use crate::relay_admission::{reset_rate_limit_gate, TEST_SERIAL};
+use tauri::Manager;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+#[tokio::test]
+async fn presence_command_preserves_query_failure_and_successful_absence() {
+    let _serial = TEST_SERIAL.lock().await;
+    reset_rate_limit_gate();
+    for (status, body) in [
+        ("200 OK", "[]"),
+        ("401 Unauthorized", r#"{"error":"unauthorized"}"#),
+        ("429 Too Many Requests", r#"{"error":"retry in 1s"}"#),
+        (
+            "500 Internal Server Error",
+            r#"{"error":"storage unavailable"}"#,
+        ),
+        ("200 OK", "not json"),
+    ] {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let server = tokio::spawn(async move {
+            let (mut stream, _) = listener.accept().await.unwrap();
+            let mut request = Vec::new();
+            loop {
+                let mut buf = [0; 4096];
+                let count = stream.read(&mut buf).await.unwrap();
+                assert!(count > 0);
+                request.extend_from_slice(&buf[..count]);
+                assert!(request.len() < 16384);
+                if let Some(end) = request.windows(4).position(|w| w == b"\r\n\r\n") {
+                    let headers = String::from_utf8_lossy(&request[..end]).to_lowercase();
+                    let length: usize = headers
+                        .lines()
+                        .find_map(|line| {
+                            line.strip_prefix("content-length:")
+                                .map(|v| v.trim().parse().unwrap())
+                        })
+                        .unwrap();
+                    if request.len() >= end + 4 + length {
+                        break;
+                    }
+                }
+            }
+            let request = String::from_utf8(request).unwrap();
+            assert!(request.starts_with("POST /query "));
+            assert!(request.to_lowercase().contains("authorization: nostr "));
+            assert!(request.contains("20001"));
+            let response = format!("HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len());
+            stream.write_all(response.as_bytes()).await.unwrap();
+        });
+        let state = build_app_state();
+        *state.relay_url_override.lock().unwrap() = Some(format!("ws://{addr}"));
+        let app = tauri::test::mock_builder()
+            .manage(state)
+            .build(tauri::test::mock_context(tauri::test::noop_assets()))
+            .unwrap();
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            get_presence(vec!["a".repeat(64)], app.state()),
+        )
+        .await
+        .unwrap();
+        tokio::time::timeout(std::time::Duration::from_secs(5), server)
+            .await
+            .unwrap()
+            .unwrap();
+        if status == "200 OK" && body == "[]" {
+            assert_eq!(
+                serde_json::to_value(result.unwrap()).unwrap(),
+                serde_json::json!({})
+            );
+        } else {
+            assert!(
+                result.is_err(),
+                "{status} / {body} must reject, not return Offline: {result:?}"
+            );
+        }
+        reset_rate_limit_gate();
+    }
+}
+
+#[tokio::test]
+async fn presence_command_transport_failure_is_not_offline() {
+    let _serial = TEST_SERIAL.lock().await;
+    reset_rate_limit_gate();
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    drop(listener);
+    let state = build_app_state();
+    *state.relay_url_override.lock().unwrap() = Some(format!("ws://{addr}"));
+    let app = tauri::test::mock_builder()
+        .manage(state)
+        .build(tauri::test::mock_context(tauri::test::noop_assets()))
+        .unwrap();
+    let result = get_presence(vec!["a".repeat(64)], app.state()).await;
+    assert!(result.is_err(), "transport failure must reject: {result:?}");
+    // Empty input does not require a relay and remains a genuine empty result.
+    assert!(get_presence(vec![], app.state()).await.unwrap().is_empty());
+}

desktop/src/features/agents/AGENTS.md

@@ -205,6 +205,17 @@ with a TypeScript lookup table or an id comparison in a component.
    select a representative or offer persona Start; a relay persona link cannot
    borrow a local sibling's management controls. See
    [the identity contract](../../../../docs/agent-profile-identity.md).
+   Availability dots read relay presence, never a saved deployment
+   receipt or runtime status. Failed/disconnected reads are unknown; lifecycle
+   actions retain their separate routing. Current exact-key Online/Away presence
+   suppresses Start for an inactive local record without granting Stop authority;
+   list/profile/member startup guards must not interpret Offline as proof of safe
+   startup. Deletion also consumes that same exact-key availability reader:
+   unknown requests shutdown when a channel exists, request failure retains the
+   record, and only established Offline keeps the intentional no-request path.
+   Unqueried persona siblings are unknown. No presence state grants deletion or
+   Stop authority; native local stop-before-remove remains independent. See
+   [the availability contract](../../../../docs/agent-availability.md).
 14. **Thinking effort has two surfaces: a local-only WRITE control and a
    read-only two-facts DISPLAY.** The write control is `EffortPickerField`
    (`ui/EffortPickerField.tsx`), a self-contained section component mounted in