返回 CodeWhale
exec_persistent_service.rs
根目录 / crates / tui / tests / integration / exec_persistent_service.rs
1 //! Real-process acceptance for `persist:true` background services on the
2 //! headless `codewhale exec` host.
3 //!
4 //! Three black-box contracts against the actual binary and real child
5 //! processes, with a `wiremock` OpenAI-compatible provider:
6 //!
7 //! - a successful exec releases the explicitly persisted service: the exec
8 //! process exits 0, emits a `service_released` receipt, and the service
9 //! process is still alive afterwards;
10 //! - a failed exec (incomplete non-limit stop) kills the pending service and
11 //! exits nonzero;
12 //! - a terminating signal mid-turn kills the pending service and exits
13 //! nonzero.
14
15 #![cfg(unix)]
16
17 use std::io::Read;
18 use std::path::{Path, PathBuf};
19 use std::process::{Command, Stdio};
20 use std::sync::atomic::{AtomicUsize, Ordering};
21 use std::sync::{Arc, Mutex, OnceLock, mpsc};
22 use std::time::{Duration, Instant};
23
24 async fn serialize_persistent_service_tests() -> tokio::sync::MutexGuard<'static, ()> {
25 static LOCK: OnceLock<tokio::sync::Mutex<()>> = OnceLock::new();
26 LOCK.get_or_init(|| tokio::sync::Mutex::new(()))
27 .lock()
28 .await
29 }
30
31 use serde_json::{Value, json};
32 use tempfile::TempDir;
33 use wait_timeout::ChildExt;
34 use wiremock::matchers::{method, path};
35 use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};
36
37 const TEST_MODEL: &str = "persist-service-model";
38 const RUN_TIMEOUT: Duration = Duration::from_secs(120);
39
40 fn sse_chunk(value: Value) -> String {
41 format!(
42 "data: {}\n\n",
43 serde_json::to_string(&value).expect("SSE JSON")
44 )
45 }
46
47 /// First model turn: one Bash tool call staging the persistent service.
48 fn stage_service_sse(command: &str) -> String {
49 let arguments = serde_json::to_string(&json!({
50 "command": command,
51 "background": true,
52 "persist": true,
53 }))
54 .expect("tool arguments JSON");
55 [
56 sse_chunk(json!({
57 "id": "chatcmpl-stage",
58 "object": "chat.completion.chunk",
59 "model": TEST_MODEL,
60 "choices": [{"index": 0, "delta": {"tool_calls": [{"index": 0, "id": "call_persist", "type": "function", "function": {"name": "Bash", "arguments": arguments}}]}, "finish_reason": null}]
61 })),
62 sse_chunk(json!({
63 "id": "chatcmpl-stage",
64 "object": "chat.completion.chunk",
65 "model": TEST_MODEL,
66 "choices": [{"index": 0, "delta": {}, "finish_reason": "tool_calls"}],
67 "usage": {"prompt_tokens": 12, "completion_tokens": 4, "total_tokens": 16}
68 })),
69 "data: [DONE]\n\n".to_string(),
70 ]
71 .join("")
72 }
73
74 /// Second model turn: an ordinary completed final answer.
75 fn final_answer_sse() -> String {
76 [
77 sse_chunk(json!({
78 "id": "chatcmpl-final",
79 "object": "chat.completion.chunk",
80 "model": TEST_MODEL,
81 "choices": [{"index": 0, "delta": {"content": "service is up"}, "finish_reason": null}]
82 })),
83 sse_chunk(json!({
84 "id": "chatcmpl-final",
85 "object": "chat.completion.chunk",
86 "model": TEST_MODEL,
87 "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
88 "usage": {"prompt_tokens": 30, "completion_tokens": 3, "total_tokens": 33}
89 })),
90 "data: [DONE]\n\n".to_string(),
91 ]
92 .join("")
93 }
94
95 /// Second model turn: incomplete non-limit stop. `length` now degrades and
96 /// continues the headless loop (no default max-turns), so this fixture uses
97 /// `content_filter` to force a failed exec and prove pending services die.
98 fn incomplete_answer_sse() -> String {
99 [
100 sse_chunk(json!({
101 "id": "chatcmpl-incomplete",
102 "object": "chat.completion.chunk",
103 "model": TEST_MODEL,
104 "choices": [{"index": 0, "delta": {"content": "partial"}, "finish_reason": null}]
105 })),
106 sse_chunk(json!({
107 "id": "chatcmpl-incomplete",
108 "object": "chat.completion.chunk",
109 "model": TEST_MODEL,
110 "choices": [{"index": 0, "delta": {}, "finish_reason": "content_filter"}],
111 "usage": {"prompt_tokens": 30, "completion_tokens": 2, "total_tokens": 32}
112 })),
113 "data: [DONE]\n\n".to_string(),
114 ]
115 .join("")
116 }
117
118 fn sse_response(body: String) -> ResponseTemplate {
119 ResponseTemplate::new(200)
120 .insert_header("content-type", "text/event-stream")
121 .insert_header("cache-control", "no-cache")
122 .set_body_string(body)
123 }
124
125 fn json_response(value: Value) -> ResponseTemplate {
126 ResponseTemplate::new(200)
127 .insert_header("content-type", "application/json")
128 .set_body_json(value)
129 }
130
131 /// Sequential provider: first POST stages the service; later POSTs get the
132 /// scenario's second turn. The failure scenario waits for explicit service
133 /// readiness; an optional delay holds the exec mid-turn for the signal case.
134 struct SequentialTurns {
135 requests: Arc<AtomicUsize>,
136 stage_command: String,
137 second_turn: String,
138 second_turn_delay: Option<Duration>,
139 second_turn_ready: Option<Mutex<mpsc::Receiver<()>>>,
140 }
141
142 impl Respond for SequentialTurns {
143 fn respond(&self, _request: &Request) -> ResponseTemplate {
144 let call = self.requests.fetch_add(1, Ordering::SeqCst);
145 if call == 0 {
146 sse_response(stage_service_sse(&self.stage_command))
147 } else {
148 if call == 1
149 && let Some(ready) = &self.second_turn_ready
150 && ready
151 .lock()
152 .expect("service readiness receiver")
153 .recv_timeout(RUN_TIMEOUT)
154 .is_err()
155 {
156 return ResponseTemplate::new(504)
157 .set_body_string("test did not confirm service readiness");
158 }
159 let response = sse_response(self.second_turn.clone());
160 match self.second_turn_delay {
161 Some(delay) => response.set_delay(delay),
162 None => response,
163 }
164 }
165 }
166 }
167
168 async fn start_mock_llm(
169 stage_command: &str,
170 second_turn: String,
171 second_turn_delay: Option<Duration>,
172 second_turn_ready: Option<mpsc::Receiver<()>>,
173 ) -> MockServer {
174 let server = MockServer::start().await;
175
176 Mock::given(method("GET"))
177 .and(path("/v1/models"))
178 .respond_with(json_response(json!({
179 "object": "list",
180 "data": [{ "id": TEST_MODEL, "object": "model" }]
181 })))
182 .mount(&server)
183 .await;
184
185 Mock::given(method("POST"))
186 .and(path("/v1/chat/completions"))
187 .respond_with(SequentialTurns {
188 requests: Arc::new(AtomicUsize::new(0)),
189 stage_command: stage_command.to_string(),
190 second_turn,
191 second_turn_delay,
192 second_turn_ready: second_turn_ready.map(Mutex::new),
193 })
194 .mount(&server)
195 .await;
196
197 server
198 }
199
200 fn preserve_host_env(command: &mut Command) {
201 command.env_clear();
202 for key in [
203 "PATH",
204 "SHELL",
205 "TEMP",
206 "TMP",
207 "TERM",
208 "COLORTERM",
209 "LANG",
210 "LC_ALL",
211 ] {
212 if let Some(value) = std::env::var_os(key) {
213 command.env(key, value);
214 }
215 }
216 }
217
218 fn exec_command(server: &MockServer, workspace: &Path, home: &Path) -> Command {
219 let mut command = Command::new(codewhale_tui_binary());
220 preserve_host_env(&mut command);
221 command
222 .current_dir(workspace)
223 .arg("--workspace")
224 .arg(workspace)
225 .arg("--no-project-config")
226 .arg("exec")
227 .arg("--auto")
228 .arg("--sandbox")
229 .arg("danger-full-access")
230 .arg("--model")
231 .arg(TEST_MODEL)
232 .arg("--output-format")
233 .arg("stream-json")
234 .arg("start the service, then confirm")
235 .env("HOME", home)
236 .env("USERPROFILE", home)
237 .env("XDG_CONFIG_HOME", home.join(".config"))
238 .env("XDG_DATA_HOME", home.join(".local").join("share"))
239 .env("XDG_CACHE_HOME", home.join(".cache"))
240 .env(
241 "CODEWHALE_CONFIG_PATH",
242 home.join(".codewhale").join("config.toml"),
243 )
244 .env(
245 "DEEPSEEK_CONFIG_PATH",
246 home.join(".deepseek").join("config.toml"),
247 )
248 .env("DEEPSEEK_API_KEY", "ci-test-key-not-real")
249 .env("DEEPSEEK_BASE_URL", server.uri())
250 .env("CODEWHALE_BASE_URL", server.uri())
251 .env("DEEPSEEK_MODEL", TEST_MODEL)
252 .env("CODEWHALE_MODEL", TEST_MODEL)
253 .env("RUST_LOG", "warn")
254 .stdout(Stdio::piped())
255 .stderr(Stdio::piped());
256 std::fs::create_dir_all(home.join(".codewhale")).expect("create codewhale config dir");
257 std::fs::create_dir_all(home.join(".deepseek")).expect("create deepseek config dir");
258 command
259 }
260
261 fn read_pipe_in_background<R>(mut reader: R) -> std::thread::JoinHandle<std::io::Result<Vec<u8>>>
262 where
263 R: Read + Send + 'static,
264 {
265 std::thread::spawn(move || {
266 let mut bytes = Vec::new();
267 reader.read_to_end(&mut bytes)?;
268 Ok(bytes)
269 })
270 }
271
272 fn join_pipe(handle: std::thread::JoinHandle<std::io::Result<Vec<u8>>>, label: &str) -> String {
273 let bytes = handle
274 .join()
275 .unwrap_or_else(|_| panic!("{label} reader thread panicked"))
276 .unwrap_or_else(|error| panic!("{label} read failed: {error}"));
277 String::from_utf8_lossy(&bytes).into_owned()
278 }
279
280 fn codewhale_tui_binary() -> PathBuf {
281 if let Some(path) = option_env!("CARGO_BIN_EXE_codewhale-tui") {
282 return PathBuf::from(path);
283 }
284 if let Ok(path) = std::env::var("CARGO_BIN_EXE_codewhale-tui") {
285 return PathBuf::from(path);
286 }
287 let mut path = std::env::current_exe().expect("current test executable path");
288 path.pop();
289 if path.ends_with("deps") {
290 path.pop();
291 }
292 path.push(format!("codewhale-tui{}", std::env::consts::EXE_SUFFIX));
293 path
294 }
295
296 fn stream_events(stdout: &str) -> Vec<Value> {
297 stdout
298 .lines()
299 .filter_map(|line| serde_json::from_str::<Value>(line.trim()).ok())
300 .collect()
301 }
302
303 fn pid_is_alive(pid: i32) -> bool {
304 // SAFETY: signal 0 performs only an existence/permission check.
305 unsafe { libc::kill(pid, 0) == 0 }
306 }
307
308 fn kill_process_group(pid: i32) {
309 // SAFETY: the pid came from this test's own staged service; the negative
310 // pid targets its process group only.
311 unsafe {
312 libc::kill(-pid, libc::SIGKILL);
313 }
314 }
315
316 fn wait_for_pid_file(path: &Path) -> Result<i32, String> {
317 let deadline = Instant::now() + RUN_TIMEOUT;
318 loop {
319 if let Ok(contents) = std::fs::read_to_string(path)
320 && let Ok(pid) = contents.trim().parse::<i32>()
321 {
322 return Ok(pid);
323 }
324 if Instant::now() >= deadline {
325 return Err(format!(
326 "service pid file never appeared at {}",
327 path.display()
328 ));
329 }
330 std::thread::sleep(Duration::from_millis(50));
331 }
332 }
333
334 fn wait_for_pid_death(pid: i32) {
335 let deadline = Instant::now() + Duration::from_secs(15);
336 while pid_is_alive(pid) {
337 assert!(
338 Instant::now() < deadline,
339 "pending persistent service (pid {pid}) must be killed"
340 );
341 std::thread::sleep(Duration::from_millis(50));
342 }
343 }
344
345 /// The staged service records its own pid, then stays alive.
346 const SERVICE_COMMAND: &str = "echo $$ > service.pid; exec sleep 600";
347
348 #[tokio::test(flavor = "multi_thread")]
349 async fn successful_exec_releases_persisted_service() {
350 let _serial = serialize_persistent_service_tests().await;
351 let server = start_mock_llm(SERVICE_COMMAND, final_answer_sse(), None, None).await;
352 let workspace = TempDir::new().expect("workspace tempdir");
353 let home = TempDir::new().expect("home tempdir");
354
355 let mut child = exec_command(&server, workspace.path(), home.path())
356 .spawn()
357 .expect("spawn codewhale-tui exec");
358 let stdout_reader = read_pipe_in_background(child.stdout.take().expect("stdout pipe"));
359 let stderr_reader = read_pipe_in_background(child.stderr.take().expect("stderr pipe"));
360 let status = child
361 .wait_timeout(RUN_TIMEOUT)
362 .expect("wait for exec")
363 .unwrap_or_else(|| {
364 let _ = child.kill();
365 let _ = child.wait();
366 panic!("exec timed out");
367 });
368 let stdout = join_pipe(stdout_reader, "stdout");
369 let stderr = join_pipe(stderr_reader, "stderr");
370
371 let service_pid = wait_for_pid_file(&workspace.path().join("service.pid"))
372 .unwrap_or_else(|error| panic!("{error}\nstdout:\n{stdout}\nstderr:\n{stderr}"));
373 let events = stream_events(&stdout);
374 let released = events
375 .iter()
376 .find(|event| event.get("type").and_then(Value::as_str) == Some("service_released"))
377 .unwrap_or_else(|| {
378 panic!("missing service_released event\nstdout:\n{stdout}\nstderr:\n{stderr}")
379 });
380
381 assert!(
382 status.success(),
383 "successful exec must exit 0 (got {status:?})\nstderr:\n{stderr}"
384 );
385 assert_eq!(
386 released.get("pid").and_then(Value::as_u64),
387 Some(u64::try_from(service_pid).expect("pid fits u64")),
388 "release receipt must carry the real service pid"
389 );
390 assert_eq!(
391 released.get("ownership").and_then(Value::as_str),
392 Some("external")
393 );
394 assert!(
395 pid_is_alive(service_pid),
396 "explicitly persisted service must survive successful headless exit"
397 );
398
399 kill_process_group(service_pid);
400 }
401
402 #[tokio::test(flavor = "multi_thread")]
403 async fn failed_exec_kills_pending_service_and_exits_nonzero() {
404 let _serial = serialize_persistent_service_tests().await;
405 let (service_ready, service_ready_receiver) = mpsc::channel();
406 let server = start_mock_llm(
407 SERVICE_COMMAND,
408 incomplete_answer_sse(),
409 None,
410 Some(service_ready_receiver),
411 )
412 .await;
413 let workspace = TempDir::new().expect("workspace tempdir");
414 let home = TempDir::new().expect("home tempdir");
415
416 let mut child = exec_command(&server, workspace.path(), home.path())
417 .spawn()
418 .expect("spawn codewhale-tui exec");
419 let stdout_reader = read_pipe_in_background(child.stdout.take().expect("stdout pipe"));
420 let stderr_reader = read_pipe_in_background(child.stderr.take().expect("stderr pipe"));
421
422 // Spawning a background process does not mean its first instruction ran.
423 // Hold the deliberate model failure until the real service is ready, so
424 // cancellation cannot kill it before it writes the PID we need to check.
425 let service_pid = match wait_for_pid_file(&workspace.path().join("service.pid")) {
426 Ok(pid) => pid,
427 Err(error) => {
428 drop(service_ready);
429 // Let the host clean up its managed services before forcing exit.
430 // SAFETY: direct child of this test.
431 unsafe {
432 libc::kill(
433 i32::try_from(child.id()).expect("child pid fits i32"),
434 libc::SIGTERM,
435 );
436 }
437 if child
438 .wait_timeout(Duration::from_secs(5))
439 .ok()
440 .flatten()
441 .is_none()
442 {
443 let _ = child.kill();
444 let _ = child.wait();
445 }
446 let stdout = join_pipe(stdout_reader, "stdout");
447 let stderr = join_pipe(stderr_reader, "stderr");
448 panic!("{error}\nstdout:\n{stdout}\nstderr:\n{stderr}");
449 }
450 };
451 assert!(
452 pid_is_alive(service_pid),
453 "service must be alive before the model fails"
454 );
455 service_ready
456 .send(())
457 .expect("release the deliberate model failure");
458
459 let status = match child.wait_timeout(RUN_TIMEOUT).expect("wait for exec") {
460 Some(status) => status,
461 None => {
462 let _ = child.kill();
463 let _ = child.wait();
464 let stdout = join_pipe(stdout_reader, "stdout");
465 let stderr = join_pipe(stderr_reader, "stderr");
466 panic!("exec timed out\nstdout:\n{stdout}\nstderr:\n{stderr}");
467 }
468 };
469 let stdout = join_pipe(stdout_reader, "stdout");
470 let stderr = join_pipe(stderr_reader, "stderr");
471 assert!(
472 !status.success(),
473 "provider incomplete stop must fail the exec\nstdout:\n{stdout}\nstderr:\n{stderr}"
474 );
475 assert!(
476 !stream_events(&stdout)
477 .iter()
478 .any(|event| event.get("type").and_then(Value::as_str) == Some("service_released")),
479 "a failed exec must never release a pending service"
480 );
481 wait_for_pid_death(service_pid);
482 }
483
484 #[tokio::test(flavor = "multi_thread")]
485 async fn terminating_signal_kills_pending_service_and_exits_nonzero() {
486 let _serial = serialize_persistent_service_tests().await;
487 // Hold the second model turn open long past the signal.
488 let server = start_mock_llm(
489 SERVICE_COMMAND,
490 final_answer_sse(),
491 Some(Duration::from_secs(300)),
492 None,
493 )
494 .await;
495 let workspace = TempDir::new().expect("workspace tempdir");
496 let home = TempDir::new().expect("home tempdir");
497
498 let mut child = exec_command(&server, workspace.path(), home.path())
499 .spawn()
500 .expect("spawn codewhale-tui exec");
501 let stdout_reader = read_pipe_in_background(child.stdout.take().expect("stdout pipe"));
502 let stderr_reader = read_pipe_in_background(child.stderr.take().expect("stderr pipe"));
503
504 // The pid file proves the service was staged before the signal.
505 let service_pid = wait_for_pid_file(&workspace.path().join("service.pid"))
506 .unwrap_or_else(|error| panic!("{error}"));
507 assert!(pid_is_alive(service_pid));
508
509 // SAFETY: direct child of this test.
510 unsafe {
511 libc::kill(
512 i32::try_from(child.id()).expect("child pid fits i32"),
513 libc::SIGTERM,
514 );
515 }
516 let status = child
517 .wait_timeout(Duration::from_secs(30))
518 .expect("wait for signalled exec")
519 .unwrap_or_else(|| {
520 let _ = child.kill();
521 let _ = child.wait();
522 panic!("signalled exec did not exit");
523 });
524 let _ = join_pipe(stdout_reader, "stdout");
525 let _ = join_pipe(stderr_reader, "stderr");
526
527 assert!(
528 !status.success(),
529 "a signalled exec must exit nonzero (got {status:?})"
530 );
531 wait_for_pid_death(service_pid);
532 }
533
533 lines RUST