speccoder_runtime/pipeline/
mod.rs

1//! The run pipeline: one task ([`task`]), a feature ([`feature`]), and the
2//! pieces they share.
3
4use std::path::Path;
5use std::sync::Arc;
6
7use speccoder_core::snapshot::{Diagnostic, ProjectSnapshot, Severity};
8use speccoder_core::{CommandData, CommandResult, CommandStatus};
9
10use crate::commands::*;
11use crate::{approvals, progress, substrate, verify, worktree};
12
13pub mod acceptance;
14pub mod bisect;
15pub mod bug;
16pub mod commit;
17pub mod feature;
18pub mod full;
19pub mod judge;
20pub mod lanes;
21pub mod outcome;
22pub mod samples;
23pub mod task;
24
25pub(crate) use acceptance::*;
26pub(crate) use bug::{reject as bug_reject, triad_now as bug_triad_now};
27pub(crate) use commit::*;
28pub use feature::*;
29pub(crate) use judge::*;
30pub(crate) use lanes::*;
31pub(crate) use outcome::*;
32pub(crate) use samples::*;
33pub use task::*;
34
35/// Where a project's run state lives, and its model configuration.
36pub(crate) struct Setup {
37    pub(crate) snapshot: Box<speccoder_core::snapshot::ProjectSnapshot>,
38    pub(crate) model: substrate::model_config::ModelConfig,
39    pub(crate) plane_dir: std::path::PathBuf,
40    /// The project's format.
41    pub(crate) format: &'static dyn crate::format::Adapter,
42}
43
44pub(crate) fn setup(
45    path: &Path,
46    feature: Option<&str>,
47    state: &Path,
48    command: &str,
49) -> Result<Setup, Box<CommandResult>> {
50    let inspected = inspect(path, feature);
51    let Some(CommandData::Inspect(snapshot)) = inspected.data() else {
52        return Err(Box::new(CommandResult::carried(command, &inspected)));
53    };
54    let snapshot = snapshot.clone();
55    preflight(&snapshot, command)?;
56    let config_error =
57        |m: String| Box::new(CommandResult::new(command, CommandStatus::ConfigError, &m));
58    let root = Path::new(&snapshot.root);
59    let config_dir = verify::find_config(root).ok_or_else(|| {
60        config_error(format!(
61            "no speccoder.toml at or above {}; add [model] provider = \"anthropic\", model = \"…\"",
62            root.display()
63        ))
64    })?;
65    let text = std::fs::read_to_string(config_dir.join(verify::config::FILE))
66        .map_err(|e| config_error(format!("speccoder.toml: {e}")))?;
67    let model = substrate::model_config::parse(&text)
68        .map_err(|e| config_error(format!("speccoder.toml: {e}")))?
69        .ok_or_else(|| config_error("speccoder.toml has no [model] section".to_owned()))?;
70    Ok(Setup {
71        format: crate::format::of(&snapshot),
72        plane_dir: worktree::project_state_dir(state, root),
73        snapshot,
74        model,
75    })
76}
77
78/// The format's preflight: errors of the selected unit (and, for `run`,
79/// what the format's own installed tools say) stop a run or proposal before
80/// any model call.
81fn preflight(snapshot: &ProjectSnapshot, command: &str) -> Result<(), Box<CommandResult>> {
82    if !matches!(command, "run" | "propose") {
83        return Ok(());
84    }
85    let format = crate::format::of(snapshot);
86    let mut extra = Vec::new();
87    if command == "run" {
88        match crate::sandbox::runner_at(Path::new(&snapshot.root)) {
89            Ok((runner, _)) => extra = format.crosscheck(snapshot, &runner),
90            Err(e) => extra.push(Diagnostic::new(
91                Severity::Info,
92                "crosscheck_skipped",
93                format!("cross-check skipped: {e}"),
94            )),
95        }
96    }
97    let problems = format.blocking(snapshot).len()
98        + extra
99            .iter()
100            .filter(|d| d.severity == Severity::Error)
101            .count();
102    if problems == 0 {
103        return Ok(());
104    }
105    let mut shown = snapshot.clone();
106    shown.diagnostics.extend(extra);
107    let name = shown
108        .feature
109        .as_ref()
110        .map(|f| f.feature.name.clone())
111        .unwrap_or_default();
112    Err(Box::new(
113        CommandResult::new(
114            command,
115            CommandStatus::DiagnosticsError,
116            &format!(
117                "{} {name} cannot run: {problems} problem(s)",
118                format.wording().unit
119            ),
120        )
121        .with_data(CommandData::Inspect(Box::new(shown))),
122    ))
123}
124
125pub(crate) fn tokio_runtime() -> Result<tokio::runtime::Runtime, String> {
126    tokio::runtime::Builder::new_current_thread()
127        .enable_all()
128        .build()
129        .map_err(|e| format!("cannot start the async runtime: {e}"))
130}
131
132/// Replaces every occurrence of `secret` (never shown, never journaled by
133/// speccoder), and every value copied from the person's secret files
134/// ([`crate::secrets`]).
135pub(crate) fn redact(text: &str, secret: Option<&str>) -> String {
136    let text = crate::secrets::mask(text);
137    match secret {
138        Some(s) if !s.is_empty() => text.replace(s, "[redacted]"),
139        _ => text,
140    }
141}
142
143/// One `speccoder run` from its fresh start to its end: `run-<unix
144/// secs>-<8 hex>`. Never one attempt's agentplane run (a retry is a new one
145/// of those, in the same run).
146#[derive(
147    Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
148)]
149#[serde(transparent)]
150pub struct RunId(pub String);
151
152impl RunId {
153    /// A fresh id.
154    #[must_use]
155    pub fn fresh() -> Self {
156        use sha2::{Digest as _, Sha256};
157        static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
158        let now = std::time::SystemTime::now()
159            .duration_since(std::time::UNIX_EPOCH)
160            .unwrap_or_default();
161        let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
162        let digest = Sha256::digest(format!(
163            "{}/{}/{}/{n}",
164            now.as_nanos(),
165            std::process::id(),
166            std::thread::current().name().unwrap_or_default()
167        ));
168        let hex: String = digest.iter().take(4).map(|b| format!("{b:02x}")).collect();
169        Self(format!("run-{}-{hex}", now.as_secs()))
170    }
171
172    /// Its text.
173    #[must_use]
174    pub fn as_str(&self) -> &str {
175        &self.0
176    }
177}
178
179impl Default for RunId {
180    fn default() -> Self {
181        Self::fresh()
182    }
183}
184
185impl std::fmt::Display for RunId {
186    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
187        f.write_str(&self.0)
188    }
189}
190
191impl RunOptions {
192    /// The run's id now.
193    #[must_use]
194    pub fn run(&self) -> RunId {
195        self.run_id
196            .lock()
197            .map(|r| r.clone())
198            .unwrap_or_else(|_| RunId(String::new()))
199    }
200
201    /// The run's prompts so far.
202    #[must_use]
203    pub fn asked_now(&self) -> u32 {
204        self.asked.lock().map(|a| *a).unwrap_or(0)
205    }
206}
207
208/// How a run behaves beyond its configuration.
209#[derive(Debug, Clone, Default)]
210pub struct RunOptions {
211    /// At most this many tasks in a feature run.
212    pub max_tasks: Option<usize>,
213    /// Who to ask inline when a call needs approval; without one,
214    /// the attempt stops and `speccoder task approve` decides.
215    pub approver: Option<Arc<dyn approvals::Approver>>,
216    /// Where live progress goes: the CLI's printer.
217    pub progress: Option<Arc<dyn progress::Progress>>,
218    /// Ready `[P]` tasks run at once, each in its own worktree; 0 or 1: one at a time.
219    pub parallel: usize,
220    /// Run the task in this lane's worktree instead of the feature's.
221    pub lane: Option<String>,
222    /// Set by a person (an editor's cancel): a feature run stops
223    /// before its next task.
224    pub cancel: Option<Arc<std::sync::atomic::AtomicBool>>,
225    /// Run although the feature's checklists have unchecked items; the
226    /// override is recorded on every commit of the run.
227    pub ignore_checklists: bool,
228    /// Set a task's held or interrupted attempt aside and write its
229    /// acceptance tests anew (`run --task T --fresh-acceptance`).
230    pub fresh_acceptance: bool,
231    /// Attempt although a check cannot run on the untouched code
232    /// (`run --allow-failing-baseline`); otherwise the run stops before any
233    /// model call (`CHECK-BROKEN`).
234    pub allow_failing_baseline: bool,
235    /// Objects a person approved "for this run": shared by every task (and
236    /// lane) of one `speccoder run`, gone with it.
237    pub run_grants: Arc<std::sync::Mutex<Vec<crate::tools::policy::Object>>>,
238    /// The run: one per `speccoder run` from its fresh start; an
239    /// invocation that resumes a suspended attempt adopts that attempt's.
240    /// Grants, the asking count and the batch are keyed by it.
241    pub run_id: Arc<std::sync::Mutex<RunId>>,
242    /// Prompts the run has shown a person (`[approvals] budget` caps them).
243    pub asked: Arc<std::sync::Mutex<u32>>,
244    /// Run in the person's checkout on `speccoder/<feature>` instead of a
245    /// worktree (`run --in-place`).
246    pub in_place: bool,
247    /// The in-place session the runtime opened for this run.
248    pub session: Option<Arc<std::sync::Mutex<crate::in_place::Session>>>,
249    /// Run a bug of the format's own bug structure (no task list names it).
250    pub bug_home: Option<Arc<bug::NativeBug>>,
251    /// How the run asks for a person: `Ask` as today, or
252    /// `Autopilot` — nothing asks, a call that would is refused for the run
253    /// unless a grant in `run_grants` (the ones the person ticked) covers it.
254    pub mode: speccoder_core::oracle::RunMode,
255}
256
257/// Where a task runs, as `worktree` reports it: speccoder's worktree, or in
258/// place the person's checkout on `speccoder/<feature>`.
259pub(crate) fn workspace(
260    path: &Path,
261    feature: Option<&str>,
262    state: &Path,
263    options: &RunOptions,
264) -> CommandResult {
265    let Some(session) = options
266        .session
267        .as_ref()
268        .and_then(|s| s.lock().ok().map(|s| s.clone()))
269    else {
270        if let Some(e) = crate::in_place::session_elsewhere(path, feature, state) {
271            return e.result("run");
272        }
273        return worktree_with(path, feature, state);
274    };
275    CommandResult::new("worktree", CommandStatus::Ok, "in place").with_data(CommandData::Worktree(
276        speccoder_core::WorktreeReport {
277            feature: session.feature.clone(),
278            path: Some(session.checkout.display().to_string()),
279            project_dir: Some(session.project.display().to_string()),
280            branch: Some(session.branch.clone()),
281            base: Some(session.tip.clone()),
282            created: false,
283            refused_paths: Vec::new(),
284        },
285    ))
286}
287
288/// Runs `f` inside an in-place session when one is asked for and none is
289/// open yet: checks and switches the checkout first, switches it back after
290/// — on every way out.
291pub(crate) fn in_session(
292    command: &str,
293    path: &Path,
294    feature: Option<&str>,
295    state: &Path,
296    options: &RunOptions,
297    f: impl FnOnce(&RunOptions) -> CommandResult,
298) -> CommandResult {
299    if !options.in_place || options.session.is_some() {
300        return f(options);
301    }
302    if options.parallel > 1 {
303        return crate::in_place::InPlaceError {
304            code: "IN-PLACE-PARALLEL",
305            message: "an in-place run has one checkout; it cannot run tasks in parallel".to_owned(),
306            paths: Vec::new(),
307            fix: vec!["speccoder run --in-place   (without --parallel)".to_owned()],
308        }
309        .result(command);
310    }
311    let opened = match crate::format::open(path, feature) {
312        Ok(o) => o,
313        Err(r) => return CommandResult::new(command, r.status, &r.message),
314    };
315    let Some(f_name) = opened
316        .snapshot
317        .feature
318        .as_ref()
319        .map(|f| f.feature.name.clone())
320    else {
321        return CommandResult::new(
322            command,
323            CommandStatus::SelectionNeeded,
324            "no feature selected; pass --feature",
325        );
326    };
327    let root = std::path::PathBuf::from(&opened.snapshot.root);
328    let plane_dir = worktree::project_state_dir(state, &root);
329    let policy = crate::tools::policy::Policy::new(&[], &[]).rules(opened.adapter.rules());
330    let mut session = match crate::in_place::Session::begin(&root, &plane_dir, &f_name, &policy) {
331        Ok(s) => s,
332        Err(e) => return e.result(command),
333    };
334    // Ticks a killed run committed but did not write, read on the branch.
335    let format = crate::format::at(&root);
336    if let Some(tasks) = format
337        .inspect(&root, Some(&f_name))
338        .and_then(|snap| snap.feature.and_then(|f| f.tasks))
339    {
340        let file = session.project.join(&tasks.path);
341        session.reconcile(&plane_dir, &file, &tasks, |file, task, hash| {
342            format.tick(file, task, hash)
343        });
344    }
345    let original = session.original.describe();
346    if let Some(p) = &options.progress {
347        p.event(&progress::Event::Notice {
348            text: format!(
349                "in place: on {} in {}; returns to {original} at the end — do not use this checkout until then",
350                session.branch,
351                session.checkout.display()
352            ),
353        });
354        for note in &session.notes {
355            p.event(&progress::Event::Notice { text: note.clone() });
356        }
357    }
358    let shared = Arc::new(std::sync::Mutex::new(session));
359    let inner = RunOptions {
360        session: Some(Arc::clone(&shared)),
361        ..options.clone()
362    };
363    let res = f(&inner);
364    let session = shared.lock().map(|s| s.clone());
365    // An attempt waiting for a person keeps the checkout as it left it: the
366    // session stays, and the next `run --in-place` takes it over and resumes.
367    if let Some(task) = waiting_task(&res) {
368        let notes: String = session
369            .as_ref()
370            .map(|s| s.notes.iter().map(|n| format!("\n{n}")).collect())
371            .unwrap_or_default();
372        let message = format!(
373            "{}{notes}\nthe checkout stays on {} while {task} waits: decide it (`speccoder task approve {task}` or `reject`), then `speccoder run --in-place` continues it — or `speccoder run --in-place --return` gives the checkout back",
374            res.message(),
375            session
376                .as_ref()
377                .map(|s| s.branch.as_str())
378                .unwrap_or("the speccoder branch"),
379        );
380        let status = res.status();
381        return res.reworded(status, &message);
382    }
383    let (back, notes) = match session {
384        Ok(s) => (s.finish(&plane_dir), s.notes),
385        Err(_) => (Ok(()), Vec::new()),
386    };
387    let told: String = notes.iter().map(|n| format!("\n{n}")).collect();
388    match back {
389        Ok(()) => {
390            let message = format!("{} — back on {original}{told}", res.message());
391            let status = res.status();
392            res.reworded(status, &message)
393        }
394        Err(e) => {
395            let message = format!("{}{told}\n{}", res.message(), e.line());
396            res.reworded(CommandStatus::NeedsHuman, &message)
397                .with_code(e.code)
398                .with_fix(e.fix.clone())
399        }
400    }
401}
402
403/// The task whose attempt the result left waiting for a person, if any.
404fn waiting_task(res: &CommandResult) -> Option<String> {
405    let waits = |r: &speccoder_core::RunReport| {
406        r.pending
407            .as_ref()
408            .is_some_and(|p| p["waiting"] == true)
409            .then(|| r.task.clone())
410    };
411    match res.data() {
412        Some(CommandData::Run(r)) => waits(r),
413        Some(CommandData::Feature(f)) => f.attempts.last().and_then(waits),
414        _ => None,
415    }
416}
417
418/// The format's checklist gate: an unchecked item in the feature's
419/// checklists stops a run before any model call, unless a person
420/// overrides it (`run --ignore-checklists`). `Ok(true)`: overridden.
421pub(crate) fn checklist_gate(
422    project: &Path,
423    feature: Option<&str>,
424    ignore: bool,
425) -> Result<bool, Box<CommandResult>> {
426    let format = crate::format::at(project);
427    let Some(snap) = format.inspect(project, feature) else {
428        return Ok(false);
429    };
430    let Some(f) = snap.feature.as_ref() else {
431        return Ok(false);
432    };
433    let lists = format.checklists(Path::new(&snap.root), &f.feature.dir);
434    let open: Vec<String> = lists
435        .iter()
436        .filter(|c| c.open > 0)
437        .map(|c| format!("{} ({} of {} open)", c.file, c.open, c.total))
438        .collect();
439    if open.is_empty() {
440        return Ok(false);
441    }
442    if ignore {
443        return Ok(true);
444    }
445    Err(Box::new(
446        CommandResult::new(
447            "run",
448            CommandStatus::NeedsHuman,
449            &(format.messages().checklists)(&open.join(", ")),
450        )
451        .with_data(CommandData::Checklists(lists)),
452    ))
453}
454
455/// One stage event of `task`; `since` (when the stage began) gives an ended
456/// stage its duration.
457pub(crate) fn stage(
458    live: &dyn crate::progress::Progress,
459    task: &str,
460    stage: speccoder_core::stage::Stage,
461    state: speccoder_core::stage::StageState,
462    detail: Option<String>,
463    since: Option<std::time::Instant>,
464) {
465    stage_with_tail(live, task, stage, state, detail, since, Vec::new());
466}
467
468/// [`stage`] with the last lines of what failed.
469pub(crate) fn stage_with_tail(
470    live: &dyn crate::progress::Progress,
471    task: &str,
472    stage: speccoder_core::stage::Stage,
473    state: speccoder_core::stage::StageState,
474    detail: Option<String>,
475    since: Option<std::time::Instant>,
476    tail: Vec<String>,
477) {
478    live.event(&crate::progress::Event::Stage {
479        task: task.to_owned(),
480        stage,
481        state,
482        detail,
483        elapsed_ms: since.map(|t| u64::try_from(t.elapsed().as_millis()).unwrap_or(u64::MAX)),
484        tail,
485    });
486}
487
488/// The tail of the first failing check of `report`, for a failed stage.
489pub(crate) fn failing_tail(report: &speccoder_core::verify::VerifyReport) -> Vec<String> {
490    report
491        .checks
492        .iter()
493        .find(|c| c.status.is_failure())
494        .map(|c| crate::progress::tail(&format!("{}\n{}", c.stdout_tail, c.stderr_tail)))
495        .unwrap_or_default()
496}
speccoder_runtime/pipeline/mod.rs

speccoder_runtime/pipeline/
mod.rs
