| 1 | //! Real Engine/process acceptance; only an in-process fake provider is used. |
| 2 | use std::io::Read; |
| 3 | use std::path::{Path, PathBuf}; |
| 4 | use std::process::{Command, Stdio}; |
| 5 | use std::sync::atomic::{AtomicUsize, Ordering}; |
| 6 | use std::sync::{Arc, Mutex}; |
| 7 | use std::time::Duration; |
| 8 | |
| 9 | use serde_json::{Value, json}; |
| 10 | use sha2::{Digest, Sha256}; |
| 11 | use tempfile::TempDir; |
| 12 | use wait_timeout::ChildExt; |
| 13 | use wiremock::matchers::{method, path}; |
| 14 | use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate}; |
| 15 | |
| 16 | const MODEL: &str = "issue-draft-fixture-model"; |
| 17 | const ORIGINAL: &str = "ORIGINAL_TASK_CONTEXT_ISSUE_REPORT"; |
| 18 | |
| 19 | #[tokio::test(flavor = "multi_thread", worker_threads = 2)] |
| 20 | async fn active_agent_drafts_converge_and_resume_in_the_same_session() { |
| 21 | let workspace = TempDir::new().unwrap(); |
| 22 | let home = TempDir::new().unwrap(); |
| 23 | let server = MockServer::start().await; |
| 24 | let count = Arc::new(AtomicUsize::new(0)); |
| 25 | let handle = Arc::new(Mutex::new(None)); |
| 26 | Mock::given(method("POST")) |
| 27 | .and(path("/v1/chat/completions")) |
| 28 | .respond_with(Scenario { |
| 29 | count: count.clone(), |
| 30 | handle: handle.clone(), |
| 31 | }) |
| 32 | .mount(&server) |
| 33 | .await; |
| 34 | let output = run_exec(workspace.path(), home.path(), &server, None); |
| 35 | assert_success(&output); |
| 36 | assert_eq!(count.load(Ordering::SeqCst), 4); |
| 37 | let (session, directory) = report_directory(home.path()).expect("saved report directory"); |
| 38 | let files = std::fs::read_dir(&directory) |
| 39 | .unwrap() |
| 40 | .collect::<Result<Vec<_>, _>>() |
| 41 | .unwrap(); |
| 42 | assert_eq!(files.len(), 1, "identical model reports must converge"); |
| 43 | let before = std::fs::read(files[0].path()).unwrap(); |
| 44 | let saved: Value = serde_json::from_slice(&before).unwrap(); |
| 45 | assert_eq!(saved["model"], MODEL); |
| 46 | assert_eq!(saved["session"], session); |
| 47 | assert_eq!( |
| 48 | saved["id"], |
| 49 | handle.lock().unwrap().as_ref().unwrap().as_str() |
| 50 | ); |
| 51 | let text = String::from_utf8(before.clone()).unwrap(); |
| 52 | for private in [ |
| 53 | "fixture-private-token", |
| 54 | "private-machine-owner", |
| 55 | "fixture-url-password", |
| 56 | ] { |
| 57 | assert!(!text.contains(private)); |
| 58 | } |
| 59 | // A new process resumes the real saved session and asks report_read for |
| 60 | // the opaque handle from the earlier model-visible result. |
| 61 | let resumed = run_exec(workspace.path(), home.path(), &server, Some(&session)); |
| 62 | assert_success(&resumed); |
| 63 | assert_eq!(count.load(Ordering::SeqCst), 6); |
| 64 | assert_eq!(std::fs::read(files[0].path()).unwrap(), before); |
| 65 | assert_eq!(std::fs::read_dir(&directory).unwrap().count(), 1); |
| 66 | assert!(!home.path().join("unexpected-gh-call").exists()); |
| 67 | let requests = server.received_requests().await.unwrap(); |
| 68 | for request in requests { |
| 69 | assert_eq!(request.url.path(), "/v1/chat/completions"); |
| 70 | assert_eq!(request.body_json::<Value>().unwrap()["model"], MODEL); |
| 71 | } |
| 72 | } |
| 73 | |
| 74 | #[tokio::test(flavor = "multi_thread", worker_threads = 2)] |
| 75 | async fn failed_provider_leaves_no_generated_draft_or_fallback_success() { |
| 76 | let workspace = TempDir::new().unwrap(); |
| 77 | let home = TempDir::new().unwrap(); |
| 78 | let server = MockServer::start().await; |
| 79 | Mock::given(method("POST")).and(path("/v1/chat/completions")) |
| 80 | .respond_with(ResponseTemplate::new(401).set_body_json(json!({"error":{"message":"fixture provider unavailable", "type":"authentication_error"}}))) |
| 81 | .mount(&server).await; |
| 82 | let output = run_exec(workspace.path(), home.path(), &server, None); |
| 83 | assert!(!output.status.success()); |
| 84 | assert!(report_directory(home.path()).is_none()); |
| 85 | let stdout = String::from_utf8_lossy(&output.stdout); |
| 86 | assert!(!stdout.contains("DRAFT_SAVED_AND_ORIGINAL_TASK_CONTINUED")); |
| 87 | assert!(!stdout.contains("ready_for_review")); |
| 88 | assert!(!home.path().join("unexpected-gh-call").exists()); |
| 89 | let requests = server.received_requests().await.unwrap(); |
| 90 | assert!(!requests.is_empty()); |
| 91 | for request in requests { |
| 92 | assert_eq!(request.url.path(), "/v1/chat/completions"); |
| 93 | assert_eq!(request.body_json::<Value>().unwrap()["model"], MODEL); |
| 94 | } |
| 95 | } |
| 96 | |
| 97 | #[derive(Clone)] |
| 98 | struct Scenario { |
| 99 | count: Arc<AtomicUsize>, |
| 100 | handle: Arc<Mutex<Option<String>>>, |
| 101 | } |
| 102 | impl Respond for Scenario { |
| 103 | fn respond(&self, request: &Request) -> ResponseTemplate { |
| 104 | let sequence = self.count.fetch_add(1, Ordering::SeqCst); |
| 105 | let body: Value = request.body_json().unwrap(); |
| 106 | assert_eq!(body["model"], MODEL); |
| 107 | assert!( |
| 108 | body["messages"].to_string().contains(ORIGINAL), |
| 109 | "same task context must reach every request" |
| 110 | ); |
| 111 | let response = match sequence { |
| 112 | 0 => draft_call("draft-first"), |
| 113 | 1 => { |
| 114 | let first = receipt(&body, "draft-first"); |
| 115 | assert_eq!(first["publication"], "unavailable"); |
| 116 | assert_eq!(first["duplicate_search"], "not_performed"); |
| 117 | assert!(first["review"].as_str().unwrap().contains(MODEL)); |
| 118 | for private in [ |
| 119 | "fixture-private-token", |
| 120 | "private-machine-owner", |
| 121 | "fixture-url-password", |
| 122 | ] { |
| 123 | assert!(!first.to_string().contains(private)); |
| 124 | } |
| 125 | *self.handle.lock().unwrap() = Some(first["report_id"].as_str().unwrap().into()); |
| 126 | draft_call("draft-repeat") |
| 127 | } |
| 128 | 2 => { |
| 129 | let repeated = receipt(&body, "draft-repeat"); |
| 130 | assert_eq!( |
| 131 | repeated["report_id"], |
| 132 | self.handle.lock().unwrap().as_ref().unwrap().as_str() |
| 133 | ); |
| 134 | tool_sse( |
| 135 | "draft-read", |
| 136 | json!({"action":"report_read", "report_id":repeated["report_id"]}), |
| 137 | ) |
| 138 | } |
| 139 | 3 => { |
| 140 | assert_eq!(receipt(&body, "draft-read")["state"], "ready_for_review"); |
| 141 | final_sse() |
| 142 | } |
| 143 | 4 => { |
| 144 | assert!( |
| 145 | body["messages"] |
| 146 | .to_string() |
| 147 | .contains("RESUME_EXISTING_DRAFT") |
| 148 | ); |
| 149 | tool_sse( |
| 150 | "draft-resume-read", |
| 151 | json!({"action":"report_read", "report_id":self.handle.lock().unwrap().as_ref().unwrap()}), |
| 152 | ) |
| 153 | } |
| 154 | 5 => { |
| 155 | assert_eq!( |
| 156 | receipt(&body, "draft-resume-read")["report_id"], |
| 157 | self.handle.lock().unwrap().as_ref().unwrap().as_str() |
| 158 | ); |
| 159 | final_sse() |
| 160 | } |
| 161 | _ => panic!("unexpected extra provider request"), |
| 162 | }; |
| 163 | ResponseTemplate::new(200) |
| 164 | .insert_header("content-type", "text/event-stream") |
| 165 | .set_body_string(response) |
| 166 | } |
| 167 | } |
| 168 | |
| 169 | fn draft_call(id: &str) -> String { |
| 170 | tool_sse( |
| 171 | id, |
| 172 | json!({"action":"report_draft", "report":{ |
| 173 | "title":"Runtime result delivery failed", "expected":"The agent receives the completed tool result", |
| 174 | "actual":"Result missing. Authorization:\nBearer\tfixture-private-token /Users/private-machine-owner/workspace https://user:fixture-url-password@example.invalid/private", |
| 175 | "impact":"Original task needs a retry", "steps":["Run a tool", "Wait for its result"], |
| 176 | "observed":["The Runtime result was absent"], "inferred":["The Runtime may have dropped an event"], |
| 177 | "reported_tool":"fixture tool", "reported_provider":"fixture provider" |
| 178 | }}), |
| 179 | ) |
| 180 | } |
| 181 | |
| 182 | fn receipt(body: &Value, id: &str) -> Value { |
| 183 | let messages = body["messages"].as_array().unwrap(); |
| 184 | let content = messages |
| 185 | .iter() |
| 186 | .find(|message| message["role"] == "tool" && message["tool_call_id"] == id) |
| 187 | .unwrap_or_else(|| panic!("missing receipt {id}"))["content"] |
| 188 | .as_str() |
| 189 | .unwrap(); |
| 190 | // The provider adapter replaces exact duplicate results with a reference |
| 191 | // to earlier full content in this same request. Follow its verified digest. |
| 192 | let content = if let Some(reference) = content.strip_prefix("<TOOL_RESULT_REF sha=\"") { |
| 193 | let digest = reference.split('"').next().unwrap(); |
| 194 | messages |
| 195 | .iter() |
| 196 | .filter_map(|message| message["content"].as_str()) |
| 197 | .find(|candidate| { |
| 198 | Sha256::digest(candidate.as_bytes()) |
| 199 | .iter() |
| 200 | .map(|byte| format!("{byte:02x}")) |
| 201 | .collect::<String>() |
| 202 | == digest |
| 203 | }) |
| 204 | .expect("duplicate receipt must reference full content in this request") |
| 205 | } else { |
| 206 | content |
| 207 | }; |
| 208 | serde_json::from_str(content).unwrap_or_else(|_| panic!("invalid draft receipt: {content}")) |
| 209 | } |
| 210 | |
| 211 | fn chunk(value: Value) -> String { |
| 212 | format!("data: {value}\n\n") |
| 213 | } |
| 214 | fn tool_sse(id: &str, args: Value) -> String { |
| 215 | format!( |
| 216 | "{}{}data: [DONE]\n\n", |
| 217 | chunk( |
| 218 | json!({"id":"fixture", "object":"chat.completion.chunk", "model":MODEL, "choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":id,"type":"function","function":{"name":"github","arguments":args.to_string()}}]},"finish_reason":null}]}) |
| 219 | ), |
| 220 | chunk( |
| 221 | json!({"id":"fixture", "model":MODEL, "choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":10,"completion_tokens":2,"total_tokens":12}}) |
| 222 | ) |
| 223 | ) |
| 224 | } |
| 225 | fn final_sse() -> String { |
| 226 | format!( |
| 227 | "{}{}data: [DONE]\n\n", |
| 228 | chunk( |
| 229 | json!({"id":"fixture", "model":MODEL, "choices":[{"index":0,"delta":{"content":"DRAFT_SAVED_AND_ORIGINAL_TASK_CONTINUED"},"finish_reason":null}]}) |
| 230 | ), |
| 231 | chunk( |
| 232 | json!({"id":"fixture", "model":MODEL, "choices":[{"index":0,"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":10,"completion_tokens":2,"total_tokens":12}}) |
| 233 | ) |
| 234 | ) |
| 235 | } |
| 236 | |
| 237 | fn report_directory(home: &Path) -> Option<(String, PathBuf)> { |
| 238 | std::fs::read_dir(home.join(".codewhale/sessions")) |
| 239 | .ok()? |
| 240 | .filter_map(Result::ok) |
| 241 | .find_map(|entry| { |
| 242 | let path = entry.path().join("artifacts/issue-reports"); |
| 243 | path.is_dir() |
| 244 | .then(|| (entry.file_name().to_string_lossy().into_owned(), path)) |
| 245 | }) |
| 246 | } |
| 247 | |
| 248 | fn run_exec( |
| 249 | workspace: &Path, |
| 250 | home: &Path, |
| 251 | server: &MockServer, |
| 252 | resume: Option<&str>, |
| 253 | ) -> std::process::Output { |
| 254 | std::fs::create_dir_all(home.join(".codewhale")).unwrap(); |
| 255 | std::fs::write( |
| 256 | home.join(".codewhale/config.toml"), |
| 257 | "allow_shell = false\ntelemetry = false\n\n[retry]\nenabled = false\n", |
| 258 | ) |
| 259 | .unwrap(); |
| 260 | let mut command = Command::new(env!("CARGO_BIN_EXE_codewhale-tui")); |
| 261 | command.env_clear(); |
| 262 | for key in [ |
| 263 | "PATH", |
| 264 | "PATHEXT", |
| 265 | "SystemRoot", |
| 266 | "SystemDrive", |
| 267 | "WINDIR", |
| 268 | "COMSPEC", |
| 269 | "TEMP", |
| 270 | "TMP", |
| 271 | "LANG", |
| 272 | ] { |
| 273 | if let Some(value) = std::env::var_os(key) { |
| 274 | command.env(key, value); |
| 275 | } |
| 276 | } |
| 277 | command |
| 278 | .current_dir(workspace) |
| 279 | .args(["--workspace"]) |
| 280 | .arg(workspace) |
| 281 | .args([ |
| 282 | "--no-project-config", |
| 283 | "exec", |
| 284 | "--auto", |
| 285 | "--provider", |
| 286 | "deepseek", |
| 287 | "--model", |
| 288 | MODEL, |
| 289 | "--allowed-tools", |
| 290 | "github", |
| 291 | "--output-format", |
| 292 | "stream-json", |
| 293 | ]) |
| 294 | .env("HOME", home) |
| 295 | .env("USERPROFILE", home) |
| 296 | .env("CODEWHALE_HOME", home.join(".codewhale")) |
| 297 | .env("CODEWHALE_CONFIG_PATH", home.join(".codewhale/config.toml")) |
| 298 | .env("DEEPSEEK_API_KEY", "fixture-key-not-real") |
| 299 | .env("DEEPSEEK_BASE_URL", server.uri()) |
| 300 | .env("CODEWHALE_BASE_URL", server.uri()) |
| 301 | .env("DEEPSEEK_MODEL", MODEL) |
| 302 | .env("CODEWHALE_MODEL", MODEL) |
| 303 | .env("CODEWHALE_TELEMETRY", "0") |
| 304 | // Any accidental use of the existing gh adapter must fail. No real gh |
| 305 | // credentials or binary is reachable through this test override. |
| 306 | .env("CODEWHALE_GH_BIN", home.join("missing-gh-fixture")) |
| 307 | .env("RUST_LOG", "warn") |
| 308 | .stdout(Stdio::piped()) |
| 309 | .stderr(Stdio::piped()); |
| 310 | if let Some(id) = resume { |
| 311 | command.args(["--resume", id]); |
| 312 | } |
| 313 | command.arg(if resume.is_some() { |
| 314 | "RESUME_EXISTING_DRAFT: read the saved draft from our earlier work and continue that task.".to_string() |
| 315 | } else { |
| 316 | format!("{ORIGINAL}: observed a Runtime result-delivery failure. Draft it locally and continue the original task.") |
| 317 | }); |
| 318 | let mut child = command.spawn().unwrap(); |
| 319 | let stdout = drain(child.stdout.take().unwrap()); |
| 320 | let stderr = drain(child.stderr.take().unwrap()); |
| 321 | let status = child |
| 322 | .wait_timeout(Duration::from_secs(60)) |
| 323 | .unwrap() |
| 324 | .unwrap_or_else(|| { |
| 325 | child.kill().ok(); |
| 326 | child.wait().ok(); |
| 327 | panic!("issue fixture timed out"); |
| 328 | }); |
| 329 | std::process::Output { |
| 330 | status, |
| 331 | stdout: stdout.join().unwrap(), |
| 332 | stderr: stderr.join().unwrap(), |
| 333 | } |
| 334 | } |
| 335 | fn drain(mut stream: impl Read + Send + 'static) -> std::thread::JoinHandle<Vec<u8>> { |
| 336 | std::thread::spawn(move || { |
| 337 | let mut bytes = Vec::new(); |
| 338 | stream.read_to_end(&mut bytes).unwrap(); |
| 339 | bytes |
| 340 | }) |
| 341 | } |
| 342 | fn assert_success(output: &std::process::Output) { |
| 343 | assert!( |
| 344 | output.status.success(), |
| 345 | "fixture failed: {}\n{}", |
| 346 | String::from_utf8_lossy(&output.stdout), |
| 347 | String::from_utf8_lossy(&output.stderr) |
| 348 | ); |
| 349 | assert!( |
| 350 | String::from_utf8_lossy(&output.stdout).contains("DRAFT_SAVED_AND_ORIGINAL_TASK_CONTINUED") |
| 351 | ); |
| 352 | } |
| 353 |