diff --git a/.github/workflows/_ci-relay.yml b/.github/workflows/_ci-relay.yml index 5a8011539ae..388d48dd940 100644 --- a/.github/workflows/_ci-relay.yml +++ b/.github/workflows/_ci-relay.yml @@ -35,7 +35,7 @@ jobs: desktop-e2e-relay: name: Desktop E2E Relay runs-on: ubuntu-latest - timeout-minutes: 30 + timeout-minutes: 45 if: inputs.lane == 'artifacts' && (github.event_name == 'push' || inputs.desktop || inputs.desktop_rust || inputs.rust) permissions: contents: read diff --git a/Cargo.lock b/Cargo.lock index 440ac5e4ffe..09d8faa9887 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -117,7 +117,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -128,7 +128,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -379,9 +379,9 @@ dependencies = [ [[package]] name = "async-utility" -version = "0.3.1" +version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a34a3b57207a7a1007832416c3e4862378c8451b4e8e093e436f48c2d3d2c151" +checksum = "188f83b9a198af8c336e505611edb00d6d2ac5c694241c5a4f9a12316938cfe9" dependencies = [ "futures-util", "gloo-timers", @@ -494,9 +494,9 @@ dependencies = [ [[package]] name = "aws-lc-rs" -version = "1.17.0" +version = "1.18.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ec2f1fc3ec205783a5da9a7e6c1509cc69dedf09a1949e412c1e18469326d00" +checksum = "b281d307588d634de920874890732659e2e7672f72b5e10e81badc1a8a83621e" dependencies = [ "aws-lc-sys", "untrusted 0.7.1", @@ -505,14 +505,15 @@ dependencies = [ [[package]] name = "aws-lc-sys" -version = "0.41.0" +version = "0.45.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1a2f9779ce85b93ab6170dd940ad0169b5766ff848247aff13bb788b832fe3f4" +checksum = "9bff6c3b54fad79a2e60b8102caf565819711497c1f5f092f49508e2f5c31b27" dependencies = [ "cc", "cmake", "dunce", "fs_extra", + "pkg-config", ] [[package]] @@ -1644,9 +1645,9 @@ dependencies = [ [[package]] name = "chacha20" -version = "0.10.0" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6f8d983286843e49675a4b7a2d174efe136dc93a18d69130dd18198a6c167601" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if 1.0.4", "cpufeatures 0.3.0", @@ -1773,7 +1774,7 @@ version = "3.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2369,7 +2370,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090" dependencies = [ "data-encoding", - "syn 2.0.117", + "syn 1.0.109", ] [[package]] @@ -2440,9 +2441,9 @@ checksum = "5729f5117e208430e437df2f4843f5e5952997175992d1414f94c57d61e270b4" [[package]] name = "der" -version = "0.8.0" +version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "71fd89660b2dc699704064e59e9dba0147b903e85319429e131620d022be411b" +checksum = "a878c850e9e421b20262e9b41f9c860e4785fa07541c266b62ff9d1ef998a80a" dependencies = [ "const-oid", "pem-rfc7468", @@ -2594,7 +2595,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -2781,7 +2782,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2992,7 +2993,7 @@ checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" dependencies = [ "futures-core", "futures-sink", - "spin 0.9.8", + "spin 0.9.9", ] [[package]] @@ -3003,7 +3004,7 @@ checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" dependencies = [ "futures-core", "futures-sink", - "spin 0.9.8", + "spin 0.9.9", ] [[package]] @@ -3089,7 +3090,7 @@ dependencies = [ "diatomic-waker", "futures-core", "pin-project-lite", - "spin 0.10.0", + "spin 0.10.1", ] [[package]] @@ -3209,7 +3210,7 @@ dependencies = [ "libc", "log", "rustversion", - "windows-link 0.2.1", + "windows-link 0.1.3", "windows-result 0.4.1", ] @@ -6091,7 +6092,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -7538,7 +7539,7 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d2e8e8bcc7961af1fdac401278c6a831614941f6164ee3bf4ce61b7edb162207" dependencies = [ - "chacha20 0.10.0", + "chacha20 0.10.2", "getrandom 0.4.3", "rand_core 0.10.1", ] @@ -8198,14 +8199,14 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] name = "rustls" -version = "0.23.42" +version = "0.23.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138" +checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634" dependencies = [ "aws-lc-rs", "log", @@ -8257,7 +8258,7 @@ dependencies = [ "security-framework 3.7.0", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -8268,9 +8269,9 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" [[package]] name = "rustls-webpki" -version = "0.103.13" +version = "0.103.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" +checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" dependencies = [ "aws-lc-rs", "ring", @@ -8526,7 +8527,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5b55fb86dfd3a2f5f76ea78310a88f96c4ea21a3031f8d212443d56123fd0521" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -9052,7 +9053,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -9074,18 +9075,18 @@ dependencies = [ [[package]] name = "spin" -version = "0.9.8" +version = "0.9.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e" dependencies = [ "lock_api", ] [[package]] name = "spin" -version = "0.10.0" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d5fe4ccb98d9c292d56fec89a5e07da7fc4cf0dc11e156b41793132775d3e591" +checksum = "023a211cb3138dbc438680b32560ad89f699977624c9f8dbb95a47d5b4c07dd3" [[package]] name = "spki" @@ -9680,7 +9681,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -9693,7 +9694,7 @@ dependencies = [ "parking_lot", "rustix 1.1.4", "signal-hook", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -10463,7 +10464,7 @@ checksum = "f2f6fb2847f6742cd76af783a2a2c49e9375d0a111c7bef6f71cd9e738c72d6e" dependencies = [ "memoffset", "tempfile", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -11016,7 +11017,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 6c272187f3c..cdb4471590e 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -4669,7 +4669,7 @@ fn handle_prompt_result( result .agent .state - .mark_scope_delivery_success(scope, false, event_ids); + .mark_scope_delivery_success(scope, false, event_ids, []); } } diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 250221badca..c188633bef7 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -20,7 +20,7 @@ //! `AcpClient` is NOT Clone — ownership moves out on claim and back on return. use std::cmp::Reverse; -use std::collections::{HashMap, HashSet}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -48,6 +48,13 @@ use crate::scope::SessionScope; /// the turn as "recently active" (eligible for requeue instead of dead-letter). const RECENT_ACTIVITY_WINDOW: Duration = Duration::from_secs(60); +/// Maximum canonical thread roots remembered by one live ACP session. +/// +/// Evicting the oldest root only causes a future prompt to include redundant +/// context again; it cannot drop user-authored context, so bounded fail-open +/// behavior is preferable to an ever-growing channel-policy session ledger. +const MAX_HYDRATED_THREAD_ROOTS_PER_SCOPE: usize = 1024; + // FlushBatch and BatchEvent derive Clone (added in queue.rs) so we can store // a recoverable copy in TaskMeta for panic recovery in Queue mode. @@ -110,6 +117,10 @@ pub struct ChannelDeliveryState { /// Buzz event IDs already delivered to this ACP session, either as trigger /// events or conversation context. pub delivered_event_ids: HashSet, + /// Canonical thread roots that have completed a successful context fetch in + /// this ACP session. Hydrated threads use bounded overfetch so exact event-ID + /// deduplication does not unnecessarily shrink the new-context window. + pub hydrated_thread_roots: VecDeque, } /// Per-channel session IDs, turn counters, and delivery state. @@ -222,10 +233,20 @@ impl SessionState { scope: SessionScope, standing_context_sent: bool, event_ids: impl IntoIterator, + hydrated_thread_roots: impl IntoIterator, ) { let delivery = self.deliveries.entry(scope).or_default(); delivery.standing_context_sent |= standing_context_sent; delivery.delivered_event_ids.extend(event_ids); + for root in hydrated_thread_roots { + if delivery.hydrated_thread_roots.contains(&root) { + continue; + } + if delivery.hydrated_thread_roots.len() >= MAX_HYDRATED_THREAD_ROOTS_PER_SCOPE { + delivery.hydrated_thread_roots.pop_front(); + } + delivery.hydrated_thread_roots.push_back(root); + } } #[cfg(test)] @@ -1170,7 +1191,7 @@ impl AgentPool { }; agent .state - .mark_scope_delivery_success(scope.clone(), false, [event_id]); + .mark_scope_delivery_success(scope.clone(), false, [event_id], []); true } @@ -2690,7 +2711,7 @@ pub async fn run_prompt_task( if !agent.has_system_prompt_support() { agent .state - .mark_scope_delivery_success(scope.clone(), true, []); + .mark_scope_delivery_success(scope.clone(), true, [], []); } let usage = agent.acp.take_turn_usage(); publish_agent_turn_metric( @@ -2816,6 +2837,10 @@ pub async fn run_prompt_task( // Event IDs represented by this prompt. Commit only after ACP reports a // successful turn; failed/cancelled prompts must be retryable without loss. let mut pending_delivered_event_ids = HashSet::new(); + // Thread hydration follows the same commit rule as event delivery. A + // failed or cancelled first turn must not make its retry assume that the + // provider retained any of that thread's context. + let mut pending_hydrated_thread_roots = HashSet::new(); let prompt_sections: Vec = if let Some(text) = prompt_text { // Heartbeats create their session before this point, so a Goose method-not-found // probe has already selected the correct framing for this process. @@ -2844,11 +2869,22 @@ pub async fn run_prompt_task( // reuse that exact typed result for prompt formatting. let channel_info = resolved_channel_info.clone(); - let conversation_context = if ctx.context_message_limit > 0 { - fetch_conversation_context(b, &channel_info, &ctx).await - } else { - None + let is_dm = channel_info + .as_ref() + .map(|info| info.channel_type == "dm") + .unwrap_or(false); + let context_target = resolve_context_target(b, is_dm); + let hydrated_thread_root = match &context_target { + ContextTarget::Thread(root) => Some(root), + ContextTarget::Dm | ContextTarget::None => None, }; + let thread_context_is_hydrated = hydrated_thread_root.is_some_and(|root| { + agent + .state + .deliveries + .get(&b.scope) + .is_some_and(|delivery| delivery.hydrated_thread_roots.contains(root)) + }); let rendered_batch_ids: HashSet = b .events .iter() @@ -2862,7 +2898,24 @@ pub async fn run_prompt_task( .map(|delivery| &delivery.delivered_event_ids) .cloned() .unwrap_or_default(); - let conversation_context_had_delivered_events = + let conversation_context = if ctx.context_message_limit > 0 { + fetch_conversation_context_for_target( + b.channel_id, + &context_target, + &ctx, + thread_context_is_hydrated, + &delivered_ids, + ) + .await + } else { + None + }; + if let Some(root) = + fetched_thread_root_to_hydrate(&context_target, conversation_context.as_ref(), is_dm) + { + pending_hydrated_thread_roots.insert(root); + } + let conversation_context_had_session_events = conversation_context.as_ref().is_some_and(|context| { conversation_context_event_ids(Some(context)) .iter() @@ -2901,7 +2954,7 @@ pub async fn run_prompt_task( huddle_instructions: standing.huddle_instructions, channel_info: channel_info.as_ref(), conversation_context: conversation_context.as_ref(), - conversation_context_had_delivered_events, + conversation_context_had_session_events, profile_lookup: profile_lookup.as_ref(), has_system_prompt_support: agent.has_system_prompt_support(), base_prompt: standing.base_prompt, @@ -3134,6 +3187,7 @@ pub async fn run_prompt_task( scope.clone(), standing_sent, &pending_delivered_event_ids, + &pending_hydrated_thread_roots, ); } apply_completed_before_control_signal( @@ -3177,6 +3231,7 @@ pub async fn run_prompt_task( scope.clone(), standing_sent, &pending_delivered_event_ids, + &pending_hydrated_thread_roots, ); } else if !agent.has_system_prompt_support() { agent.state.heartbeat_standing_context_sent = true; @@ -3824,14 +3879,18 @@ fn conversation_context_delta( triggering: &HashSet, ) -> Option { let filter = |messages: Vec| { - messages + let omitted_from_prior_session = messages + .iter() + .any(|message| !message.event_id.is_empty() && delivered.contains(&message.event_id)); + let messages = messages .into_iter() .filter(|message| { message.event_id.is_empty() || (!delivered.contains(&message.event_id) && !triggering.contains(&message.event_id)) }) - .collect::>() + .collect::>(); + (messages, omitted_from_prior_session) }; match context? { @@ -3841,12 +3900,12 @@ fn conversation_context_delta( root_present, truncated, } => { - let messages = filter(messages); + let (messages, omitted_from_prior_session) = filter(messages); (!messages.is_empty()).then_some(ConversationContext::Thread { messages, total, root_present, - truncated, + truncated: truncated || omitted_from_prior_session, }) } ConversationContext::Dm { @@ -3854,11 +3913,11 @@ fn conversation_context_delta( total, truncated, } => { - let messages = filter(messages); + let (messages, omitted_from_prior_session) = filter(messages); (!messages.is_empty()).then_some(ConversationContext::Dm { messages, total, - truncated, + truncated: truncated || omitted_from_prior_session, }) } } @@ -3886,29 +3945,28 @@ fn conversation_context_delta( /// The delivery-delta filter (`conversation_context_delta`) then removes any /// events this scope's live session already received, so subsequent turns /// deliver only intervening same-thread messages plus the trigger. -async fn fetch_conversation_context( - batch: &FlushBatch, - channel_info: &Option, +async fn fetch_conversation_context_for_target( + channel_id: Uuid, + target: &ContextTarget, ctx: &PromptContext, + overfetch_session_delta: bool, + delivered_ids: &HashSet, ) -> Option { let limit = ctx.context_message_limit; - let is_dm = channel_info - .as_ref() - .map(|ci| ci.channel_type == "dm") - .unwrap_or(false); - - match resolve_context_target(batch, is_dm) { + match target { ContextTarget::Thread(root_id) => { fetch_thread_context( - batch.channel_id, - &root_id, + channel_id, + root_id, limit, ctx.agent_keys.public_key(), &ctx.rest_client, + overfetch_session_delta, + delivered_ids, ) .await } - ContextTarget::Dm => fetch_dm_context(batch.channel_id, limit, &ctx.rest_client).await, + ContextTarget::Dm => fetch_dm_context(channel_id, limit, &ctx.rest_client).await, ContextTarget::None => None, } } @@ -3933,13 +3991,13 @@ enum ContextTarget { /// conversation history; a plain top-level channel message has none. fn resolve_context_target(batch: &FlushBatch, is_dm: bool) -> ContextTarget { if let Some(root_id) = batch.scope.root_event_id() { - return ContextTarget::Thread(root_id.to_string()); + return ContextTarget::Thread(root_id.to_ascii_lowercase()); } let Some(last_event) = batch.events.last() else { return ContextTarget::None; }; if let Some(root_id) = crate::queue::parse_thread_tags(&last_event.event).root_event_id { - return ContextTarget::Thread(root_id); + return ContextTarget::Thread(root_id.to_ascii_lowercase()); } if is_dm { return ContextTarget::Dm; @@ -3947,6 +4005,24 @@ fn resolve_context_target(batch: &FlushBatch, is_dm: bool) -> ContextTarget { ContextTarget::None } +/// Return the one canonical thread whose history this prompt successfully +/// fetched. This is deliberately narrower than the set of events rendered in +/// a merged channel-policy batch: only fetched context proves that a provider +/// session can use bounded overfetch for exact event-ID delta filtering later. +fn fetched_thread_root_to_hydrate( + target: &ContextTarget, + context: Option<&ConversationContext>, + is_dm: bool, +) -> Option { + if is_dm || !matches!(context, Some(ConversationContext::Thread { .. })) { + return None; + } + match target { + ContextTarget::Thread(root) => Some(root.clone()), + ContextTarget::Dm | ContextTarget::None => None, + } +} + /// Normalize AND validate a pubkey for the batch profile API request. /// Returns `None` for malformed input — only valid 64-char hex passes. /// See also: `normalize_lookup_key` in queue.rs (normalize-only, no validation). @@ -4104,16 +4180,20 @@ async fn fetch_prompt_profile_lookup( /// when the relay has more thread history, instead of reporting the capped page /// as the total. When the window is full, a best-effort `/count` attempts to /// improve that lower-bound total; because it is a separate racy request, the -/// result is clamped to the sentinel-proven minimum. The query also asks for the -/// agent's newest reply separately so the next prompt can include the agent's -/// own prior turn even in busy threads where the recent-message window would -/// otherwise push it out. +/// result is clamped to the sentinel-proven minimum. The query also asks for +/// the agent's newest reply separately so it survives a busy recent-message +/// window. A hydrated session overfetches a bounded 2x window so removing exact +/// event IDs already delivered to that session does not unnecessarily displace +/// new context. Author identity alone is never delivery evidence because +/// independent provider sessions can share one signing key. async fn fetch_thread_context( channel_id: Uuid, root_event_id: &str, limit: u32, agent_pubkey: nostr::PublicKey, rest: &RestClient, + overfetch_session_delta: bool, + delivered_ids: &HashSet, ) -> Option { fetch_thread_context_with( channel_id, @@ -4122,10 +4202,20 @@ async fn fetch_thread_context( agent_pubkey, |filters| async move { rest.query(&filters).await }, |filters| async move { rest.count(&filters).await }, + ThreadContextSessionDelta { + overfetch: overfetch_session_delta, + delivered_ids, + }, ) .await } +#[derive(Clone, Copy)] +struct ThreadContextSessionDelta<'a> { + overfetch: bool, + delivered_ids: &'a HashSet, +} + async fn fetch_thread_context_with( channel_id: Uuid, root_event_id: &str, @@ -4133,6 +4223,7 @@ async fn fetch_thread_context_with( agent_pubkey: nostr::PublicKey, query: Query, count: Count, + session_delta: ThreadContextSessionDelta<'_>, ) -> Option where Query: Fn(Vec) -> QueryFut, @@ -4161,6 +4252,7 @@ where // Three filters: (1) root event by ID, (2) recent replies with #e=root + // #h=channel plus a sentinel, and (3) the agent's newest reply for pinning. let root_filter = nostr::Filter::new().id(nostr::EventId::from_hex(root_event_id).ok()?); + let reply_fetch_limit = thread_reply_fetch_limit(limit, session_delta.overfetch); let replies_filter = nostr::Filter::new() .kinds([ nostr::Kind::Custom(buzz_core::kind::KIND_STREAM_MESSAGE as u16), @@ -4168,23 +4260,22 @@ where ]) .custom_tags(e_tag, [root_event_id]) .custom_tags(h_tag, [ch_str.as_str()]) - .limit(limit.saturating_add(1) as usize); + .limit(reply_fetch_limit.saturating_add(1) as usize); let agent_reply_filter = replies_filter.clone().author(agent_pubkey).limit(1); let context = fetch_with_retry(|| async { - match timeout( - CONTEXT_FETCH_TIMEOUT, - query(vec![ - root_filter.clone(), - replies_filter.clone(), - agent_reply_filter.clone(), - ]), - ) - .await - { - Ok(Ok(json)) => { - parse_nostr_thread_response_with_meta(json, root_event_id, limit, &agent_pubkey) - } + let mut filters = vec![root_filter.clone(), replies_filter.clone()]; + filters.push(agent_reply_filter.clone()); + match timeout(CONTEXT_FETCH_TIMEOUT, query(filters)).await { + Ok(Ok(json)) => parse_nostr_thread_response_with_meta( + json, + root_event_id, + limit, + &agent_pubkey, + session_delta.overfetch, + reply_fetch_limit, + session_delta.delivered_ids, + ), Ok(Err(e)) => { tracing::warn!( channel_id = %channel_id, @@ -4207,13 +4298,7 @@ where let mut parsed = context?; - if matches!( - parsed.context, - ConversationContext::Thread { - truncated: true, - .. - } - ) { + if parsed.needs_exact_count { let replies_count_filter = replies_filter.clone().limit(0); if let Some(total) = fetch_thread_total( channel_id, @@ -4280,6 +4365,21 @@ async fn fetch_dm_context( limit: u32, rest: &RestClient, ) -> Option { + fetch_dm_context_with(channel_id, limit, |filters| async move { + rest.query(&filters).await + }) + .await +} + +async fn fetch_dm_context_with( + channel_id: Uuid, + limit: u32, + query: Query, +) -> Option +where + Query: Fn(Vec) -> QueryFut, + QueryFut: std::future::Future>, +{ use nostr::{Alphabet, SingleLetterTag}; let h_tag = SingleLetterTag::lowercase(Alphabet::H); @@ -4293,12 +4393,7 @@ async fn fetch_dm_context( .limit(limit as usize); fetch_with_retry(|| async { - match timeout( - CONTEXT_FETCH_TIMEOUT, - rest.query(std::slice::from_ref(&filter)), - ) - .await - { + match timeout(CONTEXT_FETCH_TIMEOUT, query(vec![filter.clone()])).await { Ok(Ok(json)) => parse_nostr_dm_response(json, limit), Ok(Err(e)) => { tracing::warn!( @@ -4432,11 +4527,11 @@ fn json_to_context_message(obj: &serde_json::Value) -> Option { /// Parse a Nostr query response (array of events) into thread context. /// -/// Separates the root event (matching `root_event_id`) from replies, keeps the -/// newest `limit` replies returned by the sentinel query, then sorts the -/// displayed window chronologically for the prompt. If the agent's newest reply -/// is outside that window, keep it instead of the oldest displayed reply so the -/// next prompt always includes the agent's most recent prior turn. +/// Separates the root event (matching `root_event_id`) from replies, then sorts +/// the selected window chronologically for the prompt. Fresh sessions pin the +/// agent's newest reply inside the `limit`; hydrated sessions reserve the full +/// limit for events not yet delivered to this provider session and retain +/// already-delivered events only until the session delta accounts for them. #[cfg(test)] fn parse_nostr_thread_response( json: serde_json::Value, @@ -4444,13 +4539,30 @@ fn parse_nostr_thread_response( limit: u32, agent_pubkey: &nostr::PublicKey, ) -> Option { - parse_nostr_thread_response_with_meta(json, root_event_id, limit, agent_pubkey) - .map(|parsed| parsed.context) + parse_nostr_thread_response_with_meta( + json, + root_event_id, + limit, + agent_pubkey, + false, + thread_reply_fetch_limit(limit, false), + &HashSet::new(), + ) + .map(|parsed| parsed.context) } struct ParsedThreadContext { context: ConversationContext, root_present: bool, + needs_exact_count: bool, +} + +fn thread_reply_fetch_limit(limit: u32, overfetch_session_delta: bool) -> u32 { + if overfetch_session_delta { + limit.saturating_mul(2) + } else { + limit + } } fn parse_nostr_thread_response_with_meta( @@ -4458,6 +4570,9 @@ fn parse_nostr_thread_response_with_meta( root_event_id: &str, limit: u32, agent_pubkey: &nostr::PublicKey, + overfetch_session_delta: bool, + reply_fetch_limit: u32, + delivered_ids: &HashSet, ) -> Option { let events = json.as_array()?; let agent_pubkey_hex = agent_pubkey.to_hex(); @@ -4472,10 +4587,12 @@ fn parse_nostr_thread_response_with_meta( root_msg = Some(msg); } else if seen_reply_ids.insert(ev_id.to_string()) { let is_agent = msg.pubkey.eq_ignore_ascii_case(&agent_pubkey_hex); + let was_delivered = delivered_ids.contains(ev_id); reply_msgs.push(( ev_id.to_string(), ev.get("created_at").and_then(|v| v.as_u64()).unwrap_or(0), is_agent, + was_delivered, msg, )); } @@ -4483,51 +4600,77 @@ fn parse_nostr_thread_response_with_meta( } let root_present = root_msg.is_some(); - let fetched_total = reply_msgs.len() + usize::from(root_present); + let fetched_reply_count = reply_msgs.len(); + let fetched_total = fetched_reply_count + usize::from(root_present); let newest_agent_reply = reply_msgs .iter() - .filter(|(_, _, is_agent, _)| *is_agent) - .max_by_key(|(_, ts, _, _)| *ts) + .filter(|(_, _, is_agent, _, _)| *is_agent) + .max_by_key(|(_, ts, _, _, _)| *ts) .cloned(); - let truncated = reply_msgs.len() > limit as usize; - if truncated { + let relay_window_truncated = fetched_reply_count > reply_fetch_limit as usize; + let new_reply_count = reply_msgs + .iter() + .filter(|(_, _, _, was_delivered, _)| !was_delivered) + .count(); + let truncated = relay_window_truncated || new_reply_count > limit as usize; + + if !overfetch_session_delta && truncated { // The relay returns limited REQ results newest-first. Sort explicitly so // the sentinel we drop is the oldest reply in the fetched window, not an // arbitrary last element if the HTTP bridge ever changes iteration order. - reply_msgs.sort_by_key(|(_, ts, _, _)| Reverse(*ts)); + reply_msgs.sort_by_key(|(_, ts, _, _, _)| Reverse(*ts)); reply_msgs.truncate(limit as usize); + } else if overfetch_session_delta { + // Keep delivered replies until `conversation_context_delta` records + // that context was omitted, but count only new replies against the + // display budget. An unseen reply signed by this same agent remains new: + // it may have come from an independent heartbeat/provider session. + reply_msgs.sort_by_key(|(_, ts, _, _, _)| Reverse(*ts)); + let mut retained_new = 0usize; + reply_msgs.retain(|(_, _, _, was_delivered, _)| { + if *was_delivered { + true + } else if retained_new < limit as usize { + retained_new += 1; + true + } else { + false + } + }); } - if let Some(agent_reply) = newest_agent_reply { - let agent_reply_already_displayed = - reply_msgs.iter().any(|(id, _, _, _)| *id == agent_reply.0); + if let Some(agent_reply) = newest_agent_reply.filter(|reply| !reply.3) { + let agent_reply_already_displayed = reply_msgs + .iter() + .any(|(id, _, _, _, _)| *id == agent_reply.0); if !agent_reply_already_displayed { - reply_msgs.sort_by_key(|(_, ts, _, _)| *ts); - if let Some(oldest) = reply_msgs.first_mut() { + reply_msgs.sort_by_key(|(_, ts, _, _, _)| *ts); + if let Some(oldest) = reply_msgs + .iter_mut() + .find(|(_, _, _, was_delivered, _)| !was_delivered) + { *oldest = agent_reply; } } } // Sort displayed replies chronologically. - reply_msgs.sort_by_key(|(_, ts, _, _)| *ts); + reply_msgs.sort_by_key(|(_, ts, _, _, _)| *ts); let mut messages = Vec::new(); if let Some(root) = root_msg { messages.push(root); } - messages.extend(reply_msgs.into_iter().map(|(_, _, _, msg)| msg)); + messages.extend(reply_msgs.into_iter().map(|(_, _, _, _, msg)| msg)); if messages.is_empty() { return None; } - let total = if truncated { - fetched_total // all distinct fetched replies plus the root are proven visible history - } else { - messages.len() - }; + // Preserve the canonical fetched count even when session-delta filtering + // intentionally omits exact replies already delivered to this session. + let total = fetched_total; Some(ParsedThreadContext { context: ConversationContext::Thread { @@ -4537,6 +4680,7 @@ fn parse_nostr_thread_response_with_meta( truncated, }, root_present, + needs_exact_count: relay_window_truncated, }) } @@ -4712,6 +4856,7 @@ fn record_scope_delivery_success( scope: SessionScope, standing_context_sent: bool, event_ids: &HashSet, + hydrated_thread_roots: &HashSet, ) { tracing::info!( target: "pool::prompt", @@ -4722,6 +4867,7 @@ fn record_scope_delivery_success( scope, standing_context_sent, event_ids.iter().cloned(), + hydrated_thread_roots.iter().cloned(), ); } @@ -6096,6 +6242,10 @@ mod tests { assert_thread_count_filter(&filters, channel_id, root_id); std::future::ready(Ok(json!({ "count": 6 }))) }, + ThreadContextSessionDelta { + overfetch: false, + delivered_ids: &HashSet::new(), + }, ) .await .expect("thread context"); @@ -6148,6 +6298,10 @@ mod tests { agent.public_key(), move |_filters| std::future::ready(Ok(json.clone())), |_filters| std::future::ready(Ok(json!({ "count": 6 }))), + ThreadContextSessionDelta { + overfetch: false, + delivered_ids: &HashSet::new(), + }, ) .await .expect("thread context"); @@ -6202,6 +6356,10 @@ mod tests { agent.public_key(), move |_filters| std::future::ready(Ok(json.clone())), |_filters| std::future::ready(Ok(json!({ "count": 1 }))), + ThreadContextSessionDelta { + overfetch: false, + delivered_ids: &HashSet::new(), + }, ) .await .expect("thread context"); @@ -6255,6 +6413,10 @@ mod tests { agent.public_key(), move |_filters| std::future::ready(Ok(json.clone())), |_filters| std::future::ready(Err(crate::relay::RelayError::Http("boom".into()))), + ThreadContextSessionDelta { + overfetch: false, + delivered_ids: &HashSet::new(), + }, ) .await .expect("thread context"); @@ -6317,6 +6479,10 @@ mod tests { agent.public_key(), move |_filters| std::future::ready(Ok(json.clone())), |_filters| std::future::ready(Ok(json!({ "count": 3 }))), + ThreadContextSessionDelta { + overfetch: false, + delivered_ids: &HashSet::new(), + }, ) .await .expect("thread context"); @@ -6350,6 +6516,101 @@ mod tests { } } + #[tokio::test] + async fn hydrated_thread_retains_same_key_reply_from_another_session() { + let agent = Keys::generate(); + let agent_hex = agent.public_key().to_hex(); + let root_id = "1111111111111111111111111111111111111111111111111111111111111111"; + let channel_id = Uuid::new_v4(); + let json = json!([ + thread_event(root_id, "rootpub", "root", 1000), + thread_event( + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "humanpub", + "newer human reply", + 5000 + ), + thread_event( + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "humanpub", + "middle human reply", + 4000 + ), + thread_event( + "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc", + &agent_hex, + "recent agent reply already in provider history", + 6000 + ), + thread_event( + "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd", + &agent_hex, + "same-key reply from another provider session", + 5500 + ) + ]); + let delivered = HashSet::from([ + "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc".to_string(), + ]); + + let ctx = fetch_thread_context_with( + channel_id, + root_id, + 3, + agent.public_key(), + move |filters| { + assert_eq!( + filters.len(), + 3, + "same-key replies still require the agent pin query" + ); + let replies = serde_json::to_value(&filters[1]).expect("serialize replies filter"); + assert!(replies.get("authors").is_none()); + assert_eq!(replies.get("limit"), Some(&json!(7))); + std::future::ready(Ok(json.clone())) + }, + |_filters| -> std::future::Ready> { + panic!("session-delta omission alone must not issue /count") + }, + ThreadContextSessionDelta { + overfetch: true, + delivered_ids: &delivered, + }, + ) + .await + .expect("thread context"); + + let ctx = conversation_context_delta(Some(ctx), &delivered, &HashSet::new()) + .expect("human context remains after hydrated-session filtering"); + + let ConversationContext::Thread { + messages, + total, + truncated, + .. + } = ctx + else { + panic!("expected thread context"); + }; + assert_eq!(total, 5); + assert!( + truncated, + "omitted session history must be represented as a truncated context window" + ); + assert!(messages + .iter() + .any(|message| message.content == "newer human reply")); + assert!(messages + .iter() + .any(|message| message.content == "middle human reply")); + assert!(messages + .iter() + .all(|message| message.content != "recent agent reply already in provider history")); + assert!(messages + .iter() + .any(|message| message.content == "same-key reply from another provider session")); + } + #[tokio::test] async fn test_fetch_thread_context_uses_distinct_fetched_replies_as_minimum() { let agent = Keys::generate(); @@ -6391,6 +6652,10 @@ mod tests { agent.public_key(), move |_filters| std::future::ready(Ok(json.clone())), |_filters| std::future::ready(Err(crate::relay::RelayError::Http("boom".into()))), + ThreadContextSessionDelta { + overfetch: false, + delivered_ids: &HashSet::new(), + }, ) .await .expect("thread context"); @@ -6814,6 +7079,10 @@ done"# turn >= 2, "channel event IDs must commit only after ACP success" ); + assert!( + delivery.hydrated_thread_roots.is_empty(), + "a successful turn without fetched thread context must not hydrate a root" + ); agent = result.agent; } agent.acp.shutdown().await; @@ -6840,6 +7109,219 @@ done"# ); } + #[tokio::test] + async fn hydrated_thread_prompt_omits_agent_reply_but_keeps_new_human_context() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let channel_id = Uuid::new_v4(); + let agent_keys = Keys::generate(); + let human_keys = Keys::generate(); + let h_tag = Tag::parse(["h", channel_id.to_string().as_str()]).unwrap(); + let root = EventBuilder::new(Kind::Custom(9), "original question") + .tags([h_tag.clone()]) + .sign_with_keys(&human_keys) + .unwrap(); + let root_id = root.id.to_hex(); + let reply_tag = Tag::parse(["e", root_id.as_str(), "", "reply"]).unwrap(); + let agent_reply = EventBuilder::new( + Kind::Custom(9), + "agent reply already retained in the provider session", + ) + .tags([h_tag.clone(), reply_tag.clone()]) + .sign_with_keys(&agent_keys) + .unwrap(); + let intervening_human = EventBuilder::new(Kind::Custom(9), "new human context") + .tags([h_tag.clone(), reply_tag.clone()]) + .sign_with_keys(&human_keys) + .unwrap(); + let trigger = EventBuilder::new(Kind::Custom(9), "follow-up mention") + .tags([h_tag, reply_tag]) + .sign_with_keys(&human_keys) + .unwrap(); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind context server"); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let first_response_body = serde_json::to_string(&vec![root.clone(), agent_reply.clone()]) + .expect("serialize first context response"); + let response_body = serde_json::to_string(&vec![ + root.clone(), + agent_reply, + intervening_human, + trigger.clone(), + ]) + .unwrap(); + let server_root_id = root_id.clone(); + let server = tokio::spawn(async move { + let mut served_first_context = false; + while let Ok((mut socket, _)) = listener.accept().await { + let mut request = vec![0; 16 * 1024]; + let read = socket.read(&mut request).await.unwrap_or(0); + let request = String::from_utf8_lossy(&request[..read]); + let body = if request.contains(&server_root_id) && !served_first_context { + served_first_context = true; + &first_response_body + } else { + &response_body + }; + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), body + ); + let _ = socket.write_all(response.as_bytes()).await; + } + }); + + let capture = std::env::temp_dir().join(format!( + "buzz-acp-hydrated-thread-wire-{}.ndjson", + Uuid::new_v4() + )); + let quoted_capture = capture.to_string_lossy().replace('\'', "'\\''"); + let script = format!( + r#"count=0 +while IFS= read -r line; do + printf '%s\n' "$line" >> '{quoted_capture}' + printf '%s\n' "{{\"jsonrpc\":\"2.0\",\"id\":$count,\"result\":{{\"stopReason\":\"end_turn\"}}}}" + count=$((count + 1)) +done"# + ); + let acp = AcpClient::spawn("bash", &["-c".into(), script], &[], false) + .await + .expect("spawn wire-capture ACP"); + let scope = SessionScope::Thread { + channel_id, + root_event_id: root_id.clone(), + }; + let mut agent = OwnedAgent { + index: 0, + acp, + state: SessionState::default(), + model_capabilities: None, + desired_model: None, + model_overridden: false, + desired_model_request_id: None, + desired_model_pending_ack: false, + startup_effort: None, + agent_name: "legacy-test-agent".into(), + goose_system_prompt_supported: None, + protocol_version: 1, + }; + agent + .state + .sessions + .insert(scope.clone(), "live-session".into()); + agent + .state + .deliveries + .insert(scope.clone(), ChannelDeliveryState::default()); + + let make_context = |context_message_limit| { + let mut ctx = make_prompt_context_impl(&agent_keys, None); + ctx.context_message_limit = context_message_limit; + ctx.rest_client.base_url = base_url.clone(); + ctx.channel_info = ChannelInfoResolver::new( + HashMap::from([( + channel_id, + crate::relay::ChannelInfo { + name: "test-thread".into(), + channel_type: "stream".into(), + description: None, + }, + )]), + RestClient { + http: reqwest::Client::new(), + base_url: base_url.clone(), + keys: agent_keys.clone(), + auth_tag_json: None, + }, + ); + ctx + }; + let first_batch = FlushBatch { + channel_id, + scope: scope.clone(), + events: vec![crate::queue::BatchEvent { + event: root, + prompt_tag: "@mention".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + let follow_up_batch = FlushBatch { + channel_id, + scope: scope.clone(), + events: vec![crate::queue::BatchEvent { + event: trigger, + prompt_tag: "@mention".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + let (result_tx, mut result_rx) = mpsc::unbounded_channel(); + run_prompt_task( + agent, + Some(first_batch), + None, + Arc::new(make_context(10)), + result_tx.clone(), + None, + "first-turn".into(), + ) + .await; + let first_result = result_rx.recv().await.expect("first prompt result"); + assert!(matches!( + first_result.outcome, + PromptOutcome::Ok(StopReason::EndTurn) + )); + assert!(first_result.agent.state.deliveries[&scope] + .hydrated_thread_roots + .contains(&root_id)); + + run_prompt_task( + first_result.agent, + Some(follow_up_batch), + None, + Arc::new(make_context(10)), + result_tx, + None, + "follow-up-turn".into(), + ) + .await; + let mut result = result_rx.recv().await.expect("prompt result"); + assert!(matches!( + result.outcome, + PromptOutcome::Ok(StopReason::EndTurn) + )); + assert!(result.agent.state.deliveries[&scope] + .hydrated_thread_roots + .contains(&root_id)); + result.agent.acp.shutdown().await; + server.abort(); + + let requests: Vec = std::fs::read_to_string(&capture) + .expect("read prompt capture") + .lines() + .map(|line| serde_json::from_str(line).expect("captured prompt JSON")) + .collect(); + std::fs::remove_file(&capture).expect("remove prompt capture"); + assert_eq!(requests.len(), 2); + let wire = requests[1]["params"]["prompt"] + .as_array() + .expect("prompt blocks") + .iter() + .filter_map(|block| block["text"].as_str()) + .collect::>() + .join("\n"); + assert!(wire.contains("new human context")); + assert!(!wire.contains("agent reply already retained in the provider session")); + assert!(wire.contains("follow-up mention")); + assert!(wire.contains("truncated=\"true\"")); + assert!(wire.contains("buzz messages thread")); + } + #[tokio::test] async fn merged_cancel_prompt_commits_and_deduplicates_all_rendered_event_ids() { use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -7174,15 +7656,37 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" let delivery = state.deliveries.get(&conv(channel)).unwrap(); assert!(!delivery.standing_context_sent); assert!(delivery.delivered_event_ids.is_empty()); + assert!(delivery.hydrated_thread_roots.is_empty()); state.mark_scope_delivery_success( conv(channel), true, ["trigger".to_string(), "context".to_string()], + ["thread-root".to_string()], ); let delivery = state.deliveries.get(&conv(channel)).unwrap(); assert!(delivery.standing_context_sent); assert_eq!(delivery.delivered_event_ids.len(), 2); + assert!(delivery + .hydrated_thread_roots + .iter() + .any(|root| root == "thread-root")); + } + + #[test] + fn delivery_state_bounds_hydrated_thread_roots() { + let channel = Uuid::new_v4(); + let mut state = SessionState::default(); + let roots = (0..=MAX_HYDRATED_THREAD_ROOTS_PER_SCOPE) + .map(|index| format!("root-{index}")) + .collect::>(); + + state.mark_scope_delivery_success(conv(channel), false, [], roots); + + let hydrated = &state.deliveries[&conv(channel)].hydrated_thread_roots; + assert_eq!(hydrated.len(), MAX_HYDRATED_THREAD_ROOTS_PER_SCOPE); + assert!(!hydrated.contains(&"root-0".to_string())); + assert!(hydrated.contains(&format!("root-{MAX_HYDRATED_THREAD_ROOTS_PER_SCOPE}"))); } #[test] @@ -7190,7 +7694,12 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" let channel = Uuid::new_v4(); let mut state = SessionState::default(); state.sessions.insert(conv(channel), "old-session".into()); - state.mark_scope_delivery_success(conv(channel), true, ["old-event".to_string()]); + state.mark_scope_delivery_success( + conv(channel), + true, + ["old-event".to_string()], + ["old-root".to_string()], + ); assert!(state.invalidate_channel(&channel) > 0); assert!(!state.deliveries.contains_key(&conv(channel))); @@ -7202,6 +7711,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" let delivery = state.deliveries.get(&conv(channel)).unwrap(); assert!(!delivery.standing_context_sent); assert!(delivery.delivered_event_ids.is_empty()); + assert!(delivery.hydrated_thread_roots.is_empty()); } #[test] @@ -7231,13 +7741,144 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" assert_eq!(messages.len(), 1); assert_eq!(messages[0].event_id, "new"); assert_eq!(total, 3); - assert!(!truncated); + assert!(truncated); assert!(root_present); } _ => panic!("expected thread context"), } } + #[tokio::test] + async fn fresh_thread_trigger_dedup_preserves_complete_context_hint() { + let channel_id = Uuid::new_v4(); + let agent = Keys::generate(); + let human = Keys::generate(); + let h_tag = Tag::parse(["h", channel_id.to_string().as_str()]).unwrap(); + let root = EventBuilder::new(Kind::Custom(9), "earlier thread root") + .tags([h_tag.clone()]) + .sign_with_keys(&human) + .unwrap(); + let root_id = root.id.to_hex(); + let reply_tag = Tag::parse(["e", root_id.as_str(), "", "reply"]).unwrap(); + let trigger = EventBuilder::new(Kind::Custom(9), "fresh thread trigger") + .tags([h_tag, reply_tag]) + .sign_with_keys(&human) + .unwrap(); + let trigger_id = trigger.id.to_hex(); + let response = serde_json::to_value(vec![root, trigger.clone()]).unwrap(); + + let context = fetch_thread_context_with( + channel_id, + &root_id, + 10, + agent.public_key(), + move |_filters| std::future::ready(Ok(response.clone())), + |_filters| -> std::future::Ready> { + panic!("a complete thread window must not issue /count") + }, + ThreadContextSessionDelta { + overfetch: false, + delivered_ids: &HashSet::new(), + }, + ) + .await + .expect("thread context"); + let context = conversation_context_delta( + Some(context), + &HashSet::new(), + &HashSet::from([trigger_id]), + ) + .expect("root remains after trigger deduplication"); + let batch = FlushBatch { + channel_id, + scope: SessionScope::Thread { + channel_id, + root_event_id: root_id, + }, + events: vec![crate::queue::BatchEvent { + event: trigger, + prompt_tag: "@mention".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + + let wire = crate::queue::format_prompt( + &batch, + &crate::queue::FormatPromptArgs { + conversation_context: Some(&context), + ..Default::default() + }, + ) + .join("\n"); + assert!(wire.contains("")); + assert!(wire.contains("Thread context included below.")); + assert!(!wire.contains("for full history if truncated")); + assert!(wire.contains("fresh thread trigger")); + } + + #[tokio::test] + async fn fresh_dm_trigger_dedup_preserves_complete_context_hint() { + let channel_id = Uuid::new_v4(); + let human = Keys::generate(); + let h_tag = Tag::parse(["h", channel_id.to_string().as_str()]).unwrap(); + let earlier = EventBuilder::new(Kind::Custom(9), "earlier DM message") + .tags([h_tag.clone()]) + .sign_with_keys(&human) + .unwrap(); + let trigger = EventBuilder::new(Kind::Custom(9), "fresh DM trigger") + .tags([h_tag]) + .sign_with_keys(&human) + .unwrap(); + let trigger_id = trigger.id.to_hex(); + let response = serde_json::to_value(vec![earlier, trigger.clone()]).unwrap(); + + let context = fetch_dm_context_with(channel_id, 10, move |_filters| { + std::future::ready(Ok(response.clone())) + }) + .await + .expect("DM context"); + let context = conversation_context_delta( + Some(context), + &HashSet::new(), + &HashSet::from([trigger_id]), + ) + .expect("earlier DM remains after trigger deduplication"); + let batch = FlushBatch { + channel_id, + scope: SessionScope::Conversation { channel_id }, + events: vec![crate::queue::BatchEvent { + event: trigger, + prompt_tag: "@mention".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + let channel_info = PromptChannelInfo { + name: "test-dm".into(), + channel_type: "dm".into(), + ..Default::default() + }; + + let wire = crate::queue::format_prompt( + &batch, + &crate::queue::FormatPromptArgs { + channel_info: Some(&channel_info), + conversation_context: Some(&context), + ..Default::default() + }, + ) + .join("\n"); + assert!( + wire.contains("") + ); + assert!(wire.contains("Conversation context included below.")); + assert!(!wire.contains("for full history if truncated")); + assert!(wire.contains("fresh DM trigger")); + } + #[test] fn conversation_context_delta_returns_none_when_no_new_events_remain() { let delivered = HashSet::from(["old".to_string()]); @@ -7263,6 +7904,63 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" ); } + #[test] + fn conversation_context_delta_preserves_same_author_message_not_delivered_to_session() { + let agent = Keys::generate(); + let human = Keys::generate(); + let context = ConversationContext::Thread { + messages: vec![ + ContextMessage { + event_id: "session-event".into(), + pubkey: agent.public_key().to_hex().to_ascii_uppercase(), + timestamp: "2026-09-11T20:23:47Z".into(), + content: "agent reply already retained by ACP".into(), + }, + ContextMessage { + event_id: "heartbeat-event".into(), + pubkey: agent.public_key().to_hex(), + timestamp: "2026-09-11T20:24:00Z".into(), + content: "same-key reply from another session".into(), + }, + ContextMessage { + event_id: "human-event".into(), + pubkey: human.public_key().to_hex(), + timestamp: "2026-09-11T20:24:52Z".into(), + content: "intervening human reply".into(), + }, + ContextMessage { + event_id: "legacy-event".into(), + pubkey: "unknown".into(), + timestamp: "2026-09-11T20:24:53Z".into(), + content: "missing valid author metadata".into(), + }, + ], + total: 4, + root_present: true, + truncated: false, + }; + + let delivered = HashSet::from(["session-event".to_string()]); + let delta = conversation_context_delta(Some(context), &delivered, &HashSet::new()) + .expect("human context remains"); + let ConversationContext::Thread { messages, .. } = delta else { + panic!("expected thread context"); + }; + assert_eq!(messages.len(), 3); + assert!(messages + .iter() + .all(|message| message.content != "agent reply already retained by ACP")); + assert!(messages + .iter() + .any(|message| message.content == "intervening human reply")); + assert!(messages + .iter() + .any(|message| message.content == "same-key reply from another session")); + assert!(messages + .iter() + .any(|message| message.content == "missing valid author metadata")); + } + #[test] fn test_json_to_context_message_missing_pubkey_uses_default() { let obj = json!({ "content": "hello" }); @@ -7320,6 +8018,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" ChannelDeliveryState { standing_context_sent: true, delivered_event_ids: HashSet::from(["event-a".into()]), + hydrated_thread_roots: VecDeque::from(["root-a".into()]), }, ); s.deliveries.insert( @@ -7327,6 +8026,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" ChannelDeliveryState { standing_context_sent: true, delivered_event_ids: HashSet::from(["event-b".into()]), + hydrated_thread_roots: VecDeque::from(["root-b".into()]), }, ); s.heartbeat_session = Some("sess-hb".into()); @@ -7451,6 +8151,54 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" assert_eq!(resolve_context_target(&batch, false), ContextTarget::None); } + #[test] + fn merged_batch_hydrates_only_successfully_fetched_non_dm_target() { + let channel = Uuid::new_v4(); + let cancelled_root = "a".repeat(64); + let fetched_root = "b".repeat(64); + let cancelled = signed_event_with_tags(vec![ + vec!["e".into(), cancelled_root, String::new(), "root".into()], + vec!["e".into(), "c".repeat(64), String::new(), "reply".into()], + ]); + let current = signed_event_with_tags(vec![ + vec![ + "e".into(), + fetched_root.clone(), + String::new(), + "root".into(), + ], + vec!["e".into(), "d".repeat(64), String::new(), "reply".into()], + ]); + let mut batch = batch_with_scope(conv(channel), current); + batch.cancelled_events.push(crate::queue::BatchEvent { + event: cancelled, + prompt_tag: "cancelled".into(), + received_at: std::time::Instant::now(), + }); + let target = resolve_context_target(&batch, false); + let context = ConversationContext::Thread { + messages: vec![context_message(&fetched_root, "thread root")], + total: 1, + root_present: true, + truncated: false, + }; + + assert_eq!( + fetched_thread_root_to_hydrate(&target, Some(&context), false), + Some(fetched_root) + ); + assert_eq!(fetched_thread_root_to_hydrate(&target, None, false), None); + assert_eq!( + fetched_thread_root_to_hydrate(&target, Some(&context), true), + None, + "DM conversations do not use per-thread hydration" + ); + assert_eq!( + fetched_thread_root_to_hydrate(&ContextTarget::None, Some(&context), false), + None + ); + } + #[test] fn context_target_dm_nonreply_is_dm_history() { let ch = Uuid::new_v4(); diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index a419b78178d..7fbfc18696a 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -1622,8 +1622,8 @@ fn format_context_hints( ); let complete_conversation_context = conversation_context_status == ConversationContextStatus::Complete; - let conversation_context_had_delivered_events = - conversation_context_status == ConversationContextStatus::PreviouslyDelivered; + let conversation_context_had_session_events = + conversation_context_status == ConversationContextStatus::PreviouslyAvailable; // 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. @@ -1639,10 +1639,10 @@ fn format_context_hints( "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." - } else if conversation_context_had_delivered_events && is_reply { - "Earlier thread context was already delivered in this session. Use `buzz messages thread --channel --event ` to re-read the reply chain." - } else if conversation_context_had_delivered_events { - "Earlier conversation context was already delivered in this session. Use `buzz messages get --channel ` to re-read it." + } else if conversation_context_had_session_events && is_reply { + "Earlier thread context is already available in this session. Use `buzz messages thread --channel --event ` to re-read the reply chain." + } else if conversation_context_had_session_events { + "Earlier conversation context is already available in this session. Use `buzz messages get --channel ` to re-read it." } else if is_reply { "Use `buzz messages thread --channel --event ` to fetch the reply chain." } else { @@ -1675,8 +1675,8 @@ fn format_context_hints( "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." + } else if conversation_context_had_session_events { + "Earlier thread context is already available in this session. Use `buzz messages thread --channel --event ` to re-read it." } else { "Use `buzz messages thread --channel --event ` to fetch thread context." }; @@ -1729,7 +1729,7 @@ fn format_context_hints( enum ConversationContextStatus { Complete, Included, - PreviouslyDelivered, + PreviouslyAvailable, Absent, } @@ -1776,7 +1776,7 @@ fn conversation_context_covers_batch( fn conversation_context_status( batch: &FlushBatch, conversation_context: Option<&ConversationContext>, - conversation_context_had_delivered_events: bool, + conversation_context_had_session_events: bool, ) -> ConversationContextStatus { let window_is_complete = matches!( conversation_context, @@ -1793,13 +1793,13 @@ fn conversation_context_status( if window_is_complete && conversation_context_covers_batch(batch, conversation_context) - && !conversation_context_had_delivered_events + && !conversation_context_had_session_events { ConversationContextStatus::Complete } else if conversation_context.is_some() { ConversationContextStatus::Included - } else if conversation_context_had_delivered_events { - ConversationContextStatus::PreviouslyDelivered + } else if conversation_context_had_session_events { + ConversationContextStatus::PreviouslyAvailable } else { ConversationContextStatus::Absent } @@ -1859,9 +1859,10 @@ pub struct FormatPromptArgs<'a> { pub huddle_instructions: Option<&'a str>, pub channel_info: Option<&'a PromptChannelInfo>, pub conversation_context: Option<&'a ConversationContext>, - /// True when delivery-delta filtering removed at least one event that this - /// live session had already received. Trigger-only context does not set it. - pub conversation_context_had_delivered_events: bool, + /// True when delta filtering removed context already available to this + /// live session, either as prior input or as the agent's own reply. + /// Trigger-only context does not set it. + pub conversation_context_had_session_events: bool, pub profile_lookup: Option<&'a PromptProfileLookup>, /// When true, base_prompt and system_prompt are delivered via the system /// role (session/new) and omitted from the user message. When false @@ -2056,7 +2057,7 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec