Genarrative/apps/ai-game-creator-shell/src-tauri/src/process_session/tests.rs

use super::*;
#[cfg(target_os = "linux")]
use std::process::Stdio;

#[cfg(target_os = "linux")]
mod owner_fixture_cleanup;
#[cfg(target_os = "linux")]
use owner_fixture_cleanup::{project_processes, OwnerFixtureCleanup};

static PROCESS_SESSION_TEST_LOCK: OnceLock<Mutex<()>> = 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::<ProcessSessionTranscript>(
        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::<Vec<_>>();
    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() {
    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 cleanup = OwnerFixtureCleanup {
        owner: &mut owner,
        root,
    };
    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"
    );

    // Linux Child::kill 发送 SIGKILL；先检查真实子树回收，再由 guard 兜底。
    cleanup.owner.kill().expect("SIGKILL owner fixture");
    cleanup.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};

    // 系统进程快照（pid, ppid），用于证明 ping 子进程真实存在并被收割。
    fn windows_process_snapshot_for_test() -> Vec<(u32, u32)> {
        use windows_sys::Win32::Foundation::{CloseHandle, INVALID_HANDLE_VALUE};
        use windows_sys::Win32::System::Diagnostics::ToolHelp::{
            CreateToolhelp32Snapshot, Process32FirstW, Process32NextW, PROCESSENTRY32W,
            TH32CS_SNAPPROCESS,
        };
        // SAFETY: 快照句柄非 INVALID_HANDLE_VALUE 时由 CloseHandle 释放。
        let snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0) };
        if snapshot == INVALID_HANDLE_VALUE {
            return Vec::new();
        }
        let mut entry = PROCESSENTRY32W::default();
        entry.dwSize = std::mem::size_of::<PROCESSENTRY32W>() as u32;
        let mut processes = Vec::new();
        // SAFETY: entry 指向可写的 PROCESSENTRY32W，dwSize 已初始化。
        let mut available = unsafe { Process32FirstW(snapshot, &mut entry) };
        while available != 0 {
            processes.push((entry.th32ProcessID, entry.th32ParentProcessID));
            // SAFETY: 同上。
            available = unsafe { Process32NextW(snapshot, &mut entry) };
        }
        // SAFETY: snapshot 是本函数持有的合法句柄。
        unsafe { CloseHandle(snapshot) };
        processes
    }

    // 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));
    // 前置检查不能只凭 Job 非空：快照里必须真的看到 cmd 拉起了 ping 子进程。
    let tree_before: Vec<u32> = {
        let snapshot = windows_process_snapshot_for_test();
        let mut tree = vec![child.id()];
        let mut index = 0;
        while index < tree.len() {
            let parent = tree[index];
            for (pid, ppid) in &snapshot {
                if *ppid == parent && !tree.contains(pid) {
                    tree.push(*pid);
                }
            }
            index += 1;
        }
        tree
    };
    assert!(
        tree_before.len() >= 2,
        "fixture 必须包含 ping 子进程，否则收割断言是空转: {tree_before:?}"
    );
    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));
    }
    // Job 为空之外，快照里的整棵树（含 ping 子进程）也必须真实消失。
    let deadline = std::time::Instant::now() + Duration::from_secs(10);
    loop {
        let snapshot = windows_process_snapshot_for_test();
        let survivors: Vec<u32> = tree_before
            .iter()
            .copied()
            .filter(|pid| snapshot.iter().any(|(live, _)| live == pid))
            .collect();
        if survivors.is_empty() {
            break;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "Job terminate 后 fixture 子进程仍存活: {survivors:?}"
        );
        thread::sleep(Duration::from_millis(50));
    }
    let _ = child.kill();
    let _ = child.wait();
}