diff --git a/docs/concepts/agent-loop.md b/docs/concepts/agent-loop.md index d063ad7..b2cce4d 100644 --- a/docs/concepts/agent-loop.md +++ b/docs/concepts/agent-loop.md @@ -165,6 +165,48 @@ agent.clear_follow_up_queue(); // Drop all pending follow-ups agent.clear_all_queues(); // Drop everything ``` +Clients that render the queue (e.g. showing pending messages while the agent +is busy) can inspect it without consuming: + +```rust +let pending = agent.steering_queue_snapshot(); // point-in-time copy +let count = pending.len(); +``` + +Use `steering_queue_len()` instead when only the count is needed — it avoids +cloning message contents, which matters for high-frequency polling. + +For an edit-and-requeue UI, drain atomically and requeue the survivors as a +single batch: + +```rust +let mut queued = agent.take_steering_queue(); // atomic drain +queued.pop(); // e.g. drop the newest entry +agent.steer_all(queued); // requeue the rest under one lock +``` + +These accessors pair with the spawn-based flows (`prompt()`, +`prompt_messages()`, `continue_loop()`). The `_with_sender` variants borrow +the agent mutably for the whole run, so the queue cannot be touched while one +is in flight. + +Only the drain is atomic — the edit round trip is not: + +- The loop keeps running while the queue is taken. If the run finishes + during the edit, requeued survivors are delivered at the start of the + *next* run. +- A message the loop has already picked up for injection is no longer in + the queue: snapshots won't show it and `take_steering_queue()` cannot + retract it. +- Messages steered concurrently during the edit window land ahead of + requeued survivors (`steer` appends to the back; delivery pops the front). +- Delivery of a requeued batch follows the steering `QueueMode`: `All` + injects it at one check, the default `OneAtATime` delivers one message + per check. +- After `reset()`, discard taken messages instead of requeueing them. +- Pending queue contents are not included in `save_messages()` persistence — + snapshot and store them separately if they must survive a restart. + ### Low-Level API When using `agent_loop()` directly, steering and follow-ups are provided via callback functions: diff --git a/docs/reference/api.md b/docs/reference/api.md index 81170cb..240ee53 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -145,6 +145,14 @@ All return `Self` for chaining (unless noted as `Result`). | `clear_steering_queue()` | Clear pending steering messages | | `clear_follow_up_queue()` | Clear pending follow-up messages | | `clear_all_queues()` | Clear both queues | +| `steer_all(msgs: Vec)` | Queue multiple steering messages under one lock | +| `follow_up_all(msgs: Vec)` | Queue multiple follow-up messages under one lock | +| `steering_queue_snapshot() -> Vec` | Copy of pending steering messages (does not consume) | +| `follow_up_queue_snapshot() -> Vec` | Copy of pending follow-up messages | +| `steering_queue_len() -> usize` | Number of pending steering messages | +| `follow_up_queue_len() -> usize` | Number of pending follow-up messages | +| `take_steering_queue() -> Vec` | Atomically drain and return pending steering messages (messages already picked up by the loop are not included) | +| `take_follow_up_queue() -> Vec` | Atomically drain and return pending follow-up messages | | `set_steering_mode(mode: QueueMode)` | Set delivery mode: `OneAtATime` or `All` | | `set_follow_up_mode(mode: QueueMode)` | Set delivery mode: `OneAtATime` or `All` | diff --git a/src/agent.rs b/src/agent.rs index 003d172..b707cc8 100644 --- a/src/agent.rs +++ b/src/agent.rs @@ -362,6 +362,20 @@ impl Agent { self.follow_up_queue.lock().unwrap().push(msg); } + /// Queue multiple steering messages under a single lock acquisition. + /// + /// Use this to requeue messages from [`Agent::take_steering_queue`] — a + /// per-message [`Agent::steer`] loop can be interleaved by the running + /// loop's steering checks, splitting the batch across turns. + pub fn steer_all(&self, msgs: Vec) { + self.steering_queue.lock().unwrap().extend(msgs); + } + + /// Queue multiple follow-up messages under a single lock acquisition. + pub fn follow_up_all(&self, msgs: Vec) { + self.follow_up_queue.lock().unwrap().extend(msgs); + } + pub fn clear_steering_queue(&self) { self.steering_queue.lock().unwrap().clear(); } @@ -375,6 +389,52 @@ impl Agent { self.clear_follow_up_queue(); } + /// Snapshot of the messages currently waiting in the steering queue. + /// + /// The snapshot is a point-in-time copy for display purposes — the agent + /// loop may drain the queue at any moment. For atomic remove-and-edit, + /// use [`Agent::take_steering_queue`]. + pub fn steering_queue_snapshot(&self) -> Vec { + self.steering_queue.lock().unwrap().clone() + } + + /// Snapshot of the messages currently waiting in the follow-up queue. + pub fn follow_up_queue_snapshot(&self) -> Vec { + self.follow_up_queue.lock().unwrap().clone() + } + + /// Number of messages currently waiting in the steering queue. + pub fn steering_queue_len(&self) -> usize { + self.steering_queue.lock().unwrap().len() + } + + /// Number of messages currently waiting in the follow-up queue. + pub fn follow_up_queue_len(&self) -> usize { + self.follow_up_queue.lock().unwrap().len() + } + + /// Atomically drain the steering queue and return its messages. + /// + /// Enables edit-and-requeue UIs: take the queue, let the user edit or + /// drop entries, then push the survivors back with [`Agent::steer_all`]. + /// + /// Only the drain itself is atomic. For edit-and-requeue flows: + /// - Messages the running loop has already picked up (but not yet + /// injected into history) are no longer in the queue and cannot be + /// retracted here. + /// - If the run finishes while the queue is taken, requeued messages + /// are delivered at the start of the *next* run. + /// - After [`Agent::reset`], discard taken messages instead of + /// requeueing them — they belong to the discarded conversation. + pub fn take_steering_queue(&self) -> Vec { + std::mem::take(&mut *self.steering_queue.lock().unwrap()) + } + + /// Atomically drain the follow-up queue and return its messages. + pub fn take_follow_up_queue(&self) -> Vec { + std::mem::take(&mut *self.follow_up_queue.lock().unwrap()) + } + pub fn set_steering_mode(&mut self, mode: QueueMode) { self.steering_mode = mode; } diff --git a/tests/agent_test.rs b/tests/agent_test.rs index 0d16d87..e74ac9a 100644 --- a/tests/agent_test.rs +++ b/tests/agent_test.rs @@ -384,3 +384,59 @@ async fn test_prompt_with_sender_tools_restored() { agent.finish().await; assert_eq!(agent.messages().len(), 4); // 2 from first + 2 from second } + +#[tokio::test] +async fn test_queue_inspection_and_take() { + let provider = MockProvider::text("Hello!"); + let agent = Agent::new(provider) + .with_system_prompt("test") + .with_model("mock") + .with_api_key("test"); + + assert_eq!(agent.steering_queue_len(), 0); + assert!(agent.steering_queue_snapshot().is_empty()); + + agent.steer(AgentMessage::Llm(Message::user("stop"))); + agent.steer(AgentMessage::Llm(Message::user("use v2 instead"))); + agent.follow_up(AgentMessage::Llm(Message::user("then run tests"))); + + // Inspection does not consume + assert_eq!(agent.follow_up_queue_len(), 1); + let snapshot = agent.steering_queue_snapshot(); + assert_eq!(snapshot.len(), 2); + assert_eq!(agent.steering_queue_len(), 2, "snapshot must not drain"); + + // Take drains atomically and returns in FIFO order + let taken = agent.take_steering_queue(); + assert_eq!(taken.len(), 2); + assert_eq!(agent.steering_queue_len(), 0); + let AgentMessage::Llm(Message::User { content, .. }) = &taken[0] else { + panic!("expected user message"); + }; + assert_eq!( + content, + &vec![Content::Text { + text: "stop".into() + }] + ); + + // Edit-and-requeue: drop the first entry, batch-requeue the survivor + agent.steer_all(vec![taken[1].clone()]); + assert_eq!(agent.steering_queue_len(), 1); + let requeued = agent.steering_queue_snapshot(); + let AgentMessage::Llm(Message::User { content, .. }) = &requeued[0] else { + panic!("expected user message"); + }; + assert_eq!( + content, + &vec![Content::Text { + text: "use v2 instead".into() + }] + ); + + // Follow-up variants + let taken = agent.take_follow_up_queue(); + assert_eq!(taken.len(), 1); + assert_eq!(agent.follow_up_queue_len(), 0); + assert!(agent.follow_up_queue_snapshot().is_empty()); +}