bmux_application_resurrection_plugin 0.0.1-alpha.4

use super::*;

fn connect(endpoint: &str, accept_restore: bool) -> BufReader<UnixStream> {
    let stream = UnixStream::connect(endpoint.strip_prefix("v1:").unwrap()).unwrap();
    stream
        .set_read_timeout(Some(std::time::Duration::from_secs(3)))
        .unwrap();
    let mut reader = BufReader::new(stream);
    send(reader.get_mut(), &serde_json::json!({"protocol":PROTOCOL,"application":"test","pid":std::process::id(),"accept_restore":accept_restore})).unwrap();
    reader
}

#[test]
fn socket_persists_restores_and_manual_launch_does_not_resume() {
    let directory = tempfile::tempdir().unwrap();
    let key = uuid::Uuid::new_v4().to_string();
    let host = Host::new(directory.path().to_owned()).unwrap();
    let endpoint = host.prepare(&key, false, false, false).unwrap();
    host.bind(&key, &endpoint, std::process::id()).unwrap();
    let mut reader = connect(&endpoint, true);
    let welcome: serde_json::Value = receive(&mut reader).unwrap();
    assert!(welcome["restore"].is_null());
    let descriptor = serde_json::json!({"application":"test","session":"first"});
    send(
        reader.get_mut(),
        &serde_json::json!({"operation":"selection","revision":1,"descriptor":descriptor}),
    )
    .unwrap();
    let ack: serde_json::Value = receive(&mut reader).unwrap();
    assert_eq!(ack["revision"], 1);
    let checkpoint: Checkpoint = serde_json::from_slice(
        &std::fs::read(directory.path().join(format!("{key}.json"))).unwrap(),
    )
    .unwrap();
    assert!(checkpoint.descriptor.is_some());
    host.close(&key, &endpoint).unwrap();
    let restored = host.prepare(&key, true, false, false).unwrap();
    host.bind(&key, &restored, std::process::id()).unwrap();
    let mut reader = connect(&restored, true);
    let welcome: serde_json::Value = receive(&mut reader).unwrap();
    assert_eq!(welcome["restore"], descriptor);
    host.close(&key, &restored).unwrap();
    let manual = host.prepare(&key, false, false, false).unwrap();
    host.bind(&key, &manual, std::process::id()).unwrap();
    let mut reader = connect(&manual, true);
    let welcome: serde_json::Value = receive(&mut reader).unwrap();
    assert!(welcome["restore"].is_null());
}

#[test]
fn shell_registration_waits_for_command_boundary_even_after_bind() {
    let directory = tempfile::tempdir().unwrap();
    let host = Host::new(directory.path().to_owned()).unwrap();
    let key = uuid::Uuid::new_v4().to_string();
    let endpoint = host.prepare(&key, false, true, false).unwrap();
    host.bind(&key, &endpoint, std::process::id()).unwrap();
    let entry = host.entry(&key, &endpoint).unwrap();
    assert!(!entry.state.lock().unwrap().registration_ready());
    host.boundary(&key, &endpoint, false).unwrap(); // startup hook
    assert!(!entry.state.lock().unwrap().registration_ready());
    host.boundary(&key, &endpoint, true).unwrap(); // initial prompt
    assert!(!entry.state.lock().unwrap().registration_ready());
    host.boundary(&key, &endpoint, false).unwrap(); // actual invocation
    assert!(entry.state.lock().unwrap().registration_ready());
    host.boundary(&key, &endpoint, true).unwrap(); // command completed
    assert!(!entry.state.lock().unwrap().registration_ready());
}

#[test]
fn shell_replay_retains_only_initial_prompt_and_command() {
    let directory = tempfile::tempdir().unwrap();
    let host = Host::new(directory.path().to_owned()).unwrap();
    let key = uuid::Uuid::new_v4().to_string();
    let endpoint = host.prepare(&key, false, false, false).unwrap();
    let entry = host.entry(&key, &endpoint).unwrap();
    {
        let mut state = entry.state.lock().unwrap();
        let registration = state.execution.register(uuid::Uuid::new_v4()).unwrap();
        let checkpoint = state
            .execution
            .update(registration, 1, Some("{\"application\":\"test\"}".into()))
            .unwrap();
        save(&entry.file, &checkpoint).unwrap();
        state.execution.committed(checkpoint).unwrap();
    }
    let restored = host.prepare(&key, true, true, true).unwrap();
    host.boundary(&key, &restored, true).unwrap();
    host.boundary(&key, &restored, false).unwrap();
    let entry = host.entry(&key, &restored).unwrap();
    assert!(
        entry
            .state
            .lock()
            .unwrap()
            .execution
            .committed
            .descriptor
            .is_some()
    );
    host.boundary(&key, &restored, true).unwrap();
    assert!(
        entry
            .state
            .lock()
            .unwrap()
            .execution
            .committed
            .descriptor
            .is_none()
    );
    assert!(host.boundary(&key, &endpoint, false).is_err());
}

#[test]
fn replacing_endpoint_drains_old_writer_and_stale_close_cannot_remove_successor() {
    let directory = tempfile::tempdir().unwrap();
    let host = Arc::new(Host::new(directory.path().to_owned()).unwrap());
    let key = uuid::Uuid::new_v4().to_string();
    let old_endpoint = host.prepare(&key, false, false, false).unwrap();
    let old = host.entry(&key, &old_endpoint).unwrap();
    let mut writer = old.state.lock().unwrap();
    let owner = writer.execution.register(uuid::Uuid::new_v4()).unwrap();
    let candidate = writer
        .execution
        .update(owner, 1, Some("{\"session\":\"latest\"}".into()))
        .unwrap();
    let replacement_host = Arc::clone(&host);
    let replacement_key = key.clone();
    let replacement = std::thread::spawn(move || {
        replacement_host
            .prepare(&replacement_key, true, false, false)
            .unwrap()
    });
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3);
    while !old.stopped.load(Ordering::SeqCst) {
        assert!(
            std::time::Instant::now() < deadline,
            "replacement did not fence old writer"
        );
        std::thread::yield_now();
    }
    save(&old.file, &candidate).unwrap();
    writer.execution.committed(candidate).unwrap();
    drop(writer);
    let endpoint = replacement.join().unwrap();
    let current = host.entry(&key, &endpoint).unwrap();
    assert_eq!(
        current
            .state
            .lock()
            .unwrap()
            .execution
            .committed
            .descriptor
            .as_deref(),
        Some("{\"session\":\"latest\"}")
    );
    assert!(host.close(&key, &old_endpoint).is_err());
    assert!(host.entry(&key, &endpoint).is_ok());
}

#[test]
fn versioned_socket_reopens_latest_session_after_host_shutdown() {
    let directory = tempfile::tempdir().unwrap();
    let key = uuid::Uuid::new_v4().to_string();
    let host = Host::new(directory.path().to_owned()).unwrap();
    let launch = host
        .prepare_checked(&key, false, false, false, IdentityCheck::Expected(None))
        .unwrap();
    host.bind(&key, &launch.endpoint, std::process::id())
        .unwrap();
    let mut client = connect(&launch.endpoint, true);
    let welcome: serde_json::Value = receive(&mut client).unwrap();
    assert!(welcome["restore"].is_null());
    for (revision, session) in [(1, "first"), (2, "selected-at-shutdown")] {
        send(
            client.get_mut(),
            &serde_json::json!({"operation":"selection", "revision":revision,
            "descriptor":{"application":"test", "session":session}}),
        )
        .unwrap();
        let ack: serde_json::Value = receive(&mut client).unwrap();
        assert_eq!(ack["revision"], revision);
    }
    drop(host);
    drop(client);
    let restarted = Host::new(directory.path().to_owned()).unwrap();
    let restored = restarted
        .prepare_checked(
            &key,
            true,
            false,
            false,
            IdentityCheck::Expected(Some(&launch.invocation)),
        )
        .unwrap();
    restarted
        .bind(&key, &restored.endpoint, std::process::id())
        .unwrap();
    let mut client = connect(&restored.endpoint, true);
    let welcome: serde_json::Value = receive(&mut client).unwrap();
    assert_eq!(welcome["restore"]["session"], "selected-at-shutdown");
    assert_eq!(restored.invocation, launch.invocation);
    send(
        client.get_mut(),
        &serde_json::json!({"operation":"selection","revision":1,"descriptor":null}),
    )
    .unwrap();
    let ack: serde_json::Value = receive(&mut client).unwrap();
    assert_eq!(ack["revision"], 1);
    drop(restarted);
    drop(client);
    let restarted = Host::new(directory.path().to_owned()).unwrap();
    let restored = restarted
        .prepare_checked(
            &key,
            true,
            false,
            false,
            IdentityCheck::Expected(Some(&launch.invocation)),
        )
        .unwrap();
    restarted
        .bind(&key, &restored.endpoint, std::process::id())
        .unwrap();
    let mut client = connect(&restored.endpoint, true);
    let welcome: serde_json::Value = receive(&mut client).unwrap();
    assert!(
        welcome["restore"].is_null(),
        "cleared draft must not reopen previous session"
    );
}

#[test]
fn closed_endpoint_cannot_replace_final_invocation() {
    let directory = tempfile::tempdir().unwrap();
    let host = Host::new(directory.path().to_owned()).unwrap();
    let key = uuid::Uuid::new_v4().to_string();
    let launch = host
        .prepare_checked(&key, false, true, false, IdentityCheck::Expected(None))
        .unwrap();
    host.boundary_identity(&key, &launch.endpoint, true)
        .unwrap();
    let invocation = host
        .boundary_identity(&key, &launch.endpoint, false)
        .unwrap();
    host.close(&key, &launch.endpoint).unwrap();
    let file = directory.path().join(format!("{key}.json"));
    let final_checkpoint = std::fs::read(&file).unwrap();
    assert!(
        host.boundary_identity(&key, &launch.endpoint, true)
            .is_err()
    );
    assert_eq!(std::fs::read(&file).unwrap(), final_checkpoint);
    let restored = host
        .prepare_checked(
            &key,
            true,
            true,
            true,
            IdentityCheck::Expected(Some(&invocation)),
        )
        .unwrap();
    assert_eq!(restored.invocation, invocation);
}

#[test]
fn versioned_prepare_rejects_checkpoint_from_another_invocation() {
    let directory = tempfile::tempdir().unwrap();
    let host = Host::new(directory.path().to_owned()).unwrap();
    let key = uuid::Uuid::new_v4().to_string();
    let launch = host
        .prepare_checked(&key, false, false, false, IdentityCheck::Expected(None))
        .unwrap();
    assert!(
        host.prepare_checked(
            &key,
            true,
            false,
            false,
            IdentityCheck::Expected(Some("wrong-invocation"))
        )
        .is_err()
    );
    let restored = host
        .prepare_checked(
            &key,
            true,
            false,
            false,
            IdentityCheck::Expected(Some(&launch.invocation)),
        )
        .unwrap();
    assert_eq!(restored.invocation, launch.invocation);
}

#[test]
fn versioned_boundaries_return_identity_and_missing_references_fail() {
    let directory = tempfile::tempdir().unwrap();
    let host = Host::new(directory.path().to_owned()).unwrap();
    let key = uuid::Uuid::new_v4().to_string();
    assert!(
        host.prepare_checked(
            &key,
            true,
            false,
            false,
            IdentityCheck::Expected(Some("missing"))
        )
        .is_err()
    );
    let launch = host
        .prepare_checked(&key, false, true, false, IdentityCheck::Expected(None))
        .unwrap();
    assert_eq!(
        host.boundary_identity(&key, &launch.endpoint, true)
            .unwrap(),
        launch.invocation
    );
    let next = host
        .boundary_identity(&key, &launch.endpoint, false)
        .unwrap();
    assert_ne!(next, launch.invocation);
    let restored = host
        .prepare_checked(&key, true, true, true, IdentityCheck::Expected(Some(&next)))
        .unwrap();
    assert_eq!(
        host.boundary_identity(&key, &restored.endpoint, true)
            .unwrap(),
        next
    );
    assert_eq!(
        host.boundary_identity(&key, &restored.endpoint, false)
            .unwrap(),
        next
    );
    assert_ne!(
        host.boundary_identity(&key, &restored.endpoint, true)
            .unwrap(),
        next
    );
}

#[test]
fn old_snapshot_cannot_adopt_or_overwrite_unassociated_checkpoint() {
    let directory = tempfile::tempdir().unwrap();
    let host = Host::new(directory.path().to_owned()).unwrap();
    let key = uuid::Uuid::new_v4().to_string();
    let launch = host
        .prepare_checked(&key, false, false, false, IdentityCheck::Expected(None))
        .unwrap();
    let file = directory.path().join(format!("{key}.json"));
    let original = std::fs::read(&file).unwrap();
    assert!(
        host.prepare_checked(&key, true, false, false, IdentityCheck::Expected(None))
            .is_err()
    );
    assert_eq!(std::fs::read(&file).unwrap(), original);
    let recovered = host
        .prepare_checked(
            &key,
            true,
            false,
            false,
            IdentityCheck::Expected(Some(&launch.invocation)),
        )
        .unwrap();
    assert_eq!(recovered.invocation, launch.invocation);
}

#[test]
fn old_snapshot_without_checkpoint_starts_without_a_restore_offer() {
    let directory = tempfile::tempdir().unwrap();
    let host = Host::new(directory.path().to_owned()).unwrap();
    let key = uuid::Uuid::new_v4().to_string();
    let launch = host
        .prepare_checked(&key, true, false, false, IdentityCheck::Expected(None))
        .unwrap();
    host.bind(&key, &launch.endpoint, std::process::id())
        .unwrap();
    let mut client = connect(&launch.endpoint, true);
    let welcome: serde_json::Value = receive(&mut client).unwrap();
    assert!(welcome["restore"].is_null());
}

#[test]
fn checkpoint_read_distinguishes_missing_from_unreadable_and_oversized() {
    let directory = tempfile::tempdir().unwrap();
    assert!(
        load_checkpoint(&directory.path().join("missing"))
            .unwrap()
            .is_none()
    );
    assert!(load_checkpoint(directory.path()).is_err());
    let large = directory.path().join("large");
    std::fs::write(&large, vec![b' '; 65_537]).unwrap();
    assert!(load_checkpoint(&large).is_err());
}

#[test]
fn corrupt_checkpoint_does_not_become_empty_installation() {
    let directory = tempfile::tempdir().unwrap();
    let host = Host::new(directory.path().to_owned()).unwrap();
    let key = uuid::Uuid::new_v4().to_string();
    std::fs::write(directory.path().join(format!("{key}.json")), "corrupt").unwrap();
    assert!(host.prepare(&key, true, false, false).is_err());
}

#[test]
fn shell_ownership_requires_current_foreground_child() {
    assert!(foreground_child("100 200 200\n", 100));
    for identity in [
        "100 200 300",
        "101 200 200",
        "100 200 -1",
        "100 0 0",
        "100 200",
        "100 200 200 extra",
        "corrupt",
    ] {
        assert!(!foreground_child(identity, 100), "accepted {identity}");
    }
}

#[test]
fn kernel_peer_credentials_identify_the_connecting_process() {
    let (client, server) = UnixStream::pair().unwrap();
    assert_eq!(peer_pid(&client).unwrap(), std::process::id());
    assert_eq!(peer_pid(&server).unwrap(), std::process::id());
}

#[test]
fn actual_peer_identity_rejects_forged_pid() {
    let directory = tempfile::tempdir().unwrap();
    let host = Host::new(directory.path().to_owned()).unwrap();
    let key = uuid::Uuid::new_v4().to_string();
    let endpoint = host.prepare(&key, false, false, false).unwrap();
    host.bind(&key, &endpoint, std::process::id()).unwrap();
    let stream = UnixStream::connect(endpoint.strip_prefix("v1:").unwrap()).unwrap();
    stream
        .set_read_timeout(Some(std::time::Duration::from_secs(3)))
        .unwrap();
    let mut reader = BufReader::new(stream);
    send(reader.get_mut(), &serde_json::json!({"protocol":PROTOCOL,"application":"test","pid":1,"accept_restore":true})).unwrap();
    let rejection: serde_json::Value = receive(&mut reader).unwrap();
    assert_eq!(rejection["protocol"], PROTOCOL);
    assert_eq!(rejection["error"]["code"], "invalid_handshake");
    assert!(receive::<serde_json::Value>(&mut reader).is_err());
}

#[test]
fn unavailable_process_inspector_is_distinct_from_rejected_ownership() {
    assert!(foreground_owner(std::process::id(), 1, "/nonexistent/resurrection-ps").is_err());
    assert!(!belongs(std::process::id(), 1, false).unwrap());
    assert!(belongs(std::process::id(), std::process::id(), false).unwrap());
    // Exercise the actual platform tool, including PATH resolution on NixOS.
    assert!(belongs(std::process::id(), 1, true).is_ok());
}

#[test]
fn ownership_rejection_is_structured_and_does_not_claim_the_launch() {
    let directory = tempfile::tempdir().unwrap();
    let host = Host::new(directory.path().to_owned()).unwrap();
    let key = uuid::Uuid::new_v4().to_string();
    let endpoint = host.prepare(&key, false, false, false).unwrap();
    host.bind(&key, &endpoint, 1).unwrap();
    let mut reader = connect(&endpoint, true);
    let rejection: serde_json::Value = receive(&mut reader).unwrap();
    assert_eq!(rejection["error"]["code"], "ownership_rejected");
    assert!(receive::<serde_json::Value>(&mut reader).is_err());
    let entry = host.entry(&key, &endpoint).unwrap();
    assert!(entry.state.lock().unwrap().claim.is_none());
}

#[test]
fn inspection_failure_returns_a_public_rejection_without_claiming_ownership() {
    let directory = tempfile::tempdir().unwrap();
    let host = Host::new(directory.path().to_owned()).unwrap();
    let key = uuid::Uuid::new_v4().to_string();
    let endpoint = host.prepare(&key, false, false, false).unwrap();
    host.bind(&key, &endpoint, std::process::id()).unwrap();
    let entry = host.entry(&key, &endpoint).unwrap();
    let (mut client, server) = UnixStream::pair().unwrap();
    send(&mut client, &serde_json::json!({"protocol": PROTOCOL, "application": "test", "pid": std::process::id(), "accept_restore": true})).unwrap();
    let mut reader = BufReader::new(server);
    let result = accept_with_inspector(&entry, &mut reader, |_, _, _| {
        Err("private OS diagnostic".into())
    });
    let Err(Rejection(code)) = result else {
        panic!("inspection failure must reject the handshake")
    };
    assert_eq!(code, "ownership_verification_unavailable");
    assert!(entry.state.lock().unwrap().claim.is_none());
}