From 3e4bcada4d182c0c8ac49da0e3cd6609cc9911e4 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 12 Aug 2026 16:23:14 +0300 Subject: [PATCH 1/2] feat(harness): extract generic active-run queue Co-authored-by: Medulla --- docs/modules/harness/README.md | 5 ++ src/harness/mod.rs | 1 + src/harness/providers/openai/local.rs | 20 +++++ src/harness/providers/openai/local_test.rs | 33 ++++++++ src/harness/providers/openai/mod.rs | 4 +- src/harness/run_queue/mod.rs | 91 ++++++++++++++++++++++ src/harness/run_queue/test.rs | 56 +++++++++++++ src/harness/run_queue/types.rs | 25 ++++++ src/lib.rs | 3 + 9 files changed, 237 insertions(+), 1 deletion(-) create mode 100644 src/harness/run_queue/mod.rs create mode 100644 src/harness/run_queue/test.rs create mode 100644 src/harness/run_queue/types.rs diff --git a/docs/modules/harness/README.md b/docs/modules/harness/README.md index c24f834..25b1689 100644 --- a/docs/modules/harness/README.md +++ b/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,7 @@ src/harness/ prompt.rs providers.rs retry.rs + run_queue.rs runtime.rs steering.rs stream.rs @@ -201,6 +204,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. diff --git a/src/harness/mod.rs b/src/harness/mod.rs index 91d52c9..449f45f 100644 --- a/src/harness/mod.rs +++ b/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; diff --git a/src/harness/providers/openai/local.rs b/src/harness/providers/openai/local.rs index b8b1888..df35a1e 100644 --- a/src/harness/providers/openai/local.rs +++ b/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 diff --git a/src/harness/providers/openai/local_test.rs b/src/harness/providers/openai/local_test.rs index ca0ad47..19c0afa 100644 --- a/src/harness/providers/openai/local_test.rs +++ b/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; diff --git a/src/harness/providers/openai/mod.rs b/src/harness/providers/openai/mod.rs index 02367c1..94a30e0 100644 --- a/src/harness/providers/openai/mod.rs +++ b/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}; diff --git a/src/harness/run_queue/mod.rs b/src/harness/run_queue/mod.rs new file mode 100644 index 0000000..5940cf2 --- /dev/null +++ b/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 { + inner: Mutex>, +} + +#[derive(Debug)] +struct RunQueueInner { + steers: Vec, + followups: Vec, + collects: Vec, +} + +impl RunQueue { + /// 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 { + 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 Default for RunQueue { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod test; diff --git a/src/harness/run_queue/test.rs b/src/harness/run_queue/test.rs new file mode 100644 index 0000000..191b3bf --- /dev/null +++ b/src/harness/run_queue/test.rs @@ -0,0 +1,56 @@ +use super::*; + +#[tokio::test] +async fn new_queue_is_empty() { + let queue = RunQueue::::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); +} diff --git a/src/harness/run_queue/types.rs b/src/harness/run_queue/types.rs new file mode 100644 index 0000000..63457ef --- /dev/null +++ b/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, +} diff --git a/src/lib.rs b/src/lib.rs index f511507..6db8b27 100644 --- a/src/lib.rs +++ b/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; From 5e026cd8c2c6432390f5c2d9e11add6e384d4a55 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 12 Aug 2026 16:29:44 +0300 Subject: [PATCH 2/2] docs(harness): correct run queue module layout Co-authored-by: Medulla --- docs/modules/harness/README.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/docs/modules/harness/README.md b/docs/modules/harness/README.md index 25b1689..61ad8c9 100644 --- a/docs/modules/harness/README.md +++ b/docs/modules/harness/README.md @@ -166,7 +166,10 @@ src/harness/ prompt.rs providers.rs retry.rs - run_queue.rs + run_queue/ + mod.rs + test.rs + types.rs runtime.rs steering.rs stream.rs