macanderson · macanderson · Aug 7, 2026 · Aug 7, 2026 · Aug 7, 2026 · Aug 7, 2026
@@ -272,6 +272,8 @@ async fn run_pipeline_one_shot(
     };
     let custom_tools = discover_custom_tools(cfg, format == OutputFormat::Text).await;
     let store = open_store(&cfg.workspace_root);
+    // Owned here so it outlives every engine the pipeline builds below (#1595).
+    let calibration = seed_calibration(&store, cfg);
 
     if format == OutputFormat::Text {
         tui::section_header("Stella (pipeline)");
@@ -453,7 +455,8 @@ async fn run_pipeline_one_shot(
         };
 
         let events = pipeline_event_sender(&tx, format);
-        let pipeline = resume_frame::pipeline(&cfg.durability, ports, events, pipeline_config);
+        let pipeline = resume_frame::pipeline(&cfg.durability, ports, events, pipeline_config)
+            .with_calibration(&calibration);
         pipeline.run(prompt, &mut messages, &mut budget).await
     };
 

@@ -56,9 +56,8 @@
 //!   the checkpoint on every terminal path, abort included, so a policy stop
 //!   retracts its resume point on the way out. A row written by a build that
 //!   predates #1653 — where a policy stop really did store `Error` — is
-//!   therefore filtered by `SkipReason::NoResumePoint` anyway (named, not
-//!   linked: this module is private, so rustdoc cannot resolve it), without
-//!   this module having to trust its status.
+//!   therefore filtered by `SkipReason::NoResumePoint` anyway, without this
+//!   module having to trust its status.
 //! - **The attempt bound still applies.** An `Error` that resumes into
 //!   another `Error` is counted like any other continuation and retired
 //!   after `MAX_BOOT_ATTEMPTS`.

@@ -707,6 +707,8 @@ async fn run_task(
     // a one-shot or deck turn. The store is rooted in the task worktree so
     // parallel workers never contend on a single SQLite writer.
     let store = agent::open_store(root);
+    // Owned above the pipeline so it outlives every engine it builds (#1595).
+    let calibration = agent::seed_calibration(&store, &cfg);
     let execution = agent::begin_execution(&store, "fleet", &task.prompt, &cfg, None);
     // From here on this attempt's spend is durable in the store even if this
     // thread never lives to report it — publish the handle that makes it
@@ -879,7 +881,8 @@ async fn run_task(
             );
             let pipeline =
                 crate::resume_frame::pipeline(&cfg.durability, ports, tx.clone(), config)
-                    .with_turn_gate(gate.as_ref());
+                    .with_turn_gate(gate.as_ref())
+                    .with_calibration(&calibration);
             // The system prompt + task prompt are already in `messages`; the
             // pipeline appends its own volatile recall+goal message, so pass the
             // raw task prompt as the goal (the pipeline never re-reads `messages`

@@ -422,14 +422,17 @@ pub static CAPABILITIES: &[Capability] = &[
         engine_home: "stella-core estimator CalibrationMap: per-model token-drift correction feeding compaction",
         engine_entries: &["with_calibration"],
         cli: SurfacePosture::ShippedUnwitnessed {
-            mechanism: "seed_calibration from the store plus with_calibration on the interactive, \
-                        raw one-shot, goal, deck and sub-session paths — NOT on the default \
-                        `stella run` staged-pipeline path or on fleet workers (#1595)",
+            mechanism: "seed_calibration from the store plus with_calibration on all seven \
+                        assembly sites: the interactive, raw one-shot, goal, deck and \
+                        sub-session paths hand it to the engine directly, and the default \
+                        `stella run` staged-pipeline path and fleet workers lend it to the \
+                        Pipeline, which attaches it to every engine it builds (#1595)",
             missing: "a CLI-side test pinning that persisted drift samples reach the engine's \
                       calibration on session start (stella-core and stella-store each test their \
-                      half; the CLI seam between them has no witness) — and the same test would \
-                      have caught the two paths above, which is why this row is unwitnessed and \
-                      wrong at the same time",
+                      half; the CLI seam between them has no witness). The pipeline half of the \
+                      #1595 gap now has one — `the_pipeline_path_sizes_its_budget_with_the_\
+                      callers_calibration` proves a lent map reaches the engines — but nothing \
+                      yet proves `run_pipeline_one_shot` and the fleet worker do the lending",
         },
         api: SurfacePosture::Shipped {
             mechanism: "a process-lifetime CalibrationMap per provider_id, fed by every \

@@ -48,7 +48,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 
 use stella_core::driver::TurnHalt;
 
-use crate::verify::normalize_command;
+use crate::verify::{FlipState, normalize_command};
 
 /// The marker the bash tool appends to a command's output.
 ///
@@ -160,6 +160,40 @@ impl FlipHalt {
     }
 }
 
+/// The halt to hand a *revision* turn: [`FlipHalt::unfired`], narrowed by what
+/// the candidate's oracle currently believes about the tracked command.
+///
+/// [`FlipHalt::unfired`] refuses a latch that already fired, which covers the
+/// halt the *execute* turn stopped on. It cannot cover the other way a
+/// revision opens on an already-green command, because that halt has never
+/// fired: `witness_on_demand` arms a **fresh** `FlipHalt` after execution, and
+/// by then the witness it names may have already flipped — that is the ordinary
+/// case, since the witness is written to pass once the work is done. A verifier
+/// can still reject such a candidate (a lint regression, a refuted verdict),
+/// and the revision would then inherit an unfired halt on a command that is
+/// green before it starts. The first time the model re-ran that test the
+/// revision would end, before it addressed a single thing the verifier
+/// objected to.
+///
+/// So the oracle decides: only [`FlipState::Failing`] means the flip this halt
+/// watches for is still *ahead* of the turn. It is the same precondition
+/// `run_candidate` arms from, and the same rule stated there — a command that
+/// is passing as the turn opens cannot flip during it, so a halt on it can
+/// only end work early.
+///
+/// [`FlipState::Flipped`] and [`FlipState::Unstable`] are both excluded for one
+/// reason: a pass has been seen. `Unstable` is the weaker case — the pass did
+/// not reproduce — but a halt cannot tell a reproducible pass from a flaky one,
+/// and stopping a revision on a flake is what the confirmation re-run (#859)
+/// exists to catch downstream.
+#[must_use]
+pub fn for_revision(armed: &Option<Arc<FlipHalt>>, tracked: FlipState) -> Option<Arc<FlipHalt>> {
+    if tracked != FlipState::Failing {
+        return None;
+    }
+    armed.as_ref().and_then(FlipHalt::unfired)
+}
+
 impl TurnHalt for FlipHalt {
     fn halt_reason(&self) -> Option<String> {
         self.is_flipped().then(|| {
@@ -260,4 +294,54 @@ mod tests {
         let unrelated = serde_json::json!({"path": "src/lib.rs"});
         assert_eq!(command_of(&unrelated), None);
     }
+
+    /// **Witness (#1793).** A revise turn carries the halt while the tracked
+    /// command is still red.
+    ///
+    /// Revise turns passed `None` unconditionally, so the measured early stop
+    /// never applied to a revision — free wall clock and tokens after the work
+    /// was done.
+    #[test]
+    fn a_revision_inherits_the_halt_while_the_tracked_command_is_still_red() {
+        assert!(
+            for_revision(&None, FlipState::Failing).is_none(),
+            "nothing to watch stays nothing to watch"
+        );
+
+        let armed = Some(Arc::new(FlipHalt::new("pytest -q")));
+        let inherited = for_revision(&armed, FlipState::Failing)
+            .expect("a red command's halt carries into the revision");
+        assert_eq!(inherited.tracked(), "pytest -q");
+        assert!(Arc::ptr_eq(&inherited, armed.as_ref().unwrap()));
+    }
+
+    /// The half that would silently break revision if it were wrong: a
+    /// revision entered on an ALREADY-GREEN command gets no halt.
+    ///
+    /// A verifier can reject a candidate whose witness flipped — the test went
+    /// green but the review found the change wanting. Arming there would end
+    /// the revision the first time the model re-ran that passing test, before
+    /// it addressed anything the verifier objected to. `Unstable` is excluded
+    /// for the same reason with less margin: a pass was seen, and a halt
+    /// cannot tell a reproducible one from a flake.
+    #[test]
+    fn a_revision_on_a_green_or_flaky_command_is_never_halted() {
+        let armed = Some(Arc::new(FlipHalt::new("pytest -q")));
+        for state in [FlipState::Flipped, FlipState::Unstable, FlipState::None] {
+            assert!(
+                for_revision(&armed, state).is_none(),
+                "{state:?} means the flip is behind the turn, not ahead of it"
+            );
+        }
+
+        // The sticky latch is refused even when the oracle still reads red: it
+        // survives the turn that fired it.
+        assert!(
+            armed
+                .as_ref()
+                .unwrap()
+                .observe("pytest -q", "ok\n[exit code: 0]")
+        );
+        assert!(for_revision(&armed, FlipState::Failing).is_none());
+    }
 }
@@ -83,10 +83,14 @@ fn management_system_block(role: ModelCallRole) -> Option<String> {
         // adopt the split these arms move to `Some(...)` and the roles join
         // the parity witness automatically.
         ModelCallRole::Plan | ModelCallRole::PlanRepair => None,
-        // Never dispatched through the management chokepoint. `Research`
-        // (#1778) rides the sub-agent primitive — its system prompt travels
-        // on the `SubAgentSpec`, not through `metered_raw_call`.
+        // Never dispatched through the management chokepoint.
+        //
+        // `Research` (#1778) joins them because its calls are sub-agents run
+        // through `Engine::run_sub_agent_with_sender`, not `metered_raw_call`
+        // — the same reason the witness roles are here. It also carries no
+        // system block of its own to declare.
         ModelCallRole::Unknown
+        | ModelCallRole::Research
         | ModelCallRole::WitnessAuthor
         | ModelCallRole::WitnessRepair
         | ModelCallRole::AgentAuthor

@@ -53,7 +53,9 @@ use stella_core::hooks::{HookRunner, Hooks};
 use stella_core::receipts::RECEIPT_SEQ_ALLOCATED_BASE;
 use stella_core::retry::{RetryPolicy, Sleeper};
 use stella_core::router::FallbackInfo;
-use stella_core::{AbortKind, BudgetGuard, Engine, EngineConfig, EventSender, Router, TurnOutcome};
+use stella_core::{
+    AbortKind, BudgetGuard, CalibrationMap, Engine, EngineConfig, EventSender, Router, TurnOutcome,
+};
 use stella_protocol::{
     AgentEvent, CompletionMessage, LadderRung, LadderSnapshot, MessageRole, ModelCallRole,
     ModelRef, OracleObservation, ProofStep, ProofTree, Provider, Role, StageKind, VerdictEvidence,
@@ -107,6 +109,7 @@ use crate::witness::{
     validate_witness_identity, validate_witness_invocation, witness_identity_matches,
     witness_prompt, witness_repair_prompt,
 };
+mod attachments;
 mod authored;
 mod candidate_result;
 mod disclosure;
@@ -881,6 +884,9 @@ pub struct Pipeline<'a> {
     /// engine this pipeline builds and consulted before every management
     /// call, so a paused pipeline-driven worker parks instead of spending.
     turn_gate: Option<&'a dyn stella_core::ports::TurnGate>,
+    /// Caller-owned token-drift model ([`Pipeline::with_calibration`]), lent to
+    /// every engine this pipeline builds. `None` leaves estimation uncorrected.
+    calibration: Option<&'a CalibrationMap>,
     events: EventSender,
     config: PipelineConfig,
     configured_test: Result<Option<TestInvocation>, crate::witness::TestInvocationError>,
@@ -941,6 +947,7 @@ impl<'a> Pipeline<'a> {
             mcp_prefetch: ports.mcp_prefetch,
             steering: ports.steering,
             turn_gate: None,
+            calibration: None,
             events: events.into(),
             config,
             configured_test,
@@ -951,22 +958,6 @@ impl<'a> Pipeline<'a> {
         }
     }
 
-    /// Attach a boundary pause gate. Every engine the pipeline builds — the
-    /// worker's execute/revise turns and the witness author's — parks at its
-    /// step boundaries while the gate holds, and every management call
-    /// (triage, verifier, guidance) parks before dispatch: the same safe
-    /// boundary as budget aborts, never mid-tool.
-    ///
-    /// This is the seam that lets a supervisor's pause reach a
-    /// pipeline-driven worker at all. Without it only the raw step-loop path
-    /// held a gate, so `Fleet::pause_task` on a pipeline worker silently did
-    /// nothing — the named follow-up in `fleet_cmd`.
-    #[must_use]
-    pub fn with_turn_gate(mut self, gate: &'a dyn stella_core::ports::TurnGate) -> Self {
-        self.turn_gate = Some(gate);
-        self
-    }
-
     /// Drive one prompt through the full staged flow. `messages` is the
     /// caller-owned history: seed it with the stable system prefix (the cached
     /// prompt prefix, L-E8); the pipeline appends the volatile recall+goal
@@ -1598,9 +1589,7 @@ impl<'a> Pipeline<'a> {
             if let Some((hooks, runner)) = self.hooks {
                 engine = engine.with_hooks(hooks, runner);
             }
-            if let Some(gate) = self.turn_gate {
-                engine = engine.with_gate(gate);
-            }
+            engine = self.attach(engine);
             let view = fan.as_ref().map(|fan| fan.candidate());
             if let Some(view) = view.as_ref() {
                 engine = engine.with_steering(view);
@@ -2849,7 +2838,7 @@ impl<'a> Pipeline<'a> {
                 &mut state.messages,
                 spend.budget,
                 &mut state.signals,
-                state.flip_halt.as_ref().and_then(FlipHalt::unfired),
+                crate::flip_halt::for_revision(&state.flip_halt, state.oracle.state()),
             )
             .await
         {

@@ -0,0 +1,74 @@
+//! The optional things a caller bolts onto a [`Pipeline`] after construction,
+//! and the rule that every engine the pipeline builds inherits them.
+//!
+//! Separate from the ports in [`PipelinePorts`] because these are not
+//! capabilities the pipeline needs in order to run — it runs without any of
+//! them. They are host-owned state that outlives a turn (a supervisor's pause
+//! flag, the session's accumulated token-drift model), which is why each is a
+//! borrow with the pipeline's own lifetime rather than an owned field: the
+//! caller keeps it across turns and lends it in, mirroring how `BudgetGuard`
+//! and `CalibrationMap` are owned above the engine rather than by it.
+//!
+//! Split out of `pipeline.rs` for the reason everything is: it is a
+//! grandfathered god file closed to growth, and this is a coherent seam to
+//! take with it.
+
+use super::*;
+
+impl<'a> Pipeline<'a> {
+    /// Attach a boundary pause gate. Every engine the pipeline builds — the
+    /// worker's execute/revise turns and the witness author's — parks at its
+    /// step boundaries while the gate holds, and every management call
+    /// (triage, verifier, guidance) parks before dispatch: the same safe
+    /// boundary as budget aborts, never mid-tool.
+    ///
+    /// This is the seam that lets a supervisor's pause reach a
+    /// pipeline-driven worker at all. Without it only the raw step-loop path
+    /// held a gate, so `Fleet::pause_task` on a pipeline worker silently did
+    /// nothing — the named follow-up in `fleet_cmd`.
+    #[must_use]
+    pub fn with_turn_gate(mut self, gate: &'a dyn stella_core::ports::TurnGate) -> Self {
+        self.turn_gate = Some(gate);
+        self
+    }
+
+    /// Attach the caller's token-drift calibration, so the engines this
+    /// pipeline builds size their compaction budget against what this model's
+    /// tokenizer actually charged rather than against the raw estimate
+    /// (#1595).
+    ///
+    /// Five of `stella-cli`'s seven assembly sites seeded a `CalibrationMap`
+    /// and handed it to the engine; the two that did not were the **default
+    /// `stella run`** path and fleet workers, because a pipeline had no way to
+    /// accept one. So the most-used path in the product was the one estimating
+    /// worst — and not only by losing the seed: with no map at all the
+    /// correction is inert for the whole run, so the drift those turns
+    /// measured was never applied to them either.
+    ///
+    /// Borrowed rather than owned because `CalibrationMap` is deliberately not
+    /// `Clone` — the caller owns it across turns and the engine reads it
+    /// through `&self` (see its type doc). An owned field here would also cost
+    /// `PipelineConfig` its `Clone`, which many call sites rely on.
+    #[must_use]
+    pub fn with_calibration(mut self, calibration: &'a CalibrationMap) -> Self {
+        self.calibration = Some(calibration);
+        self
+    }
+
+    /// Bolt this pipeline's attachments onto one freshly-built engine.
+    ///
+    /// Every engine the pipeline constructs goes through here, which is the
+    /// point: the worker's turns, the best-of-N children and the witness
+    /// author are all engines, and attaching to some but not others is how
+    /// #1595 happened one level up. A new attachment is added once, here,
+    /// rather than at each construction site.
+    pub(super) fn attach<'e>(&'e self, mut engine: Engine<'e>) -> Engine<'e> {
+        if let Some(gate) = self.turn_gate {
+            engine = engine.with_gate(gate);
+        }
+        if let Some(calibration) = self.calibration {
+            engine = engine.with_calibration(calibration);
+        }
+        engine
+    }
+}
@@ -293,9 +293,7 @@ impl<'a> Pipeline<'a> {
         if let Some((hooks, runner)) = self.hooks {
             engine = engine.with_hooks(hooks, surface.hook_runner.unwrap_or(runner));
         }
-        if let Some(gate) = self.turn_gate {
-            engine = engine.with_gate(gate);
-        }
+        engine = self.attach(engine);
         let view = fan.map(SteeringFanOut::candidate);
         if let Some(view) = view.as_ref() {
             engine = engine.with_steering(view);

@@ -76,6 +76,7 @@ impl Pipeline<'_> {
         if let Some((hooks, runner)) = self.hooks {
             engine = engine.with_hooks(hooks, runner);
         }
+        engine = self.attach(engine);
 
         let width = u32::try_from(questions.len()).unwrap_or(u32::MAX);
         let fan = FanOutBudget::new(*budget, width);

@@ -41,7 +41,7 @@ impl Pipeline<'_> {
                     research,
                     &repo_structure,
                     revision.as_deref(),
-                    &mut spend,
+                    &mut Spend { budget, total },
                 )
                 .await
             {

@@ -16,6 +16,11 @@ use crate::triage::TaskClass;
 /// mutation must land there — an owned copy would need a write-back on each
 /// of `run`'s early returns, and one missed return is a silently vanished
 /// spend.
+///
+/// A signature still threading the loose pair adopts this the next time it
+/// gains an input: `plan_stage` did so when #1778's `research` argument pushed
+/// the loose form past clippy's arity limit. The bundle is the fix rather than
+/// an `#[allow]` precisely because the pair was never two things.
 pub(super) struct Spend<'a> {
     /// Gates each paid call; consulted between model calls only (invariant #6).
     pub(super) budget: &'a mut BudgetGuard,

@@ -3,6 +3,7 @@
 //! private surface (`CandidateSurface`, `Pipeline::gather_diff`, ...)
 //! stays reachable via `super::*`.
 
+mod calibration;
 mod conversational_window;
 mod management_accounting;
 mod telemetry;