diff --git a/crates/stella-pipeline/src/pipeline.rs b/crates/stella-pipeline/src/pipeline.rs
index 01e4dea13..b4d546b62 100644
--- a/crates/stella-pipeline/src/pipeline.rs
+++ b/crates/stella-pipeline/src/pipeline.rs
@@ -100,7 +100,7 @@ use crate::verify::{
 use crate::witness::airlock::{
     DisclosureGrain, FailureFingerprint, SealedFailure, grain_for_repeats, redact, scrub,
 };
-use crate::witness::warrant::warrant;
+use crate::witness::warrant::{ChangeSignals, warrant};
 use crate::witness::{
     Witness, parse_test_invocation, parse_witness_command, validate_witness_artifact,
     validate_witness_identity, validate_witness_invocation, witness_identity_matches,
@@ -117,6 +117,7 @@ mod repair_gate;
 mod run_error;
 mod scope_stage;
 mod stage_budget;
+mod task_frame;
 mod verifier_stage;
 mod verify_probes;
 use verify_probes::DiffProbe;
@@ -126,7 +127,8 @@ use fanout_stage::SerialCreates;
 use raw_usage::{RawCall, RawCallError};
 pub use run_error::{PipelineError, PipelineRunError};
 use run_error::{RoleResolveError, WitnessAuthorIndependence};
-use stage_budget::{PipelineBudgetAbort, budget_abort};
+use stage_budget::{PipelineBudgetAbort, Spend, budget_abort};
+use task_frame::TaskFrame;
 use witness_stage::{BoundHookRunner, WitnessAuthoring};
 /// Make a diff that verification hands downstream *incapable of lying*.
 ///
@@ -627,20 +629,13 @@ struct ResolvedRole<'a> {
 struct CandidateState {
     messages: Vec<CompletionMessage>,
     final_text: String,
-    /// `FileChange` events observed across this candidate's engine turns —
-    /// one half of the zero-diff guard's "touched files" signal (L-E2).
-    file_changes: u32,
-    /// Tool calls this candidate dispatched whose tool is not advertised
-    /// `read_only` — the ladder's only evidence-of-absence channel
-    /// ([`crate::verify::LadderInputs::mutating_actions`]). Counted off the
-    /// engine's own `ToolStart` stream rather than probed for, which is what
-    /// lets `0` mean "never tried" instead of "could not tell".
-    mutating_actions: u32,
-    /// The subset of `mutating_actions` whose effects the diff cannot be
-    /// trusted to account for (`witness::warrant::diff_accountable_mutator`)
-    /// — the warrant's guard against waiving a witness on a diff that is not
-    /// the whole change.
-    opaque_actions: u32,
+    /// What this candidate's engine turns did, accumulated in the exact form
+    /// [`warrant`] and the ladder read it — each field's meaning is documented
+    /// on [`ChangeSignals`] itself. One typed field rather than three loose
+    /// `u32`s so the counts can never be transposed on their way to the
+    /// warrant (the #1701 recurrence a projection method used to guard
+    /// against by hand).
+    signals: ChangeSignals,
     /// Ends the execute turn as soon as the tracked test goes fail→pass.
     ///
     /// `None` whenever there is nothing to watch: no configured test command,
@@ -1055,6 +1050,15 @@ impl<'a> Pipeline<'a> {
         // --- 5. Witness + execute + verify (single-shot or best-of-N). ------
         let n = self.config.candidate_count();
         let base_messages = messages.clone();
+        // The one frame every candidate stage below reads (#1809). Built here
+        // because this is where its last field settles: nothing after this
+        // point changes the goal, the staged prefix, the plan, or the class.
+        let frame = TaskFrame {
+            goal,
+            base_messages: &base_messages,
+            plan: plan.as_deref(),
+            assessment,
+        };
         // Decided above, before the user message was assembled (the worker's
         // test-first contract keys off it) and before this single-shot/
         // best-of-N split, because an authored witness is the *only* reason a
@@ -1102,14 +1106,13 @@ impl<'a> Pipeline<'a> {
             let worker_model_label = worker.model_ref.to_string();
             let mut single = self
                 .run_shared_candidates(
-                    goal,
-                    &base_messages,
-                    plan.as_deref(),
-                    assessment,
+                    frame,
                     &worker,
                     1,
-                    budget,
-                    &mut total_cost,
+                    &mut Spend {
+                        budget: &mut *budget,
+                        total: &mut total_cost,
+                    },
                 )
                 .await;
             let ran = executed_count(&single);
@@ -1120,15 +1123,14 @@ impl<'a> Pipeline<'a> {
         } else {
             match self
                 .run_best_of_n(
-                    goal,
-                    &base_messages,
-                    plan.as_deref(),
-                    assessment,
+                    frame,
                     n,
                     &frames,
                     authored_witness,
-                    budget,
-                    &mut total_cost,
+                    &mut Spend {
+                        budget: &mut *budget,
+                        total: &mut total_cost,
+                    },
                 )
                 .await
             {
@@ -1148,12 +1150,11 @@ impl<'a> Pipeline<'a> {
         let (best, candidates_run) = if best.aborted.is_some() && best.degradable {
             match self
                 .degrade_to_bare_execution(
-                    goal,
-                    &base_messages,
-                    plan.as_deref(),
-                    assessment,
-                    budget,
-                    &mut total_cost,
+                    frame,
+                    &mut Spend {
+                        budget: &mut *budget,
+                        total: &mut total_cost,
+                    },
                 )
                 .await
             {
@@ -1577,15 +1578,10 @@ impl<'a> Pipeline<'a> {
     /// work. Returns `None` only when there is no resolvable worker provider
     /// (a true impossibility, not a degradable setup failure), in which case
     /// the caller keeps the original setup abort.
-    #[allow(clippy::too_many_arguments)]
     async fn degrade_to_bare_execution(
         &self,
-        goal: &str,
-        base_messages: &[CompletionMessage],
-        plan: Option<&[PlanStep]>,
-        assessment: TaskAssessment,
-        budget: &mut BudgetGuard,
-        total: &mut f64,
+        frame: TaskFrame<'_>,
+        spend: &mut Spend<'_>,
     ) -> Option<CandidateResult> {
         let worker = self.resolve_provider(Role::Worker).ok()?;
         if let Some(fallback) = &worker.fallback {
@@ -1596,34 +1592,20 @@ impl<'a> Pipeline<'a> {
              working tree so the turn still does the work it was asked to"
                 .to_string(),
         );
-        self.run_shared_candidates(
-            goal,
-            base_messages,
-            plan,
-            assessment,
-            &worker,
-            1,
-            budget,
-            total,
-        )
-        .await
-        .pop()
+        self.run_shared_candidates(frame, &worker, 1, spend)
+            .await
+            .pop()
     }
 
     /// Run `n` candidates sequentially over the session ports (the real
     /// working tree): the single-shot path, and the shared-tree degradation
     /// of best-of-N when no [`CandidateWorkspacePort`] is wired.
-    #[allow(clippy::too_many_arguments)]
     async fn run_shared_candidates(
         &self,
-        goal: &str,
-        base_messages: &[CompletionMessage],
-        plan: Option<&[PlanStep]>,
-        assessment: TaskAssessment,
+        frame: TaskFrame<'_>,
         worker: &ResolvedRole<'a>,
         n: u32,
-        budget: &mut BudgetGuard,
-        total: &mut f64,
+        spend: &mut Spend<'_>,
     ) -> Vec<CandidateResult> {
         let surface = CandidateSurface {
             diagnostics: self.diagnostics,
@@ -1660,18 +1642,11 @@ impl<'a> Pipeline<'a> {
             }
             results.push(
                 self.run_candidate(
-                    goal,
-                    base_messages,
-                    plan,
-                    assessment,
+                    frame,
                     // A shared-tree run has no workspace to graft into and no
                     // pristine snapshot to author blind in, so it never buys a
                     // witness — exactly as before, when it was passed `None`.
-                    None,
-                    &engine,
-                    surface,
-                    budget,
-                    total,
+                    None, &engine, surface, spend,
                 )
                 .await,
             );
@@ -1686,27 +1661,25 @@ impl<'a> Pipeline<'a> {
     /// unconditional (success or failure) with one deliberate exception: a
     /// winner whose adoption failed keeps its workspace, named in the error,
     /// so completed work is never destroyed.
-    #[allow(clippy::too_many_arguments)]
     async fn run_best_of_n(
         &self,
-        goal: &str,
-        base_messages: &[CompletionMessage],
-        plan: Option<&[PlanStep]>,
-        assessment: TaskAssessment,
+        frame: TaskFrame<'_>,
         n: u32,
         frames: &[RecalledFrame],
         author_witness: bool,
-        budget: &mut BudgetGuard,
-        total: &mut f64,
+        spend: &mut Spend<'_>,
     ) -> Result<(CandidateResult, Option<String>, u32), PipelineError> {
         // Orchestrator pre-fetch (issue #248) — see `crate::mcp_prefetch::fold`.
-        let prefetched = crate::mcp_prefetch::fold(self.mcp_prefetch, n, base_messages).await;
-        let base_messages: &[CompletionMessage] = prefetched.as_deref().unwrap_or(base_messages);
+        let prefetched = crate::mcp_prefetch::fold(self.mcp_prefetch, n, frame.base_messages).await;
+        let frame = TaskFrame {
+            base_messages: prefetched.as_deref().unwrap_or(frame.base_messages),
+            ..frame
+        };
         let Some(port) = self.candidate_workspaces else {
             if author_witness {
                 return Ok((
                     CandidateResult::setup_aborted(
-                        base_messages.to_vec(),
+                        frame.base_messages.to_vec(),
                         "authored witness requires candidate isolation, but no candidate \
                          workspace port is available"
                             .to_string(),
@@ -1733,18 +1706,7 @@ impl<'a> Pipeline<'a> {
                 self.emit_fallback(fallback);
             }
             let label = worker.model_ref.to_string();
-            let candidates = self
-                .run_shared_candidates(
-                    goal,
-                    base_messages,
-                    plan,
-                    assessment,
-                    &worker,
-                    n,
-                    budget,
-                    total,
-                )
-                .await;
+            let candidates = self.run_shared_candidates(frame, &worker, n, spend).await;
             let best_idx = best_index(&candidates);
             let ran = executed_count(&candidates);
             self.emit_text(candidate_winner_notice(best_idx, n, ran));
@@ -1830,18 +1792,7 @@ impl<'a> Pipeline<'a> {
             frames,
         });
         let candidates = self
-            .dispatch_isolated_candidates(
-                goal,
-                base_messages,
-                plan,
-                assessment,
-                &worker,
-                authoring,
-                &workspaces,
-                width,
-                budget,
-                total,
-            )
+            .dispatch_isolated_candidates(frame, &worker, authoring, &workspaces, width, spend)
             .await;
 
         let best_idx = best_index(&candidates);
@@ -1936,18 +1887,13 @@ impl<'a> Pipeline<'a> {
 
     // Stages: execute + verify + revise (one candidate)
 
-    #[allow(clippy::too_many_arguments)]
     async fn run_candidate(
         &self,
-        goal: &str,
-        base_messages: &[CompletionMessage],
-        plan: Option<&[PlanStep]>,
-        assessment: TaskAssessment,
+        frame: TaskFrame<'_>,
         authoring: Option<WitnessAuthoring<'_>>,
         engine: &Engine<'_>,
         surface: CandidateSurface<'_>,
-        budget: &mut BudgetGuard,
-        total: &mut f64,
+        spend: &mut Spend<'_>,
     ) -> CandidateResult {
         // Flip oracle: for classes we always verify, take a pre-execute
         // baseline of the test command so a later pass counts as a genuine
@@ -1968,7 +1914,7 @@ impl<'a> Pipeline<'a> {
         // that was already green cannot flip, so arming on it would hand the
         // engine a stop signal for work the turn never did.
         let mut flip_halt: Option<Arc<FlipHalt>> = None;
-        if assessment.class.verifies_unconditionally()
+        if frame.assessment.class.verifies_unconditionally()
             && let Some(cmd) = self.effective_test_command(None)
         {
             let pre = self.run_test_observed(surface.tests, cmd.invocation).await;
@@ -2010,7 +1956,7 @@ impl<'a> Pipeline<'a> {
         // classes that verify: a fast-submit needs a flip, so where no flip
         // is possible the snapshot would be spend without a consumer.
         let lint_baseline = if surface.workspace.is_none()
-            && assessment.class.verifies_unconditionally()
+            && frame.assessment.class.verifies_unconditionally()
             && (self.effective_test_command(None).is_some() || self.config.witness_writer)
         {
             match surface.lint {
@@ -2029,11 +1975,12 @@ impl<'a> Pipeline<'a> {
         let untracked_before = surface.repo_status.untracked_fingerprints().await;
 
         let mut state = CandidateState {
-            messages: candidate_narration::messages_rooted_at(base_messages, surface.workspace),
+            messages: candidate_narration::messages_rooted_at(
+                frame.base_messages,
+                surface.workspace,
+            ),
             final_text: String::new(),
-            file_changes: 0,
-            mutating_actions: 0,
-            opaque_actions: 0,
+            signals: ChangeSignals::default(),
             flip_halt,
             oracle,
             oracle_trace,
@@ -2059,7 +2006,7 @@ impl<'a> Pipeline<'a> {
         };
 
         if let Err(abort) = self
-            .execute_plan(plan, engine, budget, total, &mut state)
+            .execute_plan(frame.plan, engine, spend, &mut state)
             .await
         {
             return CandidateResult::turn_aborted(state.messages, abort);
@@ -2078,10 +2025,10 @@ impl<'a> Pipeline<'a> {
         // over someone else's edit.
         let probe = self.gather_diff(surface, &state.untracked_before).await;
         self.absorb_probe(&mut state, probe);
-        let files_touched = state.file_changes > 0
-            || (state.mutating_actions > 0 && !state.diff_text.trim().is_empty());
-        let should_verify = assessment.class.verifies_unconditionally()
-            || (assessment.class == TaskClass::SimpleLookup && files_touched);
+        let files_touched = state.signals.file_changes > 0
+            || (state.signals.mutating_actions > 0 && !state.diff_text.trim().is_empty());
+        let should_verify = frame.assessment.class.verifies_unconditionally()
+            || (frame.assessment.class == TaskClass::SimpleLookup && files_touched);
         if !should_verify {
             // A clean lookup: nothing to verify.
             return state.into_unverified();
@@ -2095,7 +2042,7 @@ impl<'a> Pipeline<'a> {
         // early when a test IS required. Emitting from either would make the
         // rail's first row appear only on some runs, which is the failure this
         // whole surface exists to end.
-        let warrant = warrant(&state.diff_text, state.change_signals());
+        let warrant = warrant(&state.diff_text, state.signals);
         self.emit_proof(ProofStep::Warrant {
             required: warrant.is_required(),
             reason: warrant.reason().map(|r| r.sentence().to_string()),
@@ -2106,7 +2053,7 @@ impl<'a> Pipeline<'a> {
         // already happened, so the diff is evidence rather than a prediction —
         // which is the whole reason authoring waits until here.
         let witness = match self
-            .witness_on_demand(goal, authoring, surface, &mut state, budget, total)
+            .witness_on_demand(frame.goal, authoring, surface, &mut state, spend)
             .await
         {
             Ok(witness) => witness,
@@ -2117,17 +2064,8 @@ impl<'a> Pipeline<'a> {
             }
         };
 
-        self.verify_candidate(
-            goal,
-            assessment,
-            witness.as_ref(),
-            engine,
-            surface,
-            budget,
-            total,
-            state,
-        )
-        .await
+        self.verify_candidate(frame, witness.as_ref(), engine, surface, spend, state)
+            .await
     }
 
     /// Execute stage: one turn for simple/single-task; one turn per plan step
@@ -2138,8 +2076,7 @@ impl<'a> Pipeline<'a> {
         &self,
         plan: Option<&[PlanStep]>,
         engine: &Engine<'_>,
-        budget: &mut BudgetGuard,
-        total: &mut f64,
+        spend: &mut Spend<'_>,
         state: &mut CandidateState,
     ) -> Result<(), TurnAbort> {
         self.emit(AgentEvent::Stage {
@@ -2153,24 +2090,22 @@ impl<'a> Pipeline<'a> {
                 .run_engine_turn(
                     engine,
                     &mut state.messages,
-                    budget,
-                    &mut state.file_changes,
-                    &mut state.mutating_actions,
-                    &mut state.opaque_actions,
+                    spend.budget,
+                    &mut state.signals,
                     state.flip_halt.clone(),
                 )
                 .await
             {
                 TurnOutcome::Completed { text, cost_usd } => {
                     state.final_text = text;
-                    *total += cost_usd;
+                    *spend.total += cost_usd;
                 }
                 TurnOutcome::Aborted {
                     reason,
                     kind,
                     cost_usd,
                 } => {
-                    *total += cost_usd;
+                    *spend.total += cost_usd;
                     return Err(TurnAbort { reason, kind });
                 }
             }
@@ -2188,16 +2123,14 @@ impl<'a> Pipeline<'a> {
                     .run_engine_turn(
                         engine,
                         &mut state.messages,
-                        budget,
-                        &mut state.file_changes,
-                        &mut state.mutating_actions,
-                        &mut state.opaque_actions,
+                        spend.budget,
+                        &mut state.signals,
                         state.flip_halt.clone(),
                     )
                     .await
                 {
                     TurnOutcome::Completed { text, cost_usd } => {
-                        *total += cost_usd;
+                        *spend.total += cost_usd;
                         // #1702: a worker that declares the whole goal done
                         // ends the walk — the remaining steps could only
                         // re-confirm finished work, and a false declaration
@@ -2213,7 +2146,7 @@ impl<'a> Pipeline<'a> {
                         kind,
                         cost_usd,
                     } => {
-                        *total += cost_usd;
+                        *spend.total += cost_usd;
                         return Err(TurnAbort { reason, kind });
                     }
                 }
@@ -2227,16 +2160,13 @@ impl<'a> Pipeline<'a> {
     /// finish with a verdict, escalate to the model verifier, or spend one of
     /// `max_revisions` on a revise pass and re-observe. Owns `state` because
     /// every exit moves it into the returned [`CandidateResult`].
-    #[allow(clippy::too_many_arguments)]
     async fn verify_candidate(
         &self,
-        goal: &str,
-        assessment: TaskAssessment,
+        frame: TaskFrame<'_>,
         witness: Option<&Witness>,
         engine: &Engine<'_>,
         surface: CandidateSurface<'_>,
-        budget: &mut BudgetGuard,
-        total: &mut f64,
+        spend: &mut Spend<'_>,
         mut state: CandidateState,
     ) -> CandidateResult {
         self.emit(AgentEvent::Stage {
@@ -2244,7 +2174,7 @@ impl<'a> Pipeline<'a> {
         });
         let effective_cmd = self.effective_test_command(witness);
         let witness_paths = Self::witness_paths(witness);
-        let meter = repair_gate::RepairMeter::start(*total);
+        let meter = repair_gate::RepairMeter::start(*spend.total);
         loop {
             if let Some(workspace) = surface.workspace {
                 if let Err(error) = workspace.seal().await {
@@ -2318,8 +2248,8 @@ impl<'a> Pipeline<'a> {
                 diff_lines: state.diff_lines,
                 diff_budget: self.config.diff_budget_lines,
                 diff_available: state.diff_available,
-                file_change_events: state.file_changes,
-                mutating_actions: state.mutating_actions,
+                file_change_events: state.signals.file_changes,
+                mutating_actions: state.signals.mutating_actions,
                 // Filled by the pre-submit audit below (#861, #870, #1291) —
                 // the lint, coverage and mutation probes only run when a
                 // fast-submit is imminent.
@@ -2536,7 +2466,7 @@ impl<'a> Pipeline<'a> {
                         passed: false,
                         evidence: evidence.clone(),
                     });
-                    if !self.affords_repair(&state, &meter, *total, budget) {
+                    if !self.affords_repair(&state, &meter, *spend.total, spend.budget) {
                         return state.into_verified(
                             false,
                             &evidence,
@@ -2547,9 +2477,8 @@ impl<'a> Pipeline<'a> {
                         .revise_candidate(
                             engine,
                             surface,
-                            budget,
                             NOTHING_ATTEMPTED_NUDGE,
-                            total,
+                            spend,
                             &mut state,
                         )
                         .await
@@ -2605,7 +2534,7 @@ impl<'a> Pipeline<'a> {
                         passed: false,
                         evidence: evidence.clone(),
                     });
-                    if !self.affords_repair(&state, &meter, *total, budget) {
+                    if !self.affords_repair(&state, &meter, *spend.total, spend.budget) {
                         return state.into_verified(
                             false,
                             &evidence,
@@ -2630,7 +2559,7 @@ impl<'a> Pipeline<'a> {
                         let stripped =
                             crate::verify::strip_witness_hunks(&state.diff_text, &witness_paths);
                         let prompt = guidance_prompt(
-                            goal,
+                            frame.goal,
                             &stripped.diff,
                             &evidence.summary,
                             // Guidance never carries a delta baseline: its
@@ -2642,7 +2571,10 @@ impl<'a> Pipeline<'a> {
                                 previous: None,
                             },
                         );
-                        match self.verifier_guidance(prompt, budget, total).await {
+                        match self
+                            .verifier_guidance(prompt, spend.budget, spend.total)
+                            .await
+                        {
                             Ok(Some(guidance)) => {
                                 if let Some(text) =
                                     self.airlock_forward(&guidance, "distress_guidance", &sealed)
@@ -2663,7 +2595,7 @@ impl<'a> Pipeline<'a> {
                         }
                     }
                     if let Err(abort) = self
-                        .revise_candidate(engine, surface, budget, &reason, total, &mut state)
+                        .revise_candidate(engine, surface, &reason, spend, &mut state)
                         .await
                     {
                         return CandidateResult::turn_aborted(state.messages, abort);
@@ -2678,7 +2610,8 @@ impl<'a> Pipeline<'a> {
                 // from a behavioral change nothing proved (§7.1:
                 // predict-then-commit is the bug).
                 LadderDecision::ModelVerdict
-                    if !assessment.wants_verifier() && Self::verifier_waiver_stands(&state) =>
+                    if !frame.assessment.wants_verifier()
+                        && Self::verifier_waiver_stands(&state) =>
                 {
                     let evidence = self.waived_completion(&snapshot);
                     return state.into_verified(
@@ -2723,7 +2656,7 @@ impl<'a> Pipeline<'a> {
                     // appended note steers the worker better than a fresh
                     // reading of an unchanged tree ever did.
                     let inputs_digest =
-                        verdict_inputs_digest(goal, &stripped.diff, &evidence_summary);
+                        verdict_inputs_digest(frame.goal, &stripped.diff, &evidence_summary);
                     let cached = state
                         .last_verdict
                         .as_ref()
@@ -2745,7 +2678,7 @@ impl<'a> Pipeline<'a> {
                             // the text a verdict actually read — so both
                             // sides of the comparison are the same shape.
                             let prompt = verifier_prompt(
-                                goal,
+                                frame.goal,
                                 &stripped.diff,
                                 &evidence_summary,
                                 &DiffContext {
@@ -2753,7 +2686,10 @@ impl<'a> Pipeline<'a> {
                                     previous: state.last_verdict_diff.as_deref(),
                                 },
                             );
-                            match self.verifier(prompt, &inputs, budget, total).await {
+                            match self
+                                .verifier(prompt, &inputs, spend.budget, spend.total)
+                                .await
+                            {
                                 Ok(verdict) => {
                                     // Only pin a real model verdict for reuse. A
                                     // heuristic fallback is a transient-outage
@@ -2823,9 +2759,7 @@ impl<'a> Pipeline<'a> {
                                 state.evidence_demands += 1;
                                 let ask = crate::verify::evidence_demand_prompt(cmd.command);
                                 if let Err(abort) = self
-                                    .revise_candidate(
-                                        engine, surface, budget, &ask, total, &mut state,
-                                    )
+                                    .revise_candidate(engine, surface, &ask, spend, &mut state)
                                     .await
                                 {
                                     return CandidateResult::turn_aborted(state.messages, abort);
@@ -2857,7 +2791,7 @@ impl<'a> Pipeline<'a> {
                             score_from_verification(false, Some(true)),
                         );
                     }
-                    if !self.affords_repair(&state, &meter, *total, budget) {
+                    if !self.affords_repair(&state, &meter, *spend.total, spend.budget) {
                         return state.into_verified(
                             false,
                             &evidence,
@@ -2873,7 +2807,7 @@ impl<'a> Pipeline<'a> {
                         .map(|text| crate::verify::bound_forwarded_reasoning(&text))
                         .unwrap_or_else(|| redact(&sealed, DisclosureGrain::Symptom).message());
                     if let Err(abort) = self
-                        .revise_candidate(engine, surface, budget, &feedback, total, &mut state)
+                        .revise_candidate(engine, surface, &feedback, spend, &mut state)
                         .await
                     {
                         return CandidateResult::turn_aborted(state.messages, abort);
@@ -2910,25 +2844,12 @@ impl<'a> Pipeline<'a> {
         &self,
         engine: &Engine<'_>,
         surface: CandidateSurface<'_>,
-        budget: &mut BudgetGuard,
         reason: &str,
-        total: &mut f64,
+        spend: &mut Spend<'_>,
         state: &mut CandidateState,
     ) -> Result<(), TurnAbort> {
         let probe = self
-            .revise_turn(
-                engine,
-                surface,
-                &mut state.messages,
-                budget,
-                reason,
-                &mut state.file_changes,
-                &mut state.mutating_actions,
-                &mut state.opaque_actions,
-                &mut state.final_text,
-                total,
-                &state.untracked_before,
-            )
+            .revise_turn(engine, surface, reason, spend, state)
             .await?;
         self.absorb_probe(state, probe);
         state.revisions += 1;
@@ -2939,51 +2860,44 @@ impl<'a> Pipeline<'a> {
     /// and re-gather the diff. Emits the `Execute`/`Verify` stage bookends so
     /// the stream shows the revise loop. Returns the fresh `(diff_lines,
     /// diff_text)` on success, or the typed abort on a budget/loop abort.
-    #[allow(clippy::too_many_arguments)]
     async fn revise_turn(
         &self,
         engine: &Engine<'_>,
         surface: CandidateSurface<'_>,
-        messages: &mut Vec<CompletionMessage>,
-        budget: &mut BudgetGuard,
         reason: &str,
-        file_changes: &mut u32,
-        mutating_actions: &mut u32,
-        opaque_actions: &mut u32,
-        final_text: &mut String,
-        total: &mut f64,
-        untracked_before: &HashMap<String, String>,
+        spend: &mut Spend<'_>,
+        state: &mut CandidateState,
     ) -> Result<DiffProbe, TurnAbort> {
-        messages.push(CompletionMessage::user(revision_prompt(reason)));
+        state
+            .messages
+            .push(CompletionMessage::user(revision_prompt(reason)));
         self.emit(AgentEvent::Stage {
             name: StageKind::Execute,
         });
         match self
             .run_engine_turn(
                 engine,
-                messages,
-                budget,
-                file_changes,
-                mutating_actions,
-                opaque_actions,
+                &mut state.messages,
+                spend.budget,
+                &mut state.signals,
                 None,
             )
             .await
         {
             TurnOutcome::Completed { text, cost_usd } => {
-                *final_text = text;
-                *total += cost_usd;
+                state.final_text = text;
+                *spend.total += cost_usd;
             }
             TurnOutcome::Aborted {
                 reason,
                 kind,
                 cost_usd,
             } => {
-                *total += cost_usd;
+                *spend.total += cost_usd;
                 return Err(TurnAbort { reason, kind });
             }
         }
-        let probe = self.gather_diff(surface, untracked_before).await;
+        let probe = self.gather_diff(surface, &state.untracked_before).await;
         self.emit(AgentEvent::Stage {
             name: StageKind::Verify,
         });
@@ -3017,28 +2931,21 @@ impl<'a> Pipeline<'a> {
     /// (a concurrent drain task, not a post-hoc flush — an execute turn can
     /// run tool loops for minutes, and buffering froze the renderer for the
     /// whole turn) **except** the engine's `Stage`/`Complete` (the pipeline
-    /// owns those), tallying `FileChange`s into `file_changes` for the
-    /// zero-diff guard and mutating-capable `ToolStart`s into
-    /// `mutating_actions` for the ladder's no-op rung.
+    /// owns those), tallying `FileChange`s into `signals.file_changes` for
+    /// the zero-diff guard and mutating-capable `ToolStart`s into
+    /// `signals.mutating_actions` for the ladder's no-op rung.
     ///
-    /// The two tallies are deliberately independent. `file_changes` answers
+    /// The tallies are deliberately independent. `file_changes` answers
     /// "did the recorder see the tree change", which a shell redirect defeats;
     /// `mutating_actions` answers "was anything even asked to change", which
     /// nothing can defeat, because it is counted off the calls this pipeline
     /// dispatched rather than off any look at the world.
-    // The eighth parameter (#1798's `opaque_actions`) tipped this over the
-    // lint's threshold; the three tallies are independent by design (doc
-    // comment above), so grouping them to satisfy the count would be shape
-    // without meaning. Tracked for a real revisit in #1809.
-    #[allow(clippy::too_many_arguments)]
     async fn run_engine_turn(
         &self,
         engine: &Engine<'_>,
         messages: &mut Vec<CompletionMessage>,
         budget: &mut BudgetGuard,
-        file_changes: &mut u32,
-        mutating_actions: &mut u32,
-        opaque_actions: &mut u32,
+        signals: &mut ChangeSignals,
         flip_halt: Option<Arc<FlipHalt>>,
     ) -> TurnOutcome {
         // The filtered sender is SYNCHRONOUS on purpose: when the outer
@@ -3161,9 +3068,9 @@ impl<'a> Pipeline<'a> {
         let outcome = engine
             .run_turn_with_sender(messages, budget, &filtered)
             .await;
-        *file_changes += seen_file_changes.load(Ordering::Relaxed);
-        *mutating_actions += seen_mutating.load(Ordering::Relaxed);
-        *opaque_actions += seen_opaque.load(Ordering::Relaxed);
+        signals.file_changes += seen_file_changes.load(Ordering::Relaxed);
+        signals.mutating_actions += seen_mutating.load(Ordering::Relaxed);
+        signals.opaque_actions += seen_opaque.load(Ordering::Relaxed);
         outcome
     }
 
diff --git a/crates/stella-pipeline/src/pipeline/disclosure.rs b/crates/stella-pipeline/src/pipeline/disclosure.rs
index 7917bb079..0cfd3089b 100644
--- a/crates/stella-pipeline/src/pipeline/disclosure.rs
+++ b/crates/stella-pipeline/src/pipeline/disclosure.rs
@@ -15,25 +15,9 @@ use stella_protocol::PolicyKind;
 
 use super::*;
 use crate::witness::airlock::FailureBrief;
-use crate::witness::warrant::{ChangeSignals, warrant};
+use crate::witness::warrant::warrant;
 
 impl CandidateState {
-    /// What this candidate *did*, in the form [`warrant`] reads it.
-    ///
-    /// One accessor rather than four literals at the four call sites, because
-    /// the two counts are both `u32` and mean opposite things: `file_changes`
-    /// is what a channel observed, `mutating_actions` is what this pipeline
-    /// dispatched. Transposing them at one call site would silently restore
-    /// #1701 — the warrant would guard on an observation channel that is
-    /// structurally dark inside a candidate.
-    pub(super) fn change_signals(&self) -> ChangeSignals {
-        ChangeSignals {
-            file_changes: self.file_changes,
-            mutating_actions: self.mutating_actions,
-            opaque_actions: self.opaque_actions,
-        }
-    }
-
     /// Record one deterministic failure and return how many times *this same*
     /// failure has now occurred, counting this one. `1` is a first sighting.
     ///
@@ -131,7 +115,7 @@ impl Pipeline<'_> {
         state: &CandidateState,
         snapshot: &stella_protocol::LadderSnapshot,
     ) -> Option<VerdictEvidence> {
-        let reason = warrant(&state.diff_text, state.change_signals()).reason()?;
+        let reason = warrant(&state.diff_text, state.signals).reason()?;
         if reason.warrants_independent_review() {
             return None;
         }
@@ -187,7 +171,7 @@ impl Pipeline<'_> {
     /// without a verifier). A behavioral diff, a deletion, or anything the diff
     /// machinery could not read keeps its reviewer, whatever triage guessed.
     pub(super) fn verifier_waiver_stands(state: &CandidateState) -> bool {
-        warrant(&state.diff_text, state.change_signals())
+        warrant(&state.diff_text, state.signals)
             .reason()
             .is_some_and(|reason| !reason.warrants_independent_review())
     }
diff --git a/crates/stella-pipeline/src/pipeline/fanout_stage.rs b/crates/stella-pipeline/src/pipeline/fanout_stage.rs
index 902ad4b84..307d0b7c0 100644
--- a/crates/stella-pipeline/src/pipeline/fanout_stage.rs
+++ b/crates/stella-pipeline/src/pipeline/fanout_stage.rs
@@ -141,26 +141,21 @@ impl<'a> Pipeline<'a> {
     /// `budget` and `total` are the turn's, and both are settled before this
     /// returns: `budget` gains every candidate's spend through
     /// [`FanOutBudget`], `total` gains the same money summed by index.
-    #[allow(clippy::too_many_arguments)]
     pub(super) async fn dispatch_isolated_candidates(
         &self,
-        goal: &str,
-        base_messages: &[CompletionMessage],
-        plan: Option<&[PlanStep]>,
-        assessment: TaskAssessment,
+        frame: TaskFrame<'_>,
         worker: &ResolvedRole<'_>,
         authoring: Option<WitnessAuthoring<'_>>,
         workspaces: &[Result<Box<dyn CandidateWorkspace>, String>],
         width: u32,
-        budget: &mut BudgetGuard,
-        total: &mut f64,
+        spend: &mut Spend<'_>,
     ) -> Vec<CandidateResult> {
         // One mirror, one cursor per candidate — see `candidate_steering`.
         // Every candidate observes every steer exactly once whether they run
         // in sequence or together; the cursor is what makes that true, and it
         // was already there.
         let fan = self.steering.map(SteeringFanOut::new);
-        let budgets = FanOutBudget::new(*budget, width);
+        let budgets = FanOutBudget::new(*spend.budget, width);
         // Set for the whole fan-out rather than per candidate: the question a
         // preview has to answer is "is anyone else writing to this stream",
         // and while a wide fan-out is in flight the answer is yes for all of
@@ -179,7 +174,7 @@ impl<'a> Pipeline<'a> {
                             return FinishedCandidate {
                                 index,
                                 result: CandidateResult::setup_aborted(
-                                    base_messages.to_vec(),
+                                    frame.base_messages.to_vec(),
                                     reason.clone(),
                                 ),
                                 cost_usd: 0.0,
@@ -194,7 +189,7 @@ impl<'a> Pipeline<'a> {
                         return FinishedCandidate {
                             index,
                             result: CandidateResult::setup_aborted(
-                                base_messages.to_vec(),
+                                frame.base_messages.to_vec(),
                                 "candidate skipped: the turn's budget was already spent"
                                     .to_string(),
                             ),
@@ -204,16 +199,15 @@ impl<'a> Pipeline<'a> {
                     let mut cost_usd = 0.0;
                     let result = self
                         .run_isolated_candidate(
-                            goal,
-                            base_messages,
-                            plan,
-                            assessment,
+                            frame,
                             worker,
                             authoring,
                             ws,
                             fan,
-                            &mut allowance,
-                            &mut cost_usd,
+                            &mut Spend {
+                                budget: &mut allowance,
+                                total: &mut cost_usd,
+                            },
                         )
                         .await;
                     // Settled here, in completion order, so the next candidate
@@ -232,12 +226,12 @@ impl<'a> Pipeline<'a> {
 
         self.shared_event_lane.store(false, Ordering::Relaxed);
         finished.sort_by_key(|candidate| candidate.index);
-        *budget = budgets.into_parent();
+        *spend.budget = budgets.into_parent();
         // Summed in index order, not completion order: float addition is not
         // associative, and a run's reported cost must not depend on which
         // candidate happened to finish first.
         for candidate in &finished {
-            *total += candidate.cost_usd;
+            *spend.total += candidate.cost_usd;
         }
         finished
             .into_iter()
@@ -250,19 +244,14 @@ impl<'a> Pipeline<'a> {
     /// function now because every local in it — the engine, the steering view,
     /// the bound hook runner — has to live in the candidate's own frame rather
     /// than in a loop iteration shared with nobody.
-    #[allow(clippy::too_many_arguments)]
     async fn run_isolated_candidate(
         &self,
-        goal: &str,
-        base_messages: &[CompletionMessage],
-        plan: Option<&[PlanStep]>,
-        assessment: TaskAssessment,
+        frame: TaskFrame<'_>,
         worker: &ResolvedRole<'_>,
         authoring: Option<WitnessAuthoring<'_>>,
         ws: &dyn CandidateWorkspace,
         fan: Option<&SteeringFanOut<'_>>,
-        budget: &mut BudgetGuard,
-        total: &mut f64,
+        spend: &mut Spend<'_>,
     ) -> CandidateResult {
         let bound_hook_runner = self.hooks.map(|(_, runner)| BoundHookRunner {
             inner: runner,
@@ -297,17 +286,7 @@ impl<'a> Pipeline<'a> {
         if let Some(view) = view.as_ref() {
             engine = engine.with_steering(view);
         }
-        self.run_candidate(
-            goal,
-            base_messages,
-            plan,
-            assessment,
-            authoring,
-            &engine,
-            surface,
-            budget,
-            total,
-        )
-        .await
+        self.run_candidate(frame, authoring, &engine, surface, spend)
+            .await
     }
 }
diff --git a/crates/stella-pipeline/src/pipeline/stage_budget.rs b/crates/stella-pipeline/src/pipeline/stage_budget.rs
index 1cbf6d5f2..e9fc89004 100644
--- a/crates/stella-pipeline/src/pipeline/stage_budget.rs
+++ b/crates/stella-pipeline/src/pipeline/stage_budget.rs
@@ -1,9 +1,29 @@
-use stella_core::{AbortKind, BudgetOutcome};
+use stella_core::{AbortKind, BudgetGuard, BudgetOutcome};
 use stella_protocol::AgentEvent;
 
 use super::{PipelineOutcome, PipelineStatus};
 use crate::triage::TaskClass;
 
+/// The turn's money, threaded through the candidate plane as one parameter
+/// (#1809): the guard that decides whether the next paid call may happen, and
+/// the running total the outcome reports. The two were passed side by side
+/// through every function between `run` and the engine turn — always together,
+/// because spending through the guard without recording it (or vice versa)
+/// is a bug — so the pairing is now a type instead of a convention.
+///
+/// Borrows rather than owns: the guard and the total live with the caller of
+/// [`super::Pipeline::run`] and in `run`'s locals respectively, and every
+/// 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.
+pub(super) struct Spend<'a> {
+    /// Gates each paid call; consulted between model calls only (invariant #6).
+    pub(super) budget: &'a mut BudgetGuard,
+    /// The run's settled cost in USD, reported on every outcome — including
+    /// aborts, which is why it cannot ride inside the guard.
+    pub(super) total: &'a mut f64,
+}
+
 /// A settled stage call crossed an enforced budget. Kept distinct from
 /// provider/routing failures so raw stages must propagate the stop.
 #[derive(Debug, Clone)]
diff --git a/crates/stella-pipeline/src/pipeline/task_frame.rs b/crates/stella-pipeline/src/pipeline/task_frame.rs
new file mode 100644
index 000000000..8c94fe9ac
--- /dev/null
+++ b/crates/stella-pipeline/src/pipeline/task_frame.rs
@@ -0,0 +1,41 @@
+//! The task frame: the read-only view of the task that every candidate stage
+//! receives (#1809).
+//!
+//! Between `run` and `run_engine_turn` the same four values — the goal, the
+//! staged conversation prefix, the plan, and triage's assessment — used to be
+//! threaded positionally through every layer, and each new stage input meant
+//! widening half a dozen signatures (and re-justifying a
+//! `clippy::too_many_arguments` allow at most of them). They travel as one
+//! borrowed `Copy` frame now. The frame is transport, not semantics: each
+//! field keeps its own meaning, documented where the field's type lives.
+//!
+//! What deliberately does NOT ride here: anything mutable (the budget and
+//! running total travel as [`super::stage_budget::Spend`]; per-candidate
+//! tallies as [`crate::witness::warrant::ChangeSignals`] inside
+//! `CandidateState`), and anything per-candidate (the engine, the surface, a
+//! workspace) — a frame outliving the candidates that read it is what makes
+//! it safe to copy freely into each one.
+
+use stella_protocol::CompletionMessage;
+
+use crate::plan::PlanStep;
+use crate::triage::TaskAssessment;
+
+/// The immutable per-turn inputs of the execute/verify plane. Constructed
+/// once in `Pipeline::run` when the candidate stages begin; every stage from
+/// `run_best_of_n` down to `verify_candidate` reads the parts it needs.
+#[derive(Clone, Copy)]
+pub(super) struct TaskFrame<'a> {
+    /// The user's goal, verbatim — what triage classified and what the
+    /// verifier is ultimately asked to judge the work against.
+    pub(super) goal: &'a str,
+    /// The staged conversation prefix every candidate starts from (recall,
+    /// skills, scope — everything before the worker's first turn).
+    pub(super) base_messages: &'a [CompletionMessage],
+    /// The scope stage's plan, when one was produced. `None` runs a single
+    /// unplanned turn.
+    pub(super) plan: Option<&'a [PlanStep]>,
+    /// Triage's classification — which stages run at all, and how strictly
+    /// the ladder verifies.
+    pub(super) assessment: TaskAssessment,
+}
diff --git a/crates/stella-pipeline/src/pipeline/verify_probes.rs b/crates/stella-pipeline/src/pipeline/verify_probes.rs
index 59a5cc413..628c0587a 100644
--- a/crates/stella-pipeline/src/pipeline/verify_probes.rs
+++ b/crates/stella-pipeline/src/pipeline/verify_probes.rs
@@ -358,7 +358,7 @@ impl<'a> Pipeline<'a> {
             .saturating_sub(state.touch_baseline);
         u32::try_from(delta)
             .unwrap_or(u32::MAX)
-            .max(state.file_changes)
+            .max(state.signals.file_changes)
     }
 
     /// Fold one working-tree observation into `state`, refreshing every
@@ -378,7 +378,7 @@ impl<'a> Pipeline<'a> {
         // are bracketed exactly the way this one's were without paying for a
         // second walk of an identical tree.
         self.touches.settle_workspace_probe();
-        state.file_changes = self.observed_mutations(state);
+        state.signals.file_changes = self.observed_mutations(state);
         // The authored channel is read AFTER settling, for the same reason the
         // count is: the probe's attributions are recorded through the same
         // emitter, so reading first would describe the turn before its opaque
@@ -398,7 +398,7 @@ impl<'a> Pipeline<'a> {
         state.diff_available = probe.available;
         state.diff_text = verification_honest_diff(
             authored::splice_authored(probe.text, &authored),
-            state.file_changes,
+            state.signals.file_changes,
         );
     }
 }
diff --git a/crates/stella-pipeline/src/pipeline/witness_stage.rs b/crates/stella-pipeline/src/pipeline/witness_stage.rs
index 6df2a9ea9..b21350106 100644
--- a/crates/stella-pipeline/src/pipeline/witness_stage.rs
+++ b/crates/stella-pipeline/src/pipeline/witness_stage.rs
@@ -236,8 +236,7 @@ impl<'a> Pipeline<'a> {
         authoring: WitnessAuthoring<'_>,
         baseline: CandidateSurface<'_>,
         candidate: CandidateSurface<'_>,
-        budget: &mut BudgetGuard,
-        total: &mut f64,
+        spend: &mut Spend<'_>,
     ) -> Result<Option<Witness>, WitnessAbort> {
         // `port` is deliberately not used here: the caller has already spent it
         // to make `baseline`, and a second snapshot from inside this stage would
@@ -295,17 +294,13 @@ impl<'a> Pipeline<'a> {
         // what the worker did toward the goal. A test file written here is not
         // the change under verification, and counting it as a mutating action
         // would let an authored witness disguise a worker that never acted.
-        let mut file_changes = 0u32;
-        let mut mutating_actions = 0u32;
-        let mut opaque_actions = 0u32;
+        let mut discarded_signals = ChangeSignals::default();
         let text = match self
             .run_engine_turn(
                 &engine,
                 &mut messages,
-                budget,
-                &mut file_changes,
-                &mut mutating_actions,
-                &mut opaque_actions,
+                spend.budget,
+                &mut discarded_signals,
                 // No flip halt: this turn is the witness AUTHOR writing the
                 // test, not the worker fixing the code. Stopping it because
                 // the tracked command went green would abandon the artifact
@@ -315,14 +310,14 @@ impl<'a> Pipeline<'a> {
             .await
         {
             TurnOutcome::Completed { text, cost_usd } => {
-                *total += cost_usd;
+                *spend.total += cost_usd;
                 text
             }
             TurnOutcome::Aborted {
                 reason, cost_usd, ..
             } => {
-                *total += cost_usd;
-                if let Some(abort) = budget_abort(budget.evaluate()) {
+                *spend.total += cost_usd;
+                if let Some(abort) = budget_abort(spend.budget.evaluate()) {
                     return Err(WitnessAbort::rejected(abort.reason));
                 }
                 return Err(WitnessAbort::degradable(format!(
@@ -391,24 +386,22 @@ impl<'a> Pipeline<'a> {
                 .run_engine_turn(
                     &repair_engine,
                     &mut messages,
-                    budget,
-                    &mut file_changes,
-                    &mut mutating_actions,
-                    &mut opaque_actions,
+                    spend.budget,
+                    &mut discarded_signals,
                     // Witness repair, same reasoning as the author turn.
                     None,
                 )
                 .await
             {
                 TurnOutcome::Completed { text, cost_usd } => {
-                    *total += cost_usd;
+                    *spend.total += cost_usd;
                     text
                 }
                 TurnOutcome::Aborted {
                     reason, cost_usd, ..
                 } => {
-                    *total += cost_usd;
-                    if let Some(abort) = budget_abort(budget.evaluate()) {
+                    *spend.total += cost_usd;
+                    if let Some(abort) = budget_abort(spend.budget.evaluate()) {
                         return Err(WitnessAbort::rejected(abort.reason));
                     }
                     return Err(WitnessAbort::degradable(format!(
@@ -553,13 +546,12 @@ impl<'a> Pipeline<'a> {
         authoring: Option<WitnessAuthoring<'_>>,
         surface: CandidateSurface<'_>,
         state: &mut CandidateState,
-        budget: &mut BudgetGuard,
-        total: &mut f64,
+        spend: &mut Spend<'_>,
     ) -> Result<Option<Witness>, String> {
         let Some(authoring) = authoring else {
             return Ok(None);
         };
-        if !warrant(&state.diff_text, state.change_signals()).is_required() {
+        if !warrant(&state.diff_text, state.signals).is_required() {
             // No Witness stage is emitted, because none runs. The reason is
             // not lost: `warranted_completion` reads the same warrant during
             // verification and records the sentence on the verdict, which is
@@ -594,7 +586,7 @@ impl<'a> Pipeline<'a> {
             workspace: Some(baseline.as_ref()),
         };
         let authored = self
-            .witness_stage(goal, authoring, baseline_surface, surface, budget, total)
+            .witness_stage(goal, authoring, baseline_surface, surface, spend)
             .await;
         baseline.remove().await;