From 90c759a2d9b1bda887e148f44958752ce6b6b582 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:15:17 +0000 Subject: [PATCH 01/29] chore(vendor): bump tinytools and tinyinference submodules Update the vendored tinytools and tinyinference submodules to their latest commits. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- vendor/tinyinference | 2 +- vendor/tinytools | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/vendor/tinyinference b/vendor/tinyinference index a1b070176..c29d5118e 160000 --- a/vendor/tinyinference +++ b/vendor/tinyinference @@ -1 +1 @@ -Subproject commit a1b0701766e8d35b77ce219e26089a6ed0774f28 +Subproject commit c29d5118eddfa12b83c266d7d5f07f6918dc3787 diff --git a/vendor/tinytools b/vendor/tinytools index f3970c173..82c0d975e 160000 --- a/vendor/tinytools +++ b/vendor/tinytools @@ -1 +1 @@ -Subproject commit f3970c173842253e7f986509e48f2363c489099f +Subproject commit 82c0d975e2667e1c525afe525006134431641785 From b4e1dc28a97d9a257858f8371c26b15f12a4825a Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:16:08 +0000 Subject: [PATCH 02/29] test(turn_runner): cover live turn ordering and queue snapshot fifo Add tests asserting that live_turns reports the scoped primary turn before parallel ones and that cancel_requests cancels both, and that queue_snapshot preserves per-lane FIFO order without draining queued work. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/turn_runner/mod_tests.rs | 107 ++++++++++++++++++ 1 file changed, 107 insertions(+) diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index a10622353..d397462d4 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -1886,3 +1886,110 @@ impl Summarizer for BlockingSummary { std::future::pending().await } } + +#[tokio::test] +async fn live_turns_and_cancel_requests_report_scoped_primary_before_parallel() { + let started = Arc::new(tokio::sync::Notify::new()); + let host = Arc::new(Host { + started: Some(started.clone()), + ..Host::default() + }); + let runner = Arc::new(runner( + host.clone(), + Arc::new(Ingress::default()), + Arc::new(Replies::default()), + Arc::new(Events::default()), + )); + + let mut primary_request = request(QueueDisposition::Interrupt, false); + primary_request.request_id = "primary".into(); + let scope = primary_request.scope.clone(); + let primary = { + let runner = runner.clone(); + tokio::spawn(async move { runner.run(primary_request).await }) + }; + started.notified().await; + + let mut parallel_request = request(QueueDisposition::Parallel, false); + parallel_request.request_id = "parallel".into(); + let parallel = { + let runner = runner.clone(); + tokio::spawn(async move { runner.run(parallel_request).await }) + }; + while host.runs.load(std::sync::atomic::Ordering::SeqCst) < 2 { + tokio::task::yield_now().await; + } + + let live = runner.live_turns().await; + assert_eq!( + live.iter() + .map(|turn| (turn.request_id.as_str(), turn.parallel)) + .collect::>(), + [("primary", false), ("parallel", true)] + ); + + assert_eq!(runner.cancel_requests(&scope, "thread-a", None).await, [ + "primary".to_owned(), + "parallel".to_owned(), + ]); + assert!(matches!(primary.await.unwrap(), Err(TurnRunnerError::Cancelled))); + assert!(matches!(parallel.await.unwrap(), Err(TurnRunnerError::Cancelled))); + assert!(runner.live_turns().await.is_empty()); +} + +#[tokio::test] +async fn queue_snapshot_preserves_lane_fifo_and_does_not_drain_work() { + let started = Arc::new(tokio::sync::Notify::new()); + let host = Arc::new(Host { + started: Some(started.clone()), + ..Host::default() + }); + let runner = Arc::new(runner( + host, + Arc::new(Ingress::default()), + Arc::new(Replies::default()), + Arc::new(Events::default()), + )); + let primary = { + let runner = runner.clone(); + tokio::spawn(async move { runner.run(request(QueueDisposition::Interrupt, false)).await }) + }; + started.notified().await; + + for (disposition, id) in [ + (QueueDisposition::Steer, "steer-1"), + (QueueDisposition::Steer, "steer-2"), + (QueueDisposition::Followup, "followup-1"), + (QueueDisposition::Collect, "collect-1"), + ] { + let mut queued = request(disposition, false); + queued.request_id = id.into(); + assert!(matches!(runner.run(queued).await, Err(TurnRunnerError::Queued))); + } + let scope = request(QueueDisposition::Interrupt, false).scope; + + let snapshot = runner.queue_snapshot(&scope, "thread-a").await; + assert_eq!( + snapshot + .iter() + .map(|(lane, item)| (*lane, item.request_id.as_str())) + .collect::>(), + [ + (QueueLane::Steer, "steer-1"), + (QueueLane::Steer, "steer-2"), + (QueueLane::Followup, "followup-1"), + (QueueLane::Collect, "collect-1"), + ] + ); + assert_eq!(runner.queue_status(&scope, "thread-a").await.total, 4); + assert_eq!( + runner + .take_queued(&scope, "thread-a", QueueLane::Steer, QueueMode::OneAtATime) + .await[0] + .request_id, + "steer-1" + ); + + runner.cancel_requests(&scope, "thread-a", None).await; + assert!(matches!(primary.await.unwrap(), Err(TurnRunnerError::Cancelled))); +} From fab8dd1572fb51137e15b178681a6debb6bee9aa Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:16:48 +0000 Subject: [PATCH 03/29] chore(vendor): bump tinytools submodule Update the vendored tinytools submodule to the latest upstream commit. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- vendor/tinytools | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinytools b/vendor/tinytools index 82c0d975e..62dbb1427 160000 --- a/vendor/tinytools +++ b/vendor/tinytools @@ -1 +1 @@ -Subproject commit 82c0d975e2667e1c525afe525006134431641785 +Subproject commit 62dbb1427fee6c5dae7e9621ec9d7aa822be11b1 From 09f9002729eebfc0e0cc0d1aa7fb4a80eaba7e38 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:17:04 +0000 Subject: [PATCH 04/29] chore(vendor): bump tinyinference submodule Update the vendored tinyinference submodule to the latest commit. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- vendor/tinyinference | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinyinference b/vendor/tinyinference index c29d5118e..8d33b806b 160000 --- a/vendor/tinyinference +++ b/vendor/tinyinference @@ -1 +1 @@ -Subproject commit c29d5118eddfa12b83c266d7d5f07f6918dc3787 +Subproject commit 8d33b806b556251e920f4a6034fde049732d1cee From 5b52406d365a282d0b41d34511d3cdd6c9e7275b Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:20:23 +0000 Subject: [PATCH 05/29] feat(turn_runner): expose live turns and return cancelled request ids Adds a public LiveTurn type describing an active turn's scope, thread, request id, and parallel flag so hosts can observe and cancel turns without touching cancellation tokens or queue internals. Cancellation now goes through cancel_requests, which returns the affected request ids with the primary turn first and parallel turns in request-id order, while the existing boolean helper delegates to it. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../tinyagents-harness/src/turn_runner/mod.rs | 2 +- .../src/turn_runner/mod_tests.rs | 34 +++++-- .../src/turn_runner/types.rs | 99 +++++++++++++++++-- 3 files changed, 118 insertions(+), 17 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/mod.rs b/crates/tinyagents-harness/src/turn_runner/mod.rs index 883f5e849..cf0665a4f 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod.rs @@ -10,7 +10,7 @@ mod types; pub use types::{ EventSink, FailureClassification, FailureClassifier, IngressDecision, IngressPolicy, - IngressStage, QueueDisposition, ReplyStore, RequeuePayload, ScopeKey, Summarizer, + IngressStage, LiveTurn, QueueDisposition, ReplyStore, RequeuePayload, ScopeKey, Summarizer, TerminalEvent, TurnHost, TurnRequest, TurnResult, TurnRunner, TurnRunnerConfig, TurnRunnerError, TurnSession, TurnSignal, WorkTracker, deterministic_reply_id, }; diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index d397462d4..77f743bd9 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -1928,12 +1928,18 @@ async fn live_turns_and_cancel_requests_report_scoped_primary_before_parallel() [("primary", false), ("parallel", true)] ); - assert_eq!(runner.cancel_requests(&scope, "thread-a", None).await, [ - "primary".to_owned(), - "parallel".to_owned(), - ]); - assert!(matches!(primary.await.unwrap(), Err(TurnRunnerError::Cancelled))); - assert!(matches!(parallel.await.unwrap(), Err(TurnRunnerError::Cancelled))); + assert_eq!( + runner.cancel_requests(&scope, "thread-a", None).await, + ["primary".to_owned(), "parallel".to_owned(),] + ); + assert!(matches!( + primary.await.unwrap(), + Err(TurnRunnerError::Cancelled) + )); + assert!(matches!( + parallel.await.unwrap(), + Err(TurnRunnerError::Cancelled) + )); assert!(runner.live_turns().await.is_empty()); } @@ -1952,7 +1958,11 @@ async fn queue_snapshot_preserves_lane_fifo_and_does_not_drain_work() { )); let primary = { let runner = runner.clone(); - tokio::spawn(async move { runner.run(request(QueueDisposition::Interrupt, false)).await }) + tokio::spawn(async move { + runner + .run(request(QueueDisposition::Interrupt, false)) + .await + }) }; started.notified().await; @@ -1964,7 +1974,10 @@ async fn queue_snapshot_preserves_lane_fifo_and_does_not_drain_work() { ] { let mut queued = request(disposition, false); queued.request_id = id.into(); - assert!(matches!(runner.run(queued).await, Err(TurnRunnerError::Queued))); + assert!(matches!( + runner.run(queued).await, + Err(TurnRunnerError::Queued) + )); } let scope = request(QueueDisposition::Interrupt, false).scope; @@ -1991,5 +2004,8 @@ async fn queue_snapshot_preserves_lane_fifo_and_does_not_drain_work() { ); runner.cancel_requests(&scope, "thread-a", None).await; - assert!(matches!(primary.await.unwrap(), Err(TurnRunnerError::Cancelled))); + assert!(matches!( + primary.await.unwrap(), + Err(TurnRunnerError::Cancelled) + )); } diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index b263f6cc6..63daa9d68 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -317,6 +317,20 @@ struct ActiveTurn

{ queue: Arc>>, } +/// A live turn owned by a runner, suitable for host-side cancellation and +/// observability without exposing cancellation tokens or queue internals. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LiveTurn { + /// Agent/profile scope that owns this turn. + pub scope: ScopeKey, + /// Conversation thread supplied by the host. + pub thread_id: String, + /// Host request identity used to scope cancellation and terminal events. + pub request_id: String, + /// Whether this turn runs on an independent parallel fork. + pub parallel: bool, +} + type ThreadLockKey = (ScopeKey, String); type ThreadLocks = Mutex>>>; type ParallelQueueKey = (ScopeKey, String, String); @@ -390,34 +404,50 @@ where thread_id: &str, request_id: Option<&str>, ) -> bool { + !self + .cancel_requests(scope, thread_id, request_id) + .await + .is_empty() + } + + /// Cancel matching active turns and return their request IDs, with the + /// primary turn first followed by parallel turns in request-ID order. + /// A stale request ID is a no-op and cannot cancel a newer generation. + pub async fn cancel_requests( + &self, + scope: &ScopeKey, + thread_id: &str, + request_id: Option<&str>, + ) -> Vec { let key = (scope.clone(), thread_id.to_owned()); let active = self.active.lock().await; - let cancelled_primary = if let Some(turn) = active + let mut cancelled = Vec::new(); + if let Some(turn) = active .get(&key) .filter(|turn| request_id.is_none_or(|id| turn.request_id == id)) { turn.queue.clear().await; turn.cancel.cancel(); - true - } else { - false - }; + cancelled.push(turn.request_id.clone()); + } drop(active); let mut parallel = self.parallel_active.lock().await; - let keys: Vec<_> = parallel + let mut keys: Vec<_> = parallel .keys() .filter(|(owner, thread, req)| { owner == scope && thread == thread_id && request_id.is_none_or(|id| req == id) }) .cloned() .collect(); + keys.sort_by(|left, right| left.2.cmp(&right.2)); for key in &keys { if let Some(token) = parallel.remove(key) { token.cancel(); } self.parallel_queues.lock().await.remove(key); + cancelled.push(key.2.clone()); } - cancelled_primary || !keys.is_empty() + cancelled } /// Whether the thread owns a primary user-authored turn cancellation slot. @@ -436,6 +466,46 @@ where .map(|turn| turn.request_id.clone()) } + /// Snapshot every active primary and parallel turn in deterministic order. + /// The result contains identities only; cancellation remains owned by the + /// runner and must go through [`Self::cancel_requests`]. + pub async fn live_turns(&self) -> Vec { + let mut live = Vec::new(); + for ((scope, thread_id), turn) in self.active.lock().await.iter() { + live.push(LiveTurn { + scope: scope.clone(), + thread_id: thread_id.clone(), + request_id: turn.request_id.clone(), + parallel: false, + }); + } + for ((scope, thread_id, request_id), _) in self.parallel_active.lock().await.iter() { + live.push(LiveTurn { + scope: scope.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + parallel: true, + }); + } + live.sort_by(|left, right| { + ( + left.scope.agent_id.as_str(), + left.scope.profile_id.as_deref(), + left.thread_id.as_str(), + left.parallel, + left.request_id.as_str(), + ) + .cmp(&( + right.scope.agent_id.as_str(), + right.scope.profile_id.as_deref(), + right.thread_id.as_str(), + right.parallel, + right.request_id.as_str(), + )) + }); + live + } + pub async fn run(&self, request: TurnRequest

) -> Result { let mut state = None; let mut next = None; @@ -965,6 +1035,21 @@ where } } + /// Non-destructively snapshot the active primary turn's queued requests in + /// lane order, preserving FIFO order within each lane. + pub async fn queue_snapshot( + &self, + scope: &ScopeKey, + thread_id: &str, + ) -> Vec<(QueueLane, TurnRequest

)> { + let key = (scope.clone(), thread_id.to_owned()); + let active = self.active.lock().await; + match active.get(&key) { + Some(turn) => turn.queue.snapshot().await, + None => Vec::new(), + } + } + pub async fn clear_queue(&self, scope: &ScopeKey, thread_id: &str) -> usize { let key = (scope.clone(), thread_id.to_owned()); let active = self.active.lock().await; From fa0970635347694e7f2af3bf847755d8efb28b87 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:22:52 +0000 Subject: [PATCH 06/29] test(turn_runner): gate parallel turn tests on a release signal Two turn-runner tests now hand the host a release notifier alongside the started one, so the parallel work can be held until the test is ready to observe it. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinyagents-harness/src/turn_runner/mod_tests.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index 77f743bd9..26e79be46 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -1890,8 +1890,10 @@ impl Summarizer for BlockingSummary { #[tokio::test] async fn live_turns_and_cancel_requests_report_scoped_primary_before_parallel() { let started = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); let host = Arc::new(Host { started: Some(started.clone()), + release: Some(release), ..Host::default() }); let runner = Arc::new(runner( @@ -1946,8 +1948,10 @@ async fn live_turns_and_cancel_requests_report_scoped_primary_before_parallel() #[tokio::test] async fn queue_snapshot_preserves_lane_fifo_and_does_not_drain_work() { let started = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); let host = Arc::new(Host { started: Some(started.clone()), + release: Some(release), ..Host::default() }); let runner = Arc::new(runner( From d7e114f0217f7f65a15d2e62bfc9371256e6e2f2 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:25:02 +0000 Subject: [PATCH 07/29] docs(turn_runner): document host inspection and cancellation APIs Adds a README section covering live_turns, cancel_requests, and queue_snapshot so hosts can inspect active turns and pending queue work without touching runner-owned tokens. It also notes that cancel remains a boolean wrapper and that unscoped cancellation orders the primary request ID before parallel ones. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../tinyagents-harness/src/turn_runner/README.md | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/crates/tinyagents-harness/src/turn_runner/README.md b/crates/tinyagents-harness/src/turn_runner/README.md index a64a4f70e..af98bc12f 100644 --- a/crates/tinyagents-harness/src/turn_runner/README.md +++ b/crates/tinyagents-harness/src/turn_runner/README.md @@ -60,4 +60,20 @@ existing sinks. When one generation failure retires queued requests, the runner rewrites each classification message to match that request's queued failure message while preserving its class and details. +## Host inspection and cancellation + +`live_turns` returns identity-only `LiveTurn` records for active primary and +parallel turns. Hosts can filter these records by `ScopeKey` and thread to +prepare cancellation notes without receiving runner-owned tokens. Use +`cancel_requests` to cancel matching turns and receive the request IDs it +actually stopped; when an unscoped cancellation matches multiple turns, the +primary request ID comes first, followed by parallel request IDs in +lexicographic order. The existing `cancel` method remains a boolean wrapper +for callers that only need to know whether anything matched. + +`queue_snapshot` copies the active primary queue without consuming it. Items +are tagged by lane in Steer, Followup, Collect order, with FIFO order preserved +within each lane. Hosts can render or classify pending work from the snapshot +and still leave subsequent draining to the runner. + Focused unit coverage lives in `mod_tests.rs` and `mod_api_tests.rs`. From be0fb60402f86dd503445bb11f791a4341104d52 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:25:49 +0000 Subject: [PATCH 08/29] test(turn_runner): cover live turns and cancellation across agents Extend the live turns and cancel requests test to run a third turn under a different agent scope, asserting that live turns report the agent id alongside the request id and that cancelling one agent's scope leaves the others running. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/turn_runner/mod_tests.rs | 33 +++++++++++++++++-- 1 file changed, 30 insertions(+), 3 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index 26e79be46..98be4103c 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -1918,16 +1918,33 @@ async fn live_turns_and_cancel_requests_report_scoped_primary_before_parallel() let runner = runner.clone(); tokio::spawn(async move { runner.run(parallel_request).await }) }; - while host.runs.load(std::sync::atomic::Ordering::SeqCst) < 2 { + + let mut other_agent_request = request(QueueDisposition::Interrupt, false); + other_agent_request.request_id = "other-agent".into(); + other_agent_request.scope.agent_id = "agent-b".into(); + let other_agent_scope = other_agent_request.scope.clone(); + let other_agent = { + let runner = runner.clone(); + tokio::spawn(async move { runner.run(other_agent_request).await }) + }; + while host.runs.load(std::sync::atomic::Ordering::SeqCst) < 3 { tokio::task::yield_now().await; } let live = runner.live_turns().await; assert_eq!( live.iter() - .map(|turn| (turn.request_id.as_str(), turn.parallel)) + .map(|turn| ( + turn.scope.agent_id.as_str(), + turn.request_id.as_str(), + turn.parallel, + )) .collect::>(), - [("primary", false), ("parallel", true)] + [ + ("agent-a", "primary", false), + ("agent-a", "parallel", true), + ("agent-b", "other-agent", false), + ] ); assert_eq!( @@ -1942,6 +1959,16 @@ async fn live_turns_and_cancel_requests_report_scoped_primary_before_parallel() parallel.await.unwrap(), Err(TurnRunnerError::Cancelled) )); + assert_eq!( + runner + .cancel_requests(&other_agent_scope, "thread-a", None) + .await, + ["other-agent".to_owned()] + ); + assert!(matches!( + other_agent.await.unwrap(), + Err(TurnRunnerError::Cancelled) + )); assert!(runner.live_turns().await.is_empty()); } From 19d4b9e880470d0cc6023eaa81dcd4c06daa7f93 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:30:04 +0000 Subject: [PATCH 09/29] refactor(turn_runner): iterate parallel_active keys directly Iterate over the map's keys instead of destructuring the key from each entry, dropping the unused value binding. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinyagents-harness/src/turn_runner/types.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index 63daa9d68..a0aa36f0c 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -479,7 +479,7 @@ where parallel: false, }); } - for ((scope, thread_id, request_id), _) in self.parallel_active.lock().await.iter() { + for (scope, thread_id, request_id) in self.parallel_active.lock().await.keys() { live.push(LiveTurn { scope: scope.clone(), thread_id: thread_id.clone(), From 82f197dca3928ed15fd4fa76380349a21d24ac87 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:33:50 +0000 Subject: [PATCH 10/29] test(turn_runner): cover completed execution not reported as cancelled Add a test asserting that cancelling a request while it is blocked in checkin does not report the already-completed execution as cancelled. The host gains a checkin_started notification so the test can cancel at the right moment. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/turn_runner/mod_tests.rs | 48 +++++++++++++++++++ 1 file changed, 48 insertions(+) diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index 98be4103c..c9765d719 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -8,6 +8,7 @@ struct Host { release: Option>, fingerprint_error: bool, hold_checkin: bool, + checkin_started: Option>, fail_run: bool, fingerprint_started: Option>, fingerprint_release: Option>, @@ -80,6 +81,12 @@ impl TurnHost for Host { } async fn checkin(&self, request: &TurnRequest) { + if self.hold_checkin + && request.request_id == "first" + && let Some(started) = &self.checkin_started + { + started.notify_one(); + } if self.hold_checkin && request.request_id == "first" && let Some(release) = &self.release @@ -92,6 +99,47 @@ impl TurnHost for Host { async fn cancel_children(&self, _request: &TurnRequest) {} } +#[tokio::test] +async fn completed_execution_is_not_reported_as_cancelled_during_checkin() { + let checkin_started = Arc::new(tokio::sync::Notify::new()); + let release_checkin = Arc::new(tokio::sync::Notify::new()); + let host = Arc::new(Host { + hold_checkin: true, + checkin_started: Some(checkin_started.clone()), + release: Some(release_checkin.clone()), + ..Host::default() + }); + let replies = Arc::new(Replies::default()); + let events = Arc::new(Events::default()); + let runner = Arc::new(runner( + host, + Arc::new(Ingress::default()), + replies.clone(), + events.clone(), + )); + let mut request = request(QueueDisposition::Interrupt, false); + request.request_id = "first".into(); + let scope = request.scope.clone(); + let run = { + let runner = runner.clone(); + tokio::spawn(async move { runner.run(request).await }) + }; + + checkin_started.notified().await; + assert!(runner + .cancel_requests(&scope, "thread-a", Some("first")) + .await + .is_empty()); + release_checkin.notify_one(); + + assert!(run.await.unwrap().is_ok()); + assert_eq!(replies.0.lock().unwrap().len(), 1); + assert!(matches!( + events.terminal.lock().unwrap().as_slice(), + [TerminalEvent::Completed { request_id, .. }] if request_id == "first" + )); +} + #[derive(Default)] struct Ingress(Mutex>); From 3ebf0d02cf9ccf5eacfc04f480b4ff77f28d079e Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:35:58 +0000 Subject: [PATCH 11/29] fix(turn_runner): close cancellation window before reporting turn completion Turns could be cancelled after their work had already finished but before completion was reported, so a successful run might be reported as cancelled. The runner now marks a turn non-cancellable once its result is ready and re-checks for a late cancellation, falling back to cancelling the generation if one arrived in the meantime. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/turn_runner/mod_tests.rs | 10 +-- .../src/turn_runner/types.rs | 64 ++++++++++++++++++- 2 files changed, 69 insertions(+), 5 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index c9765d719..113b08fd8 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -126,10 +126,12 @@ async fn completed_execution_is_not_reported_as_cancelled_during_checkin() { }; checkin_started.notified().await; - assert!(runner - .cancel_requests(&scope, "thread-a", Some("first")) - .await - .is_empty()); + assert!( + runner + .cancel_requests(&scope, "thread-a", Some("first")) + .await + .is_empty() + ); release_checkin.notify_one(); assert!(run.await.unwrap().is_ok()); diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index a0aa36f0c..7adf28411 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -313,6 +313,7 @@ pub enum TurnRunnerError { struct ActiveTurn

{ request_id: String, cancel: crate::CancellationToken, + cancellable: bool, generation: u64, queue: Arc>>, } @@ -424,7 +425,7 @@ where let mut cancelled = Vec::new(); if let Some(turn) = active .get(&key) - .filter(|turn| request_id.is_none_or(|id| turn.request_id == id)) + .filter(|turn| turn.cancellable && request_id.is_none_or(|id| turn.request_id == id)) { turn.queue.clear().await; turn.cancel.cancel(); @@ -472,6 +473,9 @@ where pub async fn live_turns(&self) -> Vec { let mut live = Vec::new(); for ((scope, thread_id), turn) in self.active.lock().await.iter() { + if !turn.cancellable { + continue; + } live.push(LiveTurn { scope: scope.clone(), thread_id: thread_id.clone(), @@ -644,6 +648,7 @@ where let state = ActiveTurn { request_id: request.request_id.clone(), cancel: cancellation.clone(), + cancellable: true, generation: self .next_generation .fetch_add(1, std::sync::atomic::Ordering::Relaxed), @@ -779,6 +784,15 @@ where deadline, ) .await; + if result.is_ok() + && !self + .close_cancellation_window(&request, active_generation, &cancellation) + .await + { + self.host.checkin(&request).await; + drop(_session_guard); + return self.cancel_generation(&request, active_generation).await; + } self.host.checkin(&request).await; drop(_session_guard); @@ -991,6 +1005,7 @@ where let reserved = ActiveTurn { request_id: request.request_id.clone(), cancel: crate::CancellationToken::new(), + cancellable: true, generation: self .next_generation .fetch_add(1, std::sync::atomic::Ordering::Relaxed), @@ -1237,6 +1252,53 @@ where } } + async fn close_cancellation_window( + &self, + request: &TurnRequest

, + generation: Option, + cancellation: &crate::CancellationToken, + ) -> bool { + if cancellation.is_cancelled() { + return false; + } + if request.host_authored { + return true; + } + if request.disposition == QueueDisposition::Parallel { + let key = ( + request.scope.clone(), + request.thread_id.clone(), + request.request_id.clone(), + ); + let mut active = self.parallel_active.lock().await; + let Some(token) = active.get(&key) else { + return !cancellation.is_cancelled(); + }; + if token.is_cancelled() { + return false; + } + active.remove(&key); + self.parallel_queues.lock().await.remove(&key); + return true; + } + let Some(generation) = generation else { + return true; + }; + let key = (request.scope.clone(), request.thread_id.clone()); + let mut active = self.active.lock().await; + let Some(turn) = active + .get_mut(&key) + .filter(|turn| turn.generation == generation) + else { + return !cancellation.is_cancelled(); + }; + if turn.cancel.is_cancelled() { + return false; + } + turn.cancellable = false; + true + } + async fn clear_parallel_tracking(&self, request: &TurnRequest

) { let key = ( request.scope.clone(), From b98ceefb540dc1b50bcaf03b850f1d5cd5b7bb87 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:38:47 +0000 Subject: [PATCH 12/29] fix(turn_runner): keep cancelled parallel turns visible until cleanup Cancelled parallel turns were removed from the active map immediately, so live_turns stopped reporting them while their cleanup was still running and a second cancel could no longer find them. Parallel turns now stay registered with a cancellable flag that is cleared on cancel, and live_turns reports active turns regardless of cancellability so callers see a turn until it actually finishes. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/turn_runner/mod_tests.rs | 58 +++++++++++++++++++ .../src/turn_runner/types.rs | 42 ++++++++------ 2 files changed, 83 insertions(+), 17 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index 113b08fd8..bf1c413ce 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -142,6 +142,64 @@ async fn completed_execution_is_not_reported_as_cancelled_during_checkin() { )); } +#[tokio::test] +async fn cancelled_parallel_turn_remains_visible_until_cleanup_finishes() { + let started = Arc::new(tokio::sync::Notify::new()); + let checkin_started = Arc::new(tokio::sync::Notify::new()); + let release_checkin = Arc::new(tokio::sync::Notify::new()); + let host = Arc::new(Host { + started: Some(started.clone()), + release: Some(release_checkin.clone()), + hold_checkin: true, + checkin_started: Some(checkin_started.clone()), + ..Host::default() + }); + let runner = Arc::new(runner( + host, + Arc::new(Ingress::default()), + Arc::new(Replies::default()), + Arc::new(Events::default()), + )); + let mut parallel_request = request(QueueDisposition::Parallel, false); + parallel_request.request_id = "first".into(); + let scope = parallel_request.scope.clone(); + let run = { + let runner = runner.clone(); + tokio::spawn(async move { runner.run(parallel_request).await }) + }; + + started.notified().await; + assert_eq!( + runner + .cancel_requests(&scope, "thread-a", Some("first")) + .await, + ["first"] + ); + checkin_started.notified().await; + assert_eq!( + runner + .live_turns() + .await + .iter() + .map(|turn| (turn.request_id.as_str(), turn.parallel)) + .collect::>(), + [("first", true)] + ); + assert!( + runner + .cancel_requests(&scope, "thread-a", Some("first")) + .await + .is_empty() + ); + + release_checkin.notify_one(); + assert!(matches!( + run.await.unwrap(), + Err(TurnRunnerError::Cancelled) + )); + assert!(runner.live_turns().await.is_empty()); +} + #[derive(Default)] struct Ingress(Mutex>); diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index 7adf28411..6b0fa3ee6 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -318,6 +318,12 @@ struct ActiveTurn

{ queue: Arc>>, } +#[derive(Clone)] +struct ActiveParallelTurn { + cancel: crate::CancellationToken, + cancellable: bool, +} + /// A live turn owned by a runner, suitable for host-side cancellation and /// observability without exposing cancellation tokens or queue internals. #[derive(Debug, Clone, PartialEq, Eq)] @@ -358,7 +364,7 @@ pub struct TurnRunner { locks: ThreadLocks, parallel_queues: ParallelQueueMap

, active: Mutex>>, - parallel_active: Mutex>, + parallel_active: Mutex>, next_generation: std::sync::atomic::AtomicU64, } @@ -434,18 +440,21 @@ where drop(active); let mut parallel = self.parallel_active.lock().await; let mut keys: Vec<_> = parallel - .keys() - .filter(|(owner, thread, req)| { - owner == scope && thread == thread_id && request_id.is_none_or(|id| req == id) + .iter() + .filter(|((owner, thread, req), turn)| { + turn.cancellable + && owner == scope + && thread == thread_id + && request_id.is_none_or(|id| req == id) }) - .cloned() + .map(|(key, _)| key.clone()) .collect(); keys.sort_by(|left, right| left.2.cmp(&right.2)); for key in &keys { - if let Some(token) = parallel.remove(key) { - token.cancel(); + if let Some(turn) = parallel.get_mut(key) { + turn.cancellable = false; + turn.cancel.cancel(); } - self.parallel_queues.lock().await.remove(key); cancelled.push(key.2.clone()); } cancelled @@ -473,9 +482,6 @@ where pub async fn live_turns(&self) -> Vec { let mut live = Vec::new(); for ((scope, thread_id), turn) in self.active.lock().await.iter() { - if !turn.cancellable { - continue; - } live.push(LiveTurn { scope: scope.clone(), thread_id: thread_id.clone(), @@ -483,7 +489,7 @@ where parallel: false, }); } - for (scope, thread_id, request_id) in self.parallel_active.lock().await.keys() { + for ((scope, thread_id, request_id), _) in self.parallel_active.lock().await.iter() { live.push(LiveTurn { scope: scope.clone(), thread_id: thread_id.clone(), @@ -617,7 +623,10 @@ where request.thread_id.clone(), request.request_id.clone(), ), - cancellation.clone(), + ActiveParallelTurn { + cancel: cancellation.clone(), + cancellable: true, + }, ); } else if let Some(reserved) = reservation.as_ref() { let mut active = self.active.lock().await; @@ -1271,14 +1280,13 @@ where request.request_id.clone(), ); let mut active = self.parallel_active.lock().await; - let Some(token) = active.get(&key) else { + let Some(turn) = active.get_mut(&key) else { return !cancellation.is_cancelled(); }; - if token.is_cancelled() { + if turn.cancel.is_cancelled() { return false; } - active.remove(&key); - self.parallel_queues.lock().await.remove(&key); + turn.cancellable = false; return true; } let Some(generation) = generation else { From 811df4a4d29ef0c1e5f0e05b2587b45deb184a07 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:39:17 +0000 Subject: [PATCH 13/29] refactor(turn_runner): store parallel cancellation tokens directly The ActiveParallelTurn wrapper was reduced to a bare CancellationToken in the parallel_active map, and the cancellable flag was dropped in favour of removing entries once they are cancelled or claimed. Cancelling a parallel turn now also clears its queue entry so stale queues do not linger after the turn is gone. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/turn_runner/types.rs | 39 +++++++------------ 1 file changed, 14 insertions(+), 25 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index 6b0fa3ee6..c2ab3cb96 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -318,12 +318,6 @@ struct ActiveTurn

{ queue: Arc>>, } -#[derive(Clone)] -struct ActiveParallelTurn { - cancel: crate::CancellationToken, - cancellable: bool, -} - /// A live turn owned by a runner, suitable for host-side cancellation and /// observability without exposing cancellation tokens or queue internals. #[derive(Debug, Clone, PartialEq, Eq)] @@ -364,7 +358,7 @@ pub struct TurnRunner { locks: ThreadLocks, parallel_queues: ParallelQueueMap

, active: Mutex>>, - parallel_active: Mutex>, + parallel_active: Mutex>, next_generation: std::sync::atomic::AtomicU64, } @@ -440,21 +434,18 @@ where drop(active); let mut parallel = self.parallel_active.lock().await; let mut keys: Vec<_> = parallel - .iter() - .filter(|((owner, thread, req), turn)| { - turn.cancellable - && owner == scope - && thread == thread_id - && request_id.is_none_or(|id| req == id) + .keys() + .filter(|(owner, thread, req)| { + owner == scope && thread == thread_id && request_id.is_none_or(|id| req == id) }) - .map(|(key, _)| key.clone()) + .cloned() .collect(); keys.sort_by(|left, right| left.2.cmp(&right.2)); for key in &keys { - if let Some(turn) = parallel.get_mut(key) { - turn.cancellable = false; - turn.cancel.cancel(); + if let Some(token) = parallel.remove(key) { + token.cancel(); } + self.parallel_queues.lock().await.remove(key); cancelled.push(key.2.clone()); } cancelled @@ -489,7 +480,7 @@ where parallel: false, }); } - for ((scope, thread_id, request_id), _) in self.parallel_active.lock().await.iter() { + for (scope, thread_id, request_id) in self.parallel_active.lock().await.keys() { live.push(LiveTurn { scope: scope.clone(), thread_id: thread_id.clone(), @@ -623,10 +614,7 @@ where request.thread_id.clone(), request.request_id.clone(), ), - ActiveParallelTurn { - cancel: cancellation.clone(), - cancellable: true, - }, + cancellation.clone(), ); } else if let Some(reserved) = reservation.as_ref() { let mut active = self.active.lock().await; @@ -1280,13 +1268,14 @@ where request.request_id.clone(), ); let mut active = self.parallel_active.lock().await; - let Some(turn) = active.get_mut(&key) else { + let Some(token) = active.get(&key) else { return !cancellation.is_cancelled(); }; - if turn.cancel.is_cancelled() { + if token.is_cancelled() { return false; } - turn.cancellable = false; + active.remove(&key); + self.parallel_queues.lock().await.remove(&key); return true; } let Some(generation) = generation else { From 16501cba736f4be59f530b21b27f0eadee870989 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:41:20 +0000 Subject: [PATCH 14/29] feat(turn_runner): add TurnQueueItem trait to decouple queue storage from requests Introduce a TurnQueueItem trait that wraps accepted requests so runners can store queue items without assuming the payload is a TurnRequest. TurnHost, TurnRunner, and the internal active-turn and parallel-queue maps are now parameterized by the queue item type, defaulting to TurnRequest for existing callers. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/turn_runner/types.rs | 60 +++++++++++++++---- 1 file changed, 50 insertions(+), 10 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index c2ab3cb96..738894519 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -37,6 +37,36 @@ pub struct TurnRequest

{ pub idle_timeout: Option, } +/// A request item stored by a runner and shared with its host's running agent +/// loop. Implementations must preserve the complete request in both directions. +pub trait TurnQueueItem

: Clone + Send + Sync + 'static { + /// Wraps an accepted request without changing its identity or payload. + fn from_turn_request(request: TurnRequest

) -> Self; + + /// Borrows the request represented by this item. + fn as_turn_request(&self) -> &TurnRequest

; + + /// Consumes this item into the request it represents. + fn into_turn_request(self) -> TurnRequest

; +} + +impl

TurnQueueItem

for TurnRequest

+where + P: Clone + Send + Sync + 'static, +{ + fn from_turn_request(request: TurnRequest

) -> Self { + request + } + + fn as_turn_request(&self) -> &TurnRequest

{ + self + } + + fn into_turn_request(self) -> TurnRequest

{ + self + } +} + /// Host-owned identity and fingerprint for the checked-out durable session. #[derive(Debug, Clone, PartialEq, Eq)] pub struct TurnSession { @@ -177,7 +207,7 @@ impl WorkTracker { } #[async_trait] -pub trait TurnHost

: Send + Sync { +pub trait TurnHost

>: Send + Sync { /// Check out or create the session for this request and verify its fingerprint. async fn checkout(&self, request: &TurnRequest

) -> Result; /// Check out an independent transcript/session snapshot for Parallel mode. @@ -200,7 +230,7 @@ pub trait TurnHost

: Send + Sync { session: &TurnSession, origin: &serde_json::Value, cancellation: crate::CancellationToken, - queue: Arc>>, + queue: Arc>, progress: mpsc::Sender, ) -> Result; /// Release the session after the run, including on cancellation or failure. @@ -310,12 +340,12 @@ pub enum TurnRunnerError { } #[derive(Clone)] -struct ActiveTurn

{ +struct ActiveTurn { request_id: String, cancel: crate::CancellationToken, cancellable: bool, generation: u64, - queue: Arc>>, + queue: Arc>, } /// A live turn owned by a runner, suitable for host-side cancellation and @@ -335,7 +365,7 @@ pub struct LiveTurn { type ThreadLockKey = (ScopeKey, String); type ThreadLocks = Mutex>>>; type ParallelQueueKey = (ScopeKey, String, String); -type ParallelQueueMap

= Mutex>>>>; +type ParallelQueueMap = Mutex>>>; fn queue_lane(disposition: QueueDisposition) -> Option { match disposition { @@ -347,7 +377,16 @@ fn queue_lane(disposition: QueueDisposition) -> Option { } /// Reusable turn lifecycle coordinator, parameterized entirely by host ports. -pub struct TurnRunner { +pub struct TurnRunner< + H, + I, + R, + E, + F, + S, + P = serde_json::Value, + Q = TurnRequest

, +> { host: Arc, ingress: Arc, replies: Arc, @@ -356,21 +395,22 @@ pub struct TurnRunner { summarizer: Arc, config: TurnRunnerConfig, locks: ThreadLocks, - parallel_queues: ParallelQueueMap

, - active: Mutex>>, + parallel_queues: ParallelQueueMap, + active: Mutex>>, parallel_active: Mutex>, next_generation: std::sync::atomic::AtomicU64, } -impl TurnRunner +impl TurnRunner where - H: TurnHost

+ 'static, + H: TurnHost + 'static, I: IngressPolicy

+ 'static, R: ReplyStore

+ 'static, E: EventSink

+ 'static, F: FailureClassifier + 'static, S: Summarizer

+ 'static, P: Clone + Send + Sync + 'static, + Q: TurnQueueItem

, { pub fn new( host: Arc, From 21e5da2c3f92a348ad7131e06aeff79de7c321d7 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:41:31 +0000 Subject: [PATCH 15/29] refactor(turn_runner): parameterize active turn state over queue item type ActiveTurn and the run queue now carry a generic queue item type Q instead of TurnRequest

directly, with conversions applied when items are pushed and drained. This lets callers supply their own queued representation while keeping the existing turn request behaviour unchanged. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/turn_runner/types.rs | 44 ++++++++++++++----- 1 file changed, 32 insertions(+), 12 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index 738894519..0fa78b178 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -601,9 +601,9 @@ where async fn run_one( &self, mut request: TurnRequest

, - reservation: Option>, - state_out: &mut Option>, - next_out: &mut Option<(TurnRequest

, ActiveTurn

)>, + reservation: Option>, + state_out: &mut Option>, + next_out: &mut Option<(TurnRequest

, ActiveTurn)>, ) -> Result { let deadline = TurnDeadline::new(request.deadline.or(self.config.deadline)); let cancellation = reservation @@ -678,7 +678,10 @@ where if let Some(lane) = queue_lane(request.disposition) && let Some(current) = active.get(&key) { - current.queue.push(lane, request.clone()).await; + current + .queue + .push(lane, Q::from_turn_request(request.clone())) + .await; return Err(TurnRunnerError::Queued); } let queue = Arc::new(RunQueue::new()); @@ -877,7 +880,7 @@ where } } - async fn parallel_queue_for(&self, key: ParallelQueueKey) -> Arc>> { + async fn parallel_queue_for(&self, key: ParallelQueueKey) -> Arc> { let mut queues = self.parallel_queues.lock().await; Arc::clone( queues @@ -926,7 +929,13 @@ where active.remove(&key); let mut queued = Vec::new(); for lane in [QueueLane::Steer, QueueLane::Followup, QueueLane::Collect] { - queued.extend(turn.queue.take(lane, QueueMode::All).await); + queued.extend( + turn.queue + .take(lane, QueueMode::All) + .await + .into_iter() + .map(TurnQueueItem::into_turn_request), + ); } queued } @@ -1009,7 +1018,10 @@ where scope: &ScopeKey, thread_id: &str, owner_generation: u64, - ) -> (Vec>, Option<(TurnRequest

, ActiveTurn

)>) { + ) -> ( + Vec>, + Option<(TurnRequest

, ActiveTurn)>, + ) { let key = (scope.clone(), thread_id.to_owned()); let mut active = self.active.lock().await; if active @@ -1021,13 +1033,20 @@ where let Some(current) = active.get(&key).cloned() else { return (Vec::new(), None); }; - let collected = current.queue.take(QueueLane::Collect, QueueMode::All).await; + let collected = current + .queue + .take(QueueLane::Collect, QueueMode::All) + .await + .into_iter() + .map(TurnQueueItem::into_turn_request) + .collect(); let next = current .queue .take(QueueLane::Steer, QueueMode::OneAtATime) .await .into_iter() - .next(); + .next() + .map(TurnQueueItem::into_turn_request); let next = if next.is_none() { current .queue @@ -1035,6 +1054,7 @@ where .await .into_iter() .next() + .map(TurnQueueItem::into_turn_request) } else { next }; @@ -1064,7 +1084,7 @@ where thread_id: &str, lane: QueueLane, mode: QueueMode, - ) -> Vec> { + ) -> Vec { let key = (scope.clone(), thread_id.to_owned()); let active = self.active.lock().await; match active.get(&key) { @@ -1093,7 +1113,7 @@ where &self, scope: &ScopeKey, thread_id: &str, - ) -> Vec<(QueueLane, TurnRequest

)> { + ) -> Vec<(QueueLane, Q)> { let key = (scope.clone(), thread_id.to_owned()); let active = self.active.lock().await; match active.get(&key) { @@ -1152,7 +1172,7 @@ where session: &TurnSession, origin: &serde_json::Value, cancellation: crate::CancellationToken, - queue: Arc>>, + queue: Arc>, deadline: TurnDeadline, ) -> Result { let (progress_tx, mut progress_rx) = mpsc::channel(32); From 59272562589b90ec0870b118aea38dce0db358c8 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:41:58 +0000 Subject: [PATCH 16/29] refactor(turn_runner): reexport TurnQueueItem and use it in queue removal The queue removal path now matches on the queue item's turn request rather than assuming the item is the request itself, and TurnQueueItem is reexported from the module. Formatting of the TurnRunner generics and queue_snapshot signature was also collapsed. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../tinyagents-harness/src/turn_runner/mod.rs | 2 +- .../src/turn_runner/types.rs | 21 ++++--------------- 2 files changed, 5 insertions(+), 18 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/mod.rs b/crates/tinyagents-harness/src/turn_runner/mod.rs index cf0665a4f..7776cf712 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod.rs @@ -11,7 +11,7 @@ mod types; pub use types::{ EventSink, FailureClassification, FailureClassifier, IngressDecision, IngressPolicy, IngressStage, LiveTurn, QueueDisposition, ReplyStore, RequeuePayload, ScopeKey, Summarizer, - TerminalEvent, TurnHost, TurnRequest, TurnResult, TurnRunner, TurnRunnerConfig, + TerminalEvent, TurnHost, TurnQueueItem, TurnRequest, TurnResult, TurnRunner, TurnRunnerConfig, TurnRunnerError, TurnSession, TurnSignal, WorkTracker, deterministic_reply_id, }; diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index 0fa78b178..a2c7104d7 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -377,16 +377,7 @@ fn queue_lane(disposition: QueueDisposition) -> Option { } /// Reusable turn lifecycle coordinator, parameterized entirely by host ports. -pub struct TurnRunner< - H, - I, - R, - E, - F, - S, - P = serde_json::Value, - Q = TurnRequest

, -> { +pub struct TurnRunner> { host: Arc, ingress: Arc, replies: Arc, @@ -1107,13 +1098,9 @@ where } } - /// Non-destructively snapshot the active primary turn's queued requests in + /// Non-destructively snapshot the active primary turn's host queue items in /// lane order, preserving FIFO order within each lane. - pub async fn queue_snapshot( - &self, - scope: &ScopeKey, - thread_id: &str, - ) -> Vec<(QueueLane, Q)> { + pub async fn queue_snapshot(&self, scope: &ScopeKey, thread_id: &str) -> Vec<(QueueLane, Q)> { let key = (scope.clone(), thread_id.to_owned()); let active = self.active.lock().await; match active.get(&key) { @@ -1159,7 +1146,7 @@ where match active.get(&key) { Some(turn) => { turn.queue - .remove_where(|turn| turn.request_id == request_id) + .remove_where(|turn| turn.as_turn_request().request_id == request_id) .await } None => 0, From 4537abddbe71a9491726c58e250460f403c5a204 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:42:11 +0000 Subject: [PATCH 17/29] refactor(turn_runner): drop unused type parameter from ActiveTurn ActiveTurn no longer stores a value of type P, so the parameter was removed from the struct and all of its uses. This simplifies the type signatures of run_one and the internal helpers without changing behaviour. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinyagents-harness/src/turn_runner/types.rs | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index a2c7104d7..b02ba4672 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -340,7 +340,7 @@ pub enum TurnRunnerError { } #[derive(Clone)] -struct ActiveTurn { +struct ActiveTurn { request_id: String, cancel: crate::CancellationToken, cancellable: bool, @@ -387,7 +387,7 @@ pub struct TurnRunner, - active: Mutex>>, + active: Mutex>>, parallel_active: Mutex>, next_generation: std::sync::atomic::AtomicU64, } @@ -592,9 +592,9 @@ where async fn run_one( &self, mut request: TurnRequest

, - reservation: Option>, - state_out: &mut Option>, - next_out: &mut Option<(TurnRequest

, ActiveTurn)>, + reservation: Option>, + state_out: &mut Option>, + next_out: &mut Option<(TurnRequest

, ActiveTurn)>, ) -> Result { let deadline = TurnDeadline::new(request.deadline.or(self.config.deadline)); let cancellation = reservation @@ -1011,7 +1011,7 @@ where owner_generation: u64, ) -> ( Vec>, - Option<(TurnRequest

, ActiveTurn)>, + Option<(TurnRequest

, ActiveTurn)>, ) { let key = (scope.clone(), thread_id.to_owned()); let mut active = self.active.lock().await; From dfdbac7535e5b19676ea94e41c4db257c63956ea Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:42:28 +0000 Subject: [PATCH 18/29] fix(turn_runner): add phantom payload marker to TurnRunner The struct's generic parameter P was otherwise unused, so a PhantomData field was added to keep it part of the type and satisfy the compiler's variance and unused-parameter checks. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinyagents-harness/src/turn_runner/types.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index b02ba4672..ca7f19fbf 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -390,6 +390,7 @@ pub struct TurnRunner>>, parallel_active: Mutex>, next_generation: std::sync::atomic::AtomicU64, + payload: std::marker::PhantomData P>, } impl TurnRunner @@ -425,6 +426,7 @@ where active: Mutex::new(HashMap::new()), parallel_active: Mutex::new(HashMap::new()), next_generation: std::sync::atomic::AtomicU64::new(1), + payload: std::marker::PhantomData, } } From f0ec783be4a18c8d51f7cfc1d4372c770d5a06ed Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:43:17 +0000 Subject: [PATCH 19/29] fix(turn_runner): keep parallel turn entries after cancellation Parallel turns are no longer removed from the active map when cancelled or completed; instead their cancellable flag is cleared so the entry stays visible for observability while repeated cancellation is prevented. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/turn_runner/types.rs | 39 ++++++++++++------- 1 file changed, 25 insertions(+), 14 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index ca7f19fbf..cf1d5b6b3 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -348,6 +348,12 @@ struct ActiveTurn { queue: Arc>, } +#[derive(Clone)] +struct ActiveParallelTurn { + cancel: crate::CancellationToken, + cancellable: bool, +} + /// A live turn owned by a runner, suitable for host-side cancellation and /// observability without exposing cancellation tokens or queue internals. #[derive(Debug, Clone, PartialEq, Eq)] @@ -388,7 +394,7 @@ pub struct TurnRunner, active: Mutex>>, - parallel_active: Mutex>, + parallel_active: Mutex>, next_generation: std::sync::atomic::AtomicU64, payload: std::marker::PhantomData P>, } @@ -467,18 +473,21 @@ where drop(active); let mut parallel = self.parallel_active.lock().await; let mut keys: Vec<_> = parallel - .keys() - .filter(|(owner, thread, req)| { - owner == scope && thread == thread_id && request_id.is_none_or(|id| req == id) + .iter() + .filter(|((owner, thread, req), turn)| { + turn.cancellable + && owner == scope + && thread == thread_id + && request_id.is_none_or(|id| req == id) }) - .cloned() + .map(|(key, _)| key.clone()) .collect(); keys.sort_by(|left, right| left.2.cmp(&right.2)); for key in &keys { - if let Some(token) = parallel.remove(key) { - token.cancel(); + if let Some(turn) = parallel.get_mut(key) { + turn.cancellable = false; + turn.cancel.cancel(); } - self.parallel_queues.lock().await.remove(key); cancelled.push(key.2.clone()); } cancelled @@ -513,7 +522,7 @@ where parallel: false, }); } - for (scope, thread_id, request_id) in self.parallel_active.lock().await.keys() { + for ((scope, thread_id, request_id), _) in self.parallel_active.lock().await.iter() { live.push(LiveTurn { scope: scope.clone(), thread_id: thread_id.clone(), @@ -647,7 +656,10 @@ where request.thread_id.clone(), request.request_id.clone(), ), - cancellation.clone(), + ActiveParallelTurn { + cancel: cancellation.clone(), + cancellable: true, + }, ); } else if let Some(reserved) = reservation.as_ref() { let mut active = self.active.lock().await; @@ -1317,14 +1329,13 @@ where request.request_id.clone(), ); let mut active = self.parallel_active.lock().await; - let Some(token) = active.get(&key) else { + let Some(turn) = active.get_mut(&key) else { return !cancellation.is_cancelled(); }; - if token.is_cancelled() { + if turn.cancel.is_cancelled() { return false; } - active.remove(&key); - self.parallel_queues.lock().await.remove(&key); + turn.cancellable = false; return true; } let Some(generation) = generation else { From f2f7b4016462c618e5b0d2407a618e9288ca8b05 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:44:13 +0000 Subject: [PATCH 20/29] refactor(turn_runner): store cancellation tokens directly for parallel turns The ActiveParallelTurn wrapper was reduced to a bare CancellationToken in the parallel_active map, since the cancellable flag was only ever used to gate cancellation and can be expressed by removing the entry instead. Cancelling a parallel turn now removes its token and queue entry outright, which also prevents stale queue state from lingering after cancellation. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/turn_runner/types.rs | 44 +++++++------------ 1 file changed, 15 insertions(+), 29 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index cf1d5b6b3..a54823414 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -348,12 +348,6 @@ struct ActiveTurn { queue: Arc>, } -#[derive(Clone)] -struct ActiveParallelTurn { - cancel: crate::CancellationToken, - cancellable: bool, -} - /// A live turn owned by a runner, suitable for host-side cancellation and /// observability without exposing cancellation tokens or queue internals. #[derive(Debug, Clone, PartialEq, Eq)] @@ -394,7 +388,7 @@ pub struct TurnRunner, active: Mutex>>, - parallel_active: Mutex>, + parallel_active: Mutex>, next_generation: std::sync::atomic::AtomicU64, payload: std::marker::PhantomData P>, } @@ -473,21 +467,18 @@ where drop(active); let mut parallel = self.parallel_active.lock().await; let mut keys: Vec<_> = parallel - .iter() - .filter(|((owner, thread, req), turn)| { - turn.cancellable - && owner == scope - && thread == thread_id - && request_id.is_none_or(|id| req == id) + .keys() + .filter(|(owner, thread, req)| { + owner == scope && thread == thread_id && request_id.is_none_or(|id| req == id) }) - .map(|(key, _)| key.clone()) + .cloned() .collect(); keys.sort_by(|left, right| left.2.cmp(&right.2)); for key in &keys { - if let Some(turn) = parallel.get_mut(key) { - turn.cancellable = false; - turn.cancel.cancel(); + if let Some(token) = parallel.remove(key) { + token.cancel(); } + self.parallel_queues.lock().await.remove(key); cancelled.push(key.2.clone()); } cancelled @@ -522,7 +513,7 @@ where parallel: false, }); } - for ((scope, thread_id, request_id), _) in self.parallel_active.lock().await.iter() { + for (scope, thread_id, request_id) in self.parallel_active.lock().await.keys() { live.push(LiveTurn { scope: scope.clone(), thread_id: thread_id.clone(), @@ -656,10 +647,7 @@ where request.thread_id.clone(), request.request_id.clone(), ), - ActiveParallelTurn { - cancel: cancellation.clone(), - cancellable: true, - }, + cancellation.clone(), ); } else if let Some(reserved) = reservation.as_ref() { let mut active = self.active.lock().await; @@ -1023,10 +1011,7 @@ where scope: &ScopeKey, thread_id: &str, owner_generation: u64, - ) -> ( - Vec>, - Option<(TurnRequest

, ActiveTurn)>, - ) { + ) -> (Vec>, Option<(TurnRequest

, ActiveTurn)>) { let key = (scope.clone(), thread_id.to_owned()); let mut active = self.active.lock().await; if active @@ -1329,13 +1314,14 @@ where request.request_id.clone(), ); let mut active = self.parallel_active.lock().await; - let Some(turn) = active.get_mut(&key) else { + let Some(token) = active.get(&key) else { return !cancellation.is_cancelled(); }; - if turn.cancel.is_cancelled() { + if token.is_cancelled() { return false; } - turn.cancellable = false; + active.remove(&key); + self.parallel_queues.lock().await.remove(&key); return true; } let Some(generation) = generation else { From 1c83176455c4b2d4648f5ca30069b74e97a8537f Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:45:29 +0000 Subject: [PATCH 21/29] feat(turn_runner): support custom queue item types in turn host The turn host trait is now generic over the queue item type, letting hosts supply their own wrapper around TurnRequest while still sharing the runner's queue. Tests cover that a custom item round-trips the full request and lands in the same queue the host observes. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/turn_runner/mod_tests.rs | 189 +++++++++++++++++- .../src/turn_runner/types.rs | 39 ++-- 2 files changed, 213 insertions(+), 15 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index bf1c413ce..0ed931f23 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -1,6 +1,26 @@ use super::*; use crate::run_queue::{QueueLane, QueueMode, RunQueue}; -use std::sync::{Arc, Mutex}; +use std::{ + sync::{Arc, Mutex}, + time::Duration, +}; + +#[derive(Clone)] +struct WrappedQueueRequest(TurnRequest); + +impl TurnQueueItem for WrappedQueueRequest { + fn from_turn_request(request: TurnRequest) -> Self { + Self(request) + } + + fn as_turn_request(&self) -> &TurnRequest { + &self.0 + } + + fn into_turn_request(self) -> TurnRequest { + self.0 + } +} #[derive(Default)] struct Host { @@ -14,6 +34,7 @@ struct Host { fingerprint_release: Option>, checkins: std::sync::atomic::AtomicUsize, runs: std::sync::atomic::AtomicUsize, + wrapped_queue: Mutex>>>, } #[async_trait::async_trait] @@ -99,6 +120,61 @@ impl TurnHost for Host { async fn cancel_children(&self, _request: &TurnRequest) {} } +#[async_trait::async_trait] +impl TurnHost for Host { + async fn checkout(&self, request: &TurnRequest) -> Result { + ::checkout(self, request).await + } + + async fn checkout_fork(&self, request: &TurnRequest) -> Result { + ::checkout_fork(self, request).await + } + + async fn validate_fingerprint( + &self, + request: &TurnRequest, + session: &TurnSession, + ) -> Result<(), String> { + ::validate_fingerprint(self, request, session).await + } + + fn run_origin(&self, request: &TurnRequest) -> serde_json::Value { + ::run_origin(self, request) + } + + async fn run( + &self, + _request: &TurnRequest, + _session: &TurnSession, + _origin: &serde_json::Value, + cancellation: crate::CancellationToken, + queue: Arc>, + _progress: tokio::sync::mpsc::Sender, + ) -> Result { + *self.wrapped_queue.lock().unwrap() = Some(queue); + self.runs.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if let (Some(started), Some(release)) = (&self.started, &self.release) { + started.notify_one(); + tokio::select! { + _ = release.notified() => {}, + _ = cancellation.cancelled() => return Err("cancelled".into()), + } + } + Ok(TurnResult { + reply: "done".into(), + data: serde_json::Value::Null, + }) + } + + async fn checkin(&self, request: &TurnRequest) { + ::checkin(self, request).await + } + + async fn cancel_children(&self, request: &TurnRequest) { + ::cancel_children(self, request).await + } +} + #[tokio::test] async fn completed_execution_is_not_reported_as_cancelled_during_checkin() { let checkin_started = Arc::new(tokio::sync::Notify::new()); @@ -200,6 +276,117 @@ async fn cancelled_parallel_turn_remains_visible_until_cleanup_finishes() { assert!(runner.live_turns().await.is_empty()); } +#[tokio::test] +async fn custom_queue_item_shares_runner_queue_and_preserves_complete_request() { + let started = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); + let host = Arc::new(Host { + started: Some(started.clone()), + release: Some(release), + ..Host::default() + }); + let runner = Arc::new(TurnRunner::< + _, + _, + _, + _, + _, + _, + serde_json::Value, + WrappedQueueRequest, + >::new( + host.clone(), + Arc::new(Ingress::default()), + Arc::new(Replies::default()), + Arc::new(Events::default()), + Arc::new(Failures), + Arc::new(Summary), + TurnRunnerConfig::default(), + )); + + let mut primary = request(QueueDisposition::Interrupt, false); + primary.request_id = "primary".into(); + let scope = primary.scope.clone(); + let primary_run = { + let runner = runner.clone(); + tokio::spawn(async move { runner.run(primary).await }) + }; + started.notified().await; + + let mut runner_queued = request(QueueDisposition::Followup, true); + runner_queued.scope.profile_id = Some("profile-x".into()); + runner_queued.thread_id = "thread-from-runner".into(); + runner_queued.request_id = "runner-queued-id".into(); + runner_queued.text = "runner text".into(); + runner_queued.payload = serde_json::json!({"client":"runner","attachment":[1,2]}); + runner_queued.deadline = Some(Duration::from_secs(31)); + runner_queued.idle_timeout = Some(Duration::from_secs(7)); + assert!(matches!( + runner.run(runner_queued.clone()).await, + Err(TurnRunnerError::Queued) + )); + + let shared_queue = host.wrapped_queue.lock().unwrap().clone().unwrap(); + let runner_item = shared_queue + .snapshot() + .await + .into_iter() + .find(|(_, item)| item.as_turn_request().request_id == "runner-queued-id") + .expect("runner-enqueued item is in the host's queue"); + assert_eq!(runner_item.0, QueueLane::Followup); + let stored_runner_request = runner_item.1.as_turn_request(); + assert_eq!(stored_runner_request.scope, runner_queued.scope); + assert_eq!(stored_runner_request.thread_id, runner_queued.thread_id); + assert_eq!(stored_runner_request.request_id, runner_queued.request_id); + assert_eq!(stored_runner_request.text, runner_queued.text); + assert_eq!(stored_runner_request.payload, runner_queued.payload); + assert_eq!( + stored_runner_request.host_authored, + runner_queued.host_authored + ); + assert_eq!(stored_runner_request.disposition, runner_queued.disposition); + assert_eq!(stored_runner_request.deadline, runner_queued.deadline); + assert_eq!( + stored_runner_request.idle_timeout, + runner_queued.idle_timeout + ); + + let mut host_queued = request(QueueDisposition::Steer, false); + host_queued.request_id = "host-queued-id".into(); + host_queued.text = "host text".into(); + host_queued.payload = serde_json::json!({"client":"host","locale":"en"}); + shared_queue + .push( + QueueLane::Steer, + WrappedQueueRequest::from_turn_request(host_queued.clone()), + ) + .await; + let runner_snapshot = runner.queue_snapshot(&scope, "thread-a").await; + let host_item = runner_snapshot + .iter() + .find(|(_, item)| item.as_turn_request().request_id == "host-queued-id") + .expect("host-enqueued item is visible in runner's snapshot"); + assert_eq!(host_item.0, QueueLane::Steer); + assert_eq!(host_item.1.as_turn_request().scope, host_queued.scope); + assert_eq!( + host_item.1.as_turn_request().request_id, + host_queued.request_id + ); + assert_eq!(host_item.1.as_turn_request().text, host_queued.text); + assert_eq!(host_item.1.as_turn_request().payload, host_queued.payload); + + assert_eq!( + runner + .cancel_requests(&scope, "thread-a", Some("primary")) + .await, + ["primary"] + ); + assert!(matches!( + primary_run.await.unwrap(), + Err(TurnRunnerError::Cancelled) + )); +} + #[derive(Default)] struct Ingress(Mutex>); diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index a54823414..7f9d15583 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -348,6 +348,12 @@ struct ActiveTurn { queue: Arc>, } +#[derive(Clone)] +struct ActiveParallelTurn { + cancel: crate::CancellationToken, + cancellable: bool, +} + /// A live turn owned by a runner, suitable for host-side cancellation and /// observability without exposing cancellation tokens or queue internals. #[derive(Debug, Clone, PartialEq, Eq)] @@ -388,7 +394,7 @@ pub struct TurnRunner, active: Mutex>>, - parallel_active: Mutex>, + parallel_active: Mutex>, next_generation: std::sync::atomic::AtomicU64, payload: std::marker::PhantomData P>, } @@ -467,18 +473,21 @@ where drop(active); let mut parallel = self.parallel_active.lock().await; let mut keys: Vec<_> = parallel - .keys() - .filter(|(owner, thread, req)| { - owner == scope && thread == thread_id && request_id.is_none_or(|id| req == id) + .iter() + .filter(|((owner, thread, req), turn)| { + turn.cancellable + && owner == scope + && thread == thread_id + && request_id.is_none_or(|id| req == id) }) - .cloned() + .map(|(key, _)| key.clone()) .collect(); keys.sort_by(|left, right| left.2.cmp(&right.2)); for key in &keys { - if let Some(token) = parallel.remove(key) { - token.cancel(); + if let Some(turn) = parallel.get_mut(key) { + turn.cancellable = false; + turn.cancel.cancel(); } - self.parallel_queues.lock().await.remove(key); cancelled.push(key.2.clone()); } cancelled @@ -513,7 +522,7 @@ where parallel: false, }); } - for (scope, thread_id, request_id) in self.parallel_active.lock().await.keys() { + for ((scope, thread_id, request_id), _) in self.parallel_active.lock().await.iter() { live.push(LiveTurn { scope: scope.clone(), thread_id: thread_id.clone(), @@ -647,7 +656,10 @@ where request.thread_id.clone(), request.request_id.clone(), ), - cancellation.clone(), + ActiveParallelTurn { + cancel: cancellation.clone(), + cancellable: true, + }, ); } else if let Some(reserved) = reservation.as_ref() { let mut active = self.active.lock().await; @@ -1314,14 +1326,13 @@ where request.request_id.clone(), ); let mut active = self.parallel_active.lock().await; - let Some(token) = active.get(&key) else { + let Some(turn) = active.get_mut(&key) else { return !cancellation.is_cancelled(); }; - if token.is_cancelled() { + if turn.cancel.is_cancelled() { return false; } - active.remove(&key); - self.parallel_queues.lock().await.remove(&key); + turn.cancellable = false; return true; } let Some(generation) = generation else { From 341e54469e56e9dfe83830fcb8a4a62d6bb21909 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:45:39 +0000 Subject: [PATCH 22/29] test(turn_runner): split wrapped queue host into its own type The wrapped-queue TurnHost implementation previously lived on the shared Host test double, which forced the wrapped_queue field onto every test that used Host. A dedicated WrappedHost now wraps Host and delegates the common methods, so the queue field only exists where it is needed. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/turn_runner/mod_tests.rs | 26 ++++++++++++------- 1 file changed, 16 insertions(+), 10 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index 0ed931f23..9bcd96ffd 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -34,7 +34,6 @@ struct Host { fingerprint_release: Option>, checkins: std::sync::atomic::AtomicUsize, runs: std::sync::atomic::AtomicUsize, - wrapped_queue: Mutex>>>, } #[async_trait::async_trait] @@ -120,14 +119,19 @@ impl TurnHost for Host { async fn cancel_children(&self, _request: &TurnRequest) {} } +struct WrappedHost { + inner: Host, + wrapped_queue: Mutex>>>, +} + #[async_trait::async_trait] -impl TurnHost for Host { +impl TurnHost for WrappedHost { async fn checkout(&self, request: &TurnRequest) -> Result { - ::checkout(self, request).await + ::checkout(&self.inner, request).await } async fn checkout_fork(&self, request: &TurnRequest) -> Result { - ::checkout_fork(self, request).await + ::checkout_fork(&self.inner, request).await } async fn validate_fingerprint( @@ -135,11 +139,11 @@ impl TurnHost for Host { request: &TurnRequest, session: &TurnSession, ) -> Result<(), String> { - ::validate_fingerprint(self, request, session).await + ::validate_fingerprint(&self.inner, request, session).await } fn run_origin(&self, request: &TurnRequest) -> serde_json::Value { - ::run_origin(self, request) + ::run_origin(&self.inner, request) } async fn run( @@ -152,8 +156,10 @@ impl TurnHost for Host { _progress: tokio::sync::mpsc::Sender, ) -> Result { *self.wrapped_queue.lock().unwrap() = Some(queue); - self.runs.fetch_add(1, std::sync::atomic::Ordering::SeqCst); - if let (Some(started), Some(release)) = (&self.started, &self.release) { + self.inner + .runs + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if let (Some(started), Some(release)) = (&self.inner.started, &self.inner.release) { started.notify_one(); tokio::select! { _ = release.notified() => {}, @@ -167,11 +173,11 @@ impl TurnHost for Host { } async fn checkin(&self, request: &TurnRequest) { - ::checkin(self, request).await + ::checkin(&self.inner, request).await } async fn cancel_children(&self, request: &TurnRequest) { - ::cancel_children(self, request).await + ::cancel_children(&self.inner, request).await } } From 15ba1bc56cff16f54ef2a827683239f25f879854 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:45:48 +0000 Subject: [PATCH 23/29] test(turn_runner): wrap host in WrappedHost for queue test The custom queue item test now constructs a WrappedHost around the default Host so the runner exercises the wrapped queue path, matching the production host setup. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../tinyagents-harness/src/turn_runner/mod_tests.rs | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index 9bcd96ffd..35de385ee 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -286,10 +286,13 @@ async fn cancelled_parallel_turn_remains_visible_until_cleanup_finishes() { async fn custom_queue_item_shares_runner_queue_and_preserves_complete_request() { let started = Arc::new(tokio::sync::Notify::new()); let release = Arc::new(tokio::sync::Notify::new()); - let host = Arc::new(Host { - started: Some(started.clone()), - release: Some(release), - ..Host::default() + let host = Arc::new(WrappedHost { + inner: Host { + started: Some(started.clone()), + release: Some(release), + ..Host::default() + }, + wrapped_queue: Mutex::default(), }); let runner = Arc::new(TurnRunner::< _, From d751ecdca4ac9ed5fd37f1592535334ad22dc035 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:47:28 +0000 Subject: [PATCH 24/29] test(turn_runner): correct queued request flag in custom queue test The test built its runner-queued request with the wrong boolean, so it no longer matched the scenario it was meant to exercise. Passing false restores the intended setup. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinyagents-harness/src/turn_runner/mod_tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index 35de385ee..d039e665e 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -322,7 +322,7 @@ async fn custom_queue_item_shares_runner_queue_and_preserves_complete_request() }; started.notified().await; - let mut runner_queued = request(QueueDisposition::Followup, true); + let mut runner_queued = request(QueueDisposition::Followup, false); runner_queued.scope.profile_id = Some("profile-x".into()); runner_queued.thread_id = "thread-from-runner".into(); runner_queued.request_id = "runner-queued-id".into(); From 82146f328e7a8c9ac05eaff71989ff8f07394c7d Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:49:14 +0000 Subject: [PATCH 25/29] test(turn_runner): align queued request scopes with primary scope The custom queue test now assigns the primary request's scope to the runner-queued and host-queued requests instead of setting profile_id directly, and the runner-queued thread id was changed to match. This keeps the queued requests consistent with the scope the runner actually uses. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinyagents-harness/src/turn_runner/mod_tests.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index d039e665e..a4b6cc4e3 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -315,6 +315,7 @@ async fn custom_queue_item_shares_runner_queue_and_preserves_complete_request() let mut primary = request(QueueDisposition::Interrupt, false); primary.request_id = "primary".into(); + primary.scope.profile_id = Some("profile-x".into()); let scope = primary.scope.clone(); let primary_run = { let runner = runner.clone(); @@ -323,8 +324,8 @@ async fn custom_queue_item_shares_runner_queue_and_preserves_complete_request() started.notified().await; let mut runner_queued = request(QueueDisposition::Followup, false); - runner_queued.scope.profile_id = Some("profile-x".into()); - runner_queued.thread_id = "thread-from-runner".into(); + runner_queued.scope = scope.clone(); + runner_queued.thread_id = "thread-a".into(); runner_queued.request_id = "runner-queued-id".into(); runner_queued.text = "runner text".into(); runner_queued.payload = serde_json::json!({"client":"runner","attachment":[1,2]}); @@ -361,6 +362,7 @@ async fn custom_queue_item_shares_runner_queue_and_preserves_complete_request() ); let mut host_queued = request(QueueDisposition::Steer, false); + host_queued.scope = scope.clone(); host_queued.request_id = "host-queued-id".into(); host_queued.text = "host text".into(); host_queued.payload = serde_json::json!({"client":"host","locale":"en"}); From 5297454e408c5578cc921386f490f8bd7775044d Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:50:50 +0000 Subject: [PATCH 26/29] docs(turn_runner): document custom queue item type and live turn tracking Added a section explaining how hosts can supply their own queue wrapper via the trailing queue-item type while sharing the runner's RunQueue allocation, and clarified that live_turns now covers turns tracked through cleanup, including canceled or completing ones. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../tinyagents-harness/src/turn_runner/README.md | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/README.md b/crates/tinyagents-harness/src/turn_runner/README.md index af98bc12f..e4780b9d8 100644 --- a/crates/tinyagents-harness/src/turn_runner/README.md +++ b/crates/tinyagents-harness/src/turn_runner/README.md @@ -60,11 +60,19 @@ existing sinks. When one generation failure retires queued requests, the runner rewrites each classification message to match that request's queued failure message while preserving its class and details. +`TurnQueueItem

` lets a host store its own queue wrapper while the runner and +agent loop share the same `RunQueue` allocation. Implement the lossless request +conversions on the host wrapper and select it as the runner's trailing +queue-item type; the default remains `TurnRequest

`. Queue snapshots and +`TurnHost::run` expose that exact queue item type, while dispatch converts the +item back to its complete `TurnRequest

`. + ## Host inspection and cancellation -`live_turns` returns identity-only `LiveTurn` records for active primary and -parallel turns. Hosts can filter these records by `ScopeKey` and thread to -prepare cancellation notes without receiving runner-owned tokens. Use +`live_turns` returns identity-only `LiveTurn` records for primary and parallel +turns tracked through cleanup, including turns already canceled or completing. +Hosts can filter these records by `ScopeKey` and thread to prepare cancellation +notes without receiving runner-owned tokens. Use `cancel_requests` to cancel matching turns and receive the request IDs it actually stopped; when an unscoped cancellation matches multiple turns, the primary request ID comes first, followed by parallel request IDs in From 52c0ba5f7887381464bb7935fda40000e1f5f776 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:52:01 +0000 Subject: [PATCH 27/29] test(turn_runner): cover single cancel report until cleanup finishes Add a test asserting that cancelling a primary turn reports the request id only once while cleanup is still in flight, and that the turn stays visible until the run resolves as cancelled. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/turn_runner/mod_tests.rs | 58 +++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index a4b6cc4e3..1b657202b 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -282,6 +282,64 @@ async fn cancelled_parallel_turn_remains_visible_until_cleanup_finishes() { assert!(runner.live_turns().await.is_empty()); } +#[tokio::test] +async fn primary_cancel_result_is_reported_once_until_cleanup_finishes() { + let started = Arc::new(tokio::sync::Notify::new()); + let checkin_started = Arc::new(tokio::sync::Notify::new()); + let release_checkin = Arc::new(tokio::sync::Notify::new()); + let host = Arc::new(Host { + started: Some(started.clone()), + release: Some(release_checkin.clone()), + hold_checkin: true, + checkin_started: Some(checkin_started.clone()), + ..Host::default() + }); + let runner = Arc::new(runner( + host, + Arc::new(Ingress::default()), + Arc::new(Replies::default()), + Arc::new(Events::default()), + )); + let mut primary_request = request(QueueDisposition::Interrupt, false); + primary_request.request_id = "first".into(); + let scope = primary_request.scope.clone(); + let run = { + let runner = runner.clone(); + tokio::spawn(async move { runner.run(primary_request).await }) + }; + + started.notified().await; + assert_eq!( + runner + .cancel_requests(&scope, "thread-a", Some("first")) + .await, + ["first"] + ); + checkin_started.notified().await; + assert!( + runner + .cancel_requests(&scope, "thread-a", Some("first")) + .await + .is_empty() + ); + assert_eq!( + runner + .live_turns() + .await + .iter() + .map(|turn| turn.request_id.as_str()) + .collect::>(), + ["first"] + ); + + release_checkin.notify_one(); + assert!(matches!( + run.await.unwrap(), + Err(TurnRunnerError::Cancelled) + )); + assert!(runner.live_turns().await.is_empty()); +} + #[tokio::test] async fn custom_queue_item_shares_runner_queue_and_preserves_complete_request() { let started = Arc::new(tokio::sync::Notify::new()); From 291f7ce54305b906ff919b4db9d089099f6ca80b Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:53:49 +0000 Subject: [PATCH 28/29] fix(turn_runner): mark cancelled turns as non-cancellable Cancelling a turn now clears its cancellable flag while holding the active lock, so a second cancel request for the same turn is a no-op instead of re-clearing the queue and re-triggering the cancellation token. A test asserts that a queued runner item survives a full round trip through into_turn_request without losing any fields. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinyagents-harness/src/turn_runner/mod_tests.rs | 10 ++++++++++ crates/tinyagents-harness/src/turn_runner/types.rs | 5 +++-- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index 1b657202b..552070080 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -418,6 +418,16 @@ async fn custom_queue_item_shares_runner_queue_and_preserves_complete_request() stored_runner_request.idle_timeout, runner_queued.idle_timeout ); + let round_tripped = runner_item.1.clone().into_turn_request(); + assert_eq!(round_tripped.scope, runner_queued.scope); + assert_eq!(round_tripped.thread_id, runner_queued.thread_id); + assert_eq!(round_tripped.request_id, runner_queued.request_id); + assert_eq!(round_tripped.text, runner_queued.text); + assert_eq!(round_tripped.payload, runner_queued.payload); + assert_eq!(round_tripped.host_authored, runner_queued.host_authored); + assert_eq!(round_tripped.disposition, runner_queued.disposition); + assert_eq!(round_tripped.deadline, runner_queued.deadline); + assert_eq!(round_tripped.idle_timeout, runner_queued.idle_timeout); let mut host_queued = request(QueueDisposition::Steer, false); host_queued.scope = scope.clone(); diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index 7f9d15583..07c6ff144 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -460,12 +460,13 @@ where request_id: Option<&str>, ) -> Vec { let key = (scope.clone(), thread_id.to_owned()); - let active = self.active.lock().await; + let mut active = self.active.lock().await; let mut cancelled = Vec::new(); if let Some(turn) = active - .get(&key) + .get_mut(&key) .filter(|turn| turn.cancellable && request_id.is_none_or(|id| turn.request_id == id)) { + turn.cancellable = false; turn.queue.clear().await; turn.cancel.cancel(); cancelled.push(turn.request_id.clone()); From bf138dd75edbdb45d77b181ddeff657c7f76640c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 22 Oct 2026 23:55:54 +0000 Subject: [PATCH 29/29] refactor(turn_runner): iterate parallel_active keys directly Use the map's keys iterator instead of destructuring each entry, since the values are unused. This removes the ignored binding and makes the intent of the loop clearer without changing behaviour. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinyagents-harness/src/turn_runner/types.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index 07c6ff144..1a4a694a8 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -523,7 +523,7 @@ where parallel: false, }); } - for ((scope, thread_id, request_id), _) in self.parallel_active.lock().await.iter() { + for (scope, thread_id, request_id) in self.parallel_active.lock().await.keys() { live.push(LiveTurn { scope: scope.clone(), thread_id: thread_id.clone(),