use super::*; #[cfg(target_os = "linux")] use std::process::Stdio; static PROCESS_SESSION_TEST_LOCK: OnceLock> = OnceLock::new(); fn process_session_test_guard() -> std::sync::MutexGuard<'static, ()> { PROCESS_SESSION_TEST_LOCK .get_or_init(|| Mutex::new(())) .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) } fn process_identity(project_id: &str) -> ProcessSessionIdentity { ProcessSessionIdentity { project_id: project_id.to_string(), agent_id: "code-prototype".to_string(), task_id: "code-prototype".to_string(), conversation_session_id: "session-process-test".to_string(), run_id: "run-process-test".to_string(), start_action_id: "action-process-start-test".to_string(), start_action_fingerprint: "a".repeat(64), } } fn process_test_command_spec(root: &Path) -> ProjectCommandSpec { fs::write( root.join("package.json"), r#"{"scripts":{"dev":"node fixture.js"}}"#, ) .expect("write process test package.json"); resolve_project_command_spec_at( root, "npm", &["run".to_string(), "dev".to_string()], ".", 30, ) .expect("resolve process test command") } #[test] fn process_session_cursor_preserves_unicode_boundaries() { let process_id = "proc-0123456789abcdef0123456789abcdef"; let state = ProcessOutputState { text: "甲乙abc".to_string(), status: "running".to_string(), exit_code: None, signal: None, stdin_open: true, reader_finished: false, output_limit_exceeded: false, source_fingerprint_after: None, source_changed: None, needs_reconciliation: false, launch_failure_kind: None, }; let first = poll_result_from_output( process_id, &state.text, &state, "test", "test", "test", "test-v1", "established", "established", None, None, 2, ) .expect("first unicode page"); assert_eq!(first.output, "甲乙"); assert!(first.has_more); let second = poll_result_from_output( process_id, &state.text, &state, "test", "test", "test", "test-v1", "established", "established", None, Some(&first.next_cursor), 3, ) .expect("second unicode page"); assert_eq!(second.output, "abc"); assert!(!second.has_more); } fn write_legacy_process_session_record( root: &Path, record: &ProcessSessionRecord, schema_version: &str, ) { let mut value = serde_json::to_value(record).expect("serialize legacy record"); let object = value.as_object_mut().expect("legacy record object"); object.insert( "schemaVersion".to_string(), serde_json::Value::String(schema_version.to_string()), ); for field in [ "sandboxEstablishment", "targetExec", "launchFailureKind", "sandboxReadyAt", "execEstablishedAt", ] { object.remove(field); } if schema_version == "1" { for field in [ "sandboxBackend", "sandboxMode", "networkAccess", "sandboxProfileVersion", ] { object.remove(field); } } write_agent_runtime_json_sidecar_with_max_bytes( root, &process_session_record_relative_path(&record.process_id), "legacy process session", &value, PROCESS_SESSION_RECORD_MAX_BYTES, ) .expect("write legacy process record"); } #[test] fn process_session_v1_v2_active_records_migrate_to_v3_reconciliation() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "legacy-active-project", "Legacy Active Project") .expect("initialize project"); for (index, schema_version) in ["1", "2"].into_iter().enumerate() { let mut identity = process_identity("legacy-active-project"); identity.start_action_id = format!("legacy-active-action-{index}"); identity.start_action_fingerprint = format!("{}", index + 1).repeat(64); let process_id = format!("proc-{:032x}", index + 1); let record = initial_process_session_record( &identity, &process_id, &format!("cmd-legacy-active-{index}"), &process_test_command_spec(root), None, &"a".repeat(64), "running", ); write_legacy_process_session_record(root, &record, schema_version); let migrated = read_process_session_record(root, &process_id) .expect("read migrated active record") .expect("active record exists"); assert_eq!(migrated.schema_version, "3"); assert_eq!(migrated.status, "needs-reconciliation"); assert_eq!(migrated.sandbox_establishment, "unknown"); assert_eq!(migrated.target_exec, "unknown"); assert_eq!( migrated.launch_failure_kind.as_deref(), Some("legacy-active-record") ); assert!(migrated.needs_reconciliation); assert!(migrated.terminal_at.is_some()); let repeated = read_process_session_record(root, &process_id) .expect("read migrated active record again") .expect("active record remains"); assert_eq!(repeated, migrated); } clear_process_session_registry_for_tests(); } #[test] fn process_session_v1_v2_terminal_records_remain_readable_without_reconciliation() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "legacy-terminal-project", "Legacy Terminal Project") .expect("initialize project"); for (index, schema_version) in ["1", "2"].into_iter().enumerate() { let mut identity = process_identity("legacy-terminal-project"); identity.start_action_id = format!("legacy-terminal-action-{index}"); identity.start_action_fingerprint = format!("{}", index + 3).repeat(64); let process_id = format!("proc-{:032x}", index + 16); let mut record = initial_process_session_record( &identity, &process_id, &format!("cmd-legacy-terminal-{index}"), &process_test_command_spec(root), None, &"b".repeat(64), "exited", ); record.exit_code = Some(index as i32); record.terminal_at = Some(record.started_at); let output = format!("LEGACY-TERMINAL-{schema_version}"); record.output_bytes = output.len(); record.output_sha256 = format!("{:x}", Sha256::digest(output.as_bytes())); record.output_ref = Some(process_session_transcript_relative_path(&process_id)); let transcript = ProcessSessionTranscript { schema_version: schema_version.to_string(), project_id: identity.project_id.clone(), agent_id: identity.agent_id.clone(), task_id: identity.task_id.clone(), conversation_session_id: identity.conversation_session_id.clone(), run_id: identity.run_id.clone(), start_action_id: identity.start_action_id.clone(), start_action_fingerprint: identity.start_action_fingerprint.clone(), process_id: process_id.clone(), output: output.clone(), output_sha256: record.output_sha256.clone(), output_bytes: output.len(), updated_at: record.updated_at, }; write_agent_runtime_json_sidecar_with_max_bytes( root, record.output_ref.as_deref().expect("legacy output ref"), "legacy process transcript", &transcript, PROCESS_SESSION_TRANSCRIPT_MAX_BYTES, ) .expect("write legacy process transcript"); write_legacy_process_session_record(root, &record, schema_version); let migrated = read_process_session_record(root, &process_id) .expect("read migrated terminal record") .expect("terminal record exists"); assert_eq!(migrated.schema_version, "3"); assert_eq!(migrated.status, "exited"); assert_eq!(migrated.sandbox_establishment, "unknown"); assert_eq!(migrated.target_exec, "unknown"); assert_eq!( migrated.launch_failure_kind.as_deref(), Some("legacy-record") ); assert!(!migrated.needs_reconciliation); let poll = poll_process_session_at(root, &identity, &process_id, None, Some(8_000), Some(0)) .expect("poll migrated terminal record"); assert_eq!(poll.status, "exited"); assert_eq!(poll.exit_code, Some(index as i32)); assert_eq!(poll.output, output); } clear_process_session_registry_for_tests(); } #[test] fn process_session_v3_rejects_untrusted_state_combinations_and_timestamps() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "v3-validation-project", "V3 Validation Project") .expect("initialize project"); let identity = process_identity("v3-validation-project"); let process_id = "proc-bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; let spec = process_test_command_spec(root); let mut record = initial_process_session_record( &identity, process_id, "cmd-v3-validation", &spec, None, &"c".repeat(64), "running", ); assert!(validate_process_session_record(root, &record, process_id).is_err()); record.sandbox_establishment = "established".to_string(); record.target_exec = "established".to_string(); record.sandbox_ready_at = Some(record.started_at); record.exec_established_at = Some(record.started_at); record.launch_failure_kind = Some("arbitrary".to_string()); assert!(validate_process_session_record(root, &record, process_id).is_err()); record.launch_failure_kind = None; record.sandbox_ready_at = Some(record.started_at.saturating_add(2)); record.exec_established_at = Some(record.started_at.saturating_add(1)); assert!(validate_process_session_record(root, &record, process_id).is_err()); record.status = "failed".to_string(); record.sandbox_establishment = "unknown".to_string(); record.target_exec = "unknown".to_string(); record.launch_failure_kind = Some("launch-unknown".to_string()); record.sandbox_ready_at = None; record.exec_established_at = None; record.terminal_at = Some(record.started_at); record.needs_reconciliation = false; assert!(validate_process_session_record(root, &record, process_id).is_err()); record.status = "needs-reconciliation".to_string(); record.needs_reconciliation = true; assert!(validate_process_session_record(root, &record, process_id).is_ok()); record.needs_reconciliation = false; assert!(validate_process_session_record(root, &record, process_id).is_err()); record.needs_reconciliation = true; record.sandbox_establishment = "established".to_string(); record.target_exec = "established".to_string(); record.launch_failure_kind = Some("start-audit-failed".to_string()); record.sandbox_ready_at = Some(record.started_at); record.exec_established_at = Some(record.started_at); assert!(validate_process_session_record(root, &record, process_id).is_ok()); record.status = "failed".to_string(); assert!(validate_process_session_record(root, &record, process_id).is_err()); record.status = "launching".to_string(); record.needs_reconciliation = false; record.target_exec = "not-attempted".to_string(); record.launch_failure_kind = None; record.exec_established_at = None; record.terminal_at = None; assert!(validate_process_session_record(root, &record, process_id).is_ok()); record.sandbox_ready_at = None; assert!(validate_process_session_record(root, &record, process_id).is_err()); clear_process_session_registry_for_tests(); } #[cfg(target_os = "linux")] #[test] fn pending_process_launch_blocks_idle_until_guard_is_dropped() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "pending-launch-project", "Pending Launch Project") .expect("initialize project"); let pending = reserve_pending_process_launch( root, "code-prototype", "proc-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", ) .expect("register pending launch"); assert!(has_active_process_sessions_at(root).expect("pending launch is active")); shutdown_all_process_sessions(); assert!( activate_pending_process_launch("proc-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", i32::MAX,) .expect_err("shutdown cancels a launch before pid activation") .contains("shutdown") ); drop(pending); assert!(!has_active_process_sessions_at(root).expect("pending launch removed")); clear_process_session_registry_for_tests(); } #[test] fn process_session_ansi_stripper_handles_split_csi_and_osc() { let mut stripper = AnsiStripper::default(); let mut visible = Vec::new(); for chunk in [ b"A\x1b[3".as_slice(), b"1mB\x1b]52;c;secret".as_slice(), b"\x07C\x1b[0m\n".as_slice(), ] { for byte in chunk { stripper.push(*byte, &mut visible); } } assert_eq!(String::from_utf8(visible).expect("utf8"), "ABC\n"); } #[cfg(windows)] #[test] fn process_session_terminal_reposition_detector_ignores_color_sequences() { let mut detector = AnsiTerminalRepositionDetector::default(); let color = b"\x1b[31m"; assert!(!color.iter().any(|byte| detector.push(*byte))); let mut reposition = AnsiTerminalRepositionDetector::default(); let sequence = b"\x1b[5;1H"; assert_eq!( sequence .iter() .filter(|byte| reposition.push(**byte)) .count(), 1 ); } #[cfg(windows)] #[test] fn process_session_windows_npm_builder_uses_node_cli_and_native_script_shell() { let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "npm-builder-project", "Npm Builder Project") .expect("initialize project"); fs::write( root.join("package.json"), r#"{"scripts":{"dev":"node fixture.js"}}"#, ) .expect("write package.json"); let spec = resolve_project_command_spec_at( root, "npm", &["run".to_string(), "dev".to_string()], ".", 30, ) .expect("resolve npm command"); let launch = prepare_project_command_launch_spec(root, &spec).expect("prepare npm launch"); let command = process_session_command_builder(&launch).expect("build npm PTY command"); let argv = command.get_argv(); assert!(argv[0].to_string_lossy().ends_with("node.exe")); assert!(argv[1] .to_string_lossy() .replace('\\', "/") .ends_with("node_modules/npm/bin/npm-cli.js")); assert!(argv .iter() .all(|argument| !argument.to_string_lossy().starts_with(r"\\?\"))); assert_eq!( command.get_env("npm_config_script_shell"), Some(std::ffi::OsStr::new(r"C:\Windows\System32\cmd.exe")) ); assert_eq!( command.get_env("NODE"), command.get_env("npm_node_execpath") ); assert!(command .get_cwd() .is_some_and(|cwd| !cwd.to_string_lossy().starts_with(r"\\?\"))); } #[test] fn process_session_real_pty_streams_stdin_and_terminates() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "process-project", "Process Project") .expect("initialize project"); fs::write( root.join("package.json"), r#"{"scripts":{"dev":"node fixture.js"}}"#, ) .expect("write package.json"); fs::write( root.join("fixture.js"), r#" process.stdin.setEncoding('utf8'); console.log('\u001b[31mREADY\u001b[0m'); console.log(`BRIDGE_ENV:${Object.keys(globalThis['process']['env']).filter((name) => name.includes('PROCESS_SESSION_BRIDGE')).join(',')}`); console.log(`TARGET_ARGV:${process.argv.join('|')}`); process.stdin.on('data', (chunk) => console.log(`ECHO:${chunk.trim()}`)); process.on('SIGTERM', () => { console.log('STOPPED'); process.exit(0); }); setInterval(() => {}, 1000); "#, ) .expect("write fixture"); let spec = resolve_project_command_spec_at( root, "npm", &["run".to_string(), "dev".to_string()], ".", 30, ) .expect("resolve npm command"); let identity = process_identity("process-project"); let source_fingerprint = project_command_source_fingerprint(root).expect("source fingerprint"); let mut poll = start_process_session_at(root, identity.clone(), &spec, source_fingerprint) .expect("start process session"); assert!(poll.output.is_empty()); assert_eq!(poll.cursor, poll.next_cursor); #[cfg(target_os = "linux")] { assert_eq!(poll.sandbox_backend, "bubblewrap"); assert_eq!(poll.sandbox_mode, "workspace-write"); assert_eq!(poll.network_access, "disabled"); assert_eq!(poll.sandbox_profile_version, "workspace-v1"); assert_eq!(poll.sandbox_establishment, "established"); assert_eq!(poll.target_exec, "established"); assert_eq!(poll.launch_failure_kind, None); } assert!(has_active_process_sessions_at(root).expect("active process probe")); let mut combined = poll.output.clone(); for _ in 0..20 { if combined.contains("READY") { break; } poll = poll_process_session_at( root, &identity, &poll.process_id, Some(&poll.next_cursor), Some(8_000), Some(500), ) .expect("poll ready"); combined.push_str(&poll.output); } assert!(combined.contains("READY"), "output: {combined}"); assert!(!combined.contains("[31m"), "output: {combined}"); let mut foreign_identity = identity.clone(); foreign_identity.agent_id = "art-director".to_string(); assert!(poll_process_session_at( root, &foreign_identity, &poll.process_id, None, Some(10), Some(0), ) .is_err()); assert!(write_process_session_stdin_at( root, &foreign_identity, &poll.process_id, "blocked", true, false, ) .is_err()); let stdin = write_process_session_stdin_at(root, &identity, &poll.process_id, "你好", true, false) .expect("write stdin"); assert_eq!(stdin.bytes_written, "你好\n".len()); let mut echo = String::new(); for _ in 0..20 { poll = poll_process_session_at( root, &identity, &poll.process_id, Some(&poll.next_cursor), Some(8_000), Some(500), ) .expect("poll echo"); echo.push_str(&poll.output); if echo.contains("ECHO:你好") { break; } } assert!(echo.contains("ECHO:你好"), "output: {echo}"); let terminal = terminate_process_session_at(root, &identity, &poll.process_id, Some(&poll.next_cursor)) .expect("terminate process session"); assert_ne!(terminal.status, "running"); let record = read_process_session_record(root, &poll.process_id) .expect("read record") .expect("record exists"); assert_eq!(record.status, "terminated"); assert!(record.output_ref.is_some()); #[cfg(target_os = "linux")] { assert_eq!(record.sandbox_backend, "bubblewrap"); assert_eq!(record.sandbox_mode, "workspace-write"); assert_eq!(record.network_access, "disabled"); assert_eq!(record.sandbox_profile_version, "workspace-v1"); } let transcript = read_agent_runtime_json_sidecar_with_max_bytes::( root, record.output_ref.as_deref().expect("transcript ref"), "Agent Runtime process transcript", PROCESS_SESSION_TRANSCRIPT_MAX_BYTES, ) .expect("read transcript") .expect("transcript exists"); let transcript_lines = transcript.output.lines().collect::>(); assert!( transcript_lines.contains(&"READY"), "{:?}", transcript.output ); assert!( transcript_lines.contains(&"ECHO:你好"), "{:?}", transcript.output ); #[cfg(unix)] assert!( transcript_lines.contains(&"STOPPED"), "{:?}", transcript.output ); assert!( transcript_lines .iter() .any(|line| line.ends_with("BRIDGE_ENV:")), "{:?}", transcript.output ); let record_json = fs::read_to_string(root.join(process_session_record_relative_path(&record.process_id))) .expect("read process record json"); for private_marker in [ "GENARRATIVE_PROCESS_SESSION_BRIDGE_ENDPOINT", "GENARRATIVE_PROCESS_SESSION_BRIDGE_NONCE", "genarrative-ps-", "sandbox_ready", "commit_exec", "exec_established", ] { assert!( !transcript.output.contains(private_marker), "private marker leaked: {private_marker}: {:?}", transcript.output ); assert!( !record_json.contains(private_marker), "private marker leaked to record: {private_marker}: {record_json}" ); } assert!(!has_active_process_sessions_at(root).expect("terminal process probe")); clear_process_session_registry_for_tests(); } #[cfg(target_os = "linux")] #[test] fn process_session_graceful_terminate_keeps_wrapper_alive_for_target_cleanup() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "graceful-process-project", "Graceful Process Project") .expect("initialize project"); let spec = resolve_project_command_spec_at( root, "bash", &[ "-lc".to_string(), "(trap 'sleep 0.4; printf done > graceful-marker.txt; exit 0' TERM; while :; do sleep 1; done) & printf 'READY\\n'; exit 0".to_string(), ], ".", 30, ) .expect("resolve graceful command"); let identity = process_identity("graceful-process-project"); let fingerprint = project_command_source_fingerprint(root).expect("fingerprint"); let mut poll = start_process_session_at(root, identity.clone(), &spec, fingerprint).expect("start"); for _ in 0..20 { if poll.output.contains("READY") { break; } poll = poll_process_session_at( root, &identity, &poll.process_id, Some(&poll.next_cursor), Some(8_000), Some(250), ) .expect("poll graceful ready"); } assert!(poll.output.contains("READY")); let terminated = terminate_process_session_at(root, &identity, &poll.process_id, Some(&poll.next_cursor)) .expect("graceful terminate"); assert_eq!(terminated.status, "terminated"); assert!(!terminated.needs_reconciliation); assert_eq!( fs::read_to_string(root.join("graceful-marker.txt")) .expect("target completed delayed SIGTERM cleanup"), "done" ); clear_process_session_registry_for_tests(); } #[cfg(target_os = "linux")] #[test] fn process_session_live_registry_blocks_when_durable_record_is_missing() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "live-record-project", "Live Record Project") .expect("initialize project"); let spec = resolve_project_command_spec_at( root, "bash", &[ "-lc".to_string(), "printf 'READY\\n'; while :; do sleep 1; done".to_string(), ], ".", 30, ) .expect("resolve live record command"); let identity = process_identity("live-record-project"); let fingerprint = project_command_source_fingerprint(root).expect("fingerprint"); let started = start_process_session_at(root, identity.clone(), &spec, fingerprint).expect("start"); fs::remove_file(root.join(process_session_record_relative_path(&started.process_id))) .expect("remove durable process record"); let active = active_process_session_records_at( root, Some(identity.agent_id.as_str()), Some(identity.run_id.as_str()), ) .expect("live registry remains authoritative blocker"); assert!(active .iter() .any(|record| record.process_id == started.process_id)); assert!(has_active_process_sessions_at(root).expect("live registry blocks idle")); terminate_process_session_at(root, &identity, &started.process_id, None) .expect("terminate live record fixture"); clear_process_session_registry_for_tests(); } #[cfg(target_os = "linux")] #[test] fn process_session_fast_exit_zero_and_seven_keep_same_process_id() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); for exit_code in [0, 7] { let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); let project_id = format!("fast-exit-{exit_code}-project"); init_local_game_project_at(root, &project_id, "Fast Exit Project") .expect("initialize project"); let spec = resolve_project_command_spec_at( root, "bash", &[ "-lc".to_string(), format!("printf 'FAST-{exit_code}\\n'; exit {exit_code}"), ], ".", 30, ) .expect("resolve fast exit command"); let mut identity = process_identity(&project_id); identity.start_action_id = format!("fast-exit-action-{exit_code}"); identity.start_action_fingerprint = format!("{}", exit_code + 1).repeat(64); let fingerprint = project_command_source_fingerprint(root).expect("fingerprint"); let started = start_process_session_at(root, identity.clone(), &spec, fingerprint).expect("start"); let process_id = started.process_id.clone(); assert!(started.output.is_empty()); assert_eq!(started.cursor, started.next_cursor); assert_eq!(started.sandbox_establishment, "established"); assert_eq!(started.target_exec, "established"); let mut poll = started; let mut output = String::new(); for _ in 0..30 { poll = poll_process_session_at( root, &identity, &process_id, Some(&poll.next_cursor), Some(8_000), Some(250), ) .expect("poll fast exit"); assert_eq!(poll.process_id, process_id); output.push_str(&poll.output); if poll.status != "running" && !poll.has_more { break; } } assert_eq!(poll.status, "exited", "{output}"); assert_eq!(poll.exit_code, Some(exit_code), "{output}"); assert!(output.contains(&format!("FAST-{exit_code}")), "{output}"); let record = read_process_session_record(root, &process_id) .expect("read fast exit record") .expect("fast exit record exists"); assert_eq!(record.process_id, process_id); assert_eq!(record.target_exec, "established"); assert!(record.exec_established_at.is_some()); clear_process_session_registry_for_tests(); } } #[cfg(target_os = "linux")] #[test] fn process_session_durable_commit_failure_runs_no_target_and_writes_no_record() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "durable-failure-project", "Durable Failure Project") .expect("initialize project"); let spec = resolve_project_command_spec_at( root, "bash", &[ "-lc".to_string(), "printf ran > durable-target-ran.txt".to_string(), ], ".", 30, ) .expect("resolve durable failure command"); let launch = prepare_project_command_launch_spec(root, &spec).expect("prepare launch"); let identity = process_identity("durable-failure-project"); let fingerprint = project_command_source_fingerprint(root).expect("fingerprint"); let error = start_prepared_process_session_at( root, identity.clone(), &spec, &launch, fingerprint, || { assert!(!root.join("durable-target-ran.txt").exists()); assert!(find_existing_start_action_record(root, &identity) .expect("record absent inside durable callback") .is_none()); Err("forced durable commit failure".to_string()) }, ) .expect_err("durable commit must fail"); assert_eq!(error.stage(), ProjectCommandErrorStage::DurableCommit); assert!(!root.join("durable-target-ran.txt").exists()); assert!(find_existing_start_action_record(root, &identity) .expect("search process record") .is_none()); assert!(!has_active_process_sessions_at(root).expect("no pending launch")); clear_process_session_registry_for_tests(); } #[cfg(target_os = "linux")] #[test] fn process_session_slow_durable_commit_keeps_target_blocked_until_commit() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "slow-commit-project", "Slow Commit Project") .expect("initialize project"); let spec = resolve_project_command_spec_at( root, "bash", &[ "-lc".to_string(), "printf committed > slow-commit-target.txt".to_string(), ], ".", 30, ) .expect("resolve slow commit command"); let launch = prepare_project_command_launch_spec(root, &spec).expect("prepare launch"); let identity = process_identity("slow-commit-project"); let fingerprint = project_command_source_fingerprint(root).expect("fingerprint"); let started_at = std::time::Instant::now(); let result = start_prepared_process_session_at( root, identity.clone(), &spec, &launch, fingerprint, || { thread::sleep(Duration::from_millis(3_200)); assert!(!root.join("slow-commit-target.txt").exists()); Ok(()) }, ) .expect("slow durable commit must not time out in child"); assert!(started_at.elapsed() >= Duration::from_millis(3_200)); assert_eq!(result.sandbox_establishment, "established"); assert_eq!(result.target_exec, "established"); let mut poll = result; for _ in 0..20 { poll = poll_process_session_at( root, &identity, &poll.process_id, Some(&poll.next_cursor), Some(8_000), Some(250), ) .expect("poll slow commit target"); if poll.status != "running" { break; } } assert_eq!(poll.status, "exited"); assert_eq!( fs::read_to_string(root.join("slow-commit-target.txt")).expect("read slow commit marker"), "committed" ); clear_process_session_registry_for_tests(); } #[cfg(windows)] #[test] fn process_session_windows_replay_runs_durable_commit_only_once() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "windows-replay-project", "Windows Replay Project") .expect("initialize project"); fs::write( root.join("package.json"), r#"{"scripts":{"dev":"node fixture.js"}}"#, ) .expect("write package.json"); fs::write(root.join("fixture.js"), "setInterval(() => {}, 1000);\n").expect("write fixture"); let spec = resolve_project_command_spec_at( root, "npm", &["run".to_string(), "dev".to_string()], ".", 30, ) .expect("resolve npm command"); let launch = prepare_project_command_launch_spec(root, &spec).expect("prepare launch"); let identity = process_identity("windows-replay-project"); let fingerprint = project_command_source_fingerprint(root).expect("fingerprint"); let commit_count = std::sync::atomic::AtomicUsize::new(0); let first = start_prepared_process_session_at( root, identity.clone(), &spec, &launch, fingerprint.clone(), || { commit_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); Ok(()) }, ) .expect("first start"); let second = start_prepared_process_session_at( root, identity.clone(), &spec, &launch, fingerprint, || { commit_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); Ok(()) }, ) .expect("idempotent replay"); assert_eq!(first.process_id, second.process_id); assert_eq!(commit_count.load(std::sync::atomic::Ordering::SeqCst), 1); let record = read_process_session_record(root, &first.process_id) .expect("read Windows replay record") .expect("Windows replay record exists"); assert_eq!(record.sandbox_ready_at, Some(record.started_at)); assert_eq!(record.exec_established_at, Some(record.started_at)); terminate_process_session_at(root, &identity, &first.process_id, None) .expect("terminate replay fixture"); clear_process_session_registry_for_tests(); } #[cfg(target_os = "linux")] #[test] fn process_session_target_exec_failure_is_known_terminal_record() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "target-exec-failure-project", "Target Exec Failure") .expect("initialize project"); let mut spec = resolve_project_command_spec_at( root, "bash", &["-lc".to_string(), "exit 0".to_string()], ".", 30, ) .expect("resolve command"); let launch = prepare_project_command_launch_spec(root, &spec).expect("prepare launch"); spec.executable = PathBuf::from("/definitely-missing-genarrative-target"); let identity = process_identity("target-exec-failure-project"); let fingerprint = project_command_source_fingerprint(root).expect("fingerprint"); let committed = std::sync::atomic::AtomicBool::new(false); let result = start_prepared_process_session_at(root, identity, &spec, &launch, fingerprint, || { committed.store(true, std::sync::atomic::Ordering::SeqCst); Ok(()) }) .expect("target exec failure is a known result"); assert!(committed.load(std::sync::atomic::Ordering::SeqCst)); assert_eq!(result.status, "failed"); assert_eq!(result.sandbox_establishment, "established"); assert_eq!(result.target_exec, "failed"); assert_eq!( result.launch_failure_kind.as_deref(), Some("target-exec-failed") ); assert!(!result.needs_reconciliation); assert_eq!(result.cursor, result.next_cursor); assert!(!has_active_process_sessions_at(root).expect("known target failure is terminal")); clear_process_session_registry_for_tests(); } #[cfg(target_os = "linux")] #[test] fn process_session_start_audit_failure_terminates_and_persists_reconciliation() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "start-audit-project", "Start Audit Project") .expect("initialize project"); let spec = resolve_project_command_spec_at( root, "bash", &["-lc".to_string(), "cat >/dev/null".to_string()], ".", 30, ) .expect("resolve command"); let identity = process_identity("start-audit-project"); let fingerprint = project_command_source_fingerprint(root).expect("fingerprint"); let started = start_process_session_at(root, identity, &spec, fingerprint).expect("start"); mark_process_session_start_audit_failure_at( root, &started.process_id, "forced agent db failure", ) .expect("mark start audit reconciliation"); let mut record = None; for _ in 0..30 { let current = read_process_session_record(root, &started.process_id) .expect("read audit failure record") .expect("audit failure record exists"); if current.status == "needs-reconciliation" && current.launch_failure_kind.as_deref() == Some("start-audit-failed") { record = Some(current); break; } thread::sleep(Duration::from_millis(50)); } let record = record.expect("audit failure reconciliation persisted"); assert!(record.needs_reconciliation); assert!(!record.stdin_open); assert_eq!(record.target_exec, "established"); assert!(active_process_session_records_at(root, None, None) .expect("audit failure blocks completion") .iter() .any(|value| value.process_id == started.process_id)); clear_process_session_registry_for_tests(); } #[cfg(target_os = "linux")] #[test] fn process_session_descendants_inherit_workspace_sandbox() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path().join("workspace"); let outside = directory.path().join("outside-secret.txt"); init_local_game_project_at(&root, "process-sandbox-project", "Process Sandbox Project") .expect("initialize project"); fs::write(&outside, "OUTSIDE_SECRET").expect("write outside secret"); fs::create_dir_all(root.join(".git")).expect("create git control directory"); fs::write(root.join(".git/marker"), "git").expect("write git marker"); let outside_literal = outside.to_string_lossy().replace('"', "\\\""); let script = format!( r#"set -u if cat "{outside_literal}" >/dev/null 2>&1; then echo OUTSIDE_VISIBLE; else echo OUTSIDE_BLOCKED; fi if (printf no > .git/blocked-write) 2>/dev/null; then echo GIT_WRITABLE; else echo GIT_BLOCKED; fi if cat .agent/manifest.json >/dev/null 2>&1; then echo AGENT_VISIBLE; else echo AGENT_HIDDEN; fi /usr/bin/setsid /bin/bash -lc 'cd /tmp; if test -e "{outside_literal}"; then echo DESCENDANT_VISIBLE; else echo DESCENDANT_BLOCKED; fi' /usr/bin/python3 - <<'PY' import socket s = socket.socket() s.settimeout(0.2) try: s.connect(("1.1.1.1", 53)) print("NETWORK_VISIBLE") except OSError: print("NETWORK_BLOCKED") finally: s.close() PY "# ); fs::write(root.join("sandbox-probe.sh"), script).expect("write sandbox probe"); let spec = resolve_project_command_spec_at(&root, "bash", &["sandbox-probe.sh".to_string()], ".", 30) .expect("resolve sandbox probe"); let identity = process_identity("process-sandbox-project"); let fingerprint = project_command_source_fingerprint(&root).expect("source fingerprint"); let mut poll = start_process_session_at(&root, identity.clone(), &spec, fingerprint) .expect("start sandbox probe"); let mut output = poll.output.clone(); for _ in 0..30 { if poll.status != "running" && !poll.has_more { break; } poll = poll_process_session_at( &root, &identity, &poll.process_id, Some(&poll.next_cursor), Some(8_000), Some(250), ) .expect("poll sandbox probe"); output.push_str(&poll.output); } assert_eq!(poll.status, "exited", "{output}"); for marker in [ "OUTSIDE_BLOCKED", "GIT_BLOCKED", "AGENT_HIDDEN", "DESCENDANT_BLOCKED", "NETWORK_BLOCKED", ] { assert!(output.contains(marker), "missing {marker}: {output}"); } for marker in [ "OUTSIDE_VISIBLE", "GIT_WRITABLE", "AGENT_VISIBLE", "DESCENDANT_VISIBLE", "NETWORK_VISIBLE", ] { assert!(!output.contains(marker), "unexpected {marker}: {output}"); } assert_eq!(poll.sandbox_backend, "bubblewrap"); assert_eq!(poll.sandbox_mode, "workspace-write"); assert_eq!(poll.network_access, "disabled"); clear_process_session_registry_for_tests(); } #[test] fn process_session_runner_shutdown_reaps_active_session() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "process-shutdown-project", "Process Shutdown Project") .expect("initialize project"); fs::write( root.join("package.json"), r#"{"scripts":{"dev":"node fixture.js"}}"#, ) .expect("write package.json"); fs::write( root.join("fixture.js"), r#" console.log('READY'); setInterval(() => {}, 1000); "#, ) .expect("write fixture"); let spec = resolve_project_command_spec_at( root, "npm", &["run".to_string(), "dev".to_string()], ".", 30, ) .expect("resolve npm command"); let identity = process_identity("process-shutdown-project"); let source_fingerprint = project_command_source_fingerprint(root).expect("source fingerprint"); let started = start_process_session_at(root, identity.clone(), &spec, source_fingerprint) .expect("start process session"); assert!(has_active_process_sessions_at(root).expect("active process probe")); let mut cursor = None; let mut output = String::new(); for _ in 0..40 { let poll = poll_process_session_at( root, &identity, &started.process_id, cursor.as_deref(), Some(2_000), Some(500), ) .expect("observe shutdown fixture"); output.push_str(&poll.output); cursor = Some(poll.next_cursor); if output.contains("READY") { break; } } assert!(output.contains("READY"), "{output}"); shutdown_all_process_sessions_and_wait(Duration::from_secs(3)) .expect("shutdown active process sessions"); let terminal = poll_process_session_at( root, &identity, &started.process_id, None, Some(8_000), Some(0), ) .expect("poll shutdown terminal state"); assert_eq!(terminal.status, "terminated"); assert_eq!(terminal.signal.as_deref(), Some("runner-shutdown")); assert!(live_process_session(&started.process_id) .expect("inspect terminal registry") .is_none()); assert!(!has_active_process_sessions_at(root).expect("terminal process probe")); clear_process_session_registry_for_tests(); } #[test] fn process_session_terminal_reconciliation_still_blocks_completion() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "process-reconciliation-project", "Process Project") .expect("initialize project"); fs::write( root.join("package.json"), r#"{"scripts":{"dev":"node fixture.js"}}"#, ) .expect("write package.json"); fs::write(root.join("fixture.js"), "setInterval(() => {}, 1000);\n").expect("write fixture"); let process_id = "proc-0123456789abcdef0123456789abcdef"; let now = unix_timestamp(); let mut record = ProcessSessionRecord { schema_version: PROCESS_SESSION_SCHEMA_VERSION.to_string(), project_id: "process-reconciliation-project".to_string(), agent_id: "code-prototype".to_string(), task_id: "code-prototype".to_string(), conversation_session_id: "session-process-test".to_string(), run_id: "run-process-test".to_string(), start_action_id: "action-process-start-test".to_string(), start_action_fingerprint: "a".repeat(64), process_id: process_id.to_string(), owner_boot_id: process_session_boot_id().to_string(), command_id: "cmd-process-test".to_string(), program: "npm".to_string(), cwd: ".".to_string(), sandbox_backend: "test-unknown".to_string(), sandbox_mode: "unknown".to_string(), network_access: "unknown".to_string(), sandbox_profile_version: "test-v1".to_string(), sandbox_establishment: "unknown".to_string(), target_exec: "unknown".to_string(), launch_failure_kind: Some("launch-unknown".to_string()), sandbox_ready_at: None, exec_established_at: None, status: "needs-reconciliation".to_string(), exit_code: None, signal: Some("output-read-failed".to_string()), stdin_open: false, output_bytes: 0, output_sha256: format!("{:x}", Sha256::digest([])), output_ref: None, source_fingerprint_before: "b".repeat(64), source_fingerprint_after: None, source_changed: None, needs_reconciliation: true, started_at: now, terminal_at: Some(now), updated_at: now, }; write_process_session_record(root, &record).expect("write reconciliation record"); let active = active_process_session_records_at(root, Some("code-prototype"), Some("run-process-test")) .expect("read reconciliation blockers"); assert_eq!(active.len(), 1); assert_eq!(active[0].process_id, process_id); let spec = resolve_project_command_spec_at( root, "npm", &["run".to_string(), "dev".to_string()], ".", 30, ) .expect("resolve blocked command"); let mut blocked_identity = process_identity("process-reconciliation-project"); blocked_identity.run_id = "run-process-blocked-test".to_string(); blocked_identity.start_action_id = "action-process-blocked-start".to_string(); let blocked = validate_process_session_start_preflight_at(root, &blocked_identity, &spec) .expect_err("reconciliation must block a new process session"); assert!(blocked.contains(process_id)); record.needs_reconciliation = false; write_process_session_record(root, &record).expect("write invalid reconciliation record"); assert!(active_process_session_records_at( root, Some("code-prototype"), Some("run-process-test") ) .expect_err("invalid reconciliation record must fail closed") .contains("可信 launch 状态组合无效")); clear_process_session_registry_for_tests(); } #[test] fn process_session_capacity_preflight_counts_durable_records() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "process-capacity-project", "Process Capacity Project") .expect("initialize project"); fs::write( root.join("package.json"), r#"{"scripts":{"dev":"node fixture.js"}}"#, ) .expect("write package.json"); fs::write(root.join("fixture.js"), "setInterval(() => {}, 1000);\n").expect("write fixture"); let spec = resolve_project_command_spec_at( root, "npm", &["run".to_string(), "dev".to_string()], ".", 30, ) .expect("resolve command"); for (index, process_id) in [ "proc-11111111111111111111111111111111", "proc-22222222222222222222222222222222", ] .into_iter() .enumerate() { let mut identity = process_identity("process-capacity-project"); identity.run_id = format!("run-process-capacity-{index}"); identity.start_action_id = format!("action-process-capacity-{index}"); identity.start_action_fingerprint = format!("{}", index + 1).repeat(64); let mut record = initial_process_session_record( &identity, process_id, &format!("cmd-process-capacity-{index}"), &spec, None, &"f".repeat(64), "running", ); record.sandbox_establishment = "established".to_string(); record.target_exec = "established".to_string(); record.sandbox_ready_at = Some(record.started_at); record.exec_established_at = Some(record.started_at); write_process_session_record(root, &record).expect("write durable running record"); } let mut blocked_identity = process_identity("process-capacity-project"); blocked_identity.run_id = "run-process-capacity-blocked".to_string(); blocked_identity.start_action_id = "action-process-capacity-blocked".to_string(); let error = validate_process_session_start_preflight_at(root, &blocked_identity, &spec) .expect_err("durable records must count toward Agent capacity"); assert!(error.contains("最多同时运行 2 个"), "{error}"); clear_process_session_registry_for_tests(); } #[test] fn process_session_real_pty_eof_reaches_terminal() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "process-eof-project", "Process EOF Project") .expect("initialize project"); fs::write( root.join("package.json"), r#"{"scripts":{"dev":"node fixture.js"}}"#, ) .expect("write package.json"); fs::write( root.join("fixture.js"), r#" process.stdin.setEncoding('utf8'); console.log('READY'); process.stdin.on('end', () => { console.log('EOF'); process.exit(0); }); process.stdin.resume(); "#, ) .expect("write fixture"); let spec = resolve_project_command_spec_at( root, "npm", &["run".to_string(), "dev".to_string()], ".", 30, ) .expect("resolve npm command"); let mut identity = process_identity("process-eof-project"); identity.run_id = "run-process-eof-test".to_string(); identity.start_action_id = "action-process-eof-start".to_string(); identity.start_action_fingerprint = "c".repeat(64); let fingerprint = project_command_source_fingerprint(root).expect("source fingerprint"); let mut poll = start_process_session_at(root, identity.clone(), &spec, fingerprint).expect("start"); for _ in 0..20 { if poll.output.contains("READY") { break; } poll = poll_process_session_at( root, &identity, &poll.process_id, Some(&poll.next_cursor), Some(8_000), Some(500), ) .expect("poll ready"); } // 自然 EOF 与本用例显式 close 存在竞态：PTY 在负载下可能先收到流 EOF， // 子进程随即退出并把会话置为终态，此时 close 会以「已进入终态」拒绝。 // 该终态正是用例要验证的收敛结果，因此只在会话仍未终止时要求 close 成功。 match write_process_session_stdin_at(root, &identity, &poll.process_id, "", false, true) { Ok(eof) => { assert!(eof.eof); assert!(!eof.stdin_open); } Err(error) => { assert!( error.contains("已进入终态"), "closing stdin failed for a non-terminal session: {error}" ); } } let mut tail = String::new(); for _ in 0..20 { poll = poll_process_session_at( root, &identity, &poll.process_id, Some(&poll.next_cursor), Some(8_000), Some(500), ) .expect("poll eof"); tail.push_str(&poll.output); if poll.status != "running" { break; } } assert_eq!(poll.status, "exited", "tail: {tail}"); #[cfg(not(windows))] assert!(tail.contains("EOF"), "tail: {tail}"); // Closing a ConPTY input pipe closes the attached Windows console. Unlike a Unix PTY, // Node's console stdin does not emit its stream-level `end` callback before that terminal // close, so the portable contract here is the trusted `exited` terminal state above. clear_process_session_registry_for_tests(); } #[test] fn process_session_stdin_accepts_trusted_terminal_race_after_successful_eof() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "stdin-race-project", "Stdin Race Project") .expect("initialize project"); fs::write( root.join("package.json"), r#"{"scripts":{"dev":"node fixture.js"}}"#, ) .expect("write package.json"); fs::write( root.join("fixture.js"), "process.stdout.write('READY\\n'); setInterval(() => {}, 1000);\n", ) .expect("write fixture"); let spec = resolve_project_command_spec_at( root, "npm", &["run".to_string(), "dev".to_string()], ".", 30, ) .expect("resolve stdin race command"); let identity = process_identity("stdin-race-project"); let fingerprint = project_command_source_fingerprint(root).expect("fingerprint"); let started = start_process_session_at(root, identity.clone(), &spec, fingerprint).expect("start"); let result = write_process_session_stdin_at_with_after_write( root, &identity, &started.process_id, "", false, true, |live| { let mut output = live.output.lock().expect("lock terminal race output"); output.status = "exited".to_string(); output.exit_code = Some(0); output.stdin_open = false; }, ) .expect("successful EOF remains successful after trusted terminal wins race"); assert!(result.eof); assert!(!result.stdin_open); let record = read_process_session_record(root, &started.process_id) .expect("read terminal race record") .expect("terminal race record exists"); assert_eq!(record.status, "exited"); assert!(!record.needs_reconciliation); clear_process_session_registry_for_tests(); } #[test] fn process_session_overlong_unterminated_line_is_stopped() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "process-output-project", "Process Output Project") .expect("initialize project"); fs::write( root.join("package.json"), r#"{"scripts":{"dev":"node fixture.js"}}"#, ) .expect("write package.json"); fs::write( root.join("fixture.js"), "process.stdout.write('x'.repeat(20000)); setInterval(() => {}, 1000);\n", ) .expect("write fixture"); let spec = resolve_project_command_spec_at( root, "npm", &["run".to_string(), "dev".to_string()], ".", 30, ) .expect("resolve npm command"); let mut identity = process_identity("process-output-project"); identity.run_id = "run-process-output-test".to_string(); identity.start_action_id = "action-process-output-start".to_string(); identity.start_action_fingerprint = "d".repeat(64); let fingerprint = project_command_source_fingerprint(root).expect("source fingerprint"); let mut poll = start_process_session_at(root, identity.clone(), &spec, fingerprint).expect("start"); let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); while std::time::Instant::now() < deadline { if poll.status != "running" { break; } poll = poll_process_session_at( root, &identity, &poll.process_id, Some(&poll.next_cursor), Some(8_000), Some(100), ) .expect("poll output limit"); } assert_eq!(poll.status, "output-limit-exceeded"); clear_process_session_registry_for_tests(); } #[test] fn process_session_old_boot_becomes_reconciliation_without_relaunch() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "stale-process-project", "Stale Process Project") .expect("initialize project"); let identity = process_identity("stale-process-project"); let process_id = "proc-fedcba9876543210fedcba9876543210"; let mut record = ProcessSessionRecord { schema_version: PROCESS_SESSION_SCHEMA_VERSION.to_string(), project_id: identity.project_id.clone(), agent_id: identity.agent_id.clone(), task_id: identity.task_id.clone(), conversation_session_id: identity.conversation_session_id.clone(), run_id: identity.run_id.clone(), start_action_id: identity.start_action_id.clone(), start_action_fingerprint: identity.start_action_fingerprint.clone(), process_id: process_id.to_string(), owner_boot_id: "old-runner-boot".to_string(), command_id: "cmd-stale".to_string(), program: "npm".to_string(), cwd: ".".to_string(), sandbox_backend: "bubblewrap".to_string(), sandbox_mode: "workspace-write".to_string(), network_access: "disabled".to_string(), sandbox_profile_version: "workspace-v1".to_string(), sandbox_establishment: "established".to_string(), target_exec: "established".to_string(), launch_failure_kind: None, sandbox_ready_at: Some(unix_timestamp()), exec_established_at: Some(unix_timestamp()), status: "running".to_string(), exit_code: None, signal: None, stdin_open: true, output_bytes: 0, output_sha256: format!("{:x}", Sha256::digest([])), output_ref: None, source_fingerprint_before: "b".repeat(64), source_fingerprint_after: None, source_changed: None, needs_reconciliation: false, started_at: unix_timestamp(), terminal_at: None, updated_at: unix_timestamp(), }; write_process_session_record(root, &record).expect("write stale record"); let poll = poll_process_session_at(root, &identity, process_id, None, Some(10), Some(0)) .expect("reconcile stale record"); assert_eq!(poll.status, "needs-reconciliation"); assert!(poll.needs_reconciliation); record = read_process_session_record(root, process_id) .expect("read reconciled record") .expect("record exists"); assert_eq!(record.status, "needs-reconciliation"); assert!(record.needs_reconciliation); let launching_process_id = "proc-abcdefabcdefabcdefabcdefabcdefab"; let spec = resolve_project_command_spec_at( root, "npm", &["run".to_string(), "dev".to_string()], ".", 30, ) .expect("resolve stale launching command"); let mut launching = initial_process_session_record( &identity, launching_process_id, "cmd-stale-launching", &spec, None, &"d".repeat(64), "launching", ); launching.owner_boot_id = "old-launching-boot".to_string(); launching.sandbox_establishment = "established".to_string(); launching.sandbox_ready_at = Some(launching.started_at); write_process_session_record(root, &launching).expect("write stale launching record"); let launching_poll = poll_process_session_at( root, &identity, launching_process_id, None, Some(10), Some(0), ) .expect("reconcile stale launching record"); assert_eq!(launching_poll.status, "needs-reconciliation"); assert_eq!(launching_poll.target_exec, "unknown"); assert_eq!( launching_poll.launch_failure_kind.as_deref(), Some("launch-unknown") ); assert!(launching_poll.needs_reconciliation); } #[cfg(target_os = "linux")] #[test] fn process_session_start_replay_reconciles_old_launching_record_once() { let _guard = process_session_test_guard(); clear_process_session_registry_for_tests(); let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "stale-replay-project", "Stale Replay Project") .expect("initialize project"); let identity = process_identity("stale-replay-project"); let spec = resolve_project_command_spec_at( root, "bash", &["-lc".to_string(), "exit 0".to_string()], ".", 30, ) .expect("resolve stale replay command"); let launch = prepare_project_command_launch_spec(root, &spec).expect("prepare launch"); let process_id = process_session_id(&identity); let mut record = initial_process_session_record( &identity, &process_id, "cmd-stale-replay", &spec, Some(&launch), &"e".repeat(64), "launching", ); record.owner_boot_id = "old-replay-boot".to_string(); record.sandbox_establishment = "established".to_string(); record.sandbox_ready_at = Some(record.started_at); write_process_session_record(root, &record).expect("write stale replay record"); let callback_called = std::sync::atomic::AtomicBool::new(false); let error = start_prepared_process_session_at(root, identity, &spec, &launch, "e".repeat(64), || { callback_called.store(true, std::sync::atomic::Ordering::SeqCst); Ok(()) }) .expect_err("old launching replay must reconcile without relaunch"); assert_eq!(error.stage(), ProjectCommandErrorStage::LaunchUnknown); assert!(!callback_called.load(std::sync::atomic::Ordering::SeqCst)); let reconciled = read_process_session_record(root, &process_id) .expect("read replay reconciliation") .expect("replay record exists"); assert_eq!(reconciled.status, "needs-reconciliation"); assert_eq!(reconciled.target_exec, "unknown"); assert!(reconciled.exec_established_at.is_none()); assert_eq!( reconciled.launch_failure_kind.as_deref(), Some("launch-unknown") ); assert!(reconciled.needs_reconciliation); clear_process_session_registry_for_tests(); } #[cfg(target_os = "linux")] #[test] fn process_session_child_wrapper_fixture() { if std::env::var_os(PROCESS_SESSION_BRIDGE_ENDPOINT_ENV).is_none() { return; } let args = vec![PROCESS_SESSION_CHILD_MODE.to_string()]; match run_process_session_child(&args) { Ok(exit_code) => std::process::exit(exit_code), Err(error) => panic!("process session child wrapper failed: {error}"), } } #[test] fn process_session_runner_owner_fixture() { let Some(root) = std::env::var_os("GENARRATIVE_PROCESS_SESSION_OWNER_FIXTURE_ROOT") else { return; }; let root = PathBuf::from(root); let spec = resolve_project_command_spec_at( &root, "npm", &["run".to_string(), "dev".to_string()], ".", 300, ) .expect("resolve owner fixture command"); let identity = process_identity("owner-process-project"); let source_fingerprint = project_command_source_fingerprint(&root).expect("owner fixture fingerprint"); let poll = start_process_session_at(&root, identity, &spec, source_fingerprint) .expect("start owner fixture process"); fs::write(root.join("owner-ready"), poll.process_id).expect("write owner ready"); loop { thread::sleep(Duration::from_secs(1)); } } #[cfg(target_os = "linux")] #[test] fn process_session_owner_sigkill_leaves_no_child_process() { fn project_processes(root: &Path) -> Vec { let canonical_root = fs::canonicalize(root).expect("canonical test project"); fs::read_dir("/proc") .into_iter() .flatten() .flatten() .filter_map(|entry| { let process_id = entry.file_name().to_string_lossy().parse::().ok()?; let cwd = fs::read_link(entry.path().join("cwd")).ok()?; (cwd == canonical_root).then_some(process_id) }) .collect() } let directory = tempfile::tempdir().expect("temp project"); let root = directory.path(); init_local_game_project_at(root, "owner-process-project", "Owner Process Project") .expect("initialize project"); fs::write( root.join("package.json"), r#"{"scripts":{"dev":"node owner-fixture.js"}}"#, ) .expect("write package.json"); fs::write( root.join("owner-fixture.js"), r#" process.on('SIGHUP', () => {}); require('fs').writeFileSync('child.pid', String(process.pid)); setInterval(() => {}, 1000); "#, ) .expect("write fixture"); let current_exe = std::env::current_exe().expect("current test binary"); let mut owner = std::process::Command::new(current_exe) .arg("--exact") .arg("process_session::tests::process_session_runner_owner_fixture") .arg("--nocapture") .env("GENARRATIVE_PROCESS_SESSION_OWNER_FIXTURE_ROOT", root) .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::null()) .spawn() .expect("spawn owner fixture test process"); let deadline = std::time::Instant::now() + Duration::from_secs(10); while (!root.join("owner-ready").is_file() || project_processes(root).is_empty()) && std::time::Instant::now() < deadline { thread::sleep(Duration::from_millis(25)); } let sandbox_processes = project_processes(root); assert!( !sandbox_processes.is_empty(), "sandbox child should be visible from host /proc" ); let owner_pid = i32::try_from(owner.id()).expect("owner pid"); assert_eq!(unsafe { libc::kill(owner_pid, libc::SIGKILL) }, 0); owner.wait().expect("reap owner fixture"); let deadline = std::time::Instant::now() + Duration::from_secs(5); loop { let remaining = project_processes(root); if remaining.is_empty() { break; } assert!( std::time::Instant::now() < deadline, "Runner owner SIGKILL 后 sandbox 子进程仍存在：pids={remaining:?}" ); thread::sleep(Duration::from_millis(25)); } } #[cfg(windows)] #[test] fn windows_process_job_terminate_reaps_process_tree() { use std::process::{Command, Stdio}; // cmd 启动第一个 ping 子进程后整树存活；terminate 必须连子进程一起收割。 // timeout.exe 在 stdio 被重定向时会立即退出，ping 才能在 null stdio 下存活。 let mut child = Command::new("cmd.exe") .args([ "/c", "ping", "127.0.0.1", "-n", "60", "&", "ping", "127.0.0.1", "-n", "60", ]) .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::null()) .spawn() .expect("spawn cmd fixture"); let job = WindowsProcessJob::assign_std(&child).expect("assign job"); thread::sleep(Duration::from_millis(500)); assert!( !job.is_empty().expect("query job"), "fixture 进程树必须先存活，否则收割断言是空转" ); job.terminate().expect("terminate job"); let deadline = std::time::Instant::now() + Duration::from_secs(10); while !job.is_empty().expect("query job") { assert!( std::time::Instant::now() < deadline, "Windows Job 进程树未被收割" ); thread::sleep(Duration::from_millis(50)); } let _ = child.kill(); let _ = child.wait(); }