From 226f5d4a315d9ea3015e37731b18dd1bcf2d8896 Mon Sep 17 00:00:00 2001 From: Stella Test Date: Thu, 6 Aug 2026 03:15:56 -0700 Subject: [PATCH 1/8] =?UTF-8?q?feat(stella-core):=20parked=20waits=20?= =?UTF-8?q?=E2=80=94=20drain=20a=20tool's=20WaitRequest=20and=20probe=20wi?= =?UTF-8?q?thout=20model=20calls?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pure decision half (crate::waiting), the ToolExecutor drain port, and the driver's park loop (driver/waiting.rs, the settlement.rs split pattern). Refs #1471 --- crates/stella-core/src/driver.rs | 32 +- crates/stella-core/src/driver/settlement.rs | 23 +- crates/stella-core/src/driver/waiting.rs | 160 ++++++++++ crates/stella-core/src/lib.rs | 2 + crates/stella-core/src/ports.rs | 33 +++ crates/stella-core/src/waiting.rs | 308 ++++++++++++++++++++ 6 files changed, 535 insertions(+), 23 deletions(-) create mode 100644 crates/stella-core/src/driver/waiting.rs create mode 100644 crates/stella-core/src/waiting.rs diff --git a/crates/stella-core/src/driver.rs b/crates/stella-core/src/driver.rs index 9d861dda3..d0ab9a28b 100644 --- a/crates/stella-core/src/driver.rs +++ b/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 { - 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; diff --git a/crates/stella-core/src/driver/settlement.rs b/crates/stella-core/src/driver/settlement.rs index 5e1543c51..76babe8b6 100644 --- a/crates/stella-core/src/driver/settlement.rs +++ b/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 { + state + .messages + .iter() + .rev() + .find(|message| { + message.role == MessageRole::Assistant && !message.content.trim().is_empty() + }) + .map(|message| message.content.clone()) +} diff --git a/crates/stella-core/src/driver/waiting.rs b/crates/stella-core/src/driver/waiting.rs new file mode 100644 index 000000000..79927f49f --- /dev/null +++ b/crates/stella-core/src/driver/waiting.rs @@ -0,0 +1,160 @@ +//! The parked wait's engine half (#1471): drain a tool's +//! [`WaitRequest`](crate::waiting::WaitRequest) at the step boundary and +//! probe on the engine's own clock until the watched state changes — zero +//! model calls, zero transcript growth, and the byte-stable prompt prefix +//! untouched while the turn sleeps. A child module of `driver` (the +//! `settlement.rs` pattern) so the engine internals stay reachable without +//! growing `driver.rs` past the size gate. +//! +//! The decision arithmetic — what counts as a change, how many polls a +//! deadline affords, what the wake message says — is pure and lives in +//! [`crate::waiting`]; this module owns only the loop that drives it through +//! the engine's existing ports (the tool executor for probes, the sleeper +//! for intervals, the cancel token and soft-stop latch for interruption). + +use stella_protocol::{AgentEvent, CompletionMessage, ToolCall, ToolOutput}; + +use super::Engine; +use crate::event_sender::EventSender; +use crate::step::{StepOutcome, TurnState}; +use crate::waiting::{WaitCall, WaitRequest, WakeReason, decide, probe_fingerprint, wake_message}; + +/// The synthetic `call_id` probe and wake replays carry. They never enter +/// the transcript, so the id only needs to be recognizable in hook payloads +/// and diagnostics. +const PARKED_CALL_ID: &str = "parked-wait"; + +impl<'a> Engine<'a> { + /// Park the turn on a deposited wait request, if the step's tools left + /// one. Called once per step, after the tool results are committed and + /// before the next model call — the same safe boundary as the budget + /// enforcer and the pause gate (invariant 6), which is what makes an + /// arbitrarily long wait cost zero model steps and force no compaction. + /// + /// Returns `Some` only when the turn was cancelled while parked; every + /// other ending — a change, the deadline, a soft stop arriving — lets + /// the turn continue into its next step, which then observes the wake + /// message (or the soft-stop latch) exactly as it observes a user steer. + pub(super) async fn maybe_park( + &self, + state: &mut TurnState, + events: &EventSender, + ) -> Option { + let request = self.tools.drain_wait_request()?; + // Structural guarantee, not a convention: a replayed call the model + // never sees must not mutate anything. Both the probe and the wake + // call must advertise `read_only` in the very schema set the model's + // requests are built from. + let read_only: std::collections::HashSet = self + .tools + .schemas() + .into_iter() + .filter(|s| s.read_only) + .map(|s| s.name) + .collect(); + let replayed = std::iter::once(&request.probe).chain(request.on_wake.as_ref()); + for call in replayed { + if !read_only.contains(&call.name) { + let _ = events.send(AgentEvent::Text { + delta: format!( + "\n⚠ Ignoring a parked-wait request: `{}` is not a read-only tool, and \ + the engine only replays read-only calls while the model is not looking.\n", + call.name + ), + }); + return None; + } + } + + let _ = events.send(AgentEvent::Text { + delta: format!( + "\n⏳ Parked: waiting until {} — probing every {}s for up to {}s, with no model \ + calls while waiting.\n", + request.description, + request.interval_secs(), + request.deadline_secs() + ), + }); + + let mut polls_used = 0u64; + let mut last_observed: Option = None; + let reason = loop { + if let Some(cancelled) = state.cancel_outcome(events) { + return Some(cancelled); + } + // A latched soft stop ends the wait early and falls through to + // the next step boundary, where the standard soft-stop exit + // keeps the transcript — the park must not make Esc wait out a + // CI run. (The latch contract says the request survives this + // read.) User steers queued mid-park stay queued: the drain is + // destructive by contract, so the boundary that injects them is + // the one that reads them. + if self + .steering + .is_some_and(|steering| steering.soft_stop_requested()) + { + return None; + } + self.sleeper + .sleep(request.interval_secs().saturating_mul(1000)) + .await; + if let Some(cancelled) = state.cancel_outcome(events) { + return Some(cancelled); + } + let output = self.replay(&request.probe, events).await; + polls_used += 1; + let observed = probe_fingerprint(&output); + if observed.is_some() { + last_observed.clone_from(&observed); + } + if let Some(reason) = decide(&request, observed.as_deref(), polls_used) { + break reason; + } + }; + + // The delta the model wakes to: the rich wake call's output when the + // request named one (and the state actually changed), otherwise the + // last thing the probe saw. + let detail = match (reason, &request.on_wake) { + (WakeReason::Changed, Some(call)) => match self.replay(call, events).await { + ToolOutput::Ok { content } => Some(content), + // The wake call failing must not hide that the wait ended — + // surface the error as the detail and let the model re-query. + ToolOutput::Error { message } => Some(format!("(wake query failed: {message})")), + }, + _ => last_observed, + }; + let message = wake_message(&request, reason, polls_used, detail.as_deref()); + let _ = events.send(AgentEvent::Text { + delta: format!( + "\n▶ Wait ended after {polls_used} probe{}: {}\n", + if polls_used == 1 { "" } else { "s" }, + match reason { + WakeReason::Changed => "the watched state changed.", + WakeReason::DeadlineExpired => "the deadline expired with no change.", + } + ), + }); + // The wake rides the conversation tail — the volatile cache zone — + // exactly where a user steer lands (invariant 7): the stable prefix + // and every cacheable block before it are byte-identical to the + // pre-park request. + state.messages.push(CompletionMessage::user(message)); + None + } + + /// Replay one deposited call through the standard dispatch path — + /// malformed-input repair, hooks, and the engine's tool timeout all + /// apply, so a probe behaves like any other tool call except that its + /// output never reaches the transcript. Hook diagnostics still surface + /// on the turn stream; a `PreToolUse` block comes back as an error + /// output, which the wait loop reads as "no observation, keep waiting". + async fn replay(&self, call: &WaitCall, events: &EventSender) -> ToolOutput { + let tool_call = ToolCall { + call_id: PARKED_CALL_ID.into(), + name: call.name.clone(), + input: call.input.clone(), + }; + self.execute_with_repair(&tool_call, Some(events)).await + } +} diff --git a/crates/stella-core/src/lib.rs b/crates/stella-core/src/lib.rs index 99f90e3cd..37e5a0a83 100644 --- a/crates/stella-core/src/lib.rs +++ b/crates/stella-core/src/lib.rs @@ -48,6 +48,7 @@ pub mod subagent; mod summarize; pub mod tasks; pub mod tool_foundry; +pub mod waiting; pub use budget::{BudgetGuard, BudgetOutcome}; // `bus::HookEvent` (the extension-bus envelope) stays module-qualified: the @@ -101,6 +102,7 @@ pub use subagent::{ forwards_to_parent, push_sub_agent_spend, }; pub use tasks::{SpawnRequest, TaskBoard, TaskBoardError}; +pub use waiting::{WaitCall, WaitRequest}; pub use tool_foundry::{ GapDetectionConfig, ParamKind, ProposedTool, ShellInvocation, ToolParameter, detect_tool_gaps, }; diff --git a/crates/stella-core/src/ports.rs b/crates/stella-core/src/ports.rs index e103b3fa6..f5dfd7627 100644 --- a/crates/stella-core/src/ports.rs +++ b/crates/stella-core/src/ports.rs @@ -50,6 +50,32 @@ pub trait ToolExecutor: Send + Sync { 0.0 } + /// The parked-wait request a tool deposited during its last `execute`, + /// **taken** (not peeked) — the engine consumes it at the next step + /// boundary and parks the turn on the engine's own clock instead of + /// letting the model poll (#1471, `crate::waiting`). + /// + /// The same "written by one object, drained by another" seam as + /// [`Self::drain_sub_agent_spend_usd`], for the same structural reason: + /// a tool returns only a `ToolOutput`, and by the time the engine is at + /// a boundary where parking is safe (invariant 6 — between model calls, + /// never mid-tool) the tool call is long finished. + /// + /// # Decorators MUST forward this + /// + /// The default returns `None`, which reads as "no tool of mine ever + /// asks to wait" — correct for a leaf, and **wrong for a wrapper**. A + /// decorator that forgets to forward silently turns parked waits off + /// for every surface composed through it, and the model falls back to + /// burning steps on polling — the exact regression #1471 removes. The + /// shipped composition is pinned by `stella-cli`'s + /// `the_production_tool_stack_forwards_wait_requests`. + /// + /// Destructive by contract: two drains of one request would park twice. + fn drain_wait_request(&self) -> Option { + None + } + /// Names of tools that are safe to run **concurrently with each other and /// with read-only calls** despite not declaring `read_only` — the /// executor-level claim behind the driver's dispatch grouping. The @@ -141,6 +167,13 @@ impl ToolExecutor for ReadOnlyTools<'_> { self.inner.drain_sub_agent_spend_usd() } + /// Forwarded: a verifier legitimately waits on external state (CI is the + /// canonical read-only evidence source), and its probe replays through + /// this same view — so the read-only restriction holds while parked too. + fn drain_wait_request(&self) -> Option { + self.inner.drain_wait_request() + } + // `parallel_safe_names` deliberately keeps the empty default rather than // forwarding (contrast with the spend drain above): the one shipped // parallel-safe tool is the sub-agent spawn, and this view exists to diff --git a/crates/stella-core/src/waiting.rs b/crates/stella-core/src/waiting.rs new file mode 100644 index 000000000..d13197170 --- /dev/null +++ b/crates/stella-core/src/waiting.rs @@ -0,0 +1,308 @@ +//! Parked waits (#1471): the types and pure decision logic behind waiting on +//! external state without spending model steps. +//! +//! Waiting used to be a model behavior: poll a tool, read the unchanged +//! output, poll again — every iteration a full model round-trip on a growing +//! transcript, and every long in-tool sleep a prompt-cache expiry. A parked +//! wait inverts that: a tool deposits a [`WaitRequest`] during its execution, +//! and the engine re-runs a cheap read-only *probe* between model calls, +//! re-invoking the model exactly once — when the probed state changes or the +//! deadline passes — with the delta riding a volatile tail message. The +//! polling itself never touches the transcript, so an arbitrarily long wait +//! costs O(1) model steps and forces no compaction. +//! +//! This module is decision logic only (invariant 2): what counts as a change, +//! how many polls a deadline affords, and what the wake message says are all +//! pure functions over owned data. The actual sleeping and probing live in +//! `driver::waiting`, driven through the engine's existing ports +//! ([`crate::ports::ToolExecutor`], [`crate::retry::Sleeper`]) — no new I/O +//! surface exists for this feature. + +use serde::{Deserialize, Serialize}; +use stella_protocol::ToolOutput; + +/// Floor on the probe interval. A probe is a real subprocess or network round +/// trip (`gh run list`, a file stat) against something that changes on +/// CI-and-deploy timescales; probing faster than this buys nothing and spends +/// someone else's rate limit (#923). +pub const MIN_POLL_INTERVAL_SECS: u64 = 5; + +/// Ceiling on a parked wait's deadline. Two hours covers the workspace's +/// longest legitimate wait (the fleet CI watcher's cumulative cap is the same +/// figure); anything longer is a stuck condition the model should be woken to +/// reconsider, not slept through. +pub const MAX_PARK_SECS: u64 = 2 * 60 * 60; + +/// Default deadline when the depositing tool does not name one. +pub const DEFAULT_PARK_SECS: u64 = 30 * 60; + +/// The marker prefix on every wake message the engine injects, so transcript +/// consumers (and tests) can recognize an engine-authored wake the same way +/// `LOOP_STEER_PREFIX` marks a loop steer. +pub const WAKE_MARKER: &str = "[parked wait"; + +/// One tool invocation a parked wait replays on the engine's own clock, +/// with no model involvement. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct WaitCall { + /// Tool name, resolved through the same executor that ran the + /// depositing call. + pub name: String, + /// The exact input to replay. Composed by the depositing tool from its + /// own already-approved input — never from model-supplied text it did + /// not validate — so replaying it opens no approval surface the + /// original call did not already pass. + pub input: serde_json::Value, +} + +/// A tool's request to park the turn until an observed condition changes. +/// +/// Deposited during [`crate::ports::ToolExecutor::execute`] and drained by +/// the engine at the following step boundary +/// ([`crate::ports::ToolExecutor::drain_wait_request`]) — the same +/// "written by one object, drained by another" seam sub-agent spend uses, +/// and for the same structural reason: the tool is finished by the time the +/// engine is at a boundary where parking is safe (invariant 6). +/// +/// Serde round-trips (invariant 4) because the value crosses the +/// `stella-tools` → `stella-core` boundary, and because a checkpointed turn +/// may someday want to persist one. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct WaitRequest { + /// Human-readable condition, e.g. `CI for --branch 'main' settles`. + /// Names the wait in the park announcement and the wake message. + pub description: String, + /// Cheap read-only call re-executed once per interval. The turn wakes + /// when its output's fingerprint first differs from `baseline`. The + /// engine refuses to park on a probe whose schema does not declare + /// `read_only` — a mutating probe would repeat a write on a timer the + /// model never sees. + pub probe: WaitCall, + /// Fingerprint of the probe's output at deposit time. The depositing + /// tool observed the condition once itself (that observation is why it + /// is asking to wait), so it supplies the baseline rather than having + /// the engine burn a redundant probe to rediscover it. + pub baseline: String, + /// Richer call run once on wake; its output rides the wake message so + /// the model resumes with the fresh state, not a stale poll history. + /// `None` wakes with the probe's own final output. + pub on_wake: Option, + /// Seconds between probes. Clamped up to [`MIN_POLL_INTERVAL_SECS`]. + pub poll_interval_secs: u64, + /// Wall-clock bound on the whole wait. `0` means + /// [`DEFAULT_PARK_SECS`]; clamped down to [`MAX_PARK_SECS`]. + pub timeout_secs: u64, +} + +impl WaitRequest { + /// The probe interval with the floor applied. + #[must_use] + pub fn interval_secs(&self) -> u64 { + self.poll_interval_secs.max(MIN_POLL_INTERVAL_SECS) + } + + /// The deadline with default and ceiling applied. + #[must_use] + pub fn deadline_secs(&self) -> u64 { + let requested = if self.timeout_secs == 0 { + DEFAULT_PARK_SECS + } else { + self.timeout_secs + }; + requested.min(MAX_PARK_SECS) + } + + /// How many probes the deadline affords, always at least one. + /// + /// The engine deliberately counts polls instead of reading a clock: the + /// deadline arithmetic stays a pure function of the request (this + /// crate holds no time source — the injected sleeper owns real time), + /// and a test with a no-op sleeper walks the same bound production + /// does. The bound under-counts by the probes' own execution time, + /// which errs toward waking early — the safe direction. + #[must_use] + pub fn max_polls(&self) -> u64 { + (self.deadline_secs() / self.interval_secs()).max(1) + } +} + +/// Why a parked wait ended. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WakeReason { + /// The probe's fingerprint diverged from the baseline. + Changed, + /// Every afforded poll observed the baseline (or failed); the deadline + /// is spent. + DeadlineExpired, +} + +/// The fingerprint of one probe observation, or `None` when the probe +/// errored — a transient `gh`/network hiccup must keep the turn waiting +/// rather than waking it with a phantom change (the deadline is the +/// backstop for a probe that never stops failing). Whitespace is trimmed so +/// a trailing-newline difference between the depositing tool's baseline and +/// the replayed probe cannot manufacture a wake. +#[must_use] +pub fn probe_fingerprint(output: &ToolOutput) -> Option { + match output { + ToolOutput::Ok { content } => Some(content.trim().to_string()), + ToolOutput::Error { .. } => None, + } +} + +/// Whether one observation ends the wait: `Some(Changed)` on the first +/// fingerprint that differs from the baseline, `Some(DeadlineExpired)` once +/// `polls_used` exhausts the afforded polls, `None` to keep waiting. +#[must_use] +pub fn decide( + request: &WaitRequest, + observed: Option<&str>, + polls_used: u64, +) -> Option { + if let Some(fingerprint) = observed { + if fingerprint != request.baseline { + return Some(WakeReason::Changed); + } + } + if polls_used >= request.max_polls() { + return Some(WakeReason::DeadlineExpired); + } + None +} + +/// The volatile tail message the model wakes to. Marked with +/// [`WAKE_MARKER`], and explicit that the wait cost no model steps — the +/// model must not conclude it has been polling and start compensating. +#[must_use] +pub fn wake_message( + request: &WaitRequest, + reason: WakeReason, + polls_used: u64, + detail: Option<&str>, +) -> String { + let waited_secs = polls_used * request.interval_secs(); + let headline = match reason { + WakeReason::Changed => format!( + "{WAKE_MARKER} complete] {} — the watched state changed after ~{waited_secs}s of \ + engine-side probing (no model steps were spent waiting).", + request.description + ), + WakeReason::DeadlineExpired => format!( + "{WAKE_MARKER} timed out] {} — still unchanged after ~{waited_secs}s of engine-side \ + probing (no model steps were spent waiting). Reassess rather than re-polling: the \ + condition may be stuck.", + request.description + ), + }; + match detail { + Some(detail) if !detail.trim().is_empty() => { + format!("{headline}\n\nCurrent state:\n{detail}") + } + _ => headline, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn request(interval: u64, timeout: u64) -> WaitRequest { + WaitRequest { + description: "CI for --branch 'main' settles".into(), + probe: WaitCall { + name: "ci_status".into(), + input: serde_json::json!({ "probe": true, "branch": "main" }), + }, + baseline: "pending".into(), + on_wake: None, + poll_interval_secs: interval, + timeout_secs: timeout, + } + } + + #[test] + fn clamps_apply_floor_default_and_ceiling() { + // A zero interval cannot busy-spin someone else's API. + assert_eq!(request(0, 600).interval_secs(), MIN_POLL_INTERVAL_SECS); + // A zero timeout means the default, not an instant expiry. + assert_eq!(request(15, 0).deadline_secs(), DEFAULT_PARK_SECS); + // A u64::MAX timeout cannot park a turn for a week. + assert_eq!(request(15, u64::MAX).deadline_secs(), MAX_PARK_SECS); + // Even a deadline shorter than one interval affords one probe. + assert_eq!(request(60, 1).max_polls(), 1); + } + + #[test] + fn a_changed_fingerprint_wakes_and_an_unchanged_one_waits() { + let req = request(15, 600); + assert_eq!(decide(&req, Some("pending"), 1), None); + assert_eq!(decide(&req, Some("settled"), 1), Some(WakeReason::Changed)); + } + + #[test] + fn a_probe_error_keeps_waiting_instead_of_waking() { + let req = request(15, 600); + // An errored probe yields no fingerprint… + assert_eq!( + probe_fingerprint(&ToolOutput::Error { + message: "gh: network unreachable".into() + }), + None + ); + // …and no fingerprint is "keep waiting", never "changed". + assert_eq!(decide(&req, None, 1), None); + } + + #[test] + fn the_deadline_expires_after_the_afforded_polls() { + let req = request(15, 600); // affords 40 polls + assert_eq!(req.max_polls(), 40); + assert_eq!(decide(&req, Some("pending"), 39), None); + assert_eq!( + decide(&req, Some("pending"), 40), + Some(WakeReason::DeadlineExpired) + ); + // A change on the final poll still reports Changed, not a timeout. + assert_eq!(decide(&req, Some("settled"), 40), Some(WakeReason::Changed)); + } + + #[test] + fn fingerprints_trim_whitespace_so_a_newline_is_not_a_change() { + assert_eq!( + probe_fingerprint(&ToolOutput::Ok { + content: "pending\n".into() + }) + .as_deref(), + Some("pending") + ); + } + + #[test] + fn wake_messages_carry_the_marker_and_the_detail() { + let req = request(15, 600); + let woke = wake_message(&req, WakeReason::Changed, 8, Some("all green")); + assert!(woke.starts_with(WAKE_MARKER), "{woke}"); + assert!(woke.contains("all green"), "{woke}"); + assert!(woke.contains("~120s"), "{woke}"); + let timed_out = wake_message(&req, WakeReason::DeadlineExpired, 40, None); + assert!(timed_out.contains("timed out"), "{timed_out}"); + // A blank detail attaches nothing — no dangling "Current state:". + assert!(!timed_out.contains("Current state"), "{timed_out}"); + } + + /// Invariant 4: the request crosses the `stella-tools` → `stella-core` + /// boundary, so it round-trips byte-for-byte. + #[test] + fn wait_request_round_trips_through_serde_json() { + let req = WaitRequest { + on_wake: Some(WaitCall { + name: "ci_status".into(), + input: serde_json::json!({ "branch": "main" }), + }), + ..request(15, 600) + }; + let json = serde_json::to_string(&req).expect("serialize"); + let back: WaitRequest = serde_json::from_str(&json).expect("deserialize"); + assert_eq!(back, req); + } +} From e81f6129d8fc207d890544a08322f0619f30218e Mon Sep 17 00:00:00 2001 From: Stella Test Date: Thu, 6 Aug 2026 03:21:54 -0700 Subject: [PATCH 2/8] feat(stella-tools): ci_status deposits a parked wait instead of blocking in-tool wait=true returns current status and parks the turn; the engine probes a stable settled/pending word (#1466 semantics kept) and wakes the model once with the fresh status. mentions_path moves to exploration.rs to keep registry.rs under its ceiling. Refs #1471 --- crates/stella-core/src/driver/waiting.rs | 2 +- crates/stella-tools/src/ci.rs | 298 ++++++++++++++---- crates/stella-tools/src/exploration.rs | 30 ++ crates/stella-tools/src/registry.rs | 55 ++-- .../src/registry/process_tools.rs | 2 +- crates/stella-tools/src/registry/tests.rs | 1 + 6 files changed, 300 insertions(+), 88 deletions(-) diff --git a/crates/stella-core/src/driver/waiting.rs b/crates/stella-core/src/driver/waiting.rs index 79927f49f..a6f093250 100644 --- a/crates/stella-core/src/driver/waiting.rs +++ b/crates/stella-core/src/driver/waiting.rs @@ -17,7 +17,7 @@ use stella_protocol::{AgentEvent, CompletionMessage, ToolCall, ToolOutput}; use super::Engine; use crate::event_sender::EventSender; use crate::step::{StepOutcome, TurnState}; -use crate::waiting::{WaitCall, WaitRequest, WakeReason, decide, probe_fingerprint, wake_message}; +use crate::waiting::{WaitCall, WakeReason, decide, probe_fingerprint, wake_message}; /// The synthetic `call_id` probe and wake replays carry. They never enter /// the transcript, so the id only needs to be recognizable in hook payloads diff --git a/crates/stella-tools/src/ci.rs b/crates/stella-tools/src/ci.rs index 2f6ebf124..fc09e875d 100644 --- a/crates/stella-tools/src/ci.rs +++ b/crates/stella-tools/src/ci.rs @@ -15,6 +15,11 @@ use crate::registry::Tool; const DEFAULT_TIMEOUT_SECS: u64 = 120; +/// Seconds between settledness probes while the turn is parked. CI runs +/// change state on minute timescales; 15s keeps the wake latency small +/// against a multi-minute wait without hammering the forge (#923). +const PARK_POLL_INTERVAL_SECS: u64 = 15; + /// `ci_status`'s `timeout_secs`, through the crate-wide clamp /// ([`crate::exec::timeout_from`]): an unclamped model-supplied u64 would /// disable the hang backstop for every `gh` sub-call (u64::MAX ≈ never). @@ -22,7 +27,15 @@ fn ci_timeout(input: &Value) -> u64 { crate::exec::timeout_from(input, DEFAULT_TIMEOUT_SECS) } -pub struct CiStatus; +/// The tool, plus the parked-wait deposit slot the registry drains +/// (`Tool::take_wait_request`, #1471). Interior mutability because +/// `execute` takes `&self`; the slot holds at most the latest request and +/// is overwritten (never appended) so a request a host never drained +/// cannot go stale across calls. +#[derive(Default)] +pub struct CiStatus { + pending_wait: std::sync::Mutex>, +} #[async_trait] impl Tool for CiStatus { @@ -31,8 +44,11 @@ impl Tool for CiStatus { name: "ci_status".into(), description: "CI state via GitHub (gh). Give branch, pr, or commit; returns runs, \ conclusions, and a failure log tail scoped to the target's current \ - head commit. wait=true blocks until every run for the target has \ - completed, or timeout_secs expires — not just the newest-created one." + head commit. wait=true returns the current status immediately and \ + PARKS the turn until every run for the target has completed (or \ + timeout_secs, default 30 minutes): the engine probes on its own \ + clock with no model calls, and you wake with the fresh status — \ + never poll ci_status in a loop or sleep in bash to wait for CI." .into(), input_schema: serde_json::json!({ "type": "object", @@ -65,9 +81,32 @@ impl Tool for CiStatus { }; } + // Probe replay for a parked wait (#1471): answer with the single + // word `settled` or `pending`, deliberately free of run names, + // elapsed times, or counts — anything that changes poll-to-poll + // would fake the state change the engine is watching for. Probe + // inputs deposited below carry the already-resolved branch, so this + // path re-resolves a PR head only when a caller passes `pr` itself. + if input.get("probe").and_then(|v| v.as_bool()).unwrap_or(false) { + let pr_head = match input.get("pr").and_then(|v| v.as_u64()) { + Some(pr) => resolve_pr_head(pr, root, timeout_secs).await, + None => None, + }; + let scope = gh_run_scope(input, pr_head.as_ref()); + return match exec::run_github(&settled_command(&scope), root, timeout_secs).await { + Ok((0, out)) => ToolOutput::Ok { + content: out.trim().to_string(), + }, + Ok((code, out)) => ToolOutput::Error { + message: format!("gh failed (exit {code}): {out}"), + }, + Err(message) => ToolOutput::Error { message }, + }; + } + let list_cmd = primary_command(input); - let (code, mut report) = match exec::run_github(&list_cmd, root, timeout_secs).await { + let (code, report) = match exec::run_github(&list_cmd, root, timeout_secs).await { Ok(pair) => pair, Err(e) => return ToolOutput::Error { message: e }, }; @@ -93,13 +132,33 @@ impl Tool for CiStatus { // attach logs from an UNRELATED branch's most-recent run. let scope = gh_run_scope(input, pr_head.as_ref()); - // Optionally block until EVERY run for THIS target has completed - // (not just the newest-created one — #1466), then re-list. + // wait=true used to BLOCK here in a composed shell poll loop, capped + // by the exec backstop at 10 minutes — every second of it inside one + // tool call, aging the provider prompt cache and ending in a + // tool-timeout error for any real CI run (#1471's evidence). It now + // checks settledness once and, when runs are still pending, deposits + // a parked-wait request: the ENGINE probes between model calls + // (`settled`/`pending` via the probe input above, #1466's + // whole-incomplete-set semantics) and the model wakes exactly once, + // to the fresh status the `on_wake` replay fetches. if wait { - let _ = exec::run_github(&wait_command(&scope), root, timeout_secs).await; - if let Ok((0, fresh)) = exec::run_github(&list_cmd, root, timeout_secs).await { - report = fresh; + let settled = exec::run_github(&settled_command(&scope), root, timeout_secs).await; + if matches!(&settled, Ok((0, out)) if out.trim() == "pending") { + *self + .pending_wait + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = + Some(park_request(input, pr_head.as_ref())); + return ToolOutput::Ok { + content: format!( + "{report}\n⏳ Runs are still in progress. Parking the turn until CI \ + settles — the engine probes every {PARK_POLL_INTERVAL_SECS}s with no \ + model calls, and you will be woken with the fresh status. Do not poll." + ), + }; } + // Already settled — or the check itself failed, in which case + // reporting the state we have beats blocking on a broken probe. } // Attach failure logs for THIS target's most recent failed run at @@ -122,12 +181,90 @@ impl Tool for CiStatus { } // A `bash -c` composer joins the `command.started` fence (#804). The - // gate sees the primary query line; the pr-head resolution, wait-watch - // and failure-log sub-queries are same-target derivatives of the same + // gate sees the primary query line — or, for a parked-wait probe + // replay, the settledness line itself; the pr-head resolution and + // failure-log sub-queries are same-target derivatives of the same // input ([`resolve_pr_head`], [`gh_run_scope`]) riding the same // approval, and the `command -v gh` probe is a constant. async fn command_for_gate(&self, input: &Value, _root: &std::path::Path) -> Option { - Some(primary_command(input)) + if input.get("probe").and_then(|v| v.as_bool()).unwrap_or(false) { + Some(settled_command(&gh_run_scope(input, None))) + } else { + Some(primary_command(input)) + } + } + + fn take_wait_request(&self) -> Option { + self.pending_wait + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take() + } +} + +/// The parked-wait request one pending `wait=true` call deposits (#1471). +/// +/// The probe input embeds the target the top-level call already resolved — +/// a PR becomes its resolved head branch, so the engine's per-poll replay +/// never re-pays the `gh pr view` round trip #1526 hoisted out of the old +/// wait loop (the unresolved-head fallback keeps the `pr` field and its +/// embedded lookup, exactly like the composed sub-queries). The wake call +/// is this same tool minus `wait`: full status plus the head-scoped failure +/// logs, fetched once, riding the wake message. +fn park_request(input: &Value, pr_head: Option<&PrHead>) -> stella_core::WaitRequest { + let mut probe = serde_json::Map::new(); + probe.insert("probe".into(), Value::Bool(true)); + let (target_key, target, description) = if let Some(pr) = input.get("pr").and_then(Value::as_u64) + { + match pr_head { + Some(head) => ( + "branch", + Value::String(head.branch.clone()), + format!("CI for PR #{pr} (branch {}) settles", head.branch), + ), + None => ("pr", Value::from(pr), format!("CI for PR #{pr} settles")), + } + } else if let Some(commit) = input.get("commit").and_then(Value::as_str) { + ( + "commit", + Value::String(commit.into()), + format!("CI for commit {commit} settles"), + ) + } else { + let branch = input + .get("branch") + .and_then(Value::as_str) + .unwrap_or("main"); + ( + "branch", + Value::String(branch.into()), + format!("CI for branch {branch} settles"), + ) + }; + probe.insert(target_key.into(), target); + let mut on_wake = input.as_object().cloned().unwrap_or_default(); + on_wake.remove("wait"); + stella_core::WaitRequest { + description, + probe: stella_core::WaitCall { + name: "ci_status".into(), + input: Value::Object(probe), + }, + // The deposit happens precisely because the settledness check said + // `pending`, so that is the baseline the first diverging probe + // (`settled`) wakes against. + baseline: "pending".into(), + on_wake: Some(stella_core::WaitCall { + name: "ci_status".into(), + input: Value::Object(on_wake), + }), + poll_interval_secs: PARK_POLL_INTERVAL_SECS, + // 0 defers to the engine's default park deadline (30 minutes) — + // only an explicit request overrides it, clamped by the engine. + timeout_secs: input + .get("timeout_secs") + .and_then(Value::as_u64) + .unwrap_or(0), } } @@ -219,30 +356,26 @@ fn gh_run_scope(input: &Value, pr_head: Option<&PrHead>) -> String { } } -/// The wait-loop composed as one shell line: repeatedly re-lists THIS -/// target's runs and polls until none remain incomplete (`status != -/// "completed"`), relying on the caller's own `timeout_secs` (the -/// process-group kill in `exec::drive`) as the hang backstop rather than a -/// bound of its own. +/// One settledness check for THIS target, composed as a single shell line: +/// `settled` when no run remains incomplete (`status != "completed"`), +/// `pending` otherwise. The successor of the pre-#1471 in-tool wait loop, +/// keeping #1466's semantics — it counts the WHOLE incomplete set for the +/// target, never a single newest-created run (`gh run list --limit 1` + +/// `gh run watch`), whose short-sibling-finished-first flakiness that issue +/// documents. A `gh run list`/`jq` hiccup yields an empty `$n`, which +/// compares unequal to `"0"` and reads as `pending` — a transient failure +/// keeps the parked turn waiting rather than falsely waking it "done". /// -/// Pulled out of `execute` so it is unit-testable the same way -/// `primary_command`/`gh_run_scope` are (#1466). The pre-fix version watched -/// only the single newest-CREATED run (`gh run list --limit 1` then `gh run -/// watch` on it) — when one push triggers several workflows, the -/// newest-created one is often a short job that is already done, so the -/// watch returned instantly while a long sibling run from the SAME push was -/// still in flight. This polls the WHOLE incomplete set for the target -/// instead of a single arbitrarily-chosen run. A `gh run list`/`jq` hiccup -/// yields an empty `$n`, which compares unequal to `"0"` — so a transient -/// failure keeps the loop waiting rather than falsely reporting "done". -fn wait_command(scope: &str) -> String { +/// The one-word output is deliberate: the engine wakes the turn on ANY +/// fingerprint change (`stella-core`'s `waiting::decide`), so the +/// probe's answer must be constant while runs are in flight — a count +/// (`3 incomplete`) would wake on every finished sibling, and a run list +/// would wake on elapsed-time noise. +fn settled_command(scope: &str) -> String { format!( - "while :; do \ - n=$(gh run list {scope} --limit 15 --json status \ - --jq '[.[] | select(.status != \"completed\")] | length' 2>/dev/null); \ - [ \"$n\" = \"0\" ] && break; \ - sleep 5; \ - done" + "n=$(gh run list {scope} --limit 15 --json status \ + --jq '[.[] | select(.status != \"completed\")] | length' 2>/dev/null); \ + [ \"$n\" = \"0\" ] && echo settled || echo pending" ) } @@ -305,7 +438,7 @@ mod tests { #[test] fn schema_is_read_only_for_verifier_use() { - assert!(CiStatus.schema().read_only); + assert!(CiStatus::default().schema().read_only); } #[test] @@ -345,14 +478,14 @@ mod tests { }; let scope = gh_run_scope(&input, Some(&head)); // The scope is the literal branch, exactly as a branch target's is — - // so the wait loop, which re-runs its scope EVERY poll iteration, no - // longer pays a forge round trip per poll. + // so the settledness probe, which the parked engine replays EVERY + // poll, never pays a forge round trip re-resolving the head. assert_eq!(scope, "--branch 'feat/x'"); - let wait = wait_command(&scope); + let settled = settled_command(&scope); let logs = failure_log_command(&scope, &input, Some(&head)); assert!( - !wait.contains("gh pr view") && !logs.contains("gh pr view"), - "no composed sub-query may re-resolve the PR head:\n{wait}\n{logs}" + !settled.contains("gh pr view") && !logs.contains("gh pr view"), + "no composed sub-query may re-resolve the PR head:\n{settled}\n{logs}" ); // The failure-log SHA filter uses the resolved OID as a literal, and // stays conjoined with the failure selection (#1470's semantics). @@ -401,17 +534,17 @@ mod tests { assert_eq!(shell_quote("a'b"), r"'a'\''b'"); } - // --- #1466: wait=true must watch every run for the target, not just --- - // --- the newest-created one. ------------------------------------------- + // --- #1466 (semantics kept through #1471): settledness must consider --- + // --- every run for the target, not just the newest-created one. -------- #[test] - fn wait_command_polls_the_whole_incomplete_set_instead_of_one_run() { - let cmd = wait_command("--branch 'main'"); - // The pre-fix bug: `gh run list --limit 1` picked a single + fn settled_command_answers_on_the_whole_incomplete_set_not_one_run() { + let cmd = settled_command("--branch 'main'"); + // The pre-#1466 bug: `gh run list --limit 1` picked a single // newest-created run to `gh run watch`, which returns instantly if // THAT run happens to be a short sibling job while a long one from - // the same push is still going. The fixed command must not select a - // single run at all. + // the same push is still going. The probe must not select a single + // run at all. assert!( // Trailing space matters: "--limit 15" also contains the bare // substring "--limit 1". @@ -422,23 +555,76 @@ mod tests { !cmd.contains("gh run watch"), "must not delegate to watching one arbitrarily-chosen run: {cmd}" ); - // It must loop, re-listing THIS target's scope, until nothing is - // left incomplete. - assert!(cmd.contains("while"), "must loop: {cmd}"); + // Single-shot: the WAIT loop is the engine's now (#1471) — a shell + // loop inside the tool call is exactly the cache-aging block this + // probe replaced. + assert!(!cmd.contains("while"), "must not loop in-tool: {cmd}"); + assert!(!cmd.contains("sleep"), "must not sleep in-tool: {cmd}"); assert!(cmd.contains("--branch 'main'"), "must stay scoped: {cmd}"); assert!( cmd.contains("status != \"completed\""), - "must terminate on 'no incomplete runs', not one run's exit: {cmd}" + "settled means 'no incomplete runs', not one run's exit: {cmd}" ); } #[test] - fn wait_command_treats_a_list_failure_as_still_incomplete() { - // An empty `$n` (gh/jq hiccup) must compare unequal to "0" so the - // loop keeps waiting rather than falsely declaring done — the outer - // `timeout_secs` process-group kill is the only backstop. - let cmd = wait_command("--branch 'main'"); + fn settled_command_reads_a_list_failure_as_pending() { + // An empty `$n` (gh/jq hiccup) must compare unequal to "0" so a + // transient failure keeps the parked turn waiting — the park + // deadline is the backstop, never a phantom "settled". + let cmd = settled_command("--branch 'main'"); assert!(cmd.contains(r#"[ "$n" = "0" ]"#), "{cmd}"); + assert!( + cmd.contains("echo settled") && cmd.contains("echo pending"), + "the probe's whole vocabulary is two stable words: {cmd}" + ); + } + + // --- #1471: the parked-wait deposit. ----------------------------------- + + #[test] + fn park_request_probes_by_resolved_head_and_wakes_without_wait() { + let input = serde_json::json!({ "pr": 42, "wait": true, "timeout_secs": 900 }); + let head = PrHead { + branch: "feat/x".into(), + oid: "abc123def".into(), + }; + let req = park_request(&input, Some(&head)); + // The probe replays by the literal resolved branch — one `gh pr + // view` per wait, never one per poll (#1526's discipline). + assert_eq!( + req.probe.input, + serde_json::json!({ "probe": true, "branch": "feat/x" }) + ); + assert_eq!(req.baseline, "pending"); + assert_eq!(req.poll_interval_secs, PARK_POLL_INTERVAL_SECS); + assert_eq!(req.timeout_secs, 900); + // The wake call is this same tool minus `wait` — full status plus + // failure logs, and no re-park loop. + let wake = req.on_wake.expect("a wake call"); + assert_eq!(wake.name, "ci_status"); + assert_eq!( + wake.input, + serde_json::json!({ "pr": 42, "timeout_secs": 900 }) + ); + } + + #[test] + fn park_request_defaults_and_fallbacks_stay_safe() { + // No explicit timeout defers to the engine's park default (0 marks + // "unset" — `stella_core::waiting` maps it to 30 minutes). + let req = park_request(&serde_json::json!({ "branch": "main", "wait": true }), None); + assert_eq!(req.timeout_secs, 0); + assert_eq!( + req.probe.input, + serde_json::json!({ "probe": true, "branch": "main" }) + ); + assert!(req.description.contains("branch main"), "{}", req.description); + // An unresolved PR head keeps the `pr` field, whose probe-side scope + // falls back to the embedded lookup exactly like every other + // composed sub-query. + let req = park_request(&serde_json::json!({ "pr": 7, "wait": true }), None); + assert_eq!(req.probe.input, serde_json::json!({ "probe": true, "pr": 7 })); } // --- #1470: the failure-log tail must be scoped to the target's --- diff --git a/crates/stella-tools/src/exploration.rs b/crates/stella-tools/src/exploration.rs index 8c2ba499e..84ef45d28 100644 --- a/crates/stella-tools/src/exploration.rs +++ b/crates/stella-tools/src/exploration.rs @@ -754,6 +754,36 @@ impl Tool for SaveExploration { } } +/// True when `haystack` mentions `path` as a whole path token: the hit may +/// not extend into path characters on either side, so a map covering +/// `lib.rs` never fires on `mylib.rs` or `graphlib.rs` — a false positive +/// would permanently consume that map's once-per-session coverage hint +/// (the registry's exploration-coverage footer is the caller). +pub(crate) fn mentions_path(haystack: &str, path: &str) -> bool { + if path.is_empty() { + return false; + } + let is_path_char = |c: char| c.is_alphanumeric() || matches!(c, '_' | '-' | '.' | '/'); + let mut from = 0; + while let Some(pos) = haystack[from..].find(path) { + let start = from + pos; + let end = start + path.len(); + // A preceding `/` is a component boundary, not an embedding: search + // results routinely print the workspace-relative map path with an + // absolute prefix (`/tmp/ws/covered.rs` covers `covered.rs`). + let clear_before = !haystack[..start] + .chars() + .next_back() + .is_some_and(|c| is_path_char(c) && c != '/'); + let clear_after = !haystack[end..].chars().next().is_some_and(is_path_char); + if clear_before && clear_after { + return true; + } + from = end; + } + false +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/stella-tools/src/registry.rs b/crates/stella-tools/src/registry.rs index 1e46630fc..816f12d12 100644 --- a/crates/stella-tools/src/registry.rs +++ b/crates/stella-tools/src/registry.rs @@ -61,6 +61,15 @@ pub trait Tool: Send + Sync { fn parallel_safe(&self) -> bool { false } + + /// The parked-wait request this tool deposited during its last + /// `execute`, taken destructively — the registry aggregates it into + /// `ToolExecutor::drain_wait_request` and the engine parks the turn on + /// it at the next step boundary (#1471, `stella_core::waiting`). + /// Defaults to `None`: almost no tool waits on external state. + fn take_wait_request(&self) -> Option { + None + } } /// A file op classified before execution: the normalized path the ledger @@ -1257,7 +1266,7 @@ impl ToolRegistry { let mut hits: Vec = coverage .by_path .iter() - .filter(|(path, _)| mentions_path(&haystack, path)) + .filter(|(path, _)| crate::exploration::mentions_path(&haystack, path)) .flat_map(|(_, slices)| slices.iter().cloned()) .filter(|slice| !coverage.hinted.contains(slice)) .collect(); @@ -2108,35 +2117,6 @@ impl ToolRegistry { } } -/// True when `haystack` mentions `path` as a whole path token: the hit may -/// not extend into path characters on either side, so a map covering -/// `lib.rs` never fires on `mylib.rs` or `graphlib.rs` — a false positive -/// would permanently consume that map's once-per-session coverage hint. -fn mentions_path(haystack: &str, path: &str) -> bool { - if path.is_empty() { - return false; - } - let is_path_char = |c: char| c.is_alphanumeric() || matches!(c, '_' | '-' | '.' | '/'); - let mut from = 0; - while let Some(pos) = haystack[from..].find(path) { - let start = from + pos; - let end = start + path.len(); - // A preceding `/` is a component boundary, not an embedding: search - // results routinely print the workspace-relative map path with an - // absolute prefix (`/tmp/ws/covered.rs` covers `covered.rs`). - let clear_before = !haystack[..start] - .chars() - .next_back() - .is_some_and(|c| is_path_char(c) && c != '/'); - let clear_after = !haystack[end..].chars().next().is_some_and(is_path_char); - if clear_before && clear_after { - return true; - } - from = end; - } - false -} - /// `ToolRegistry` is the production implementation of `stella-core`'s /// `ToolExecutor` port — the engine drives every tool call through this /// impl, never through `stella-tools` types directly. @@ -2157,6 +2137,21 @@ impl ToolExecutor for ToolRegistry { stella_core::subagent::drain_sub_agent_spend(&self.sub_agent_spend) } + /// Collect the parked-wait request a tool deposited this step, if any + /// (#1471). Consults the primary map and the late-enabled overlay — the + /// same two sources every other read path reads. First hit wins: a step + /// dispatches at most a handful of calls, and `Tool::take_wait_request` + /// is destructive, so at most one request exists per boundary. + fn drain_wait_request(&self) -> Option { + let from_primary = self.tools.values().find_map(|tool| tool.take_wait_request()); + from_primary.or_else(|| { + self.late_tools + .read() + .ok() + .and_then(|late| late.values().find_map(|tool| tool.take_wait_request())) + }) + } + /// Aggregate each registered tool's [`Tool::parallel_safe`] claim for the /// engine's dispatch grouping. Consults the primary map and the /// late-enabled overlay — the same two sources every other read path diff --git a/crates/stella-tools/src/registry/process_tools.rs b/crates/stella-tools/src/registry/process_tools.rs index 3a0912cdd..3c182d437 100644 --- a/crates/stella-tools/src/registry/process_tools.rs +++ b/crates/stella-tools/src/registry/process_tools.rs @@ -34,7 +34,7 @@ pub(super) fn builtins( Arc::new(crate::repo::RepoPush(repo.clone())), Arc::new(crate::repo::RepoPull(repo.clone())), Arc::new(crate::repo::RepoRollback(repo)), - Arc::new(crate::ci::CiStatus), + Arc::new(crate::ci::CiStatus::default()), Arc::new(crate::screenshot::Screenshot), Arc::new(crate::tasks::TaskAssign(task_board, spawn_queue)), ] diff --git a/crates/stella-tools/src/registry/tests.rs b/crates/stella-tools/src/registry/tests.rs index 82b8948e4..23ef50e8a 100644 --- a/crates/stella-tools/src/registry/tests.rs +++ b/crates/stella-tools/src/registry/tests.rs @@ -38,6 +38,7 @@ fn with_ambient_search(mut names: Vec<&'static str>) -> Vec<&'static str> { /// once-per-session hint on a file it doesn't cover. #[test] fn mentions_path_requires_token_boundaries() { + use crate::exploration::mentions_path; assert!(mentions_path("src/lib.rs:12: pub fn x()", "src/lib.rs")); assert!(mentions_path("lib.rs", "lib.rs")); // An absolute spelling of a covered relative path still matches. From 69088044d9b0069f5be3cc9a11f7bbd591477a3a Mon Sep 17 00:00:00 2001 From: Stella Test Date: Thu, 6 Aug 2026 03:25:07 -0700 Subject: [PATCH 3/8] =?UTF-8?q?test(stella-core):=20witness=20for=20#1471?= =?UTF-8?q?=20=E2=80=94=20Nth-probe=20change=20costs=20one=20model=20re-in?= =?UTF-8?q?vocation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs #1471 --- crates/stella-core/src/driver/tests.rs | 4 +- .../src/driver/tests/parked_wait.rs | 279 ++++++++++++++++++ 2 files changed, 281 insertions(+), 2 deletions(-) create mode 100644 crates/stella-core/src/driver/tests/parked_wait.rs diff --git a/crates/stella-core/src/driver/tests.rs b/crates/stella-core/src/driver/tests.rs index 02b7c08cd..a9ca8f0d4 100644 --- a/crates/stella-core/src/driver/tests.rs +++ b/crates/stella-core/src/driver/tests.rs @@ -3,8 +3,7 @@ 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::{CompletionUsage, ToolSchema}; use stella_protocol::event::BudgetMode; 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; diff --git a/crates/stella-core/src/driver/tests/parked_wait.rs b/crates/stella-core/src/driver/tests/parked_wait.rs new file mode 100644 index 000000000..53c6561a5 --- /dev/null +++ b/crates/stella-core/src/driver/tests/parked_wait.rs @@ -0,0 +1,279 @@ +//! Witness for #1471: waiting on external state costs O(1) model steps. +//! +//! Before parked waits, a monitoring turn polled as model behavior — one +//! full model round-trip per poll on a growing transcript, and any long +//! in-tool sleep aged the provider prompt cache to nothing. These tests pin +//! the replacement contract end-to-end through `run_turn`: a condition that +//! changes on the Nth engine-side probe re-invokes the model exactly once, +//! the poll history never enters the transcript, and the wake rides the +//! volatile tail as a marked user message. + +use super::*; +use crate::waiting::{WAKE_MARKER, WaitCall, WaitRequest}; + +/// A `ToolExecutor` whose `ci_status` deposits a parked-wait request on +/// `wait=true`, answers `pending` until the `change_on`th probe replay, and +/// serves a recognizable fresh status to the wake call — the tool half of +/// the `ci.rs` composition, scripted. +struct ParkingTools { + deposit: std::sync::Mutex>, + probe_calls: Arc, + change_on: u32, + wake_calls: Arc, +} + +impl ParkingTools { + fn depositing(request: WaitRequest, change_on: u32) -> Self { + Self { + deposit: std::sync::Mutex::new(Some(request)), + probe_calls: Arc::new(AtomicU32::new(0)), + change_on, + wake_calls: Arc::new(AtomicU32::new(0)), + } + } +} + +#[async_trait] +impl ToolExecutor for ParkingTools { + fn schemas(&self) -> Vec { + vec![ToolSchema { + name: "ci_status".into(), + description: "scripted CI".into(), + input_schema: serde_json::json!({"type": "object"}), + read_only: true, + speculation_safe: false, + }] + } + + async fn execute(&self, _name: &str, input: &Value) -> ToolOutput { + if input.get("probe").and_then(Value::as_bool).unwrap_or(false) { + let n = self.probe_calls.fetch_add(1, Ordering::SeqCst) + 1; + return ToolOutput::Ok { + content: if n >= self.change_on { + "settled".into() + } else { + "pending".into() + }, + }; + } + if input.get("wait").and_then(Value::as_bool).unwrap_or(false) { + return ToolOutput::Ok { + content: "2 runs in progress — parking".into(), + }; + } + self.wake_calls.fetch_add(1, Ordering::SeqCst); + ToolOutput::Ok { + content: "fresh status: all green".into(), + } + } + + fn drain_wait_request(&self) -> Option { + self.deposit.lock().unwrap().take() + } +} + +fn ci_wait_request(timeout_secs: u64) -> WaitRequest { + WaitRequest { + description: "CI for branch main settles".into(), + probe: WaitCall { + name: "ci_status".into(), + input: serde_json::json!({ "probe": true, "branch": "main" }), + }, + baseline: "pending".into(), + on_wake: Some(WaitCall { + name: "ci_status".into(), + input: serde_json::json!({ "branch": "main" }), + }), + poll_interval_secs: 5, + timeout_secs, + } +} + +fn ci_wait_call() -> CompletionResultAlias { + CompletionResultAlias { + text: String::new(), + tool_calls: vec![ToolCall { + call_id: "w1".into(), + name: "ci_status".into(), + input: serde_json::json!({ "branch": "main", "wait": true }), + }], + usage: CompletionUsage::reported_zero(), + model: "scripted".into(), + cost_usd: 0.0001, + finish_reason: None, + } +} + +/// The witness: the watched state changes on the THIRD engine-side probe, +/// and the model is re-invoked exactly once — not three times — with the +/// wake delta on the transcript tail and zero poll debris anywhere in it. +#[tokio::test] +async fn a_change_on_the_nth_probe_re_invokes_the_model_exactly_once() { + let provider = ScriptedProvider { + id: "scripted".into(), + script: TokioMutex::new(vec![ + Ok(ci_wait_call()), + Ok(text_result("CI settled — the fix is green, done")), + ]), + calls: Arc::new(AtomicU32::new(0)), + }; + let tools = ParkingTools::depositing(ci_wait_request(600), 3); + let probe_calls = tools.probe_calls.clone(); + let wake_calls = tools.wake_calls.clone(); + let sleeper = NoopSleeper; + let engine = Engine::with_sleeper(&provider, &tools, EngineConfig::default(), &sleeper); + let mut messages = vec![ + CompletionMessage::system("sys"), + CompletionMessage::user("monitor CI and report"), + ]; + let mut budget = BudgetGuard::new(BudgetMode::Off, None, None); + let (tx, mut rx) = mpsc::unbounded_channel(); + + let outcome = engine.run_turn(&mut messages, &mut budget, &tx).await; + + assert!( + matches!(outcome, TurnOutcome::Completed { ref text, .. } if text.contains("green")), + "got {outcome:?}" + ); + // THE contract (#1471): one model call requested the wait, one answered + // the wake. Three probes happened, and none of them was a model step. + assert_eq!( + provider.calls.load(Ordering::SeqCst), + 2, + "a change on the 3rd probe must cost exactly one re-invocation" + ); + assert_eq!(probe_calls.load(Ordering::SeqCst), 3); + assert_eq!(wake_calls.load(Ordering::SeqCst), 1, "one wake replay"); + + // The wake rides the tail as a marked user message carrying the fresh + // status — and the probes' outputs never entered the transcript. + let wake_idx = messages + .iter() + .position(|m| m.role == MessageRole::User && m.content.starts_with(WAKE_MARKER)) + .expect("the wake message must enter the conversation"); + assert!( + messages[wake_idx].content.contains("fresh status: all green"), + "the wake carries the on_wake output: {}", + messages[wake_idx].content + ); + assert!( + messages + .iter() + .rposition(|m| m.role == MessageRole::Assistant) + .expect("final reply") + > wake_idx, + "the model call that answers the wake must observe it" + ); + let polls_in_transcript = messages + .iter() + .filter(|m| m.content.contains("pending") || m.content.contains("settled")) + .count(); + assert_eq!( + polls_in_transcript, 0, + "poll observations must never reach the transcript" + ); + assert_tool_pairing(&messages); + + // The user-facing stream announced the park and the wake. + let events = drain_events(&mut rx); + let texts: Vec<&str> = events + .iter() + .filter_map(|e| match e { + AgentEvent::Text { delta } => Some(delta.as_str()), + _ => None, + }) + .collect(); + assert!( + texts.iter().any(|t| t.contains("Parked: waiting until")), + "the park must be announced: {texts:?}" + ); + assert!( + texts.iter().any(|t| t.contains("Wait ended after 3 probes")), + "the wake must be announced with its probe count: {texts:?}" + ); +} + +/// A condition that never changes wakes the model once with the timeout +/// marked — never N poll-steps, and never a silent hang past the deadline. +#[tokio::test] +async fn an_unchanged_condition_wakes_once_at_the_deadline() { + let provider = ScriptedProvider { + id: "scripted".into(), + script: TokioMutex::new(vec![ + Ok(ci_wait_call()), + Ok(text_result("CI is stuck; escalating instead of waiting")), + ]), + calls: Arc::new(AtomicU32::new(0)), + }; + // 20s deadline at the 5s interval floor: exactly 4 afforded probes. + let tools = ParkingTools::depositing(ci_wait_request(20), u32::MAX); + let probe_calls = tools.probe_calls.clone(); + let wake_calls = tools.wake_calls.clone(); + let sleeper = NoopSleeper; + let engine = Engine::with_sleeper(&provider, &tools, EngineConfig::default(), &sleeper); + let mut messages = vec![ + CompletionMessage::system("sys"), + CompletionMessage::user("monitor CI"), + ]; + let mut budget = BudgetGuard::new(BudgetMode::Off, None, None); + let (tx, _rx) = mpsc::unbounded_channel(); + + let outcome = engine.run_turn(&mut messages, &mut budget, &tx).await; + + assert!(matches!(outcome, TurnOutcome::Completed { .. }), "{outcome:?}"); + assert_eq!(provider.calls.load(Ordering::SeqCst), 2); + assert_eq!(probe_calls.load(Ordering::SeqCst), 4, "deadline / interval"); + assert_eq!( + wake_calls.load(Ordering::SeqCst), + 0, + "no change, no wake replay — the model re-queries if it wants to" + ); + let wake = messages + .iter() + .find(|m| m.role == MessageRole::User && m.content.starts_with(WAKE_MARKER)) + .expect("the timeout wake must enter the conversation"); + assert!( + wake.content.contains("timed out"), + "the model must see the deadline, not a phantom change: {}", + wake.content + ); +} + +/// A request whose replayed calls are not read-only is refused outright — +/// the engine must never mutate on a timer the model cannot see. +#[tokio::test] +async fn a_non_read_only_probe_is_refused_not_replayed() { + let provider = ScriptedProvider { + id: "scripted".into(), + script: TokioMutex::new(vec![ + Ok(ci_wait_call()), + Ok(text_result("done without parking")), + ]), + calls: Arc::new(AtomicU32::new(0)), + }; + let mut request = ci_wait_request(600); + request.probe.name = "bash".into(); // not in the read-only schema set + let tools = ParkingTools::depositing(request, 1); + let probe_calls = tools.probe_calls.clone(); + let sleeper = NoopSleeper; + let engine = Engine::with_sleeper(&provider, &tools, EngineConfig::default(), &sleeper); + let mut messages = vec![ + CompletionMessage::system("sys"), + CompletionMessage::user("monitor CI"), + ]; + let mut budget = BudgetGuard::new(BudgetMode::Off, None, None); + let (tx, mut rx) = mpsc::unbounded_channel(); + + let outcome = engine.run_turn(&mut messages, &mut budget, &tx).await; + + assert!(matches!(outcome, TurnOutcome::Completed { .. }), "{outcome:?}"); + assert_eq!(probe_calls.load(Ordering::SeqCst), 0, "never replayed"); + assert!( + !messages.iter().any(|m| m.content.starts_with(WAKE_MARKER)), + "a refused park must not fabricate a wake" + ); + let refused = drain_events(&mut rx).into_iter().any(|e| { + matches!(&e, AgentEvent::Text { delta } if delta.contains("not a read-only tool")) + }); + assert!(refused, "the refusal must be visible on the stream"); +} From e2b5a0a59b7803f35434a6b5e5b63e7b75a88916 Mon Sep 17 00:00:00 2001 From: Stella Test Date: Thu, 6 Aug 2026 03:34:30 -0700 Subject: [PATCH 4/8] test(stella-core): tighten the poll-debris assertion to verbatim probe words Refs #1471 --- crates/stella-core/src/driver/tests/parked_wait.rs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/crates/stella-core/src/driver/tests/parked_wait.rs b/crates/stella-core/src/driver/tests/parked_wait.rs index 53c6561a5..46819c75e 100644 --- a/crates/stella-core/src/driver/tests/parked_wait.rs +++ b/crates/stella-core/src/driver/tests/parked_wait.rs @@ -164,9 +164,19 @@ async fn a_change_on_the_nth_probe_re_invokes_the_model_exactly_once() { > wake_idx, "the model call that answers the wake must observe it" ); + // Probe outputs are the verbatim words `pending`/`settled`; neither may + // surface anywhere — not as a message, not as a tool result. let polls_in_transcript = messages .iter() - .filter(|m| m.content.contains("pending") || m.content.contains("settled")) + .flat_map(|m| { + std::iter::once(m.content.trim().to_string()).chain(m.tool_results.iter().map(|r| { + match &r.output { + ToolOutput::Ok { content } => content.trim().to_string(), + ToolOutput::Error { message } => message.trim().to_string(), + } + })) + }) + .filter(|content| content == "pending" || content == "settled") .count(); assert_eq!( polls_in_transcript, 0, From 7ab09fc5ca7a6de550edd0eb4ddf16905b8f122f Mon Sep 17 00:00:00 2001 From: Stella Test Date: Thu, 6 Aug 2026 03:53:20 -0700 Subject: [PATCH 5/8] feat(stella-cli,serve,mcp,tools): forward drain_wait_request through every tool-stack decorator TaskTap moves to command_deck/task_tap.rs (the settlement.rs split pattern) so the god file shrinks instead of growing. Pinned end-to-end by the_production_tool_stack_forwards_wait_requests, the wait twin of the spend-forwarding witness. Refs #1471 --- crates/stella-cli/src/claims.rs | 7 ++ crates/stella-cli/src/command_deck.rs | 48 +----------- .../stella-cli/src/command_deck/task_tap.rs | 63 ++++++++++++++++ crates/stella-cli/src/discovery.rs | 7 ++ crates/stella-cli/src/fleet_commits.rs | 7 ++ crates/stella-cli/src/interactive.rs | 7 ++ crates/stella-cli/src/subagent/tests.rs | 62 +++++++++++++++ crates/stella-cli/src/tool_policy.rs | 7 ++ crates/stella-core/src/driver/tests.rs | 2 +- .../src/driver/tests/parked_wait.rs | 24 ++++-- crates/stella-core/src/lib.rs | 2 +- crates/stella-mcp/src/toolset.rs | 17 +++++ crates/stella-serve/src/subagents.rs | 7 ++ crates/stella-tools/src/ci.rs | 75 +++++++++++-------- crates/stella-tools/src/custom.rs | 7 ++ crates/stella-tools/src/hunk_review.rs | 7 ++ crates/stella-tools/src/registry.rs | 5 +- 17 files changed, 268 insertions(+), 86 deletions(-) create mode 100644 crates/stella-cli/src/command_deck/task_tap.rs diff --git a/crates/stella-cli/src/claims.rs b/crates/stella-cli/src/claims.rs index e0716bb8c..46463aea3 100644 --- a/crates/stella-cli/src/claims.rs +++ b/crates/stella-cli/src/claims.rs @@ -349,6 +349,13 @@ impl ToolExecutor for ClaimTap<'_> { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.drain_sub_agent_spend_usd() } + + /// 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 { + self.inner.drain_wait_request() + } } #[cfg(test)] diff --git a/crates/stella-cli/src/command_deck.rs b/crates/stella-cli/src/command_deck.rs index 02f4ef221..5eba96499 100644 --- a/crates/stella-cli/src/command_deck.rs +++ b/crates/stella-cli/src/command_deck.rs @@ -117,6 +117,8 @@ mod profile_cmd; mod scope_gate; mod session_clear; mod settle; +mod task_tap; +use task_tap::TaskTap; mod theme_cmd; use crate::memory::{SessionMemory, inject_recall_block}; use crate::runtime::{SystemClock, TokioSleeper}; @@ -4690,51 +4692,5 @@ impl AskUserIo for DeckAskUserIo { } } -/// Mirrors the task board into the event stream: after any `task_*` tool -/// call the FULL board snapshot rides the turn's channel as -/// `AgentEvent::TaskUpdate` — persisted by the forwarder, so replay shows -/// the checklist exactly as it moved — and `task_assign`'s spawn requests -/// are handed to the driver's supervisor channel. `supervisor: None` is the -/// worker configuration (v1 delegation runs from the lead only; a worker's -/// stranded requests are reported on its lane by `crate::subsession`). -pub(crate) struct TaskTap<'a> { - pub(crate) inner: &'a dyn ToolExecutor, - pub(crate) events: UnboundedSender, - pub(crate) registry: &'a ToolRegistry, - pub(crate) supervisor: Option>, -} - -#[async_trait] -impl ToolExecutor for TaskTap<'_> { - fn schemas(&self) -> Vec { - self.inner.schemas() - } - - async fn execute(&self, name: &str, input: &Value) -> ToolOutput { - let output = self.inner.execute(name, input).await; - if name.starts_with("task_") { - let tasks: Vec = { - let board = self.registry.task_board(); - let guard = board.lock().unwrap_or_else(|p| p.into_inner()); - guard.items().to_vec() - }; - let _ = self.events.send(AgentEvent::TaskUpdate { tasks }); - if let Some(sup) = &self.supervisor { - for request in self.registry.take_spawn_requests() { - let _ = sup.send(SupervisorMsg::SpawnTask(request)); - } - } - } - output - } - - /// Forwarded: this is a decorator, and a decorator that let the default - /// `0.0` stand would silently drop sub-agent spend out of the parent's - /// budget (see the port's contract). - fn drain_sub_agent_spend_usd(&self) -> f64 { - self.inner.drain_sub_agent_spend_usd() - } -} - #[cfg(test)] mod tests; diff --git a/crates/stella-cli/src/command_deck/task_tap.rs b/crates/stella-cli/src/command_deck/task_tap.rs new file mode 100644 index 000000000..5ce27e793 --- /dev/null +++ b/crates/stella-cli/src/command_deck/task_tap.rs @@ -0,0 +1,63 @@ +//! 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; +use stella_core::ports::ToolExecutor; +use stella_protocol::{AgentEvent, TaskItem, ToolOutput, ToolSchema}; +use stella_tools::ToolRegistry; +use tokio::sync::mpsc::UnboundedSender; + +use crate::subsession::SupervisorMsg; + +/// Mirrors the task board into the event stream: after any `task_*` tool +/// call the FULL board snapshot rides the turn's channel as +/// `AgentEvent::TaskUpdate` — persisted by the forwarder, so replay shows +/// the checklist exactly as it moved — and `task_assign`'s spawn requests +/// are handed to the driver's supervisor channel. `supervisor: None` is the +/// worker configuration (v1 delegation runs from the lead only; a worker's +/// stranded requests are reported on its lane by `crate::subsession`). +pub(crate) struct TaskTap<'a> { + pub(crate) inner: &'a dyn ToolExecutor, + pub(crate) events: UnboundedSender, + pub(crate) registry: &'a ToolRegistry, + pub(crate) supervisor: Option>, +} + +#[async_trait] +impl ToolExecutor for TaskTap<'_> { + fn schemas(&self) -> Vec { + self.inner.schemas() + } + + async fn execute(&self, name: &str, input: &Value) -> ToolOutput { + let output = self.inner.execute(name, input).await; + if name.starts_with("task_") { + let tasks: Vec = { + let board = self.registry.task_board(); + let guard = board.lock().unwrap_or_else(|p| p.into_inner()); + guard.items().to_vec() + }; + let _ = self.events.send(AgentEvent::TaskUpdate { tasks }); + if let Some(sup) = &self.supervisor { + for request in self.registry.take_spawn_requests() { + let _ = sup.send(SupervisorMsg::SpawnTask(request)); + } + } + } + output + } + + /// Forwarded: this is a decorator, and a decorator that let the default + /// `0.0` stand would silently drop sub-agent spend out of the parent's + /// budget (see the port's contract). + fn drain_sub_agent_spend_usd(&self) -> f64 { + self.inner.drain_sub_agent_spend_usd() + } + + /// 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 { + self.inner.drain_wait_request() + } +} diff --git a/crates/stella-cli/src/discovery.rs b/crates/stella-cli/src/discovery.rs index 5875b1ab2..0d0b2112b 100644 --- a/crates/stella-cli/src/discovery.rs +++ b/crates/stella-cli/src/discovery.rs @@ -785,6 +785,13 @@ impl ToolExecutor for DiscoveryToolSet<'_> { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.drain_sub_agent_spend_usd() } + + /// 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 { + self.inner.drain_wait_request() + } } /// Split an advertised `mcp__server__tool` name into (server, tool). diff --git a/crates/stella-cli/src/fleet_commits.rs b/crates/stella-cli/src/fleet_commits.rs index a2909b19a..5562eae5b 100644 --- a/crates/stella-cli/src/fleet_commits.rs +++ b/crates/stella-cli/src/fleet_commits.rs @@ -131,6 +131,13 @@ impl ToolExecutor for CommitObserver<'_> { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.drain_sub_agent_spend_usd() } + + /// 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 { + self.inner.drain_wait_request() + } } /// One attempt's commits, oldest first — and the two ways of knowing which diff --git a/crates/stella-cli/src/interactive.rs b/crates/stella-cli/src/interactive.rs index d63fa7330..153ba0994 100644 --- a/crates/stella-cli/src/interactive.rs +++ b/crates/stella-cli/src/interactive.rs @@ -527,6 +527,13 @@ impl ToolExecutor for InteractiveToolSet<'_> { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.drain_sub_agent_spend_usd() } + + /// 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 { + self.inner.drain_wait_request() + } } #[cfg(test)] diff --git a/crates/stella-cli/src/subagent/tests.rs b/crates/stella-cli/src/subagent/tests.rs index 75cc9f42d..07fb146ad 100644 --- a/crates/stella-cli/src/subagent/tests.rs +++ b/crates/stella-cli/src/subagent/tests.rs @@ -101,6 +101,68 @@ async fn the_production_tool_stack_forwards_sub_agent_spend() { ); } +/// 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_wait_requests() { + /// A leaf holding one deposited request, standing in for the registry. + struct WaitingBase(std::sync::Mutex>); + + #[async_trait] + impl ToolExecutor for WaitingBase { + fn schemas(&self) -> Vec { + Vec::new() + } + async fn execute(&self, _name: &str, _input: &Value) -> ToolOutput { + ToolOutput::Ok { + content: String::new(), + } + } + fn drain_wait_request(&self) -> Option { + 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(".")); + let (stub_tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + let interactive = crate::interactive::InteractiveToolSet::new( + &customs, + stub_tx, + crate::interactive::default_ask_io(false), + ); + let permitted = crate::agent::PolicyToolSet::new(&interactive, Default::default()); + let discovery = + crate::discovery::DiscoveryToolSet::new(&permitted, std::path::PathBuf::from(".")); + + 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" + ); +} + /// A dispatcher whose registry has been dropped reports a refusal rather /// than panicking a torn-down session. #[tokio::test] diff --git a/crates/stella-cli/src/tool_policy.rs b/crates/stella-cli/src/tool_policy.rs index 1fa2361fc..814ce2928 100644 --- a/crates/stella-cli/src/tool_policy.rs +++ b/crates/stella-cli/src/tool_policy.rs @@ -102,6 +102,13 @@ impl ToolExecutor for PolicyToolSet<'_> { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.get().drain_sub_agent_spend_usd() } + + /// 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 { + self.inner.get().drain_wait_request() + } } /// Which `"tools"` key withheld `name` — the exact name, its group, or the diff --git a/crates/stella-core/src/driver/tests.rs b/crates/stella-core/src/driver/tests.rs index a9ca8f0d4..e9c453074 100644 --- a/crates/stella-core/src/driver/tests.rs +++ b/crates/stella-core/src/driver/tests.rs @@ -3,8 +3,8 @@ use std::sync::atomic::{AtomicU32, Ordering}; use async_trait::async_trait; use serde_json::Value; -use stella_protocol::{CompletionUsage, ToolSchema}; use stella_protocol::event::BudgetMode; +use stella_protocol::{CompletionUsage, ToolSchema}; use tokio::sync::Mutex as TokioMutex; use tokio::sync::mpsc; diff --git a/crates/stella-core/src/driver/tests/parked_wait.rs b/crates/stella-core/src/driver/tests/parked_wait.rs index 46819c75e..6c4db0b34 100644 --- a/crates/stella-core/src/driver/tests/parked_wait.rs +++ b/crates/stella-core/src/driver/tests/parked_wait.rs @@ -152,7 +152,9 @@ async fn a_change_on_the_nth_probe_re_invokes_the_model_exactly_once() { .position(|m| m.role == MessageRole::User && m.content.starts_with(WAKE_MARKER)) .expect("the wake message must enter the conversation"); assert!( - messages[wake_idx].content.contains("fresh status: all green"), + messages[wake_idx] + .content + .contains("fresh status: all green"), "the wake carries the on_wake output: {}", messages[wake_idx].content ); @@ -198,7 +200,9 @@ async fn a_change_on_the_nth_probe_re_invokes_the_model_exactly_once() { "the park must be announced: {texts:?}" ); assert!( - texts.iter().any(|t| t.contains("Wait ended after 3 probes")), + texts + .iter() + .any(|t| t.contains("Wait ended after 3 probes")), "the wake must be announced with its probe count: {texts:?}" ); } @@ -230,7 +234,10 @@ async fn an_unchanged_condition_wakes_once_at_the_deadline() { let outcome = engine.run_turn(&mut messages, &mut budget, &tx).await; - assert!(matches!(outcome, TurnOutcome::Completed { .. }), "{outcome:?}"); + assert!( + matches!(outcome, TurnOutcome::Completed { .. }), + "{outcome:?}" + ); assert_eq!(provider.calls.load(Ordering::SeqCst), 2); assert_eq!(probe_calls.load(Ordering::SeqCst), 4, "deadline / interval"); assert_eq!( @@ -276,14 +283,17 @@ async fn a_non_read_only_probe_is_refused_not_replayed() { let outcome = engine.run_turn(&mut messages, &mut budget, &tx).await; - assert!(matches!(outcome, TurnOutcome::Completed { .. }), "{outcome:?}"); + assert!( + matches!(outcome, TurnOutcome::Completed { .. }), + "{outcome:?}" + ); assert_eq!(probe_calls.load(Ordering::SeqCst), 0, "never replayed"); assert!( !messages.iter().any(|m| m.content.starts_with(WAKE_MARKER)), "a refused park must not fabricate a wake" ); - let refused = drain_events(&mut rx).into_iter().any(|e| { - matches!(&e, AgentEvent::Text { delta } if delta.contains("not a read-only tool")) - }); + let refused = drain_events(&mut rx).into_iter().any( + |e| matches!(&e, AgentEvent::Text { delta } if delta.contains("not a read-only tool")), + ); assert!(refused, "the refusal must be visible on the stream"); } diff --git a/crates/stella-core/src/lib.rs b/crates/stella-core/src/lib.rs index 37e5a0a83..c647a1e08 100644 --- a/crates/stella-core/src/lib.rs +++ b/crates/stella-core/src/lib.rs @@ -102,7 +102,7 @@ pub use subagent::{ forwards_to_parent, push_sub_agent_spend, }; pub use tasks::{SpawnRequest, TaskBoard, TaskBoardError}; -pub use waiting::{WaitCall, WaitRequest}; pub use tool_foundry::{ GapDetectionConfig, ParamKind, ProposedTool, ShellInvocation, ToolParameter, detect_tool_gaps, }; +pub use waiting::{WaitCall, WaitRequest}; diff --git a/crates/stella-mcp/src/toolset.rs b/crates/stella-mcp/src/toolset.rs index 5d1821eb2..6097458a2 100644 --- a/crates/stella-mcp/src/toolset.rs +++ b/crates/stella-mcp/src/toolset.rs @@ -560,6 +560,15 @@ impl ToolExecutor for McpToolSet { .as_ref() .map_or(0.0, |native| native.drain_sub_agent_spend_usd()) } + + /// 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 { + self.native + .as_ref() + .and_then(|native| native.drain_wait_request()) + } } /// A Best-of-N candidate's tool surface (issue #248 Phase 1): built by @@ -613,6 +622,14 @@ impl ToolExecutor for CandidateMcpView { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.drain_sub_agent_spend_usd() } + + /// Forwarded so a swallowed wait request cannot turn parked waits + /// (#1471) back into model-step polling — to `native`, not `inner`: + /// remote MCP tools never deposit wait requests, and the native layer + /// this view executes (`ci_status` included) is the only depositor. + fn drain_wait_request(&self) -> Option { + self.native.drain_wait_request() + } } /// Compose the namespaced tool name for a server/tool pair. diff --git a/crates/stella-serve/src/subagents.rs b/crates/stella-serve/src/subagents.rs index 6d5733ed8..4497f13ad 100644 --- a/crates/stella-serve/src/subagents.rs +++ b/crates/stella-serve/src/subagents.rs @@ -437,6 +437,13 @@ impl ToolExecutor for DelegatingTools<'_> { self.inner.drain_sub_agent_spend_usd() + stella_core::subagent::drain_sub_agent_spend(&self.spend) } + + /// 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 { + self.inner.drain_wait_request() + } } /// Turn a child's outcome into a model-visible result. diff --git a/crates/stella-tools/src/ci.rs b/crates/stella-tools/src/ci.rs index fc09e875d..8a9eea87e 100644 --- a/crates/stella-tools/src/ci.rs +++ b/crates/stella-tools/src/ci.rs @@ -87,7 +87,11 @@ impl Tool for CiStatus { // would fake the state change the engine is watching for. Probe // inputs deposited below carry the already-resolved branch, so this // path re-resolves a PR head only when a caller passes `pr` itself. - if input.get("probe").and_then(|v| v.as_bool()).unwrap_or(false) { + if input + .get("probe") + .and_then(|v| v.as_bool()) + .unwrap_or(false) + { let pr_head = match input.get("pr").and_then(|v| v.as_u64()) { Some(pr) => resolve_pr_head(pr, root, timeout_secs).await, None => None, @@ -187,7 +191,11 @@ impl Tool for CiStatus { // input ([`resolve_pr_head`], [`gh_run_scope`]) riding the same // approval, and the `command -v gh` probe is a constant. async fn command_for_gate(&self, input: &Value, _root: &std::path::Path) -> Option { - if input.get("probe").and_then(|v| v.as_bool()).unwrap_or(false) { + if input + .get("probe") + .and_then(|v| v.as_bool()) + .unwrap_or(false) + { Some(settled_command(&gh_run_scope(input, None))) } else { Some(primary_command(input)) @@ -214,33 +222,33 @@ impl Tool for CiStatus { fn park_request(input: &Value, pr_head: Option<&PrHead>) -> stella_core::WaitRequest { let mut probe = serde_json::Map::new(); probe.insert("probe".into(), Value::Bool(true)); - let (target_key, target, description) = if let Some(pr) = input.get("pr").and_then(Value::as_u64) - { - match pr_head { - Some(head) => ( + let (target_key, target, description) = + if let Some(pr) = input.get("pr").and_then(Value::as_u64) { + match pr_head { + Some(head) => ( + "branch", + Value::String(head.branch.clone()), + format!("CI for PR #{pr} (branch {}) settles", head.branch), + ), + None => ("pr", Value::from(pr), format!("CI for PR #{pr} settles")), + } + } else if let Some(commit) = input.get("commit").and_then(Value::as_str) { + ( + "commit", + Value::String(commit.into()), + format!("CI for commit {commit} settles"), + ) + } else { + let branch = input + .get("branch") + .and_then(Value::as_str) + .unwrap_or("main"); + ( "branch", - Value::String(head.branch.clone()), - format!("CI for PR #{pr} (branch {}) settles", head.branch), - ), - None => ("pr", Value::from(pr), format!("CI for PR #{pr} settles")), - } - } else if let Some(commit) = input.get("commit").and_then(Value::as_str) { - ( - "commit", - Value::String(commit.into()), - format!("CI for commit {commit} settles"), - ) - } else { - let branch = input - .get("branch") - .and_then(Value::as_str) - .unwrap_or("main"); - ( - "branch", - Value::String(branch.into()), - format!("CI for branch {branch} settles"), - ) - }; + Value::String(branch.into()), + format!("CI for branch {branch} settles"), + ) + }; probe.insert(target_key.into(), target); let mut on_wake = input.as_object().cloned().unwrap_or_default(); on_wake.remove("wait"); @@ -619,12 +627,19 @@ mod tests { req.probe.input, serde_json::json!({ "probe": true, "branch": "main" }) ); - assert!(req.description.contains("branch main"), "{}", req.description); + assert!( + req.description.contains("branch main"), + "{}", + req.description + ); // An unresolved PR head keeps the `pr` field, whose probe-side scope // falls back to the embedded lookup exactly like every other // composed sub-query. let req = park_request(&serde_json::json!({ "pr": 7, "wait": true }), None); - assert_eq!(req.probe.input, serde_json::json!({ "probe": true, "pr": 7 })); + assert_eq!( + req.probe.input, + serde_json::json!({ "probe": true, "pr": 7 }) + ); } // --- #1470: the failure-log tail must be scoped to the target's --- diff --git a/crates/stella-tools/src/custom.rs b/crates/stella-tools/src/custom.rs index 60327e788..6a6691da6 100644 --- a/crates/stella-tools/src/custom.rs +++ b/crates/stella-tools/src/custom.rs @@ -768,6 +768,13 @@ impl ToolExecutor for CustomToolSet<'_> { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.get().drain_sub_agent_spend_usd() } + + /// 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 { + self.inner.get().drain_wait_request() + } } #[cfg(test)] diff --git a/crates/stella-tools/src/hunk_review.rs b/crates/stella-tools/src/hunk_review.rs index 910d4348a..43e798913 100644 --- a/crates/stella-tools/src/hunk_review.rs +++ b/crates/stella-tools/src/hunk_review.rs @@ -456,6 +456,13 @@ impl ToolExecutor for HunkGate<'_> { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.drain_sub_agent_spend_usd() } + + /// 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 { + self.inner.drain_wait_request() + } } #[cfg(test)] diff --git a/crates/stella-tools/src/registry.rs b/crates/stella-tools/src/registry.rs index 816f12d12..8da402164 100644 --- a/crates/stella-tools/src/registry.rs +++ b/crates/stella-tools/src/registry.rs @@ -2143,7 +2143,10 @@ impl ToolExecutor for ToolRegistry { /// dispatches at most a handful of calls, and `Tool::take_wait_request` /// is destructive, so at most one request exists per boundary. fn drain_wait_request(&self) -> Option { - let from_primary = self.tools.values().find_map(|tool| tool.take_wait_request()); + let from_primary = self + .tools + .values() + .find_map(|tool| tool.take_wait_request()); from_primary.or_else(|| { self.late_tools .read() From d07ef33a6f04c4d948b35be7b65e6dada641fcd8 Mon Sep 17 00:00:00 2001 From: Stella Test Date: Thu, 6 Aug 2026 03:55:28 -0700 Subject: [PATCH 6/8] fix(stella-cli): drop the ToolSchema import stranded by the TaskTap split Refs #1471 --- crates/stella-cli/src/command_deck.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/stella-cli/src/command_deck.rs b/crates/stella-cli/src/command_deck.rs index 5eba96499..b79e3e185 100644 --- a/crates/stella-cli/src/command_deck.rs +++ b/crates/stella-cli/src/command_deck.rs @@ -88,7 +88,7 @@ use stella_pipeline::{ }; use stella_protocol::{ AgentEvent, CiStatus, CompletionMessage, CompletionRequest, ModelRef, PrStatus, TaskItem, - ToolOutput, ToolSchema, + ToolOutput, }; use stella_store::Store; use stella_tools::ToolRegistry; From 40d8c534469c3e0213a38f2d3a82a58250ad64d8 Mon Sep 17 00:00:00 2001 From: Stella Test Date: Thu, 6 Aug 2026 03:57:55 -0700 Subject: [PATCH 7/8] docs(stella-core): add parked waits to the README module map and step-phase list The step loop's phase sequence gains its tail phase (maybe_park) and the layout table names the two new waiting modules. Closes #1471 --- crates/stella-core/README.md | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/crates/stella-core/README.md b/crates/stella-core/README.md index 9644765e6..59b1b1572 100644 --- a/crates/stella-core/README.md +++ b/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 From 4f28e74a0c3bb476c9b26efe26780211aac45288 Mon Sep 17 00:00:00 2001 From: Stella Test Date: Thu, 6 Aug 2026 03:59:16 -0700 Subject: [PATCH 8/8] fix(stella-core): collapse the wake decision's nested if (clippy) Refs #1471 --- crates/stella-core/src/waiting.rs | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/crates/stella-core/src/waiting.rs b/crates/stella-core/src/waiting.rs index d13197170..1d24d98ac 100644 --- a/crates/stella-core/src/waiting.rs +++ b/crates/stella-core/src/waiting.rs @@ -159,10 +159,8 @@ pub fn decide( observed: Option<&str>, polls_used: u64, ) -> Option { - if let Some(fingerprint) = observed { - if fingerprint != request.baseline { - return Some(WakeReason::Changed); - } + if observed.is_some_and(|fingerprint| fingerprint != request.baseline) { + return Some(WakeReason::Changed); } if polls_used >= request.max_polls() { return Some(WakeReason::DeadlineExpired);