feat(harness): extract reusable run queue mechanics

docs/modules/harness/README.md

@@ -119,6 +119,8 @@ tests for every adapter.
 - Support parent-orchestrator and human steering of sub-agents, orchestrator
   agents, graph tasks, and harness loops through typed commands delivered at
   safe boundaries.
+- Queue host-owned payloads in steer, follow-up, and collected-context lanes
+  without importing host transport or UI metadata into the harness.
 - Support durable graph runs with pause/resume, checkpoint listing, and
   inspectable node transitions.
 - Support per-agent execution blueprints that describe how an agent runs
@@ -164,6 +166,10 @@ src/harness/
   prompt.rs
   providers.rs
   retry.rs
+  run_queue/
+    mod.rs
+    test.rs
+    types.rs
   runtime.rs
   steering.rs
   stream.rs
@@ -201,6 +207,8 @@ Feature ownership:
 - `prompt`: prompt templates, rendering, and dynamic prompt context.
 - `providers`: feature-gated provider adapters.
 - `retry`: retry classification, backoff, attempt accounting.
+- `run_queue`: generic FIFO mechanics for steer, follow-up, and
+  collected-context payloads; hosts retain ownership of payload metadata.
 - `runtime`: high-level `AgentHarness` builder/facade.
 - `steering`: policy-checked parent/human steering of orchestrators,
   sub-agents, graph tasks, and harness loops.

src/harness/mod.rs

@@ -31,6 +31,7 @@ pub mod observability;
 pub mod prompt;
 pub mod providers;
 pub mod retry;
+pub mod run_queue;
 pub mod runtime;
 pub mod steering;
 pub mod store;

src/harness/providers/openai/local.rs

@@ -326,6 +326,26 @@ pub(super) fn local_options_object(provider_options: &Value) -> Option<&Value> {
 // Error classification
 // ---------------------------------------------------------------------------
 
+/// Returns `true` when a provider failure says the model's Jinja chat template
+/// rejected the message list.
+///
+/// This is distinct from a rejected model id, sampling parameter, or
+/// credential. The markers are emitted by local OpenAI-compatible runtimes
+/// such as LM Studio, llama.cpp, and Ollama while rendering model-owned chat
+/// templates. Matching is deliberately narrow and case-insensitive so hosts
+/// can present accurate remediation without misclassifying ordinary 400s.
+pub fn is_chat_template_rejection_message(body: &str) -> bool {
+    const PHRASES: &[&str] = &[
+        "no user query found in messages",
+        "unable to generate parser for this template",
+        "automatic parser generation failed",
+        "jinja exception",
+    ];
+
+    let lower = body.to_ascii_lowercase();
+    PHRASES.iter().any(|phrase| lower.contains(phrase))
+}
+
 /// Rewrites a local runtime's opaque 404 into a message naming the fix.
 ///
 /// The embeddings adapter has done this for a while — "Run `ollama pull

src/harness/providers/openai/local_test.rs

@@ -3,6 +3,39 @@
 use super::*;
 use serde_json::json;
 
+const LMSTUDIO_CHAT_TEMPLATE_REJECTION: &str = "lmstudio returned: Engine protocol predict \
+    request returned 400: {\"error\":{\"code\":400,\"message\":\"Unable to generate parser \
+    for this template. Automatic parser generation failed: While executing CallExpression at \
+    line 79, column 24 in source: {{- raise_exception('No user query found in messages.') }}. \
+    Error: Jinja Exception: No user query found in messages.\"}}";
+
+#[test]
+fn chat_template_rejections_are_classified_inside_runtime_wrappers() {
+    let aggregate =
+        format!("The model may not be available. Attempts: {LMSTUDIO_CHAT_TEMPLATE_REJECTION}");
+    assert!(is_chat_template_rejection_message(&aggregate));
+}
+
+#[test]
+fn chat_template_rejection_detection_is_case_insensitive() {
+    assert!(is_chat_template_rejection_message(
+        "Error: JINJA EXCEPTION: No User Query Found In Messages."
+    ));
+}
+
+#[test]
+fn unrelated_provider_rejections_are_not_chat_template_failures() {
+    for body in [
+        "openai API error (400): invalid temperature: only 1 is allowed for this model",
+        "The model `gpt-5.5` does not exist or you do not have access to it.",
+        "lmstudio returned: model 'qwen3.5-9b' does not support tools",
+        "openrouter API error (429): rate limited",
+        "Failed to render the prompt template file on disk",
+    ] {
+        assert!(!is_chat_template_rejection_message(body), "{body:?}");
+    }
+}
+
 #[test]
 fn native_root_strips_the_openai_compat_suffix() {
     let kind = LocalRuntimeKind::Ollama;

src/harness/providers/openai/mod.rs

@@ -92,7 +92,9 @@ mod sse;
 mod transport;
 
 pub use convert::CacheTokenAccounting;
-pub use local::{CONTEXT_OVERFLOW_CODE, LocalProbe, LocalRuntimeKind};
+pub use local::{
+    CONTEXT_OVERFLOW_CODE, LocalProbe, LocalRuntimeKind, is_chat_template_rejection_message,
+};
 pub use reasoning_tags::ReasoningTagExtraction;
 pub use transport::{AuthStyle, OpenAiModel};
 

src/harness/run_queue/mod.rs

@@ -0,0 +1,91 @@
+//! Generic multi-lane queue for messages arriving during an active run.
+//!
+//! Hosts decide which incoming events should be queued and retain ownership of
+//! the queued payload. TinyAgents owns the reusable FIFO mechanics for the
+//! three lanes an agent runtime can consume at safe iteration boundaries:
+//! immediate steering, deferred follow-up work, and collected context.
+
+mod types;
+
+use tokio::sync::Mutex;
+
+pub use types::{QueueLane, QueueStatus};
+
+/// Thread-safe FIFO queue split into steer, follow-up, and collect lanes.
+#[derive(Debug)]
+pub struct RunQueue<T> {
+    inner: Mutex<RunQueueInner<T>>,
+}
+
+#[derive(Debug)]
+struct RunQueueInner<T> {
+    steers: Vec<T>,
+    followups: Vec<T>,
+    collects: Vec<T>,
+}
+
+impl<T> RunQueue<T> {
+    /// Creates an empty queue.
+    pub fn new() -> Self {
+        Self {
+            inner: Mutex::new(RunQueueInner {
+                steers: Vec::new(),
+                followups: Vec::new(),
+                collects: Vec::new(),
+            }),
+        }
+    }
+
+    /// Appends `item` to `lane`.
+    pub async fn push(&self, lane: QueueLane, item: T) {
+        let mut inner = self.inner.lock().await;
+        match lane {
+            QueueLane::Steer => inner.steers.push(item),
+            QueueLane::Followup => inner.followups.push(item),
+            QueueLane::Collect => inner.collects.push(item),
+        }
+    }
+
+    /// Drains one lane in FIFO order.
+    pub async fn drain(&self, lane: QueueLane) -> Vec<T> {
+        let mut inner = self.inner.lock().await;
+        match lane {
+            QueueLane::Steer => std::mem::take(&mut inner.steers),
+            QueueLane::Followup => std::mem::take(&mut inner.followups),
+            QueueLane::Collect => std::mem::take(&mut inner.collects),
+        }
+    }
+
+    /// Returns the current queue depth per lane.
+    pub async fn status(&self) -> QueueStatus {
+        let inner = self.inner.lock().await;
+        let steers = inner.steers.len();
+        let followups = inner.followups.len();
+        let collects = inner.collects.len();
+        QueueStatus {
+            steers,
+            followups,
+            collects,
+            total: steers + followups + collects,
+        }
+    }
+
+    /// Clears every lane and returns the number of dropped items.
+    pub async fn clear(&self) -> usize {
+        let mut inner = self.inner.lock().await;
+        let total = inner.steers.len() + inner.followups.len() + inner.collects.len();
+        inner.steers.clear();
+        inner.followups.clear();
+        inner.collects.clear();
+        total
+    }
+}
+
+impl<T> Default for RunQueue<T> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod test;

src/harness/run_queue/test.rs

@@ -0,0 +1,56 @@
+use super::*;
+
+#[tokio::test]
+async fn new_queue_is_empty() {
+    let queue = RunQueue::<String>::new();
+    assert_eq!(
+        queue.status().await,
+        QueueStatus {
+            steers: 0,
+            followups: 0,
+            collects: 0,
+            total: 0,
+        }
+    );
+}
+
+#[tokio::test]
+async fn push_routes_items_to_the_requested_lane() {
+    let queue = RunQueue::new();
+    queue.push(QueueLane::Steer, "steer").await;
+    queue.push(QueueLane::Followup, "followup").await;
+    queue.push(QueueLane::Collect, "collect").await;
+
+    assert_eq!(
+        queue.status().await,
+        QueueStatus {
+            steers: 1,
+            followups: 1,
+            collects: 1,
+            total: 3,
+        }
+    );
+}
+
+#[tokio::test]
+async fn drain_is_fifo_and_does_not_affect_other_lanes() {
+    let queue = RunQueue::new();
+    queue.push(QueueLane::Steer, "first").await;
+    queue.push(QueueLane::Steer, "second").await;
+    queue.push(QueueLane::Followup, "later").await;
+
+    assert_eq!(queue.drain(QueueLane::Steer).await, vec!["first", "second"]);
+    assert_eq!(queue.status().await.followups, 1);
+    assert_eq!(queue.status().await.steers, 0);
+}
+
+#[tokio::test]
+async fn clear_empties_every_lane_and_reports_the_drop_count() {
+    let queue = RunQueue::new();
+    queue.push(QueueLane::Steer, 1).await;
+    queue.push(QueueLane::Followup, 2).await;
+    queue.push(QueueLane::Collect, 3).await;
+
+    assert_eq!(queue.clear().await, 3);
+    assert_eq!(queue.status().await.total, 0);
+}

src/harness/run_queue/types.rs

@@ -0,0 +1,25 @@
+//! Public types for the active-run queue.
+
+/// A queue lane consumed by the agent runtime.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum QueueLane {
+    /// Inject at the next safe iteration boundary as an instruction.
+    Steer,
+    /// Dispatch as a fresh turn after the active run completes.
+    Followup,
+    /// Inject at the next safe boundary as additional context.
+    Collect,
+}
+
+/// Snapshot of the queue depth per lane.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
+pub struct QueueStatus {
+    /// Number of pending steer items.
+    pub steers: usize,
+    /// Number of pending follow-up items.
+    pub followups: usize,
+    /// Number of pending collected-context items.
+    pub collects: usize,
+    /// Total number of pending items across all lanes.
+    pub total: usize,
+}

src/lib.rs

@@ -159,6 +159,9 @@ pub use harness::steering::{
     SteeringCommand, SteeringCommandKind, SteeringHandle, SteeringOutcome, SteeringPolicy,
 };
 
+// --- Harness: active-run message queues ---
+pub use harness::run_queue::{QueueLane, QueueStatus, RunQueue};
+
 // --- Cooperative run cancellation ---
 pub use harness::cancel::CancellationToken;