From a37189c9213bc89aad7bce888f0453d716b05933 Mon Sep 17 00:00:00 2001 From: Mac Anderson Date: Wed, 9 Sep 2026 21:07:53 -0700 Subject: [PATCH] fix(stella-core): take retry jitter and hook stamps through ports, and guard the engine against I/O The retry ladder drew its jitter from the OS entropy pool through `rand::rng()`, and the hook bus stamped every event from `SystemTime::now()`, while both files' headers said the crate reads nothing directly. `Sleeper` gains a required `jitter` draw, `HookBus::new` takes a `Clock` counting from the Unix epoch, and `rand` leaves stella-core's shipping dependencies for its dev-dependencies. `make core-no-io` (scripts/check-core-no-io.py) now reads the crate's shipping source and manifest on every gate run: a floor over every I/O, wall-clock, sleep and entropy surface, and a down-only count of the `Instant::now()` reads the deadline arithmetic still makes. --- .github/workflows/ci.yml | 9 + .github/workflows/guard-self-tests.yml | 4 + AGENTS.md | 24 ++ CONTRIBUTING.md | 1 + Cargo.lock | 1 + Makefile | 13 + crates/stella-cli/Cargo.toml | 3 + crates/stella-cli/src/agent/tool_stack.rs | 2 +- crates/stella-cli/src/rules.rs | 3 +- crates/stella-cli/src/runtime.rs | 10 +- crates/stella-core/Cargo.toml | 5 +- crates/stella-core/README.md | 6 +- crates/stella-core/src/accounted_call.rs | 11 + crates/stella-core/src/bus.rs | 101 +++-- crates/stella-core/src/bus/attribution.rs | 8 +- crates/stella-core/src/bus/policy_bridge.rs | 6 +- crates/stella-core/src/driver/capabilities.rs | 5 +- crates/stella-core/src/driver/dispatch.rs | 5 + crates/stella-core/src/driver/drive.rs | 5 + crates/stella-core/src/driver/restore.rs | 5 + crates/stella-core/src/driver/tests.rs | 5 + .../src/driver/tests/audit_fixes.rs | 5 + .../src/driver/tests/context_overflow.rs | 5 + .../src/driver/tests/deadline_notice.rs | 5 + .../src/driver/tests/lifecycle_bus.rs | 7 +- .../src/driver/tests/model_fallback.rs | 5 + .../src/driver/tests/output_budget.rs | 5 + .../src/driver/tests/provider_outcomes.rs | 5 + .../stella-core/src/driver/tests/requery.rs | 5 + .../src/driver/tests/user_hooks/verdicts.rs | 2 +- crates/stella-core/src/goal.rs | 5 + crates/stella-core/src/ports.rs | 23 +- crates/stella-core/src/retry.rs | 159 +++++-- crates/stella-core/src/subagent/tests.rs | 5 + .../stella-core/src/subagent/tests/seams.rs | 4 +- .../tests/engine_emits_no_stage.rs | 5 + .../stella-core/tests/hard_drop_write_back.rs | 5 + .../tests/hook_bus_stamps_from_the_clock.rs | 32 ++ crates/stella-core/tests/parallel_dispatch.rs | 5 + crates/stella-core/tests/spend_gate.rs | 5 + crates/stella-core/tests/tool_wall_clock.rs | 5 + crates/stella-engine/src/tests.rs | 5 + crates/stella-engine/tests/embedding.rs | 5 + crates/stella-serve/src/extensions.rs | 2 +- crates/stella-serve/src/remote.rs | 32 +- crates/stella-serve/src/remote/tests.rs | 8 +- crates/stella-serve/src/session.rs | 3 +- crates/stella-tools/src/ctx.rs | 2 +- .../src/custom/tests/claims_and_plugins.rs | 4 +- crates/stella-tools/src/gated.rs | 2 +- crates/stella-tools/src/hook_bridge.rs | 6 +- crates/stella-tools/src/registry/approval.rs | 2 +- crates/stella-tools/src/subagent/tests.rs | 2 +- crates/stella-tools/tests/approval_witness.rs | 2 +- .../tests/ask_question_a2a_witness.rs | 2 +- scripts/check-core-no-io.py | 389 ++++++++++++++++++ scripts/check-gate-parity.sh | 1 + scripts/core-no-io-baseline.txt | 18 + scripts/test-core-no-io.sh | 222 ++++++++++ 59 files changed, 1118 insertions(+), 118 deletions(-) create mode 100644 crates/stella-core/tests/hook_bus_stamps_from_the_clock.rs create mode 100755 scripts/check-core-no-io.py create mode 100644 scripts/core-no-io-baseline.txt create mode 100755 scripts/test-core-no-io.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9b2997e190..82ab4b8a32 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -331,6 +331,15 @@ jobs: if: ${{ !cancelled() && steps.checkout.outcome == 'success' }} run: python3 ./scripts/check-core-reachability.py + # Invariant 2 says the engine does no I/O, and until this step nothing + # 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. + - 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 + # Also toolchain-free. AGENTS.md requires that anything noticed and not # fixed becomes an issue written as a handoff; nothing checked it, and an # unenforced standard reads as one the codebase is meeting. The decidable diff --git a/.github/workflows/guard-self-tests.yml b/.github/workflows/guard-self-tests.yml index 5eed9b8c7f..9369e8001f 100644 --- a/.github/workflows/guard-self-tests.yml +++ b/.github/workflows/guard-self-tests.yml @@ -355,6 +355,10 @@ jobs: if: ${{ !cancelled() }} run: ./scripts/test-core-reachability.sh + - name: the core-no-io floor, manifest check and clock ratchet + if: ${{ !cancelled() }} + run: ./scripts/test-core-no-io.sh + - name: the unverified-main detector, and that it fails open if: ${{ !cancelled() }} run: ./scripts/test-main-verified.sh diff --git a/AGENTS.md b/AGENTS.md index 43adb3fc8d..ca717bc7a5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -129,6 +129,9 @@ make gate # = no-scratch + no-secrets + design-refs # + stat-portability + module-reachability # + 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) # + typed-errors # + tool-error-class (#3167 unclassified-ToolOutput::error ratchet) # + dead-code-allows @@ -938,6 +941,27 @@ Append; do not renumber. `scripts/check-invariants.sh` enforces both halves. property-testable. Anything that spawns processes, reads files, or hits the network belongs in `stella-tools`, `stella-model`, `stella-cli`, or `stella-store` — injected as a port/trait, not called directly. + + Enforced by `scripts/check-core-no-io.py` (`make core-no-io`), which + reads the crate's shipping source — `#[cfg(test)]` bodies, `tests/` + 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 + 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 + now — `retry::Sleeper::jitter` and the `Clock` handed to + `bus::HookBus::new`. 3. **Zero telemetry egress by default.** Community/default Stella sends no telemetry anywhere; model-provider traffic remains the normal network exception selected by the user. The sole additional egress is an explicitly diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 305cbd8331..82de023ffa 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -98,6 +98,7 @@ python3 ./scripts/check-retired-model-keys.py ./scripts/check-stat-portability.sh python3 ./scripts/check-module-reachability.py python3 ./scripts/check-core-reachability.py +python3 ./scripts/check-core-no-io.py python3 ./scripts/check-typed-errors.py python3 ./scripts/check-tool-error-class.py python3 ./scripts/check-dead-code-allows.py diff --git a/Cargo.lock b/Cargo.lock index 9e84c645b6..2821389a13 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3214,6 +3214,7 @@ dependencies = [ "hmac", "libc", "proptest", + "rand 0.10.2", "ratatui", "reqwest", "rpassword", diff --git a/Makefile b/Makefile index 7357ed7b38..afab21caae 100644 --- a/Makefile +++ b/Makefile @@ -45,6 +45,7 @@ GATE_GUARDS_FAST := no-scratch no-secrets design-refs action-pins cargo-install- release-wiring left-behind \ retired-model-keys \ stat-portability module-reachability core-reachability \ + core-no-io \ typed-errors \ tool-error-class \ dead-code-allows measured-constants diagnostic-codes consumer-sites \ @@ -927,6 +928,18 @@ core-reachability-update: ## Shrink the core-reachability baseline after an evic core-reachability-test: ## Test the core-reachability walker (hermetic; not part of `gate`) ./scripts/test-core-reachability.sh +.PHONY: core-no-io +core-no-io: ## Assert no shipping stella-core source names an I/O surface, and no new Instant::now() read (down-only) + @python3 ./scripts/check-core-no-io.py + +.PHONY: core-no-io-update +core-no-io-update: ## Shrink the core-no-io clock-read baseline after removing a read (refuses to grow) + @python3 ./scripts/check-core-no-io.py --update + +.PHONY: core-no-io-test +core-no-io-test: ## Test the core-no-io guard (hermetic; not part of `gate`) + ./scripts/test-core-no-io.sh + .PHONY: god-files god-files: ## Assert AGENTS.md and the crate READMEs name the baselined god files (#1435) @./scripts/check-god-files.sh diff --git a/crates/stella-cli/Cargo.toml b/crates/stella-cli/Cargo.toml index b17fb6dec0..e40d57942f 100644 --- a/crates/stella-cli/Cargo.toml +++ b/crates/stella-cli/Cargo.toml @@ -91,6 +91,9 @@ serde.workspace = true serde_json.workspace = true sha2.workspace = true hmac.workspace = true +# The entropy behind retry jitter. `stella-core` takes it through +# `retry::Sleeper::jitter` and links no entropy source of its own. +rand.workspace = true reqwest.workspace = true futures-util.workspace = true tokio = { workspace = true, features = ["sync", "signal", "net"] } diff --git a/crates/stella-cli/src/agent/tool_stack.rs b/crates/stella-cli/src/agent/tool_stack.rs index 0a5cb115cd..35cb5632c7 100644 --- a/crates/stella-cli/src/agent/tool_stack.rs +++ b/crates/stella-cli/src/agent/tool_stack.rs @@ -514,7 +514,7 @@ mod tests { let registry = Arc::new(stella_tools::registry::ToolRegistry::new( dir.path().to_path_buf(), )); - let bus = HookBus::new("gate-2793"); + let bus = HookBus::new("gate-2793", stella_core::ports::FixedClock(0)); bus.on_blocking(hook_names::TOOL_CALL_REQUESTED, |event| { match event.payload["tool"].as_str() { Some("mcp__vendor__deploy") | Some("my_tool") => { diff --git a/crates/stella-cli/src/rules.rs b/crates/stella-cli/src/rules.rs index c2998d69a2..96a91ec8d3 100644 --- a/crates/stella-cli/src/rules.rs +++ b/crates/stella-cli/src/rules.rs @@ -50,6 +50,7 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; +use crate::runtime::WallClock; use stella_core::bus::{HookBus, HookDecision, names as hook_names}; use stella_learn::rules::{ LoadRulesOptions, ProposedAction, Rule, RuleFile, RuleSource, evaluate_guards, load_rules, @@ -587,7 +588,7 @@ pub(crate) fn attach_rule_guards(registry: &ToolRegistry, rules: &ResolvedRules) return; } let rules = Arc::clone(&rules.rules); - let bus = HookBus::new(format!("rules-{}", std::process::id())); + let bus = HookBus::new(format!("rules-{}", std::process::id()), WallClock); bus.on_blocking(hook_names::TOOL_CALL_REQUESTED, move |event| { let tool = canonical_tool(event.payload["tool"].as_str().unwrap_or_default()); let input = &event.payload["input"]; diff --git a/crates/stella-cli/src/runtime.rs b/crates/stella-cli/src/runtime.rs index 645a8b0ccd..16b4d74eab 100644 --- a/crates/stella-cli/src/runtime.rs +++ b/crates/stella-cli/src/runtime.rs @@ -5,6 +5,7 @@ //! keeps `stella-core` free of production `tokio::time` calls. use async_trait::async_trait; +use rand::RngExt; use stella_core::ports::Clock; use stella_core::retry::Sleeper; @@ -55,7 +56,10 @@ impl Clock for WallClock { } } -/// The production [`Sleeper`]: a thin wrapper over `tokio::time::sleep`. +/// The production [`Sleeper`]: `tokio::time::sleep` for the wait, and the +/// OS entropy pool for the jitter that spreads concurrent retriers across +/// the backoff window. Both live here so `stella-core` links neither a +/// timer nor an entropy source. #[derive(Debug, Default, Clone, Copy)] pub struct TokioSleeper; @@ -64,6 +68,10 @@ impl Sleeper for TokioSleeper { async fn sleep(&self, duration_ms: u64) { tokio::time::sleep(std::time::Duration::from_millis(duration_ms)).await; } + + fn jitter(&self, upper: u64) -> u64 { + rand::rng().random_range(0..=upper) + } } /// The budget guard for a one-shot invocation, with its wall-clock task diff --git a/crates/stella-core/Cargo.toml b/crates/stella-core/Cargo.toml index 4afa690653..961339e65e 100644 --- a/crates/stella-core/Cargo.toml +++ b/crates/stella-core/Cargo.toml @@ -17,9 +17,12 @@ thiserror = { workspace = true } tokio = { workspace = true, features = ["sync", "time"] } async-trait = { workspace = true } futures-util = { workspace = true } -rand = { workspace = true } sha2 = { workspace = true } [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } +# Seeds the retry tests' jitter double. Shipping code draws nothing: the +# entropy behind backoff jitter is a port (`retry::Sleeper::jitter`), and +# `make core-no-io` refuses this crate an entropy source. +rand = { workspace = true } proptest = { workspace = true } diff --git a/crates/stella-core/README.md b/crates/stella-core/README.md index a932d4feb2..fb9657a29b 100644 --- a/crates/stella-core/README.md +++ b/crates/stella-core/README.md @@ -12,7 +12,11 @@ filesystem, never spawns a process, never opens a socket. Anything needing the outside world is a trait the caller implements: `ToolExecutor`, `Clock`, `TurnGate`, `TurnSteering` ([`src/ports.rs`](src/ports.rs)), `Sleeper` ([`src/retry.rs`](src/retry.rs)), `HookRunner` ([`src/hooks.rs`](src/hooks.rs)) -— plus `Provider` from `stella-protocol`. +— 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. 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 diff --git a/crates/stella-core/src/accounted_call.rs b/crates/stella-core/src/accounted_call.rs index 5355e48036..089bd32aa1 100644 --- a/crates/stella-core/src/accounted_call.rs +++ b/crates/stella-core/src/accounted_call.rs @@ -461,6 +461,11 @@ mod tests { #[async_trait] impl Sleeper for NoopSleeper { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } struct RetryThenSuccess { @@ -976,6 +981,12 @@ mod tests { async fn sleep(&self, duration_ms: u64) { tokio::time::sleep(Duration::from_millis(duration_ms)).await; } + + // The floor: the timeout under test is placed against the exact + // backoff, so the draw must not move it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } struct AlwaysRetryable; diff --git a/crates/stella-core/src/bus.rs b/crates/stella-core/src/bus.rs index 0e9e818024..696ce9f5c5 100644 --- a/crates/stella-core/src/bus.rs +++ b/crates/stella-core/src/bus.rs @@ -57,22 +57,26 @@ //! # No I/O in this module //! //! The bus is plain synchronous logic over owned data: registration lists, -//! an atomic sequence counter, and inline dispatch. Timestamping reads the -//! system clock (a pure computation over `SystemTime`), and observer dispatch -//! reads a monotonic `Instant` to enforce the per-handler latency budget -//! (#459) — both are clock reads, not I/O in the architectural sense (no -//! filesystem, no network, no processes). +//! an atomic sequence counter, and inline dispatch. The stamp on every +//! 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. 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, SystemTime, UNIX_EPOCH}; +use std::time::{Duration, Instant}; use serde::{Deserialize, Serialize}; use serde_json::Value; use stella_protocol::Denial; +use crate::ports::Clock; + // The event-name catalog, split to a sibling file (the `driver/settlement.rs` // pattern) so this file stays under its file-size ceiling (#1857). pub mod names; @@ -312,6 +316,9 @@ struct BusInner { /// The per-observer-dispatch latency budget (#459). Immutable for the /// bus's lifetime; defaults to [`SLOW_OBSERVER_BUDGET`]. slow_observer_budget: Duration, + /// Where every event's `timestamp` comes from. Unix-epoch milliseconds, + /// by the contract on [`HookBus::new`]. + clock: Box, } /// The session-scoped hook bus. Cheap to clone (shared inner); every clone @@ -322,8 +329,15 @@ pub struct HookBus { } impl HookBus { - pub fn new(session_id: impl Into) -> Self { - Self::with_slow_observer_budget(session_id, SLOW_OBSERVER_BUDGET) + /// A bus for one session, stamping its events from `clock`. + /// + /// `clock` must count milliseconds from the Unix epoch. The stamp it + /// produces leaves the process — a hook script parses it, an + /// observatory orders by it — so a clock counting from process start + /// would write a number that means nothing to either reader. A test + /// passes [`crate::ports::FixedClock`] and asserts on the exact stamp. + pub fn new(session_id: impl Into, clock: impl Clock + 'static) -> Self { + Self::with_slow_observer_budget(session_id, clock, SLOW_OBSERVER_BUDGET) } /// Like [`HookBus::new`] but with an explicit observer latency budget @@ -331,6 +345,7 @@ impl HookBus { /// tiny budget to exercise quarantine without real-time sleeps. pub fn with_slow_observer_budget( session_id: impl Into, + clock: impl Clock + 'static, slow_observer_budget: Duration, ) -> Self { Self { @@ -343,10 +358,16 @@ impl HookBus { context: Mutex::new(AmbientContext::default()), recent_failures: Mutex::new(VecDeque::new()), slow_observer_budget, + clock: Box::new(clock), }), } } + /// The clock's reading now, as the ISO 8601 stamp events carry. + fn stamp(&self) -> String { + iso8601_utc_millis(i64::try_from(self.inner.clock.now_ms()).unwrap_or(i64::MAX)) + } + pub fn session_id(&self) -> &str { &self.inner.session_id } @@ -581,7 +602,7 @@ impl HookBus { HookEvent { id: format!("evt_{}_{sequence}", self.inner.session_id), name: draft.name, - timestamp: now_iso8601_utc(), + timestamp: self.stamp(), session_id: self.inner.session_id.clone(), turn_id: draft.turn_id.or(ambient_turn), agent_id: draft.agent_id.or(ambient_agent), @@ -677,7 +698,7 @@ impl HookBus { pattern: pattern.to_string(), event_name: event.name.clone(), message: message.clone(), - timestamp: now_iso8601_utc(), + timestamp: self.stamp(), }); } self.emit(HookEventDraft { @@ -711,7 +732,7 @@ impl HookBus { pattern: pattern.to_string(), event_name: event.name.clone(), message: message.clone(), - timestamp: now_iso8601_utc(), + timestamp: self.stamp(), }); } if event.name != names::EXTENSION_ERROR { @@ -971,14 +992,6 @@ pub fn is_sensitive_path(path: &str) -> bool { // Timestamps — ISO 8601 UTC without a date-time dependency -fn now_iso8601_utc() -> String { - let millis = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.as_millis() as i64) - .unwrap_or(0); - iso8601_utc_millis(millis) -} - /// Format Unix milliseconds as `YYYY-MM-DDThh:mm:ss.mmmZ`. Fixed-width and /// zero-padded, so lexicographic order equals time order (same property as /// `stella-context`'s clock, extended to millisecond precision). @@ -1020,7 +1033,7 @@ mod tests { /// Bus + a `Vec` capturing every event a `"*"` observer sees. fn observed_bus(session: &str) -> (HookBus, Arc>>) { - let bus = HookBus::new(session); + let bus = HookBus::new(session, crate::ports::FixedClock(0)); let seen = Arc::new(Mutex::new(Vec::new())); let sink = seen.clone(); bus.on("*", move |event| { @@ -1044,7 +1057,7 @@ mod tests { #[test] fn bridge_maps_the_audit_plane_onto_policy_decision_events() { use stella_protocol::PolicyKind; - let bus = HookBus::new("bridge-test"); + let bus = HookBus::new("bridge-test", crate::ports::FixedClock(0)); bus.on_blocking(names::TOOL_CALL_REQUESTED, |_| { HookDecision::Deny("not on my watch".into()) }) @@ -1107,7 +1120,7 @@ mod tests { #[test] fn dropping_the_bridge_subscription_stops_the_flow() { - let bus = HookBus::new("bridge-drop"); + let bus = HookBus::new("bridge-drop", crate::ports::FixedClock(0)); let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); let bridge = bridge_policy_plane(&bus, crate::event_sender::EventSender::new(tx)); drop(bridge); // un-detached subscription unsubscribes on drop @@ -1181,7 +1194,7 @@ mod tests { #[test] fn exact_and_wildcard_subscriptions_each_receive_matching_events() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); let exact = Arc::new(AtomicUsize::new(0)); let ns = Arc::new(AtomicUsize::new(0)); let all = Arc::new(AtomicUsize::new(0)); @@ -1212,7 +1225,7 @@ mod tests { #[test] fn envelope_serializes_to_the_documented_shape() { - let bus = HookBus::new("sess-1"); + let bus = HookBus::new("sess-1", crate::ports::FixedClock(0)); bus.set_turn(Some("turn-9".into())); let event = bus.emit( HookEventDraft::new(names::FILE_READ, serde_json::json!({"path": "src/a.rs"})) @@ -1239,7 +1252,7 @@ mod tests { #[test] fn absent_turn_and_agent_ids_are_omitted_from_the_wire_shape() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); let event = bus.emit_named(names::FILE_READ, Value::Null); let json = serde_json::to_string(&event).unwrap(); assert!(!json.contains("turn_id")); @@ -1301,8 +1314,8 @@ mod tests { #[test] fn sequences_are_per_session_not_global() { - let a = HookBus::new("a"); - let b = HookBus::new("b"); + let a = HookBus::new("a", crate::ports::FixedClock(0)); + let b = HookBus::new("b", crate::ports::FixedClock(0)); assert_eq!(a.emit_named(names::FILE_READ, Value::Null).sequence, 1); assert_eq!(a.emit_named(names::FILE_READ, Value::Null).sequence, 2); assert_eq!(b.emit_named(names::FILE_READ, Value::Null).sequence, 1); @@ -1338,7 +1351,7 @@ mod tests { #[test] fn failing_observers_never_stop_delivery_and_surface_extension_error() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); let delivered = Arc::new(AtomicUsize::new(0)); let errors = Arc::new(Mutex::new(Vec::new())); @@ -1378,7 +1391,7 @@ mod tests { #[test] fn a_broken_extension_error_handler_does_not_recurse() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); // This handler fails on EVERY event — including extension.error. // Without the recursion guard this loops forever. bus.on("*", |_| Err("always broken".to_string())).detach(); @@ -1390,7 +1403,7 @@ mod tests { #[test] fn failure_log_is_bounded() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); bus.on(names::FILE_READ, |_| Err("broken".to_string())) .detach(); for _ in 0..(MAX_RECENT_FAILURES + 20) { @@ -1486,7 +1499,7 @@ mod tests { #[test] fn blocking_modify_folds_payload_and_later_handlers_see_it() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); let _a = bus.on_blocking(names::COMMAND_STARTED, |event| { let mut payload = event.payload.clone(); payload["command"] = Value::String("ls -la".into()); @@ -1516,7 +1529,7 @@ mod tests { #[test] fn blocking_handlers_run_in_registration_order() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); let order = Arc::new(Mutex::new(Vec::new())); for tag in ["first", "second", "third"] { let order = order.clone(); @@ -1532,7 +1545,7 @@ mod tests { #[test] fn a_panicking_policy_handler_fails_closed() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); let _a = bus.on_blocking(names::FILE_DELETED, |_| panic!("broken extension")); let outcome = bus.emit_blocking(HookEventDraft::new( names::FILE_DELETED, @@ -1549,7 +1562,7 @@ mod tests { #[test] fn blocking_events_consume_session_sequence_numbers_too() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); let first = bus.emit_named(names::FILE_READ, Value::Null); let outcome = bus.emit_blocking(HookEventDraft::new(names::FILE_CREATED, Value::Null)); let last = bus.emit_named(names::FILE_READ, Value::Null); @@ -1563,7 +1576,7 @@ mod tests { #[test] fn unsubscribe_stops_delivery() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); let count = Arc::new(AtomicUsize::new(0)); let sink = count.clone(); let sub = bus.on(names::FILE_READ, move |_| { @@ -1578,7 +1591,7 @@ mod tests { #[test] fn dropping_the_subscription_unsubscribes() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); let count = Arc::new(AtomicUsize::new(0)); { let sink = count.clone(); @@ -1594,7 +1607,7 @@ mod tests { #[test] fn detach_keeps_the_handler_for_the_bus_lifetime() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); let count = Arc::new(AtomicUsize::new(0)); let sink = count.clone(); bus.on("*", move |_| { @@ -1609,7 +1622,7 @@ mod tests { #[test] fn off_removes_blocking_handlers_and_outlives_the_bus_safely() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); let sub = bus.on_blocking(names::FILE_CREATED, |_| HookDecision::Deny("no".into())); assert!( !bus.emit_blocking(HookEventDraft::new(names::FILE_CREATED, Value::Null)) @@ -1623,7 +1636,7 @@ mod tests { // A subscription outliving its bus unsubscribes into nothing. let orphan = { - let short_lived = HookBus::new("gone"); + let short_lived = HookBus::new("gone", crate::ports::FixedClock(0)); short_lived.on("*", |_| Ok(())) }; orphan.unsubscribe(); // must not panic @@ -1652,7 +1665,7 @@ mod tests { #[tokio::test] async fn forward_to_bridges_events_into_a_bounded_channel() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); let (tx, mut rx) = tokio::sync::mpsc::channel(8); let (fwd, dropped) = forward_to(tx); bus.on("file.*", fwd).detach(); @@ -1672,7 +1685,7 @@ mod tests { /// outcome, not a handler failure — so no `extension.error` storm. #[tokio::test] async fn forward_to_drops_newest_when_the_bounded_channel_is_full() { - let bus = HookBus::new("s"); + let bus = HookBus::new("s", crate::ports::FixedClock(0)); // `_rx` is never drained but stays alive, so the channel fills (Full), // it is not closed (which would be a real Err). let (tx, _rx) = tokio::sync::mpsc::channel(2); @@ -1700,7 +1713,11 @@ mod tests { /// it), to stay non-flaky under CI load. #[test] fn a_persistently_slow_observer_is_quarantined_then_skipped() { - let bus = HookBus::with_slow_observer_budget("s", Duration::from_millis(10)); + let bus = HookBus::with_slow_observer_budget( + "s", + crate::ports::FixedClock(0), + Duration::from_millis(10), + ); let slow_calls = Arc::new(AtomicU32::new(0)); let sc = slow_calls.clone(); bus.on("file.created", move |_event| { diff --git a/crates/stella-core/src/bus/attribution.rs b/crates/stella-core/src/bus/attribution.rs index 94af599f82..f3a4daace5 100644 --- a/crates/stella-core/src/bus/attribution.rs +++ b/crates/stella-core/src/bus/attribution.rs @@ -132,7 +132,7 @@ mod tests { /// that had finished. #[test] fn overlapping_siblings_do_not_corrupt_the_ambient_agent() { - let bus = HookBus::new("session-1653"); + let bus = HookBus::new("session-1653", crate::ports::FixedClock(0)); assert_eq!(bus.current_agent(), None, "nothing entered yet"); let a = bus.push_agent("a".to_string()); @@ -158,7 +158,7 @@ mod tests { /// Strict nesting — the only case the slot got right — still behaves. #[test] fn nested_scopes_restore_their_parent() { - let bus = HookBus::new("session-1653"); + let bus = HookBus::new("session-1653", crate::ports::FixedClock(0)); let parent = bus.push_agent("parent".to_string()); let child = bus.push_agent("child".to_string()); assert_eq!(bus.current_agent(), Some("child".to_string())); @@ -180,7 +180,7 @@ mod tests { /// else's entry — which a pop-the-top implementation would do. #[test] fn dropping_a_scope_twice_does_not_disturb_a_live_sibling() { - let bus = HookBus::new("session-1653"); + let bus = HookBus::new("session-1653", crate::ports::FixedClock(0)); let a = bus.push_agent("a".to_string()); let b = bus.push_agent("b".to_string()); @@ -199,7 +199,7 @@ mod tests { /// a slot, and the one a fan-out of siblings actually produces. #[test] fn entry_order_exits_leave_the_innermost_survivor_attributed() { - let bus = HookBus::new("session-1653"); + let bus = HookBus::new("session-1653", crate::ports::FixedClock(0)); let a = bus.push_agent("a".to_string()); let b = bus.push_agent("b".to_string()); let c = bus.push_agent("c".to_string()); diff --git a/crates/stella-core/src/bus/policy_bridge.rs b/crates/stella-core/src/bus/policy_bridge.rs index 9032326510..f98e249c4d 100644 --- a/crates/stella-core/src/bus/policy_bridge.rs +++ b/crates/stella-core/src/bus/policy_bridge.rs @@ -125,7 +125,7 @@ mod tests { /// rows per ask, and a count of asks read double. #[test] fn one_ask_is_one_row_however_many_emissions_it_makes() { - let bus = HookBus::new("approval-pair"); + let bus = HookBus::new("approval-pair", crate::ports::FixedClock(0)); bus.on_blocking(names::TOOL_CALL_REQUESTED, |_| { HookDecision::RequireApproval { reason: "destructive".into(), @@ -161,7 +161,7 @@ mod tests { /// a new row, not the other half of the last one. #[test] fn a_resolved_ask_lets_the_next_one_through() { - let bus = HookBus::new("approval-reopen"); + let bus = HookBus::new("approval-reopen", crate::ports::FixedClock(0)); let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); let _bridge = bridge_policy_plane(&bus, crate::event_sender::EventSender::new(tx)); @@ -189,7 +189,7 @@ mod tests { /// is what keeps them apart. #[test] fn two_gates_are_two_asks() { - let bus = HookBus::new("approval-two-gates"); + let bus = HookBus::new("approval-two-gates", crate::ports::FixedClock(0)); let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); let _bridge = bridge_policy_plane(&bus, crate::event_sender::EventSender::new(tx)); diff --git a/crates/stella-core/src/driver/capabilities.rs b/crates/stella-core/src/driver/capabilities.rs index 7805cca883..679e3e0fd3 100644 --- a/crates/stella-core/src/driver/capabilities.rs +++ b/crates/stella-core/src/driver/capabilities.rs @@ -447,7 +447,10 @@ mod tests { /// owned struct actually arrives. #[test] fn as_borrowed_carries_every_slot_it_was_given() { - let bus = Arc::new(crate::bus::HookBus::new("owned-caps-test")); + let bus = Arc::new(crate::bus::HookBus::new( + "owned-caps-test", + crate::ports::FixedClock(0), + )); let mut owned = OwnedTurnCapabilities::none(); owned.bus = Some(Arc::clone(&bus)); diff --git a/crates/stella-core/src/driver/dispatch.rs b/crates/stella-core/src/driver/dispatch.rs index 264efc9782..c9c694e639 100644 --- a/crates/stella-core/src/driver/dispatch.rs +++ b/crates/stella-core/src/driver/dispatch.rs @@ -453,6 +453,11 @@ mod tests { #[async_trait] impl Sleeper for NoopSleeper { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// Fires once the executor's flag is up — the shape of the pipeline's diff --git a/crates/stella-core/src/driver/drive.rs b/crates/stella-core/src/driver/drive.rs index 257f0f83b5..a25de75593 100644 --- a/crates/stella-core/src/driver/drive.rs +++ b/crates/stella-core/src/driver/drive.rs @@ -283,6 +283,11 @@ mod tests { #[async_trait] impl Sleeper for NoopSleeper { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// Armed from the start, so the first committed boundary ends the turn. diff --git a/crates/stella-core/src/driver/restore.rs b/crates/stella-core/src/driver/restore.rs index 8e605d7899..7abece0543 100644 --- a/crates/stella-core/src/driver/restore.rs +++ b/crates/stella-core/src/driver/restore.rs @@ -583,6 +583,11 @@ mod tests { #[async_trait] impl Sleeper for NoSleep { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// Always answers "SUMMARY" — the summarizer path under test is the diff --git a/crates/stella-core/src/driver/tests.rs b/crates/stella-core/src/driver/tests.rs index f38b0bc22f..a65fa240e0 100644 --- a/crates/stella-core/src/driver/tests.rs +++ b/crates/stella-core/src/driver/tests.rs @@ -19,6 +19,11 @@ struct NoopSleeper; #[async_trait] impl Sleeper for NoopSleeper { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// A `ToolExecutor` that always succeeds and counts real invocations — the diff --git a/crates/stella-core/src/driver/tests/audit_fixes.rs b/crates/stella-core/src/driver/tests/audit_fixes.rs index 06915e81ca..61f528c939 100644 --- a/crates/stella-core/src/driver/tests/audit_fixes.rs +++ b/crates/stella-core/src/driver/tests/audit_fixes.rs @@ -394,6 +394,11 @@ impl crate::retry::Sleeper for HangingSleeper { self.sleeping.notify_one(); std::future::pending().await } + + // The floor: this double never wakes, so the draw decides nothing. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// The other half of F9's contract: a hard cancel landing in a backoff sleep diff --git a/crates/stella-core/src/driver/tests/context_overflow.rs b/crates/stella-core/src/driver/tests/context_overflow.rs index d6ff7d935a..0143ade702 100644 --- a/crates/stella-core/src/driver/tests/context_overflow.rs +++ b/crates/stella-core/src/driver/tests/context_overflow.rs @@ -106,6 +106,11 @@ struct NoSleep; #[async_trait::async_trait] impl crate::retry::Sleeper for NoSleep { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// Completes every call — the wider-window replacement the spent ladder diff --git a/crates/stella-core/src/driver/tests/deadline_notice.rs b/crates/stella-core/src/driver/tests/deadline_notice.rs index c70ebfed07..8a6580b37d 100644 --- a/crates/stella-core/src/driver/tests/deadline_notice.rs +++ b/crates/stella-core/src/driver/tests/deadline_notice.rs @@ -64,6 +64,11 @@ struct NoSleep; #[async_trait::async_trait] impl crate::retry::Sleeper for NoSleep { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } struct NoTools; diff --git a/crates/stella-core/src/driver/tests/lifecycle_bus.rs b/crates/stella-core/src/driver/tests/lifecycle_bus.rs index 5e503a31c5..c2d9c44447 100644 --- a/crates/stella-core/src/driver/tests/lifecycle_bus.rs +++ b/crates/stella-core/src/driver/tests/lifecycle_bus.rs @@ -29,7 +29,7 @@ struct Recorder { impl Recorder { /// Register on a fresh bus and hand back both halves. fn attach() -> (HookBus, Arc) { - let bus = HookBus::new("lifecycle-test"); + let bus = HookBus::new("lifecycle-test", crate::ports::FixedClock(0)); let recorder = Arc::new(Recorder::default()); let sink = Arc::clone(&recorder); bus.on("*", move |event| { @@ -134,6 +134,11 @@ struct NoSleep; #[async_trait::async_trait] impl crate::retry::Sleeper for NoSleep { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } async fn run_one_turn(provider: &dyn Provider, bus: &HookBus) -> TurnOutcome { diff --git a/crates/stella-core/src/driver/tests/model_fallback.rs b/crates/stella-core/src/driver/tests/model_fallback.rs index c60b62442f..dd457a06f7 100644 --- a/crates/stella-core/src/driver/tests/model_fallback.rs +++ b/crates/stella-core/src/driver/tests/model_fallback.rs @@ -170,6 +170,11 @@ struct NoSleep; #[async_trait::async_trait] impl crate::retry::Sleeper for NoSleep { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// One real turn against `provider` with `resolver` attached (when given), diff --git a/crates/stella-core/src/driver/tests/output_budget.rs b/crates/stella-core/src/driver/tests/output_budget.rs index 8de97bde36..dd949d549b 100644 --- a/crates/stella-core/src/driver/tests/output_budget.rs +++ b/crates/stella-core/src/driver/tests/output_budget.rs @@ -158,6 +158,11 @@ struct NoSleep; #[async_trait::async_trait] impl crate::retry::Sleeper for NoSleep { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// One real turn against `provider` with a configured output ceiling, and diff --git a/crates/stella-core/src/driver/tests/provider_outcomes.rs b/crates/stella-core/src/driver/tests/provider_outcomes.rs index fdcc3e05cf..70dc4378dd 100644 --- a/crates/stella-core/src/driver/tests/provider_outcomes.rs +++ b/crates/stella-core/src/driver/tests/provider_outcomes.rs @@ -81,6 +81,11 @@ struct NoSleep; #[async_trait::async_trait] impl crate::retry::Sleeper for NoSleep { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// Cooldown timing is irrelevant here (nothing advances), so a frozen clock diff --git a/crates/stella-core/src/driver/tests/requery.rs b/crates/stella-core/src/driver/tests/requery.rs index d0c5ea7fda..2f1dd874ce 100644 --- a/crates/stella-core/src/driver/tests/requery.rs +++ b/crates/stella-core/src/driver/tests/requery.rs @@ -28,6 +28,11 @@ struct NoopSleeper; #[async_trait] impl Sleeper for NoopSleeper { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } struct OkTools; diff --git a/crates/stella-core/src/driver/tests/user_hooks/verdicts.rs b/crates/stella-core/src/driver/tests/user_hooks/verdicts.rs index 8d446205d3..57feb253f5 100644 --- a/crates/stella-core/src/driver/tests/user_hooks/verdicts.rs +++ b/crates/stella-core/src/driver/tests/user_hooks/verdicts.rs @@ -44,7 +44,7 @@ async fn a_structured_stop_denial_reaches_the_model_and_the_journal_intact() { let hooks: Hooks = serde_json::from_str(r#"{ "Stop": [ { "hooks": [{ "command": "verify" }] } ] }"#).unwrap(); - let bus = crate::bus::HookBus::new("denial-test"); + let bus = crate::bus::HookBus::new("denial-test", crate::ports::FixedClock(0)); let blocked: Arc>> = Arc::default(); let sink = Arc::clone(&blocked); bus.on(crate::bus::names::HOOK_STOP_BLOCKED, move |event| { diff --git a/crates/stella-core/src/goal.rs b/crates/stella-core/src/goal.rs index 9bb0da8b23..6941f786f9 100644 --- a/crates/stella-core/src/goal.rs +++ b/crates/stella-core/src/goal.rs @@ -561,6 +561,11 @@ mod tests { #[async_trait] impl Sleeper for NoSleep { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// A provider that returns a fixed sequence of results, then errors. diff --git a/crates/stella-core/src/ports.rs b/crates/stella-core/src/ports.rs index a72923c439..6f1a441766 100644 --- a/crates/stella-core/src/ports.rs +++ b/crates/stella-core/src/ports.rs @@ -620,11 +620,32 @@ impl ToolExecutor for GrantedTools<'_> { /// here — the production implementation belongs to the binary that wires /// the engine (the CLI's `runtime` module), so `stella-core` never carries /// a concrete time source of its own. +/// +/// The epoch is the holder's choice, and each holder says which it needs. +/// The router's circuit breaker subtracts two reads, so any origin serves +/// it; [`crate::bus::HookBus`] writes the reading into a stamp a hook script +/// reads on the other side of a process boundary, so it needs one counting +/// from the Unix epoch. pub trait Clock: Send + Sync { - /// Monotonic milliseconds since an arbitrary epoch. + /// Milliseconds since the holder's epoch, never decreasing. fn now_ms(&self) -> u64; } +/// A [`Clock`] pinned at one reading. +/// +/// A replay stamps its events from the record it is replaying, and a test +/// asserting on a stamp needs one it can predict. Both want a clock that +/// does not move, and neither is a reason for `stella-core` to read the +/// real one. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct FixedClock(pub u64); + +impl Clock for FixedClock { + fn now_ms(&self) -> u64 { + self.0 + } +} + /// Call-outcome feedback into provider routing — the write half of /// [`crate::router::Router`]'s circuit breaker (§7 reliability rules; L-M7), /// which implements it. The engine reports each *logical* model call's diff --git a/crates/stella-core/src/retry.rs b/crates/stella-core/src/retry.rs index fe4a95bd5d..20b6fbf9ef 100644 --- a/crates/stella-core/src/retry.rs +++ b/crates/stella-core/src/retry.rs @@ -24,32 +24,53 @@ //! keep-alive, budget-derived allowance — and only an unaffordable wait //! still fails fast. Which failures those are is decided at the adapter //! and asked here through [`ProviderError::is_park_eligible`], never -//! re-derived (L-M7 again). Nothing here reads a clock or a budget -//! directly; the supervisor is a port, like [`Sleeper`]. +//! re-derived (L-M7 again). Nothing here reads a budget; the supervisor +//! is a port, like [`Sleeper`]. The one clock read left is the monotonic +//! `Instant` that times each attempt for the failure observer, and +//! `make core-no-io` counts it. //! //! Per-call timeouts (L-E4) are a caller concern layered on top of //! `attempt_fn`; this module owns "should we try again, and if so after how //! long" — and the jitter on that delay is part of the answer //! ([`compute_backoff_delay_ms`]'s equal jitter, `server_hint_delay_ms`'s -//! additive nudge), not something a caller layers on. +//! additive nudge), not something a caller layers on. The entropy behind +//! that jitter is the one input this module cannot compute, so it comes +//! through [`Sleeper::jitter`] rather than from an RNG this crate seeds +//! itself: a retry ladder that rolled its own dice could not be replayed, +//! and `stella-core` would link an entropy source for one draw. use std::future::Future; use async_trait::async_trait; -use rand::{Rng, RngExt}; use stella_protocol::ProviderError; -/// The delay port `retry_with_backoff` sleeps through between attempts. +/// The backoff port: how `retry_with_backoff` waits between attempts, and +/// where the entropy that spreads those waits apart comes from. +/// /// Injectable so retry-loop tests run instantly and deterministically /// instead of paying real wall-clock delays — the same seam /// [`crate::ports::Clock`] provides for reading time, but for the one place /// this crate needs to actually suspend a task. Only the trait lives here — /// the production tokio-backed impl belongs to the binary that constructs /// the engine (the CLI's `runtime` module). +/// +/// Both methods are required, with no default. A default `jitter` of zero +/// would let a host forget the draw and ship a fleet whose workers all wake +/// on the same millisecond after a shared 429, silently — the herd this +/// jitter exists to break up. A host that wants no jitter writes that down. #[async_trait] pub trait Sleeper: Send + Sync { /// Suspend the current task for `duration_ms` milliseconds. async fn sleep(&self, duration_ms: u64); + + /// A uniform draw from `0..=upper`, the spread between the backoff floor + /// and its cap that this attempt actually waits. + /// + /// Production draws from the OS entropy pool; a test double returns a + /// fixed value or a seeded sequence, which is what makes a retry ladder + /// assertable to the millisecond. A draw past `upper` is clamped by the + /// caller, so a careless impl widens nothing. + fn jitter(&self, upper: u64) -> u64; } /// Milliseconds per parked-wait chunk: a long rate-limit wait is slept in @@ -324,9 +345,9 @@ pub struct RetryOutcome { /// (thundering herd). /// /// Pure and synchronous by design — the only -/// non-determinism is the injected `rng`, so bounds and shape are directly -/// assertable without sleeping. -pub fn compute_backoff_delay_ms(policy: &RetryPolicy, attempt: u32, rng: &mut impl Rng) -> u64 { +/// non-determinism is the injected [`Sleeper::jitter`] draw, so bounds and +/// shape are directly assertable without sleeping. +pub fn compute_backoff_delay_ms(policy: &RetryPolicy, attempt: u32, sleeper: &dyn Sleeper) -> u64 { let base = policy.base_delay_ms; let cap = policy.max_delay_ms.max(base); let exponential = base.saturating_mul(2u64.saturating_pow(attempt)); @@ -335,10 +356,11 @@ pub fn compute_backoff_delay_ms(policy: &RetryPolicy, attempt: u32, rng: &mut im if high <= base { // Degenerate range (attempt 0, or a misconfigured policy where the // cap doesn't exceed the floor): nothing to jitter, return the - // floor rather than call `rng.random_range` on an empty range. + // floor rather than ask for a draw from an empty range. return base; } - rng.random_range(base..=high) + let span = high - base; + base + sleeper.jitter(span).min(span) } /// Turn a server `Retry-After` hint into the actual delay to sleep: raised to @@ -352,14 +374,14 @@ pub fn compute_backoff_delay_ms(policy: &RetryPolicy, attempt: u32, rng: &mut im /// [`compute_backoff_delay_ms`], because the whole point of honoring a server /// hint is to retry *no earlier* than it asked. So the result is always at or /// above the floored hint. Pure and synchronous; the only non-determinism is -/// the injected `rng`. -fn server_hint_delay_ms(policy: &RetryPolicy, hint_ms: u64, rng: &mut impl Rng) -> u64 { +/// the injected [`Sleeper::jitter`] draw. +fn server_hint_delay_ms(policy: &RetryPolicy, hint_ms: u64, sleeper: &dyn Sleeper) -> u64 { let floored = hint_ms.max(policy.base_delay_ms); let jitter_span = floored / 8; if jitter_span == 0 { return floored; } - floored.saturating_add(rng.random_range(0..=jitter_span)) + floored.saturating_add(sleeper.jitter(jitter_span).min(jitter_span)) } /// Drive `attempt_fn` to completion, retrying retryable @@ -466,16 +488,14 @@ where // is for (#2677, #2742). let ladder_open = attempt < policy.max_retries; let inline_delay_ms = match park_hint { - None if ladder_open => { - Some(compute_backoff_delay_ms(policy, attempt, &mut rand::rng())) - } + None if ladder_open => Some(compute_backoff_delay_ms(policy, attempt, sleeper)), None => return Err(error), Some(hint) if ladder_open && hint.is_none_or(|h| h <= policy.max_server_hint_ms) => { Some(match hint { - Some(h) => server_hint_delay_ms(policy, h, &mut rand::rng()), - None => compute_backoff_delay_ms(policy, attempt, &mut rand::rng()), + Some(h) => server_hint_delay_ms(policy, h, sleeper), + None => compute_backoff_delay_ms(policy, attempt, sleeper), }) } Some(_) => None, @@ -498,7 +518,7 @@ where } let hint_ms = park_hint.flatten(); let hint_delay_ms = - hint_ms.map(|h| server_hint_delay_ms(policy, h, &mut rand::rng())); + hint_ms.map(|h| server_hint_delay_ms(policy, h, sleeper)); let allowance_ms = park.wait_allowance_ms(parked_total_ms); let total_ms = match plan_park(hint_delay_ms, park_streak, allowance_ms) { ParkPlan::Wait { total_ms } => total_ms, @@ -573,17 +593,34 @@ mod tests { use std::sync::Mutex; use std::sync::atomic::{AtomicU32, Ordering}; - use rand::SeedableRng; use rand::rngs::StdRng; + use rand::{RngExt, SeedableRng}; use super::*; /// A [`Sleeper`] that never actually waits — it just records every /// requested delay so async-loop tests can assert on retry timing - /// without paying real wall-clock cost or flaking under load. - #[derive(Default)] + /// without paying real wall-clock cost or flaking under load. Its + /// jitter is a seeded `StdRng`, so a test that wants the draw to vary + /// gets a real spread, and one that reruns gets the same spread. struct NoopSleeper { delays_ms: Mutex>, + rng: Mutex, + } + + impl NoopSleeper { + fn seeded(seed: u64) -> Self { + Self { + delays_ms: Mutex::new(Vec::new()), + rng: Mutex::new(StdRng::seed_from_u64(seed)), + } + } + } + + impl Default for NoopSleeper { + fn default() -> Self { + Self::seeded(0) + } } #[async_trait] @@ -594,6 +631,13 @@ mod tests { .expect("mutex poisoned") .push(duration_ms); } + + fn jitter(&self, upper: u64) -> u64 { + self.rng + .lock() + .expect("mutex poisoned") + .random_range(0..=upper) + } } // ---- RetryPolicy ---------------------------------------------------- @@ -636,16 +680,53 @@ mod tests { // attempt 0: base * 2^0 == base, so the jitter range is degenerate // and the result is deterministic. let policy = RetryPolicy::new(5, 250, 8_000); - let mut rng = StdRng::seed_from_u64(1); - assert_eq!(compute_backoff_delay_ms(&policy, 0, &mut rng), 250); + let sleeper = NoopSleeper::seeded(1); + assert_eq!(compute_backoff_delay_ms(&policy, 0, &sleeper), 250); + } + + /// A [`Sleeper`] whose draw is always the same number. + struct FixedJitter(u64); + + #[async_trait] + impl Sleeper for FixedJitter { + async fn sleep(&self, _duration_ms: u64) {} + + fn jitter(&self, _upper: u64) -> u64 { + self.0 + } + } + + /// The draw comes from the port. A sleeper answering a fixed number + /// lands the delay at exactly the floor plus that number, which no RNG + /// this crate seeded itself could promise. A ladder drawing from + /// `rand::rng()` cannot pass this, and that is what it did before. + #[test] + fn the_jitter_is_the_ports_draw_and_a_wide_draw_is_clamped() { + let policy = RetryPolicy::new(5, 100, 8_000); + // Attempt 3: 100 * 2^3 = 800, so the span above the floor is 700. + assert_eq!(compute_backoff_delay_ms(&policy, 3, &FixedJitter(0)), 100); + assert_eq!(compute_backoff_delay_ms(&policy, 3, &FixedJitter(250)), 350); + assert_eq!( + compute_backoff_delay_ms(&policy, 3, &FixedJitter(u64::MAX)), + 800 + ); + // A server hint of 8000ms: the nudge is at most an eighth of it. + assert_eq!( + server_hint_delay_ms(&policy, 8_000, &FixedJitter(40)), + 8_040 + ); + assert_eq!( + server_hint_delay_ms(&policy, 8_000, &FixedJitter(5_000)), + 9_000 + ); } #[test] fn delay_stays_within_base_and_cap_bounds_across_many_attempts() { let policy = RetryPolicy::new(10, 100, 5_000); - let mut rng = StdRng::seed_from_u64(7); + let sleeper = NoopSleeper::seeded(7); for attempt in 0..30 { - let delay = compute_backoff_delay_ms(&policy, attempt, &mut rng); + let delay = compute_backoff_delay_ms(&policy, attempt, &sleeper); assert!( (policy.base_delay_ms..=policy.max_delay_ms).contains(&delay), "attempt {attempt}: delay {delay} out of [{}, {}]", @@ -658,7 +739,7 @@ mod tests { #[test] fn delay_grows_with_attempt_number_up_to_the_cap() { let policy = RetryPolicy::new(10, 50, 4_000); - let mut rng = StdRng::seed_from_u64(42); + let sleeper = NoopSleeper::seeded(42); // Use the upper bound of what's achievable at each attempt (the // exponential envelope) rather than one jittered sample, since a // single draw can dip low even as the ceiling climbs. @@ -681,7 +762,7 @@ mod tests { assert_eq!(envelope(20), policy.max_delay_ms); // Sanity: real samples at a late attempt never exceed the cap. for _ in 0..20 { - assert!(compute_backoff_delay_ms(&policy, 20, &mut rng) <= policy.max_delay_ms); + assert!(compute_backoff_delay_ms(&policy, 20, &sleeper) <= policy.max_delay_ms); } } @@ -691,9 +772,9 @@ mod tests { // wide [base, cap] range sampled many times at a fixed attempt // should not collapse to a single value. let policy = RetryPolicy::new(10, 100, 10_000); - let mut rng = StdRng::seed_from_u64(99); + let sleeper = NoopSleeper::seeded(99); let samples: Vec = (0..50) - .map(|_| compute_backoff_delay_ms(&policy, 5, &mut rng)) + .map(|_| compute_backoff_delay_ms(&policy, 5, &sleeper)) .collect(); let first = samples[0]; assert!( @@ -705,9 +786,9 @@ mod tests { #[test] fn zero_policy_never_panics_and_returns_zero() { let policy = RetryPolicy::deterministic(); - let mut rng = StdRng::seed_from_u64(3); - assert_eq!(compute_backoff_delay_ms(&policy, 0, &mut rng), 0); - assert_eq!(compute_backoff_delay_ms(&policy, 7, &mut rng), 0); + let sleeper = NoopSleeper::seeded(3); + assert_eq!(compute_backoff_delay_ms(&policy, 0, &sleeper), 0); + assert_eq!(compute_backoff_delay_ms(&policy, 7, &sleeper), 0); } #[test] @@ -715,8 +796,8 @@ mod tests { // cap < base is a misconfiguration, but must degrade safely rather // than panic the sampler on an empty range. let policy = RetryPolicy::new(3, 5_000, 100); - let mut rng = StdRng::seed_from_u64(11); - let delay = compute_backoff_delay_ms(&policy, 4, &mut rng); + let sleeper = NoopSleeper::seeded(11); + let delay = compute_backoff_delay_ms(&policy, 4, &sleeper); assert_eq!(delay, policy.base_delay_ms); } @@ -728,8 +809,8 @@ mod tests { attempt in 0u32..40, ) { let policy = RetryPolicy::new(10, base, base + extra); - let mut rng = StdRng::seed_from_u64(u64::from(attempt) ^ base ^ extra); - let delay = compute_backoff_delay_ms(&policy, attempt, &mut rng); + let sleeper = NoopSleeper::seeded(u64::from(attempt) ^ base ^ extra); + let delay = compute_backoff_delay_ms(&policy, attempt, &sleeper); proptest::prop_assert!(delay >= policy.base_delay_ms); proptest::prop_assert!(delay <= policy.max_delay_ms); } @@ -1312,12 +1393,12 @@ mod tests { // [floored, floored + floored/8], and the draw actually varies so a // fleet decorrelates instead of waking in lockstep. let policy = RetryPolicy::new(3, 250, 8_000); - let mut rng = StdRng::seed_from_u64(2024); + let sleeper = NoopSleeper::seeded(2024); let hint = 30_000u64; let floor = hint; // hint already above base_delay_ms let ceil = floor + floor / 8; let samples: Vec = (0..64) - .map(|_| server_hint_delay_ms(&policy, hint, &mut rng)) + .map(|_| server_hint_delay_ms(&policy, hint, &sleeper)) .collect(); for &d in &samples { assert!( diff --git a/crates/stella-core/src/subagent/tests.rs b/crates/stella-core/src/subagent/tests.rs index 4a8c56f840..a57ae6c408 100644 --- a/crates/stella-core/src/subagent/tests.rs +++ b/crates/stella-core/src/subagent/tests.rs @@ -31,6 +31,11 @@ pub(crate) struct NoSleep; #[async_trait] impl Sleeper for NoSleep { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// A provider that returns a fixed sequence of results, then errors. diff --git a/crates/stella-core/src/subagent/tests/seams.rs b/crates/stella-core/src/subagent/tests/seams.rs index 55bf434de5..dd25973b9f 100644 --- a/crates/stella-core/src/subagent/tests/seams.rs +++ b/crates/stella-core/src/subagent/tests/seams.rs @@ -250,7 +250,7 @@ fn empty_turn_controls_leave_an_engine_exactly_as_it_was() { #[test] fn attribution_leaves_its_own_scope_including_on_an_unwind() { - let bus = HookBus::new("session-1"); + let bus = HookBus::new("session-1", crate::ports::FixedClock(0)); let parent = bus.push_agent("parent".into()); { @@ -451,7 +451,7 @@ async fn subagent_start_and_stop_hooks_fire_around_a_child_turn() { /// turn of its own. #[tokio::test] async fn a_forked_child_stamps_the_subagent_fork_lane() { - let bus = HookBus::new("fork-lane-test"); + let bus = HookBus::new("fork-lane-test", crate::ports::FixedClock(0)); let seen: std::sync::Arc>> = std::sync::Arc::new(Mutex::new(Vec::new())); let sink = std::sync::Arc::clone(&seen); diff --git a/crates/stella-core/tests/engine_emits_no_stage.rs b/crates/stella-core/tests/engine_emits_no_stage.rs index 25610e9c03..2e61263845 100644 --- a/crates/stella-core/tests/engine_emits_no_stage.rs +++ b/crates/stella-core/tests/engine_emits_no_stage.rs @@ -34,6 +34,11 @@ struct NoopSleeper; #[async_trait] impl Sleeper for NoopSleeper { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// Answers once, with text and no tool calls, so the turn completes in one diff --git a/crates/stella-core/tests/hard_drop_write_back.rs b/crates/stella-core/tests/hard_drop_write_back.rs index 0263910452..f40c7fcee9 100644 --- a/crates/stella-core/tests/hard_drop_write_back.rs +++ b/crates/stella-core/tests/hard_drop_write_back.rs @@ -33,6 +33,11 @@ struct NoopSleeper; #[async_trait] impl Sleeper for NoopSleeper { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// Always answers with the same single tool call, so the turn reaches tool diff --git a/crates/stella-core/tests/hook_bus_stamps_from_the_clock.rs b/crates/stella-core/tests/hook_bus_stamps_from_the_clock.rs new file mode 100644 index 0000000000..2a48a0a2bb --- /dev/null +++ b/crates/stella-core/tests/hook_bus_stamps_from_the_clock.rs @@ -0,0 +1,32 @@ +//! Witness: a hook event's `timestamp` is the injected clock's reading, not +//! the process's wall clock. +//! +//! A bus that stamps from `SystemTime::now()` cannot pass this: no test can +//! say what such a stamp will be. Two readings decades apart show the stamp +//! follows the clock the bus was handed. + +use std::sync::{Arc, Mutex}; + +use stella_core::bus::{HookBus, names}; +use stella_core::ports::FixedClock; + +fn stamp_at(unix_ms: u64) -> String { + let bus = HookBus::new("witness", FixedClock(unix_ms)); + let seen = Arc::new(Mutex::new(None)); + let sink = seen.clone(); + let _observer = bus.on("*", move |event| { + *sink.lock().expect("mutex poisoned") = Some(event.timestamp.clone()); + Ok(()) + }); + bus.emit_named(names::FILE_READ, serde_json::json!({"path": "a"})); + seen.lock() + .expect("mutex poisoned") + .clone() + .expect("the observer saw the event") +} + +#[test] +fn the_stamp_is_the_clocks_reading() { + assert_eq!(stamp_at(0), "1970-01-01T00:00:00.000Z"); + assert_eq!(stamp_at(1_700_000_000_123), "2023-11-14T22:13:20.123Z"); +} diff --git a/crates/stella-core/tests/parallel_dispatch.rs b/crates/stella-core/tests/parallel_dispatch.rs index f373fff87d..39cc06c38c 100644 --- a/crates/stella-core/tests/parallel_dispatch.rs +++ b/crates/stella-core/tests/parallel_dispatch.rs @@ -37,6 +37,11 @@ struct NoopSleeper; #[async_trait] impl Sleeper for NoopSleeper { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// First call: one step carrying two sibling `delegate` calls. Second call: done. diff --git a/crates/stella-core/tests/spend_gate.rs b/crates/stella-core/tests/spend_gate.rs index b6c75d37f9..5845239e73 100644 --- a/crates/stella-core/tests/spend_gate.rs +++ b/crates/stella-core/tests/spend_gate.rs @@ -63,6 +63,11 @@ struct NoopSleeper; #[async_trait] impl Sleeper for NoopSleeper { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// A scripted `Provider`: one entry per call, repeating the last entry once diff --git a/crates/stella-core/tests/tool_wall_clock.rs b/crates/stella-core/tests/tool_wall_clock.rs index be95da365b..867bd551f9 100644 --- a/crates/stella-core/tests/tool_wall_clock.rs +++ b/crates/stella-core/tests/tool_wall_clock.rs @@ -38,6 +38,11 @@ struct NoopSleeper; #[async_trait] impl Sleeper for NoopSleeper { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// First call: one long shell command. Second call: an answer. The turn only diff --git a/crates/stella-engine/src/tests.rs b/crates/stella-engine/src/tests.rs index 0d674fbc22..2be4bd1121 100644 --- a/crates/stella-engine/src/tests.rs +++ b/crates/stella-engine/src/tests.rs @@ -28,6 +28,11 @@ struct NoopSleeper; #[async_trait] impl Sleeper for NoopSleeper { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// A scripted provider: one response per call, in order, looping the last diff --git a/crates/stella-engine/tests/embedding.rs b/crates/stella-engine/tests/embedding.rs index 27df0ec742..fe3022e26b 100644 --- a/crates/stella-engine/tests/embedding.rs +++ b/crates/stella-engine/tests/embedding.rs @@ -309,6 +309,11 @@ struct NoopSleeper; #[async_trait] impl Sleeper for NoopSleeper { async fn sleep(&self, _duration_ms: u64) {} + + // The floor: a test that asserts on retry timing wants no spread in it. + fn jitter(&self, _upper: u64) -> u64 { + 0 + } } /// What one consult of the re-query port was shown, owned so the port can keep diff --git a/crates/stella-serve/src/extensions.rs b/crates/stella-serve/src/extensions.rs index 453447d6c5..d1262bb28e 100644 --- a/crates/stella-serve/src/extensions.rs +++ b/crates/stella-serve/src/extensions.rs @@ -144,7 +144,7 @@ pub(crate) fn install_for_turn( if extensions.is_empty() { return None; } - let bus = HookBus::new(turn_id); + let bus = HookBus::new(turn_id, crate::remote::WallClock); for extension in extensions { extension.install(&bus); } diff --git a/crates/stella-serve/src/remote.rs b/crates/stella-serve/src/remote.rs index 53cfc49de5..bcc6323975 100644 --- a/crates/stella-serve/src/remote.rs +++ b/crates/stella-serve/src/remote.rs @@ -16,10 +16,13 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, Instant}; use async_trait::async_trait; +use rand::RngExt; use serde_json::Value; use stella_core::bus::{self, HookBus, HookEventDraft, names as hook_names}; use stella_core::hooks::decision::{GateVerdict, OperatorPosture, resolve_precedence}; -use stella_core::ports::{AuthzGate, DispatchAdmission, DispatchGate, Principal, ToolExecutor}; +use stella_core::ports::{ + AuthzGate, Clock, DispatchAdmission, DispatchGate, Principal, ToolExecutor, +}; use stella_core::retry::Sleeper; use stella_protocol::{ CompletionRequestRef, CompletionResult, Provider, ProviderError, ToolCallObserver, ToolOutput, @@ -128,7 +131,9 @@ fn forward_delta(observer: Option<&dyn ToolCallObserver>, delta: &ProviderDelta) /// A Tokio-backed [`Sleeper`] for the session runtime's retry backoff. The /// session runtime is built with the time driver enabled, so `sleep` resolves -/// there. +/// there. The jitter draws from the OS entropy pool, which is the host's to +/// hold: `stella-core` takes the draw through the port and links no entropy +/// source of its own. pub(crate) struct TokioSleeper; #[async_trait] @@ -136,6 +141,29 @@ impl Sleeper for TokioSleeper { async fn sleep(&self, duration_ms: u64) { tokio::time::sleep(std::time::Duration::from_millis(duration_ms)).await; } + + fn jitter(&self, upper: u64) -> u64 { + rand::rng().random_range(0..=upper) + } +} + +/// The host's own clock, counting from the Unix epoch, for the stamp on +/// every hook event an installed extension sees. Those stamps are read on +/// the other side of the wire and compared with the host's own, so they have +/// to share the host's origin; `stella-cli`'s `WallClock` answers the same +/// port the same way. A system clock set before the epoch reads as `0` +/// rather than failing the turn. +#[derive(Debug, Default, Clone, Copy)] +pub(crate) struct WallClock; + +impl Clock for WallClock { + fn now_ms(&self) -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |since| { + u64::try_from(since.as_millis()).unwrap_or(u64::MAX) + }) + } } /// The `Provider` port as a reverse-RPC to the host. `complete_ref` emits a diff --git a/crates/stella-serve/src/remote/tests.rs b/crates/stella-serve/src/remote/tests.rs index 41b797b8a7..8c8ef8d4b8 100644 --- a/crates/stella-serve/src/remote/tests.rs +++ b/crates/stella-serve/src/remote/tests.rs @@ -33,7 +33,7 @@ fn disconnected_port() -> (RemoteToolExecutor, Arc>>) { crate::observe::null_observer(), TurnRef::new("turn-remotetest"), ); - let bus = HookBus::new("turn-remotetest"); + let bus = HookBus::new("turn-remotetest", stella_core::ports::FixedClock(0)); let seen = Arc::new(Mutex::new(Vec::new())); let recorder = Arc::clone(&seen); bus.on("tool.call.*", move |event: &HookEvent| { @@ -164,7 +164,7 @@ async fn a_tool_call_the_host_never_answers_still_reports_an_outcome() { crate::observe::null_observer(), TurnRef::new("turn-remotetest"), ); - let bus = HookBus::new("turn-remotetest"); + let bus = HookBus::new("turn-remotetest", stella_core::ports::FixedClock(0)); let seen = Arc::new(Mutex::new(Vec::new())); let recorder = Arc::clone(&seen); bus.on("tool.call.*", move |event: &HookEvent| { @@ -387,7 +387,7 @@ async fn a_denying_gate_refuses_a_remoted_call_before_any_frame_leaves() { }); // A bus recording `policy.evaluated`: the served surface must journal // the same rule-by-rule account the CLI's `GatedToolSet` does (#3362). - let bus = HookBus::new("turn-authztest0"); + let bus = HookBus::new("turn-authztest0", stella_core::ports::FixedClock(0)); let seen: Arc>> = Arc::default(); let journal = Arc::clone(&seen); bus.on(hook_names::POLICY_EVALUATED, move |event: &HookEvent| { @@ -501,7 +501,7 @@ fn policed_port(refused: &'static str) -> RemoteToolExecutor { crate::observe::null_observer(), TurnRef::new("turn-policedtest"), ); - let bus = HookBus::new("turn-policedtest"); + let bus = HookBus::new("turn-policedtest", stella_core::ports::FixedClock(0)); bus.on_blocking(hook_names::TOOL_CALL_REQUESTED, move |event| { if event.payload["tool"].as_str() == Some(refused) { return stella_core::bus::HookDecision::Deny( diff --git a/crates/stella-serve/src/session.rs b/crates/stella-serve/src/session.rs index 8a796f343b..df0e46264f 100644 --- a/crates/stella-serve/src/session.rs +++ b/crates/stella-serve/src/session.rs @@ -1153,7 +1153,8 @@ mod tests { ); let calibration = stella_core::estimator::CalibrationMap::default(); - let bus = stella_core::bus::HookBus::new("serve-lane-test"); + let bus = + stella_core::bus::HookBus::new("serve-lane-test", stella_core::ports::FixedClock(0)); let requery = SilentRequery; let host_supplied_all = served_capabilities( Some(&calibration), diff --git a/crates/stella-tools/src/ctx.rs b/crates/stella-tools/src/ctx.rs index 31712baa36..b56a9356c5 100644 --- a/crates/stella-tools/src/ctx.rs +++ b/crates/stella-tools/src/ctx.rs @@ -539,7 +539,7 @@ mod tests { type Seen = Arc>>; fn recording_bus() -> (HookBus, Seen) { - let bus = HookBus::new("test-session"); + let bus = HookBus::new("test-session", stella_core::ports::FixedClock(0)); let seen: Arc>> = Arc::default(); let sink = seen.clone(); bus.on("tool.call.*", move |event| { diff --git a/crates/stella-tools/src/custom/tests/claims_and_plugins.rs b/crates/stella-tools/src/custom/tests/claims_and_plugins.rs index 1459af296b..02240119d2 100644 --- a/crates/stella-tools/src/custom/tests/claims_and_plugins.rs +++ b/crates/stella-tools/src/custom/tests/claims_and_plugins.rs @@ -436,7 +436,7 @@ fn gated_fixture( tool.name = gated_name.to_string(); let registry = crate::registry::ToolRegistry::new(dir.to_path_buf()); - let bus = HookBus::new("custom-gate-test"); + let bus = HookBus::new("custom-gate-test", stella_core::ports::FixedClock(0)); let gated_name = gated_name.to_string(); bus.on_blocking(hook_names::TOOL_CALL_REQUESTED, move |event| { if event.payload["tool"] == gated_name.as_str() { @@ -596,7 +596,7 @@ async fn a_name_that_falls_through_is_gated_exactly_once() { let dir = tempfile::tempdir().unwrap(); let registry = crate::registry::ToolRegistry::new(dir.path().to_path_buf()); - let bus = HookBus::new("once-test"); + let bus = HookBus::new("once-test", stella_core::ports::FixedClock(0)); let seen = Arc::new(AtomicUsize::new(0)); let counter = seen.clone(); bus.on_blocking(hook_names::TOOL_CALL_REQUESTED, move |_| { diff --git a/crates/stella-tools/src/gated.rs b/crates/stella-tools/src/gated.rs index 75f9ea72a2..e7b0721cd6 100644 --- a/crates/stella-tools/src/gated.rs +++ b/crates/stella-tools/src/gated.rs @@ -890,7 +890,7 @@ mod tests { /// denied" is reconstructable after the fact. #[tokio::test] async fn an_attached_bus_journals_the_decision_with_its_trace() { - let bus = stella_core::bus::HookBus::new("test-session"); + let bus = stella_core::bus::HookBus::new("test-session", stella_core::ports::FixedClock(0)); let seen: Arc>> = Arc::default(); let sink = Arc::clone(&seen); bus.on(stella_core::bus::names::POLICY_EVALUATED, move |event| { diff --git a/crates/stella-tools/src/hook_bridge.rs b/crates/stella-tools/src/hook_bridge.rs index e63838c5c6..a2fa67185c 100644 --- a/crates/stella-tools/src/hook_bridge.rs +++ b/crates/stella-tools/src/hook_bridge.rs @@ -171,7 +171,7 @@ mod tests { /// bit, and reason. #[tokio::test] async fn a_hook_approval_resolves_through_the_broker_with_the_audit_trail() { - let bus = HookBus::new("hook-bridge-test"); + let bus = HookBus::new("hook-bridge-test", stella_core::ports::FixedClock(0)); let events = collect_approval_events(&bus); let responder = Arc::new(Scripted { answer: ApprovalResponse::Approve, @@ -219,7 +219,7 @@ mod tests { /// config. #[tokio::test] async fn a_renamed_route_stamps_its_producer_on_the_audit_trail() { - let bus = HookBus::new("hook-bridge-test"); + let bus = HookBus::new("hook-bridge-test", stella_core::ports::FixedClock(0)); let events = collect_approval_events(&bus); let route = BrokerApprovalRoute::new( ApprovalBroker::interactive( @@ -249,7 +249,7 @@ mod tests { /// denial is audited. #[tokio::test] async fn a_denying_responder_reaches_the_bridge_with_its_reason() { - let bus = HookBus::new("hook-bridge-test"); + let bus = HookBus::new("hook-bridge-test", stella_core::ports::FixedClock(0)); let events = collect_approval_events(&bus); let route = BrokerApprovalRoute::new( ApprovalBroker::interactive( diff --git a/crates/stella-tools/src/registry/approval.rs b/crates/stella-tools/src/registry/approval.rs index 41614fb69b..edb6c2e432 100644 --- a/crates/stella-tools/src/registry/approval.rs +++ b/crates/stella-tools/src/registry/approval.rs @@ -463,7 +463,7 @@ mod tests { fn fixture(gate_event: &str, reason: &str) -> (tempfile::TempDir, ToolRegistry, HookBus) { let dir = tempfile::tempdir().unwrap(); let reg = ToolRegistry::new(dir.path().to_path_buf()); - let bus = HookBus::new("approval-test"); + let bus = HookBus::new("approval-test", stella_core::ports::FixedClock(0)); let reason = reason.to_string(); bus.on_blocking(gate_event, move |_| HookDecision::RequireApproval { reason: reason.clone(), diff --git a/crates/stella-tools/src/subagent/tests.rs b/crates/stella-tools/src/subagent/tests.rs index a9b323ef09..a4f24ea68d 100644 --- a/crates/stella-tools/src/subagent/tests.rs +++ b/crates/stella-tools/src/subagent/tests.rs @@ -442,7 +442,7 @@ fn minting_is_a_function_of_call_order_alone() { async fn dispatching_a_child_emits_declared_progress_on_the_bus() { let (tool, _dispatcher) = tool_with(SubAgentOutcome::Completed(report("done", 0.0, false))); - let bus = stella_core::bus::HookBus::new("test-session"); + let bus = stella_core::bus::HookBus::new("test-session", stella_core::ports::FixedClock(0)); let seen: Arc>> = Arc::default(); let sink = seen.clone(); bus.on("tool.call.progress", move |event| { diff --git a/crates/stella-tools/tests/approval_witness.rs b/crates/stella-tools/tests/approval_witness.rs index 05f0f8225b..ca5ff62adc 100644 --- a/crates/stella-tools/tests/approval_witness.rs +++ b/crates/stella-tools/tests/approval_witness.rs @@ -18,7 +18,7 @@ use stella_tools::ToolRegistry; async fn headless_require_approval_names_the_missing_surface_and_grant_path() { let dir = tempfile::tempdir().unwrap(); let reg = ToolRegistry::new(dir.path().to_path_buf()); - let bus = HookBus::new("witness-2676"); + let bus = HookBus::new("witness-2676", stella_core::ports::FixedClock(0)); bus.on_blocking(hook_names::TOOL_CALL_REQUESTED, |_| { HookDecision::RequireApproval { reason: "policy wants a human".into(), diff --git a/crates/stella-tools/tests/ask_question_a2a_witness.rs b/crates/stella-tools/tests/ask_question_a2a_witness.rs index b1843dd2c6..2afc2123fb 100644 --- a/crates/stella-tools/tests/ask_question_a2a_witness.rs +++ b/crates/stella-tools/tests/ask_question_a2a_witness.rs @@ -139,7 +139,7 @@ async fn a_child_behind_read_only_tools_can_see_and_call_ask_question() { #[tokio::test] async fn a_childs_question_is_attributed_to_the_child_not_the_turn() { let (registry, responder) = registry_with_driver(); - let bus = HookBus::new("ses-a2a-witness"); + let bus = HookBus::new("ses-a2a-witness", stella_core::ports::FixedClock(0)); registry.attach_bus(bus.clone()); // Top-level: no agent is entered, so the question is the driver's own. diff --git a/scripts/check-core-no-io.py b/scripts/check-core-no-io.py new file mode 100755 index 0000000000..fc2b420128 --- /dev/null +++ b/scripts/check-core-no-io.py @@ -0,0 +1,389 @@ +#!/usr/bin/env python3 +"""Guard: no I/O in `stella-core`, read from the source rather than assumed. + +AGENTS.md rule 2 says the engine does no I/O: anything that spawns a +process, reads a file, or hits the network belongs in an adapter crate and +reaches the engine as a port. The crate's own manifest says the same, and so +does the header of every module that takes a `Sleeper` or a `Clock`. Nothing +checked it. On 2026-09-09 an audit found the retry ladder drawing its jitter +from the OS entropy pool through `rand::rng()`, and the hook bus stamping +every event from `SystemTime::now()`, while both files' headers said the +crate reads nothing directly. A rule that only prose enforces is met until +somebody reads the prose. + +## Three questions, one run + +**The floor.** No shipping source in the crate may name a filesystem, +process, network, environment, entropy, wall-clock, or standard-stream API. +Absolute: one hit fails, and there is no baseline, because the tree has +none. `#[cfg(test)]` bodies, `tests/` directories, and `tests.rs` files are +stripped first — a test may read a fixture; the engine may not. + +**The manifest.** `[dependencies]` may not name a crate whose purpose is I/O +or entropy, and `tokio` may take only the features that schedule and time +(`sync`, `time`, `macros`, `rt`). `[dev-dependencies]` is exempt for the +same reason `#[cfg(test)]` is. + +**The ratchet.** Reads of the monotonic clock — `Instant::now()` — are not +I/O, and the engine's deadline arithmetic is full of them. They are ambient +state all the same: a turn that reads the clock itself cannot be replayed +from its record, which is why `ports::Clock` exists. Each file's count is +recorded in `scripts/core-no-io-baseline.txt` and may only go down. +`--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, never by +recording the read. + +This is a fact about the repository rather than about a crate, so it is never +scoped by CARGO_SCOPE (AGENTS.md § "The gate"), and a text-level walk keeps it +in the toolchain-free `guards-fast` rung. + +Usage: + ./scripts/check-core-no-io.py [ROOT] + ./scripts/check-core-no-io.py --update [ROOT] # shrink the baseline +""" + +from __future__ import annotations + +import importlib.util +import re +import sys +from pathlib import Path + +CRATE = "crates/stella-core" +BASELINE = "scripts/core-no-io-baseline.txt" + +# The strippers are the sibling guard's. That way both guards agree on what +# counts as shipping code. +_SIBLING = Path(__file__).with_name("check-core-reachability.py") +_spec = importlib.util.spec_from_file_location("core_reachability", _SIBLING) +if _spec is None or _spec.loader is None: + sys.exit(f"check-core-no-io: cannot load {_SIBLING}") +_sibling = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_sibling) +strip_comments = _sibling.strip_comments +strip_cfg_test = _sibling.strip_cfg_test + +# The floor: a label, a pattern, and what to do instead. Comments, strings +# and test bodies are blanked first. So a pattern named in a doc comment, as +# this file's own header does, is not a hit. +FLOOR: list[tuple[str, re.Pattern[str], str]] = [ + ( + "std::fs", + re.compile(r"\bstd\s*::\s*fs\b|\buse\s+std\s*::\s*\{[^}]*\bfs\b"), + "the filesystem is a port — take a trait here, implement it in stella-tools or stella-cli", + ), + ( + "std::process", + re.compile(r"\bstd\s*::\s*process\b|\buse\s+std\s*::\s*\{[^}]*\bprocess\b"), + "process spawning belongs to stella-tools; the engine asks a ToolExecutor", + ), + ( + "std::net", + re.compile(r"\bstd\s*::\s*net\b|\buse\s+std\s*::\s*\{[^}]*\bnet\b"), + "the network is a Provider or an MCP adapter, never the engine's own socket", + ), + ( + "std::env", + re.compile(r"\bstd\s*::\s*env\b|\buse\s+std\s*::\s*\{[^}]*\benv\b"), + "the environment is read once by the binary and handed in as config", + ), + ( + "standard streams", + re.compile( + r"\bstd\s*::\s*io\s*::\s*std(?:in|out|err)\b" + r"|\bio\s*::\s*std(?:in|out|err)\s*\(" + r"|\bstd(?:in|out|err)\s*\(\s*\)" + ), + "the engine talks through AgentEvent, not a terminal", + ), + ( + "tokio I/O", + re.compile(r"\btokio\s*::\s*(?:fs|process|net|io)\b"), + "tokio's I/O drivers belong to the host that owns the runtime", + ), + ( + "SystemTime::now", + re.compile(r"\bSystemTime\s*::\s*now\b"), + "take a ports::Clock counting from the Unix epoch, the way bus::HookBus does", + ), + ( + "a blocking sleep", + re.compile(r"\bstd\s*::\s*thread\s*::\s*sleep\b|\bthread\s*::\s*sleep\s*\("), + "the engine suspends through retry::Sleeper, never a thread", + ), + ( + "tokio::time::sleep", + re.compile(r"\btokio\s*::\s*time\s*::\s*sleep\b"), + "the engine suspends through retry::Sleeper, never a timer of its own", + ), + ( + "an entropy source", + re.compile(r"\brand\s*::\s*rng\b|\bthread_rng\b|\bOsRng\b|\bgetrandom\b"), + "a draw is a port — retry::Sleeper::jitter is the one the backoff uses", + ), + ( + "a print macro", + re.compile(r"\b(?:e?print(?:ln)?|dbg)\s*!"), + "the engine reports through AgentEvent; a print reaches nothing a host reads", + ), +] + +# The ratchet: clock reads, counted per file. +INSTANT_NOW = re.compile(r"\bInstant\s*::\s*now\s*\(") + +# The manifest: crates built for I/O or entropy, and the tokio features that +# do no I/O. `rand` heads the list. It is the one this guard was written over. +DENIED_DEPS = { + "rand", + "rand_core", + "getrandom", + "reqwest", + "hyper", + "hyper-util", + "ureq", + "rusqlite", + "sqlx", + "dirs", + "home", + "walkdir", + "ignore", + "notify", + "tempfile", + "libc", + "nix", + "mio", + "socket2", + "which", + "git2", +} +TOKIO_FEATURES_ALLOWED = {"sync", "time", "macros", "rt"} + +SECTION = re.compile(r"^\s*\[([^\]]+)\]\s*$") +DEP_KEY = re.compile(r"^\s*([A-Za-z0-9_-]+)\s*(?:\.\s*workspace\s*)?=") +FEATURES = re.compile(r"features\s*=\s*\[([^\]]*)\]") + + +def shipping_sources(src: Path) -> list[Path]: + """Every `.rs` file under `src/` that is not a test file. + + A `tests/` directory anywhere under `src/`, and a file named `tests.rs`, + are the two spellings this workspace uses for a module's tests; both are + compiled only under `cfg(test)` by the `mod` line that names them, and + the sibling guard skips them the same way. + """ + out: list[Path] = [] + for path in sorted(src.rglob("*.rs")): + rel = path.relative_to(src) + if "tests" in rel.parts[:-1] or rel.name == "tests.rs": + continue + out.append(path) + return out + + +def shipping_text(path: Path) -> str: + return strip_cfg_test(strip_comments(path.read_text(encoding="utf-8", errors="replace"))) + + +def line_of(text: str, offset: int) -> int: + return text.count("\n", 0, offset) + 1 + + +def floor_hits(root: Path, src: Path) -> list[str]: + """Every shipping line that names an I/O surface, with its remedy.""" + hits: list[str] = [] + for path in shipping_sources(src): + text = shipping_text(path) + rel = path.relative_to(root) + for label, pattern, remedy in FLOOR: + for match in pattern.finditer(text): + hits.append(f" {rel}:{line_of(text, match.start())}: {label} — {remedy}") + return hits + + +def clock_reads(root: Path, src: Path) -> dict[str, int]: + """`Instant::now()` reads per shipping file, files with none omitted.""" + counts: dict[str, int] = {} + for path in shipping_sources(src): + n = len(INSTANT_NOW.findall(shipping_text(path))) + if n: + counts[str(path.relative_to(root))] = n + return counts + + +def manifest_hits(manifest: Path) -> list[str]: + """Denied crates and tokio I/O features in `[dependencies]`.""" + if not manifest.is_file(): + return [f" {manifest}: missing — nothing declares what this crate links"] + hits: list[str] = [] + section = "" + for lineno, line in enumerate(manifest.read_text(encoding="utf-8").splitlines(), 1): + header = SECTION.match(line) + if header: + section = header.group(1).strip() + # `[dependencies.tokio]`, the table form of one dependency. + if section.startswith("dependencies."): + name = section[len("dependencies.") :] + if name in DENIED_DEPS: + hits.append(f" Cargo.toml:{lineno}: `{name}` in [dependencies]") + section = f"dependencies.{name}" + continue + if section == "dependencies": + key = DEP_KEY.match(line) + if not key: + continue + name = key.group(1) + if name in DENIED_DEPS: + hits.append(f" Cargo.toml:{lineno}: `{name}` in [dependencies]") + if name == "tokio": + hits.extend(tokio_feature_hits(line, lineno)) + elif section == "dependencies.tokio": + hits.extend(tokio_feature_hits(line, lineno)) + return hits + + +def tokio_feature_hits(line: str, lineno: int) -> list[str]: + features = FEATURES.search(line) + if not features: + return [] + named = {f.strip().strip("\"'") for f in features.group(1).split(",") if f.strip()} + extra = sorted(named - TOKIO_FEATURES_ALLOWED) + return [ + f" Cargo.toml:{lineno}: tokio feature `{f}` — only " + f"{', '.join(sorted(TOKIO_FEATURES_ALLOWED))} schedule without I/O" + for f in extra + ] + + +def read_baseline(path: Path) -> dict[str, int]: + """` ` per line, comments and blanks skipped.""" + if not path.is_file(): + return {} + out: dict[str, int] = {} + for raw in path.read_text(encoding="utf-8").splitlines(): + line = raw.strip() + if not line or line.startswith("#"): + continue + count, _, rel = line.partition(" ") + out[rel.strip()] = int(count) + return out + + +BASELINE_HEADER = """\ +# `Instant::now()` reads in stella-core's shipping source, per file. +# +# A DOWN-ONLY ratchet. It records the clock reads that predate the guard. It +# may never gain a file or a larger count, and `--update` refuses both. To +# clear a red run, read the clock once at the edge and pass `now` in, or take +# `ports::Clock`. Never record the read. The list is meant to reach empty. +# +# Regenerate after removing a read: ./scripts/check-core-no-io.py --update +""" + + +def write_baseline(path: Path, counts: dict[str, int]) -> None: + body = "".join(f"{n} {rel}\n" for rel, n in sorted(counts.items())) + path.write_text(BASELINE_HEADER + body, encoding="utf-8") + + +def main() -> int: + args = [a for a in sys.argv[1:] if a != "--update"] + update = "--update" in sys.argv[1:] + root = Path(args[0] if args else ".").resolve() + src = root / CRATE / "src" + baseline_path = root / BASELINE + + if not src.is_dir(): + print(f"check-core-no-io: no {CRATE}/src at {root} — nothing to check") + return 0 + + floor = floor_hits(root, src) + manifest = manifest_hits(root / CRATE / "Cargo.toml") + current = clock_reads(root, src) + baseline = read_baseline(baseline_path) + + failed = False + if floor: + failed = True + print("check-core-no-io: FAILED — I/O in the engine\n") + print("These shipping lines in stella-core reach outside the process:\n") + print("\n".join(floor)) + print( + "\nAGENTS.md rule 2: the engine does no I/O. Anything that reads a\n" + "file, spawns a process, hits the network, or draws from an ambient\n" + "source is a port, implemented outside stella-core and handed in.\n" + "There is no baseline for this — the tree has none, and gains none.\n" + ) + if manifest: + failed = True + print("check-core-no-io: FAILED — an I/O crate in the manifest\n") + print("These [dependencies] of stella-core link an I/O or entropy source:\n") + print("\n".join(manifest)) + print( + "\nA test that needs one lists it under [dev-dependencies]. Shipping\n" + "code takes the capability as a port.\n" + ) + + if update: + grew = sorted( + rel for rel, n in current.items() if rel not in baseline or n > baseline[rel] + ) + if grew: + print("check-core-no-io: REFUSING to grow the baseline.\n") + print("These files read Instant::now() more than the baseline records:") + for rel in grew: + print(f" {rel}: {current[rel]} (baseline {baseline.get(rel, 0)})") + print( + "\nThe ratchet only goes down. Read the clock once at the edge and\n" + "pass `now` in, or take ports::Clock — do not record the read here." + ) + return 1 + if failed: + return 1 + write_baseline(baseline_path, current) + for rel in sorted(set(baseline) - set(current)): + print(f"check-core-no-io: retired {rel} — no clock reads left") + for rel in sorted(current): + if current[rel] < baseline.get(rel, current[rel]): + print(f"check-core-no-io: lowered {rel} to {current[rel]}") + print(f"check-core-no-io: baseline holds {sum(current.values())} read(s) in {len(current)} file(s)") + return 0 + + grew = sorted(rel for rel, n in current.items() if n > baseline.get(rel, 0)) + if grew: + failed = True + print("check-core-no-io: FAILED — more clock reads than the baseline\n") + for rel in grew: + print(f" {rel}: {current[rel]} Instant::now() read(s), baseline {baseline.get(rel, 0)}") + print( + "\nThe monotonic clock is ambient state: a turn that reads it cannot be\n" + "replayed from its record. Read it once at the edge and pass `now` in,\n" + "or take ports::Clock. Do NOT add a baseline entry — the ratchet only\n" + "goes down.\n" + ) + + stale = sorted( + rel for rel, n in baseline.items() if current.get(rel, 0) < n + ) + if stale: + failed = True + print("check-core-no-io: baseline is STALE\n") + for rel in stale: + print(f" {rel}: baseline {baseline[rel]}, now {current.get(rel, 0)}") + print( + "\nSomebody removed a clock read and left the ceiling where it was. Run\n" + "`make core-no-io-update` to lower it; the ratchet never goes back up.\n" + ) + + if failed: + return 1 + + reads = sum(current.values()) + print( + f"check-core-no-io: OK — {len(shipping_sources(src))} shipping file(s) name no I/O; " + f"{reads} Instant::now() read(s) in {len(current)} file(s), at or under the baseline" + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/check-gate-parity.sh b/scripts/check-gate-parity.sh index 48ca30b6b4..1b2fba7708 100755 --- a/scripts/check-gate-parity.sh +++ b/scripts/check-gate-parity.sh @@ -156,6 +156,7 @@ step_command() { doc-links) echo 'check-doc-links' ;; module-reachability) echo 'check-module-reachability' ;; core-reachability) echo 'check-core-reachability' ;; + core-no-io) echo 'check-core-no-io' ;; retired-model-keys) echo 'check-retired-model-keys' ;; typed-errors) echo 'check-typed-errors' ;; tool-error-class) echo 'check-tool-error-class' ;; diff --git a/scripts/core-no-io-baseline.txt b/scripts/core-no-io-baseline.txt new file mode 100644 index 0000000000..17d3d1a9ee --- /dev/null +++ b/scripts/core-no-io-baseline.txt @@ -0,0 +1,18 @@ +# `Instant::now()` reads in stella-core's shipping source, per file. +# +# A DOWN-ONLY ratchet. It records the clock reads that predate the guard. It +# may never gain a file or a larger count, and `--update` refuses both. To +# clear a red run, read the clock once at the edge and pass `now` in, or take +# `ports::Clock`. Never record the read. The list is meant to reach empty. +# +# Regenerate after removing a read: ./scripts/check-core-no-io.py --update +2 crates/stella-core/src/accounted_call.rs +1 crates/stella-core/src/bus.rs +4 crates/stella-core/src/driver.rs +1 crates/stella-core/src/driver/completion.rs +2 crates/stella-core/src/driver/dispatch.rs +2 crates/stella-core/src/driver/rate_limit.rs +1 crates/stella-core/src/driver/settlement.rs +1 crates/stella-core/src/retry.rs +3 crates/stella-core/src/step.rs +2 crates/stella-core/src/subagent.rs diff --git a/scripts/test-core-no-io.sh b/scripts/test-core-no-io.sh new file mode 100755 index 0000000000..6ed815214e --- /dev/null +++ b/scripts/test-core-no-io.sh @@ -0,0 +1,222 @@ +#!/usr/bin/env bash +# +# Tests for check-core-no-io.py. +# +# ./scripts/test-core-no-io.sh +# +# Run it after touching that script. Not part of `make gate`: it builds +# throwaway crate fixtures, the same posture as `core-reachability-test`. +# +# ── Why a fixture instead of the real tree ─────────────────────────────────── +# +# A green run over the real crate proves the tree is clean today. What needs +# proving is each way the guard can be wrong: +# +# misses a file read in shipping source passes. That is the defect +# the guard exists for. +# fabricates a test body, a comment, or a test file is reported as I/O. +# Then the reader stops trusting the guard. +# manifest an I/O crate in `[dependencies]` passes, or one in +# `[dev-dependencies]` fails. +# ratchet a clock read past the baseline passes. Or `--update` records +# new debt. Or a lowered count is never reclaimed. +# +# bash 3.2 compatible. + +set -uo pipefail + +repo_root="$(cd "$(dirname "$0")/.." && pwd -P)" +SCRIPT="$repo_root/scripts/check-core-no-io.py" +TMP="$(mktemp -d)" +trap 'rm -rf "$TMP"' EXIT INT TERM + +pass=0 +fail=0 + +# A throwaway tree shaped like the real one. One crate, a clean manifest, +# one source file. +new_core() { # + local dir="$TMP/$1/crates/stella-core" + mkdir -p "$dir/src" "$TMP/$1/scripts" + printf '[package]\nname = "stella-core"\n\n[dependencies]\nserde = "1"\ntokio = { version = "1", features = ["sync", "time"] }\n\n[dev-dependencies]\nrand = "0.10"\n' >"$dir/Cargo.toml" + printf 'pub mod driver;\n' >"$dir/src/lib.rs" + printf 'pub fn drive() {}\n' >"$dir/src/driver.rs" + : >"$TMP/$1/scripts/core-no-io-baseline.txt" + echo "$dir/src" +} + +seed_baseline() { # + local case="$1" + shift + : >"$TMP/$case/scripts/core-no-io-baseline.txt" + for line in "$@"; do + echo "$line" >>"$TMP/$case/scripts/core-no-io-baseline.txt" + done +} + +# want [substring] +want() { + local name="$1" expect="$2" case="$3" sub="${4:-}" out rc + out="$(python3 "$SCRIPT" "$TMP/$case" 2>&1)" + rc=$? + if [ "$expect" = "expect-pass" ]; then + if [ "$rc" -eq 0 ]; then + pass=$((pass + 1)); echo "ok $name" + else + fail=$((fail + 1)); echo "FAIL $name — expected OK, got:"; echo "$out" + fi + return + fi + if [ "$rc" -eq 0 ]; then + fail=$((fail + 1)); echo "FAIL $name — the guard passed it:"; echo "$out" + return + fi + case "$out" in + *"$sub"*) pass=$((pass + 1)); echo "ok $name" ;; + *) fail=$((fail + 1)); echo "FAIL $name — failed for the wrong reason (wanted '$sub'):"; echo "$out" ;; + esac +} + +# ── The floor ──────────────────────────────────────────────────────────────── +src="$(new_core clean)" +want "a clean crate passes" expect-pass clean + +src="$(new_core fs)" +printf 'pub fn drive() { let _ = std::fs::read_to_string("x"); }\n' >"$src/driver.rs" +want "std::fs in shipping source is reported" expect-fail fs "driver.rs:1: std::fs" + +src="$(new_core grouped)" +printf 'use std::{fs, path::Path};\npub fn drive() { let _ = fs::read_to_string(Path::new("x")); }\n' >"$src/driver.rs" +want "a grouped use of std::fs is reported" expect-fail grouped "std::fs" + +src="$(new_core wallclock)" +printf 'pub fn now() -> u64 { std::time::SystemTime::now().elapsed().unwrap().as_millis() as u64 }\n' >"$src/driver.rs" +want "SystemTime::now is reported" expect-fail wallclock "SystemTime::now" + +src="$(new_core entropy)" +printf 'pub fn draw() -> u64 { rand::rng().random_range(0..10) }\n' >"$src/driver.rs" +want "rand::rng is reported" expect-fail entropy "an entropy source" + +src="$(new_core printing)" +printf 'pub fn drive() { println!("hi"); }\n' >"$src/driver.rs" +want "a print macro is reported" expect-fail printing "a print macro" + +src="$(new_core spawn)" +printf 'pub fn drive() { let _ = std::process::Command::new("sh"); }\n' >"$src/driver.rs" +want "std::process is reported" expect-fail spawn "std::process" + +# ── Fabrication: what is not shipping code ─────────────────────────────────── +src="$(new_core cfgtest)" +cat >"$src/driver.rs" <<'RS' +pub fn drive() {} + +#[cfg(test)] +mod tests { + #[test] + fn reads_a_fixture() { + let _ = std::fs::read_to_string("fixture"); + let _ = std::time::SystemTime::now(); + println!("{}", rand::rng().random_range(0..10)); + } +} +RS +want "a #[cfg(test)] body may do anything" expect-pass cfgtest + +src="$(new_core commented)" +printf '// Never std::fs::read here; SystemTime::now() is a port. See println!.\npub fn drive() { let _ = "std::fs::read"; }\n' >"$src/driver.rs" +want "a comment or a string is not a hit" expect-pass commented + +src="$(new_core testfiles)" +mkdir -p "$src/driver/tests" "$TMP/testfiles/crates/stella-core/tests" +printf 'pub fn drive() {}\n#[cfg(test)]\nmod tests;\n' >"$src/driver.rs" +printf 'pub fn t() { let _ = std::fs::read_to_string("x"); }\n' >"$src/driver/tests.rs" +printf 'pub fn t() { let _ = std::fs::read_to_string("x"); }\n' >"$src/driver/tests/more.rs" +printf 'fn t() { let _ = std::fs::read_to_string("x"); }\n' >"$TMP/testfiles/crates/stella-core/tests/witness.rs" +want "tests.rs, a tests/ module, and an integration test are not shipping code" expect-pass testfiles + +# ── The manifest ───────────────────────────────────────────────────────────── +src="$(new_core denied)" +printf '[package]\nname = "stella-core"\n\n[dependencies]\nrand = "0.10"\n' >"$TMP/denied/crates/stella-core/Cargo.toml" +want "an entropy crate in [dependencies] is reported" expect-fail denied "rand\` in [dependencies]" + +src="$(new_core devdep)" +want "the same crate in [dev-dependencies] is allowed" expect-pass devdep + +src="$(new_core tokiofs)" +printf '[package]\nname = "stella-core"\n\n[dependencies]\ntokio = { version = "1", features = ["sync", "fs"] }\n' >"$TMP/tokiofs/crates/stella-core/Cargo.toml" +want "a tokio I/O feature is reported" expect-fail tokiofs "tokio feature \`fs\`" + +src="$(new_core tokiotable)" +printf '[package]\nname = "stella-core"\n\n[dependencies.tokio]\nversion = "1"\nfeatures = ["process"]\n' >"$TMP/tokiotable/crates/stella-core/Cargo.toml" +want "the table spelling of a tokio dependency is read too" expect-fail tokiotable "tokio feature \`process\`" + +# ── The ratchet ────────────────────────────────────────────────────────────── +src="$(new_core clock)" +printf 'pub fn drive() { let _ = std::time::Instant::now(); let _ = std::time::Instant::now(); }\n' >"$src/driver.rs" +want "a clock read with no baseline entry is reported" expect-fail clock "Instant::now" + +seed_baseline clock "2 crates/stella-core/src/driver.rs" +want "and passes at the recorded count" expect-pass clock + +seed_baseline clock "1 crates/stella-core/src/driver.rs" +want "and fails one read over it" expect-fail clock "baseline 1" + +# --update refuses to grow. The baseline can grow two ways. Both are refused. +seed_baseline clock "1 crates/stella-core/src/driver.rs" +out="$(python3 "$SCRIPT" --update "$TMP/clock" 2>&1)" +rc=$? +if [ "$rc" -ne 0 ] && [ -z "${out##*REFUSING*}" ]; then + pass=$((pass + 1)); echo "ok --update refuses to raise a count" +else + fail=$((fail + 1)); echo "FAIL --update raised a count (rc=$rc):"; echo "$out" +fi +case "$(cat "$TMP/clock/scripts/core-no-io-baseline.txt")" in + "1 crates/stella-core/src/driver.rs") pass=$((pass + 1)); echo "ok and left the baseline untouched" ;; + *) fail=$((fail + 1)); echo "FAIL --update wrote the baseline anyway:"; cat "$TMP/clock/scripts/core-no-io-baseline.txt" ;; +esac + +seed_baseline clock +out="$(python3 "$SCRIPT" --update "$TMP/clock" 2>&1)" +rc=$? +if [ "$rc" -ne 0 ] && [ -z "${out##*REFUSING*}" ]; then + pass=$((pass + 1)); echo "ok --update refuses to add a file" +else + fail=$((fail + 1)); echo "FAIL --update added a file (rc=$rc):"; echo "$out" +fi + +# A count that dropped is stale. --update reclaims it. +src="$(new_core lowered)" +printf 'pub fn drive() { let _ = std::time::Instant::now(); }\n' >"$src/driver.rs" +seed_baseline lowered "3 crates/stella-core/src/driver.rs" +want "a count under its ceiling is reported stale" expect-fail lowered "STALE" +python3 "$SCRIPT" --update "$TMP/lowered" >/dev/null 2>&1 +case "$(grep -v '^#' "$TMP/lowered/scripts/core-no-io-baseline.txt")" in + "1 crates/stella-core/src/driver.rs") pass=$((pass + 1)); echo "ok --update lowers a count to what the file reads now" ;; + *) fail=$((fail + 1)); echo "FAIL --update did not lower the count:"; cat "$TMP/lowered/scripts/core-no-io-baseline.txt" ;; +esac + +# An entry whose file reads the clock no more is retired. +src="$(new_core retired)" +seed_baseline retired "1 crates/stella-core/src/driver.rs" +python3 "$SCRIPT" --update "$TMP/retired" >/dev/null 2>&1 +if [ -z "$(grep -v '^#' "$TMP/retired/scripts/core-no-io-baseline.txt" | tr -d '[:space:]')" ]; then + pass=$((pass + 1)); echo "ok --update retires an entry with no reads left" +else + fail=$((fail + 1)); echo "FAIL --update kept a dead entry:"; cat "$TMP/retired/scripts/core-no-io-baseline.txt" +fi + +# --update never writes over a red floor. A run judges more than the clock +# count. A baseline written beside an I/O hit would read as a pass. +src="$(new_core redfloor)" +printf 'pub fn drive() { let _ = std::fs::read_to_string("x"); }\n' >"$src/driver.rs" +out="$(python3 "$SCRIPT" --update "$TMP/redfloor" 2>&1)" +rc=$? +if [ "$rc" -ne 0 ] && [ -z "${out##*std::fs*}" ]; then + pass=$((pass + 1)); echo "ok --update still fails on a floor hit" +else + fail=$((fail + 1)); echo "FAIL --update passed a floor hit:"; echo "$out" +fi + +echo +echo "core-no-io: $pass passed, $fail failed" +[ "$fail" -eq 0 ]