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

speccoder_runtime/pipeline/
mod.rs
