From ded2c7deadba3e97d0f27cba66338e75f665a8e5 Mon Sep 17 00:00:00 2001 From: Salman Mohammed Date: Mon, 24 Aug 2026 14:03:50 -0400 Subject: [PATCH] Deduplicate ACP thread prompt context Signed-off-by: Salman Mohammed --- crates/buzz-acp/src/pool.rs | 60 ++++++ crates/buzz-acp/src/queue.rs | 360 ++++++++++++++++++++++++++++++++--- 2 files changed, 393 insertions(+), 27 deletions(-) diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index f5176d3f159..d1f241746e2 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -3278,12 +3278,14 @@ fn conversation_context_delta( ConversationContext::Thread { messages, total, + root_present, truncated, } => { let messages = filter(messages); (!messages.is_empty()).then_some(ConversationContext::Thread { messages, total, + root_present, truncated, }) } @@ -3753,6 +3755,7 @@ fn parse_thread_response(json: serde_json::Value) -> Option Some(ConversationContext::Thread { messages, total, + root_present: json.get("root").and_then(json_to_context_message).is_some(), truncated, }) } @@ -3931,6 +3934,7 @@ fn parse_nostr_thread_response_with_meta( context: ConversationContext::Thread { messages, total, + root_present, truncated, }, root_present, @@ -5109,11 +5113,13 @@ mod tests { ConversationContext::Thread { messages, total, + root_present, truncated, } => { assert_eq!(messages.len(), 2); // root + 1 reply assert_eq!(total, 2); // 1 reply + 1 root assert!(!truncated); + assert!(root_present); assert_eq!(messages[0].content, "root message"); assert_eq!(messages[1].content, "first reply"); } @@ -5146,11 +5152,13 @@ mod tests { ConversationContext::Thread { messages, total, + root_present, truncated, } => { assert_eq!(messages.len(), 2); assert_eq!(total, 11); // 10 replies + 1 root assert!(truncated); + assert!(root_present); } _ => panic!("expected Thread context"), } @@ -5321,11 +5329,13 @@ mod tests { ConversationContext::Thread { messages, total, + root_present, truncated, } => { assert_eq!(messages.len(), 3); // root + 2 displayed replies assert_eq!(total, 4); // root + displayed replies + sentinel assert!(truncated); + assert!(root_present); assert_eq!(messages[0].content, "root"); assert_eq!(messages[1].content, "middle reply"); assert_eq!(messages[2].content, "newest agent reply"); @@ -5362,11 +5372,50 @@ mod tests { ConversationContext::Thread { messages, total, + root_present, truncated, } => { assert_eq!(messages.len(), 2); assert_eq!(total, 2); assert!(!truncated); + assert!(root_present); + } + _ => panic!("expected Thread context"), + } + } + + #[test] + fn test_parse_nostr_thread_response_marks_missing_root_incomplete() { + let agent = Keys::generate(); + let root_id = "1111111111111111111111111111111111111111111111111111111111111111"; + let json = json!([ + { + "id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "pubkey": "replypub1", + "content": "first reply", + "created_at": 2000 + }, + { + "id": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "pubkey": "replypub2", + "content": "second reply", + "created_at": 3000 + } + ]); + + let ctx = parse_nostr_thread_response(json, root_id, 12, &agent.public_key()) + .expect("reply context should still be available"); + match ctx { + ConversationContext::Thread { + messages, + total, + root_present, + truncated, + } => { + assert_eq!(messages.len(), 2); + assert_eq!(total, 2); + assert!(!truncated); + assert!(!root_present); } _ => panic!("expected Thread context"), } @@ -5483,6 +5532,7 @@ mod tests { messages, total, truncated, + .. } => { assert!(truncated); assert_eq!(messages.len(), 3); @@ -5533,11 +5583,13 @@ mod tests { ConversationContext::Thread { messages, total, + root_present, truncated, } => { assert!(truncated); assert_eq!(messages.len(), 2); assert_eq!(total, 6); + assert!(!root_present); } _ => panic!("expected Thread context"), } @@ -5586,6 +5638,7 @@ mod tests { messages, total, truncated, + .. } => { assert!(truncated); assert_eq!(messages.len(), 3); @@ -5638,6 +5691,7 @@ mod tests { messages, total, truncated, + .. } => { assert!(truncated); assert_eq!(messages.len(), 3); @@ -5699,6 +5753,7 @@ mod tests { messages, total, truncated, + .. } => { assert!(truncated); assert_eq!(total, 4); @@ -5772,6 +5827,7 @@ mod tests { messages, total, truncated, + .. } => { assert!(truncated); assert_eq!(messages.len(), 3); @@ -5910,6 +5966,7 @@ mod tests { content: "follow up".into(), }], total: 1, + root_present: true, truncated: false, }; @@ -6579,6 +6636,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" context_message("new", "new context"), ], total: 3, + root_present: true, truncated: false, }; @@ -6588,12 +6646,14 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" ConversationContext::Thread { messages, total, + root_present, truncated, } => { assert_eq!(messages.len(), 1); assert_eq!(messages[0].event_id, "new"); assert_eq!(total, 3); assert!(!truncated); + assert!(root_present); } _ => panic!("expected thread context"), } diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 8fd0a02fe7e..ab58267186a 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -992,13 +992,21 @@ pub enum ConversationContext { /// Thread context for a reply event. Thread { messages: Vec, + /// Exact visible count when complete; otherwise a proven lower bound. total: usize, + /// Whether the fetched context included the thread-opening event. + /// A reply-only window cannot be treated as complete even when it was + /// not capped by the configured message limit. + root_present: bool, + /// Whether replies exceeded the configured display window. truncated: bool, }, /// DM conversation history. Dm { messages: Vec, + /// Exact visible count when below the fetch limit; otherwise a lower bound. total: usize, + /// Whether the fetch filled its configured window and may omit history. truncated: bool, }, } @@ -1144,7 +1152,9 @@ pub(crate) fn format_event_block( let thread = parse_thread_tags(&be.event); let mut parsed_parts = Vec::new(); if let Some(ref p) = thread.parent_event_id { - parsed_parts.push(format!("parent={p}")); + if thread.root_event_id.as_ref() != Some(p) { + parsed_parts.push(format!("parent={p}")); + } } if let Some(ref r) = thread.root_event_id { parsed_parts.push(format!("root={r}")); @@ -1303,14 +1313,21 @@ fn format_context_hints( channel_info: Option<&PromptChannelInfo>, thread_tags: &ThreadTags, is_dm: bool, - has_conversation_context: bool, - conversation_context_had_delivered_events: bool, + conversation_context_status: ConversationContextStatus, reply_anchor: Option<&str>, ) -> String { let channel_display = match channel_info { Some(ci) => format!("{} (#{channel_id})", ci.name), None => channel_id.to_string(), }; + let has_conversation_context = matches!( + conversation_context_status, + ConversationContextStatus::Complete | ConversationContextStatus::Included + ); + let complete_conversation_context = + conversation_context_status == ConversationContextStatus::Complete; + let conversation_context_had_delivered_events = + conversation_context_status == ConversationContextStatus::PreviouslyDelivered; // DM check comes first — a DM reply has both thread tags AND is_dm=true, // and the scope should be "dm" (not "thread") because the agent is in a DM. @@ -1318,7 +1335,11 @@ fn format_context_hints( let is_reply = thread_tags.root_event_id.is_some(); // DM replies use thread command because /messages excludes thread replies. // DM non-replies use get for recent conversation. - let ctx_hint = if has_conversation_context && is_reply { + let ctx_hint = if complete_conversation_context && is_reply { + "Thread context included below." + } else if complete_conversation_context { + "Conversation context included below." + } else if has_conversation_context && is_reply { "Thread context included below. Use `buzz messages thread --channel --event ` for full history if truncated." } else if has_conversation_context { "Conversation context included below. Use `buzz messages get --channel ` for full history if truncated." @@ -1350,7 +1371,9 @@ fn format_context_hints( } crate::prompt_framing::semantic_section("context", &s) } else if let Some(ref root) = thread_tags.root_event_id { - let ctx_hint = if has_conversation_context { + let ctx_hint = if complete_conversation_context { + "Thread context included below." + } else if has_conversation_context { "Thread context included below. Use `buzz messages thread --channel --event ` for full history if truncated." } else if conversation_context_had_delivered_events { "Earlier thread context was already delivered in this session. Use `buzz messages thread --channel --event ` to re-read it." @@ -1389,6 +1412,86 @@ fn format_context_hints( } } +#[derive(Clone, Copy, Eq, PartialEq)] +enum ConversationContextStatus { + Complete, + Included, + PreviouslyDelivered, + Absent, +} + +/// Whether the fetched context covers every event rendered in this turn. +/// +/// Thread context is fetched for the last event's root only, so a mixed batch +/// must keep the retrieval hint. A thread window that omitted its root is also +/// incomplete even when it did not hit the reply limit. DM history covers only +/// top-level DM events, not reply threads. +fn conversation_context_covers_batch( + batch: &FlushBatch, + conversation_context: Option<&ConversationContext>, +) -> bool { + match conversation_context { + Some(ConversationContext::Thread { + root_present: true, .. + }) => { + let Some(expected_root) = batch + .events + .last() + .and_then(|event| parse_thread_tags(&event.event).root_event_id) + else { + return false; + }; + + batch + .cancelled_events + .iter() + .chain(&batch.events) + .all(|event| { + parse_thread_tags(&event.event).root_event_id.as_deref() + == Some(expected_root.as_str()) + }) + } + Some(ConversationContext::Dm { .. }) => batch + .cancelled_events + .iter() + .chain(&batch.events) + .all(|event| parse_thread_tags(&event.event).root_event_id.is_none()), + _ => false, + } +} + +fn conversation_context_status( + batch: &FlushBatch, + conversation_context: Option<&ConversationContext>, + conversation_context_had_delivered_events: bool, +) -> ConversationContextStatus { + let window_is_complete = matches!( + conversation_context, + Some( + ConversationContext::Thread { + truncated: false, + .. + } | ConversationContext::Dm { + truncated: false, + .. + } + ) + ); + + if window_is_complete + && conversation_context_covers_batch(batch, conversation_context) + && !conversation_context_had_delivered_events + { + ConversationContextStatus::Complete + } else if conversation_context.is_some() { + ConversationContextStatus::Included + } else if conversation_context_had_delivered_events { + ConversationContextStatus::PreviouslyDelivered + } else { + ConversationContextStatus::Absent + } +} + /// Format a conversation context section (thread or DM). fn format_conversation_context( ctx: &ConversationContext, @@ -1399,6 +1502,7 @@ fn format_conversation_context( messages, total, truncated, + .. } => ("thread-context", messages, total, truncated), ConversationContext::Dm { messages, @@ -1634,8 +1738,11 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec")); - assert!(prompt.contains("Let's refactor auth")); - assert!(prompt.contains("Thread context included below")); + assert!(complete_prompt.contains("Thread context included below.")); + assert!(!complete_prompt.contains("buzz messages thread")); + assert!(!complete_prompt.contains("full history")); + assert!(complete_prompt + .contains("")); + assert!(complete_prompt.contains("Let's refactor auth")); + assert!(complete_prompt.contains(&format!( + "IMPORTANT: For ordinary replies in this turn, use `--reply-to {root}`" + ))); + + let prompt_with_prior_delivery = format_prompt( + &batch, + &FormatPromptArgs { + conversation_context: Some(&ctx), + conversation_context_had_delivered_events: true, + ..Default::default() + }, + ) + .join("\n\n"); + assert!(prompt_with_prior_delivery.contains("buzz messages thread")); + assert!(prompt_with_prior_delivery + .contains("")); + assert!(prompt_with_prior_delivery.contains("Let's refactor auth")); + + if let ConversationContext::Thread { + total, truncated, .. + } = &mut ctx + { + *total = 5; + *truncated = true; + } + let truncated_prompt = format_prompt( + &batch, + &FormatPromptArgs { + conversation_context: Some(&ctx), + ..Default::default() + }, + ) + .join("\n\n"); + assert!(truncated_prompt + .contains("")); + assert!(truncated_prompt.contains("buzz messages thread")); + assert!(truncated_prompt.contains("for full history if truncated")); + + if let ConversationContext::Thread { + total, + root_present, + truncated, + .. + } = &mut ctx + { + *total = 2; + *root_present = false; + *truncated = false; + } + let missing_root_prompt = format_prompt( + &batch, + &FormatPromptArgs { + conversation_context: Some(&ctx), + ..Default::default() + }, + ) + .join("\n\n"); + assert!(missing_root_prompt + .contains("")); + assert!(missing_root_prompt.contains("Let's refactor auth")); + assert!(missing_root_prompt.contains("buzz messages thread")); + } + + #[test] + fn test_thread_context_retrieval_hint_requires_batch_coverage() { + let ch = Uuid::new_v4(); + let root_a = "a".repeat(64); + let root_b = "b".repeat(64); + let reply = |content: &str, root: &str| BatchEvent { + event: make_event_with_tags( + content, + vec![vec!["e".into(), root.into(), "".into(), "reply".into()]], + ), + prompt_tag: "@mention".into(), + received_at: Instant::now(), + }; + let ctx = ConversationContext::Thread { + messages: vec![ContextMessage { + event_id: root_b.clone(), + pubkey: "npub1xyz".into(), + timestamp: "2026-03-15T16:30:00Z".into(), + content: "thread B root question".into(), + }], + total: 1, + root_present: true, + truncated: false, + }; + + let mixed_batch = FlushBatch { + channel_id: ch, + events: vec![ + reply("older reply in thread A", &root_a), + reply("newer reply in thread B", &root_b), + ], + cancelled_events: vec![], + cancel_reason: None, + }; + let mixed_prompt = format_prompt( + &mixed_batch, + &FormatPromptArgs { + conversation_context: Some(&ctx), + ..Default::default() + }, + ) + .join("\n\n"); + assert!(mixed_prompt + .contains("")); + assert!(mixed_prompt.contains("thread B root question")); + assert!(mixed_prompt.contains("older reply in thread A")); + assert!(mixed_prompt.contains("newer reply in thread B")); + assert!(mixed_prompt.contains("buzz messages thread")); + + let same_thread_batch = FlushBatch { + channel_id: ch, + events: vec![ + reply("older reply in thread B", &root_b), + reply("newer reply in thread B", &root_b), + ], + cancelled_events: vec![], + cancel_reason: None, + }; + let same_thread_prompt = format_prompt( + &same_thread_batch, + &FormatPromptArgs { + conversation_context: Some(&ctx), + ..Default::default() + }, + ) + .join("\n\n"); + assert!(same_thread_prompt + .contains("")); + assert!(same_thread_prompt.contains("thread B root question")); + assert!(!same_thread_prompt.contains("buzz messages thread")); } #[test] @@ -3468,6 +3709,9 @@ mod tests { ) .join("\n\n"); assert!(prompt.contains("Scope: dm")); + assert!(prompt.contains("Conversation context included below.")); + assert!(!prompt.contains("buzz messages get")); + assert!(!prompt.contains("full history")); assert!(prompt .contains("")); assert!(prompt.contains("Can you deploy?")); @@ -3502,6 +3746,7 @@ mod tests { content: "follow up".into(), }], total: 1, + root_present: true, truncated: false, }; let profiles = HashMap::from([ @@ -3680,7 +3925,7 @@ mod tests { } #[test] - fn test_format_prompt_dm_reply_hints_get_thread() { + fn test_format_prompt_dm_reply_with_complete_thread_context_omits_retrieval_hint() { let ch = Uuid::new_v4(); // DM reply event — has thread e-tags. let event = make_event_with_tags( @@ -3716,6 +3961,7 @@ mod tests { content: "Should I deploy?".into(), }], total: 1, + root_present: true, truncated: false, }; @@ -3733,11 +3979,9 @@ mod tests { prompt.contains("Scope: dm"), "DM reply should have Scope: dm, got:\n{prompt}" ); - // Hint should point to the thread command, not get. - assert!( - prompt.contains("buzz messages thread"), - "DM reply hint should mention `buzz messages thread`, got:\n{prompt}" - ); + assert!(prompt.contains("Thread context included below.")); + assert!(!prompt.contains("buzz messages thread")); + assert!(!prompt.contains("full history")); // Thread structural info should be present. assert!( prompt.contains( @@ -3746,6 +3990,7 @@ mod tests { "DM reply should include thread root" ); // Thread context should be included. + assert!(prompt.contains("")); assert!(prompt.contains("Should I deploy?")); } @@ -3949,6 +4194,67 @@ mod tests { ); } + #[test] + fn test_format_event_block_only_omits_parent_when_it_duplicates_root() { + let ch = Uuid::new_v4(); + let root = "a".repeat(64); + let parent = "b".repeat(64); + let mention = "c".repeat(64); + + let direct_event = make_event_with_tags( + "direct reply", + vec![ + vec!["e".into(), root.clone(), "".into(), "reply".into()], + vec!["p".into(), mention.clone()], + ], + ); + let direct_event_id = direct_event.id.to_hex(); + let direct_block = format_event_block( + ch, + None, + &BatchEvent { + event: direct_event, + prompt_tag: "test".into(), + received_at: Instant::now(), + }, + None, + ); + + assert!(direct_block.contains(&format!("Event ID: {direct_event_id}"))); + assert!(direct_block.contains("From:")); + assert!(direct_block.contains(&format!( + "Tags: [[\"e\",\"{root}\",\"\",\"reply\"],[\"p\",\"{mention}\"]]" + ))); + assert!(direct_block.contains(&format!("Parsed: root={root}, mentions=[{mention}]"))); + assert!(!direct_block.contains(&format!("parent={root}"))); + + let nested_event = make_event_with_tags( + "nested reply", + vec![ + vec!["e".into(), root.clone(), "".into(), "root".into()], + vec!["e".into(), parent.clone(), "".into(), "reply".into()], + vec!["p".into(), mention.clone()], + ], + ); + let nested_block = format_event_block( + ch, + None, + &BatchEvent { + event: nested_event, + prompt_tag: "test".into(), + received_at: Instant::now(), + }, + None, + ); + + assert!(nested_block.contains(&format!( + "Tags: [[\"e\",\"{root}\",\"\",\"root\"],[\"e\",\"{parent}\",\"\",\"reply\"],[\"p\",\"{mention}\"]]" + ))); + assert!(nested_block.contains(&format!( + "Parsed: parent={parent}, root={root}, mentions=[{mention}]" + ))); + } + #[test] fn test_drain_channel_removes_pending_events() { let mut q = EventQueue::new(DedupMode::Queue);