mirror of
https://github.com/block/buzz.git
synced 2026-11-05 17:01:35 +00:00
fix(acp): sort flushed batch events chronologically so replies thread under the right message (#1872)
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -314,7 +314,7 @@ impl EventQueue {
|
||||
// Drain up to MAX_BATCH_EVENTS; leave any remainder in the queue.
|
||||
let queue = self.queues.entry(channel_id).or_default();
|
||||
let drain_count = MAX_BATCH_EVENTS.min(queue.len());
|
||||
let events: Vec<BatchEvent> = queue
|
||||
let mut events: Vec<BatchEvent> = queue
|
||||
.drain(..drain_count)
|
||||
.map(|qe| BatchEvent {
|
||||
event: qe.event,
|
||||
@@ -322,6 +322,11 @@ impl EventQueue {
|
||||
received_at: qe.received_at,
|
||||
})
|
||||
.collect();
|
||||
// Relay replay delivers stored events newest-first (`ORDER BY
|
||||
// created_at DESC`), but batch consumers — `format_prompt` scope and
|
||||
// reply-anchor selection — require the LAST event to be the newest.
|
||||
// Stable sort: same-second events keep delivery order.
|
||||
events.sort_by_key(|be| be.event.created_at);
|
||||
|
||||
// Remove the queue entry if now empty.
|
||||
if self.queues.get(&channel_id).is_some_and(|q| q.is_empty()) {
|
||||
@@ -1592,7 +1597,7 @@ pub(crate) fn native_steer_framing() -> (&'static str, &'static str) {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use nostr::{EventBuilder, Keys, Kind};
|
||||
use nostr::{EventBuilder, Keys, Kind, Timestamp};
|
||||
use std::time::Duration;
|
||||
|
||||
/// Build a test event with the given content and kind.
|
||||
@@ -1624,6 +1629,26 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
/// Build a QueuedEvent with an explicit Nostr `created_at` timestamp.
|
||||
fn make_queued_created_at(
|
||||
channel_id: Uuid,
|
||||
content: &str,
|
||||
created_at_secs: u64,
|
||||
) -> QueuedEvent {
|
||||
let keys = Keys::generate();
|
||||
let event = EventBuilder::new(Kind::Custom(9), content)
|
||||
.custom_created_at(Timestamp::from(created_at_secs))
|
||||
.tags([])
|
||||
.sign_with_keys(&keys)
|
||||
.unwrap();
|
||||
QueuedEvent {
|
||||
channel_id,
|
||||
event,
|
||||
received_at: Instant::now(),
|
||||
prompt_tag: "test".into(),
|
||||
}
|
||||
}
|
||||
|
||||
fn pending_count(q: &EventQueue) -> usize {
|
||||
q.queues.values().map(|q| q.len()).sum()
|
||||
}
|
||||
@@ -1722,6 +1747,34 @@ mod tests {
|
||||
assert_eq!(q.queues.len(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_flush_orders_replayed_events_chronologically() {
|
||||
// Relay replay after a reconnect/restart delivers stored events newest
|
||||
// first (`ORDER BY created_at DESC`), but `format_prompt` derives the
|
||||
// reply anchor and scope from the LAST batch event on the assumption
|
||||
// that it is the newest. The drained batch must therefore be
|
||||
// chronological regardless of delivery order.
|
||||
let mut q = EventQueue::new(DedupMode::Queue);
|
||||
let ch = Uuid::new_v4();
|
||||
|
||||
q.push(make_queued_created_at(ch, "newest", 2_000));
|
||||
q.push(make_queued_created_at(ch, "oldest", 1_000));
|
||||
|
||||
let batch = q.flush_next().expect("should return batch");
|
||||
assert_eq!(batch.events.len(), 2);
|
||||
assert_eq!(batch.events[0].event.content, "oldest");
|
||||
assert_eq!(batch.events[1].event.content, "newest");
|
||||
|
||||
// The rendered prompt's reply anchor must cite the newest event, so
|
||||
// the agent's reply threads under the message it is responding to.
|
||||
let newest_id = batch.events[1].event.id.to_hex();
|
||||
let prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n");
|
||||
assert!(
|
||||
prompt.contains(&format!("--reply-to {newest_id}")),
|
||||
"reply anchor must target the newest event; prompt was:\n{prompt}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_fifo_fairness_picks_oldest_channel() {
|
||||
let mut q = EventQueue::new(DedupMode::Queue);
|
||||
|
||||
Reference in New Issue
Block a user