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(acp): flush prepared native steer fallback

Browse Source 
Signed-off-by: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz>
This commit is contained in:
Brother Darryl
2026-08-10 16:27:58 -04:00
parent 12d94766ab
commit 5fced1e6e0
2 changed files with 223 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-acp/src/lib.rs
+168 -23
View File
@@ -2622,31 +2622,41 @@ async fn tokio_main() -> Result<()> {
if queue::edit_target_id(&event_for_steer).is_some() {
let event_id = event_for_steer.id.to_hex();
let channel_id = buzz_event.channel_id;
let membership_generation = reserve_native_edit_preparation(
match reserve_native_edit_preparation(
&mut queue,
&membership_generations,
channel_id,
&event_id,
)
.expect("accepted edit must still be queued");
let tx = native_steer_tx.clone();
let ctx = Arc::clone(&ctx);
tokio::spawn(async move {
let prompt_blocks = pool::format_native_steer_prompt(
channel_id,
event_for_steer.clone(),
prompt_tag_for_steer,
&ctx,
)
.await;
let _ = tx.send(NativeSteerPrepared {
channel_id,
membership_generation,
event: event_for_steer,
prompt_blocks,
});
});
true
) {
Some(membership_generation) => {
let tx = native_steer_tx.clone();
let ctx = Arc::clone(&ctx);
tokio::spawn(async move {
let prompt_blocks = pool::format_native_steer_prompt(
channel_id,
event_for_steer.clone(),
prompt_tag_for_steer,
&ctx,
)
.await;
let _ = tx.send(NativeSteerPrepared {
channel_id,
membership_generation,
event: event_for_steer,
prompt_blocks,
});
});
true
}
None => {
tracing::warn!(
%channel_id,
%event_id,
"native edit steer preparation could not reserve queued event; falling back to cancel+merge"
);
false
}
}
} else {
let prompt_blocks =
pool::format_native_steer_prompt_sync(
@@ -2883,8 +2893,15 @@ async fn tokio_main() -> Result<()> {
prompt_blocks,
&steer_ack_tx,
) {
queue.release_native_steer(channel_id, &event.id.to_hex());
signal_in_flight_task(&mut pool, channel_id, ControlSignal::Steer);
release_prepared_native_steer_fallback(
&mut pool,
&mut queue,
channel_id,
&event.id.to_hex(),
&ctx,
&mut last_activity,
&mut typing_channels,
);
}
}
Some(PoolEvent::SteerAck(SteerAckEvent {
@@ -3360,6 +3377,26 @@ fn try_native_steer(
}
}
/// Recover a prepared edit whose native steer could not be accepted. The edit
/// was withheld before asynchronous preparation, so release it, request the
/// universal cancel+merge fallback, and immediately try dispatch in case the
/// original turn already ended.
fn release_prepared_native_steer_fallback(
pool: &mut AgentPool,
queue: &mut EventQueue,
channel_id: Uuid,
event_id: &str,
ctx: &Arc<PromptContext>,
last_activity: &mut tokio::time::Instant,
typing_channels: &mut HashMap<Uuid, ThreadTags>,
) {
queue.release_native_steer(channel_id, event_id);
signal_in_flight_task(pool, channel_id, ControlSignal::Steer);
for (channel_id, thread_tags) in dispatch_pending(pool, queue, ctx, last_activity) {
typing_channels.insert(channel_id, thread_tags);
}
}
// ── dispatch_pending ──────────────────────────────────────────────────────────
/// Flush queued work to available agents.
@@ -8288,6 +8325,114 @@ mod native_edit_membership_lifecycle_tests {
.expect("signed edit")
}
async fn dummy_agent(index: usize) -> OwnedAgent {
OwnedAgent {
index,
acp: AcpClient::spawn("cat", &[], &[], false)
.await
.expect("spawn inert agent"),
state: Default::default(),
model_capabilities: None,
desired_model: None,
model_overridden: false,
agent_name: "unknown".into(),
protocol_version: 1,
goose_system_prompt_supported: None,
}
}
fn prompt_context() -> Arc<PromptContext> {
let keys = Keys::generate();
let rest_client = relay::RestClient {
http: reqwest::Client::new(),
base_url: "http://127.0.0.1:0".into(),
keys: keys.clone(),
auth_tag_json: None,
};
Arc::new(PromptContext {
mcp_servers: vec![],
initial_message: None,
idle_timeout: Duration::from_secs(60),
max_turn_duration: Duration::from_secs(120),
turn_liveness_interval: Duration::ZERO,
dedup_mode: DedupMode::Queue,
system_prompt: None,
session_title: None,
team_instructions: None,
heartbeat_prompt: None,
base_prompt: None,
cwd: ".".into(),
channel_info: pool::ChannelInfoResolver::new(HashMap::new(), rest_client.clone()),
rest_client,
context_message_limit: 0,
max_turns_per_session: 0,
permission_mode: config::PermissionMode::Default,
agent_keys: keys,
agent_owner_pubkey: None,
memory_enabled: false,
harness_name: "goose".into(),
relay_url: "ws://127.0.0.1:0".into(),
})
}
#[tokio::test]
async fn prepared_native_steer_rejection_releases_and_dispatches_immediately() {
let channel_id = Uuid::new_v4();
let event = edit_event(&"ab".repeat(32));
let event_id = event.id.to_hex();
let mut queue = EventQueue::new(config::DedupMode::Queue);
assert!(queue.push(QueuedEvent {
channel_id,
event,
received_at: std::time::Instant::now(),
prompt_tag: "@mention".into(),
}));
assert!(queue.mark_native_steer_pending(channel_id, &event_id));
// The original turn completed while edit enrichment ran. The native
// transport rejection must synchronously release and re-dispatch the
// edit; no maintenance tick may be needed to observe the work.
let mut pool = AgentPool::from_slots(vec![Some(dummy_agent(0).await)]);
let mut last_activity = tokio::time::Instant::now();
let mut typing_channels = HashMap::new();
release_prepared_native_steer_fallback(
&mut pool,
&mut queue,
channel_id,
&event_id,
&prompt_context(),
&mut last_activity,
&mut typing_channels,
);
assert!(typing_channels.contains_key(&channel_id));
assert_eq!(
pool.task_map().len(),
1,
"released edit dispatched exactly once"
);
assert!(
queue.flush_next().is_none(),
"dispatch consumed the released edit"
);
}
#[test]
fn unreservable_edit_leaves_universal_fallback_work_queued() {
let channel_id = Uuid::new_v4();
let event = edit_event(&"ab".repeat(32));
let event_id = event.id.to_hex();
let mut queue = EventQueue::new(config::DedupMode::Queue);
let generations = HashMap::new();
assert_eq!(
reserve_native_edit_preparation(&mut queue, &generations, channel_id, &event_id),
None,
"a missing queue entry must not panic or claim a native attempt"
);
assert!(queue.flush_next().is_none());
}
#[test]
fn late_prepared_edit_is_discarded_across_remove_then_readd() {
let channel_id = Uuid::new_v4();
crates/buzz-acp/src/queue.rs
+55 -12
View File
@@ -96,6 +96,7 @@ pub struct FlushBatch {
/// ```text
/// State:
/// queues: Map<channel_id, VecDeque<QueuedEvent>> (capped at MAX_PENDING_PER_CHANNEL)
/// withheld_native_steer: Map<channel_id, Vec<QueuedEvent>> (native reservations)
/// in_flight_channels: HashSet<Uuid>
/// in_flight_deadlines: Map<channel_id, Instant> (auto-expire after in_flight_deadline)
/// retry_after: Map<channel_id, Instant>
@@ -106,8 +107,8 @@ pub struct FlushBatch {
/// push(event):
/// if dedup_mode == Drop AND in_flight_channels.contains(event.channel_id):
/// debug log + discard
/// else if queues[channel].len() >= MAX_PENDING_PER_CHANNEL:
/// drop oldest (pop_front), warn, push_back new event
/// else if queued + withheld events for channel reach MAX_PENDING_PER_CHANNEL:
/// drop oldest queued event, or reject the new event if all slots are withheld
/// else:
/// queues[event.channel_id].push_back(event)
///
@@ -237,17 +238,39 @@ impl EventQueue {
);
return false;
}
let queue = self.queues.entry(event.channel_id).or_default();
// Enforce per-channel depth cap: drop oldest to make room.
if queue.len() >= MAX_PENDING_PER_CHANNEL {
queue.pop_front();
tracing::warn!(
channel_id = %event.channel_id,
limit = MAX_PENDING_PER_CHANNEL,
"queue depth cap reached — dropped oldest event"
);
let channel_id = event.channel_id;
let withheld = self
.withheld_native_steer
.get(&channel_id)
.map_or(0, Vec::len);
let queued = self.queues.get(&channel_id).map_or(0, VecDeque::len);
// A native-steer reservation owns its event until its asynchronous
// attempt resolves, so it counts toward the per-channel cap too.
// Never evict a withheld reservation: its completion may still steer.
// If every slot is reserved, reject the new event rather than creating
// an unbounded preparation task or violating exact-once delivery.
if queued + withheld >= MAX_PENDING_PER_CHANNEL {
if self
.queues
.get_mut(&channel_id)
.and_then(VecDeque::pop_front)
.is_some()
{
tracing::warn!(
%channel_id,
limit = MAX_PENDING_PER_CHANNEL,
"queue depth cap reached — dropped oldest queued event"
);
} else {
tracing::warn!(
%channel_id,
limit = MAX_PENDING_PER_CHANNEL,
"queue depth cap reached by native steer reservations — dropped new event"
);
return false;
}
}
queue.push_back(event);
self.queues.entry(channel_id).or_default().push_back(event);
true
}
@@ -4542,6 +4565,26 @@ mod tests {
/// steer must be invisible to both `flush_next` and `has_flushable_work`.
/// The withhold is the whole point of the side table — it must close the
/// `mark_complete` → ack race window.
#[test]
fn test_native_steer_reservations_count_toward_channel_cap() {
let mut q = EventQueue::new(DedupMode::Queue);
let ch = Uuid::new_v4();
for i in 0..MAX_PENDING_PER_CHANNEL {
let qe = make_queued(ch, &format!("reserved-{i}"));
let event_id = qe.event.id.to_hex();
assert!(q.push(qe));
assert!(q.mark_native_steer_pending(ch, &event_id));
}
assert_eq!(q.withheld_native_steer[&ch].len(), MAX_PENDING_PER_CHANNEL);
assert!(
!q.push(make_queued(ch, "over-cap")),
"all cap slots are reserved, so a new event must not create another preparation"
);
assert_eq!(q.withheld_native_steer[&ch].len(), MAX_PENDING_PER_CHANNEL);
assert!(q.queues.get(&ch).is_none());
}
#[test]
fn test_native_steer_withhold_only_channel_not_flushable() {
let mut q = EventQueue::new(DedupMode::Queue);