feat(stella-core): parked waits — waiting on external state without model steps (#1471)

crates/stella-cli/src/claims.rs

@@ -350,12 +350,11 @@ impl ToolExecutor for ClaimTap<'_> {
         self.inner.drain_sub_agent_spend_usd()
     }
 
-    /// Forwarded: letting the empty default stand would silently serialize the
-    /// inner executor's sibling spawns (see the port's contract). The spawn
-    /// tool names no mutating path and no transient lane, so concurrent
-    /// siblings pass through `execute` above without touching a claim.
-    fn parallel_safe_names(&self) -> std::collections::HashSet<String> {
-        self.inner.parallel_safe_names()
+    /// Forwarded for the same reason as the spend drain above: a swallowed
+    /// wait request silently turns parked waits (#1471) back into
+    /// model-step polling.
+    fn drain_wait_request(&self) -> Option<stella_core::WaitRequest> {
+        self.inner.drain_wait_request()
     }
 }
 

crates/stella-cli/src/command_deck.rs

@@ -119,6 +119,7 @@ mod session_clear;
 mod sessions_view;
 mod settle;
 mod task_tap;
+use task_tap::TaskTap;
 mod theme_cmd;
 use crate::memory::{SessionMemory, inject_recall_block};
 use crate::runtime::{SystemClock, TokioSleeper};

crates/stella-cli/src/command_deck/task_tap.rs

@@ -1,5 +1,5 @@
-//! The deck's task-board tap — split out of `command_deck.rs` (a god file
-//! closed to growth) when the tap grew its `parallel_safe_names` forward.
+//! The deck's task-board decorator, split out of `command_deck.rs` to keep
+//! it under the size gate (the `driver/settlement.rs` pattern).
 
 use async_trait::async_trait;
 use serde_json::Value;
@@ -55,53 +55,9 @@ impl ToolExecutor for TaskTap<'_> {
         self.inner.drain_sub_agent_spend_usd()
     }
 
-    /// Forwarded: letting the empty default stand would silently serialize
-    /// the inner executor's sibling spawns (see the port's contract). The
-    /// spawn tool is `task`, not `task_*` — the tap never fires for it.
-    fn parallel_safe_names(&self) -> std::collections::HashSet<String> {
-        self.inner.parallel_safe_names()
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    /// A leaf claiming one parallel-safe name, standing in for the registry.
-    struct Claiming;
-
-    #[async_trait]
-    impl ToolExecutor for Claiming {
-        fn schemas(&self) -> Vec<ToolSchema> {
-            Vec::new()
-        }
-        async fn execute(&self, _name: &str, _input: &Value) -> ToolOutput {
-            ToolOutput::Ok {
-                content: String::new(),
-            }
-        }
-        fn parallel_safe_names(&self) -> std::collections::HashSet<String> {
-            std::collections::HashSet::from(["task".to_string()])
-        }
-    }
-
-    /// The deck's lead lane wraps the whole stack in this tap last, so a tap
-    /// that swallowed the claim would kill concurrent sibling spawns for
-    /// every deck session no matter what the layers below advertised.
-    #[test]
-    fn the_task_tap_forwards_parallel_safe_names() {
-        let inner = Claiming;
-        let registry = ToolRegistry::with_issue_backend(std::path::PathBuf::from("."), None);
-        let (events, _rx) = tokio::sync::mpsc::unbounded_channel();
-        let tap = TaskTap {
-            inner: &inner,
-            events,
-            registry: &registry,
-            supervisor: None,
-        };
-        assert!(
-            tap.parallel_safe_names().contains("task"),
-            "the tap must forward the inner executor's concurrency claims"
-        );
+    /// Forwarded for the same reason: a swallowed wait request silently
+    /// turns parked waits (#1471) back into model-step polling.
+    fn drain_wait_request(&self) -> Option<stella_core::WaitRequest> {
+        self.inner.drain_wait_request()
     }
 }

crates/stella-cli/src/discovery.rs

@@ -786,12 +786,11 @@ impl ToolExecutor for DiscoveryToolSet<'_> {
         self.inner.drain_sub_agent_spend_usd()
     }
 
-    /// Forwarded: letting the empty default stand would silently serialize the
-    /// inner executor's sibling spawns (see the port's contract). No lean-mode
-    /// intersection — hiding is a prompt-budget measure, and a hidden tool
-    /// called by name still executes above.
-    fn parallel_safe_names(&self) -> std::collections::HashSet<String> {
-        self.inner.parallel_safe_names()
+    /// Forwarded for the same reason as the spend drain above: a swallowed
+    /// wait request silently turns parked waits (#1471) back into
+    /// model-step polling.
+    fn drain_wait_request(&self) -> Option<stella_core::WaitRequest> {
+        self.inner.drain_wait_request()
     }
 }
 

crates/stella-cli/src/fleet_commits.rs

@@ -132,11 +132,11 @@ impl ToolExecutor for CommitObserver<'_> {
         self.inner.drain_sub_agent_spend_usd()
     }
 
-    /// Forwarded: letting the empty default stand would silently serialize
-    /// the inner executor's sibling spawns (see the port's contract) — the
-    /// spawn tool is not commit-lane routed, so nothing is observed for it.
-    fn parallel_safe_names(&self) -> std::collections::HashSet<String> {
-        self.inner.parallel_safe_names()
+    /// Forwarded for the same reason as the spend drain above: a swallowed
+    /// wait request silently turns parked waits (#1471) back into
+    /// model-step polling.
+    fn drain_wait_request(&self) -> Option<stella_core::WaitRequest> {
+        self.inner.drain_wait_request()
     }
 }
 

crates/stella-cli/src/interactive.rs

@@ -528,11 +528,11 @@ impl ToolExecutor for InteractiveToolSet<'_> {
         self.inner.drain_sub_agent_spend_usd()
     }
 
-    /// Forwarded: letting the empty default stand would silently serialize the
-    /// inner executor's sibling spawns (see the port's contract). The tools
-    /// added here (`ask_user`, skills) make no concurrency claim of their own.
-    fn parallel_safe_names(&self) -> std::collections::HashSet<String> {
-        self.inner.parallel_safe_names()
+    /// Forwarded for the same reason as the spend drain above: a swallowed
+    /// wait request silently turns parked waits (#1471) back into
+    /// model-step polling.
+    fn drain_wait_request(&self) -> Option<stella_core::WaitRequest> {
+        self.inner.drain_wait_request()
     }
 }
 

crates/stella-cli/src/subagent/tests.rs

@@ -106,18 +106,42 @@ async fn the_production_tool_stack_forwards_sub_agent_spend() {
     );
 }
 
-/// **The parallel-dispatch counterpart of the spend witness above.**
-///
-/// `parallel_safe_names` has an empty default, so any decorator that forgets
-/// to forward it silently serializes sibling `task` calls in every real
-/// session — the registry's claim never reaches the engine, and the feature
-/// (#1776) is dead exactly where it shipped. Asserted through the shipped
-/// composition for the same reason as the spend test: a future decorator
-/// inserted into the real stack fails here, and nowhere else.
+/// The wait-request twin of the spend witness above: `drain_wait_request`
+/// has a `None` default, so any one decorator forgetting to forward would
+/// silently turn parked waits (#1471) back into model-step polling for
+/// every session composed through it — and no compiler would say so.
 #[tokio::test]
-async fn the_production_tool_stack_forwards_parallel_safe_names() {
-    let ledger: SubAgentSpendLedger = Arc::default();
-    let base = LedgerBase(ledger);
+async fn the_production_tool_stack_forwards_wait_requests() {
+    /// A leaf holding one deposited request, standing in for the registry.
+    struct WaitingBase(std::sync::Mutex<Option<stella_core::WaitRequest>>);
+
+    #[async_trait]
+    impl ToolExecutor for WaitingBase {
+        fn schemas(&self) -> Vec<ToolSchema> {
+            Vec::new()
+        }
+        async fn execute(&self, _name: &str, _input: &Value) -> ToolOutput {
+            ToolOutput::Ok {
+                content: String::new(),
+            }
+        }
+        fn drain_wait_request(&self) -> Option<stella_core::WaitRequest> {
+            self.0.lock().unwrap().take()
+        }
+    }
+
+    let request = stella_core::WaitRequest {
+        description: "CI for branch main settles".into(),
+        probe: stella_core::WaitCall {
+            name: "ci_status".into(),
+            input: json!({ "probe": true, "branch": "main" }),
+        },
+        baseline: "pending".into(),
+        on_wake: None,
+        poll_interval_secs: 15,
+        timeout_secs: 0,
+    };
+    let base = WaitingBase(std::sync::Mutex::new(Some(request.clone())));
 
     let customs =
         stella_tools::custom::CustomToolSet::new(&base, Vec::new(), std::path::PathBuf::from("."));
@@ -131,11 +155,16 @@ async fn the_production_tool_stack_forwards_parallel_safe_names() {
     let discovery =
         crate::discovery::DiscoveryToolSet::new(&permitted, std::path::PathBuf::from("."));
 
-    assert!(
-        discovery.parallel_safe_names().contains("task"),
-        "the registry's concurrency claim must survive every decorator \
-         between the engine and the registry — one that swallows it silently \
-         serializes sibling spawns"
+    assert_eq!(
+        discovery.drain_wait_request(),
+        Some(request),
+        "a deposited wait request must survive every decorator between the \
+         engine and the registry — one that swallows it re-enables polling"
+    );
+    assert_eq!(
+        discovery.drain_wait_request(),
+        None,
+        "and the drain stays destructive through the stack"
     );
 }
 

crates/stella-cli/src/tool_policy.rs

@@ -103,14 +103,11 @@ impl ToolExecutor for PolicyToolSet<'_> {
         self.inner.get().drain_sub_agent_spend_usd()
     }
 
-    /// Forwarded minus what the policy withholds. Letting the empty default
-    /// stand would silently serialize the inner executor's sibling spawns —
-    /// but blindly delegating would advertise names `execute` above refuses,
-    /// an empty promise. Same two-sided shape as `schemas`/`execute`.
-    fn parallel_safe_names(&self) -> std::collections::HashSet<String> {
-        let mut names = self.inner.get().parallel_safe_names();
-        names.retain(|name| self.policy.allows(name));
-        names
+    /// Forwarded for the same reason as the spend drain above: a swallowed
+    /// wait request silently turns parked waits (#1471) back into
+    /// model-step polling.
+    fn drain_wait_request(&self) -> Option<stella_core::WaitRequest> {
+        self.inner.get().drain_wait_request()
     }
 }
 

crates/stella-core/README.md

@@ -101,6 +101,7 @@ lib.rs), never as a planning assumption.
 | [`src/lib.rs`](src/lib.rs) | Module list and the crate's re-export surface. Read the `pub use` block to see what callers are meant to touch. |
 | [`src/driver.rs`](src/driver.rs) | `Engine`, `EngineConfig`, `TurnOutcome`, `run_turn`. The one file that sequences every other module against real I/O. Start here. |
 | [`src/driver/settlement.rs`](src/driver/settlement.rs) | The between-steps budget check and `BudgetTick`/warning emission, split out of the step loop. |
+| [`src/waiting.rs`](src/waiting.rs) + [`src/driver/waiting.rs`](src/driver/waiting.rs) | Parked waits (#1471): the pure change/deadline decision logic and `WaitRequest` types, and the driver's park loop that probes through the existing ports with zero model calls. |
 | [`src/ports.rs`](src/ports.rs) | The port boundary: `ToolExecutor`, `ReadOnlyTools`, `Clock`, `TurnGate`, `TurnSteering`. |
 | [`src/budget.rs`](src/budget.rs) | `BudgetGuard` — USD spend against a turn and/or session cap. Returns `BudgetOutcome`; aborts nothing itself. |
 | [`src/compaction.rs`](src/compaction.rs) | `compact()` — dedup, supersession, aging, eviction. Open when the conversation is being rewritten wrongly. |
@@ -134,9 +135,10 @@ lib.rs), never as a planning assumption.
 and each step runs the same phases in the same order: pause gate → drain
 steering / check soft stop → budget check → snapshot tool-result identities →
 compaction pass → loop detection → model call (wrapped in retry+backoff) →
-committed-step bookkeeping → dispatch. Each phase is one sub-method
-(`run_compaction_pass`, `check_loop_detection`, `run_model_call`,
-`dispatch_completion`). The order is load-bearing, not stylistic: identities are
+committed-step bookkeeping → dispatch → parked wait (when a tool deposited
+one). Each phase is one sub-method (`run_compaction_pass`,
+`check_loop_detection`, `run_model_call`, `dispatch_completion`,
+`maybe_park`). The order is load-bearing, not stylistic: identities are
 snapshotted *before* compaction because the compaction pass rewrites tool
 results in place and loop detection then runs on the rewritten history in that
 same step (#554). The engine holds no conversation state — `messages` is

crates/stella-core/src/driver.rs

@@ -132,6 +132,7 @@ use tokio::sync::mpsc::UnboundedSender;
 
 mod dispatch;
 mod settlement;
+mod waiting;
 use settlement::{BudgetWarnings, emit_budget_warning, record_settled_cost};
 
 /// Everything about a turn's execution that isn't the provider/tools
@@ -530,27 +531,6 @@ pub fn step_cap_reason(max_steps: usize) -> String {
     )
 }
 
-/// The most recent non-empty assistant text in a turn's transcript.
-///
-/// Used only by the halt path ([`TurnHalt`]), which ends a turn at a step
-/// boundary and therefore has no "final" model text of its own to report. An
-/// assistant message that only made tool calls carries empty `content`, so
-/// this walks back to the last thing the model actually *said* rather than
-/// reporting a blank answer for a turn that did real work.
-///
-/// `None` when the model has said nothing yet — a turn halted after a first
-/// step of pure tool calls — which the caller renders as the halt reason.
-fn last_assistant_text(state: &crate::step::TurnState) -> Option<String> {
-    state
-        .messages
-        .iter()
-        .rev()
-        .find(|message| {
-            message.role == MessageRole::Assistant && !message.content.trim().is_empty()
-        })
-        .map(|message| message.content.clone())
-}
-
 /// Upper bound on tool calls from one step executing concurrently. Tools
 /// are I/O-bound (process spawns, file reads), so this caps descriptor and
 /// process pressure, not CPU.
@@ -900,7 +880,8 @@ impl<'a> Engine<'a> {
                 // predicate's reason when the final step was pure tool calls
                 // (an assistant message that only called tools has empty
                 // `content`), so the turn never reports an empty answer.
-                let text = last_assistant_text(&turn.state).unwrap_or_else(|| reason.clone());
+                let text =
+                    settlement::last_assistant_text(&turn.state).unwrap_or_else(|| reason.clone());
                 let outcome = TurnOutcome::Completed {
                     text,
                     cost_usd: turn.state.total_cost_usd,
@@ -1262,6 +1243,13 @@ impl<'a> Engine<'a> {
             return completed.into();
         }
 
+        // A tool may have asked to park the turn (#1471, `driver::waiting`):
+        // the engine probes on its own clock and the model wakes to the
+        // delta. `Some` only when cancelled while parked.
+        if let Some(cancelled) = self.maybe_park(state, events).await {
+            return cancelled;
+        }
+
         // Advanced only by a step that committed and continued, so the index
         // a checkpoint carries is always "the step that runs next".
         state.step += 1;

crates/stella-core/src/driver/settlement.rs

@@ -1,4 +1,4 @@
-use stella_protocol::AgentEvent;
+use stella_protocol::{AgentEvent, MessageRole};
 
 use super::TurnOutcome;
 use crate::budget::{BudgetAxis, BudgetGuard, BudgetOutcome, DeadlineOutcome};
@@ -197,3 +197,24 @@ fn check_budget(
         cost_usd: total_cost_usd,
     })
 }
+
+/// The most recent non-empty assistant text in a turn's transcript.
+///
+/// Used only by the halt path (`TurnHalt`), which ends a turn at a step
+/// boundary and therefore has no "final" model text of its own to report. An
+/// assistant message that only made tool calls carries empty `content`, so
+/// this walks back to the last thing the model actually *said* rather than
+/// reporting a blank answer for a turn that did real work.
+///
+/// `None` when the model has said nothing yet — a turn halted after a first
+/// step of pure tool calls — which the caller renders as the halt reason.
+pub(super) fn last_assistant_text(state: &crate::step::TurnState) -> Option<String> {
+    state
+        .messages
+        .iter()
+        .rev()
+        .find(|message| {
+            message.role == MessageRole::Assistant && !message.content.trim().is_empty()
+        })
+        .map(|message| message.content.clone())
+}

crates/stella-core/src/driver/tests.rs

@@ -3,9 +3,8 @@ use std::sync::atomic::{AtomicU32, Ordering};
 
 use async_trait::async_trait;
 use serde_json::Value;
-use stella_protocol::CompletionUsage;
-use stella_protocol::ToolSchema;
 use stella_protocol::event::BudgetMode;
+use stella_protocol::{CompletionUsage, ToolSchema};
 use tokio::sync::Mutex as TokioMutex;
 use tokio::sync::mpsc;
 
@@ -3377,6 +3376,7 @@ mod compute_passes;
 mod context_efficiency;
 mod lifecycle_bus;
 mod loop_abort;
+mod parked_wait;
 mod steer_midturn;
 mod usage_completeness;
 mod zero_copy_request;