Requeue unapplied steers synchronously when a forwarder guard drops

CHANGELOG.md

@@ -1,5 +1,14 @@
 # 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: workspace transcripts on driver ports (tinyagents#453, transcript half)
 
 - `port::WorkspaceStores::open(workspace, agent_id)` (feature `legacy-import`)

crates/tinyagents-harness/src/run_queue/README.md

@@ -28,7 +28,9 @@ stops, pauses, and deferrals leave the queue untouched. Details:
 
 - [`RunQueue<T>`] — 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.

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<bool>;
 
 fn lock_gate(gate: &ClosedGate) -> std::sync::MutexGuard<'_, bool> {
@@ -112,11 +114,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<T: QueuedMessage>(
+fn forward_lane<T: QueuedMessage>(
     queue: &RunQueue<T>,
     handle: &SteeringHandle,
     thread_label: &str,
@@ -126,41 +129,27 @@ async fn forward_lane<T: QueuedMessage>(
     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 +185,7 @@ pub async fn forward_steers<T: QueuedMessage>(
         STEER_PREFIX,
         sink,
         None,
-    )
-    .await;
+    );
 }
 
 /// Forward any queued **collect** messages as injected user turns so they reach
@@ -223,8 +211,7 @@ pub async fn forward_collects<T: QueuedMessage>(
         COLLECT_PREFIX,
         sink,
         None,
-    )
-    .await;
+    );
 }
 
 /// Stop-on-drop guard around the steering-forwarder poll task.
@@ -274,10 +261,9 @@ impl<T: QueuedMessage> SteeringForwarderGuard<T> {
             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),
@@ -300,8 +286,7 @@ impl<T: QueuedMessage> SteeringForwarderGuard<T> {
                         prefix,
                         &loop_sink,
                         Some(&loop_closed),
-                    )
-                    .await;
+                    );
                     if !open {
                         break;
                     }
@@ -354,8 +339,8 @@ impl<T: QueuedMessage> Drop for SteeringForwarderGuard<T> {
         //    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 {
@@ -391,36 +376,12 @@ impl<T: QueuedMessage> Drop for SteeringForwarderGuard<T> {
             .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!(

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::<Item>::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<String> = 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::<Item>::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<String> = queue
+        .drain_now(QueueLane::Steer)
+        .into_iter()
+        .map(|i| i.id)
+        .collect();
+    assert_eq!(ids, ["q"]);
+}