diff --git a/crates/ai/src/agent.rs b/crates/ai/src/agent.rs index f9a45d9b..d732a488 100644 --- a/crates/ai/src/agent.rs +++ b/crates/ai/src/agent.rs @@ -8,7 +8,7 @@ use crate::context; use crate::model::{Block, ChatModel, Delta, Effort, Msg, Speaker, Stop, StopReason, ToolDef, TurnRequest, Usage}; use crate::price::{self, Meter}; use crate::prompt; -use crate::{AgentEvent, Conversation}; +use crate::{AgentEvent, Conversation, TurnFailure}; use craft_anything_engine::{CallOptions, Caller, CancelToken, Consent, Engine, Registry, ToolOutput, Toolset}; use craft_anything_evidence::{ErrorCode, ToolError}; use craft_anything_platform::sync::Mutex; @@ -276,6 +276,7 @@ pub struct AgentLoop { last_input: u64, /// Estimated history size sent with that turn, before its reply and tool results were added. last_history_estimate: u64, + failure: Arc>>, } impl AgentLoop { @@ -300,16 +301,26 @@ impl AgentLoop { /// Handle one user message to completion. pub fn handle(&mut self, text: &str) { + self.handle_turn(Some(text)); + } + + /// A retry has no new user message: keep completed tool calls and their results, then + /// continue the failed model request using the same history. + fn handle_turn(&mut self, text: Option<&str>) { self.stop.reset(); let cancel = CancelToken::default(); *self.cancel.lock().unwrap_or_else(PoisonError::into_inner) = cancel.clone(); - self.host.journal("user_message", json!({"text": crate::journal_text(text)})); + if let Some(text) = text { + self.host.journal("user_message", json!({"text": crate::journal_text(text)})); + } if self.system.is_none() { self.emit(AgentEvent::Status { text: "Reading the investigation so far…".into() }); let inv = self.host.investigation(); self.system = Some(prompt::system_prompt(self.cfg.beginner, self.host.clean_room(), &inv)); } - context::push_user_text(&mut self.history, text); + if let Some(text) = text { + context::push_user_text(&mut self.history, text); + } let configured = self.chain.get(self.current).and_then(|s| s.effort); let effort = prompt::choose_effort(configured); let max_turns = self.chain.get(self.current).map(|s| s.max_turns).unwrap_or(DEFAULT_MAX_TURNS).max(1); @@ -564,6 +575,7 @@ impl AgentLoop { self.emit(AgentEvent::Status { text: format!("{name} failed ({}). Trying {next}…", e.message) }); continue; } + *self.failure.lock().unwrap_or_else(PoisonError::into_inner) = Some(TurnFailure::from_api(&e)); self.emit(AgentEvent::Error { message: format!("{name}: {}", e.message), remediation: e.remediation, log: None }); return None; } @@ -685,7 +697,7 @@ fn wait_answer(rx: &Receiver, stop: &Stop) -> Option { /// A conversation with an API backend: a worker thread runs [`AgentLoop`]. pub struct ApiConversation { - to_worker: Sender, + to_worker: Sender>, events: Receiver, busy: Arc, stop: Stop, @@ -693,6 +705,7 @@ pub struct ApiConversation { waiters: Waiters, queued: Vec, resumed: bool, + failure: Arc>>, } impl ApiConversation { @@ -706,11 +719,12 @@ impl ApiConversation { return Err(ToolError::unavailable("No AI backend could be started.").with_remediation("Add one in Settings → AI.")); } let (etx, erx) = std::sync::mpsc::channel(); - let (wtx, wrx) = std::sync::mpsc::channel::(); + let (wtx, wrx) = std::sync::mpsc::channel::>(); let busy = Arc::new(AtomicBool::new(false)); let stop = Stop::default(); let cancel = Arc::new(Mutex::new(CancelToken::default())); let waiters: Waiters = Arc::default(); + let failure: Arc>> = Arc::default(); // A kept conversation continues where it stopped. let history = cfg.transcript.as_deref().and_then(context::load).unwrap_or_default(); let resumed = !history.is_empty(); @@ -729,13 +743,14 @@ impl ApiConversation { meter: Meter::default(), last_input: 0, last_history_estimate: 0, + failure: failure.clone(), }; let busy2 = busy.clone(); std::thread::Builder::new() .name("ai-agent".into()) .spawn(move || { while let Ok(text) = wrx.recv() { - let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| agent.handle(&text))); + let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| agent.handle_turn(text.as_deref()))); if r.is_err() { let _ = etx.send(AgentEvent::Error { message: "The assistant hit a bug and stopped safely.".into(), @@ -747,19 +762,18 @@ impl ApiConversation { } }) .map_err(|e| ToolError::internal(format!("couldn't start the AI worker: {e}")))?; - Ok(ApiConversation { to_worker: wtx, events: erx, busy, stop, cancel, waiters, queued: Vec::new(), resumed }) + Ok(ApiConversation { to_worker: wtx, events: erx, busy, stop, cancel, waiters, queued: Vec::new(), resumed, failure }) } /// It continues a conversation kept in its transcript file. pub fn resumed(&self) -> bool { self.resumed } -} -impl Conversation for ApiConversation { - fn send(&mut self, text: &str) { + fn submit(&mut self, text: Option) { + *self.failure.lock().unwrap_or_else(PoisonError::into_inner) = None; self.busy.store(true, Ordering::SeqCst); - if self.to_worker.send(text.to_string()).is_err() { + if self.to_worker.send(text).is_err() { self.busy.store(false, Ordering::SeqCst); self.queued.push(AgentEvent::Error { message: "The assistant isn't running any more.".into(), @@ -768,6 +782,20 @@ impl Conversation for ApiConversation { }); } } +} + +impl Conversation for ApiConversation { + fn send(&mut self, text: &str) { + self.submit(Some(text.to_string())); + } + + fn retry(&mut self, _text: &str) { + self.submit(None); + } + + fn last_failure(&self) -> Option { + *self.failure.lock().unwrap_or_else(PoisonError::into_inner) + } fn poll(&mut self) -> Vec { let mut v = std::mem::take(&mut self.queued); diff --git a/crates/ai/src/anthropic.rs b/crates/ai/src/anthropic.rs index 9a6f5871..432b0262 100644 --- a/crates/ai/src/anthropic.rs +++ b/crates/ai/src/anthropic.rs @@ -253,9 +253,9 @@ impl StreamState { "authentication_error" | "permission_error" => ErrorCode::NoPermission, _ => ErrorCode::ProgramFailed, }; - return Err( - ToolError::new(code, format!("Anthropic stream error ({et}): {msg}")).with_remediation("Try again; if it keeps happening, switch backend.") - ); + return Err(ToolError::new(code, format!("Anthropic stream error ({et}): {msg}")) + .with_remediation("Try again; if it keeps happening, switch backend.") + .with_details(json!({"provider_error": et}))); } _ => {} } diff --git a/crates/ai/src/cli_agents/acp.rs b/crates/ai/src/cli_agents/acp.rs index 0b0a6f6b..ec88c2ea 100644 --- a/crates/ai/src/cli_agents/acp.rs +++ b/crates/ai/src/cli_agents/acp.rs @@ -8,7 +8,7 @@ //! process; [`AcpConversation`] runs it against the agent through `craft_anything_platform`. use super::{McpCommand, display_for}; -use crate::{AgentEvent, Conversation}; +use crate::{AgentEvent, Conversation, FailureKind, TurnFailure}; use craft_anything_platform::sync::Mutex; use craft_anything_platform::{Process, RunningProcess, StdinMode}; use serde_json::{Value, json}; @@ -120,6 +120,7 @@ pub struct AcpCore { tools: HashMap, /// Permission requests go to an approver ([`Out::Decide`]) instead of the person's chat. decides: bool, + failure: Option, } impl AcpCore { @@ -138,6 +139,7 @@ impl AcpCore { asks: HashMap::new(), tools: HashMap::new(), decides: false, + failure: None, } } @@ -171,6 +173,7 @@ impl AcpCore { /// A message from the person: sent now, or once the session is ready. pub fn prompt(&mut self, text: &str) -> Vec { + self.failure = None; self.queued.push_back(text.to_string()); // A start that failed earlier (an `initialize` error) left nothing under way: start again, // or the message would wait for a session that never comes (#271). @@ -226,6 +229,11 @@ impl AcpCore { let params = json!({"cwd": self.cwd, "mcpServers": self.mcp_servers()}); return vec![self.request("session/new", params, Pending::NewSession)]; } + self.failure = Some(TurnFailure::new(match err.get("code").and_then(Value::as_i64) { + Some(-32000) => FailureKind::Authentication, + Some(-32602..=-32600) => FailureKind::Configuration, + _ => FailureKind::Unknown, + })); let fix = matches!(kind, Pending::Initialize | Pending::NewSession) .then(|| format!("Check that {} is installed and signed in: run it once in a terminal.", self.agent_name)); return self.fail(format!("{}: {text}", self.agent_name), fix); @@ -447,6 +455,7 @@ pub struct AcpConversation { approver: Option, preamble: String, logs: Option, + failure: Option, } impl AcpConversation { @@ -488,6 +497,7 @@ impl AcpConversation { approver: None, preamble: String::new(), logs: None, + failure: None, } } @@ -558,6 +568,10 @@ impl AcpConversation { } impl Conversation for AcpConversation { + fn last_failure(&self) -> Option { + self.failure.or_else(|| self.lock().as_ref().and_then(|s| s.core.failure)) + } + fn send(&mut self, text: &str) { if self.busy.swap(true, Ordering::SeqCst) { let _ = self.tx.send(AgentEvent::Error { @@ -567,11 +581,13 @@ impl Conversation for AcpConversation { }); return; } + self.failure = None; let running = self.proc.as_mut().is_some_and(|p| !p.is_finished()); if !running { let _ = self.tx.send(AgentEvent::Status { text: format!("Starting {}…", display_for(&self.agent)) }); if let Err(e) = self.spawn() { self.busy.store(false, Ordering::SeqCst); + self.failure = Some(TurnFailure::from_process(&e)); let _ = self.tx.send(AgentEvent::Error { message: e.message, remediation: e.remediation, log: None }); let _ = self.tx.send(AgentEvent::TurnDone { backend: self.backend.clone(), @@ -596,6 +612,8 @@ impl Conversation for AcpConversation { // An agent that exits mid-message would otherwise leave the conversation busy forever. if self.busy.load(Ordering::SeqCst) && self.proc.as_mut().is_some_and(RunningProcess::is_finished) { self.busy.store(false, Ordering::SeqCst); + let ready = self.lock().as_ref().is_some_and(|s| s.core.ready); + self.failure = Some(TurnFailure::new(if ready { FailureKind::Transient } else { FailureKind::Unknown })); let tail = self.lock().as_ref().map(|s| s.stderr.iter().cloned().collect::>().join("\n")).unwrap_or_default(); events.extend(self.rx.try_iter()); let command = craft_anything_platform::process::printable_command(std::ffi::OsStr::new(&self.program), &self.args); @@ -818,6 +836,7 @@ mod tests { c.start(); c.prompt("hi"); let outs = c.line(r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"Authentication required"}}"#); + assert_eq!(c.failure, Some(TurnFailure::new(FailureKind::Authentication))); assert!(outs.contains(&Out::TurnEnded)); assert!(matches!(&events(outs)[0], AgentEvent::Error { message, remediation: Some(_), .. } if message.contains("Authentication required"))); let refused = sent(&c.line(r#"{"jsonrpc":"2.0","id":9,"method":"fs/read_text_file","params":{"path":"/etc/passwd"}}"#)); diff --git a/crates/ai/src/cli_agents/adapters.rs b/crates/ai/src/cli_agents/adapters.rs index 4b7033a2..01b9644d 100644 --- a/crates/ai/src/cli_agents/adapters.rs +++ b/crates/ai/src/cli_agents/adapters.rs @@ -287,7 +287,39 @@ impl AgentAdapter for ClaudeCode { p.session_id = Some(id.to_string()); } match s("type").as_str() { + "system" if s("subtype") == "init" => { + let failed = v.get("mcp_servers").and_then(Value::as_array).and_then(|servers| { + servers.iter().find(|server| { + server.get("name").and_then(Value::as_str) == Some(MCP_NAME) + && matches!(server.get("status").and_then(Value::as_str), Some("failed" | "needs-auth")) + }) + }); + if let Some(server) = failed { + p.failure = Some(TurnFailure::new(FailureKind::ToolsUnavailable)); + let status = server.get("status").and_then(Value::as_str).unwrap_or_default(); + out.push(AgentEvent::Error { + message: format!("Craft Anything's tools did not start in Claude Code (server status: {status})."), + remediation: Some( + "Check the Craft Anything CLI and the agent failure log, then start a new build attempt after fixing the tool connection.".into(), + ), + log: None, + }); + } + } "assistant" => { + if let Some(code) = v.get("error").and_then(Value::as_str) { + let failure = TurnFailure::claude(code); + p.failure = Some(failure); + let why = v.pointer("/message/content").map(content_text).unwrap_or_else(|| code.to_string()); + let (message, hint) = claude_code_failure(&why); + let remediation = hint.or_else(|| match failure.kind { + FailureKind::Authentication => self.sign_in_fix(), + FailureKind::Billing => Some("Check the account's credits, plan and spending limit before trying again.".into()), + FailureKind::Configuration => Some("Check the chosen model and Claude Code settings before trying again.".into()), + _ => None, + }); + return vec![AgentEvent::Error { message, remediation, log: None }]; + } for b in v.pointer("/message/content").and_then(Value::as_array).cloned().unwrap_or_default() { match b.get("type").and_then(Value::as_str).unwrap_or_default() { "text" => out.push(AgentEvent::TextDelta { text: b.get("text").and_then(Value::as_str).unwrap_or_default().to_string() }), @@ -332,6 +364,7 @@ impl AgentAdapter for ClaudeCode { log: None, }); } else if v.get("is_error").and_then(Value::as_bool).unwrap_or(false) { + p.failure.get_or_insert(TurnFailure::new(FailureKind::Unknown)); let why = if s("result").is_empty() { s("subtype") } else { s("result") }; let (message, remediation) = claude_code_failure(&why); out.push(AgentEvent::Error { message, remediation, log: None }); diff --git a/crates/ai/src/cli_agents/adapters/opencode.rs b/crates/ai/src/cli_agents/adapters/opencode.rs index f5d33ef0..68a9a4ef 100644 --- a/crates/ai/src/cli_agents/adapters/opencode.rs +++ b/crates/ai/src/cli_agents/adapters/opencode.rs @@ -21,13 +21,107 @@ impl AgentAdapter for OpenCode { a.extend(["--model".into(), m.clone()]); } a.extend(cx.options.args.iter().cloned()); + a.extend(["--format".into(), "json".into()]); a.push(format!("{}{}", cx.preamble, cx.prompt)); a } - fn streams_json(&self) -> bool { + fn required_flags(&self) -> (&'static [&'static str], &'static [&'static str]) { + (&["run", "--help"], &["--format"]) + } + + fn plain_fallback(&self) -> bool { false } + + fn parse(&self, p: &mut StreamParser, v: &Value) -> Vec { + let mut events = Vec::new(); + let part = v.get("part").unwrap_or(&Value::Null); + match v.get("type").and_then(Value::as_str) { + Some("text") => { + if let Some(text) = part.get("text").and_then(Value::as_str) { + events.push(AgentEvent::TextDelta { text: text.into() }); + } + } + Some("reasoning") => { + if let Some(text) = part.get("text").and_then(Value::as_str) { + events.push(AgentEvent::Status { text: summary_of(text) }); + } + } + Some("tool_use") => { + if let (Some(id), Some(name), Some(state)) = + (part.get("callID").and_then(Value::as_str), part.get("tool").and_then(Value::as_str), part.get("state")) + { + if !p.tool_names.contains_key(id) { + events.push(p.started(id, name, state.get("input").cloned().unwrap_or_else(|| json!({})))); + } + match state.get("status").and_then(Value::as_str) { + Some("completed") => events.push(p.finished(id, true, &content_text(state.get("output").unwrap_or(&Value::Null)))), + Some("error") => events.push(p.finished(id, false, &content_text(state.get("error").unwrap_or(&Value::Null)))), + _ => {} + } + } + } + Some("step_finish") => { + let tokens = part.get("tokens").unwrap_or(&Value::Null); + let count = |path| tokens.pointer(path).and_then(Value::as_u64).unwrap_or(0); + // OpenCode's step-finish counters exclude cache from input and reasoning from + // output, so these components make the inclusive totals the feed records. + p.input_tokens = p.input_tokens.saturating_add(count("/input")).saturating_add(count("/cache/read")).saturating_add(count("/cache/write")); + p.output_tokens = p.output_tokens.saturating_add(count("/output")).saturating_add(count("/reasoning")); + if !p.cost_incomplete { + p.cost_usd = part.get("cost").and_then(Value::as_f64).filter(|n| n.is_finite() && *n >= 0.0).and_then(|cost| { + let total = p.cost_usd.unwrap_or(0.0) + cost; + total.is_finite().then_some(total) + }); + p.cost_incomplete = p.cost_usd.is_none(); + } + } + Some("error") => { + let error = v.get("error").unwrap_or(&Value::Null); + let data = error.get("data").unwrap_or(&Value::Null); + let why = data.get("message").and_then(Value::as_str).unwrap_or("the service returned no error details"); + p.failure = Some(opencode_failure(error)); + p.saw_result = true; + events.push(AgentEvent::Error { message: format!("OpenCode stopped: {why}"), remediation: None, log: None }); + } + _ => {} + } + events + } +} + +/// OpenCode's JSON error carries the provider status and its own retry decision. In particular, +/// an unclassified exit or an unsupported CLI flag cannot become retryable from display text. +fn opencode_failure(error: &Value) -> TurnFailure { + let data = error.get("data").unwrap_or(&Value::Null); + match error.get("name").and_then(Value::as_str) { + Some("ProviderAuthError") => TurnFailure::new(FailureKind::Authentication), + Some("APIError") => { + let status = data.get("statusCode").and_then(Value::as_u64).and_then(|n| u16::try_from(n).ok()); + let mut failure = status.map_or_else( + || TurnFailure::new(FailureKind::Unknown), + |status| { + TurnFailure::from_api(&crate::http::status_error("OpenCode", status, data.get("responseBody").and_then(Value::as_str).unwrap_or_default())) + }, + ); + if failure.kind == FailureKind::Unknown && data.get("isRetryable").and_then(Value::as_bool) == Some(true) { + failure.kind = FailureKind::Transient; + } else if failure.retryable() && data.get("isRetryable").and_then(Value::as_bool) == Some(false) { + failure.kind = FailureKind::Unknown; + } + failure.retry_after_secs = data.get("responseHeaders").and_then(Value::as_object).and_then(|headers| { + headers + .iter() + .find(|(name, _)| name.eq_ignore_ascii_case("retry-after")) + .and_then(|(_, value)| value.as_str()) + .and_then(|n| n.trim().parse::().ok()) + .map(|n| n.clamp(1, crate::failure::MAX_RETRY_WAIT)) + }); + failure + } + _ => TurnFailure::new(FailureKind::Unknown), + } } #[cfg(test)] @@ -41,10 +135,87 @@ mod tests { #[test] fn the_prompt_goes_to_run_after_the_flags() { - assert_eq!(args(false, &BackendOptions::default()), ["run", "PRE make a game"]); + assert_eq!(args(false, &BackendOptions::default()), ["run", "--format", "json", "PRE make a game"]); let options = BackendOptions { cli_model: Some("anthropic/claude-sonnet-4-5".into()), args: vec!["--agent".into(), "build".into()], ..Default::default() }; - assert_eq!(args(true, &options), ["run", "--auto", "--model", "anthropic/claude-sonnet-4-5", "--agent", "build", "PRE make a game"]); + assert_eq!( + args(true, &options), + ["run", "--auto", "--model", "anthropic/claude-sonnet-4-5", "--agent", "build", "--format", "json", "PRE make a game"] + ); assert!(AgentAdapters::built_in().for_agent("opencode").ids().contains(&"opencode"), "registered, not the plain adapter"); } + + #[test] + fn structured_errors_select_retry_without_reading_english() { + let error = |status, retryable| json!({"name":"APIError","data":{"message":"服务限流。Please retry in 19.05s.","statusCode":status,"isRetryable":retryable,"responseHeaders":{"Retry-After":"4"}}}); + let rate = opencode_failure(&error(429, true)); + assert_eq!(rate, TurnFailure { kind: FailureKind::RateLimited, retry_after_secs: Some(4) }); + for (status, kind) in + [(401, FailureKind::Authentication), (402, FailureKind::Billing), (400, FailureKind::Configuration), (503, FailureKind::Transient)] + { + assert_eq!(opencode_failure(&error(status, true)).kind, kind); + } + assert!(!opencode_failure(&error(429, false)).retryable(), "an explicit nonretryable response is kept"); + assert!(!opencode_failure(&json!({"name":"UnknownError","data":{"message":"Please retry in 2s"}})).retryable()); + assert_eq!(opencode_failure(&json!({"name":"ProviderAuthError","data":{"message":"重新登录"}})).kind, FailureKind::Authentication); + assert_eq!(opencode_failure(&json!({"name":"APIError","data":{"isRetryable":true}})).kind, FailureKind::Transient); + let quota = json!({"name":"APIError","data":{"statusCode":429,"isRetryable":true,"responseBody":"{\"error\":{\"code\":\"insufficient_quota\"}}"}}); + assert_eq!(opencode_failure("a).kind, FailureKind::Billing); + } + + #[test] + fn json_output_keeps_text_tools_and_usage_without_ending_the_process_turn() { + let mut p = StreamParser::new(Arc::new(OpenCode)); + let lines = [ + json!({"type":"step_start","sessionID":"reported-but-not-resumable","part":{}}), + json!({"type":"text","part":{"text":"Built a page"}}), + json!({"type":"tool_use","part":{"callID":"write-1","tool":"write","state":{"status":"completed","input":{"path":"index.html"},"output":"wrote the page"}}}), + json!({"type":"step_finish","part":{"tokens":{"input":10,"output":5,"reasoning":2,"cache":{"read":3,"write":1}},"cost":0.01}}), + json!({"type":"step_finish","part":{"tokens":{"input":1,"output":2},"cost":0.02}}), + ]; + let events: Vec<_> = lines.iter().flat_map(|v| p.line(&v.to_string())).collect(); + assert!( + matches!(events.as_slice(), [AgentEvent::TextDelta { text }, AgentEvent::ToolStarted { tool, .. }, AgentEvent::ToolFinished { ok:true, .. }] if text=="Built a page" && tool=="write") + ); + assert_eq!((p.input_tokens, p.output_tokens), (15, 9)); + assert_eq!(p.cost_usd, Some(0.03)); + assert!(p.session_id.is_none(), "the adapter does not pretend it can resume"); + let events = p.line(&json!({"type":"error","error":{"name":"APIError","data":{"message":"暂时失败","statusCode":429,"isRetryable":true}}}).to_string()); + assert!(matches!(events.as_slice(), [AgentEvent::Error { .. }]), "run_cli alone owns TurnDone"); + assert_eq!(p.failure.unwrap().kind, FailureKind::RateLimited); + } + + #[test] + fn missing_invalid_and_overflowed_costs_stay_unknown_for_the_turn() { + let step = |cost| json!({"type":"step_finish","part":{"cost":cost,"tokens":{}}}).to_string(); + for invalid in [Value::Null, json!(-1), json!("unknown")] { + for unknown_first in [true, false] { + let mut p = StreamParser::new(Arc::new(OpenCode)); + let costs = if unknown_first { [invalid.clone(), json!(0.02)] } else { [json!(0.02), invalid.clone()] }; + for cost in costs { + p.line(&step(cost)); + } + p.line(&step(json!(0.01))); + assert_eq!(p.cost_usd, None, "a later known step cannot fill an earlier gap"); + } + } + for unknown_first in [true, false] { + let mut missing = StreamParser::new(Arc::new(OpenCode)); + if !unknown_first { + missing.line(&step(json!(0.02))); + } + missing.line(r#"{"type":"step_finish","part":{"tokens":{}}}"#); + missing.line(&step(json!(0.01))); + assert_eq!(missing.cost_usd, None); + } + let mut overflow = StreamParser::new(Arc::new(OpenCode)); + overflow.line(&step(json!(f64::MAX))); + overflow.line(&step(json!(f64::MAX))); + overflow.line(&step(json!(0))); + assert_eq!(overflow.cost_usd, None); + let mut zero = StreamParser::new(Arc::new(OpenCode)); + zero.line(&step(json!(0))); + zero.line(&step(json!(0))); + assert_eq!(zero.cost_usd, Some(0.0), "an explicit zero is a known cost"); + } } diff --git a/crates/ai/src/cli_agents/mod.rs b/crates/ai/src/cli_agents/mod.rs index 1798b3b0..8fa4c6ce 100644 --- a/crates/ai/src/cli_agents/mod.rs +++ b/crates/ai/src/cli_agents/mod.rs @@ -16,7 +16,7 @@ pub use adapters::{AgentAdapter, AgentAdapters, ArgsCx}; use crate::config::BackendOptions; use crate::model::Stop; -use crate::{AgentEvent, BackendConfig, Conversation}; +use crate::{AgentEvent, BackendConfig, Conversation, FailureKind, TurnFailure}; use craft_anything_engine::CancelToken; use craft_anything_evidence::{ErrorCode, ToolError}; use craft_anything_platform::process::Process; @@ -258,10 +258,14 @@ pub struct StreamParser { pub input_tokens: u64, pub output_tokens: u64, pub cost_usd: Option, + /// A missing or invalid per-step cost cannot become a complete turn total later. + cost_incomplete: bool, pub saw_result: bool, /// The run was asked to continue a saved session the agent doesn't have (it lives on another /// computer, or under another folder), so it did nothing. pub session_missing: bool, + /// A protocol fact about the failed turn, independent of the text shown to the person. + pub failure: Option, } fn short_tool(name: &str) -> String { @@ -303,8 +307,10 @@ impl StreamParser { input_tokens: 0, output_tokens: 0, cost_usd: None, + cost_incomplete: false, saw_result: false, session_missing: false, + failure: None, } } @@ -452,6 +458,7 @@ pub struct CliConversation { verified: Arc>>, /// The last run's saved session wasn't there (see [`Conversation::session_lost`]). lost: Arc, + failure: Arc>>, } impl CliConversation { @@ -460,7 +467,17 @@ impl CliConversation { pub fn new(spec: CliSpec) -> Self { let (tx, rx) = std::sync::mpsc::channel(); let session = Arc::new(Mutex::new(spec.resume.clone())); - CliConversation { spec, tx, rx, busy: Arc::default(), stop: Stop::default(), session, verified: Arc::default(), lost: Arc::default() } + CliConversation { + spec, + tx, + rx, + busy: Arc::default(), + stop: Stop::default(), + session, + verified: Arc::default(), + lost: Arc::default(), + failure: Arc::default(), + } } /// The agent's session id once a run has reported it (to continue later). @@ -504,6 +521,7 @@ struct RunState { session: Arc>>, verified: Arc>>, lost: Arc, + failure: Arc>>, } /// End a turn: take the next message first, then say the turn is done (`send(done)`), since @@ -525,7 +543,7 @@ fn nothing_done(backend: &str) -> AgentEvent { /// next message; one it was asked to continue but doesn't have is forgotten and flagged in /// `state.lost` before the error that says so is sent. fn run_cli(spec: &CliSpec, text: &str, tx: &Sender, stop: &Stop, state: &RunState) -> AgentEvent { - let RunState { session, verified, lost } = state; + let RunState { session, verified, lost, failure } = state; let name = spec.config.name.clone(); let agent = spec.config.model.clone(); let program = spec.options.command.clone().unwrap_or_else(|| program_for(&agent)); @@ -592,7 +610,9 @@ fn run_cli(spec: &CliSpec, text: &str, tx: &Sender, stop: &Stop, sta let _ = tx.send(AgentEvent::Status { text: format!("Asking {}{}…", display_for(&agent), remote.as_ref().map(|h| format!(" on {h}")).unwrap_or_default()) }); let parser = Arc::new(Mutex::new(StreamParser::new(adapter.clone()))); - let (p2, tx2, tx_err, lost2) = (parser.clone(), tx.clone(), tx.clone(), lost.clone()); + let (p2, tx2, tx_err, lost2, failure2) = (parser.clone(), tx.clone(), tx.clone(), lost.clone(), failure.clone()); + let failed_tools = Arc::new(AtomicBool::new(false)); + let (failed_tools2, failed_tools3) = (failed_tools.clone(), failed_tools.clone()); let stop2 = stop.clone(); let mut proc = (if remote.is_none() { adapter.process(&prog) } else { Process::new(&prog) }) .args(&argv) @@ -603,14 +623,23 @@ fn run_cli(spec: &CliSpec, text: &str, tx: &Sender, stop: &Stop, sta } let mut proc = proc .stdout_cap(4 * 1024 * 1024) - .cancel_when(move || stop2.is_stopped()) + .cancel_when(move || stop2.is_stopped() || failed_tools2.load(Ordering::SeqCst)) .on_stdout_line(move |l| { + if failed_tools3.load(Ordering::SeqCst) { + return; + } let mut p = p2.lock().unwrap_or_else(PoisonError::into_inner); let events = p.line(l); // Said before the error that follows it, so whoever reads the error can tell. if p.session_missing { lost2.store(true, Ordering::SeqCst); } + if let Some(failed) = p.failure { + *failure2.lock().unwrap_or_else(PoisonError::into_inner) = Some(failed); + if failed.kind == FailureKind::ToolsUnavailable { + failed_tools3.store(true, Ordering::SeqCst); + } + } for e in events { let _ = tx2.send(e); } @@ -635,7 +664,26 @@ fn run_cli(spec: &CliSpec, text: &str, tx: &Sender, stop: &Stop, sta } else if let Some(id) = &p.session_id { *session.lock().unwrap_or_else(PoisonError::into_inner) = Some(id.clone()); } + if p.failure.is_none() { + let failed = match &result { + Ok(out) if !out.success => Some(TurnFailure::process_exit(out.code)), + Err(e) if e.code != ErrorCode::Cancelled => Some(TurnFailure::from_process(e)), + _ => None, + }; + *failure.lock().unwrap_or_else(PoisonError::into_inner) = failed; + } match result { + // The protocol already named the required tool server. Stop the owned process without + // replacing that useful error with the cancellation used to stop it. + result if failed_tools.load(Ordering::SeqCst) => { + let (out, error) = match &result { + Ok(out) => (Some(out), None), + Err(e) => (None, Some(e.message.as_str())), + }; + if let Some(path) = shown.save_with(&agent, out, Some("Craft Anything's tools did not start."), error).0 { + let _ = tx.send(AgentEvent::Status { text: format!("The details are in {}", path.display()) }); + } + } // It said why already, and there's nothing more in its output to keep. Ok(_) if p.session_missing => {} Ok(out) if !out.success && !p.saw_result => { @@ -775,6 +823,15 @@ impl FailureContext { } impl Conversation for CliConversation { + fn retry(&mut self, text: &str) { + let resumable = self.spec.adapters.for_agent(&self.spec.config.model).resumes() && self.session_id().is_some(); + self.send(if resumable { crate::failure::CONTINUE_TURN } else { text }); + } + + fn last_failure(&self) -> Option { + *self.failure.lock().unwrap_or_else(PoisonError::into_inner) + } + fn session_id(&self) -> Option { CliConversation::session_id(self) } @@ -799,8 +856,9 @@ impl Conversation for CliConversation { } self.stop.reset(); self.lost.store(false, Ordering::SeqCst); + *self.failure.lock().unwrap_or_else(PoisonError::into_inner) = None; let (spec, tx, stop, busy) = (self.spec.clone(), self.tx.clone(), self.stop.clone(), self.busy.clone()); - let state = RunState { session: self.session.clone(), verified: self.verified.clone(), lost: self.lost.clone() }; + let state = RunState { session: self.session.clone(), verified: self.verified.clone(), lost: self.lost.clone(), failure: self.failure.clone() }; let text = text.to_string(); let spawned = std::thread::Builder::new().name("ai-cli".into()).spawn(move || { let done = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| run_cli(&spec, &text, &tx, &stop, &state))).unwrap_or_else(|_| { @@ -846,6 +904,47 @@ impl Drop for CliConversation { mod tests { use super::*; + #[test] + fn failed_craft_tool_server_is_reported_from_its_init_status() { + let mut p = StreamParser::new(adapter_for("claude-code")); + let events = p.line(r#"{"type":"system","subtype":"init","session_id":"cannot-build","mcp_servers":[{"name":"craft-anything","status":"failed"}]}"#); + assert!( + events.iter().any(|e| matches!(e, AgentEvent::Error { message, .. } if message.contains("tools") && message.contains("start"))), + "a failed required tool server must end the turn with an actionable error: {events:?}" + ); + assert_eq!(p.failure, Some(TurnFailure::new(FailureKind::ToolsUnavailable))); + } + + #[test] + fn optional_or_pending_mcp_servers_do_not_block_the_builder() { + for status in ["connected", "pending"] { + let mut p = StreamParser::new(adapter_for("claude-code")); + let init = json!({"type":"system","subtype":"init","mcp_servers":[{"name":"craft-anything","status":status},{"name":"optional-service","status":"failed"}]}); + assert!(p.line(&init.to_string()).is_empty()); + assert!(p.failure.is_none()); + } + let mut p = StreamParser::new(adapter_for("claude-code")); + assert!(!p.line(r#"{"type":"system","subtype":"init","mcp_servers":[{"name":"craft-anything","status":"needs-auth"}]}"#).is_empty()); + assert_eq!(p.failure, Some(TurnFailure::new(FailureKind::ToolsUnavailable))); + } + + #[test] + fn claude_assistant_error_codes_survive_localized_display_messages() { + for (code, kind) in [ + ("authentication_failed", FailureKind::Authentication), + ("billing_error", FailureKind::Billing), + ("rate_limit", FailureKind::RateLimited), + ("server_error", FailureKind::Transient), + ] { + let mut p = StreamParser::new(adapter_for("claude-code")); + let events = p.line(&json!({"type":"assistant","error":code,"message":{"content":[{"type":"text","text":"服务暂时不可用"}]}}).to_string()); + assert!(matches!(events.as_slice(), [AgentEvent::Error { .. }])); + assert_eq!(p.failure, Some(TurnFailure::new(kind))); + p.line(r#"{"type":"result","is_error":true,"result":"the turn ended"}"#); + assert_eq!(p.failure, Some(TurnFailure::new(kind)), "result must keep the more specific code"); + } + } + /// #506: under a Flatpak `~/.claude` is read-only, so Claude Code gets a folder of ours to save /// its conversations in (so `--resume` finds them); the person's own `CLAUDE_CONFIG_DIR` and a /// normal install are left as they are, and the sign-in hint says how to sign in there. diff --git a/crates/ai/src/failure.rs b/crates/ai/src/failure.rs new file mode 100644 index 00000000..bb94e99f --- /dev/null +++ b/crates/ai/src/failure.rs @@ -0,0 +1,175 @@ +//! A failed turn's retry policy, kept apart from the words shown to the person. Protocol and +//! process boundaries supply these facts; display text never decides whether to retry. + +use craft_anything_evidence::{ErrorCode, ToolError}; + +/// Why a conversation's last turn failed. Unknown failures need attention: repeating an +/// unclassified setup error can spend quota without doing any work. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FailureKind { + Authentication, + Billing, + MissingProgram, + ToolsUnavailable, + Configuration, + Transient, + RateLimited, + Unknown, +} + +/// A typed failure, and a provider's optional wait in seconds. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct TurnFailure { + pub kind: FailureKind, + pub retry_after_secs: Option, +} + +/// No failed turn waits more than fifteen minutes before its next attempt. +pub(crate) const MAX_RETRY_WAIT: u64 = 15 * 60; +pub(crate) const CONTINUE_TURN: &str = + "Your last turn was cut off. Carry on from where you stopped: look at what's already in app/ and your checklist, then finish the task."; + +impl TurnFailure { + pub const fn new(kind: FailureKind) -> Self { + Self { kind, retry_after_secs: None } + } + + pub fn retryable(self) -> bool { + matches!(self.kind, FailureKind::Transient | FailureKind::RateLimited) + } + + pub(crate) fn from_api(error: &ToolError) -> Self { + let status = error.details.as_ref().and_then(|d| d.get("status")).and_then(serde_json::Value::as_u64); + let provider = error.details.as_ref().and_then(|d| d.get("provider_error")).and_then(serde_json::Value::as_str); + let kind = match (status, provider) { + (Some(401 | 403), _) | (_, Some("authentication_error" | "permission_error" | "invalid_api_key")) => FailureKind::Authentication, + (Some(402), _) | (_, Some("billing_error" | "insufficient_quota")) => FailureKind::Billing, + (Some(429), _) | (_, Some("rate_limit_error" | "rate_limit_exceeded")) => FailureKind::RateLimited, + (Some(408 | 500..=599), _) | (_, Some("overloaded_error" | "api_error")) => FailureKind::Transient, + (Some(400..=499), _) => FailureKind::Configuration, + (_, Some(_)) => FailureKind::Unknown, + _ => match error.code { + ErrorCode::TimedOut | ErrorCode::EngineMissing | ErrorCode::Truncated => FailureKind::Transient, + ErrorCode::NoPermission => FailureKind::Authentication, + ErrorCode::BadInput | ErrorCode::BadSettings | ErrorCode::NotSupported | ErrorCode::FeatureMissing => FailureKind::Configuration, + _ => FailureKind::Unknown, + }, + }; + let retry_after_secs = + error.details.as_ref().and_then(|d| d.get("retry_after_secs")).and_then(serde_json::Value::as_u64).map(|n| n.clamp(1, MAX_RETRY_WAIT)); + Self { kind, retry_after_secs } + } + + pub(crate) fn from_process(error: &ToolError) -> Self { + Self::new(match error.code { + ErrorCode::EngineMissing | ErrorCode::NotFound => FailureKind::MissingProgram, + ErrorCode::TimedOut => FailureKind::Transient, + ErrorCode::NoPermission | ErrorCode::BadSettings | ErrorCode::BadInput => FailureKind::Configuration, + _ => FailureKind::Unknown, + }) + } + + /// Claude Code's documented `assistant.error` values, not the assistant's display text. + pub(crate) fn claude(code: &str) -> Self { + Self::new(match code { + "authentication_failed" | "oauth_org_not_allowed" | "cloud_credential_error" => FailureKind::Authentication, + "billing_error" | "account_on_hold" => FailureKind::Billing, + "invalid_request" | "model_not_found" => FailureKind::Configuration, + "rate_limit" => FailureKind::RateLimited, + "server_error" | "overloaded" => FailureKind::Transient, + _ => FailureKind::Unknown, + }) + } + + /// Without a protocol error, a process may have crashed or lost its connection. Signals + /// and nonzero exits keep the same bounded retry policy; a successful exit never opts in. + pub(crate) fn process_exit(code: Option) -> Self { + Self::new(if code == Some(0) { FailureKind::Unknown } else { FailureKind::Transient }) + } +} + +/// Compatibility for a provider's display-only wait hint, after a typed status already made +/// the failure retryable. This never decides whether to retry. The number is bounded even when +/// an error asks for hours, or sends infinity or a negative delay. +pub(crate) fn retry_after(message: &str) -> Option { + let m = message.to_ascii_lowercase(); + let tail = ["retry in ", "retry after ", "try again in "].iter().find_map(|s| m.find(s).and_then(|at| m.get(at + s.len()..)))?; + let number: String = tail.chars().take_while(|c| c.is_ascii_digit() || *c == '.').collect(); + let n: f64 = number.parse().ok()?; + if !n.is_finite() || n < 0.0 { + return None; + } + let unit = tail.get(number.len()..)?.trim_start(); + let scale = if unit.starts_with("ms") || unit.starts_with("millisecond") { + 0.001 + } else if unit.starts_with("min") || unit.starts_with('m') { + 60.0 + } else if unit.starts_with('h') { + 3600.0 + } else { + 1.0 + }; + Some((n * scale).ceil().clamp(1.0, MAX_RETRY_WAIT as f64) as u64) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn typed_failures_do_not_depend_on_their_display_language() { + for message in ["请更新凭证", "The provider is unavailable", "anything"] { + let auth = ToolError::new(ErrorCode::NoPermission, message).with_details(json!({"status":401})); + assert_eq!(TurnFailure::from_api(&auth).kind, FailureKind::Authentication); + assert!(!TurnFailure::from_api(&auth).retryable()); + let bill = ToolError::new(ErrorCode::BadInput, message).with_details(json!({"status":402})); + assert_eq!(TurnFailure::from_api(&bill).kind, FailureKind::Billing); + let absent = ToolError::new(ErrorCode::EngineMissing, message); + assert_eq!(TurnFailure::from_process(&absent).kind, FailureKind::MissingProgram); + assert!(!TurnFailure::from_process(&absent).retryable()); + } + let transient = ToolError::new(ErrorCode::TimedOut, "The authentication service is restarting; billing and login are temporarily unavailable."); + assert!(TurnFailure::from_api(&transient).retryable(), "English words don't override a typed timeout"); + let limit = ToolError::new(ErrorCode::OverLimit, "provider response").with_details(json!({"status":429,"retry_after_secs":2})); + assert_eq!(TurnFailure::from_api(&limit), TurnFailure { kind: FailureKind::RateLimited, retry_after_secs: Some(2) }); + let quota = crate::http::status_error("Provider", 429, r#"{"error":{"code":"insufficient_quota","message":"余额不足"}}"#); + assert_eq!(TurnFailure::from_api("a).kind, FailureKind::Billing); + assert!(!TurnFailure::from_api(&ToolError::new(ErrorCode::OverLimit, "oversized input")).retryable(), "a local size limit is not rate limiting"); + } + + #[test] + fn streamed_provider_errors_keep_their_type() { + for (provider, kind) in [ + ("rate_limit_exceeded", FailureKind::RateLimited), + ("invalid_api_key", FailureKind::Authentication), + ("insufficient_quota", FailureKind::Billing), + ("unknown", FailureKind::Unknown), + ] { + let mut stream = crate::openai::ChatStream::default(); + let event = crate::http::SseEvent { event: None, data: json!({"error":{"code":provider,"message":"服务返回错误"}}).to_string() }; + let error = stream.apply(&event, &mut |_| {}).unwrap_err(); + assert_eq!(TurnFailure::from_api(&error).kind, kind); + } + } + + #[test] + fn structured_claude_errors_and_plain_waits_have_bounded_retry_advice() { + assert!(!TurnFailure::claude("billing_error").retryable()); + assert!(!TurnFailure::claude("authentication_failed").retryable()); + assert!(TurnFailure::claude("server_error").retryable()); + assert!(TurnFailure::claude("rate_limit").retryable()); + assert_eq!(retry_after("Please retry in 19.052906498s."), Some(20)); + assert_eq!(retry_after("Rate limit reached. Retry after 2 minutes"), Some(120)); + assert_eq!(retry_after("try again in 500ms"), Some(1)); + assert_eq!(retry_after("Retry in 45 seconds"), Some(45)); + for code in [None, Some(1), Some(2), Some(255)] { + assert!(TurnFailure::process_exit(code).retryable()); + } + assert!(!TurnFailure::process_exit(Some(0)).retryable()); + assert_eq!(retry_after("retry after 3 h"), Some(MAX_RETRY_WAIT)); + for text in ["retry after NaN", "retry after inf", "retry after -1s", "invalid command", "please retry in a moment", "retry in ."] { + assert_eq!(retry_after(text), None); + } + } +} diff --git a/crates/ai/src/http.rs b/crates/ai/src/http.rs index b5d57db0..87caafc9 100644 --- a/crates/ai/src/http.rs +++ b/crates/ai/src/http.rs @@ -55,7 +55,12 @@ pub fn status_error(what: &str, status: u16, body: &str) -> ToolError { 500..=599 => (ErrorCode::EngineMissing, "The service is having trouble. Try again soon, or use another backend."), _ => (ErrorCode::BadInput, "The request was rejected; see the message."), }; - ToolError::new(code, format!("{what} answered {status}: {detail}")).with_remediation(remediation).with_details(serde_json::json!({"status": status})) + let provider = serde_json::from_str::(body) + .ok() + .and_then(|v| v.pointer("/error/code").and_then(Value::as_str).or_else(|| v.pointer("/error/type").and_then(Value::as_str)).map(str::to_string)); + ToolError::new(code, format!("{what} answered {status}: {detail}")) + .with_remediation(remediation) + .with_details(serde_json::json!({"status": status, "provider_error": provider})) } /// Read only the human-readable fields, never serialise an arbitrary provider payload. @@ -112,8 +117,13 @@ pub fn post_json(agent: &ureq::Agent, what: &str, url: &str, headers: &[(&str, S let resp = req.send(body.to_string()).map_err(|e| net_error(what, e))?; let status = resp.status().as_u16(); if !(200..300).contains(&status) { + let retry_after = resp.headers().get("retry-after").and_then(|h| h.to_str().ok()).and_then(|s| s.trim().parse::().ok()); let text = resp.into_body().read_to_string().unwrap_or_default(); - return Err(status_error(what, status, &text)); + let mut error = status_error(what, status, &text); + if let (Some(seconds), Some(Value::Object(details))) = (retry_after, error.details.as_mut()) { + details.insert("retry_after_secs".into(), Value::from(seconds.min(crate::failure::MAX_RETRY_WAIT))); + } + return Err(error); } Ok(Box::new(resp.into_body().into_reader())) } @@ -337,4 +347,39 @@ mod tests { assert!(e.message.contains("invalid x-api-key")); assert_eq!(status_error("x", 429, "slow down").code, ErrorCode::OverLimit); } + + #[test] + fn a_post_keeps_a_structured_retry_after_header() { + use std::io::Write; + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + stream.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + let mut reader = BufReader::new(&mut stream); + let mut len = 0; + loop { + let mut line = String::new(); + assert!(reader.read_line(&mut line).unwrap() > 0); + if line == "\r\n" { + break; + } + if let Some((name, value)) = line.split_once(':') + && name.eq_ignore_ascii_case("content-length") + { + len = value.trim().parse::().unwrap(); + } + } + reader.read_exact(&mut vec![0; len]).unwrap(); + let body = r#"{"error":{"type":"rate_limit_error","message":"服务限流"}}"#; + write!(stream, "HTTP/1.1 429 Too Many Requests\r\nRetry-After: 7\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).unwrap(); + }); + let result = post_json(&quick_agent(Duration::from_secs(5)), "Stand-in", &url, &[], &serde_json::json!({})); + let error = match result { + Err(error) => error, + Ok(_) => panic!("a rate limit is not a response stream"), + }; + assert_eq!(crate::TurnFailure::from_api(&error), crate::TurnFailure { kind: crate::FailureKind::RateLimited, retry_after_secs: Some(7) }); + server.join().unwrap(); + } } diff --git a/crates/ai/src/lib.rs b/crates/ai/src/lib.rs index d53dffb5..d58c496d 100644 --- a/crates/ai/src/lib.rs +++ b/crates/ai/src/lib.rs @@ -12,6 +12,7 @@ pub mod cli_agents; pub mod config; pub mod context; pub mod detect; +mod failure; pub mod failure_log; pub mod http; pub mod key_tools; @@ -24,6 +25,7 @@ pub mod rebuild; pub mod ssh; pub use config::{AiConfig, BackendEntry, BackendOptions, Price}; +pub use failure::{FailureKind, TurnFailure}; pub use keys::KeyStore; pub use model::Effort; @@ -213,6 +215,17 @@ pub enum AgentEvent { /// A chat with the agent about the current session. pub trait Conversation: Send { fn send(&mut self, text: &str); + /// Retry an interrupted turn. Stateless adapters resend `text`; a conversation that keeps + /// its own history can continue it without appending the user's message again. + fn retry(&mut self, text: &str) { + let prompt = if self.session_id().is_some() { failure::CONTINUE_TURN } else { text }; + self.send(prompt); + } + /// Why the last turn failed, supplied by its protocol or process boundary. An unknown + /// failure is never retried automatically. Cleared when the next turn starts. + fn last_failure(&self) -> Option { + None + } /// A saved session to continue later, for agents that have one (agent CLIs). fn session_id(&self) -> Option { None diff --git a/crates/ai/src/openai.rs b/crates/ai/src/openai.rs index 50bdd24f..3313e78a 100644 --- a/crates/ai/src/openai.rs +++ b/crates/ai/src/openai.rs @@ -235,7 +235,11 @@ impl ChatStream { serde_json::from_str(data).map_err(|e| ToolError::new(ErrorCode::GarbledOutput, format!("The server sent a chunk we couldn't read: {e}")))?; if let Some(err) = v.get("error") { let msg = err.get("message").and_then(Value::as_str).unwrap_or("unknown error"); - return Err(ToolError::new(ErrorCode::EngineMissing, format!("Stream error: {msg}"))); + let provider = err.get("code").and_then(Value::as_str).or_else(|| err.get("type").and_then(Value::as_str)); + let status = err.get("code").and_then(Value::as_u64); + return Err( + ToolError::new(ErrorCode::ProgramFailed, format!("Stream error: {msg}")).with_details(json!({"provider_error": provider, "status": status})) + ); } if let Some(m) = v.get("model").and_then(Value::as_str) { self.model = m.to_string(); diff --git a/crates/ai/src/rebuild/mod.rs b/crates/ai/src/rebuild/mod.rs index b1b6ef91..45cc1947 100644 --- a/crates/ai/src/rebuild/mod.rs +++ b/crates/ai/src/rebuild/mod.rs @@ -848,11 +848,29 @@ impl Worker { /// `fresh`: what to send instead when the agent's saved session turns out not to be on this /// computer. - fn turn(&mut self, ws: &mut Workspace, agent: &mut dyn Conversation, prompt: &str, mut fresh: Option) -> Result { + fn turn(&mut self, ws: &mut Workspace, agent: &mut dyn Conversation, prompt: &str, fresh: Option) -> Result { + self.turn_with_wait(ws, agent, prompt, fresh, |stop, seconds| { + for _ in 0..seconds { + stop.check()?; + std::thread::sleep(Duration::from_secs(1)); + } + stop.check() + }) + } + + fn turn_with_wait( + &mut self, + ws: &mut Workspace, + agent: &mut dyn Conversation, + prompt: &str, + mut fresh: Option, + mut wait_for: impl FnMut(&CancelToken, u64) -> Result<(), ToolError>, + ) -> Result { let mut prompt = prompt.to_string(); let mut tries = 0u64; + let mut retry = false; loop { - let result = self.converse(ws, agent, &prompt); + let result = self.converse_turn(ws, agent, &prompt, retry); let lost = result.is_err() && agent.session_lost(); if lost { // The id is for a session this computer doesn't have: never ask for it again. @@ -871,30 +889,31 @@ impl Worker { "The builder's earlier conversation isn't on this computer. Starting a new one from what the project already has…", )?; prompt = f; + retry = false; } - Err(e) if e.code != ErrorCode::Cancelled && tries < 3 && agent.session_id().is_some() && worth_retrying(&e.message) => { + // A missing session can only start fresh above, never resume. Only backend + // Error events become ProgramFailed here; a simultaneous backend failure must + // not make an inbox or workspace write failure retryable. + Err(e) if !lost && e.code == ErrorCode::ProgramFailed && tries < 3 && agent.last_failure().is_some_and(crate::TurnFailure::retryable) => { tries += 1; let why = e.message.chars().take(160).collect::(); - // A service that says how long to wait is waited for; otherwise a dropped - // connection or a busy service gets a moment, longer each time. - let wait = match retry_after(&e.message) { + // Typed status decides whether to retry. A display-only wait remains a + // compatibility hint, and never overrides a structured provider delay. + let delay = agent.last_failure().and_then(|f| f.retry_after_secs).or_else(|| crate::failure::retry_after(&e.message)); + let seconds = match delay { Some(secs) => { let note = format!("The AI service is busy or its rate limit was reached, and asked to wait {secs} s. Trying again then…"); self.step(ws, Step::Build, Status::Building, ¬e)?; - secs + 1 + secs.saturating_add(1).min(crate::failure::MAX_RETRY_WAIT) } None => { self.step(ws, Step::Build, Status::Building, &format!("The builder stopped ({why}). Picking up where it left off…"))?; 30 * tries } }; - for _ in 0..wait { - self.stop.check()?; - std::thread::sleep(Duration::from_secs(1)); - } - prompt = format!( - "Your last turn was cut off ({why}). Carry on from where you stopped: look at what's already in app/ and your checklist, then finish the task." - ); + wait_for(&self.stop, seconds)?; + self.stop.check()?; + retry = true; } r => return r, } @@ -904,7 +923,15 @@ impl Worker { /// Send one message and pump the agent until its turn ends, relaying events and inbox /// requests. Returns the agent's reply text. fn converse(&mut self, ws: &mut Workspace, agent: &mut dyn Conversation, text: &str) -> Result { - agent.send(text); + self.converse_turn(ws, agent, text, false) + } + + fn converse_turn(&mut self, ws: &mut Workspace, agent: &mut dyn Conversation, text: &str, retry: bool) -> Result { + if retry { + agent.retry(text); + } else { + agent.send(text); + } let mut reply = String::new(); let mut failed = None; loop { @@ -1028,36 +1055,3 @@ pub fn builder_backend(cfg: &crate::AiConfig, named: Option<&str>) -> Result bool { - let m = message.to_ascii_lowercase(); - !["sign in", "log in", "login", "logged in", "auth", "credit", "billing", "usage limit", "not installed", "install ", crate::cli_agents::TOO_OLD_FOR_MODEL] - .iter() - .any(|k| m.contains(k)) -} - -/// The longest wait a service's "retry in …" is followed for. -const MAX_RETRY_WAIT: u64 = 15 * 60; - -/// How long a busy or rate-limited service asked to wait ("Please retry in 19.05s", "retry after -/// 2 minutes"), in whole seconds rounded up and at most [`MAX_RETRY_WAIT`] (#142). -fn retry_after(message: &str) -> Option { - let m = message.to_ascii_lowercase(); - let at = ["retry in ", "retry after ", "try again in "].iter().find_map(|p| m.find(p).map(|i| i + p.len()))?; - let rest = m.get(at..)?; - let number: String = rest.chars().take_while(|c| c.is_ascii_digit() || *c == '.').collect(); - let value: f64 = number.parse().ok()?; - let unit = rest.get(number.len()..).unwrap_or_default().trim_start(); - let seconds = if unit.starts_with("ms") || unit.starts_with("millisecond") { - value / 1000.0 - } else if unit.starts_with("min") || (unit.starts_with('m') && !unit.starts_with("ms")) { - value * 60.0 - } else if unit.starts_with('h') { - value * 3600.0 - } else { - value - }; - Some((seconds.ceil().max(1.0) as u64).min(MAX_RETRY_WAIT)) -} diff --git a/crates/ai/src/rebuild/tests.rs b/crates/ai/src/rebuild/tests.rs index 73aefe31..f22b23bf 100644 --- a/crates/ai/src/rebuild/tests.rs +++ b/crates/ai/src/rebuild/tests.rs @@ -323,35 +323,432 @@ fn pictures_of_the_original_reach_the_builder() { assert!(pictures_reminder(&ws).contains("1 picture(s)")); } -#[test] -fn a_cut_off_turn_is_resumed_unless_the_person_must_act() { - for resumable in ["Claude Code took longer than 86400 s and was stopped.", "API Error: 529 overloaded", "connection reset by peer"] { - assert!(worth_retrying(resumable), "{resumable}"); - } - for needs_you in [ - "Not logged in. Please run /login", - "Your credit balance is too low", - "Claude Code is not installed", - "Invalid API key · Please run /login", - "Claude Code 2.1.81 does not support this model", - ] { - assert!(!worth_retrying(needs_you), "{needs_you}"); - } - assert_eq!(RunOptions::default().max_rounds, 6, "room to keep fixing until it works"); -} +mod typed_retries { + use super::*; + use crate::{FailureKind, TurnFailure}; + use std::collections::VecDeque; -/// #142: a rate-limited service's own wait is followed (rounded up, capped), not a fixed one. -#[test] -fn a_services_requested_wait_is_read() { - assert_eq!(retry_after("OpenCode stopped with exit code 1: Please retry in 19.052906498s."), Some(20)); - assert_eq!(retry_after("Rate limit reached. Retry after 2 minutes"), Some(120)); - assert_eq!(retry_after("try again in 500ms"), Some(1)); - assert_eq!(retry_after("Please try again in 3h"), Some(MAX_RETRY_WAIT)); - assert_eq!(retry_after("Retry in 45 seconds"), Some(45)); - for no_wait in ["API Error: 529 overloaded", "please retry in a moment", "retry in ."] { - assert_eq!(retry_after(no_wait), None, "{no_wait}"); + struct FailingConversation { + failures: VecDeque>, + failure: Option, + queued: Vec, + sent: Vec, + session: Option, + message: &'static str, + } + + impl FailingConversation { + fn new(failures: impl IntoIterator>, session: bool) -> Self { + Self { + failures: failures.into_iter().collect(), + failure: None, + queued: Vec::new(), + sent: Vec::new(), + session: session.then(|| "saved".into()), + message: "provider response", + } + } + } + + impl Conversation for FailingConversation { + fn send(&mut self, text: &str) { + self.sent.push(text.into()); + self.failure = self.failures.pop_front().flatten(); + self.queued.push(if self.failure.is_some() { + AgentEvent::Error { message: self.message.into(), remediation: None, log: None } + } else { + AgentEvent::TextDelta { text: "Done".into() } + }); + self.queued.push(AgentEvent::TurnDone { backend: "scripted".into(), input_tokens: 0, output_tokens: 0, cost_usd: None, cost_is_estimate: false }); + } + fn session_id(&self) -> Option { + self.session.clone() + } + fn last_failure(&self) -> Option { + self.failure + } + fn poll(&mut self) -> Vec { + std::mem::take(&mut self.queued) + } + fn answer_consent(&mut self, _: &str, _: bool) {} + fn stop(&mut self) {} + fn is_busy(&self) -> bool { + false + } + } + + #[test] + fn transient_failures_retry_without_requiring_an_external_session() { + for saved in [false, true] { + let dir = tempfile::tempdir().unwrap(); + let mut ws = Workspace::create(dir.path(), "x", "Build a thing", TargetRef::None).unwrap(); + let mut worker = check_worker(&ws, false); + let failure = TurnFailure { kind: FailureKind::RateLimited, retry_after_secs: Some(2) }; + let mut agent = FailingConversation::new([Some(failure), None], saved); + agent.message = "Authentication and billing service restarting. Please retry in 90s."; + let mut waits = Vec::new(); + let result = worker.turn_with_wait(&mut ws, &mut agent, "Build the requested app", None, |_, seconds| { + waits.push(seconds); + Ok(()) + }); + assert_eq!(result.unwrap(), "Done"); + assert_eq!(waits, [3], "typed provider wait overrides the text hint"); + assert_eq!(agent.sent.len(), 2); + assert_eq!(agent.sent[0], "Build the requested app"); + assert_eq!(agent.sent[1], if saved { crate::failure::CONTINUE_TURN } else { "Build the requested app" }); + } + } + + #[test] + fn setup_errors_do_not_retry_even_with_a_wait_in_the_message() { + for kind in [ + FailureKind::Authentication, + FailureKind::Billing, + FailureKind::MissingProgram, + FailureKind::ToolsUnavailable, + FailureKind::Configuration, + FailureKind::Unknown, + ] { + let dir = tempfile::tempdir().unwrap(); + let mut ws = Workspace::create(dir.path(), "x", "Build a thing", TargetRef::None).unwrap(); + let mut worker = check_worker(&ws, false); + let mut agent = FailingConversation::new([Some(TurnFailure::new(kind))], true); + agent.message = "需要修改设置。Please retry in 1s."; + let result = worker.turn_with_wait(&mut ws, &mut agent, "Build it", None, |_, _| panic!("a setup error must not wait and retry")); + assert!(result.is_err(), "{kind:?}"); + assert_eq!(agent.sent.len(), 1, "{kind:?}"); + } + } + + #[test] + fn retries_are_bounded_and_cancelled_waits_send_nothing_more() { + assert_eq!(RunOptions::default().max_rounds, 6, "room to keep fixing until it works"); + let dir = tempfile::tempdir().unwrap(); + let mut ws = Workspace::create(dir.path(), "x", "Build a thing", TargetRef::None).unwrap(); + let mut worker = check_worker(&ws, false); + let failure = TurnFailure::new(FailureKind::Transient); + let mut agent = FailingConversation::new([Some(failure); 5], false); + let mut waits = Vec::new(); + assert!( + worker + .turn_with_wait(&mut ws, &mut agent, "Build it", None, |_, seconds| { + waits.push(seconds); + Ok(()) + }) + .is_err() + ); + assert_eq!(agent.sent.len(), 4, "initial attempt and at most three retries"); + assert_eq!(waits, [30, 60, 90]); + let mut agent = FailingConversation::new([Some(failure)], false); + let error = worker + .turn_with_wait(&mut ws, &mut agent, "Build it", None, |stop, _| { + stop.cancel(); + stop.check() + }) + .unwrap_err(); + assert_eq!(error.code, ErrorCode::Cancelled); + assert_eq!(agent.sent.len(), 1); + } + + #[test] + fn a_local_self_check_write_error_is_not_retried_as_a_backend_rate_limit() { + #[derive(Clone)] + struct BlockReply { + answer: PathBuf, + } + impl Check for BlockReply { + fn id(&self) -> &'static str { + "blocked-reply" + } + fn run(&self, cx: &mut CheckCx) -> Result { + // The request was pending when the run read it; its answer now cannot be + // written. The workspace itself remains writable, including retry status. + std::fs::create_dir_all(&self.answer).unwrap(); + Ok(CheckResult::new(Check::id(self), cx.round, true, "The check finished.")) + } + } + impl TargetKind for BlockReply { + fn id(&self) -> &'static str { + "blocked-reply" + } + fn words(&self) -> KindWords { + SlidesKind.words() + } + fn detect(&self, _: &DetectCx) -> Option { + Some(1) + } + fn default_stack(&self, cx: &DetectCx) -> &'static str { + SlidesKind.default_stack(cx) + } + fn goal(&self) -> &'static str { + SlidesKind.goal() + } + fn checks(&self, _: &Workspace) -> Vec> { + vec![Box::new(self.clone())] + } + } + struct RateLimited { + sent: usize, + first_poll: bool, + } + impl Conversation for RateLimited { + fn send(&mut self, _: &str) { + self.sent += 1; + self.first_poll = true; + } + fn poll(&mut self) -> Vec { + vec![if std::mem::take(&mut self.first_poll) { + AgentEvent::Error { message: "The provider is rate limited.".into(), remediation: None, log: None } + } else { + AgentEvent::TurnDone { backend: "scripted".into(), input_tokens: 0, output_tokens: 0, cost_usd: None, cost_is_estimate: false } + }] + } + fn last_failure(&self) -> Option { + Some(TurnFailure::new(FailureKind::RateLimited)) + } + fn answer_consent(&mut self, _: &str, _: bool) {} + fn stop(&mut self) {} + fn is_busy(&self) -> bool { + false + } + } + let dir = tempfile::tempdir().unwrap(); + let mut ws = Workspace::create(dir.path(), "x", "Build a thing", TargetRef::None).unwrap(); + let asks = Inbox::new(self_check::asks_dir(&ws.root)); + let id = asks.post(craft_anything_engine::inbox::RequestKind::Question, "Check the build", "", Vec::new(), false, None).unwrap(); + let mut worker = check_worker(&ws, false); + worker.self_check = Some((Arc::new(BlockReply { answer: asks.dir().join(format!("{id}.answer.json")) }), 90.0)); + let mut agent = RateLimited { sent: 0, first_poll: false }; + let mut waits = Vec::new(); + let error = worker + .turn_with_wait(&mut ws, &mut agent, "Build the app", None, |_, seconds| { + waits.push(seconds); + Ok(()) + }) + .unwrap_err(); + assert_eq!(agent.sent, 1, "a backend's retry hint cannot retry a local persistence failure"); + assert!(waits.is_empty()); + assert_ne!(error.code, ErrorCode::ProgramFailed); + assert!(error.message.contains(".answer.json"), "the original local failure is retained: {error:?}"); + } + + #[test] + fn an_api_rate_limit_continues_history_without_repeating_users_or_completed_tools() { + use crate::agent::tests::{Scripted, turn}; + use crate::agent::{ApiConversation, EngineHost, LoopConfig, ModelSlot}; + use crate::model::{Block, Speaker, StopReason}; + for write_first in [false, true] { + let dir = tempfile::tempdir().unwrap(); + let mut ws = Workspace::create(dir.path(), "x", "Build a notes app", TargetRef::None).unwrap(); + let mut worker = check_worker(&ws, false); + worker.attach(&ws).unwrap(); + let (tx, rx) = std::sync::mpsc::channel(); + worker.tx = tx; + let seen = Arc::default(); + let mut script = Scripted { turns: VecDeque::new(), seen: Arc::clone(&seen) }; + if write_first { + script.turns.push_back(turn( + vec![Block::ToolUse { + id: "write-once".into(), + name: "build_file_write".into(), + input: json!({"path":"app/README.md","content":"Built once"}), + }], + StopReason::ToolUse, + )); + } + script.turns.push_back(Err(ToolError::new(ErrorCode::OverLimit, "服务限流").with_details(json!({"status":429,"retry_after_secs":3})))); + script.turns.push_back(turn(vec![Block::Text { text: "Built the notes app".into() }], StopReason::EndTurn)); + let slot = ModelSlot { name: "Scripted".into(), model: Box::new(script), local: false, effort: None, max_turns: 10, context_tokens: 0 }; + let enabled = std::collections::BTreeSet::from([craft_anything_engine::Toolset::Core, craft_anything_engine::Toolset::Build]); + let mut api = ApiConversation::start_with(Arc::new(EngineHost(worker.engine.clone())), vec![slot], LoopConfig::default(), enabled).unwrap(); + let mut waits = Vec::new(); + assert_eq!( + worker + .turn_with_wait(&mut ws, &mut api, "Build the notes app", None, |_, seconds| { + waits.push(seconds); + Ok(()) + }) + .unwrap(), + "Built the notes app" + ); + assert_eq!(waits, [4]); + assert!(ws.manifest.agent_session.is_none()); + let requests = seen.lock().unwrap(); + assert_eq!(requests.len(), if write_first { 3 } else { 2 }); + let before_retry = &requests[requests.len() - 2]; + let after_retry = requests.last().unwrap(); + assert_eq!(before_retry, after_retry, "retry must continue the same model request"); + assert_eq!(after_retry.iter().filter(|m| m.role == Speaker::User && m.text() == "Build the notes app").count(), 1); + let calls = rx.try_iter().filter(|e| matches!(e, RunEvent::Agent(AgentEvent::ToolStarted { tool, .. }) if tool == "build_file_write")).count(); + assert_eq!(calls, usize::from(write_first), "a completed write is not replayed"); + if write_first { + assert_eq!(std::fs::read_to_string(ws.app_dir().join("README.md")).unwrap(), "Built once"); + } + } + } + + #[cfg(unix)] + fn executable_cli(ws: &Workspace, model: &str, script: &str) -> crate::cli_agents::CliConversation { + use std::os::unix::fs::PermissionsExt; + let program = ws.root.join("agent-stand-in.sh"); + std::fs::write(&program, script).unwrap(); + std::fs::set_permissions(&program, std::fs::Permissions::from_mode(0o755)).unwrap(); + let config = crate::BackendConfig { + name: "Scripted CLI".into(), + kind: BackendKind::AgentCli, + model: model.into(), + base_url: None, + ssh_host: None, + enabled: true, + }; + let options = crate::BackendOptions { command: Some(program.display().to_string()), timeout_secs: Some(10), ..Default::default() }; + let mut spec = CliSpec::new(config, options, false, None); + spec.project = Some(ws.root.clone()); + spec.preamble = Some(String::new()); + crate::cli_agents::CliConversation::new(spec) + } + + #[cfg(unix)] + #[test] + fn a_failed_required_tool_server_stops_the_owned_cli_without_resuming_it() { + let dir = tempfile::tempdir().unwrap(); + let mut ws = Workspace::create(dir.path(), "x", "Build a thing", TargetRef::None).unwrap(); + let mut worker = check_worker(&ws, false); + let (tx, rx) = std::sync::mpsc::channel(); + worker.tx = tx; + let mut cli = executable_cli( + &ws, + "claude-code", + concat!( + "#!/bin/sh\ncat >/dev/null\nprintf 'attempt\\n' >> attempts\n", + "printf '%s\\n' '{\"type\":\"system\",\"subtype\":\"init\",\"session_id\":\"cannot-build\",\"mcp_servers\":[{\"name\":\"craft-anything\",\"status\":\"failed\"}]}'\n", + "sleep 5\nprintf 'unreachable' > continued-without-tools\n" + ), + ); + let error = worker.turn_with_wait(&mut ws, &mut cli, "Build the app", None, |_, _| panic!("missing tools must not be retried")).unwrap_err(); + assert!(error.message.contains("tools did not start"), "{error:?}"); + assert_eq!(cli.last_failure().unwrap().kind, FailureKind::ToolsUnavailable); + assert!(!cli.is_busy()); + assert_eq!(std::fs::read_to_string(ws.root.join("attempts")).unwrap(), "attempt\n"); + assert!(!ws.root.join("continued-without-tools").exists(), "the owned child was stopped at the failed init"); + assert_eq!(rx.try_iter().filter(|e| matches!(e, RunEvent::Agent(AgentEvent::TurnDone { .. }))).count(), 1); + } + + #[cfg(unix)] + #[test] + fn a_plain_nonzero_cli_exit_resumes_its_saved_session() { + let dir = tempfile::tempdir().unwrap(); + let mut ws = Workspace::create(dir.path(), "x", "Build a thing", TargetRef::None).unwrap(); + let mut worker = check_worker(&ws, false); + let (tx, rx) = std::sync::mpsc::channel(); + worker.tx = tx; + let mut cli = executable_cli( + &ws, + "claude-code", + concat!( + "#!/bin/sh\ncat >> prompts\nprintf '\\n' >> prompts\nprintf 'attempt\\n' >> attempts\n", + "case \" $* \" in *\" --resume interrupted-session \"*)\n", + "printf '%s\\n' '{\"type\":\"assistant\",\"message\":{\"content\":[{\"type\":\"text\",\"text\":\"Built after the interruption\"}]}}'\n", + "printf '%s\\n' '{\"type\":\"result\",\"is_error\":false,\"result\":\"Built after the interruption\"}'\nexit 0;; esac\n", + "printf '%s\\n' '{\"type\":\"system\",\"subtype\":\"init\",\"session_id\":\"interrupted-session\"}'\nexit 1\n" + ), + ); + let mut waits = Vec::new(); + let result = worker.turn_with_wait(&mut ws, &mut cli, "Make a notes app", None, |_, seconds| { + waits.push(seconds); + Ok(()) + }); + assert_eq!(std::fs::read_to_string(ws.root.join("attempts")).unwrap(), "attempt\nattempt\n", "a plain nonzero exit must resume the reported session"); + assert_eq!(result.unwrap(), "Built after the interruption"); + assert_eq!(waits, [30]); + assert_eq!(std::fs::read_to_string(ws.root.join("prompts")).unwrap(), format!("Make a notes app\n{}\n", crate::failure::CONTINUE_TURN)); + assert_eq!(cli.session_id().as_deref(), Some("interrupted-session")); + assert!(cli.last_failure().is_none(), "successful recovery clears the process failure"); + assert!(!cli.is_busy()); + assert_eq!(rx.try_iter().filter(|e| matches!(e, RunEvent::Agent(AgentEvent::TurnDone { .. }))).count(), 2); + } + + #[cfg(unix)] + #[test] + fn typed_cli_setup_errors_take_precedence_over_a_nonzero_exit() { + for (code, kind) in + [("authentication_failed", FailureKind::Authentication), ("billing_error", FailureKind::Billing), ("model_not_found", FailureKind::Configuration)] + { + let dir = tempfile::tempdir().unwrap(); + let mut ws = Workspace::create(dir.path(), "x", "Build a thing", TargetRef::None).unwrap(); + let mut worker = check_worker(&ws, false); + let error = json!({"type":"assistant","error":code,"message":{"content":[{"type":"text","text":"Please retry in 1s."}]}}); + let script = format!("#!/bin/sh\ncat >/dev/null\nprintf 'attempt\\n' >> attempts\nprintf '%s\\n' '{error}'\nexit 1\n"); + let mut cli = executable_cli(&ws, "claude-code", &script); + let result = worker.turn_with_wait(&mut ws, &mut cli, "Make a notes app", None, |_, _| panic!("typed setup failures must not retry")); + assert!(result.is_err(), "{kind:?}"); + assert_eq!(cli.last_failure().unwrap().kind, kind); + assert_eq!(std::fs::read_to_string(ws.root.join("attempts")).unwrap(), "attempt\n"); + assert!(!cli.is_busy()); + } + } + + #[cfg(unix)] + #[test] + fn opencode_retries_a_typed_rate_limit_with_its_whole_original_request() { + let dir = tempfile::tempdir().unwrap(); + let mut ws = Workspace::create(dir.path(), "x", "Build a thing", TargetRef::None).unwrap(); + let mut worker = check_worker(&ws, false); + let (tx, rx) = std::sync::mpsc::channel(); + worker.tx = tx; + let mut cli = executable_cli( + &ws, + "opencode", + concat!( + "#!/bin/sh\nif [ \"$2\" = '--help' ]; then printf '%s\\n' '--format'; exit 0; fi\n", + "for arg do last=\"$arg\"; done\nprintf '%s\\n' \"$last\" >> prompts\n", + "if [ ! -f rate-was-returned ]; then touch rate-was-returned\n", + "printf '%s\\n' '{\"type\":\"error\",\"error\":{\"name\":\"APIError\",\"data\":{\"message\":\"Please retry in 19.052906498s.\",\"statusCode\":429,\"isRetryable\":true}}}'\nexit 1\nfi\n", + "printf '%s\\n' '{\"type\":\"text\",\"part\":{\"text\":\"Built the app\"}}'\n" + ), + ); + let mut waits = Vec::new(); + assert_eq!( + worker + .turn_with_wait(&mut ws, &mut cli, "Make a notes app", None, |_, seconds| { + waits.push(seconds); + Ok(()) + }) + .unwrap(), + "Built the app" + ); + assert_eq!(waits, [21]); + assert_eq!(std::fs::read_to_string(ws.root.join("prompts")).unwrap(), "Make a notes app\nMake a notes app\n"); + assert!(cli.session_id().is_none()); + assert!(cli.last_failure().is_none(), "successful retry clears the prior failure"); + assert!(!cli.is_busy()); + assert_eq!(rx.try_iter().filter(|e| matches!(e, RunEvent::Agent(AgentEvent::TurnDone { .. }))).count(), 2, "one final event per process attempt"); + } + + #[cfg(unix)] + #[test] + fn an_untyped_opencode_exit_keeps_its_error_and_the_three_retry_cap() { + let dir = tempfile::tempdir().unwrap(); + let mut ws = Workspace::create(dir.path(), "x", "Build a thing", TargetRef::None).unwrap(); + let mut worker = check_worker(&ws, false); + let mut cli = executable_cli( + &ws, + "opencode", + "#!/bin/sh\nif [ \"$2\" = '--help' ]; then printf 'old help\\n'; exit 0; fi\nprintf 'attempt\\n' >> attempts\nprintf 'Unknown option: --format. Please retry in 1s.\\n' >&2\nexit 2\n", + ); + let mut waits = Vec::new(); + let error = worker + .turn_with_wait(&mut ws, &mut cli, "Make a notes app", None, |_, seconds| { + waits.push(seconds); + Ok(()) + }) + .unwrap_err(); + assert!(error.message.contains("Unknown option"), "{error:?}"); + assert_eq!(cli.last_failure().unwrap().kind, FailureKind::Transient); + assert!(!cli.is_busy()); + assert_eq!(waits, [2, 2, 2], "the display wait hint affects only the delay after a retryable exit"); + assert_eq!(std::fs::read_to_string(ws.root.join("attempts")).unwrap(), "attempt\nattempt\nattempt\nattempt\n"); } - assert!(worth_retrying("Please retry in 19.05s"), "a rate limit is waited out, not handed to the person"); } #[test] @@ -1874,7 +2271,7 @@ fn a_missing_session_with_nothing_to_start_from_is_said_plainly() { usage: None, self_check: None, }; - let e = w.turn(&mut ws, &mut agent, "Carry on.", None).unwrap_err(); + let e = w.turn_with_wait(&mut ws, &mut agent, "Carry on.", None, |_, _| panic!("a missing session without a fresh start must not be retried")).unwrap_err(); assert!(e.message.contains("isn't on this computer"), "{e:?}"); assert!(ws.manifest.agent_session.is_none()); } diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 25c62a97..b010cf2b 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -170,18 +170,18 @@ Every round spends the person's money or quota. One that can't do anything shoul across runs: a run ends at 7 checks (`max_rounds`) but check numbers carry on, so #44's check 28 and #417's 35+ look like several runs (a guess: Try again is the only restart found, and Auto mode doesn't start runs) -- [ ] A builder whose tools didn't start isn't resumed. `turn` (`rebuild/mod.rs`) tries 3 more times +- [x] A builder whose tools didn't start isn't resumed. `turn` (`rebuild/mod.rs`) tries 3 more times ("Picking up where it left off", #377) because `worth_retrying` doesn't know this error. Give it a typed signal, as `Conversation::session_lost` does, not one more word in that function's - list. Say what failed and stop (pitfalls: never retry what can't succeed) -- [ ] `worth_retrying` decides on English words in a message (AGENTS.md: structured results). Move - its other cases (sign-in, credit, a missing install) to the same typed signal + list. Say what failed and stop (pitfalls: never retry what can't succeed) — commit `f80a16a66834` +- [x] `worth_retrying` decides on English words in a message (AGENTS.md: structured results). Move + its other cases (sign-in, credit, a missing install) to the same typed signal — commit `f80a16a66834` - [ ] Find what repeated #64's attempts and stop it: it logged 41, about 38 s apart, on 0.2.0, but `turn` allows 3 retries 30, 60 and 90 s apart, there too, so something else starts them -- [ ] A rate limit is waited for only on agents with a session: `turn` retries only when +- [x] A rate limit is waited for only on agents with a session: `turn` retries only when `agent.session_id()` is set, which API backends and OpenCode's plain adapter never set (read in the code, not run). So #148 didn't reach #142, which was OpenCode. Wait and send the turn - again for them too, then close #142 + again for them too, then close #142 — commit `f80a16a66834` - [ ] **decision** Whether a build gets a token or cost limit of its own, beyond rounds: #399 used 9.4M tokens in 19 minutes of the first stage, then failed to compile diff --git a/docs/STATUS.md b/docs/STATUS.md index fdd9b072..f8c98c25 100644 --- a/docs/STATUS.md +++ b/docs/STATUS.md @@ -41,6 +41,39 @@ workspace clippy, layers, fmt and the secret scan of all history were clean. Chi | SQLite database files (`db_sqlite_show`) | A rowid alias column (`INTEGER PRIMARY KEY`, including the table-constraint form) shows the rowid; `DESC`, non-`INTEGER` and composite keys keep their stored values | Hand-built fixtures; generated files compared with the official SQLite library | | SQLite WITHOUT ROWID tables | Read from the index b-tree, interior records and overflow included; rows in primary-key order, values in declaration order, `rowid` null | Hand-built fixtures; generated files compared with the official SQLite library | +### Builder retry policy + +Builder failures carry a typed reason through `Conversation::last_failure`. Typed authentication, +billing, missing programs, unavailable tools, configuration and unknown protocol failures stop +for attention. Transport interruptions, rate limits and nonzero process exits without a typed +error get at most three retries; the decision no longer searches English error messages. +Structured provider waits win; the earlier numeric “retry in” +hint is only a delay fallback after a failure has already been classified as retryable. All waits +remain cancellable and bounded to fifteen minutes. +Local inbox and workspace write failures are not retried using a concurrent backend failure's +retry advice; the original local error is returned. +A missing saved session starts afresh only when a fresh request is available; otherwise the run +stops without retrying the unavailable session. + +Claude Code's failed or unauthenticated `craft-anything` MCP server in its init event stops the +owned CLI process. Pending servers and failures in unrelated optional servers do not stop it. +Claude assistant error codes, HTTP status/provider errors and ACP authentication errors retain +their typed meanings. A later generic result does not erase a more specific error. + +API retries continue the existing history without inserting the user's message again or replaying +completed tools. Stateless CLI retries receive the whole original request. OpenCode now uses +`run --format json`: its API error status and retry flag decide retryability while text, tools and +usage still reach the feed. Missing or invalid step costs keep the turn's cost unknown; explicit +zero remains a known cost. A CLI that fails with only a numeric exit also uses bounded retries; +its display error remains available if recovery fails. The process wrapper sends one final turn +event and becomes available for the next request. + +Protocol fixtures, a loopback HTTP server, scripted API turns and executable CLI stand-ins cover +these paths, including cancellation and the retry cap. An executable that exits 1 without a +structured error resumes its saved session, while typed setup errors retain priority over the +same exit status. No live provider or paid-model run was performed. The separate source of #64's +41 attempts remains unverified. + ### Bun stored-text encoding Packed Bun members with encoding value 2 are labelled `utf16le`, matching their stored UTF-16LE