| 1 | //! Process-level acceptance for the v0.9.4 Terminal-Bench P0 exit-path fix. |
| 2 | //! |
| 3 | //! Benchmark evidence (Terminal-Bench 2.1, codewhale 0.9.4): five tasks were |
| 4 | //! forfeited when the DeepSeek stream dropped mid-response ("error decoding |
| 5 | //! response body" after partial content). The engine surfaced the warning |
| 6 | //! and `codewhale exec` exited 1, and Harbor raised |
| 7 | //! `NonZeroAgentExitCodeError`. The fix: headless turns re-issue the request |
| 8 | //! after a mid-stream network drop (bounded by MAX_STREAM_RETRIES), and a |
| 9 | //! turn that still fails exits `EX_TEMPFAIL` (75) — a retryable |
| 10 | //! infrastructure failure the harness can distinguish from a genuine task |
| 11 | //! failure (exit 1). |
| 12 | //! |
| 13 | //! These tests drive the real binary against a raw TCP server that sends a |
| 14 | //! partial SSE body and then closes mid-`content-length`, reproducing the |
| 15 | //! exact reqwest decode failure from the bench artifacts. |
| 16 | |
| 17 | use std::io::Read; |
| 18 | use std::path::{Path, PathBuf}; |
| 19 | use std::process::{Command, Stdio}; |
| 20 | use std::sync::Arc; |
| 21 | use std::sync::atomic::{AtomicUsize, Ordering}; |
| 22 | use std::time::Duration; |
| 23 | |
| 24 | use serde_json::{Value, json}; |
| 25 | use tempfile::TempDir; |
| 26 | use wait_timeout::ChildExt; |
| 27 | |
| 28 | const MODEL: &str = "stream-drop-test"; |
| 29 | /// The server claims a body far larger than it delivers on a "drop" response, |
| 30 | /// so hyper raises `error decoding response body` after the first SSE chunk — |
| 31 | /// the exact failure string in the Terminal-Bench crash artifacts. |
| 32 | const CLAIMED_DROP_BODY_LEN: usize = 1_048_576; |
| 33 | |
| 34 | #[tokio::test(flavor = "multi_thread", worker_threads = 2)] |
| 35 | async fn headless_exec_recovers_from_mid_stream_drop() { |
| 36 | let workspace = TempDir::new().expect("workspace"); |
| 37 | let home = TempDir::new().expect("home"); |
| 38 | let (base_url, chat_posts, _server) = start_flaky_server(1).await; |
| 39 | |
| 40 | let output = run_exec(workspace.path(), home.path(), &base_url); |
| 41 | |
| 42 | assert!( |
| 43 | output.status.success(), |
| 44 | "a recovered stream drop must exit 0\nstdout:\n{}\nstderr:\n{}", |
| 45 | String::from_utf8_lossy(&output.stdout), |
| 46 | String::from_utf8_lossy(&output.stderr) |
| 47 | ); |
| 48 | assert_eq!( |
| 49 | chat_posts.load(Ordering::SeqCst), |
| 50 | 2, |
| 51 | "the dropped attempt must be re-issued exactly once" |
| 52 | ); |
| 53 | let stdout = String::from_utf8_lossy(&output.stdout); |
| 54 | assert!( |
| 55 | stdout.contains("recovered after retry"), |
| 56 | "the retried turn's content must stream: {stdout}" |
| 57 | ); |
| 58 | assert!( |
| 59 | !stdout.contains(r#""type":"error""#), |
| 60 | "a transient drop that the retry recovers must not surface an error event: {stdout}" |
| 61 | ); |
| 62 | let meta = terminal_metadata(&stdout); |
| 63 | assert_eq!( |
| 64 | meta["meta"]["status"].as_str(), |
| 65 | Some("completed"), |
| 66 | "recovered run must record a completed terminal receipt: {meta}" |
| 67 | ); |
| 68 | } |
| 69 | |
| 70 | #[tokio::test(flavor = "multi_thread", worker_threads = 2)] |
| 71 | async fn headless_exec_exits_ex_tempfail_after_drop_budget_exhausted() { |
| 72 | let workspace = TempDir::new().expect("workspace"); |
| 73 | let home = TempDir::new().expect("home"); |
| 74 | // More drops than the engine can consume: initial attempt + |
| 75 | // MAX_STREAM_RETRIES (3) resumes, then the turn must fail. |
| 76 | let (base_url, chat_posts, _server) = start_flaky_server(usize::MAX).await; |
| 77 | |
| 78 | let output = run_exec(workspace.path(), home.path(), &base_url); |
| 79 | |
| 80 | assert_eq!( |
| 81 | output.status.code(), |
| 82 | Some(75), |
| 83 | "retry budget exhausted on a network-class failure must exit EX_TEMPFAIL (75), \ |
| 84 | not the generic task-failure 1\nstdout:\n{}\nstderr:\n{}", |
| 85 | String::from_utf8_lossy(&output.stdout), |
| 86 | String::from_utf8_lossy(&output.stderr) |
| 87 | ); |
| 88 | assert_eq!( |
| 89 | chat_posts.load(Ordering::SeqCst), |
| 90 | 4, |
| 91 | "initial attempt plus the bounded resume budget (MAX_STREAM_RETRIES = 3)" |
| 92 | ); |
| 93 | let stdout = String::from_utf8_lossy(&output.stdout); |
| 94 | let error_events = stdout |
| 95 | .lines() |
| 96 | .filter(|line| line.contains(r#""type":"error""#)) |
| 97 | .count(); |
| 98 | assert_eq!( |
| 99 | error_events, 1, |
| 100 | "only the final, budget-exhausted attempt may emit an error event: {stdout}" |
| 101 | ); |
| 102 | let error_line = stdout |
| 103 | .lines() |
| 104 | .find(|line| line.contains(r#""type":"error""#)) |
| 105 | .expect("terminal error event"); |
| 106 | assert!( |
| 107 | error_line.contains("Provider stream connection dropped"), |
| 108 | "the error channel must carry the real failure: {error_line}" |
| 109 | ); |
| 110 | let meta = terminal_metadata(&stdout); |
| 111 | assert_eq!(meta["meta"]["status"].as_str(), Some("failed"), "{meta}"); |
| 112 | assert_eq!( |
| 113 | meta["meta"]["error_category"].as_str(), |
| 114 | Some("network"), |
| 115 | "the terminal receipt must classify the failure as retryable infra: {meta}" |
| 116 | ); |
| 117 | } |
| 118 | |
| 119 | /// Extract the terminal `metadata` event from the stream-json stdout. |
| 120 | fn terminal_metadata(stdout: &str) -> Value { |
| 121 | let line = stdout |
| 122 | .lines() |
| 123 | .rev() |
| 124 | .find(|line| line.contains(r#""type":"metadata""#)) |
| 125 | .unwrap_or_else(|| panic!("stream-json metadata event missing: {stdout}")); |
| 126 | serde_json::from_str(line).expect("metadata event is valid JSON") |
| 127 | } |
| 128 | |
| 129 | /// Spawn a raw HTTP server that answers `GET /v1/models` and, for every POST |
| 130 | /// to the chat endpoint, either truncates the SSE body mid-stream (the first |
| 131 | /// `drops` requests) or completes a normal text turn. Returns the base URL |
| 132 | /// and a counter of chat-completion POSTs. |
| 133 | async fn start_flaky_server( |
| 134 | drops: usize, |
| 135 | ) -> (String, Arc<AtomicUsize>, tokio::task::JoinHandle<()>) { |
| 136 | let listener = tokio::net::TcpListener::bind("127.0.0.1:0") |
| 137 | .await |
| 138 | .expect("bind flaky server"); |
| 139 | let addr = listener.local_addr().expect("flaky server addr"); |
| 140 | let chat_posts = Arc::new(AtomicUsize::new(0)); |
| 141 | let server_posts = Arc::clone(&chat_posts); |
| 142 | let task = tokio::spawn(async move { |
| 143 | loop { |
| 144 | let (mut socket, _) = match listener.accept().await { |
| 145 | Ok(pair) => pair, |
| 146 | Err(_) => break, |
| 147 | }; |
| 148 | let request = match read_http_request(&mut socket).await { |
| 149 | Some(request) => request, |
| 150 | None => continue, |
| 151 | }; |
| 152 | if request.starts_with("GET ") { |
| 153 | write_all( |
| 154 | &mut socket, |
| 155 | &http_response( |
| 156 | "application/json", |
| 157 | &json!({"object":"list","data":[{"id":MODEL,"object":"model"}]}) |
| 158 | .to_string(), |
| 159 | ), |
| 160 | ) |
| 161 | .await; |
| 162 | continue; |
| 163 | } |
| 164 | let call = server_posts.fetch_add(1, Ordering::SeqCst) + 1; |
| 165 | if call <= drops { |
| 166 | // Partial SSE (one real content chunk), then the socket |
| 167 | // closes with the declared content-length unmet — the |
| 168 | // production "Provider stream connection dropped" failure. |
| 169 | let partial = sse_chunk( |
| 170 | json!({"id":"drop","object":"chat.completion.chunk","model":MODEL,"choices":[{"index":0,"delta":{"content":"partial answer that must be discarded"},"finish_reason":null}]}), |
| 171 | ); |
| 172 | let head = format!( |
| 173 | "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ncontent-length: {CLAIMED_DROP_BODY_LEN}\r\nconnection: close\r\n\r\n" |
| 174 | ); |
| 175 | write_all(&mut socket, &format!("{head}{partial}")).await; |
| 176 | } else { |
| 177 | let body = [ |
| 178 | sse_chunk(json!({"id":"final","object":"chat.completion.chunk","model":MODEL,"choices":[{"index":0,"delta":{"content":"recovered after retry"},"finish_reason":null}]})), |
| 179 | sse_chunk(json!({"id":"final","object":"chat.completion.chunk","model":MODEL,"choices":[{"index":0,"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":20,"completion_tokens":2,"total_tokens":22}})), |
| 180 | "data: [DONE]\n\n".to_string(), |
| 181 | ] |
| 182 | .join(""); |
| 183 | write_all(&mut socket, &http_response("text/event-stream", &body)).await; |
| 184 | } |
| 185 | } |
| 186 | }); |
| 187 | (format!("http://{addr}/v1"), chat_posts, task) |
| 188 | } |
| 189 | |
| 190 | /// Read one HTTP request (headers plus the content-length body, when any). |
| 191 | async fn read_http_request(socket: &mut tokio::net::TcpStream) -> Option<String> { |
| 192 | use tokio::io::AsyncReadExt; |
| 193 | let mut buffer = Vec::new(); |
| 194 | let mut chunk = [0u8; 4096]; |
| 195 | let header_end = loop { |
| 196 | if let Some(pos) = find_subslice(&buffer, b"\r\n\r\n") { |
| 197 | break pos + 4; |
| 198 | } |
| 199 | let read = socket.read(&mut chunk).await.ok()?; |
| 200 | if read == 0 { |
| 201 | return None; |
| 202 | } |
| 203 | buffer.extend_from_slice(&chunk[..read]); |
| 204 | if buffer.len() > 1 << 20 { |
| 205 | return None; |
| 206 | } |
| 207 | }; |
| 208 | let headers = String::from_utf8_lossy(&buffer[..header_end]).to_string(); |
| 209 | let content_length = headers |
| 210 | .lines() |
| 211 | .find_map(|line| { |
| 212 | line.to_ascii_lowercase() |
| 213 | .strip_prefix("content-length:") |
| 214 | .and_then(|value| value.trim().parse::<usize>().ok()) |
| 215 | }) |
| 216 | .unwrap_or(0); |
| 217 | while buffer.len() < header_end + content_length { |
| 218 | let read = socket.read(&mut chunk).await.ok()?; |
| 219 | if read == 0 { |
| 220 | break; |
| 221 | } |
| 222 | buffer.extend_from_slice(&chunk[..read]); |
| 223 | } |
| 224 | Some(headers) |
| 225 | } |
| 226 | |
| 227 | fn find_subslice(haystack: &[u8], needle: &[u8]) -> Option<usize> { |
| 228 | haystack |
| 229 | .windows(needle.len()) |
| 230 | .position(|window| window == needle) |
| 231 | } |
| 232 | |
| 233 | async fn write_all(socket: &mut tokio::net::TcpStream, bytes: &str) { |
| 234 | use tokio::io::AsyncWriteExt; |
| 235 | let _ = socket.write_all(bytes.as_bytes()).await; |
| 236 | let _ = socket.shutdown().await; |
| 237 | } |
| 238 | |
| 239 | fn http_response(content_type: &str, body: &str) -> String { |
| 240 | format!( |
| 241 | "HTTP/1.1 200 OK\r\ncontent-type: {content_type}\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", |
| 242 | body.len() |
| 243 | ) |
| 244 | } |
| 245 | |
| 246 | fn sse_chunk(value: Value) -> String { |
| 247 | format!( |
| 248 | "data: {}\n\n", |
| 249 | serde_json::to_string(&value).expect("SSE JSON") |
| 250 | ) |
| 251 | } |
| 252 | |
| 253 | fn run_exec(workspace: &Path, home: &Path, base_url: &str) -> std::process::Output { |
| 254 | std::fs::create_dir_all(home.join(".codewhale")).expect("config directory"); |
| 255 | std::fs::create_dir_all(home.join(".deepseek")).expect("legacy config directory"); |
| 256 | std::fs::write( |
| 257 | home.join(".codewhale/config.toml"), |
| 258 | "allow_shell = true\n\n[retry]\nenabled = false\n", |
| 259 | ) |
| 260 | .expect("headless test config"); |
| 261 | let mut command = Command::new(binary()); |
| 262 | preserve_host_env(&mut command); |
| 263 | command |
| 264 | .current_dir(workspace) |
| 265 | .args(["--workspace", workspace.to_str().expect("workspace utf8")]) |
| 266 | .arg("--no-project-config") |
| 267 | .args([ |
| 268 | "exec", |
| 269 | "--auto", |
| 270 | "--model", |
| 271 | MODEL, |
| 272 | "--output-format", |
| 273 | "stream-json", |
| 274 | ]) |
| 275 | .arg("answer briefly") |
| 276 | .env("HOME", home) |
| 277 | .env("USERPROFILE", home) |
| 278 | .env("XDG_CONFIG_HOME", home.join(".config")) |
| 279 | .env("XDG_DATA_HOME", home.join(".local/share")) |
| 280 | .env("XDG_CACHE_HOME", home.join(".cache")) |
| 281 | .env("CODEWHALE_CONFIG_PATH", home.join(".codewhale/config.toml")) |
| 282 | .env("DEEPSEEK_CONFIG_PATH", home.join(".deepseek/config.toml")) |
| 283 | .env("DEEPSEEK_API_KEY", "ci-test-key-not-real") |
| 284 | .env("DEEPSEEK_BASE_URL", base_url) |
| 285 | .env("CODEWHALE_BASE_URL", base_url) |
| 286 | .env("DEEPSEEK_MODEL", MODEL) |
| 287 | .env("CODEWHALE_MODEL", MODEL) |
| 288 | .env("RUST_LOG", "warn") |
| 289 | .stdout(Stdio::piped()) |
| 290 | .stderr(Stdio::piped()); |
| 291 | run_with_timeout(command, Duration::from_secs(45)) |
| 292 | } |
| 293 | |
| 294 | fn binary() -> PathBuf { |
| 295 | std::env::var_os("CARGO_BIN_EXE_codewhale-tui") |
| 296 | .map(PathBuf::from) |
| 297 | .unwrap_or_else(|| { |
| 298 | PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../target/debug/codewhale-tui") |
| 299 | }) |
| 300 | } |
| 301 | |
| 302 | fn preserve_host_env(command: &mut Command) { |
| 303 | command.env_clear(); |
| 304 | for key in [ |
| 305 | "PATH", |
| 306 | "PATHEXT", |
| 307 | "SystemRoot", |
| 308 | "SystemDrive", |
| 309 | "WINDIR", |
| 310 | "COMSPEC", |
| 311 | "TEMP", |
| 312 | "TMP", |
| 313 | "TERM", |
| 314 | "LANG", |
| 315 | "LC_ALL", |
| 316 | ] { |
| 317 | if let Some(value) = std::env::var_os(key) { |
| 318 | command.env(key, value); |
| 319 | } |
| 320 | } |
| 321 | } |
| 322 | |
| 323 | fn run_with_timeout(mut command: Command, timeout: Duration) -> std::process::Output { |
| 324 | let mut child = command.spawn().expect("spawn codewhale exec"); |
| 325 | let stdout = read_in_background(child.stdout.take().expect("stdout")); |
| 326 | let stderr = read_in_background(child.stderr.take().expect("stderr")); |
| 327 | let status = child |
| 328 | .wait_timeout(timeout) |
| 329 | .expect("wait") |
| 330 | .unwrap_or_else(|| { |
| 331 | let _ = child.kill(); |
| 332 | let _ = child.wait(); |
| 333 | panic!("codewhale exec timed out") |
| 334 | }); |
| 335 | std::process::Output { |
| 336 | status, |
| 337 | stdout: stdout.join().expect("stdout thread").expect("read stdout"), |
| 338 | stderr: stderr.join().expect("stderr thread").expect("read stderr"), |
| 339 | } |
| 340 | } |
| 341 | |
| 342 | fn read_in_background<R: Read + Send + 'static>( |
| 343 | mut reader: R, |
| 344 | ) -> std::thread::JoinHandle<std::io::Result<Vec<u8>>> { |
| 345 | std::thread::spawn(move || { |
| 346 | let mut bytes = Vec::new(); |
| 347 | reader.read_to_end(&mut bytes).map(|_| bytes) |
| 348 | }) |
| 349 | } |
| 350 |