fix(stella-core): read every instant, wait and timeout through the Sleeper port

.github/workflows/ci.yml

@@ -335,7 +335,8 @@ jobs:
       # read the source to check: the retry ladder drew its jitter from the
       # OS entropy pool and the hook bus stamped events from SystemTime while
       # both files' headers said otherwise. A floor over the I/O surfaces and
-      # the manifest, plus a down-only count of Instant::now() reads.
+      # the manifest, plus a down-only count of Instant::now() reads that
+      # reached zero on 2026-09-10 and may not rise.
       - name: no I/O in stella-core, and no new clock read
         if: ${{ !cancelled() && steps.checkout.outcome == 'success' }}
         run: python3 ./scripts/check-core-no-io.py

AGENTS.md

@@ -130,8 +130,8 @@ make gate                # = no-scratch + no-secrets + design-refs
                          #   + core-reachability (a stella-core module is
                          #     reachable from the engine's step path; down-only)
                          #   + core-no-io (no shipping stella-core source or
-                         #     dependency names an I/O surface; Instant::now()
-                         #     reads are a down-only count)
+                         #     dependency names an I/O surface, a timer or a
+                         #     clock; Instant::now() is a down-only count, at 0)
                          #   + typed-errors
                          #   + tool-error-class (#3167 unclassified-ToolOutput::error ratchet)
                          #   + dead-code-allows
@@ -947,16 +947,19 @@ Append; do not renumber. `scripts/check-invariants.sh` enforces both halves.
    directories and `tests.rs` files stripped — and its `[dependencies]`.
    Three questions. A **floor**: no line names the filesystem, a process,
    the network, the environment, a standard stream, the wall clock
-   (`SystemTime::now`), a sleep, an entropy source, or a print macro, and
-   no dependency is an I/O or entropy crate (`tokio` may take only
-   `sync`, `time`, `macros` and `rt`). There is no baseline for the floor:
-   the tree has none and gains none. A **ratchet**: `Instant::now()` reads
-   are counted per file in `scripts/core-no-io-baseline.txt`, down-only,
-   because the monotonic clock is ambient state a replay cannot reproduce
-   even though it is not I/O; `make core-no-io-update` refuses to add a
-   file or raise a count, so a red run is cleared by reading the clock once
-   at the edge and passing `now` in, or by taking `ports::Clock`. The
-   guard was written on 2026-09-09 over two breaches this rule had
+   (`SystemTime::now`), a thread sleep, tokio's timer, a hidden clock read
+   (`.elapsed()`), an entropy source, or a print macro, and no dependency
+   is an I/O or entropy crate (`tokio` may take only `sync`, `macros` and
+   `rt` — not `time`). There is no baseline for the floor: the tree has
+   none and gains none. A **ratchet**: `Instant::now()` reads are counted
+   per file in `scripts/core-no-io-baseline.txt`, down-only, because the
+   monotonic clock is ambient state a replay cannot reproduce even though
+   it is not I/O. The count reached zero on 2026-09-10 and
+   `make core-no-io-update` refuses to add a file or raise it, so any read
+   fails. The engine's time comes through one port, `retry::Sleeper`: `now`
+   for every deadline and every elapsed time, `sleep` for every wait, and
+   `retry::bounded` — the port's sleep racing a call — for every timeout.
+   The guard was written on 2026-09-09 over two breaches this rule had
    carried unread: the retry ladder drew its jitter from `rand::rng()`, and
    the hook bus stamped every event from `SystemTime::now()`, while each
    file's header said the crate reads nothing directly. Both take a port

crates/stella-cli/src/agent/resume.rs

@@ -194,7 +194,11 @@ pub(crate) async fn run_resume(cfg: &Config, id: Option<&str>) -> Result<(), Cli
         );
         let hook_runner = HostHookRunner;
         let engine_config = engine_config_for(cfg);
-        let state = stella_core::step::TurnState::from_checkpoint(checkpoint, &engine_config);
+        let state = stella_core::step::TurnState::from_checkpoint(
+            checkpoint,
+            &engine_config,
+            std::time::Instant::now(),
+        );
         // Assembled, not built up by optional builders. This turn is the
         // `Resume` lane and says so. `lane_capabilities::resume` answers
         // every seam, so nothing is left to a chain.
@@ -496,7 +500,8 @@ mod tests {
             checkpoint_sink: Some(sink.clone()),
             ..EngineConfig::default()
         };
-        let state = TurnState::from_checkpoint(killed_mid_turn(), &config);
+        let state =
+            TurnState::from_checkpoint(killed_mid_turn(), &config, std::time::Instant::now());
         let seams = TurnCapabilities::none();
         let engine = Engine::assemble(
             &provider,
@@ -578,7 +583,7 @@ mod tests {
         };
         let mut at_cap = killed_mid_turn();
         at_cap.step = HOST_CAP;
-        let state = TurnState::from_checkpoint(at_cap, &config);
+        let state = TurnState::from_checkpoint(at_cap, &config, std::time::Instant::now());
         let seams = TurnCapabilities::none();
         let engine = Engine::assemble(
             &provider,
@@ -633,7 +638,7 @@ mod tests {
         };
         let mut at_cap = killed_mid_turn();
         at_cap.step = HOST_CAP;
-        let state = TurnState::from_checkpoint(at_cap, &config);
+        let state = TurnState::from_checkpoint(at_cap, &config, std::time::Instant::now());
         let seams = TurnCapabilities::none();
         let engine = Engine::assemble(
             &provider,

crates/stella-cli/src/runtime.rs

@@ -69,6 +69,10 @@ impl Sleeper for TokioSleeper {
         tokio::time::sleep(std::time::Duration::from_millis(duration_ms)).await;
     }
 
+    fn now(&self) -> std::time::Instant {
+        std::time::Instant::now()
+    }
+
     fn jitter(&self, upper: u64) -> u64 {
         rand::rng().random_range(0..=upper)
     }

crates/stella-core/Cargo.toml

@@ -14,7 +14,7 @@ serde = { workspace = true }
 serde_json = { workspace = true }
 serde_json_canonicalizer = { workspace = true }
 thiserror = { workspace = true }
-tokio = { workspace = true, features = ["sync", "time"] }
+tokio = { workspace = true, features = ["sync"] }
 async-trait = { workspace = true }
 futures-util = { workspace = true }
 sha2 = { workspace = true }

crates/stella-core/README.md

@@ -14,9 +14,11 @@ outside world is a trait the caller implements: `ToolExecutor`, `Clock`,
 ([`src/retry.rs`](src/retry.rs)), `HookRunner` ([`src/hooks.rs`](src/hooks.rs))
 — plus `Provider` from `stella-protocol`. `make core-no-io`
 (`scripts/check-core-no-io.py`) reads the shipping source and the manifest
-and fails on any of those surfaces named directly; the `Instant::now()` reads
-the deadline arithmetic still makes are a down-only count in
-`scripts/core-no-io-baseline.txt`, cleared by passing `now` in from the edge.
+and fails on any of those surfaces named directly — a hidden clock read
+(`.elapsed()`) and a tokio timer included. Every deadline is an `Instant`
+from `Sleeper::now`, every timeout is `retry::bounded` racing the port's
+sleep, and `scripts/core-no-io-baseline.txt` holds the `Instant::now()`
+count at zero.
 Even the working directory is passed in (`EngineConfig::cwd`) rather than read
 from `std::env`. That is what makes compaction, eviction, loop detection and
 budget arithmetic plain synchronous functions over owned data, testable against

crates/stella-core/src/accounted_call.rs

@@ -1,13 +1,12 @@
 //! I/O-free one-shot provider accounting shared by non-engine callers.
 
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 use stella_protocol::{
     AgentEvent, CompletionRequest, CompletionResult, ModelCallRole, Provider, ProviderError,
     ToolCall, ToolCallObserver, UsageIncompleteReason,
 };
-use tokio::time::timeout;
 
 use crate::budget::{BudgetGuard, BudgetOutcome, DeadlineOutcome};
 use crate::event_sender::EventSender;
@@ -157,7 +156,7 @@ pub async fn run_accounted_call(
     events: &EventSender,
     sleeper: &dyn Sleeper,
 ) -> Result<CompletionResult, AccountedCallError> {
-    let started = Instant::now();
+    let started = sleeper.now();
     // The task's wall clock, checked BEFORE the dispatch (#2238). This seam is
     // between model calls by construction — an `AccountedCall` carries no
     // tools, so there is never anything in flight to interrupt — which makes
@@ -245,12 +244,12 @@ pub async fn run_accounted_call(
             let mut future = std::pin::pin!(future);
             let mut seen = progress.count();
             loop {
-                match timeout(limit, &mut future).await {
-                    Ok(Ok(outcome)) => break outcome,
-                    Ok(Err(error)) => {
+                match crate::retry::bounded(sleeper, limit, &mut future).await {
+                    Some(Ok(outcome)) => break outcome,
+                    Some(Err(error)) => {
                         return Err(AccountedCallError::Provider(error));
                     }
-                    Err(_) => {
+                    None => {
                         // The window elapsed with the call still unresolved.
                         // Whether that is the deadline this bound exists for
                         // depends on what arrived during it: any fragment at
@@ -271,7 +270,13 @@ pub async fn run_accounted_call(
                                 // so nothing was salvaged: the stream is
                                 // still open and its usage frame may yet
                                 // have been in flight.
-                                emit_incomplete(&call, events, started.elapsed(), None, None);
+                                emit_incomplete(
+                                    &call,
+                                    events,
+                                    sleeper.now().duration_since(started),
+                                    None,
+                                    None,
+                                );
                             }
                             return Err(AccountedCallError::Timeout);
                         }
@@ -353,7 +358,7 @@ pub async fn run_accounted_call(
         reasoning_tokens: result.usage.reasoning_tokens,
         estimated_input_tokens: call.estimated_input_tokens,
         cost_usd: result.cost_usd,
-        duration_ms: started.elapsed().as_millis() as u64,
+        duration_ms: sleeper.now().duration_since(started).as_millis() as u64,
         retries: outcome.retries.len() as u32,
         tool_calls: result.tool_calls.len(),
         complete: result.usage.is_complete(),
@@ -372,7 +377,7 @@ pub async fn run_accounted_call(
         task_id: None,
     });
     let budget_outcome = budget.record_spend(result.cost_usd);
-    let _ = events.send(budget.tick_event(Instant::now()));
+    let _ = events.send(budget.tick_event(sleeper.now()));
     if let BudgetOutcome::Warn {
         spent_usd,
         limit_usd,
@@ -455,6 +460,7 @@ mod tests {
     use stella_protocol::{BudgetMode, CompletionMessage, CompletionRequestRef, CompletionUsage};
 
     use super::*;
+    use std::time::Instant;
 
     struct NoopSleeper;
 
@@ -463,6 +469,10 @@ mod tests {
         async fn sleep(&self, _duration_ms: u64) {}
 
         // The floor: a test that asserts on retry timing wants no spread in it.
+        fn now(&self) -> std::time::Instant {
+            std::time::Instant::now()
+        }
+
         fn jitter(&self, _upper: u64) -> u64 {
             0
         }
@@ -984,6 +994,10 @@ mod tests {
 
         // The floor: the timeout under test is placed against the exact
         // backoff, so the draw must not move it.
+        fn now(&self) -> std::time::Instant {
+            std::time::Instant::now()
+        }
+
         fn jitter(&self, _upper: u64) -> u64 {
             0
         }
@@ -1156,7 +1170,7 @@ mod tests {
             },
             &mut budget,
             &EventSender::new(tx),
-            &NoopSleeper,
+            &TokioSleeper,
         )
         .await
         .expect("the trailing gap must not abandon a call that was actively answering");

crates/stella-core/src/bus.rs

@@ -61,15 +61,16 @@
 //! event comes from the [`Clock`] the host hands [`HookBus::new`], never
 //! from `SystemTime` — a hook script reads that stamp on the far side of a
 //! process boundary, so the host's wall clock is the right one and the
-//! host owns it. Observer dispatch still reads a monotonic `Instant` to
-//! enforce the per-handler latency budget (#459); `make core-no-io` counts
-//! that read and lets it go no higher.
+//! host owns it. Observer dispatch times each handler off that same clock
+//! to enforce the per-handler latency budget (#459), so a wall-clock step
+//! can misjudge one dispatch; quarantine needs three in a row, which one
+//! step cannot supply.
 
 use std::collections::VecDeque;
 use std::panic::{AssertUnwindSafe, catch_unwind};
 use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex, Weak};
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
@@ -645,9 +646,9 @@ impl HookBus {
             if quarantined.load(Ordering::Relaxed) {
                 continue;
             }
-            let started = Instant::now();
+            let started = self.inner.clock.now_ms();
             let outcome = catch_unwind(AssertUnwindSafe(|| handler(event)));
-            let elapsed = started.elapsed();
+            let elapsed = Duration::from_millis(self.inner.clock.now_ms().saturating_sub(started));
             match outcome {
                 Ok(Ok(())) => {}
                 Ok(Err(message)) => self.report_failure(&pattern, event, message),
@@ -1711,18 +1712,31 @@ mod tests {
     /// every later event, so it can't keep stalling the emitting (tool) thread.
     /// Positive direction only (a tiny budget + a handler that sleeps ~10x past
     /// it), to stay non-flaky under CI load.
+    /// A [`Clock`] the test advances by hand.
+    struct SteppingClock(Arc<AtomicU64>);
+
+    impl Clock for SteppingClock {
+        fn now_ms(&self) -> u64 {
+            self.0.load(Ordering::Relaxed)
+        }
+    }
+
     #[test]
     fn a_persistently_slow_observer_is_quarantined_then_skipped() {
+        // The bus reads its clock before and after each dispatch, so a
+        // handler that moves the clock is a slow handler — with no real
+        // sleep to flake under CI load.
+        let clock = Arc::new(AtomicU64::new(0));
         let bus = HookBus::with_slow_observer_budget(
             "s",
-            crate::ports::FixedClock(0),
+            SteppingClock(clock.clone()),
             Duration::from_millis(10),
         );
         let slow_calls = Arc::new(AtomicU32::new(0));
         let sc = slow_calls.clone();
         bus.on("file.created", move |_event| {
             sc.fetch_add(1, Ordering::Relaxed);
-            std::thread::sleep(Duration::from_millis(100));
+            clock.fetch_add(100, Ordering::Relaxed);
             Ok(())
         })
         .detach();