magi-code 0.77.1

//! Real processes and sockets; synthetic execution is confined to this test binary.
use super::*;
use serde_json::{Value, json};
use std::process::{Child, Command, Stdio};

const CHILD_TEST: &str = "service::unix::survival_tests::synthetic_process_child";
const ROOT_ENV: &str = "MAGI_SYNTHETIC_SOCKET_TEST_ROOT";
const ROLE_ENV: &str = "MAGI_SYNTHETIC_SOCKET_TEST_ROLE";

struct TestProcess(Child);
impl Drop for TestProcess {
    fn drop(&mut self) {
        let _ = self.0.kill();
        let _ = self.0.wait();
    }
}

fn spawn_child(root: &Path, role: &str) -> TestProcess {
    TestProcess(
        Command::new(std::env::current_exe().unwrap())
            .args(["--exact", CHILD_TEST, "--nocapture"])
            .env(ROOT_ENV, root)
            .env(ROLE_ENV, role)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .spawn()
            .unwrap(),
    )
}

fn wait_until(mut ready: impl FnMut() -> bool) {
    let deadline = Instant::now() + Duration::from_secs(15);
    while !ready() {
        assert!(
            Instant::now() < deadline,
            "synthetic process did not settle"
        );
        thread::sleep(Duration::from_millis(10));
    }
}

struct Frontend {
    socket: UnixStream,
    instance: Value,
    connection: Value,
}
impl Frontend {
    fn connect(identity: &Identity) -> Self {
        let mut socket = UnixStream::connect(&identity.socket).unwrap();
        socket
            .set_read_timeout(Some(Duration::from_secs(5)))
            .unwrap();
        socket
            .set_write_timeout(Some(Duration::from_secs(5)))
            .unwrap();
        send_json(
            &mut socket,
            &Hello {
                version: VERSION.into(),
                workspace: identity.workspace.clone(),
                state_root: identity.state_root.clone(),
                action: "connect".into(),
            },
        )
        .unwrap();
        let hello: Reply =
            serde_json::from_slice(&read_line(&mut socket, HANDSHAKE_LIMIT).unwrap()).unwrap();
        assert_eq!(hello.status, "ready");
        let mut frontend = Self {
            socket,
            instance: Value::Null,
            connection: Value::Null,
        };
        let init = frontend.call(
            "initialize",
            Value::Null,
            Value::Null,
            Value::Null,
            json!({"supported_protocol_versions":[2],"requested_capabilities":[]}),
        );
        frontend.instance = init["instance_id"].clone();
        frontend.connection = init["connection_id"].clone();
        frontend
    }

    fn call(
        &mut self,
        method: &str,
        session: Value,
        control: Value,
        operation: Value,
        payload: Value,
    ) -> Value {
        let request_id = uuid::Uuid::new_v4().to_string();
        send_json(
            &mut self.socket,
            &json!({
                "protocol_version":2,"kind":"request","request_id":request_id,
                "instance_id":self.instance,"connection_id":self.connection,
                "session_id":session,"control":control,"operation_id":operation,
                "method":method,"payload":payload,
            }),
        )
        .unwrap();
        loop {
            let reply: Value = serde_json::from_slice(
                &read_line(&mut self.socket, crate::service::protocol::MAX_RECORD_BYTES).unwrap(),
            )
            .unwrap();
            if reply["kind"] == "response" && reply["request_id"] == request_id {
                assert!(reply["error"].is_null(), "{reply}");
                return reply["payload"].clone();
            }
        }
    }
}

#[test]
fn synthetic_process_child() {
    let Some(root) = std::env::var_os(ROOT_ENV) else {
        return;
    };
    let root = PathBuf::from(root);
    let identity = Identity::resolve(&root, &root.join("state")).unwrap();
    match std::env::var(ROLE_ENV).unwrap().as_str() {
        "daemon" => {
            let (release, wait) = crossbeam_channel::bounded(1);
            let (cleanup, cleanup_wait) = crossbeam_channel::bounded(1);
            let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
            let service = Arc::new(PersistentService::synthetic_socket_fixture(
                &root,
                wait,
                cleanup_wait,
                Arc::clone(&calls),
            ));
            let _lock = lock_identity(&identity).unwrap();
            if identity.socket.exists() {
                check_path(&identity.socket, Kind::Socket).unwrap();
                assert_eq!(
                    connect::connect(&identity.socket).unwrap_err().kind(),
                    std::io::ErrorKind::ConnectionRefused
                );
                fs::remove_file(&identity.socket).unwrap();
            }
            let listener = UnixListener::bind(&identity.socket).unwrap();
            fs::set_permissions(&identity.socket, fs::Permissions::from_mode(0o600)).unwrap();
            listener.set_nonblocking(true).unwrap();
            fs::write(root.join("ready"), "").unwrap();
            let mut clients = Vec::new();
            let mut released = false;
            let mut cleaned = false;
            let deadline = Instant::now() + Duration::from_secs(30);
            while !service.is_finished() {
                assert!(Instant::now() < deadline, "daemon fixture timed out");
                fs::write(
                    root.join("provider-calls"),
                    calls.load(Ordering::SeqCst).to_string(),
                )
                .unwrap();
                if !released && root.join("release").exists() {
                    release.send(()).unwrap();
                    released = true;
                }
                if !cleaned && root.join("cleanup").exists() {
                    cleanup.send(()).unwrap();
                    cleaned = true;
                }
                match listener.accept() {
                    Ok((stream, _)) => {
                        let first_frontend = clients.is_empty();
                        let service = Arc::clone(&service);
                        let identity = identity.clone();
                        let root = root.clone();
                        clients.push(thread::spawn(move || {
                            let _ = serve(stream, &identity, &service, &AtomicBool::new(false));
                            // The first socket is the submitting frontend. serve has now
                            // synchronously disconnected it from the real coordinator.
                            if first_frontend {
                                fs::write(root.join("disconnected"), "").unwrap();
                            }
                        }));
                    }
                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
                        thread::sleep(Duration::from_millis(5))
                    }
                    Err(error) => panic!("listener failed: {error}"),
                }
            }
            for client in clients {
                client.join().unwrap();
            }
            fs::write(
                root.join("provider-calls"),
                calls.load(Ordering::SeqCst).to_string(),
            )
            .unwrap();
            drop(service);
            fs::remove_file(&identity.socket).unwrap();
        }
        "frontend" => {
            let mut frontend = Frontend::connect(&identity);
            let session = frontend.call(
                "session.create",
                Value::Null,
                Value::Null,
                json!("create"),
                json!({}),
            )["session_id"]
                .clone();
            let claim = frontend.call(
                "session.claim",
                session.clone(),
                Value::Null,
                json!("claim"),
                json!({}),
            );
            let accepted = frontend.call(
                "turn.start",
                session.clone(),
                json!({"grant_id":claim["grant_id"],"generation":claim["generation"]}),
                json!("surviving-turn"),
                json!({"prompt":"Read the synthetic fixture"}),
            );
            assert_eq!(accepted["status"], "accepted");
            fs::write(
                root.join("accepted.tmp"),
                serde_json::to_vec(&json!({"session":session,"instance":frontend.instance}))
                    .unwrap(),
            )
            .unwrap();
            fs::rename(root.join("accepted.tmp"), root.join("accepted")).unwrap();
            // Keep the socket alive until the parent kills this process, not graceful EOF.
            loop {
                thread::sleep(Duration::from_secs(1));
            }
        }
        role => panic!("unknown test role: {role}"),
    }
}

#[test]
fn accepted_synthetic_turn_survives_frontend_process_death_and_rejects_busy_stop() {
    let root = tempfile::TempDir::new().unwrap();
    let identity = Identity::resolve(root.path(), &root.path().join("state")).unwrap();
    let mut daemon = spawn_child(root.path(), "daemon");
    wait_until(|| root.path().join("ready").exists());
    let mut frontend = spawn_child(root.path(), "frontend");
    wait_until(|| root.path().join("accepted").exists());
    let accepted: Value =
        serde_json::from_slice(&fs::read(root.path().join("accepted")).unwrap()).unwrap();
    frontend.0.kill().unwrap();
    assert!(!frontend.0.wait().unwrap().success());
    wait_until(|| root.path().join("disconnected").exists());
    // No frontend remains, and the worker is blocked before producing output.
    assert_eq!(request(&identity, "stop").unwrap().status, "busy");
    assert!(daemon.0.try_wait().unwrap().is_none());
    fs::write(root.path().join("release"), "").unwrap();

    let mut reconnected = Frontend::connect(&identity);
    assert_eq!(reconnected.instance, accepted["instance"]);
    let sessions = crate::sessions::SessionManager::new(root.path().join("state/sessions"));
    let session = sessions
        .open_existing(accepted["session"].as_str().unwrap())
        .unwrap();
    wait_until(|| {
        let replay = match session.frontend_replay(None, 32) {
            Ok(replay) => replay,
            Err(error) if error.is::<crate::sessions::FrontendSnapshotBusy>() => return false,
            Err(error) => panic!("replay failed: {error}"),
        };
        serde_json::to_value(replay)
            .unwrap()
            .to_string()
            .contains("after tool")
    });
    assert!(session.try_frontend_writer().unwrap().is_none());
    let pending = reconnected.call(
        "operation.lookup",
        Value::Null,
        Value::Null,
        Value::Null,
        json!({"target_instance_id":accepted["instance"],"operation_id":"surviving-turn"}),
    );
    assert_eq!(pending["state"], "accepted", "cleanup still owns the lease");
    assert_eq!(request(&identity, "stop").unwrap().status, "busy");
    fs::write(root.path().join("cleanup"), "").unwrap();
    let mut outcome = Value::Null;
    wait_until(|| {
        outcome = reconnected.call(
            "operation.lookup",
            Value::Null,
            Value::Null,
            Value::Null,
            json!({"target_instance_id":accepted["instance"],"operation_id":"surviving-turn"}),
        );
        outcome["state"] == "terminal"
    });
    assert_eq!(outcome["result"]["persistence"], "committed");
    assert!(session.try_frontend_writer().unwrap().is_some());
    let claimed = reconnected.call(
        "session.claim",
        accepted["session"].clone(),
        Value::Null,
        json!("reclaim"),
        json!({}),
    );
    assert_eq!(claimed["snapshot"]["phase"], "idle");
    assert_eq!(claimed["snapshot"]["terminal"]["status"], "completed");
    assert_eq!(
        claimed["snapshot"]["terminal"]["assistant_text"],
        "before tool\n\nafter tool"
    );
    assert!(daemon.0.try_wait().unwrap().is_none());
    assert!(
        session.try_frontend_writer().unwrap().is_none(),
        "an idle controller must exclude a standalone writer"
    );
    drop(reconnected);
    wait_until(|| request(&identity, "stop").unwrap().status == "stopped");
    wait_until(|| daemon.0.try_wait().unwrap().is_some());
    assert!(daemon.0.wait().unwrap().success());
    assert!(!identity.socket.exists());
}

#[test]
fn daemon_crash_does_not_replay_an_accepted_active_turn_after_restart() {
    use crate::sessions::{SessionEventKind, SessionManager};

    let root = tempfile::TempDir::new().unwrap();
    let identity = Identity::resolve(root.path(), &root.path().join("state")).unwrap();
    let mut daemon = spawn_child(root.path(), "daemon");
    wait_until(|| root.path().join("ready").exists());
    let mut frontend = spawn_child(root.path(), "frontend");
    wait_until(|| root.path().join("accepted").exists());
    let accepted: Value =
        serde_json::from_slice(&fs::read(root.path().join("accepted")).unwrap()).unwrap();
    // One provider invocation proves this is executing work, not merely queued admission.
    // The provider blocks before any output until the release file exists.
    wait_until(|| fs::read_to_string(root.path().join("provider-calls")).unwrap() == "1");
    let sessions = SessionManager::new(root.path().join("state/sessions"));
    let session = sessions
        .open_existing(accepted["session"].as_str().unwrap())
        .unwrap();
    assert_eq!(request(&identity, "stop").unwrap().status, "busy");
    assert!(session.try_frontend_writer().unwrap().is_none());
    daemon.0.kill().unwrap();
    assert!(!daemon.0.wait().unwrap().success());
    frontend.0.kill().unwrap();
    assert!(!frontend.0.wait().unwrap().success());
    assert!(
        identity.socket.exists(),
        "crash must leave a stale endpoint"
    );
    assert_eq!(
        connect::connect(&identity.socket).unwrap_err().kind(),
        std::io::ErrorKind::ConnectionRefused
    );
    assert!(session.try_frontend_writer().unwrap().is_some());
    let events = session.read_events().unwrap();
    assert_eq!(
        events
            .iter()
            .filter(|event| event.kind() == Some(SessionEventKind::UserInput))
            .count(),
        1
    );
    assert!(
        !events.iter().any(|event| matches!(
            event.kind(),
            Some(
                SessionEventKind::TurnStatus
                    | SessionEventKind::AssistantOutput
                    | SessionEventKind::AssistantChunk
                    | SessionEventKind::ToolCall
                    | SessionEventKind::ToolResult
            )
        )),
        "a crash must not invent output or a durable terminal record"
    );
    let interrupted_history = fs::read(session.path()).unwrap();

    fs::remove_file(root.path().join("ready")).unwrap();
    // Unblock both execution and cleanup on restart: accidental replay cannot hide
    // behind the fixture gates or leave an unjoinable worker during shutdown.
    fs::write(root.path().join("release"), "").unwrap();
    fs::write(root.path().join("cleanup"), "").unwrap();
    let mut restarted = spawn_child(root.path(), "daemon");
    wait_until(|| root.path().join("ready").exists());
    let mut reconnected = Frontend::connect(&identity);
    assert_ne!(reconnected.instance, accepted["instance"]);
    let outcome = reconnected.call(
        "operation.lookup",
        Value::Null,
        Value::Null,
        Value::Null,
        json!({"target_instance_id":accepted["instance"],"operation_id":"surviving-turn"}),
    );
    assert_eq!(outcome["state"], "unknown");
    assert!(outcome["result"].is_null());
    let claimed = reconnected.call(
        "session.claim",
        accepted["session"].clone(),
        Value::Null,
        json!("claim-after-crash"),
        json!({}),
    );
    assert_eq!(claimed["snapshot"]["phase"], "idle");
    assert!(claimed["snapshot"]["turn"].is_null());
    assert!(claimed["snapshot"]["terminal"].is_null());
    drop(reconnected);
    wait_until(|| request(&identity, "stop").unwrap().status == "stopped");
    wait_until(|| restarted.0.try_wait().unwrap().is_some());
    assert!(restarted.0.wait().unwrap().success());
    assert!(!identity.socket.exists());
    assert_eq!(
        fs::read_to_string(root.path().join("provider-calls")).unwrap(),
        "0",
        "restart must not invoke a provider for the interrupted turn"
    );
    assert_eq!(
        fs::read(session.path()).unwrap(),
        interrupted_history,
        "restart and claim must preserve interrupted history without continuing or completing it"
    );
}