use std::path::{Path, PathBuf};
use std::sync::Arc;
use agentplane::model::ModelProvider;
use serde_json::Value;
use speccoder_core::{CommandData, CommandResult, CommandStatus, ConvergeReport};
use crate::{redact, substrate, tools, verify, worktree};
const SEVERITY: [&str; 4] = ["CRITICAL", "HIGH", "MEDIUM", "LOW"];
const STOP: [&str; 24] = [
"a", "an", "the", "and", "or", "so", "to", "of", "for", "in", "on", "is", "are", "be", "it",
"its", "every", "each", "must", "only", "per", "that", "this", "with",
];
fn words(text: &str) -> std::collections::BTreeSet<String> {
text.to_lowercase()
.split(|c: char| !c.is_alphanumeric())
.filter(|w| !w.is_empty() && !STOP.contains(w))
.map(str::to_owned)
.collect()
}
fn overlap(a: &str, b: &str) -> f64 {
let (a, b) = (words(a), words(b));
let union = a.union(&b).count();
if union == 0 {
return 0.0;
}
#[allow(clippy::cast_precision_loss)]
let share = a.intersection(&b).count() as f64 / union as f64;
share
}
#[must_use]
pub fn same_task(a: &str, b: &str) -> bool {
let (a, b) = (words(a), words(b));
let union = a.union(&b).count();
union > 0 && a.intersection(&b).count() * 2 >= union
}
fn cite(id: &str) -> String {
match id.split_once('-') {
Some((story, n))
if story.starts_with("US")
&& !n.is_empty()
&& n.chars().all(|c| c.is_ascii_digit()) =>
{
format!("{story}/AC{n}")
}
_ => format!("(scenarios {id})"),
}
}
fn scenario_cites(text: &str) -> Vec<String> {
let mut out: Vec<String> = text
.split(|c: char| !(c.is_alphanumeric() || c == '/' || c == '-'))
.filter_map(|w| {
let w = w.trim_matches('-').to_ascii_uppercase();
let (story, n) = w.split_once("/AC").or_else(|| w.split_once('-'))?;
let num = story.strip_prefix("US")?;
(!num.is_empty()
&& num.chars().all(|c| c.is_ascii_digit())
&& !n.is_empty()
&& n.chars().all(|c| c.is_ascii_digit()))
.then(|| format!("{story}-{n}"))
})
.collect();
out.sort();
out.dedup();
out
}
#[must_use]
pub fn task_line(finding: &Value) -> String {
let text = |k: &str| finding[k].as_str().unwrap_or_default().trim().to_owned();
format!(
"{} per {} ({})",
text("task").trim_end_matches('.'),
text("source_ref"),
text("gap_type")
)
}
pub fn converge_with(
path: &Path,
feature: Option<&str>,
state: &Path,
provider: Arc<dyn ModelProvider>,
secret: Option<&str>,
) -> CommandResult {
converge_with_options(
path,
feature,
state,
provider,
secret,
&crate::RunOptions::default(),
)
}
pub fn converge_with_options(
path: &Path,
feature: Option<&str>,
state: &Path,
provider: Arc<dyn ModelProvider>,
secret: Option<&str>,
options: &crate::RunOptions,
) -> CommandResult {
if options.in_place && options.session.is_none() {
return crate::pipeline::in_session("converge", path, feature, state, options, |o| {
converge_with_options(path, feature, state, Arc::clone(&provider), secret, o)
});
}
let before = match crate::hosting::run_hooks(
"before_converge",
"",
path,
state,
&provider,
secret,
options,
) {
Ok(lines) => lines,
Err(stop) => return CommandResult::carried("converge", &stop),
};
let result = assess(path, feature, state, Arc::clone(&provider), secret, options);
let Some(CommandData::Converge(report)) = result.data() else {
return result;
};
let mut report = (**report).clone();
report.hooks = before;
let (status, message) = (result.status(), result.message().to_owned());
if status == CommandStatus::Ok {
match crate::hosting::run_hooks(
"after_converge",
"",
path,
state,
&provider,
secret,
options,
) {
Ok(lines) => report.hooks.extend(lines),
Err(stop) => {
return CommandResult::carried("converge", &stop)
.with_data(CommandData::Converge(Box::new(report)));
}
}
}
CommandResult::new("converge", status, &message)
.with_data(CommandData::Converge(Box::new(report)))
}
fn changed_requirements(project: &Path, feature: &str) -> Vec<String> {
let report = crate::status::report(project, Some(feature));
let Some(CommandData::Report(r)) = report.data() else {
return Vec::new();
};
r.tasks
.iter()
.filter_map(|t| t.drift.as_ref().map(|d| (t, d)))
.flat_map(|(t, d)| {
d.requirements.iter().map(move |q| {
let quote = |l: &Option<String>| {
l.as_deref()
.map_or_else(|| "(absent)".to_owned(), |l| format!("\"{}\"", l.trim()))
};
format!(
"- {} {}: was {}; now {}",
t.id,
q.id,
quote(&q.then),
quote(&q.now)
)
})
})
.collect()
}
fn assess(
path: &Path,
feature: Option<&str>,
state: &Path,
provider: Arc<dyn ModelProvider>,
secret: Option<&str>,
options: &crate::RunOptions,
) -> CommandResult {
let refuse = |status: CommandStatus, m: String| {
CommandResult::new("converge", status, &redact(&m, secret))
};
let s = match crate::setup(path, feature, state, "converge") {
Ok(s) => s,
Err(r) => return *r,
};
let Some(f) = s.snapshot.feature.as_ref() else {
return refuse(
CommandStatus::SelectionNeeded,
"no feature selected; pass --feature".to_owned(),
);
};
let feature_name = f.feature.name.clone();
let wt = crate::pipeline::workspace(path, Some(&feature_name), state, options);
let Some(CommandData::Worktree(w)) = wt.data() else {
return CommandResult::carried("converge", &wt);
};
if wt.status() != CommandStatus::Ok {
return CommandResult::carried("converge", &wt).with_data(CommandData::Worktree(w.clone()));
}
let (Some(wt_path), Some(project_dir)) = (w.path.clone(), w.project_dir.clone()) else {
return refuse(CommandStatus::Error, "worktree without a path".to_owned());
};
let wt_root = PathBuf::from(&wt_path);
let project = PathBuf::from(&project_dir);
if options.session.is_none()
&& let Err(r) = worktree::require_clean(&wt_root)
{
return refuse(
CommandStatus::GitPrecondition,
format!(
"the worktree has uncommitted changes ({}); a failed attempt's work? Commit or undo it in {wt_path}",
r.paths
.iter()
.map(|(p, _)| p.as_str())
.collect::<Vec<_>>()
.join(", ")
),
)
.with_code("WORKTREE-DIRTY");
}
let format = crate::format::at(&project);
let Some(snap) = format.inspect(&project, Some(&feature_name)) else {
return refuse(
CommandStatus::NotAProject,
format.messages().no_project_in_worktree.to_owned(),
);
};
let Some(tasks) = snap.feature.as_ref().and_then(|f| f.tasks.as_ref()) else {
return refuse(
CommandStatus::NotFound,
format!(
"{feature_name} has no {}; {}",
format.messages().tasks_file,
format.messages().make_tasks
),
);
};
let open: Vec<&str> = tasks
.tasks
.iter()
.filter(|t| !t.done)
.map(|t| t.id.as_str())
.collect();
if !open.is_empty() {
return refuse(
CommandStatus::TaskNotReady,
format!(
"open tasks: {}; converge assesses a finished feature — run it first",
open.join(", ")
),
);
}
let tasks_path = project.join(&tasks.path);
let feature_dir = tasks_path
.parent()
.map_or_else(|| project.clone(), Path::to_path_buf);
let read = |p: &Path| std::fs::read_to_string(p).unwrap_or_default();
let tasks_text = read(&tasks_path);
let tasks_hash = crate::integrity::sha256_hex(tasks_text.as_bytes());
let Some(config_root) = verify::find_config(&project) else {
return refuse(
CommandStatus::ConfigError,
"the worktree has no speccoder.toml (is it committed?)".to_owned(),
)
.with_code("CONFIG-MISSING");
};
let config_text =
std::fs::read_to_string(config_root.join(verify::config::FILE)).unwrap_or_default();
let run_config = match substrate::run_config::parse(&config_text) {
Ok(c) => c,
Err(e) => {
return refuse(CommandStatus::ConfigError, format!("speccoder.toml: {e}"))
.with_code("CONFIG-INVALID");
}
};
let is_convergence = |phase: &str| format.is_convergence(phase);
let done_rounds = tasks
.phases
.iter()
.filter(|p| is_convergence(&p.title))
.count();
let round = u32::try_from(done_rounds)
.unwrap_or(u32::MAX)
.saturating_add(1);
let mut report = ConvergeReport {
feature: feature_name.clone(),
run_id: None,
round,
outcome: String::new(),
findings: Vec::new(),
appended: Vec::new(),
unrequested: Vec::new(),
repeated: Vec::new(),
dropped: Vec::new(),
stories: Vec::new(),
commit: None,
usage: None,
tool_calls: Vec::new(),
hooks: Vec::new(),
};
let done = |report: ConvergeReport, status: CommandStatus, message: String| {
CommandResult::new("converge", status, &redact(&message, secret))
.with_data(CommandData::Converge(Box::new(report)))
};
if round > run_config.max_converge_rounds {
report.outcome = "round_limit".to_owned();
return done(
report,
CommandStatus::NeedsHuman,
format!(
"{done_rounds} convergence rounds already ([converge] max_rounds = {}); a person decides what remains",
run_config.max_converge_rounds
),
);
}
let stories: Vec<String> = if !run_config.acceptance {
Vec::new()
} else if format.kind() == speccoder_core::spec::FormatKind::Openspec {
vec![feature_name.clone()]
} else {
let mut keys: Vec<String> = tasks.tasks.iter().filter_map(|t| t.story.clone()).collect();
keys.sort();
keys.dedup();
keys
};
let mut holds = std::collections::BTreeSet::new();
let mut red = std::collections::BTreeSet::new();
let mut evidence: Vec<Value> = Vec::new();
for key in &stories {
let checked = crate::pipeline::story::checked(
path,
Some(&feature_name),
key,
state,
&provider,
options,
);
let c = match checked {
Ok(c) if c.finished => c,
Ok(_) => continue,
Err(r) => {
report
.stories
.push(format!("{key}: no oracle — {}", r.message()));
continue;
}
};
let green = c
.scenarios
.iter()
.filter(|s| c.states.get(&s.id) == Some(&speccoder_core::story::ScenarioState::Green))
.count();
report.usage = crate::pipeline::commit::add_usage(report.usage.as_ref(), c.usage.as_ref());
report.stories.push(format!(
"{key}: {green} of {} hold{}",
c.scenarios.len(),
if c.wrote { " (oracle written now)" } else { "" }
));
for sc in &c.scenarios {
match c.states.get(&sc.id) {
Some(speccoder_core::story::ScenarioState::Green) => {
holds.insert(sc.id.clone());
}
Some(speccoder_core::story::ScenarioState::Red)
if !c.untestable.contains(&sc.id) =>
{
red.insert(sc.id.clone());
evidence.push(serde_json::json!({
"task": format!("Make {} hold through the story's interface", sc.id),
"source_ref": cite(&sc.id),
"gap_type": "scenario",
"severity": "HIGH",
"source": "oracle",
"evidence": sc.text,
}));
}
_ => {}
}
}
}
let tmp = s.plane_dir.join("tmp");
let files: Vec<String> = worktree::checkpoint(&wt_root, &tmp)
.and_then(|c| worktree::files(&wt_root, &c))
.unwrap_or_default()
.into_iter()
.take(400)
.collect();
let [constitution, spec, plan, _] = format.convergence_texts(&project, &feature_dir);
let changed = changed_requirements(&project, &feature_name);
let input = substrate::converge::input(
&feature_name,
[&constitution, &spec, &plan, &tasks_text],
&files,
&changed,
run_config.max_turns,
run_config.max_tool_calls,
);
let rt = match crate::tokio_runtime() {
Ok(rt) => rt,
Err(e) => return refuse(CommandStatus::RunFailed, e),
};
let runner = match crate::sandbox::runner_at(&wt_root) {
Ok((r, _)) => r,
Err(e) => return refuse(CommandStatus::ConfigError, e),
};
let policy = tools::policy::Policy::new(&[], &[]).rules(format.rules());
let (policy, runner) = match options.session.as_ref().and_then(|s| s.lock().ok()) {
Some(s) => {
if let Err(e) = s.check(&policy) {
return e.result("converge");
}
let (policy, runner, _) = s.confine(policy, runner);
(policy, runner)
}
None => (policy, runner),
};
let server = match tools::WorkspaceTools::new(&wt_root, policy) {
Ok(t) => Arc::new(t.with_runner(runner)),
Err(e) => return refuse(CommandStatus::Error, e.to_string()),
};
let outcome = rt.block_on(async {
let plane = substrate::Plane::open_with(
&s.plane_dir,
&s.model,
provider,
Some(server),
substrate::Hooks {
format: Some(format),
..substrate::Hooks::default()
},
)?;
plane.converge(&feature_name, input).await
});
let outcome = match outcome {
Ok(o) => o,
Err(e) => return refuse(CommandStatus::RunFailed, e.0),
};
report.run_id = Some(outcome.run_id.to_string());
let output = outcome
.output
.as_ref()
.map(|o| o.peek().clone())
.unwrap_or(Value::Null);
report.usage = crate::pipeline::commit::add_usage(report.usage.as_ref(), output.get("usage"));
report.tool_calls = output["tool_calls"].as_array().cloned().unwrap_or_default();
if output["status"] != "completed" {
report.outcome = "failed".to_owned();
let why = output["reason"]
.as_str()
.map(str::to_owned)
.unwrap_or_else(|| format!("{:?}", outcome.status));
return done(
report,
CommandStatus::NeedsHuman,
format!("the assessment did not finish: {why}"),
);
}
report.findings = output["report"]["findings"]
.as_array()
.cloned()
.unwrap_or_default();
let earlier: Vec<String> = tasks
.tasks
.iter()
.filter(|t| t.phase.as_deref().is_some_and(is_convergence))
.map(|t| {
t.title
.rsplit_once(" per ")
.map_or(t.title.clone(), |(task, _)| task.to_owned())
})
.collect();
let ticked: Vec<&speccoder_core::snapshot::Task> = tasks
.tasks
.iter()
.filter(|t| t.done && !t.phase.as_deref().is_some_and(is_convergence))
.collect();
let mut actionable: Vec<Value> = Vec::new();
for finding in &evidence {
if earlier
.iter()
.any(|e| same_task(e, finding["task"].as_str().unwrap_or_default()))
{
report.repeated.push(finding.clone());
} else {
actionable.push(finding.clone());
}
}
for finding in &report.findings {
let text = format!(
"{} {}",
finding["task"].as_str().unwrap_or_default(),
finding["source_ref"].as_str().unwrap_or_default()
);
let cites = scenario_cites(&text);
let said = finding["task"].as_str().unwrap_or_default();
let restates = ticked
.iter()
.find(|t| {
same_task(said, &t.title)
|| (text
.split(|c: char| !c.is_alphanumeric())
.any(|w| w == t.id)
&& overlap(said, &t.title) >= 0.3)
})
.map(|t| t.id.as_str());
let why = if !cites.is_empty() && cites.iter().all(|c| holds.contains(c)) {
Some(format!(
"{} already hold{}",
cites.join(", "),
if cites.len() == 1 { "s" } else { "" }
))
} else if cites.iter().any(|c| red.contains(c)) {
Some("its scenario's gap is appended from the oracle".to_owned())
} else {
restates
.filter(|_| !cites.iter().any(|c| red.contains(c)))
.map(|id| format!("restates {id}, which is done, with no red scenario"))
};
if let Some(why) = why {
let mut dropped = finding.clone();
dropped["why"] = Value::String(why);
report.dropped.push(dropped);
} else if finding["gap_type"] == "unrequested" {
report.unrequested.push(finding.clone());
} else if earlier
.iter()
.any(|e| same_task(e, finding["task"].as_str().unwrap_or_default()))
{
report.repeated.push(finding.clone());
} else {
actionable.push(finding.clone());
}
}
actionable.sort_by_key(|f| {
SEVERITY
.iter()
.position(|s| f["severity"] == *s)
.unwrap_or(SEVERITY.len())
});
let for_person = if report.unrequested.is_empty() {
String::new()
} else {
format!(
"; {} unrequested finding(s) for a person",
report.unrequested.len()
)
};
if actionable.is_empty() {
return if report.repeated.is_empty() && !report.unrequested.is_empty() {
report.outcome = "needs_review".to_owned();
let n = report.unrequested.len();
done(
report,
CommandStatus::NeedsHuman,
format!("no tasks appended; {n} unrequested finding(s) for a person to review"),
)
} else if report.repeated.is_empty() {
report.outcome = "converged".to_owned();
done(
report,
CommandStatus::Ok,
format!(
"converged: the implementation satisfies the spec, plan and tasks{for_person}"
),
)
} else {
report.outcome = "no_progress".to_owned();
let n = report.repeated.len();
done(
report,
CommandStatus::NeedsHuman,
format!("no progress: {n} finding(s) repeat earlier convergence tasks{for_person}"),
)
};
}
let items: Vec<String> = actionable.iter().map(task_line).collect();
let tasks_before = std::fs::read(&tasks_path).unwrap_or_default();
let ids = match format.append(&tasks_path, &tasks_hash, &items) {
Ok(ids) => ids,
Err(e) => {
report.outcome = "failed".to_owned();
return done(report, CommandStatus::Error, e);
}
};
report.appended = ids
.iter()
.zip(&items)
.map(|(id, text)| format!("{id} {text}"))
.collect();
let message = format!(
"speccoder: converge round {round} ({} task{})\n\n{}\n\nSpec-Feature: {feature_name}\nRun: {}\n{}",
ids.len(),
if ids.len() == 1 { "" } else { "s" },
report.appended.join("\n"),
report.run_id.clone().unwrap_or_default(),
crate::attempts::usage_trailer(report.usage.as_ref())
);
let committed = match options.session.as_ref().and_then(|s| s.lock().ok()) {
None => crate::commit_attempt(&wt_root, &message),
Some(mut s) => {
let policy = tools::policy::Policy::new(&[], &[]).rules(format.rules());
let digest =
crate::in_place::intent_digest(&project, &policy, format.messages().tasks_file);
s.commit_round(
&plane_dir_of(state, &project),
&tasks_path,
&tasks_before,
&message,
&digest,
)
}
};
match committed {
Ok(id) => report.commit = Some(id),
Err(e) => {
report.outcome = "failed".to_owned();
return done(report, CommandStatus::Error, format!("commit failed: {e}"));
}
}
report.outcome = "tasks_appended".to_owned();
let range = match (ids.first(), ids.last()) {
(Some(a), Some(b)) if a != b => format!("{a}–{b}"),
(Some(a), _) => a.clone(),
_ => String::new(),
};
done(
report,
CommandStatus::Ok,
format!(
"appended {range} as Phase: Convergence (round {round}); run the feature again{for_person}"
),
)
}
fn plane_dir_of(state: &Path, project: &Path) -> PathBuf {
worktree::project_state_dir(state, project)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn rewordings_are_the_same_task_and_distinct_tasks_are_not() {
let t005 = "Add tests and tighten pre-release validation so every identifier is rejected unless it is non-empty ASCII letters, digits, or hyphen only";
let t007 = "Tighten identifier validation so every pre-release and build identifier is non-empty and consists only of ASCII letters, digits, or hyphen";
let t003 = "Tighten build metadata parsing so every dot-separated build identifier must be non-empty and valid, and reject inputs with empty build identifiers";
let t006 = "Add unit tests for every invalid-input category named in the specification and ensure compare returns None for each of them";
assert!(same_task(t005, t007));
assert!(!same_task(t003, t007));
assert!(!same_task(t006, t007));
assert!(!same_task("", ""));
}
}