External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(presence): clear on disconnect, fix heartbeat/TTL, drop broken REST path (#877)

Browse Source
This commit is contained in:
Will Pfleger
2026-06-05 13:28:00 -07:00
committed by GitHub
parent ef98ae942a
commit 5d7c748969
10 changed files with 36 additions and 340 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/sprout-relay/src/connection.rs
+8
View File
@@ -192,6 +192,14 @@ pub async fn handle_connection(socket: WebSocket, state: Arc<AppState>, addr: So
state.sub_registry.remove_connection(conn.conn_id);
state.conn_manager.deregister(conn.conn_id);
if let AuthState::Authenticated(ref auth_ctx) = *conn.auth_state.read().await {
let remaining = state
.conn_manager
.connection_ids_for_pubkey(auth_ctx.pubkey.to_bytes().as_slice());
if remaining.is_empty() {
let _ = state.pubsub.clear_presence(&auth_ctx.pubkey).await;
}
}
metrics::gauge!("sprout_ws_connections_active").decrement(1.0);
info!(conn_id = %conn_id, addr = %addr, "WebSocket connection closed");
desktop/src-tauri/src/commands/profile.rs
+1 -23
View File
@@ -7,10 +7,7 @@ use tauri::State;
use crate::{
app_state::AppState,
events,
models::{
ProfileInfo, SearchUsersResponse, SetPresenceResponse, UserNotesResponse,
UsersBatchResponse,
},
models::{ProfileInfo, SearchUsersResponse, UserNotesResponse, UsersBatchResponse},
nostr_convert,
relay::{query_relay, submit_event},
};
@@ -270,25 +267,6 @@ pub async fn get_presence(
.collect())
}
#[tauri::command]
pub async fn set_presence(
status: PresenceStatus,
state: State<'_, AppState>,
) -> Result<SetPresenceResponse, String> {
let status_str = match status {
PresenceStatus::Online => "online",
PresenceStatus::Away => "away",
PresenceStatus::Offline => "offline",
};
let builder = events::build_presence(status_str)?;
submit_event(builder, &state).await?;
Ok(SetPresenceResponse {
status,
ttl_seconds: 60,
})
}
fn current_pubkey_hex(state: &AppState) -> Result<String, String> {
let keys = state.keys.lock().map_err(|e| e.to_string())?;
Ok(keys.public_key().to_hex())
desktop/src-tauri/src/events.rs
-9
View File
@@ -690,15 +690,6 @@ pub fn build_dm_hide(channel_id: &str) -> Result<EventBuilder, String> {
Ok(EventBuilder::new(Kind::Custom(41012), "").tags(tags))
}
/// Kind 20001 — ephemeral presence broadcast (`online` / `away` / `offline`).
pub fn build_presence(status: &str) -> Result<EventBuilder, String> {
match status {
"online" | "away" | "offline" => {}
other => return Err(format!("invalid presence status: {other}")),
};
Ok(EventBuilder::new(Kind::Custom(20001), status.to_string()))
}
/// Kind 30620 — replaceable workflow definition.
///
/// The `d` tag carries the workflow id; `h` tag carries the channel id; the
desktop/src-tauri/src/lib.rs
-2
View File
@@ -482,7 +482,6 @@ pub fn run() {
// this worktree's data directory. Must run before
// restore_managed_agents_on_launch (which reads managed-agents.json).
migration::sync_shared_agent_data(&app_handle);
migration::reconcile_provider_mcp_commands(&app_handle);
migration::reconcile_persona_pack_paths(&app_handle);
// Resolve persisted identity key (env var → file → generate+save).
@@ -585,7 +584,6 @@ pub fn run() {
get_user_notes,
search_users,
get_presence,
set_presence,
get_default_relay_url,
is_shared_identity,
get_relay_ws_url,
desktop/src-tauri/src/migration.rs
-208
View File
@@ -6,11 +6,6 @@
//! `SPROUT_SHARE_IDENTITY=1` and `SPROUT_PRIVATE_KEY` is set. All dev
//! instances share the same physical files — edits in any worktree are
//! immediately visible to all others.
//!
//! **Provider reconciliation** (`reconcile_provider_mcp_commands`): Per-launch
//! fix-up of `mcp_command` values in `managed-agents.json` against the
//! discovery table. Ensures known providers always have their canonical
//! `mcp_command`; unknown/custom agents are left untouched.
use std::path::{Path, PathBuf};
use tauri::Manager;
@@ -264,54 +259,6 @@ pub fn sync_shared_agent_data(app: &tauri::AppHandle) {
}
}
fn reconcile_mcp_commands_in_file(path: &Path) {
patch_json_records(path, |obj| {
let agent_command = match obj.get("agent_command").and_then(|v| v.as_str()) {
Some(cmd) => cmd.to_string(),
None => return false,
};
let Some(provider) = crate::managed_agents::known_acp_provider(&agent_command) else {
return false;
};
let expected = provider.mcp_command.unwrap_or("");
let current = obj
.get("mcp_command")
.and_then(|v| v.as_str())
.unwrap_or("");
// Only clear the known stale default — never touch user-customized values.
if current == "sprout-mcp-server" {
eprintln!(
"sprout-desktop: provider-reconcile: {:?} ({:?}): mcp_command {:?} → {:?}",
obj.get("name").and_then(|v| v.as_str()).unwrap_or("?"),
agent_command,
current,
expected,
);
obj.insert(
"mcp_command".to_string(),
serde_json::Value::String(expected.to_string()),
);
true
} else {
false
}
});
}
/// Reconcile `mcp_command` values in managed-agents.json against the
/// discovery table. Known providers get their canonical mcp_command;
/// unknown/custom agents are left untouched.
pub fn reconcile_provider_mcp_commands(app: &tauri::AppHandle) {
let Ok(dir) = app.path().app_data_dir() else {
return;
};
let path = dir.join("agents/managed-agents.json");
if !path.exists() {
return;
}
reconcile_mcp_commands_in_file(&path);
}
fn reconcile_pack_paths_in_file(path: &Path, canonical_dir: &Path) {
let canonical_packs = canonical_dir.join("agents/packs");
patch_json_records(path, |obj| {
@@ -667,161 +614,6 @@ mod tests {
serde_json::from_str(&content).unwrap()
}
#[test]
fn reconcile_clears_mcp_command_for_goose() {
let dir = tempfile::tempdir().unwrap();
write_agents_json(
dir.path(),
&serde_json::json!([{
"name": "Scout",
"agent_command": "goose",
"mcp_command": "sprout-mcp-server"
}]),
);
reconcile_mcp_commands_in_file(&dir.path().join("agents/managed-agents.json"));
let records = read_agents_json(dir.path());
assert_eq!(records[0]["mcp_command"], "");
}
#[test]
fn reconcile_clears_mcp_command_for_claude() {
let dir = tempfile::tempdir().unwrap();
write_agents_json(
dir.path(),
&serde_json::json!([{
"name": "Claude Agent",
"agent_command": "claude-agent-acp",
"mcp_command": "sprout-mcp-server"
}]),
);
reconcile_mcp_commands_in_file(&dir.path().join("agents/managed-agents.json"));
let records = read_agents_json(dir.path());
assert_eq!(records[0]["mcp_command"], "");
}
#[test]
fn reconcile_preserves_sprout_dev_mcp() {
let dir = tempfile::tempdir().unwrap();
write_agents_json(
dir.path(),
&serde_json::json!([{
"name": "Solo",
"agent_command": "sprout-agent",
"mcp_command": "sprout-dev-mcp"
}]),
);
let before =
std::fs::read_to_string(dir.path().join("agents/managed-agents.json")).unwrap();
reconcile_mcp_commands_in_file(&dir.path().join("agents/managed-agents.json"));
let after = std::fs::read_to_string(dir.path().join("agents/managed-agents.json")).unwrap();
assert_eq!(
before, after,
"file should not be rewritten when already correct"
);
}
#[test]
fn reconcile_fixes_sprout_agent_if_stale() {
let dir = tempfile::tempdir().unwrap();
write_agents_json(
dir.path(),
&serde_json::json!([{
"name": "Solo",
"agent_command": "sprout-agent",
"mcp_command": "sprout-mcp-server"
}]),
);
reconcile_mcp_commands_in_file(&dir.path().join("agents/managed-agents.json"));
let records = read_agents_json(dir.path());
assert_eq!(records[0]["mcp_command"], "sprout-dev-mcp");
}
#[test]
fn reconcile_leaves_unknown_agent_untouched() {
let dir = tempfile::tempdir().unwrap();
write_agents_json(
dir.path(),
&serde_json::json!([{
"name": "Custom Bot",
"agent_command": "my-custom-agent",
"mcp_command": "my-custom-mcp"
}]),
);
reconcile_mcp_commands_in_file(&dir.path().join("agents/managed-agents.json"));
let records = read_agents_json(dir.path());
assert_eq!(records[0]["mcp_command"], "my-custom-mcp");
}
#[test]
fn reconcile_is_idempotent() {
let dir = tempfile::tempdir().unwrap();
write_agents_json(
dir.path(),
&serde_json::json!([{
"name": "Scout",
"agent_command": "goose",
"mcp_command": "sprout-mcp-server"
}]),
);
let path = dir.path().join("agents/managed-agents.json");
reconcile_mcp_commands_in_file(&path);
let after_first = std::fs::read_to_string(&path).unwrap();
reconcile_mcp_commands_in_file(&path);
let after_second = std::fs::read_to_string(&path).unwrap();
assert_eq!(after_first, after_second);
}
#[test]
fn reconcile_handles_mixed_records() {
let dir = tempfile::tempdir().unwrap();
write_agents_json(
dir.path(),
&serde_json::json!([
{"name": "Scout", "agent_command": "goose", "mcp_command": "sprout-mcp-server"},
{"name": "Claude", "agent_command": "claude-agent-acp", "mcp_command": "sprout-mcp-server"},
{"name": "Solo", "agent_command": "sprout-agent", "mcp_command": "sprout-dev-mcp"},
{"name": "Custom", "agent_command": "my-bot", "mcp_command": "my-mcp"},
{"name": "Codex", "agent_command": "codex-acp", "mcp_command": "sprout-mcp-server"}
]),
);
reconcile_mcp_commands_in_file(&dir.path().join("agents/managed-agents.json"));
let records = read_agents_json(dir.path());
assert_eq!(records[0]["mcp_command"], "", "goose should be cleared");
assert_eq!(records[1]["mcp_command"], "", "claude should be cleared");
assert_eq!(
records[2]["mcp_command"], "sprout-dev-mcp",
"sprout-agent preserved"
);
assert_eq!(
records[3]["mcp_command"], "my-mcp",
"custom agent untouched"
);
assert_eq!(records[4]["mcp_command"], "", "codex should be cleared");
}
#[test]
fn reconcile_leaves_absent_mcp_command_untouched() {
let dir = tempfile::tempdir().unwrap();
let json = serde_json::json!([{"name": "Solo", "agent_command": "sprout-agent"}]);
write_agents_json(dir.path(), &json);
let path = dir.path().join("agents/managed-agents.json");
let before = std::fs::read_to_string(&path).unwrap();
reconcile_mcp_commands_in_file(&path);
assert_eq!(before, std::fs::read_to_string(&path).unwrap());
}
#[test]
fn reconcile_leaves_null_mcp_command_untouched() {
let dir = tempfile::tempdir().unwrap();
let json =
serde_json::json!([{"name":"Solo","agent_command":"sprout-agent","mcp_command":null}]);
write_agents_json(dir.path(), &json);
let path = dir.path().join("agents/managed-agents.json");
let before = std::fs::read_to_string(&path).unwrap();
reconcile_mcp_commands_in_file(&path);
assert_eq!(before, std::fs::read_to_string(&path).unwrap());
}
#[test]
fn sync_creates_packs_directory_symlink() {
let (_parent, canonical, worktree) = setup_sync_layout();
desktop/src-tauri/src/models.rs
-8
View File
@@ -2,8 +2,6 @@ use std::collections::HashMap;
use serde::{Deserialize, Deserializer, Serialize};
use sprout_core::PresenceStatus;
#[derive(Serialize)]
pub struct IdentityInfo {
pub pubkey: String,
@@ -74,12 +72,6 @@ pub struct UserNotesResponse {
pub next_cursor: Option<UserNotesCursor>,
}
#[derive(Serialize, Deserialize)]
pub struct SetPresenceResponse {
pub status: PresenceStatus,
pub ttl_seconds: u64,
}
#[derive(Serialize, Deserialize)]
pub struct ChannelInfo {
pub id: String,
desktop/src/features/presence/hooks.ts
+16 -20
View File
@@ -2,11 +2,11 @@ import * as React from "react";
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
import { relayClient } from "@/shared/api/relayClient";
import { getPresence, setPresence } from "@/shared/api/tauri";
import { getPresence } from "@/shared/api/tauri";
import { normalizePubkey } from "@/shared/lib/pubkey";
import type { PresenceLookup, PresenceStatus } from "@/shared/api/types";
const PRESENCE_HEARTBEAT_INTERVAL_MS = 60_000;
const PRESENCE_HEARTBEAT_INTERVAL_MS = 30_000;
const PRESENCE_IDLE_TIMEOUT_MS = 5 * 60_000;
const PRESENCE_STATUS_TICK_INTERVAL_MS = 30_000;
const PRESENCE_TTL_SECONDS = 90;
@@ -103,12 +103,18 @@ export function usePresenceSubscription() {
let isCancelled = false;
let retryTimer: ReturnType<typeof setTimeout> | null = null;
function handlePresenceEvent(event: { pubkey: string; content: string }) {
function handlePresenceEvent(event: {
pubkey: string;
content: string;
tags?: string[][];
}) {
if (isCancelled) return;
const status = event.content;
if (status !== "online" && status !== "away" && status !== "offline")
return;
const pubkey = event.pubkey.toLowerCase();
const pubkey = (
event.tags?.find((t) => t[0] === "p")?.[1] ?? event.pubkey
).toLowerCase();
queryClient.setQueriesData<PresenceLookup>(
{ queryKey: ["presence"] },
(old) => {
@@ -162,20 +168,13 @@ export function useSetPresenceMutation(pubkey?: string) {
return useMutation({
mutationFn: async (status: PresenceStatus) => {
// Prefer WS — triggers fan-out to subscribers. REST fallback if WS fails.
try {
await relayClient.sendPresence(status);
return {
status,
ttlSeconds: status === "offline" ? 0 : PRESENCE_TTL_SECONDS,
viaWs: true,
};
} catch {
const result = await setPresence(status);
return { ...result, viaWs: false };
}
await relayClient.sendPresence(status);
return {
status,
ttlSeconds: status === "offline" ? 0 : PRESENCE_TTL_SECONDS,
};
},
onSuccess: ({ status, viaWs }) => {
onSuccess: ({ status }) => {
if (normalizedPubkey.length === 0) return;
// Update all cached presence queries containing this pubkey.
queryClient.setQueriesData<PresenceLookup>(
@@ -186,9 +185,6 @@ export function useSetPresenceMutation(pubkey?: string) {
return { ...old, [normalizedPubkey]: status };
},
);
// REST fallback: no WS echo will arrive, invalidate for full refresh.
if (!viaWs)
void queryClient.invalidateQueries({ queryKey: ["presence"] });
},
});
}
desktop/src/shared/api/tauri.ts
-19
View File
@@ -28,7 +28,6 @@ import type {
SendChannelMessageResult,
SetCanvasInput,
SetCanvasResult,
SetPresenceResult,
SetChannelPurposeInput,
SetChannelTopicInput,
UpdateProfileInput,
@@ -83,11 +82,6 @@ type RawSearchUsersResponse = {
type RawPresenceLookup = Record<string, PresenceStatus>;
type RawSetPresenceResult = {
status: PresenceStatus;
ttl_seconds: number;
};
type RawChannel = {
id: string;
name: string;
@@ -526,19 +520,6 @@ export async function getPresence(pubkeys: string[]): Promise<PresenceLookup> {
);
}
export async function setPresence(
status: PresenceStatus,
): Promise<SetPresenceResult> {
const response = await invokeTauri<RawSetPresenceResult>("set_presence", {
status,
});
return {
status: response.status,
ttlSeconds: response.ttl_seconds,
};
}
export function getDefaultRelayUrl(): Promise<string> {
return invokeTauri<string>("get_default_relay_url");
}
desktop/src/shared/api/types.ts
-5
View File
@@ -149,11 +149,6 @@ export type UserStatus = {
export type UserStatusLookup = Record<string, UserStatus | null>;
export type SetPresenceResult = {
status: PresenceStatus;
ttlSeconds: number;
};
export type RelayEvent = {
id: string;
pubkey: string;
desktop/src/testing/e2eBridge.ts
+11 -46
View File
@@ -114,11 +114,6 @@ type PresenceStatus = "online" | "away" | "offline";
type RawPresenceLookup = Record<string, PresenceStatus>;
type RawSetPresenceResponse = {
status: PresenceStatus;
ttl_seconds: number;
};
type RawChannel = {
id: string;
name: string;
@@ -596,7 +591,6 @@ const CHARLIE_PUBKEY =
const OUTSIDER_PUBKEY =
"df8e91b86fda13a9a67896df77232f7bdab2ba9c3e165378e1ba3d24c13a328e";
const MOCK_IDENTITY_PUBKEY = DEFAULT_MOCK_IDENTITY.pubkey;
const MOCK_PRESENCE_TTL_SECONDS = 90;
const mockDisplayNames = new Map<string, string>([
[MOCK_IDENTITY_PUBKEY, DEFAULT_MOCK_IDENTITY.display_name],
@@ -2821,7 +2815,7 @@ async function handleGetPresence(
return {} satisfies RawPresenceLookup;
}
// Presence is ephemeral (kind:20001) — query via bridge which synthesizes from Redis.
// Presence is ephemeral (kind:20001) — mock returns from in-memory map.
const events = await relayQuery(config, [
{ kinds: [20001], authors: args.pubkeys, limit: args.pubkeys.length },
]);
@@ -2841,40 +2835,6 @@ async function handleGetPresence(
return result;
}
async function handleSetPresence(
args: {
status: PresenceStatus;
},
config: E2eConfig | undefined,
) {
const identity = getIdentity(config);
if (!identity) {
setMockPresenceStatus(getMockMemberPubkey(config), args.status);
return {
status: args.status,
ttl_seconds: args.status === "offline" ? 0 : MOCK_PRESENCE_TTL_SECONDS,
} satisfies RawSetPresenceResponse;
}
// Presence is ephemeral kind:20001 — submit via POST /events.
// Note: the relay may reject this with "kind 20001 is only accepted via WebSocket"
// in which case we just return the expected shape (presence is best-effort in e2e).
try {
await submitSignedEvent(config, {
kind: 20001,
content: args.status,
tags: [],
});
} catch {
// Expected: ephemeral events may be WS-only
}
return {
status: args.status,
ttl_seconds: args.status === "offline" ? 0 : 90,
};
}
async function handleCreateChannel(
args: {
name: string;
@@ -5168,6 +5128,16 @@ function sendToMockSocket(args: {
return;
}
if (event.kind === 20001) {
const status = event.content;
if (status === "online" || status === "away" || status === "offline") {
setMockPresenceStatus(event.pubkey, status);
}
emitMockGlobalEvent(event);
sendWsText(socket.handler, ["OK", event.id, true, ""]);
return;
}
if (event.kind === KIND_USER_STATUS) {
const hasGeneralDTag = event.tags.some(
(tag) => tag[0] === "d" && tag[1] === "general",
@@ -5497,11 +5467,6 @@ export function maybeInstallE2eTauriMocks() {
},
activeConfig,
);
case "set_presence":
return handleSetPresence(
payload as Parameters<typeof handleSetPresence>[0],
activeConfig,
);
case "get_relay_ws_url":
return getRelayWsUrl(activeConfig);
case "get_default_relay_url":