From f89b4e898bf072a6fb7bf006448d9eeef9a2253e Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 15 Oct 2026 10:36:58 +0530 Subject: [PATCH 1/4] feat(run_queue): forward queued runs to the executor Added a forwarder that drains the run queue and hands each entry to the executor, so queued runs are actually started instead of sitting idle. Tests cover the forwarding path and the module now exports the new component. Auto-committed-on: macbook Co-authored-by: Medulla --- .../src/run_queue/forwarder.rs | 102 ++++++------------ .../src/run_queue/forwarder_tests.rs | 54 ++++++++++ .../tinyagents-harness/src/run_queue/mod.rs | 42 ++++++-- 3 files changed, 118 insertions(+), 80 deletions(-) diff --git a/crates/tinyagents-harness/src/run_queue/forwarder.rs b/crates/tinyagents-harness/src/run_queue/forwarder.rs index 6078f2a0a..2e0fd22e1 100644 --- a/crates/tinyagents-harness/src/run_queue/forwarder.rs +++ b/crates/tinyagents-harness/src/run_queue/forwarder.rs @@ -112,11 +112,12 @@ fn lock_gate(gate: &ClosedGate) -> std::sync::MutexGuard<'_, bool> { } } -/// Drain `lane` and forward it into `handle`. With a `gate`, delivery happens -/// under it, and items drained after the gate closed are pushed back onto the -/// queue; returns `false` in that case (the caller should stop polling). +/// Drain `lane` and forward it into `handle`. With a `gate`, the drain and +/// the delivery both happen while holding it, so once the gate is closed no +/// item is ever out of the queue without being in the handle; returns `false` +/// when the gate is already closed (the caller should stop polling). #[allow(clippy::too_many_arguments)] -async fn forward_lane( +fn forward_lane( queue: &RunQueue, handle: &SteeringHandle, thread_label: &str, @@ -126,41 +127,27 @@ async fn forward_lane( sink: &ForwardEventSink, gate: Option<&ClosedGate>, ) -> bool { - let drained = queue.drain(lane).await; - if drained.is_empty() { - return true; - } - let delivered = drained.len(); - let (item_id, text) = drained - .first() - .map(|msg| (Some(msg.id().to_string()), Some(msg.text().to_string()))) - .unwrap_or((None, None)); - let undelivered = { + let (delivered, item_id, text) = { let closed = gate.map(lock_gate); if closed.as_deref().copied().unwrap_or(false) { - Some(drained) - } else { - for msg in &drained { - handle.send(SteeringCommand::InjectMessage(TaMessage::user(format!( - "{prefix}{}", - msg.text() - )))); - } - None + return false; + } + let drained = queue.drain_now(lane); + if drained.is_empty() { + return true; + } + for msg in &drained { + handle.send(SteeringCommand::InjectMessage(TaMessage::user(format!( + "{prefix}{}", + msg.text() + )))); } + let (item_id, text) = drained + .first() + .map(|msg| (Some(msg.id().to_string()), Some(msg.text().to_string()))) + .unwrap_or((None, None)); + (drained.len(), item_id, text) }; - if let Some(items) = undelivered { - tracing::debug!( - thread_id = thread_label, - returned = items.len(), - mode, - "[run_queue] forwarder closed mid-delivery; returned drained message(s) to the queue" - ); - // Back at the front, in order: anything queued after the drain is - // newer and must stay behind these. - queue.requeue_front(lane, items).await; - return false; - } tracing::debug!( thread_id = thread_label, delivered, @@ -196,8 +183,7 @@ pub async fn forward_steers( STEER_PREFIX, sink, None, - ) - .await; + ); } /// Forward any queued **collect** messages as injected user turns so they reach @@ -223,8 +209,7 @@ pub async fn forward_collects( COLLECT_PREFIX, sink, None, - ) - .await; + ); } /// Stop-on-drop guard around the steering-forwarder poll task. @@ -300,8 +285,7 @@ impl SteeringForwarderGuard { prefix, &loop_sink, Some(&loop_closed), - ) - .await; + ); if !open { break; } @@ -391,36 +375,12 @@ impl Drop for SteeringForwarderGuard { .map(|(text, _lane, id)| (Some(id.clone()), Some(text.clone()))) .unwrap_or((None, None)); - // `RunQueue::push` is async (tokio `Mutex`); `Drop` is synchronous. Push - // the recovered steers back on a detached task so they land in the - // session queue and become the next turn's input. The forwarder poll - // loop keeps re-draining the queue, so even if the requeue completes - // after the next turn starts, its next tick picks them up. - match tokio::runtime::Handle::try_current() { - Ok(rt) => { - let label = thread_label.clone(); - rt.spawn(async move { - // Appended, not prepended: a later guard's residuals must - // not jump ahead of these, and successive drops each add - // theirs behind the last. - for (text, lane, id) in requeue_items { - queue - .push(lane, T::requeued(id, text, &label, now_ms())) - .await; - } - }); - } - Err(_) => { - // No runtime to spawn on (should not happen on live paths, which - // always drop inside a tokio task). The steers are lost rather - // than silently mis-handled: log loudly. - tracing::warn!( - thread_id = thread_label.as_str(), - requeued, - "[run_queue] could not requeue residual steers: no tokio runtime at guard drop" - ); - return; - } + // Pushed synchronously, before `drop` returns, so a host reading the + // queue once the turn has ended finds every residual steer there. + // Appended, not prepended: a later guard's residuals must not jump + // ahead of these, and successive drops each add theirs behind the last. + for (text, lane, id) in requeue_items { + queue.push_now(lane, T::requeued(id, text, &thread_label, now_ms())); } tracing::debug!( diff --git a/crates/tinyagents-harness/src/run_queue/forwarder_tests.rs b/crates/tinyagents-harness/src/run_queue/forwarder_tests.rs index 6ba6a2bb4..923a70adb 100644 --- a/crates/tinyagents-harness/src/run_queue/forwarder_tests.rs +++ b/crates/tinyagents-harness/src/run_queue/forwarder_tests.rs @@ -254,3 +254,57 @@ async fn requeue_front_keeps_returned_items_ahead_of_newer_ones() { .collect(); assert_eq!(order, ["old-1", "old-2", "newer"]); } + +#[tokio::test] +async fn residual_steers_are_back_in_the_queue_as_soon_as_the_guard_drops() { + // A host that ends a turn and then reads the queue for what the run never + // applied must find every residual steer there, without waiting for a + // background task: nothing may be in flight once the guard has dropped. + let queue = Arc::new(RunQueue::::new()); + let handle = SteeringHandle::allow_all(); + let (sink, _events) = recorder(); + let guard = SteeringForwarderGuard::new( + handle.clone(), + Some(queue.clone()), + None, + "t".to_string(), + sink, + ); + handle.send(SteeringCommand::InjectMessage(TaMessage::user(format!( + "{STEER_PREFIX}delivered but never applied" + )))); + drop(guard); + let texts: Vec = queue + .drain_now(QueueLane::Steer) + .into_iter() + .map(|i| i.text) + .collect(); + assert_eq!(texts, ["delivered but never applied"]); +} + +#[tokio::test] +async fn a_steer_queued_while_the_guard_closes_stays_in_the_queue() { + // Queued steers the poll loop has not forwarded yet are never taken out + // of the queue once the guard is dropped: they are there to read at once. + let queue = Arc::new(RunQueue::::new()); + let handle = SteeringHandle::allow_all(); + let (sink, _events) = recorder(); + let guard = SteeringForwarderGuard::new( + handle.clone(), + Some(queue.clone()), + None, + "t".to_string(), + sink, + ); + queue.push_now(QueueLane::Steer, item("q", "queued")); + drop(guard); + // Let a poll tick pass: a closed forwarder must not take it. + tokio::time::sleep(Duration::from_millis(120)).await; + assert!(handle.drain().is_empty()); + let ids: Vec = queue + .drain_now(QueueLane::Steer) + .into_iter() + .map(|i| i.id) + .collect(); + assert_eq!(ids, ["q"]); +} diff --git a/crates/tinyagents-harness/src/run_queue/mod.rs b/crates/tinyagents-harness/src/run_queue/mod.rs index 9502ae1b9..f837131f1 100644 --- a/crates/tinyagents-harness/src/run_queue/mod.rs +++ b/crates/tinyagents-harness/src/run_queue/mod.rs @@ -35,7 +35,7 @@ mod forwarder; mod types; -use tokio::sync::Mutex; +use std::sync::{Mutex, MutexGuard, PoisonError}; pub use forwarder::{ COLLECT_PREFIX, ForwardEvent, ForwardEventSink, ForwarderCleanup, QueuedMessage, STEER_PREFIX, @@ -45,6 +45,11 @@ pub use forwarder::{ pub use types::{QueueLane, QueueMode, QueueStatus, RunQueueHandle}; /// Thread-safe FIFO queue split into steer, follow-up, and collect lanes. +/// +/// The lock guards only in-memory vector operations and is never held across +/// an `.await`, so every operation also has a synchronous form +/// ([`Self::push_now`], [`Self::drain_now`]) usable from `Drop` or other +/// non-async code without spawning a task. #[derive(Debug)] pub struct RunQueue { inner: Mutex>, @@ -75,9 +80,22 @@ impl RunQueue { } } + /// Locks the lanes. A panic while the lock was held cannot leave a lane + /// half-updated (every operation is a single vector call), so a poisoned + /// lock is recovered rather than propagated. + fn lock(&self) -> MutexGuard<'_, RunQueueInner> { + self.inner.lock().unwrap_or_else(PoisonError::into_inner) + } + /// Appends `item` to `lane`. pub async fn push(&self, lane: QueueLane, item: T) { - let mut inner = self.inner.lock().await; + self.push_now(lane, item); + } + + /// Appends `item` to `lane` without awaiting: the synchronous form of + /// [`Self::push`], for `Drop` impls and other non-async callers. + pub fn push_now(&self, lane: QueueLane, item: T) { + let mut inner = self.lock(); match lane { QueueLane::Steer => inner.steers.push(item), QueueLane::Followup => inner.followups.push(item), @@ -92,7 +110,7 @@ impl RunQueue { if items.is_empty() { return; } - let mut inner = self.inner.lock().await; + let mut inner = self.lock(); let target = match lane { QueueLane::Steer => &mut inner.steers, QueueLane::Followup => &mut inner.followups, @@ -103,7 +121,13 @@ impl RunQueue { /// Drains one lane in FIFO order. pub async fn drain(&self, lane: QueueLane) -> Vec { - let mut inner = self.inner.lock().await; + self.drain_now(lane) + } + + /// Drains one lane in FIFO order without awaiting: the synchronous form + /// of [`Self::drain`]. + pub fn drain_now(&self, lane: QueueLane) -> Vec { + let mut inner = self.lock(); match lane { QueueLane::Steer => std::mem::take(&mut inner.steers), QueueLane::Followup => std::mem::take(&mut inner.followups), @@ -118,7 +142,7 @@ impl RunQueue { match mode { QueueMode::All => self.drain(lane).await, QueueMode::OneAtATime => { - let mut inner = self.inner.lock().await; + let mut inner = self.lock(); let items = match lane { QueueLane::Steer => &mut inner.steers, QueueLane::Followup => &mut inner.followups, @@ -135,7 +159,7 @@ impl RunQueue { /// Returns the current queue depth per lane. pub async fn status(&self) -> QueueStatus { - let inner = self.inner.lock().await; + let inner = self.lock(); let steers = inner.steers.len(); let followups = inner.followups.len(); let collects = inner.collects.len(); @@ -149,7 +173,7 @@ impl RunQueue { /// Clears every lane and returns the number of dropped items. pub async fn clear(&self) -> usize { - let mut inner = self.inner.lock().await; + let mut inner = self.lock(); let total = inner.steers.len() + inner.followups.len() + inner.collects.len(); inner.steers.clear(); inner.followups.clear(); @@ -168,7 +192,7 @@ impl RunQueue { where T: Clone, { - let inner = self.inner.lock().await; + let inner = self.lock(); let mut items = Vec::with_capacity(inner.steers.len() + inner.followups.len() + inner.collects.len()); items.extend( @@ -202,7 +226,7 @@ impl RunQueue { /// Use to retract a specific queued item (e.g. one the host decided not /// to apply after all) without clearing the rest of the queue. pub async fn remove_where(&self, pred: impl Fn(&T) -> bool) -> usize { - let mut inner = self.inner.lock().await; + let mut inner = self.lock(); let before = inner.steers.len() + inner.followups.len() + inner.collects.len(); inner.steers.retain(|item| !pred(item)); inner.followups.retain(|item| !pred(item)); From 26530154a493ccedc13fb158dbc7f730f586a659 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 15 Oct 2026 10:37:50 +0530 Subject: [PATCH 2/4] docs(run_queue): clarify forwarder gate and requeue invariants Update the forwarder's doc comments to describe the current design: the poll task drains the queue and delivers under the shutdown gate, so an item is always either in the queue or in the handle and never in flight. The requeue is synchronous, so hosts can read leftover steers as soon as the guard drops. Auto-committed-on: macbook Co-authored-by: Medulla --- .../src/run_queue/forwarder.rs | 23 ++++++++++--------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/crates/tinyagents-harness/src/run_queue/forwarder.rs b/crates/tinyagents-harness/src/run_queue/forwarder.rs index 2e0fd22e1..3394b2138 100644 --- a/crates/tinyagents-harness/src/run_queue/forwarder.rs +++ b/crates/tinyagents-harness/src/run_queue/forwarder.rs @@ -15,6 +15,8 @@ //! task, runs the host's cleanup hook (e.g. deregistering a steering-registry //! entry), and drains residual (delivered-but-unapplied) steers back into the //! queue so a late steer becomes the *next* turn's input instead of vanishing. +//! The requeue is synchronous: once the guard has dropped, every steer the run +//! did not apply is in the queue, so a host can read the leftovers right away. //! //! The queued payload, the delivery notifications and the cleanup are host //! seams: [`QueuedMessage`] for the payload, [`ForwardEvent`] for notifications. @@ -98,11 +100,11 @@ fn now_ms() -> u64 { /// Shared shutdown gate between the poll task and the guard's `Drop`. /// -/// The poll task sends into the steering handle only while holding this lock -/// and seeing `false`; `Drop` sets it to `true` under the same lock before it -/// drains the handle. So once `Drop` has drained, no later send can land in a -/// handle nobody will read again: the task pushes what it drained back onto the -/// queue instead. +/// The poll task drains the queue and sends into the steering handle only +/// while holding this lock and seeing `false`; `Drop` sets it to `true` under +/// the same lock before it drains the handle. So an item is always either in +/// the queue or in the handle, never in flight, and once `Drop` has drained, +/// no later send can land in a handle nobody will read again. type ClosedGate = std::sync::Mutex; fn lock_gate(gate: &ClosedGate) -> std::sync::MutexGuard<'_, bool> { @@ -259,10 +261,9 @@ impl SteeringForwarderGuard { let loop_sink = sink.clone(); let loop_closed = closed.clone(); tokio::spawn(async move { - // Not aborted on drop: an abort could land between draining the - // queue and delivering, losing what was drained. The loop exits - // on its own once the gate is closed, returning anything it - // drained after closing to the queue. + // Not aborted on drop; the loop exits on its own at its next + // check once the gate is closed. It drains and delivers under + // the gate, so a closed gate means it takes nothing more. for (lane, mode, prefix) in [ (QueueLane::Steer, "steer", STEER_PREFIX), (QueueLane::Collect, "collect", COLLECT_PREFIX), @@ -338,8 +339,8 @@ impl Drop for SteeringForwarderGuard { // once the run is gone and are intentionally dropped. let residual = self.handle.drain(); // Each residual steer gets its requeued id minted here (Drop is - // synchronous) so the `Requeued` event below and the item actually - // pushed onto the queue in the spawned task agree on the same id. + // synchronous) so the `Requeued` event below and the item pushed onto + // the queue agree on the same id. let requeue_items: Vec<(String, QueueLane, String)> = residual .into_iter() .filter_map(|cmd| match cmd { From 2462186a6e0ff38bd4ffaee4415d053ce0cfe37a Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 15 Oct 2026 10:38:01 +0530 Subject: [PATCH 3/4] docs(run_queue): document the run queue module Add a README describing the run queue's purpose and how it schedules work within the harness. Auto-committed-on: macbook Co-authored-by: Medulla --- crates/tinyagents-harness/src/run_queue/README.md | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/crates/tinyagents-harness/src/run_queue/README.md b/crates/tinyagents-harness/src/run_queue/README.md index efceb29b6..c2c6ae0fc 100644 --- a/crates/tinyagents-harness/src/run_queue/README.md +++ b/crates/tinyagents-harness/src/run_queue/README.md @@ -28,7 +28,9 @@ stops, pauses, and deferrals leave the queue untouched. Details: - [`RunQueue`] — the queue itself. `new`/`Default` create an empty queue; `push(lane, item)` appends; `drain(lane)` empties one lane in FIFO order and - returns its contents; `take(lane, mode)` is the `QueueMode`-aware form; + returns its contents; `push_now`/`drain_now` are their synchronous forms, + for `Drop` impls and other non-async callers; `take(lane, mode)` is the + `QueueMode`-aware form; `status()` snapshots per-lane depth; `clear()` empties every lane and returns how many items were dropped. - [`QueueLane`] — which lane an item belongs to: `Steer` (inject at the next @@ -51,11 +53,16 @@ stops, pauses, and deferrals leave the queue untouched. Details: ## Operational constraints -- All mutating/reading methods are `async` and serialize through a single - `tokio::sync::Mutex` guarding all three lanes together, so a caller draining - one lane briefly blocks a concurrent push to another — acceptable given the +- Every method serializes through a single `std::sync::Mutex` guarding all + three lanes together. It is held only for one vector operation and never + across an `.await`, so the `async` methods never block a runtime thread for + long and the synchronous `push_now`/`drain_now` are safe anywhere. The queue is meant to hold a handful of pending items, not a high-throughput channel. +- `SteeringForwarderGuard` drains and delivers under its shutdown gate and + requeues residual (delivered-but-unapplied) steers synchronously in `Drop`, + so once the guard has dropped, every steer the run did not apply is back in + the queue and readable at once. - `drain` removes and returns items in FIFO order; it does not re-queue on failure — a caller that fails to act on a drained item is responsible for re-pushing it if that is the desired behavior. From b6b7f994a54f43a7b7bac106dec082d56c05fbef Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 15 Oct 2026 10:38:48 +0530 Subject: [PATCH 4/4] docs(changelog): note unapplied steers returning to the run queue Document that SteeringForwarderGuard now drains and delivers under its shutdown gate and requeues residual steers synchronously in Drop, so a host can deterministically collect a turn's leftovers. Also note that RunQueue guards its lanes with a std::sync::Mutex and gains synchronous push_now and drain_now while its async methods stay unchanged. Auto-committed-on: macbook Co-authored-by: Medulla --- CHANGELOG.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 22ac50711..1b4fe4a30 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,15 @@ # Changelog +## Unreleased: unapplied steers are back in the run queue when a turn ends + +`SteeringForwarderGuard` now drains and delivers under its shutdown gate and +requeues residual (delivered-but-unapplied) steers synchronously in `Drop`, +instead of on a spawned task. Once the guard has dropped, every steer the run +did not apply is in the queue, so a host can collect the leftovers of a turn +deterministically. `RunQueue` guards its lanes with a `std::sync::Mutex` +(never held across an `.await`) and gains synchronous `push_now` and +`drain_now`; its `async` methods are unchanged. + ## Unreleased: tinystoragedrivers by git tag The workspace takes `tinystoragedrivers-core` / `-sqlite` by git tag `v0.11.0`