diff --git a/crates/stella-cli/src/agent.rs b/crates/stella-cli/src/agent.rs index 2e8232e8d..1868ec4d3 100644 --- a/crates/stella-cli/src/agent.rs +++ b/crates/stella-cli/src/agent.rs @@ -22,7 +22,7 @@ use stella_mcp::{McpConfig, McpServerConfig, McpToolSet}; use stella_model::credential::ApiKey; use stella_model::provider::Provider; use stella_pipeline::{ - AlwaysAbortGate, CmdOutcome, ContextRecallPort, McpPrefetchPort, NoContextRecall, Pipeline, + AlwaysAbortGate, CmdOutcome, ContextRecallPort, McpPrefetchPort, NoContextRecall, PipelineConfig, PipelinePorts, PipelineStatus, ProviderResolver, RepoStatusPort, RepoStructurePort, }; @@ -43,7 +43,7 @@ use crate::memory::{ }; use crate::runtime::{SystemClock, TokioSleeper}; use crate::tui; -use crate::{OutputFormat, config::Config}; +use crate::{OutputFormat, config::Config, resume_frame}; use stella_context::EpisodeOutcome; mod coverage; @@ -452,8 +452,8 @@ async fn run_pipeline_one_shot( steering: None, }; - crate::resume_frame::declare(&cfg.durability, &pipeline_config); - let pipeline = Pipeline::new(ports, pipeline_event_sender(&tx, format), pipeline_config); + let events = pipeline_event_sender(&tx, format); + let pipeline = resume_frame::pipeline(&cfg.durability, ports, events, pipeline_config); pipeline.run(prompt, &mut messages, &mut budget).await }; diff --git a/crates/stella-cli/src/agent/goal.rs b/crates/stella-cli/src/agent/goal.rs index 60ea47ec4..944a73317 100644 --- a/crates/stella-cli/src/agent/goal.rs +++ b/crates/stella-cli/src/agent/goal.rs @@ -563,7 +563,7 @@ pub(crate) async fn run_goal_turn( /// `Engine::run_turn`. /// /// The goal-loop verifier is distinct from the pipeline's verify verifier: the verify -/// verifier (inside [`Pipeline::run`]) answers "did this change pass its tests?", +/// verifier (inside [`stella_pipeline::Pipeline::run`]) answers "did this change pass its tests?", /// while the goal verifier here answers "does the whole effort meet the goal?". /// Both are independent of the worker model. #[allow(clippy::too_many_arguments)] @@ -769,7 +769,8 @@ async fn run_goal_pipeline_turn( // Goal pipeline rounds run without an interactive steer tap. steering: None, }; - let pipeline = Pipeline::new(ports, tx.clone(), pipeline_config); + let pipeline = + crate::resume_frame::pipeline(&cfg.durability, ports, tx.clone(), pipeline_config); // The same kickoff wording as the raw goal loop and the served // one — `stella_core::goal` owns the bytes for all three. let round_goal = stella_core::goal::goal_kickoff_text(goal); diff --git a/crates/stella-cli/src/command_deck.rs b/crates/stella-cli/src/command_deck.rs index ac437ee98..557ec0ab7 100644 --- a/crates/stella-cli/src/command_deck.rs +++ b/crates/stella-cli/src/command_deck.rs @@ -83,7 +83,7 @@ use stella_core::router::CircuitBreaker; use stella_core::{BudgetGuard, CalibrationMap, Engine, Router, TurnOutcome}; use stella_model::provider::Provider; use stella_pipeline::{ - ContextRecallPort, McpPrefetchPort, NoContextRecall, Pipeline, PipelineConfig, PipelinePorts, + ContextRecallPort, McpPrefetchPort, NoContextRecall, PipelineConfig, PipelinePorts, PipelineStatus, }; use stella_protocol::{ @@ -4497,9 +4497,9 @@ async fn run_lead_pipeline_turn( headless_bypass_scope_review: false, ..agent::apply_pipeline_tuning(cfg, PipelineConfig::default()) }; - // #1214's seam, now driven from the deck: the pipeline attaches the gate - // to every engine it builds and parks its management calls behind it. - let pipeline = Pipeline::new(ports, tx.clone(), config).with_turn_gate(pause.turn_gate()); + // #1214's seam, driven from the deck — see `resume_frame::pipeline`. + let pipeline = crate::resume_frame::pipeline(&cfg.durability, ports, tx.clone(), config) + .with_turn_gate(pause.turn_gate()); pipeline.run(prompt, messages, budget).await }; // Same settle window as `run_lead_turn` — see `SteeringTap::mark_settling`. diff --git a/crates/stella-cli/src/daemon/boot.rs b/crates/stella-cli/src/daemon/boot.rs index 240ff8b97..75804c526 100644 --- a/crates/stella-cli/src/daemon/boot.rs +++ b/crates/stella-cli/src/daemon/boot.rs @@ -127,7 +127,7 @@ pub(super) struct BootCandidate { /// Whether the workspace the turn must continue in still exists. pub(super) workspace_exists: bool, /// Whether the run left an unanswered approval request in its sidecar - /// ([`supervised::APPROVAL_REQUEST`]). + /// ([`stella_store::supervised::APPROVAL_REQUEST`]). /// /// Such a run does not fail on resume — it *parks*, waiting for a human /// who is not there. The sweep resumes one run at a time and streams each diff --git a/crates/stella-cli/src/fleet_cmd.rs b/crates/stella-cli/src/fleet_cmd.rs index 1a3e5cec4..31a424e8a 100644 --- a/crates/stella-cli/src/fleet_cmd.rs +++ b/crates/stella-cli/src/fleet_cmd.rs @@ -802,9 +802,7 @@ async fn run_task( let (summary, success, outcome_label, force_incomplete): (String, bool, &str, bool) = if use_pipeline { use stella_core::router::{CircuitBreaker, Router}; - use stella_pipeline::{ - NoContextRecall, Pipeline, PipelineConfig, PipelinePorts, PipelineStatus, - }; + use stella_pipeline::{NoContextRecall, PipelineConfig, PipelinePorts, PipelineStatus}; let model_ref = stella_protocol::ModelRef::new(cfg.provider.id, cfg.model_id.clone()); // Role wiring from `agent_engine_config` — fleet workers honor the // same worker/triage/verifier pins and per-role overrides as `stella run`. @@ -879,7 +877,9 @@ async fn run_task( let _controls = registry.attach_turn_controls( stella_core::ports::TurnControls::none().with_gate(gate.clone()), ); - let pipeline = Pipeline::new(ports, tx.clone(), config).with_turn_gate(gate.as_ref()); + let pipeline = + crate::resume_frame::pipeline(&cfg.durability, ports, tx.clone(), config) + .with_turn_gate(gate.as_ref()); // 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` diff --git a/crates/stella-cli/src/resume_frame.rs b/crates/stella-cli/src/resume_frame.rs index e2d6b6dd2..8c92237ee 100644 --- a/crates/stella-cli/src/resume_frame.rs +++ b/crates/stella-cli/src/resume_frame.rs @@ -243,6 +243,31 @@ pub fn declare( } } +/// Build a [`stella_pipeline::Pipeline`] that has already declared its frame. +/// +/// **The one construction path**, so a surface cannot come to hold a pipeline +/// whose checkpoints do not say what they are. The frame was previously +/// declared beside `Pipeline::new` at one of four call sites, and the three +/// that forgot left checkpoints that resume as plain engine turns with no +/// notice that their stages are gone (#1672). +/// +/// Pairing the two here rather than asking each surface to remember is the +/// difference between a convention and a mechanism. +/// +/// Callers still chain what is theirs. The deck and fleet workers add +/// `with_turn_gate` — #1214's seam, where the pipeline attaches the gate to +/// every engine it builds and parks its management calls behind it — because +/// the gate is per-surface while the frame is not. +pub fn pipeline<'a>( + durability: &crate::durability::SessionDurability, + ports: stella_pipeline::PipelinePorts<'a>, + events: impl Into, + config: stella_pipeline::PipelineConfig, +) -> stella_pipeline::Pipeline<'a> { + declare(durability, &config); + stella_pipeline::Pipeline::new(ports, events, config) +} + #[cfg(test)] mod tests { use super::*; @@ -339,4 +364,68 @@ mod tests { assert!(!frame.witness_writer); assert_eq!(frame.candidates, 0); } + + /// **Witness (#1672).** Every surface that builds a `Pipeline` declares + /// its frame first. + /// + /// The frame was declared at exactly one of four call sites, so a + /// checkpoint left by the deck, the goal loop or a fleet worker read as a + /// plain engine turn and any resume from it degraded in silence — the + /// failure #1615 closed for `stella run` alone. + /// + /// This greps the source rather than driving a turn, deliberately. What + /// went wrong was **wiring**, not logic: every unit here already passed + /// while three surfaces never called it. A behavioural test would need one + /// scripted run per surface and would still only cover the surfaces + /// somebody remembered to write a test for, which is the same gap one + /// level up. The repo uses source-grep guards for exactly this shape (the + /// `stella fullauto` wrapper guards from #1619). + #[test] + fn every_pipeline_construction_declares_its_resume_frame() { + let src = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src"); + let mut sites = Vec::new(); + let mut undeclared = Vec::new(); + + let mut stack = vec![src]; + while let Some(dir) = stack.pop() { + let Ok(entries) = std::fs::read_dir(&dir) else { + continue; + }; + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + stack.push(path); + continue; + } + if path.extension().and_then(|e| e.to_str()) != Some("rs") { + continue; + } + // This guard's own source carries the needle in a string + // literal, so it would report itself forever. + if path.file_name().and_then(|n| n.to_str()) == Some("resume_frame.rs") { + continue; + } + let Ok(text) = std::fs::read_to_string(&path) else { + continue; + }; + let lines: Vec<&str> = text.lines().collect(); + for (i, line) in lines.iter().enumerate() { + if !line.contains("Pipeline::new(") { + continue; + } + sites.push(format!("{}:{}", path.display(), i + 1)); + undeclared.push(format!("{}:{}", path.display(), i + 1)); + } + } + } + + let _ = sites; + assert!( + undeclared.is_empty(), + "these sites call `Pipeline::new` directly and so declare no resume frame — \ + a checkpoint they leave resumes as a plain engine turn with no notice that \ + its stages are gone. Build through `resume_frame::pipeline` instead: \ + {undeclared:?}" + ); + } }