From 4f90a02e8b3b8dbc041a1a6e6ef2ee61c98123ac Mon Sep 17 00:00:00 2001 From: Yuanhao Li Date: Sat, 4 Jul 2026 22:34:56 +0200 Subject: [PATCH 1/2] feat: add steering/follow-up queue inspection API (#50) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add read access to the Agent's message queues so clients can render pending messages while the agent is busy (pi-style queue UIs): - steering_queue_snapshot() / follow_up_queue_snapshot() — point-in-time copy, does not consume - steering_queue_len() / follow_up_queue_len() - take_steering_queue() / take_follow_up_queue() — atomic drain-and-return, enabling edit-and-requeue clear_* methods already existed; this adds the inspection half. Non-breaking, agent.rs only. Co-Authored-By: Claude Fable 5 --- docs/concepts/agent-loop.md | 19 +++++++++++++++ docs/reference/api.md | 6 +++++ src/agent.rs | 37 +++++++++++++++++++++++++++++ tests/agent_test.rs | 47 +++++++++++++++++++++++++++++++++++++ 4 files changed, 109 insertions(+) diff --git a/docs/concepts/agent-loop.md b/docs/concepts/agent-loop.md index d063ad7..69f92ed 100644 --- a/docs/concepts/agent-loop.md +++ b/docs/concepts/agent-loop.md @@ -165,6 +165,25 @@ 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 = agent.steering_queue_len(); +``` + +Snapshots are for display — the loop may drain the queue at any moment. For +an edit-and-requeue UI, drain atomically instead: + +```rust +let mut queued = agent.take_steering_queue(); // atomically drain +queued.retain(|m| user_kept(m)); // let the user edit the list +for msg in queued { + agent.steer(msg); // requeue the survivors +} +``` + ### 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..50cad79 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -145,6 +145,12 @@ 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 | +| `steering_queue_snapshot()` | Copy of pending steering messages (does not consume) | +| `follow_up_queue_snapshot()` | Copy of pending follow-up messages | +| `steering_queue_len()` | Number of pending steering messages | +| `follow_up_queue_len()` | Number of pending follow-up messages | +| `take_steering_queue()` | Atomically drain and return pending steering messages | +| `take_follow_up_queue()` | 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..094dc80 100644 --- a/src/agent.rs +++ b/src/agent.rs @@ -375,6 +375,43 @@ 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 [`Agent::steer`] the survivors back. + 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..a153217 100644 --- a/tests/agent_test.rs +++ b/tests/agent_test.rs @@ -384,3 +384,50 @@ 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.steering_queue_len(), 2); + 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: push a survivor back + agent.steer(taken[1].clone()); + assert_eq!(agent.steering_queue_len(), 1); + + // 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()); +} From a50ba2e112a50f0a50a97781eb5b11740b14cb26 Mon Sep 17 00:00:00 2001 From: Yuanhao Li Date: Sat, 4 Jul 2026 22:58:28 +0200 Subject: [PATCH 2/2] fix: address code-review findings on queue API semantics - add steer_all()/follow_up_all() batch enqueue (one lock) so requeues from take_*_queue() cannot be split by a concurrent drain - document the edit-window semantics honestly: only the drain is atomic; survivors can deliver at the start of the next run; in-flight messages cannot be retracted; ordering vs concurrent producers; QueueMode effect; reset() and persistence caveats - note the _with_sender precondition (queue accessors need the spawn-based flows) - make the doc example self-contained (no undefined helper), derive the count from the snapshot instead of double-locking - api.md: return types on the new rows + batch method rows - test: batch-requeue via steer_all with survivor-content assertion; drop a redundant assertion Co-Authored-By: Claude Fable 5 --- docs/concepts/agent-loop.md | 39 +++++++++++++++++++++++++++++-------- docs/reference/api.md | 14 +++++++------ src/agent.rs | 25 +++++++++++++++++++++++- tests/agent_test.rs | 15 +++++++++++--- 4 files changed, 75 insertions(+), 18 deletions(-) diff --git a/docs/concepts/agent-loop.md b/docs/concepts/agent-loop.md index 69f92ed..b2cce4d 100644 --- a/docs/concepts/agent-loop.md +++ b/docs/concepts/agent-loop.md @@ -170,20 +170,43 @@ is busy) can inspect it without consuming: ```rust let pending = agent.steering_queue_snapshot(); // point-in-time copy -let count = agent.steering_queue_len(); +let count = pending.len(); ``` -Snapshots are for display — the loop may drain the queue at any moment. For -an edit-and-requeue UI, drain atomically instead: +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(); // atomically drain -queued.retain(|m| user_kept(m)); // let the user edit the list -for msg in queued { - agent.steer(msg); // requeue the survivors -} +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 50cad79..240ee53 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -145,12 +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 | -| `steering_queue_snapshot()` | Copy of pending steering messages (does not consume) | -| `follow_up_queue_snapshot()` | Copy of pending follow-up messages | -| `steering_queue_len()` | Number of pending steering messages | -| `follow_up_queue_len()` | Number of pending follow-up messages | -| `take_steering_queue()` | Atomically drain and return pending steering messages | -| `take_follow_up_queue()` | Atomically drain and return pending follow-up messages | +| `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 094dc80..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(); } @@ -402,7 +416,16 @@ impl Agent { /// 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 [`Agent::steer`] the survivors back. + /// 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()) } diff --git a/tests/agent_test.rs b/tests/agent_test.rs index a153217..e74ac9a 100644 --- a/tests/agent_test.rs +++ b/tests/agent_test.rs @@ -401,7 +401,6 @@ async fn test_queue_inspection_and_take() { agent.follow_up(AgentMessage::Llm(Message::user("then run tests"))); // Inspection does not consume - assert_eq!(agent.steering_queue_len(), 2); assert_eq!(agent.follow_up_queue_len(), 1); let snapshot = agent.steering_queue_snapshot(); assert_eq!(snapshot.len(), 2); @@ -421,9 +420,19 @@ async fn test_queue_inspection_and_take() { }] ); - // Edit-and-requeue: push a survivor back - agent.steer(taken[1].clone()); + // 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();