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: make branch reply scoping relay-driven

Browse Source 
Move drilled thread reply scoping out of MCP by having relay ingest suppress broadcast for branch-owned replies and letting desktop inherit branch ownership from the parent chain for live events.

Made-with: Cursor
This commit is contained in:
Thomas Petersen
2026-04-13 12:33:07 -04:00
parent 594a43998f
commit c3d4807002
5 changed files with 104 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/sprout-cli/src/commands/messages.rs
-1
View File
@@ -201,7 +201,6 @@ pub async fn cmd_send_message(
channel_uuid,
content,
thread_ref.as_ref(),
None,
&mention_refs,
broadcast,
&[],
crates/sprout-mcp/src/server.rs
+11 -79
View File
@@ -103,29 +103,6 @@ fn find_agent_reply_parent_from_tags(tags: &serde_json::Value) -> Option<String>
None
}
/// Extract a Sprout-specific UI branch head from serialized tags.
///
/// Tag format:
/// - `["sprout", "thread_branch_head", "<event-id>"]`
fn find_thread_branch_head_from_tags(tags: &serde_json::Value) -> Option<String> {
let arr = tags.as_array()?;
for tag in arr {
let Some(parts) = tag.as_array() else {
continue;
};
if parts.len() >= 3
&& parts[0].as_str() == Some("sprout")
&& parts[1].as_str() == Some("thread_branch_head")
{
let branch_head = parts[2].as_str()?;
if branch_head.len() == 64 && branch_head.chars().all(|c| c.is_ascii_hexdigit()) {
return Some(branch_head.to_ascii_lowercase());
}
}
}
None
}
/// Maximum allowed content size for a single message (64 KiB).
const MAX_CONTENT_BYTES: usize = 65_536;
@@ -858,11 +835,6 @@ pub struct SproutMcpServer {
tool_router: ToolRouter<Self>,
}
struct ResolvedThreadRef {
thread_ref: sprout_sdk::ThreadRef,
thread_branch_head_id: Option<String>,
}
#[tool_router]
impl SproutMcpServer {
/// Create a new [`SproutMcpServer`] backed by the given relay client.
@@ -893,7 +865,7 @@ impl SproutMcpServer {
async fn resolve_thread_ref(
&self,
parent_event_id: &str,
) -> Result<ResolvedThreadRef, String> {
) -> Result<sprout_sdk::ThreadRef, String> {
let resp = self
.client
.get(&format!("/api/events/{}", parent_event_id))
@@ -903,7 +875,6 @@ impl SproutMcpServer {
let event_json: serde_json::Value = serde_json::from_str(&resp)
.map_err(|e| format!("failed to parse parent event: {e}"))?;
let thread_branch_head_id = find_thread_branch_head_from_tags(&event_json["tags"]);
let effective_parent_id = find_agent_reply_parent_from_tags(&event_json["tags"])
.unwrap_or_else(|| parent_event_id.to_ascii_lowercase());
let parent_eid = EventId::from_hex(&effective_parent_id)
@@ -915,12 +886,9 @@ impl SproutMcpServer {
_ => parent_eid,
};
Ok(ResolvedThreadRef {
thread_ref: sprout_sdk::ThreadRef {
root_event_id: root_eid,
parent_event_id: parent_eid,
},
thread_branch_head_id,
Ok(sprout_sdk::ThreadRef {
root_event_id: root_eid,
parent_event_id: parent_eid,
})
}
@@ -1013,14 +981,14 @@ Default kind is 9 (stream message)."
None => return "Error: kind 45003 requires parent_event_id".to_string(),
};
// Fetch parent to resolve thread root for NIP-10 markers.
let resolved_thread_ref = match self.resolve_thread_ref(parent_id).await {
let thread_ref = match self.resolve_thread_ref(parent_id).await {
Ok(tr) => tr,
Err(e) => return format!("Error: {e}"),
};
match sprout_sdk::build_forum_comment(
channel_uuid,
&p.content,
&resolved_thread_ref.thread_ref,
&thread_ref,
&mention_refs,
&[],
) {
@@ -1030,7 +998,7 @@ Default kind is 9 (stream message)."
}
_ => {
// kind 9 (default) and any other stream message kinds.
let resolved_thread_ref = if let Some(ref parent_id) = p.parent_event_id {
let thread_ref = if let Some(ref parent_id) = p.parent_event_id {
match self.resolve_thread_ref(parent_id).await {
Ok(tr) => Some(tr),
Err(e) => return format!("Error: {e}"),
@@ -1038,23 +1006,12 @@ Default kind is 9 (stream message)."
} else {
None
};
let effective_broadcast = broadcast
&& p.parent_event_id.is_some()
&& resolved_thread_ref
.as_ref()
.and_then(|resolved| resolved.thread_branch_head_id.as_deref())
.is_none();
match sprout_sdk::build_message(
channel_uuid,
&p.content,
resolved_thread_ref
.as_ref()
.map(|resolved| &resolved.thread_ref),
resolved_thread_ref
.as_ref()
.and_then(|resolved| resolved.thread_branch_head_id.as_deref()),
thread_ref.as_ref(),
&mention_refs,
effective_broadcast,
broadcast && p.parent_event_id.is_some(),
&[],
) {
Ok(b) => b,
@@ -1140,7 +1097,7 @@ Default kind is 9 (stream message)."
};
// 5. Resolve optional thread ref
let resolved_thread_ref = if let Some(ref parent_id) = parent_event_id {
let thread_ref = if let Some(ref parent_id) = parent_event_id {
match self.resolve_thread_ref(parent_id).await {
Ok(tr) => Some(tr),
Err(e) => return format!("Error: {e}"),
@@ -1166,9 +1123,7 @@ Default kind is 9 (stream message)."
channel_uuid,
&diff_content,
&diff_meta,
resolved_thread_ref
.as_ref()
.map(|resolved| &resolved.thread_ref),
thread_ref.as_ref(),
) {
Ok(b) => b,
Err(e) => return format!("Error: {e}"),
@@ -2795,29 +2750,6 @@ mod tests {
assert!(find_agent_reply_parent_from_tags(&tags).is_none());
}
#[test]
fn find_thread_branch_head_from_tags_matches_valid_tag() {
let tags = serde_json::json!([
["h", "channel-id"],
["sprout", "thread_branch_head", "CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC"]
]);
assert_eq!(
find_thread_branch_head_from_tags(&tags).as_deref(),
Some("cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc")
);
}
#[test]
fn find_thread_branch_head_from_tags_ignores_invalid_values() {
let tags = serde_json::json!([
["sprout", "thread_branch_head", "short"],
["sprout", "agent_reply_parent", "DDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD"]
]);
assert!(find_thread_branch_head_from_tags(&tags).is_none());
}
// ── MAX_CONTENT_BYTES ─────────────────────────────────────────────────────
#[test]
crates/sprout-relay/src/handlers/ingest.rs
+47 -1
View File
@@ -324,6 +324,22 @@ fn check_token_channel_access(auth: &IngestAuth, channel_id: Uuid) -> Result<(),
Ok(())
}
fn find_thread_branch_head_in_event(event: &Event) -> Option<String> {
event.tags.iter().find_map(|tag| {
let parts = tag.as_slice();
if parts.len() >= 3
&& parts[0] == "sprout"
&& parts[1] == "thread_branch_head"
&& parts[2].len() == 64
&& parts[2].chars().all(|c| c.is_ascii_hexdigit())
{
Some(parts[2].to_ascii_lowercase())
} else {
None
}
})
}
// ── NIP-10 thread resolution ─────────────────────────────────────────────────
/// Owned thread metadata for the DB insert.
@@ -412,6 +428,9 @@ pub(crate) async fn resolve_nip10_thread_meta(
let parent_created =
chrono::DateTime::from_timestamp(parent_event.event.created_at.as_u64() as i64, 0)
.unwrap_or_else(Utc::now);
let explicit_branch_head_hex = find_thread_branch_head_in_event(event);
let parent_branch_head_hex = find_thread_branch_head_in_event(&parent_event.event);
let effective_branch_head_hex = explicit_branch_head_hex.or(parent_branch_head_hex);
let client_root_bytes =
hex::decode(&root_hex).map_err(|_| "invalid root event ID hex".to_string())?;
@@ -484,7 +503,7 @@ pub(crate) async fn resolve_nip10_thread_meta(
let broadcast = event.tags.iter().any(|t| {
let parts = t.as_slice();
parts.len() >= 2 && parts[0] == "broadcast" && parts[1] == "1"
});
}) && effective_branch_head_hex.is_none();
let event_created_at = chrono::DateTime::from_timestamp(event.created_at.as_u64() as i64, 0)
.unwrap_or_else(Utc::now);
@@ -1603,6 +1622,33 @@ mod tests {
.unwrap()
}
#[test]
fn find_thread_branch_head_in_event_matches_valid_tag() {
let event = make_event_with_tags(
KIND_STREAM_MESSAGE,
"hello",
&[&[
"sprout",
"thread_branch_head",
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
]],
);
assert_eq!(
find_thread_branch_head_in_event(&event).as_deref(),
Some("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
);
}
#[test]
fn find_thread_branch_head_in_event_ignores_invalid_tag() {
let event = make_event_with_tags(
KIND_STREAM_MESSAGE,
"hello",
&[&["sprout", "thread_branch_head", "short"]],
);
assert!(find_thread_branch_head_in_event(&event).is_none());
}
#[test]
fn count_e_tags_includes_malformed() {
// A deletion event with one valid e-tag and one malformed e-tag
crates/sprout-sdk/src/builders.rs
+8 -44
View File
@@ -50,17 +50,6 @@ fn thread_tags(thread_ref: &ThreadRef, tags: &mut Vec<Tag>) -> Result<(), SdkErr
Ok(())
}
fn thread_branch_head_tag(thread_branch_head_id: &str) -> Result<Tag, SdkError> {
if thread_branch_head_id.len() != 64
|| !thread_branch_head_id.chars().all(|c| c.is_ascii_hexdigit())
{
return Err(SdkError::InvalidInput(
"thread_branch_head_id must be exactly 64 hex characters".into(),
));
}
tag(&["sprout", "thread_branch_head", thread_branch_head_id])
}
/// Deduplicate and cap mentions, emitting p-tags.
fn mention_tags(mentions: &[&str], tags: &mut Vec<Tag>) -> Result<(), SdkError> {
if mentions.len() > 50 {
@@ -92,7 +81,6 @@ fn imeta_tags(media_tags: &[Vec<String>], tags: &mut Vec<Tag>) -> Result<(), Sdk
/// - `channel_id`: target channel UUID
/// - `content`: message text (max 64 KiB)
/// - `thread_ref`: optional NIP-10 reply context
/// - `thread_branch_head_id`: optional UI branch head for deterministic thread rendering
/// - `mentions`: pubkey hex strings to p-tag (deduped, max 50)
/// - `broadcast`: if true, adds `["broadcast", "1"]` tag
/// - `media_tags`: raw imeta tag vectors
@@ -100,7 +88,6 @@ pub fn build_message(
channel_id: Uuid,
content: &str,
thread_ref: Option<&ThreadRef>,
thread_branch_head_id: Option<&str>,
mentions: &[&str],
broadcast: bool,
media_tags: &[Vec<String>],
@@ -109,13 +96,6 @@ pub fn build_message(
let mut tags = vec![tag(&["h", &channel_id.to_string()])?];
if let Some(tr) = thread_ref {
thread_tags(tr, &mut tags)?;
if let Some(branch_head_id) = thread_branch_head_id {
tags.push(thread_branch_head_tag(branch_head_id)?);
}
} else if thread_branch_head_id.is_some() {
return Err(SdkError::InvalidInput(
"thread_branch_head_id requires thread_ref".into(),
));
}
mention_tags(mentions, &mut tags)?;
if broadcast {
@@ -653,7 +633,7 @@ mod tests {
#[test]
fn message_happy_path() {
let cid = uuid();
let ev = sign(build_message(cid, "hello", None, None, &[], false, &[]).unwrap());
let ev = sign(build_message(cid, "hello", None, &[], false, &[]).unwrap());
assert_eq!(ev.kind.as_u16(), 9);
assert_eq!(ev.content, "hello");
assert!(has_tag(&ev, "h", &cid.to_string()));
@@ -667,7 +647,7 @@ mod tests {
root_event_id: eid,
parent_event_id: eid,
};
let ev = sign(build_message(cid, "reply", Some(&tr), None, &[], false, &[]).unwrap());
let ev = sign(build_message(cid, "reply", Some(&tr), &[], false, &[]).unwrap());
// Direct reply: only one e-tag with "reply" marker
let e_tags: Vec<_> = ev
.tags
@@ -690,7 +670,7 @@ mod tests {
root_event_id: root,
parent_event_id: parent,
};
let ev = sign(build_message(cid, "nested", Some(&tr), None, &[], false, &[]).unwrap());
let ev = sign(build_message(cid, "nested", Some(&tr), &[], false, &[]).unwrap());
let e_tags: Vec<_> = ev
.tags
.iter()
@@ -705,25 +685,9 @@ mod tests {
assert!(markers.contains(&"reply"));
}
#[test]
fn message_thread_branch_head_tag() {
let cid = uuid();
let root = event_id();
let tr = ThreadRef {
root_event_id: root,
parent_event_id: root,
};
let branch_head = "abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234";
let ev = sign(
build_message(cid, "hi", Some(&tr), Some(branch_head), &[], false, &[]).unwrap(),
);
assert!(has_tag(&ev, "sprout", "thread_branch_head"));
}
#[test]
fn message_broadcast_flag() {
let cid = uuid();
let ev = sign(build_message(cid, "hi", None, None, &[], true, &[]).unwrap());
let ev = sign(build_message(cid, "hi", None, &[], true, &[]).unwrap());
assert!(has_tag(&ev, "broadcast", "1"));
}
@@ -731,7 +695,7 @@ mod tests {
fn message_mentions_deduped() {
let cid = uuid();
let hex = "abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234";
let ev = sign(build_message(cid, "hi", None, None, &[hex, hex], false, &[]).unwrap());
let ev = sign(build_message(cid, "hi", None, &[hex, hex], false, &[]).unwrap());
let p_tags = tag_values(&ev, "p");
assert_eq!(p_tags.len(), 1);
}
@@ -752,7 +716,7 @@ mod tests {
})
.collect();
let refs: Vec<&str> = hexes.iter().map(|s| s.as_str()).collect();
let result = build_message(cid, "hi", None, None, &refs, false, &[]);
let result = build_message(cid, "hi", None, &refs, false, &[]);
assert!(matches!(result, Err(SdkError::TooManyMentions)));
}
@@ -760,7 +724,7 @@ mod tests {
fn message_content_too_large() {
let cid = uuid();
let big = "x".repeat(64 * 1024 + 1);
let result = build_message(cid, &big, None, None, &[], false, &[]);
let result = build_message(cid, &big, None, &[], false, &[]);
assert!(matches!(result, Err(SdkError::ContentTooLarge { .. })));
}
@@ -768,7 +732,7 @@ mod tests {
fn message_max_content_ok() {
let cid = uuid();
let max = "x".repeat(64 * 1024);
assert!(build_message(cid, &max, None, None, &[], false, &[]).is_ok());
assert!(build_message(cid, &max, None, &[], false, &[]).is_ok());
}
// ── build_forum_post ─────────────────────────────────────────────────────
desktop/src/features/messages/lib/formatTimelineMessages.ts
+38 -1
View File
@@ -231,8 +231,10 @@ export function formatTimelineMessages(
const authorPubkeyByEventId = new Map<string, string>();
const authorLabelByEventId = new Map<string, string>();
const branchHeadByEventId = new Map<string, string | null>();
const depthByEventId = new Map<string, number>();
const resolvingEventIds = new Set<string>();
const resolvingBranchHeadEventIds = new Set<string>();
function getAuthorLabel(event: RelayEvent) {
const cached = authorLabelByEventId.get(event.id);
@@ -284,6 +286,41 @@ export function formatTimelineMessages(
return depth;
}
function getBranchHeadId(event: RelayEvent): string | null {
const cached = branchHeadByEventId.get(event.id);
if (cached !== undefined) {
return cached;
}
if (resolvingBranchHeadEventIds.has(event.id)) {
return null;
}
const explicitBranchHeadId = getThreadBranchHeadFromTags(event.tags);
if (explicitBranchHeadId) {
branchHeadByEventId.set(event.id, explicitBranchHeadId);
return explicitBranchHeadId;
}
const thread = getThreadReference(event.tags);
if (!thread.parentId) {
branchHeadByEventId.set(event.id, null);
return null;
}
const parent = eventsById.get(thread.parentId);
if (!parent) {
branchHeadByEventId.set(event.id, null);
return null;
}
resolvingBranchHeadEventIds.add(event.id);
const inheritedBranchHeadId = getBranchHeadId(parent);
resolvingBranchHeadEventIds.delete(event.id);
branchHeadByEventId.set(event.id, inheritedBranchHeadId);
return inheritedBranchHeadId;
}
return visibleEvents.map((event) => {
const author = getAuthorLabel(event);
const authorPubkey =
@@ -317,7 +354,7 @@ export function formatTimelineMessages(
body: edit ? edit.content : event.content,
parentId: thread.parentId,
rootId: thread.rootId,
branchHeadId: getThreadBranchHeadFromTags(event.tags),
branchHeadId: getBranchHeadId(event),
depth: getDepth(event),
accent: currentPubkey === authorPubkey,
pending: event.pending,