use std::path::Path;
use std::sync::Arc;
use speccoder_core::{CommandData, CommandResult, CommandStatus};
use crate::pipeline::*;
use crate::{controls, convergence, hosting, progress, substrate, worktree};
#[must_use]
pub fn run_feature(
path: &Path,
feature: Option<&str>,
options: &RunOptions,
converge: bool,
) -> CommandResult {
with_driver("run", path, feature, |state, provider, key| {
if converge {
run_converged_with(path, feature, state, provider, Some(key), options)
} else {
run_feature_with(path, feature, state, provider, Some(key), options)
}
})
}
#[must_use]
pub fn converge(path: &Path, feature: Option<&str>) -> CommandResult {
converge_in(path, feature, false)
}
#[must_use]
pub fn converge_in(path: &Path, feature: Option<&str>, in_place: bool) -> CommandResult {
let options = RunOptions {
in_place,
..RunOptions::default()
};
with_driver("converge", path, feature, |state, provider, key| {
convergence::converge_with_options(
path,
feature,
state,
Arc::clone(provider),
Some(key),
&options,
)
})
}
pub(crate) fn with_driver(
command: &str,
path: &Path,
feature: Option<&str>,
f: impl FnOnce(&Path, &Arc<dyn agentplane::model::ModelProvider>, &str) -> CommandResult,
) -> CommandResult {
let Some(state) = worktree::state_dir() else {
return CommandResult::new(
command,
CommandStatus::ConfigError,
"no state directory (set SPECCODER_STATE_DIR or HOME)",
)
.with_code("NO-STATE-DIR");
};
let model = match setup(path, feature, &state, command) {
Ok(s) => s.model,
Err(r) => return *r,
};
let (driver, key) = match substrate::model_config::open_driver(&model) {
Ok((d, k)) => (Ok::<_, String>(d), k.unwrap_or_default()),
Err(e) => return substrate::model_config::refused(command, &e),
};
match driver {
Ok(provider) => f(&state, &provider, &key),
Err(e) => CommandResult::new(command, CommandStatus::ConfigError, &redact(&e, Some(&key))),
}
}
pub fn run_converged_with(
path: &Path,
feature: Option<&str>,
state: &Path,
provider: &Arc<dyn agentplane::model::ModelProvider>,
secret: Option<&str>,
options: &RunOptions,
) -> CommandResult {
if options.in_place && options.session.is_none() {
return in_session("run", path, feature, state, options, |o| {
run_converged_with(path, feature, state, provider, secret, o)
});
}
let mut attempts = Vec::new();
let mut rounds = Vec::new();
let cap = match cap_of(path, options) {
Ok(c) => c,
Err(stop) => return *stop,
};
loop {
let round = RunOptions {
cap: cap.map(|c| c.cap.less(spent(&c, &attempts))),
..options.clone()
};
let res = run_feature_with(path, feature, state, provider, secret, &round);
let Some(CommandData::Feature(fr)) = res.data() else {
return res;
};
let mut fr = (**fr).clone();
attempts.append(&mut fr.attempts);
let finished = res.status() == CommandStatus::Ok && fr.stop.starts_with("feature done");
if !finished {
fr.attempts = attempts;
fr.converge = rounds;
return CommandResult::carried("run", &res)
.with_data(CommandData::Feature(Box::new(fr)));
}
if let Some(p) = &options.progress {
p.event(&progress::Event::Notice {
text: format!("converge: assessing {} against its spec", fr.feature),
});
}
let c = convergence::converge_with_options(
path,
feature,
state,
Arc::clone(provider),
secret,
options,
);
if let Some(p) = &options.progress {
p.event(&progress::Event::Notice {
text: format!("converge: {}", c.message()),
});
}
let outcome = match c.data() {
Some(CommandData::Converge(r)) => {
let outcome = r.outcome.clone();
rounds.push((**r).clone());
outcome
}
_ => String::new(),
};
if outcome != "tasks_appended" {
fr.attempts = attempts;
fr.converge = rounds;
fr.stop = format!("converge: {}", c.message());
return CommandResult::new("run", c.status(), &fr.stop.clone())
.with_data(CommandData::Feature(Box::new(fr)));
}
}
}
pub fn run_feature_with(
path: &Path,
feature: Option<&str>,
state: &Path,
provider: &std::sync::Arc<dyn agentplane::model::ModelProvider>,
secret: Option<&str>,
options: &RunOptions,
) -> CommandResult {
if options.in_place && options.session.is_none() {
return in_session("run", path, feature, state, options, |o| {
run_feature_with(path, feature, state, provider, secret, o)
});
}
let max_tasks = options.max_tasks;
use speccoder_core::FeatureReport;
use speccoder_core::snapshot::TaskState;
let mut attempts = Vec::new();
let mut feature_name = feature.map(str::to_owned);
let hook_lines: std::cell::RefCell<Vec<String>> = std::cell::RefCell::new(Vec::new());
let parallel_lines: std::cell::RefCell<Vec<String>> = std::cell::RefCell::new(Vec::new());
let full_tier: std::cell::RefCell<Option<speccoder_core::FullTier>> =
std::cell::RefCell::new(None);
let finish = |name: Option<String>,
attempts: Vec<speccoder_core::RunReport>,
status: CommandStatus,
stop: String| {
let autopilot_denied: Vec<speccoder_core::oracle::AutopilotDenial> =
if options.mode == speccoder_core::oracle::RunMode::Autopilot {
attempts
.iter()
.flat_map(|a| crate::oracle::autopilot_denials(&a.task, &a.tool_calls))
.collect()
} else {
Vec::new()
};
if let Some(n) = name.as_deref() {
crate::oracle::record_denials(
&worktree::project_state_dir(state, path),
n,
&autopilot_denied,
);
}
let stop = if autopilot_denied.is_empty() {
stop
} else {
format!(
"{stop} — autopilot refused {} call(s) instead of asking; they are listed for you (speccoder inbox)",
autopilot_denied.len()
)
};
CommandResult::new("run", status, &stop).with_data(CommandData::Feature(Box::new(
FeatureReport {
feature: name.unwrap_or_default(),
attempts,
stop,
converge: Vec::new(),
hooks: hook_lines.borrow().clone(),
parallel: parallel_lines.borrow().clone(),
full_tier: full_tier.borrow().clone(),
mode: options.mode,
autopilot_denied,
},
)))
};
let mut hinted: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
let mut passed: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
let mut one_at_a_time: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
let mut parallel_off = false;
let mut budget = max_tasks.unwrap_or(usize::MAX);
if let Err(stop) = checklist_gate(path, feature, options.ignore_checklists) {
return *stop;
}
let cap = match cap_of(path, options) {
Ok(c) => c,
Err(stop) => return *stop,
};
let mut told_uncounted = false;
match hosting::run_hooks(
"before_implement",
"",
path,
state,
provider,
secret,
options,
) {
Ok(lines) => hook_lines.borrow_mut().extend(lines),
Err(stop) => {
return finish(
feature_name,
attempts,
stop.status(),
stop.message().to_owned(),
);
}
}
let mut planned = false;
while budget > 0 {
if options
.cancel
.as_ref()
.is_some_and(|c| c.load(std::sync::atomic::Ordering::SeqCst))
{
return finish(
feature_name,
attempts,
CommandStatus::NeedsHuman,
"cancelled by the person before the next task".to_owned(),
);
}
let wt = workspace(path, feature_name.as_deref(), state, options);
let Some(CommandData::Worktree(w)) = wt.data() else {
return CommandResult::carried("run", &wt);
};
if wt.status() != CommandStatus::Ok {
return CommandResult::carried("run", &wt).with_data(CommandData::Worktree(w.clone()));
}
feature_name = Some(w.feature.clone());
let project = std::path::PathBuf::from(w.project_dir.clone().unwrap_or_default());
let format = crate::format::at(&project);
let Some(snap) = format.inspect(&project, Some(&w.feature)) else {
return CommandResult::new(
"run",
CommandStatus::NotAProject,
format.messages().no_project_in_worktree,
);
};
let schedule = snap
.feature
.as_ref()
.and_then(|f| f.schedule.clone())
.unwrap_or_else(|| speccoder_core::snapshot::Schedule {
tasks: Vec::new(),
totals: speccoder_core::snapshot::ScheduleTotals::default(),
});
if !planned {
planned = true;
if let Some(p) = &options.progress {
let skipped = controls::skipped(&project, &w.feature);
let titles: std::collections::HashMap<&str, &str> = snap
.feature
.as_ref()
.and_then(|f| f.tasks.as_ref())
.map(|t| {
t.tasks
.iter()
.map(|x| (x.id.as_str(), x.title.as_str()))
.collect()
})
.unwrap_or_default();
p.event(&crate::progress::Event::RunStarted {
feature: w.feature.clone(),
tasks: schedule
.tasks
.iter()
.map(|t| crate::progress::PlannedTask {
id: t.id.clone(),
title: titles
.get(t.id.as_str())
.copied()
.unwrap_or_default()
.to_owned(),
state: if skipped.contains_key(&t.id) {
"skipped".to_owned()
} else {
match t.state {
TaskState::Done => "done",
TaskState::Ready => "ready",
_ => "blocked",
}
.to_owned()
},
})
.collect(),
});
let open = schedule
.tasks
.iter()
.filter(|t| t.state != TaskState::Done && !skipped.contains_key(&t.id))
.count();
if let Some(line) = controls::spend_line(path, open, cap.map(|c| c.cap)) {
p.event(&crate::progress::Event::Notice { text: line });
}
if let Some(l) = crate::status::lanes(path, Some(&w.feature)) {
p.event(&crate::progress::Event::Notice { text: l.line() });
}
let warnings = crate::plan_lint::lints(format, &snap, &project).len();
if warnings > 0 {
p.event(&crate::progress::Event::Notice {
text: format!(
"plan: {warnings} warning{} — speccoder doctor --feature {}",
if warnings == 1 { "" } else { "s" },
w.feature
),
});
}
}
}
if let Some(stop) = snap.feature.as_ref().and_then(|f| no_tasks(format, f)) {
return finish(
feature_name,
attempts,
CommandStatus::TaskNotReady,
stop.message,
)
.with_code("NO-TASKS")
.with_fix(stop.fix)
.with_next(stop.next);
}
let skipped = controls::skipped(&project, feature_name.as_deref().unwrap_or_default());
let parsed: Vec<&speccoder_core::snapshot::Task> = snap
.feature
.as_ref()
.and_then(|f| f.tasks.as_ref())
.map(|t| t.tasks.iter().collect())
.unwrap_or_default();
let saved = crate::attempts::list(&project, &w.feature);
let stopped = |id: &str| {
saved
.iter()
.find(|a| a.task == id && !controls::is_set_aside(a))
.filter(|a| a.outcome == "blocked")
.map(|a| a.reason.clone())
};
let yours = |id: &str| {
!hinted.contains(id)
&& (parsed.iter().any(|t| t.id == id && t.yours) || stopped(id).is_some())
};
let Some(next) = schedule.tasks.iter().find(|t| {
t.state == TaskState::Ready && !skipped.contains_key(&t.id) && !passed.contains(&t.id)
}) else {
let open: Vec<String> = schedule
.tasks
.iter()
.filter(|t| t.state != TaskState::Done)
.map(|t| {
if skipped.contains_key(&t.id) {
format!("{} (skipped)", t.id)
} else {
t.id.clone()
}
})
.collect();
return if open.is_empty() {
if attempts
.iter()
.any(|a: &speccoder_core::RunReport| a.ticked)
{
let wt_root = std::path::PathBuf::from(w.path.clone().unwrap_or_default());
let tier = full::run_once(&project, &wt_root);
full::record(&wt_root, feature_name.as_deref().unwrap_or_default(), &tier);
let failed = (!tier.passed).then(|| tier.failed.join(", "));
*full_tier.borrow_mut() = Some(tier);
if let Some(failed) = failed {
return finish(
feature_name,
attempts,
CommandStatus::VerificationFailed,
format!(
"every task is ticked, but the full-tier checks failed: {failed}; fix them, or `speccoder pr --force`"
),
);
}
}
if let (Some(name), Ok(opened)) =
(feature_name.as_deref(), crate::format::open(path, feature))
{
let wt_root = std::path::PathBuf::from(w.path.clone().unwrap_or_default());
let plane = crate::worktree::project_state_dir(
state,
std::path::Path::new(&opened.snapshot.root),
);
if let Err(e) =
crate::held_out::publish(&wt_root, &plane, name, &plane.join("tmp"))
{
hook_lines
.borrow_mut()
.push(format!("held-out sets not published: {e}"));
}
}
match hosting::run_hooks(
"after_implement",
"",
path,
state,
provider,
secret,
options,
) {
Ok(lines) => hook_lines.borrow_mut().extend(lines),
Err(stop) => {
return finish(
feature_name,
attempts,
stop.status(),
stop.message().to_owned(),
);
}
}
finish(
feature_name,
attempts,
CommandStatus::Ok,
"feature done: every task is ticked".to_owned(),
)
} else if let Some(first) = passed.iter().next() {
let fix = speccoder_core::codes::lookup("TASK-YOURS")
.map(|e| {
speccoder_core::codes::fixes(e, "run", Some(first), feature_name.as_deref())
})
.unwrap_or_default();
let mine: Vec<String> = passed.iter().cloned().collect();
let message = match mine.as_slice() {
[one] => format!(
"{one} is yours — {}: a step only a person can take",
parsed
.iter()
.find(|t| t.id == *one)
.map(|t| t.title.as_str())
.unwrap_or_default()
),
many => format!(
"{} are yours: steps only a person can take",
many.join(", ")
),
};
finish(feature_name, attempts, CommandStatus::NeedsHuman, message)
.with_code("TASK-YOURS")
.with_fix(fix)
} else {
finish(
feature_name,
attempts,
CommandStatus::TaskNotReady,
format!("no task can run; open: {}", open.join(", ")),
)
};
};
let title_of = |id: &str| {
parsed
.iter()
.find(|t| t.id == id)
.map(|t| t.title.clone())
.unwrap_or_default()
};
if yours(&next.id) {
let id = next.id.clone();
let stop = crate::approvals::StopRequest {
task: format!("{}/{id}", w.feature),
title: title_of(&id),
claim: stopped(&id),
dependents: speccoder_core::graph::dependents(&schedule, &id),
};
if let Some(failed) = settle_stop(
path,
feature_name.as_deref(),
state,
options,
&stop,
&mut hinted,
&mut passed,
) {
return finish(
feature_name,
attempts,
failed.status(),
format!("{id}: {}", failed.message()),
)
.caused_by(&failed);
}
continue;
}
if let Some(c) = cap
&& let Some(stop) = capped(&c, &attempts, &schedule, &skipped, &w.feature)
{
return finish(feature_name, attempts, CommandStatus::NeedsHuman, stop.0)
.with_code("RUN-CAP")
.with_fix(vec![stop.1]);
}
let left = cap.map(|c| c.cap.less(spent(&c, &attempts)));
let parallel_ids: std::collections::BTreeSet<String> = snap
.feature
.as_ref()
.and_then(|f| f.tasks.as_ref())
.map(|t| {
t.tasks
.iter()
.filter(|t| t.parallel)
.map(|t| t.id.clone())
.collect()
})
.unwrap_or_default();
let group: Vec<String> = schedule
.tasks
.iter()
.filter(|t| t.state == TaskState::Ready && !skipped.contains_key(&t.id))
.filter(|t| {
parallel_ids.contains(&t.id) && !one_at_a_time.contains(&t.id) && !yours(&t.id)
})
.filter(|t| {
!super::story::oracle_pending(
std::path::Path::new(w.path.as_deref().unwrap_or_default()),
&crate::worktree::project_state_dir(state, path),
format,
&snap,
&t.id,
)
})
.map(|t| t.id.clone())
.take(options.parallel.min(budget))
.collect();
if !parallel_off && group.len() >= 2 {
budget = budget.saturating_sub(group.len());
let feature = feature_name.clone().unwrap_or_default();
let feature_wt = std::path::PathBuf::from(w.path.clone().unwrap_or_default());
let shared = RunOptions {
left: left.map(|l| l.less(l.amount() - l.amount() / group.len() as u64)),
..options.clone()
};
let lanes = run_lanes(path, &feature, &group, state, provider, secret, &shared);
let outcome = merge_lanes(path, &feature_wt, &project, &feature, lanes);
attempts.extend(outcome.reports);
parallel_lines.borrow_mut().extend(outcome.lines);
one_at_a_time.extend(outcome.one_at_a_time);
parallel_off |= outcome.reverted;
if let Some((task, res)) = outcome.stop {
let claim = attempts
.iter()
.rev()
.find(|a| a.task == task && a.attempt_status == "blocked")
.map(|a| a.reason.clone().unwrap_or_default());
let Some(claim) = claim else {
return finish(
feature_name,
attempts,
res.status(),
format!("{task}: {}", res.message()),
)
.caused_by(&res);
};
let stop = crate::approvals::StopRequest {
task: format!("{}/{task}", w.feature),
title: title_of(&task),
claim: Some(claim),
dependents: speccoder_core::graph::dependents(&schedule, &task),
};
if let Some(failed) = settle_stop(
path,
feature_name.as_deref(),
state,
options,
&stop,
&mut hinted,
&mut passed,
) {
return finish(
feature_name,
attempts,
failed.status(),
format!("{task}: {}", failed.message()),
)
.caused_by(&failed);
}
}
continue;
}
budget -= 1;
let task = next.id.clone();
let res = run_task(
path,
feature_name.as_deref(),
&task,
state,
std::sync::Arc::clone(provider),
secret,
&RunOptions {
left,
..options.clone()
},
);
if let Some(CommandData::Run(r)) = res.data() {
if cap.is_some()
&& !told_uncounted
&& !r.run_id.is_empty()
&& super::commit::all_usage(r).is_none()
&& let Some(p) = &options.progress
{
told_uncounted = true;
p.event(&crate::progress::Event::Notice {
text: "this run's agent reports no usage: the cap cannot count its attempts"
.to_owned(),
});
}
attempts.push((**r).clone());
}
if let Some(c) = cap
&& res.status() != CommandStatus::Ok
&& let Some(stop) = capped(&c, &attempts, &schedule, &skipped, &w.feature)
{
return finish(feature_name, attempts, CommandStatus::NeedsHuman, stop.0)
.with_code("RUN-CAP")
.with_fix(vec![stop.1]);
}
let claim = match res.data() {
Some(CommandData::Run(r)) if r.attempt_status == "blocked" => {
Some(r.reason.clone().unwrap_or_default())
}
_ => None,
};
if let Some(claim) = claim.filter(|_| res.status() != CommandStatus::Ok) {
let stop = crate::approvals::StopRequest {
task: format!("{}/{task}", w.feature),
title: title_of(&task),
claim: Some(claim),
dependents: speccoder_core::graph::dependents(&schedule, &task),
};
if let Some(failed) = settle_stop(
path,
feature_name.as_deref(),
state,
options,
&stop,
&mut hinted,
&mut passed,
) {
return finish(
feature_name,
attempts,
failed.status(),
format!("{task}: {}", failed.message()),
)
.caused_by(&failed);
}
continue;
}
if res.status() != CommandStatus::Ok {
return finish(
feature_name,
attempts,
res.status(),
format!("{task}: {}", res.message()),
)
.caused_by(&res);
}
}
finish(
feature_name,
attempts,
CommandStatus::Ok,
"stopped at --max-tasks".to_owned(),
)
}
fn cap_of(
path: &Path,
options: &RunOptions,
) -> Result<Option<substrate::attempt::Left>, Box<CommandResult>> {
let text = crate::verify::find_config(path)
.and_then(|d| std::fs::read_to_string(d.join(crate::verify::config::FILE)).ok())
.unwrap_or_default();
let configured = substrate::run_config::parse(&text).ok().and_then(|c| c.cap);
let Some(cap) = options.cap.or(configured) else {
return Ok(None);
};
let prices = substrate::model_config::parse(&text)
.ok()
.flatten()
.and_then(|m| m.prices);
if matches!(cap, substrate::run_config::Cap::Usd(_)) && prices.is_none() {
let fix = speccoder_core::codes::lookup("RUN-CAP-UNPRICED")
.map(|e| speccoder_core::codes::fixes(e, "run", None, None))
.unwrap_or_default();
return Err(Box::new(
CommandResult::new(
"run",
CommandStatus::ConfigError,
&format!(
"a cap of {cap} needs the model's prices: set them in [model], or cap tokens"
),
)
.with_code("RUN-CAP-UNPRICED")
.with_fix(fix),
));
}
Ok(Some(substrate::attempt::Left { cap, prices }))
}
fn spent(cap: &substrate::attempt::Left, attempts: &[speccoder_core::RunReport]) -> u64 {
attempts
.iter()
.map(|a| {
cap.spent(&substrate::attempt::usage_of(
super::commit::all_usage(a).as_ref(),
))
})
.sum()
}
fn capped(
cap: &substrate::attempt::Left,
attempts: &[speccoder_core::RunReport],
schedule: &speccoder_core::snapshot::Schedule,
skipped: &std::collections::BTreeMap<String, String>,
feature: &str,
) -> Option<(String, String)> {
let used = spent(cap, attempts);
if used < cap.cap.amount() {
return None;
}
let open: Vec<&str> = schedule
.tasks
.iter()
.filter(|t| {
t.state != speccoder_core::snapshot::TaskState::Done && !skipped.contains_key(&t.id)
})
.map(|t| t.id.as_str())
.collect();
let used = cap.cap.with(used);
Some((
format!(
"the run spent its cap: {} of {}; {} left — a new run starts its cap over",
used.rounded(),
cap.cap,
if open.is_empty() {
"nothing".to_owned()
} else {
open.join(", ")
}
),
format!(
"speccoder run --cap {} --feature {feature}",
cap.cap.doubled()
),
))
}
fn settle_stop(
path: &Path,
feature: Option<&str>,
state: &Path,
options: &RunOptions,
stop: &crate::approvals::StopRequest,
hinted: &mut std::collections::BTreeSet<String>,
passed: &mut std::collections::BTreeSet<String>,
) -> Option<CommandResult> {
let id = stop
.task
.rsplit('/')
.next()
.unwrap_or(&stop.task)
.to_owned();
match ask_at_stop(path, feature, state, options, stop) {
Some((answer, done)) if done.status() == CommandStatus::Ok => {
if matches!(answer, crate::approvals::StopAnswer::Hint(_)) {
hinted.insert(id);
}
None
}
Some((_, failed)) => Some(failed),
None => {
passed.insert(id);
None
}
}
}
fn ask_at_stop(
path: &Path,
feature: Option<&str>,
state: &Path,
options: &RunOptions,
stop: &crate::approvals::StopRequest,
) -> Option<(crate::approvals::StopAnswer, CommandResult)> {
let approver = options
.approver
.as_ref()
.filter(|_| options.lane.is_none())?;
let answer = approver.at_stop(stop)?;
let task = stop.task.rsplit('/').next().unwrap_or(&stop.task);
let done = crate::actions::answered_at_stop(
path,
feature,
task,
&answer,
(approver.via(), options),
state,
);
Some((answer, done))
}
#[derive(Debug, Clone)]
pub(crate) struct NoTasks {
pub message: String,
pub fix: Vec<String>,
pub next: Vec<speccoder_core::next::NextStep>,
}
pub(crate) fn no_tasks(
format: &dyn crate::format::Adapter,
feature: &speccoder_core::snapshot::FeatureSnapshot,
) -> Option<NoTasks> {
if feature.tasks.as_ref().is_some_and(|t| !t.tasks.is_empty()) {
return None;
}
let messages = format.messages();
let skills: Vec<String> = (format.planning().toward_tasks)(feature.artifacts.plan)
.iter()
.map(|s| (*s).to_owned())
.collect();
let name = &feature.feature.name;
Some(NoTasks {
message: format!(
"no tasks yet: {name} {}; {}",
if feature.tasks.is_some() {
format!("has a {} without a task", messages.tasks_file)
} else {
format!("has no {}", messages.tasks_file)
},
messages.make_tasks
),
fix: skills
.iter()
.map(|s| format!("speccoder spec {s}"))
.collect(),
next: speccoder_core::next::next(
&speccoder_core::next::Outcome::Planning { skills },
Some(name),
),
})
}