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

speccoder_runtime/pipeline/
mod.rs
