From eface7be2017a21b6d162f398616962a9c098832 Mon Sep 17 00:00:00 2001 From: kev1n77 Date: Sat, 5 Sep 2026 17:35:00 +0800 Subject: [PATCH] fix(agent): stop retrying deterministic provider errors --- src/apps/cli/AGENTS.md | 6 + .../exec_cli_contracts.rs | 60 +++- src/apps/cli/tests/support/mod.rs | 18 ++ .../core/src/agentic/execution/AGENTS.md | 6 + .../src/agentic/execution/round_executor.rs | 282 +++++++++++++++++- 5 files changed, 358 insertions(+), 14 deletions(-) diff --git a/src/apps/cli/AGENTS.md b/src/apps/cli/AGENTS.md index 8a2972816c..f7c72cd21d 100644 --- a/src/apps/cli/AGENTS.md +++ b/src/apps/cli/AGENTS.md @@ -142,6 +142,12 @@ cargo check -p openbitfun-cli cargo test -p openbitfun-cli ``` +For streaming `exec` retry, context recovery, and final-event contracts: + +```bash +cargo test --locked -p openbitfun-cli --test cli_command_contracts exec_cli_contracts::stream_json_ +``` + When a CLI change crosses a shared boundary, use the focused command maintained by that owner: Agent Runtime for port/SDK behavior, the IPC adapter for shared protocol behavior, Core for turn/tool/persistence behavior, Terminal for diff --git a/src/apps/cli/tests/cli_command_contracts/exec_cli_contracts.rs b/src/apps/cli/tests/cli_command_contracts/exec_cli_contracts.rs index dd58a7bec4..dc1fb00eaa 100644 --- a/src/apps/cli/tests/cli_command_contracts/exec_cli_contracts.rs +++ b/src/apps/cli/tests/cli_command_contracts/exec_cli_contracts.rs @@ -448,6 +448,58 @@ fn stream_json_malformed_sse_retries_then_completes() { ); } +#[test] +fn stream_json_context_overflow_compresses_before_reissuing_the_model_request() { + let server = MockOpenAiServer::context_overflow_then_immediate(); + let environment = CliTestEnvironment::new(); + environment.configure_mock_model(server.base_url()); + let mut command = environment.std_command(); + command.args([ + "exec", + "Remember this request and continue after context recovery", + "--output-format", + "stream-json", + ]); + let output = command_output_with_timeout(&mut command, std::time::Duration::from_secs(30)); + let stdout = stdout(&output); + assert!(output.status.success(), "{}\n{stdout}", stderr(&output)); + // One rejected request, one summary request, then the recovered model round. + server.assert_chat_completion_requests(3); + let requests = server.chat_completion_request_bodies(); + assert_ne!( + requests[0]["messages"], requests[1]["messages"], + "overflow must enter compression instead of replaying the original request" + ); + assert_ne!( + requests[0]["messages"], requests[2]["messages"], + "recovery must send the compressed context" + ); + let events = jsonl_events(&stdout); + let compression_started = events + .iter() + .position(|value| { + value["event"]["type"] == "ContextCompressionStarted" + && value["event"]["trigger"] == "context_overflow_recovery" + }) + .expect("overflow should start recovery compression"); + let compression_completed = events + .iter() + .position(|value| value["event"]["type"] == "ContextCompressionCompleted") + .expect("recovery compression should complete"); + assert!(compression_started < compression_completed); + assert_eq!( + events + .iter() + .filter(|value| is_terminal_event(value)) + .count(), + 1 + ); + assert_eq!( + events.last().unwrap()["event"]["type"], + "DialogTurnCompleted" + ); +} + #[test] fn stream_json_provider_http_403_emits_one_error_terminal() { let server = MockOpenAiServer::http_403("provider authorization denied"); @@ -461,7 +513,7 @@ fn stream_json_provider_http_403_emits_one_error_terminal() { "stream-json", ]); let output = command_output_with_timeout(&mut command, std::time::Duration::from_secs(30)); - server.assert_chat_completion_requests(10); + server.assert_chat_completion_requests(1); let stdout = stdout(&output); assert!(!output.status.success(), "{stdout}"); @@ -524,7 +576,7 @@ fn stream_json_provider_and_patch_failures_publish_one_final_classification() { &output_target, ]); let output = command_output_with_timeout(&mut command, std::time::Duration::from_secs(30)); - server.assert_chat_completion_requests(10); + server.assert_chat_completion_requests(1); let stdout = stdout(&output); let stderr = stderr(&output); @@ -562,7 +614,7 @@ fn stream_json_provider_and_patch_failures_publish_one_final_classification() { } #[test] -fn stream_json_disconnect_then_exhausted_retry_failure_emits_one_error_terminal() { +fn stream_json_disconnect_then_authorization_failure_emits_one_error_terminal() { let server = MockOpenAiServer::disconnect_then_http_403(); let environment = CliTestEnvironment::new(); environment.configure_mock_model(server.base_url()); @@ -574,7 +626,7 @@ fn stream_json_disconnect_then_exhausted_retry_failure_emits_one_error_terminal( "stream-json", ]); let output = command_output_with_timeout(&mut command, std::time::Duration::from_secs(30)); - server.assert_chat_completion_requests(10); + server.assert_chat_completion_requests(2); let stdout = stdout(&output); assert!(!output.status.success(), "{stdout}"); diff --git a/src/apps/cli/tests/support/mod.rs b/src/apps/cli/tests/support/mod.rs index 3bd0012197..edf2d04361 100644 --- a/src/apps/cli/tests/support/mod.rs +++ b/src/apps/cli/tests/support/mod.rs @@ -320,6 +320,7 @@ enum MockModelResponse { Http403 { reason: String }, DisconnectThenHttp403, MalformedSseThenImmediate, + ContextOverflowThenImmediate, } impl MockOpenAiServer { @@ -349,6 +350,10 @@ impl MockOpenAiServer { Self::spawn(MockModelResponse::MalformedSseThenImmediate) } + pub(crate) fn context_overflow_then_immediate() -> Self { + Self::spawn(MockModelResponse::ContextOverflowThenImmediate) + } + pub(crate) fn base_url(&self) -> &str { &self.base_url } @@ -437,6 +442,10 @@ impl MockOpenAiServer { | MockModelResponse::DisconnectThenHttp403 ) || (matches!(response, MockModelResponse::MalformedSseThenImmediate) && attempt < 2) + || (matches!( + response, + MockModelResponse::ContextOverflowThenImmediate + ) && attempt < 3) || (matches!(response, MockModelResponse::ProductControlLoop) && attempt < 5); if accepts_more_requests { @@ -488,6 +497,15 @@ fn serve_model_response( release_stream: &mpsc::Receiver<()>, stream_disconnected: &mpsc::Sender<()>, ) { + if matches!(response, MockModelResponse::ContextOverflowThenImmediate) && attempt == 0 { + let body = json!({ + "error": {"code": "context_length_exceeded", "message": "Maximum context length exceeded"} + }).to_string(); + write!(stream, "HTTP/1.1 400 Bad Request\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()) + .expect("write context overflow response"); + stream.flush().expect("flush context overflow response"); + return; + } if matches!(response, MockModelResponse::DisconnectThenHttp403) && attempt > 0 { write_http_403(stream, "provider stream remained unavailable") .expect("write post-disconnect HTTP error"); diff --git a/src/crates/assembly/core/src/agentic/execution/AGENTS.md b/src/crates/assembly/core/src/agentic/execution/AGENTS.md index b5044b29c8..c632a73721 100644 --- a/src/crates/assembly/core/src/agentic/execution/AGENTS.md +++ b/src/crates/assembly/core/src/agentic/execution/AGENTS.md @@ -1,5 +1,11 @@ If you modify `stream_processor.rs`, run the stream integration tests before finishing. +For model retry admission and recovery, use: + +```bash +cargo test --locked -p openbitfun-core --no-default-features --features agent-runtime,git --lib agentic::execution::round_executor::tests +``` + For complete shell constraint checks, use: ```bash diff --git a/src/crates/assembly/core/src/agentic/execution/round_executor.rs b/src/crates/assembly/core/src/agentic/execution/round_executor.rs index 19095884a2..2a55a2542d 100644 --- a/src/crates/assembly/core/src/agentic/execution/round_executor.rs +++ b/src/crates/assembly/core/src/agentic/execution/round_executor.rs @@ -122,7 +122,28 @@ impl RoundExecutor { const MAX_RATE_LIMIT_DELAY_MS: u64 = 60_000; const MAX_RETRY_EXPONENT_SHIFT: u32 = 6; - fn exhausted_request_error(error: &anyhow::Error, attempts: u32) -> OpenBitFunError { + /// Unknown and malformed provider responses retain the bounded recovery + /// introduced with the unified attempt budget. Only classified rejections + /// leave this loop immediately; context overflow is recovered by the caller. + fn should_retry_provider_error(category: &ErrorCategory) -> bool { + match category { + ErrorCategory::Auth + | ErrorCategory::Permission + | ErrorCategory::ProviderQuota + | ErrorCategory::ProviderBilling + | ErrorCategory::InvalidRequest + | ErrorCategory::ContentPolicy + | ErrorCategory::ContextOverflow => false, + ErrorCategory::Network + | ErrorCategory::RateLimit + | ErrorCategory::Timeout + | ErrorCategory::ProviderUnavailable + | ErrorCategory::ModelError + | ErrorCategory::Unknown => true, + } + } + + fn terminal_request_error(error: &anyhow::Error, attempts: u32) -> OpenBitFunError { let mut provider_error = error .downcast_ref::() .cloned() @@ -451,7 +472,9 @@ impl RoundExecutor { error!("AI request failed: {:#}", e); let provider_error = e.downcast_ref::().cloned(); let err_msg = format!("{e:#}"); - if local_attempt_index < max_attempts - 1 { + let error = Self::terminal_request_error(&e, lifecycle.attempts_started()); + let retryable = Self::should_retry_provider_error(&error.error_category()); + if retryable && local_attempt_index < max_attempts - 1 { self.record_retry_diagnostic( &context, &round_id, @@ -481,12 +504,12 @@ impl RoundExecutor { local_attempt_index += 1; continue; } - let error = Self::exhausted_request_error(&e, lifecycle.attempts_started()); warn!( - "AI request retry budget exhausted: session_id={}, round_id={}, attempts={}, category={:?}, error={}", + "AI request stopped: session_id={}, round_id={}, attempts={}, reason={}, category={:?}, error={}", context.session_id, round_id, lifecycle.attempts_started(), + if retryable { "retry_budget_exhausted" } else { "non_retryable_error" }, error.error_category(), error ); @@ -808,6 +831,7 @@ impl RoundExecutor { Err(stream_err) => { let err_msg = stream_err.error.to_string(); let stream_error_category = stream_err.error.error_category(); + let retryable = Self::should_retry_provider_error(&stream_error_category); let provider_error = match &stream_err.error { OpenBitFunError::AIProvider(error) | OpenBitFunError::RecoverableContextOverflow(error) => Some(error), @@ -819,7 +843,7 @@ impl RoundExecutor { Self::error_trace_response("error", err_msg.clone()), ) .await; - if local_attempt_index < max_attempts - 1 { + if retryable && local_attempt_index < max_attempts - 1 { self.record_retry_diagnostic( &context, &round_id, @@ -852,10 +876,11 @@ impl RoundExecutor { continue; } warn!( - "Stream retry budget exhausted: session_id={}, round_id={}, attempts={}, effective_output={}, category={:?}, error={}", + "Stream stopped: session_id={}, round_id={}, attempts={}, reason={}, effective_output={}, category={:?}, error={}", context.session_id, round_id, - max_attempts, + lifecycle.attempts_started(), + if retryable { "retry_budget_exhausted" } else { "non_retryable_error" }, stream_err.has_effective_output, stream_error_category, err_msg @@ -1746,6 +1771,243 @@ mod tests { } } + struct RetryTestServer { + url: String, + requests: Arc>>, + stop: Arc, + thread: Option>, + } + + impl RetryTestServer { + fn new(replies: Vec<(u16, String)>) -> Self { + use std::io::{BufRead, Read, Write}; + use std::sync::atomic::{AtomicBool, Ordering}; + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let url = format!( + "http://{}/v1/chat/completions", + listener.local_addr().unwrap() + ); + let requests = Arc::new(std::sync::Mutex::new(Vec::new())); + let captured = requests.clone(); + let stop = Arc::new(AtomicBool::new(false)); + let stopped = stop.clone(); + assert!(!replies.is_empty()); + let thread = std::thread::spawn(move || { + while !stopped.load(Ordering::Relaxed) { + let mut socket = match listener.accept() { + Ok((socket, _)) => socket, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + std::thread::sleep(Duration::from_millis(5)); + continue; + } + Err(error) => panic!("accept retry fixture request: {error}"), + }; + socket + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + socket + .set_write_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut reader = std::io::BufReader::new(&mut socket); + let mut content_length = 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(':') { + if name.eq_ignore_ascii_case("content-length") { + content_length = value.trim().parse::().unwrap(); + } + } + } + let mut body = vec![0; content_length]; + reader.read_exact(&mut body).unwrap(); + let mut requests = captured.lock().unwrap(); + let index = requests.len().min(replies.len() - 1); + requests.push(serde_json::from_slice(&body).unwrap()); + drop(requests); + let (status, body) = &replies[index]; + let content_type = if *status == 200 { + "text/event-stream" + } else { + "application/json" + }; + write!(socket, "HTTP/1.1 {status} Fixture\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).unwrap(); + } + }); + Self { + url, + requests, + stop, + thread: Some(thread), + } + } + + fn client(&self) -> Arc { + Arc::new(crate::infrastructure::ai::AIClient::new( + openbitfun_core_types::AIConfig { + name: "retry-test".to_string(), + base_url: self.url.clone(), + request_url: self.url.clone(), + api_key: "retry-test-key".to_string(), + model: "retry-test-model".to_string(), + format: "openai".to_string(), + context_window: 4096, + max_tokens: Some(128), + temperature: None, + top_p: None, + inline_think_in_text: false, + custom_headers: None, + custom_headers_mode: None, + skip_ssl_verify: false, + custom_request_body: None, + custom_request_body_mode: None, + }, + )) + } + } + + impl Drop for RetryTestServer { + fn drop(&mut self) { + self.stop.store(true, std::sync::atomic::Ordering::Relaxed); + if let Some(thread) = self.thread.take() { + if let Err(error) = thread.join() { + if !std::thread::panicking() { + std::panic::resume_unwind(error); + } + } + } + } + } + + fn retry_test_success() -> (u16, String) { + ( + 200, + format!( + "data: {}\n\ndata: [DONE]\n\n", + json!({ + "id": "retry-test", + "object": "chat.completion.chunk", + "created": 1, + "model": "retry-test-model", + "choices": [{"index": 0, "delta": {"content": "Recovered"}, "finish_reason": "stop"}] + }) + ), + ) + } + + #[tokio::test] + async fn provider_rejections_stop_after_one_request_for_http_and_stream_errors() { + for (status, code, category) in [ + (401, "invalid_api_key", ErrorCategory::Auth), + (403, "permission_error", ErrorCategory::Permission), + (413, "invalid_request_error", ErrorCategory::InvalidRequest), + (402, "insufficient_quota", ErrorCategory::ProviderQuota), + ( + 400, + "context_length_exceeded", + ErrorCategory::ContextOverflow, + ), + ] { + for in_stream in [false, true] { + let body = + json!({"error": {"code": code, "message": "Request rejected"}}).to_string(); + let reply = if in_stream { + (200, format!("data: {body}\n\n")) + } else { + (status, body) + }; + // A second request would succeed, making an accidental retry + // fail this test immediately instead of waiting for the budget. + let server = RetryTestServer::new(vec![reply, retry_test_success()]); + let executor = test_round_executor(); + let error = tokio::time::timeout( + Duration::from_secs(5), + executor.execute_round( + server.client(), + test_round_context(), + vec![super::AIMessage::user("Original request".to_string())], + None, + None, + ), + ) + .await + .expect("deterministic error should return promptly") + .expect_err("rejection must not retry"); + assert_eq!( + error.error_category(), + category, + "code={code}, in_stream={in_stream}" + ); + assert_eq!( + error.is_recoverable_context_overflow(), + category == ErrorCategory::ContextOverflow + ); + assert_eq!(server.requests.lock().unwrap().len(), 1); + let events = executor.event_queue.dequeue_batch(100).await; + assert!(!events.iter().any(|event| matches!( + event.event, + AgenticEvent::ModelRoundAttemptSuperseded { .. } + ))); + } + } + } + + #[tokio::test] + async fn transient_and_malformed_provider_responses_still_retry() { + for reply in [ + ( + 429, + json!({"error": {"code": "rate_limit_exceeded", "message": "Try later"}}) + .to_string(), + ), + ( + 503, + json!({"error": {"message": "Temporarily unavailable"}}).to_string(), + ), + (200, "data: not-json\n\n".to_string()), + ( + 200, + format!( + "data: {}\n\n", + json!({"error": {"code": "unrecognized", "message": "Unclassified provider failure"}}) + ), + ), + ] { + let server = RetryTestServer::new(vec![reply, retry_test_success()]); + let result = tokio::time::timeout( + Duration::from_secs(5), + test_round_executor().execute_round( + server.client(), + test_round_context(), + vec![super::AIMessage::user("Retry safely".to_string())], + None, + None, + ), + ) + .await + .expect("one retry should complete") + .expect("recoverable response should retry"); + assert!(result.had_assistant_text); + assert_eq!(server.requests.lock().unwrap().len(), 2); + } + } + + #[test] + fn terminal_request_classification_preserves_full_error_chain() { + let source = anyhow::anyhow!("invalid api key").context("Provider request failed"); + let error = RoundExecutor::terminal_request_error(&source, 1); + assert_eq!(error.error_category(), ErrorCategory::Auth); + assert!(!RoundExecutor::should_retry_provider_error( + &error.error_category() + )); + assert!(error.to_string().contains("invalid api key")); + } + #[test] fn resolves_global_project_and_agent_permission_rules_before_execution() { let mut global = GlobalConfig::default(); @@ -2150,7 +2412,7 @@ mod tests { "OpenAI Streaming API TTFT timeout after 30s waiting for first effective stream output" ) .context("OpenAI Streaming API failed after 1 attempts"); - let error = RoundExecutor::exhausted_request_error(&source, 10); + let error = RoundExecutor::terminal_request_error(&source, 10); assert_eq!(error.error_category(), ErrorCategory::Timeout); assert!(error .to_string() @@ -2168,7 +2430,7 @@ mod tests { ); provider.category = ErrorCategory::RateLimit; let source = anyhow::Error::new(provider); - let error = RoundExecutor::exhausted_request_error(&source, 10); + let error = RoundExecutor::terminal_request_error(&source, 10); let detail = error.error_detail(); assert_eq!(detail.category, ErrorCategory::RateLimit); assert_eq!(detail.http_status, Some(403)); @@ -2203,7 +2465,7 @@ mod tests { Some(400), )); assert!(matches!( - RoundExecutor::exhausted_request_error(&source, 10), + RoundExecutor::terminal_request_error(&source, 10), OpenBitFunError::RecoverableContextOverflow(_) )); }