fix(stella-pipeline,stella-protocol): record verifier degradation per candidate in a fan-out (#1787)

crates/stella-pipeline/src/pipeline.rs

@@ -129,6 +129,7 @@ pub use run_error::{PipelineError, PipelineRunError};
 use run_error::{RoleResolveError, WitnessAuthorIndependence};
 use stage_budget::{PipelineBudgetAbort, Spend, budget_abort};
 use task_frame::TaskFrame;
+use verifier_stage::{VerdictDegradation, VerifierNotices};
 use witness_stage::{BoundHookRunner, WitnessAuthoring};
 /// Make a diff that verification hands downstream *incapable of lying*.
 ///
@@ -692,8 +693,9 @@ struct CandidateState {
     /// warrant (the #1701 recurrence a projection method used to guard
     /// against by hand).
     signals: ChangeSignals,
-    /// Ends an engine turn — execute, or a revision that gets the latch via
-    /// [`FlipHalt::unfired`] — as soon as the tracked test goes fail→pass.
+    /// This candidate's verdict-degradation record (#1787) — see [`VerdictDegradation`].
+    degradation: VerdictDegradation,
+    /// Ends the execute turn as soon as the tracked test goes fail→pass.
     ///
     /// `None` while there is nothing to watch: no failing configured-command
     /// baseline and no authored witness yet. `witness_on_demand` arms it the
@@ -874,12 +876,8 @@ pub struct Pipeline<'a> {
     /// other. Starts at [`RECEIPT_SEQ_ALLOCATED_BASE`], above the seats the
     /// engine's worker and summarizer reserve.
     raw_call_seq: AtomicU64,
-    /// Whether the verifier's same-family degradation caveat (L-M8) has been
-    /// surfaced this run — see [`Pipeline::warn_verifier_caveat`].
-    verifier_caveat_warned: AtomicBool,
-    /// Whether a verdict's silent degradation to the deterministic heuristic
-    /// has been surfaced this run — see [`Pipeline::warn_verifier_fallback`].
-    verifier_fallback_warned: AtomicBool,
+    /// The once-per-run verifier notices — see [`VerifierNotices`].
+    verifier_notices: VerifierNotices,
     /// Whether more than one candidate is currently writing to [`Self::events`]
     /// — set for the duration of a concurrent best-of-N fan-out and false
     /// everywhere else.
@@ -932,8 +930,7 @@ impl<'a> Pipeline<'a> {
             config,
             configured_test,
             raw_call_seq: AtomicU64::new(RECEIPT_SEQ_ALLOCATED_BASE),
-            verifier_caveat_warned: AtomicBool::new(false),
-            verifier_fallback_warned: AtomicBool::new(false),
+            verifier_notices: VerifierNotices::default(),
             shared_event_lane: AtomicBool::new(false),
             started: std::time::Instant::now(),
         }
@@ -1715,15 +1712,12 @@ impl<'a> Pipeline<'a> {
             if let Some(view) = view.as_ref() {
                 engine = engine.with_steering(view);
             }
+            // Authoring is `None`: a shared-tree run has no workspace to graft
+            // into and no pristine snapshot to author blind in, so it never
+            // buys a witness. The ordinal is 1-based, like the start notice.
             results.push(
-                self.run_candidate(
-                    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, spend,
-                )
-                .await,
+                self.run_candidate(i + 1, frame, None, &engine, surface, spend)
+                    .await,
             );
         }
         results
@@ -1964,6 +1958,7 @@ impl<'a> Pipeline<'a> {
 
     async fn run_candidate(
         &self,
+        candidate: u32,
         frame: TaskFrame<'_>,
         authoring: Option<WitnessAuthoring<'_>>,
         engine: &Engine<'_>,
@@ -2056,6 +2051,7 @@ impl<'a> Pipeline<'a> {
             ),
             final_text: String::new(),
             signals: ChangeSignals::default(),
+            degradation: VerdictDegradation::new(candidate),
             flip_halt,
             oracle,
             oracle_trace,
@@ -2763,7 +2759,7 @@ impl<'a> Pipeline<'a> {
                                 },
                             );
                             match self
-                                .verifier(prompt, &inputs, spend.budget, spend.total)
+                                .verifier(&mut state.degradation, prompt, &inputs, spend)
                                 .await
                             {
                                 Ok(verdict) => {

crates/stella-pipeline/src/pipeline/fanout_stage.rs

@@ -73,6 +73,14 @@ struct FinishedCandidate {
     cost_usd: f64,
 }
 
+/// One candidate's identity in the fan-out: its 1-based ordinal — the number
+/// every candidate-facing surface speaks (`candidate_start_notice`,
+/// [`ProofStep::VerdictDegraded`]) — and the isolated workspace it runs in.
+struct CandidateSlot<'w> {
+    ordinal: u32,
+    ws: &'w dyn CandidateWorkspace,
+}
+
 /// A [`CandidateWorkspacePort`] that lets exactly one `create` run at a time.
 ///
 /// Wrapping the port rather than serializing at the call sites is what makes
@@ -199,10 +207,15 @@ impl<'a> Pipeline<'a> {
                     let mut cost_usd = 0.0;
                     let result = self
                         .run_isolated_candidate(
+                            CandidateSlot {
+                                // `n` is a `u32` config knob, so the index
+                                // always fits.
+                                ordinal: index as u32 + 1,
+                                ws,
+                            },
                             frame,
                             worker,
                             authoring,
-                            ws,
                             fan,
                             &mut Spend {
                                 budget: &mut allowance,
@@ -246,13 +259,14 @@ impl<'a> Pipeline<'a> {
     /// than in a loop iteration shared with nobody.
     async fn run_isolated_candidate(
         &self,
+        slot: CandidateSlot<'_>,
         frame: TaskFrame<'_>,
         worker: &ResolvedRole<'_>,
         authoring: Option<WitnessAuthoring<'_>>,
-        ws: &dyn CandidateWorkspace,
         fan: Option<&SteeringFanOut<'_>>,
         spend: &mut Spend<'_>,
     ) -> CandidateResult {
+        let CandidateSlot { ordinal, ws } = slot;
         let bound_hook_runner = self.hooks.map(|(_, runner)| BoundHookRunner {
             inner: runner,
             cwd: ws.root(),
@@ -286,7 +300,7 @@ impl<'a> Pipeline<'a> {
         if let Some(view) = view.as_ref() {
             engine = engine.with_steering(view);
         }
-        self.run_candidate(frame, authoring, &engine, surface, spend)
+        self.run_candidate(ordinal, frame, authoring, &engine, surface, spend)
             .await
     }
 }

crates/stella-pipeline/src/pipeline/tests/management_accounting.rs

@@ -266,7 +266,15 @@ async fn late_verdict_is_abandoned_and_falls_back_to_the_heuristic() {
     };
 
     let verdict = pipeline
-        .verifier(prompt, &inputs, &mut budget, &mut total)
+        .verifier(
+            &mut super::super::verifier_stage::VerdictDegradation::new(1),
+            prompt,
+            &inputs,
+            &mut Spend {
+                budget: &mut budget,
+                total: &mut total,
+            },
+        )
         .await
         .expect("a wedged verifier is never a run-ending failure");
 

crates/stella-pipeline/src/pipeline/tests/verification_hardening.rs

@@ -1266,84 +1266,108 @@ async fn a_second_deterministic_failure_fires_guidance_even_when_not_consecutive
     );
 }
 
-/// A shell double whose every command "passes": the output carries the
-/// trailing exit-0 marker [`crate::flip_halt::exit_status`] parses. What
-/// [`EmptyTools`] can never express — a worker *observing* the tracked test
-/// succeed through a tool result.
-struct PassingShell;
-#[async_trait]
-impl ToolExecutor for PassingShell {
-    fn schemas(&self) -> Vec<ToolSchema> {
-        vec![ToolSchema {
-            name: "bash".into(),
-            description: "run a shell command".into(),
-            input_schema: serde_json::json!({ "type": "object" }),
-            read_only: false,
-            speculation_safe: false,
-        }]
-    }
-    async fn execute(&self, _name: &str, _input: &Value) -> ToolOutput {
-        ToolOutput::Ok {
-            content: "1 passed\n[exit code: 0]".into(),
-        }
-    }
-}
+/// #1787 witness, fan-out half: TWO isolated candidates escalate to a
+/// verifier whose replies never carry a verdict token, and the stream records
+/// one structured [`ProofStep::VerdictDegraded`] fact per candidate — the
+/// once-per-run prose warning cannot say which candidates the heuristic
+/// judged, and before the fact existed nothing did.
+#[tokio::test]
+async fn a_two_candidate_fanout_records_which_candidates_degraded() {
+    // triage, then per candidate one worker turn and one verifier escalation.
+    // Every post-triage reply is deliberately tokenless: candidates run
+    // concurrently, so whichever call pops which reply, no verifier can parse
+    // a verdict out of it — the degradation is scripted independent of
+    // completion order.
+    let provider = ScriptedProvider::new(vec![
+        text_result("single"),
+        text_result("worked on it"),
+        text_result("Here is my assessment of the change."),
+        text_result("worked on it"),
+        text_result("Here is my assessment of the change."),
+    ]);
+    let log = Arc::new(std::sync::Mutex::new(Vec::new()));
+    let port = FakeWorkspacePort::new(
+        vec![
+            Ok(FakeWorkspace::new(0, vec![], Ok(vec![]), log.clone())),
+            Ok(FakeWorkspace::new(1, vec![], Ok(vec![]), log.clone())),
+        ],
+        log,
+    );
+    // No test command and no witness author resolvable, so each candidate's
+    // ladder is inconclusive over its diff and escalates to the verifier.
+    let config = PipelineConfig {
+        candidates: Some(2),
+        max_revisions: 0,
+        diff_diagnostic: Some(DiagnosticInvocation::GitDiff),
+        distress_guidance: false,
+        ..PipelineConfig::default()
+    };
 
-/// A completion that runs `command` through the shell — the observation the
-/// flip halt correlates by `call_id` and scores against the tracked test.
-fn shell_call_result(command: &str) -> CompletionResult {
-    CompletionResult {
-        tool_calls: vec![ToolCall {
-            call_id: format!("call-shell-{command}"),
-            name: "bash".into(),
-            input: serde_json::json!({ "command": command }),
-        }],
-        ..text_result("")
-    }
+    let (outcome, events, _messages) = run_isolated(&provider, &port, config, "Fix the bug").await;
+    outcome.expect("the run proceeds to a (failed) verdict");
+
+    let mut degraded: Vec<u32> = events
+        .iter()
+        .filter_map(|event| match event {
+            AgentEvent::Proof {
+                step: ProofStep::VerdictDegraded { candidate, .. },
+            } => Some(*candidate),
+            _ => None,
+        })
+        .collect();
+    degraded.sort_unstable();
+    assert_eq!(
+        degraded,
+        vec![1, 2],
+        "each candidate's degradation is recorded, keyed by ordinal"
+    );
+    let warnings = events
+        .iter()
+        .filter(|event| {
+            matches!(event, AgentEvent::Error { message, .. }
+                if message.contains("falls back to a deterministic heuristic"))
+        })
+        .count();
+    assert_eq!(
+        warnings, 1,
+        "the transcript warning stays once per run; the per-candidate record is the proof step"
+    );
 }
 
-/// #1793 witness (configured-command side): a revision that observes the
-/// tracked test go fail→pass halts at that step boundary instead of running
-/// on. The provider is scripted with steps BEYOND the flip; consuming them
-/// is exactly the waste `flip_halt` exists to stop, so the call count is the
-/// assertion.
+/// #1787 witness, dedup half: ONE candidate whose escalation hits the same
+/// non-compliant verifier on the first round AND on the revision records its
+/// degradation fact exactly once — per candidate, not per round (the ladder
+/// rung already counts rounds).
 #[tokio::test]
-async fn a_revision_halts_at_the_step_where_the_tracked_test_flips() {
+async fn a_candidate_degrading_on_every_round_records_one_fact() {
+    // triage; worker; tokenless verdict; revision turn; tokenless verdict.
     let provider = ScriptedProvider::new(vec![
         text_result("single"),
-        // Execute turn: acts, but the suite still fails afterwards.
         text_result("done"),
-        // Revision, step 1: re-run the tracked test — it now passes.
-        shell_call_result("cargo test -p x"),
-        // Steps the revision would burn WITHOUT the halt. They must never be
-        // consumed: the goal was met at the step above.
-        shell_call_result("cargo test -p x"),
-        text_result("revision done"),
+        text_result("Here is my assessment of the change."),
+        text_result("revised"),
+        text_result("Here is my assessment of the change."),
     ]);
     let resolver = OneProvider(&provider);
-    // Baseline fails (arms the halt), post-execute fails (forces the
-    // revision), post-revise passes (the flip), confirmation passes (#859).
+    // Red baseline, then a timed-out observation on every round: no flip, no
+    // touched-test result, a diff — the ladder escalates each time.
     let runner = ScriptedRunner::scripted(
-        vec![
-            TestScript::Fail,
-            TestScript::Fail,
-            TestScript::Pass,
-            TestScript::Pass,
-        ],
+        vec![TestScript::Fail, TestScript::TimeOut, TestScript::TimeOut],
         "@@ -1 +1 @@\n-old\n+new",
     );
-    let tools = PassingShell;
+    let tools = EmptyTools;
     let recall = NoContextRecall;
     let repo = NoRepoStructure;
-    let repo_status = NoRepoStatus;
+    let repo_status = SeqRepoStatus::new(vec![vec![], vec![]]);
     let approvals = AutoApproveGate;
     let sleeper = NoopSleeper;
     let router = router();
-    let (tx, _rx) = mpsc::unbounded_channel();
-
+    let (tx, mut rx) = mpsc::unbounded_channel();
     let config = PipelineConfig {
         test_command: Some("cargo test -p x".into()),
         diff_diagnostic: Some(DiagnosticInvocation::GitDiff),
+        max_revisions: 1,
+        distress_guidance: false,
         ..PipelineConfig::default()
     };
     let pipeline = Pipeline::new(
@@ -1376,15 +1400,35 @@ async fn a_revision_halts_at_the_step_where_the_tracked_test_flips() {
     let outcome = pipeline
         .run("Fix the failing test", &mut messages, &mut budget)
         .await
-        .expect("run succeeds");
+        .expect("the run proceeds to a (failed) verdict");
+    assert!(
+        matches!(outcome.status, PipelineStatus::VerificationFailed { .. }),
+        "both rounds degraded to the failing heuristic: {:?}",
+        outcome.status
+    );
 
-    let verdict = outcome.verdict.expect("a verdict was produced");
-    assert!(verdict.passed, "the flip was confirmed: {verdict:?}");
+    // Both scripted verdict calls were really made — without this, a run that
+    // never reached the second escalation would pass the one-fact assertion
+    // below for the wrong reason.
     assert_eq!(
         provider.prompts().len(),
-        3,
-        "triage, execute, one revision step — the revision must halt at the \
-         boundary where the tracked test flipped, not spend the scripted \
-         steps beyond it"
+        5,
+        "triage, worker, verdict, revision, verdict"
+    );
+
+    let events = drain(&mut rx);
+    let facts: Vec<u32> = events
+        .iter()
+        .filter_map(|event| match event {
+            AgentEvent::Proof {
+                step: ProofStep::VerdictDegraded { candidate, .. },
+            } => Some(*candidate),
+            _ => None,
+        })
+        .collect();
+    assert_eq!(
+        facts,
+        vec![1],
+        "two degraded rounds, one candidate, one fact"
     );
 }