magi-code 0.77.1

#[cfg(test)]
use super::dispatcher::{ServiceDispatcher, ServiceSnapshot};
use super::{
    coordinator::{ServiceCoordinator, ServiceWorkerOutput},
    dispatcher::{ServiceOutbound, ServiceTransportCapabilities},
    protocol,
    runtime::ServiceRuntime,
    turns::TurnWorkerMessage,
};
use crossbeam_channel::{Receiver, Sender, bounded};
use std::io::{self, BufRead, BufReader, BufWriter, Write};
use std::sync::Arc;
use std::thread;

const STDIO_TRANSPORT: &str = "stdio";

pub(crate) fn run_stdio() -> anyhow::Result<()> {
    let runtime = Arc::new(ServiceRuntime::load().map_err(|_| {
        anyhow::anyhow!("application service startup failed: configuration unavailable")
    })?);
    let coordinator = ServiceCoordinator::new(
        runtime,
        ServiceTransportCapabilities::new(vec![STDIO_TRANSPORT.to_string()]),
    );
    let reader = BufReader::new(io::stdin());
    let writer = BufWriter::new(io::stdout());
    run_stdio_with_io(reader, writer, coordinator)
        .map(|_| ())
        .map_err(|_| anyhow::anyhow!("application service I/O failed"))
}

/// Run the service with an owned reader and writer.
///
/// The input reader is isolated on one bounded worker so the coordinator can write streamed turn
/// events while stdin is waiting for another request. On an I/O failure the reader is deliberately
/// detached: a blocking OS read cannot be interrupted portably, while the process is already on its
/// way out and active turns have been cancelled and drained first.
fn run_stdio_with_io<R: BufRead + Send + 'static, W: Write>(
    reader: R,
    writer: W,
    mut coordinator: ServiceCoordinator,
) -> io::Result<W> {
    let (input_sender, input_receiver) = bounded(super::turns::INPUT_QUEUE_CAPACITY);
    let reader_handle = thread::Builder::new()
        .name("magi-service-stdin".to_string())
        .spawn(move || feed_input(reader, input_sender))?;

    let mut writer = writer;
    let result = run_service_loop(&mut writer, &mut coordinator, input_receiver);
    if result.is_ok() {
        if reader_handle.join().is_err() {
            return Err(io::Error::other("service input reader panicked"));
        }
    } else {
        // See the function comment above. Dropping the handle detaches a reader that may be inside
        // a blocking stdin read; no service worker is left running by this path.
        drop(reader_handle);
    }
    result.map(|()| writer)
}

fn feed_input<R: BufRead>(mut reader: R, sender: Sender<InputRecord>) {
    loop {
        let record = match read_record(&mut reader, protocol::MAX_RECORD_BYTES) {
            Ok(record) => record,
            Err(_) => InputRecord::ReadError,
        };
        let done = matches!(record, InputRecord::Eof | InputRecord::ReadError);
        if sender.send(record).is_err() || done {
            return;
        }
    }
}

fn run_service_loop<W: Write>(
    writer: &mut W,
    coordinator: &mut ServiceCoordinator,
    input_receiver: Receiver<InputRecord>,
) -> io::Result<()> {
    loop {
        crossbeam_channel::select! {
            recv(input_receiver) -> record => {
                let record = match record {
                    Ok(record) => record,
                    Err(_) => {
                        return shutdown_active(
                            writer,
                            coordinator,
                            Some(io::Error::other("service input channel closed")),
                        );
                    }
                };
                match record {
                    InputRecord::Eof => return shutdown_active(writer, coordinator, None),
                    InputRecord::ReadError => {
                        return shutdown_active(
                            writer,
                            coordinator,
                            Some(io::Error::other("service input failed")),
                        );
                    }
                    InputRecord::Oversized => {
                        let outbound = coordinator.error_outbound(
                            protocol::RequestIdentity::default(),
                            protocol::ServiceErrorCode::RecordTooLarge,
                        );
                        if let Err(error) = write_messages(writer, outbound) {
                            return shutdown_active(writer, coordinator, Some(error));
                        }
                    }
                    InputRecord::Data(mut bytes) => {
                        strip_line_ending(&mut bytes);
                        let outbound = match protocol::decode_request(&bytes) {
                            Ok(request) => coordinator.dispatch_request(request),
                            Err(error) => coordinator.error_outbound(error.identity, error.code),
                        };
                        if let Err(error) = write_messages(writer, outbound) {
                            return shutdown_active(writer, coordinator, Some(error));
                        }
                    }
                }
            }
            recv(coordinator.auth_receiver()) -> message => {
                let message = message.map_err(|_| io::Error::other("service auth channel closed"))?;
                if let Some(outbound) = coordinator.auth_output(message)
                    && let Err(error) = write_messages(writer, outbound) {
                        return shutdown_active(writer, coordinator, Some(error));
                    }
            }
            recv(coordinator.worker_receiver()) -> message => {
                let message = match message {
                    Ok(message) => message,
                    Err(_) => {
                        return shutdown_active(
                            writer,
                            coordinator,
                            Some(io::Error::other("service turn channel closed")),
                        );
                    }
                };
                if let Err(error) = write_worker_message(writer, coordinator, message) {
                    return shutdown_active(writer, coordinator, Some(error));
                }
            }
        }
    }
}

fn write_worker_message<W: Write>(
    writer: &mut W,
    coordinator: &mut ServiceCoordinator,
    message: TurnWorkerMessage,
) -> io::Result<()> {
    let Some(ServiceWorkerOutput {
        outbound,
        terminal_turn_id,
    }) = coordinator.worker_output(message)
    else {
        return Ok(());
    };
    let result = write_messages(writer, outbound);
    if let Some(turn_id) = terminal_turn_id {
        // The coordinator marks a turn terminal before output. Releasing it after the write keeps
        // the request ID and session reservation through the terminal event.
        coordinator.finish_worker_output(&turn_id);
    }
    result
}

fn shutdown_active<W: Write>(
    writer: &mut W,
    coordinator: &mut ServiceCoordinator,
    mut first_error: Option<io::Error>,
) -> io::Result<()> {
    loop {
        let output = match coordinator.shutdown_next() {
            Ok(Some(output)) => output,
            Ok(None) => break,
            Err(_) => {
                if first_error.is_none() {
                    first_error = Some(io::Error::other("service turn channel closed"));
                }
                break;
            }
        };
        let ServiceWorkerOutput {
            outbound,
            terminal_turn_id,
        } = output;
        if first_error.is_none()
            && let Err(error) = write_messages(writer, outbound)
        {
            first_error = Some(error);
            coordinator.cancel_all();
        }
        if let Some(turn_id) = terminal_turn_id {
            coordinator.finish_worker_output(&turn_id);
        }
    }
    coordinator.join_all();
    match first_error {
        Some(error) => Err(error),
        None => writer.flush(),
    }
}

/// Write a complete outbound batch in order. The batch owns any request ID reservation until all
/// messages are encoded, written, and flushed; this keeps a correlated ID in flight through its
/// terminal event.
fn write_messages<W: Write>(writer: &mut W, outbound: ServiceOutbound) -> io::Result<()> {
    // Encode the whole batch before writing any of it. An invalid or oversized later message must
    // not leave a partial batch that could be mistaken for a complete correlated response.
    let encoded_messages = outbound
        .messages()
        .iter()
        .map(protocol::encode_message)
        .collect::<Result<Vec<_>, _>>()
        .map_err(|_| {
            io::Error::new(
                io::ErrorKind::InvalidData,
                protocol::ServiceErrorCode::InternalError.message(),
            )
        })?;

    for encoded in encoded_messages {
        writer.write_all(&encoded)?;
    }
    writer.flush()?;
    drop(outbound);
    Ok(())
}

enum InputRecord {
    Eof,
    Data(Vec<u8>),
    Oversized,
    ReadError,
}

fn read_record<R: BufRead>(reader: &mut R, max_bytes: usize) -> io::Result<InputRecord> {
    let mut bytes = Vec::with_capacity(max_bytes.min(4096));
    loop {
        let available = reader.fill_buf()?;
        if available.is_empty() {
            return if bytes.is_empty() {
                Ok(InputRecord::Eof)
            } else {
                Ok(InputRecord::Data(bytes))
            };
        }
        let take = available
            .iter()
            .position(|byte| *byte == b'\n')
            .map_or(available.len(), |index| index + 1);
        let ends_record = available[..take].last() == Some(&b'\n');
        if bytes
            .len()
            .checked_add(take)
            .is_none_or(|length| length > max_bytes)
        {
            reader.consume(take);
            if !ends_record {
                discard_record(reader)?;
            }
            return Ok(InputRecord::Oversized);
        }
        bytes.extend_from_slice(&available[..take]);
        reader.consume(take);
        if ends_record {
            return Ok(InputRecord::Data(bytes));
        }
    }
}

fn discard_record<R: BufRead>(reader: &mut R) -> io::Result<()> {
    loop {
        let available = reader.fill_buf()?;
        if available.is_empty() {
            return Ok(());
        }
        let take = available
            .iter()
            .position(|byte| *byte == b'\n')
            .map_or(available.len(), |index| index + 1);
        let ends_record = available[..take].last() == Some(&b'\n');
        reader.consume(take);
        if ends_record {
            return Ok(());
        }
    }
}

fn strip_line_ending(bytes: &mut Vec<u8>) {
    if bytes.last() == Some(&b'\n') {
        bytes.pop();
    }
    if bytes.last() == Some(&b'\r') {
        bytes.pop();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        config::{
            Auth, AuthProviderRecord, EffectiveConfig, McPaths, ProviderCredential, Settings,
            write_auth,
        },
        thinking::ThinkingLevel,
    };
    use serde_json::{Value, json};
    use std::{collections::BTreeMap, io::Cursor};
    use tempfile::TempDir;
    fn runtime(temp: &TempDir) -> Arc<ServiceRuntime> {
        let paths = McPaths::from_root(temp.path().join("mc"));
        Arc::new(ServiceRuntime {
            config: EffectiveConfig {
                provider: Some("test".to_string()),
                model: Some("test-model".to_string()),
                no_color: true,
                file_autocomplete_respects_gitignore: true,
                custom_providers: BTreeMap::new(),
                thinking_level: ThinkingLevel::Default,
                auth: Some(ProviderCredential::NoAuth),
                paths: paths.clone(),
            },
            settings: Settings::default(),
            instructions: Vec::new(),
            skills: crate::skills::SkillDiscovery::default(),
            session_manager: crate::sessions::SessionManager::new(paths.sessions),
            cwd: temp.path().to_path_buf(),
            mcp: None,
        })
    }

    fn codex_runtime(temp: &TempDir) -> Arc<ServiceRuntime> {
        let paths = McPaths::from_root(temp.path().join("mc"));
        paths.ensure_runtime_dirs().unwrap();
        let mut settings = Settings::default();
        settings.selected_model.provider =
            Some(crate::providers::OPENAI_CODEX_PROVIDER.to_string());
        settings.selected_model.model = Some("test-model".to_string());
        crate::config::write_settings(&paths, &settings).unwrap();
        Arc::new(ServiceRuntime {
            config: EffectiveConfig {
                provider: Some(crate::providers::OPENAI_CODEX_PROVIDER.to_string()),
                model: Some("test-model".to_string()),
                no_color: true,
                file_autocomplete_respects_gitignore: true,
                custom_providers: BTreeMap::new(),
                thinking_level: ThinkingLevel::Default,
                // A cached startup credential must not override current-store status.
                auth: Some(ProviderCredential::OAuth {
                    access: "cached-access".to_string(),
                    account_id: Some("cached-account".to_string()),
                }),
                paths: paths.clone(),
            },
            settings: Settings::default(),
            instructions: Vec::new(),
            skills: crate::skills::SkillDiscovery::default(),
            session_manager: crate::sessions::SessionManager::new(paths.sessions),
            cwd: temp.path().to_path_buf(),
            mcp: None,
        })
    }

    fn dispatch_value(
        coordinator: &mut ServiceCoordinator,
        id: &str,
        method: &str,
        payload: Value,
        session_id: Option<&str>,
    ) -> Value {
        let mut request = protocol::ServiceRequest::new(id, method, payload);
        request.session_id = session_id.map(str::to_string);
        let outbound = coordinator.dispatch_request(request);
        serde_json::to_value(outbound.messages().first().unwrap()).unwrap()
    }

    fn assert_status(value: &Value, id: &str, session_id: &str, ready: bool) {
        assert_eq!(value["protocol_version"], protocol::PROTOCOL_VERSION);
        assert_eq!(value["kind"], "response");
        assert_eq!(value["request_id"], id);
        assert_eq!(value["session_id"], session_id);
        assert_eq!(value["method"], "status");
        assert_eq!(value["payload"]["service"], "ready");
        assert_eq!(value["payload"]["provider_auth_ready"], ready);
        assert!(value["payload"].get("provider").is_none());
        assert!(value["payload"].get("model").is_none());
        assert!(value["error"].is_null());
    }

    fn coordinator(temp: &TempDir) -> ServiceCoordinator {
        ServiceCoordinator::new(
            runtime(temp),
            ServiceTransportCapabilities::new(vec![STDIO_TRANSPORT.to_string()]),
        )
    }

    fn request(id: &str, method: &str, payload: Value) -> String {
        serde_json::json!({
            "protocol_version": protocol::PROTOCOL_VERSION,
            "kind": "request",
            "request_id": id,
            "method": method,
            "payload": payload,
        })
        .to_string()
    }

    #[test]
    fn failed_negotiation_blocks_mutation_routes_before_workers_or_storage() {
        for (payload, code) in [
            (
                json!({"supported_protocol_versions": [99]}),
                "unsupported_version",
            ),
            (
                json!({"supported_protocol_versions": [1], "requested_capabilities": ["turn.activity"]}),
                "unsupported_capability",
            ),
        ] {
            let temp = TempDir::new().unwrap();
            let mut coordinator = coordinator(&temp);
            let rejected = dispatch_value(&mut coordinator, "init", "initialize", payload, None);
            assert_eq!(rejected["error"]["code"], code);
            for (method, payload, session_id) in [
                ("session.create", json!({}), None),
                ("session.open", json!({}), Some("session-test")),
                ("session.close", json!({}), Some("session-test")),
                ("turn.start", json!({"prompt": "synthetic prompt"}), None),
                (
                    "turn.cancel",
                    json!({"turn_id": "turn-test"}),
                    Some("session-test"),
                ),
                (
                    "auth.login.start",
                    json!({"provider_id": "openai-codex"}),
                    None,
                ),
                (
                    "auth.login.callback",
                    json!({"login_id": "login-test", "input": "synthetic callback"}),
                    None,
                ),
                ("auth.login.cancel", json!({"login_id": "login-test"}), None),
                (
                    "auth.logout",
                    json!({"provider_id": "openai-codex", "confirmed": true}),
                    None,
                ),
                ("catalog.refresh", json!({}), None),
                ("config.set", json!({"scope": "global", "fast": true}), None),
            ] {
                let rejected =
                    dispatch_value(&mut coordinator, "mutation", method, payload, session_id);
                assert_eq!(rejected["error"]["code"], "not_initialized", "{method}");
                assert_eq!(rejected["request_id"], "mutation");
                assert_eq!(rejected["method"], method);
                assert!(rejected["payload"].is_null());
            }
            assert!(coordinator.worker_receiver().is_empty());
            assert!(coordinator.auth_receiver().is_empty());
            assert!(!temp.path().join("mc").exists());
            assert!(coordinator.shutdown_next().unwrap().is_none());
        }
    }

    fn error_message(id: &str, code: protocol::ServiceErrorCode) -> protocol::ServiceMessage {
        protocol::ServiceMessage::response(protocol::ServiceResponse::error(
            protocol::RequestIdentity {
                request_id: Some(id.to_string()),
                ..protocol::RequestIdentity::default()
            },
            code,
        ))
    }

    fn event_message(id: &str, event: &str, payload: Value) -> protocol::ServiceMessage {
        protocol::ServiceMessage::Event(protocol::ServiceEvent {
            protocol_version: protocol::PROTOCOL_VERSION,
            kind: protocol::MessageKind::Event,
            event_id: format!("{event}-1"),
            request_id: id.to_string(),
            session_id: None,
            event: event.to_string(),
            payload,
        })
    }

    struct GuardCheckingWriter<'a> {
        dispatcher: &'a ServiceDispatcher,
        request_id: &'static str,
        output: Vec<u8>,
        checked_write: bool,
        checked_flush: bool,
    }

    impl std::io::Write for GuardCheckingWriter<'_> {
        fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
            assert!(matches!(
                self.dispatcher.begin_request(self.request_id),
                Err(protocol::ServiceErrorCode::DuplicateRequestId)
            ));
            self.checked_write = true;
            self.output.extend_from_slice(bytes);
            Ok(bytes.len())
        }

        fn flush(&mut self) -> std::io::Result<()> {
            assert!(matches!(
                self.dispatcher.begin_request(self.request_id),
                Err(protocol::ServiceErrorCode::DuplicateRequestId)
            ));
            self.checked_flush = true;
            Ok(())
        }
    }

    #[test]
    fn configuration_worker_is_bounded_and_retains_request_id_through_output() {
        let temp = TempDir::new().unwrap();
        let mut coordinator = coordinator(&temp);
        drop(coordinator.dispatch_request(protocol::ServiceRequest::new(
            "init",
            "initialize",
            json!({"supported_protocol_versions":[1]}),
        )));
        let pending = coordinator.dispatch_request(protocol::ServiceRequest::new(
            "read",
            "catalog.providers",
            json!({}),
        ));
        assert!(pending.messages().is_empty());
        let busy = dispatch_value(
            &mut coordinator,
            "other",
            "catalog.providers",
            json!({}),
            None,
        );
        assert_eq!(busy["error"]["code"], "configuration_busy");
        let message = coordinator
            .worker_receiver()
            .recv_timeout(std::time::Duration::from_secs(5))
            .unwrap();
        let output = coordinator.worker_output(message).unwrap();
        let duplicate = dispatch_value(
            &mut coordinator,
            "read",
            "catalog.providers",
            json!({}),
            None,
        );
        assert_eq!(duplicate["error"]["code"], "duplicate_request_id");
        drop(output);
        let pending = coordinator.dispatch_request(protocol::ServiceRequest::new(
            "read",
            "catalog.providers",
            json!({}),
        ));
        assert!(pending.messages().is_empty());
        assert!(coordinator.shutdown_next().unwrap().is_some());
        assert!(coordinator.shutdown_next().unwrap().is_none());
    }

    #[test]
    fn configuration_stdio_acknowledges_durable_write_and_reads_it_after_restart() {
        let env = crate::test_support::env::env_lock();
        let _provider = env.save("MC_PROVIDER");
        let _model = env.save("MC_MODEL");
        env.remove_var("MC_PROVIDER");
        env.remove_var("MC_MODEL");
        let temp = TempDir::new().unwrap();
        let runtime = runtime(&temp);
        let paths = &runtime.config.paths;
        paths.ensure_runtime_dirs().unwrap();
        std::fs::write(
            &paths.settings_file,
            r#"{"schema_version":2,"extension":{"api_key":"must-stay-local"}}"#,
        )
        .unwrap();
        crate::model_catalog::write_catalog_cache(
            paths,
            "anthropic",
            &[crate::model_catalog::ModelCatalogEntry::new(
                "anthropic",
                "claude-test",
            )],
        )
        .unwrap();
        for (method, payload) in [
            (
                "config.set",
                json!({"scope":"global", "selection":{"provider_id":"anthropic", "model_id":"claude-test"},"fast":true}),
            ),
            ("config.get", json!({})),
        ] {
            let input = format!(
                "{}\n{}\n",
                request(
                    "init",
                    "initialize",
                    json!({"supported_protocol_versions":[1]})
                ),
                request("configuration", method, payload)
            );
            let coordinator = ServiceCoordinator::new(
                Arc::clone(&runtime),
                ServiceTransportCapabilities::new(vec!["stdio".into()]),
            );
            let output =
                run_stdio_with_io(Cursor::new(input.into_bytes()), Vec::new(), coordinator)
                    .unwrap();
            let text = String::from_utf8(output).unwrap();
            assert!(!text.contains("must-stay-local"));
            assert!(!text.contains("api_key"));
            let records: Vec<Value> = text
                .lines()
                .map(|line| serde_json::from_str(line).unwrap())
                .collect();
            assert_eq!(records.len(), 2);
            assert_eq!(records[1]["request_id"], "configuration");
            assert!(records[1]["error"].is_null());
            let effective = if method == "config.set" {
                assert_eq!(records[1]["payload"]["persisted"], true);
                &records[1]["payload"]["effective"]
            } else {
                &records[1]["payload"]
            };
            assert_eq!(effective["selection"]["model_id"], "claude-test");
            assert_eq!(effective["fast"], true);
        }
        assert!(
            std::fs::read_to_string(&paths.settings_file)
                .unwrap()
                .contains("must-stay-local")
        );
    }

    #[test]
    fn bounded_reader_resynchronizes_after_an_oversized_record() {
        let mut input = vec![b'x'; protocol::MAX_RECORD_BYTES + 1];
        input.push(b'\n');
        input.extend_from_slice(
            request(
                "ok",
                "initialize",
                json!({
                    "supported_protocol_versions": [1]
                }),
            )
            .as_bytes(),
        );
        input.push(b'\n');
        let mut reader = Cursor::new(input);

        assert!(matches!(
            read_record(&mut reader, protocol::MAX_RECORD_BYTES).unwrap(),
            InputRecord::Oversized
        ));
        let InputRecord::Data(bytes) =
            read_record(&mut reader, protocol::MAX_RECORD_BYTES).unwrap()
        else {
            panic!("expected the bounded reader to continue after the oversized record");
        };
        let value: Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(value["request_id"], "ok");
    }

    #[test]
    fn stdio_loop_handles_crlf_and_final_unterminated_record() {
        let temp = TempDir::new().unwrap();
        let mut input = request(
            "init",
            "initialize",
            json!({
                "supported_protocol_versions": [1]
            }),
        );
        input.push_str("\r\nnot-json\r\n");
        input.push_str(&request("status", "status", json!({})));

        let output = run_stdio_with_io(
            Cursor::new(input.into_bytes()),
            Vec::new(),
            coordinator(&temp),
        )
        .unwrap();

        let lines = String::from_utf8(output)
            .unwrap()
            .lines()
            .map(|line| serde_json::from_str::<Value>(line).unwrap())
            .collect::<Vec<_>>();
        assert_eq!(lines.len(), 3);
        assert_eq!(lines[0]["request_id"], "init");
        assert_eq!(lines[1]["error"]["code"], "invalid_json");
        assert_eq!(lines[2]["request_id"], "status");
        assert_eq!(lines[2]["payload"]["provider_auth_ready"], false);
    }

    #[test]
    fn stdio_disconnect_settles_turn_and_next_client_replays_without_session_files() {
        let temp = TempDir::new().unwrap();
        let input = format!(
            "{}\n{}\n",
            request(
                "init",
                "initialize",
                json!({"supported_protocol_versions": [1]})
            ),
            request("turn", "turn.start", json!({"prompt": "offline question"})),
        );
        let output = run_stdio_with_io(
            Cursor::new(input.into_bytes()),
            Vec::new(),
            coordinator(&temp),
        )
        .unwrap();
        let messages = String::from_utf8(output)
            .unwrap()
            .lines()
            .map(|line| serde_json::from_str::<Value>(line).unwrap())
            .collect::<Vec<_>>();
        let accepted = messages
            .iter()
            .find(|message| message["method"] == "turn.start")
            .unwrap();
        let id = accepted["payload"]["session_id"].as_str().unwrap();
        assert_eq!(
            messages
                .iter()
                .filter(|message| message["event"] == "turn.terminal")
                .count(),
            1
        );
        // This client only uses the returned opaque ID; it never opens a session file.
        let mut input = format!(
            "{}\n{}\n",
            request(
                "init",
                "initialize",
                json!({"supported_protocol_versions": [1]})
            ),
            request("list", "session.list", json!({"limit": 32})),
        );
        for method in ["session.open", "session.replay", "session.close"] {
            let mut value: Value =
                serde_json::from_str(&request(method, method, json!({}))).unwrap();
            value["session_id"] = json!(id);
            input.push_str(&value.to_string());
            input.push('\n');
        }
        let output = run_stdio_with_io(
            Cursor::new(input.into_bytes()),
            Vec::new(),
            coordinator(&temp),
        )
        .unwrap();
        let text = String::from_utf8(output).unwrap();
        assert!(!text.contains(temp.path().to_str().unwrap()));
        assert!(!text.contains(".jsonl"));
        let messages = text
            .lines()
            .map(|line| serde_json::from_str::<Value>(line).unwrap())
            .collect::<Vec<_>>();
        assert_eq!(messages.len(), 5);
        assert!(messages.iter().all(|message| message["error"].is_null()));
        assert_eq!(messages[1]["payload"]["sessions"][0]["session_id"], id);
        assert_eq!(messages[2]["payload"]["status"], "open");
        let events = messages[3]["payload"]["events"].as_array().unwrap();
        assert!(events.iter().any(|event| event["kind"] == "terminal"));
        assert_eq!(messages[4]["payload"]["status"], "closed");
    }

    #[test]
    fn service_status_reads_current_auth_for_one_long_lived_runtime() {
        let env = crate::test_support::env::env_lock();
        let _provider = env.save("MC_PROVIDER");
        let _model = env.save("MC_MODEL");
        env.remove_var("MC_PROVIDER");
        env.remove_var("MC_MODEL");
        let temp = TempDir::new().unwrap();
        let runtime = codex_runtime(&temp);
        let paths = runtime.config.paths.clone();
        let mut coordinator = ServiceCoordinator::new(
            runtime,
            ServiceTransportCapabilities::new(vec![STDIO_TRANSPORT.to_string()]),
        );

        let initialize = dispatch_value(
            &mut coordinator,
            "init-1",
            "initialize",
            json!({"supported_protocol_versions": [1]}),
            None,
        );
        assert!(initialize["error"].is_null());

        let missing = dispatch_value(
            &mut coordinator,
            "status-missing",
            "status",
            json!({}),
            Some("session-1"),
        );
        assert_status(&missing, "status-missing", "session-1", false);

        let mut auth = Auth::default();
        auth.providers.insert(
            crate::providers::OPENAI_CODEX_PROVIDER.to_string(),
            AuthProviderRecord::OAuth {
                access: "access".to_string(),
                refresh: Some("refresh".to_string()),
                expires: Some(chrono::Utc::now().timestamp() + 3_600),
                account_id: Some("account".to_string()),
            },
        );
        write_auth(&paths, &auth).unwrap();
        let ready = dispatch_value(
            &mut coordinator,
            "status-ready",
            "status",
            json!({}),
            Some("session-1"),
        );
        assert_status(&ready, "status-ready", "session-1", true);

        auth.providers.insert(
            crate::providers::OPENAI_CODEX_PROVIDER.to_string(),
            AuthProviderRecord::OAuth {
                access: String::new(),
                refresh: Some("refresh".to_string()),
                expires: Some(1),
                account_id: None,
            },
        );
        write_auth(&paths, &auth).unwrap();
        let refreshable_auth_bytes = std::fs::read(&paths.auth_file).unwrap();
        let refreshable = dispatch_value(
            &mut coordinator,
            "status-refreshable",
            "status",
            json!({}),
            Some("session-1"),
        );
        assert_status(&refreshable, "status-refreshable", "session-1", true);
        assert_eq!(
            std::fs::read(&paths.auth_file).unwrap(),
            refreshable_auth_bytes
        );

        write_auth(&paths, &Auth::default()).unwrap();
        let removed = dispatch_value(
            &mut coordinator,
            "status-removed",
            "status",
            json!({}),
            Some("session-1"),
        );
        assert_status(&removed, "status-removed", "session-1", false);
    }

    #[test]
    fn service_status_sanitizes_shared_auth_read_failures_and_keeps_correlation() {
        let env = crate::test_support::env::env_lock();
        let _provider = env.save("MC_PROVIDER");
        let _model = env.save("MC_MODEL");
        env.remove_var("MC_PROVIDER");
        env.remove_var("MC_MODEL");
        let temp = TempDir::new().unwrap();
        let runtime = codex_runtime(&temp);
        let paths = runtime.config.paths.clone();
        let mut coordinator = ServiceCoordinator::new(
            runtime,
            ServiceTransportCapabilities::new(vec![STDIO_TRANSPORT.to_string()]),
        );
        let initialize = dispatch_value(
            &mut coordinator,
            "init-1",
            "initialize",
            json!({"supported_protocol_versions": [1]}),
            None,
        );
        assert!(initialize["error"].is_null());

        write_auth(&paths, &Auth::default()).unwrap();
        std::fs::write(&paths.auth_file, b"{\"malformed\":\"raw-auth-data\"").unwrap();
        let failure = dispatch_value(
            &mut coordinator,
            "status-read-failure",
            "status",
            json!({}),
            Some("session-1"),
        );

        assert_eq!(failure["protocol_version"], protocol::PROTOCOL_VERSION);
        assert_eq!(failure["kind"], "response");
        assert_eq!(failure["request_id"], "status-read-failure");
        assert_eq!(failure["session_id"], "session-1");
        assert_eq!(failure["method"], "status");
        assert!(failure["payload"].is_null());
        assert_eq!(failure["error"]["code"], "internal_error");
        assert_eq!(
            failure["error"]["message"],
            protocol::ServiceErrorCode::InternalError.message()
        );
        let encoded = failure.to_string();
        assert!(!encoded.contains("raw-auth-data"));
        assert!(!encoded.contains(&paths.auth_file.to_string_lossy().to_string()));
        assert!(!encoded.contains(crate::providers::OPENAI_CODEX_PROVIDER));
    }

    #[test]
    fn outbound_writer_preserves_every_message_in_order() {
        let mut output = Vec::new();
        write_messages(
            &mut output,
            ServiceOutbound::unguarded(vec![
                error_message("first", protocol::ServiceErrorCode::InvalidJson),
                error_message("second", protocol::ServiceErrorCode::InvalidPayload),
            ]),
        )
        .unwrap();

        let lines = String::from_utf8(output)
            .unwrap()
            .lines()
            .map(|line| serde_json::from_str::<Value>(line).unwrap())
            .collect::<Vec<_>>();
        assert_eq!(lines.len(), 2);
        assert_eq!(lines[0]["request_id"], "first");
        assert_eq!(lines[0]["error"]["code"], "invalid_json");
        assert_eq!(lines[1]["request_id"], "second");
        assert_eq!(lines[1]["error"]["code"], "invalid_payload");
    }

    #[test]
    fn outbound_batch_keeps_request_id_reserved_through_terminal_event_write() {
        let dispatcher = ServiceDispatcher::new(
            ServiceSnapshot::new(|| Ok(false)),
            ServiceTransportCapabilities::new(vec![STDIO_TRANSPORT.to_string()]),
        );
        let guard = dispatcher.begin_request("stream-1").unwrap();
        let outbound = ServiceOutbound::guarded(
            vec![
                error_message("stream-1", protocol::ServiceErrorCode::InvalidJson),
                event_message("stream-1", "terminal", json!({"done": true})),
            ],
            guard,
        );
        let mut writer = GuardCheckingWriter {
            dispatcher: &dispatcher,
            request_id: "stream-1",
            output: Vec::new(),
            checked_write: false,
            checked_flush: false,
        };

        write_messages(&mut writer, outbound).unwrap();

        assert!(writer.checked_write);
        assert!(writer.checked_flush);
        let lines = String::from_utf8(writer.output)
            .unwrap()
            .lines()
            .map(|line| serde_json::from_str::<Value>(line).unwrap())
            .collect::<Vec<_>>();
        assert_eq!(lines.len(), 2);
        assert_eq!(lines[0]["kind"], "response");
        assert_eq!(lines[1]["kind"], "event");
        assert!(lines.iter().all(|line| line["request_id"] == "stream-1"));
        assert!(dispatcher.begin_request("stream-1").is_ok());
    }

    #[test]
    fn oversized_correlated_response_fails_without_uncorrelated_fallback() {
        let mut response = protocol::ServiceResponse::error(
            protocol::RequestIdentity {
                request_id: Some("response-1".to_string()),
                session_id: Some("session-1".to_string()),
                method: Some("status".to_string()),
            },
            protocol::ServiceErrorCode::InvalidPayload,
        );
        response.error.as_mut().unwrap().message =
            "x".repeat(protocol::MAX_ERROR_MESSAGE_BYTES + 1);
        let mut output = Vec::new();

        let error = write_messages(
            &mut output,
            ServiceOutbound::unguarded(vec![protocol::ServiceMessage::response(response)]),
        )
        .unwrap_err();

        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert!(output.is_empty());
    }

    #[test]
    fn oversized_correlated_event_fails_before_writing_any_batch_message() {
        let text = "x".repeat(protocol::MAX_STRING_BYTES);
        let oversized_event = event_message(
            "event-request-1",
            "activity",
            json!({"values": [text.clone(), text.clone(), text]}),
        );
        let mut output = Vec::new();

        let error = write_messages(
            &mut output,
            ServiceOutbound::unguarded(vec![
                error_message("event-request-1", protocol::ServiceErrorCode::InvalidJson),
                oversized_event,
            ]),
        )
        .unwrap_err();

        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert!(output.is_empty());
    }

    #[test]
    fn auth_status_and_confirmed_logout_preserve_provider_boundaries() {
        let temp = TempDir::new().unwrap();
        let runtime = codex_runtime(&temp);
        let paths = runtime.config.paths.clone();
        let mut auth = Auth::default();
        auth.providers.insert(
            "openai-codex".into(),
            AuthProviderRecord::OAuth {
                access: "synthetic-access-secret".into(),
                refresh: Some("synthetic-refresh-secret".into()),
                expires: Some(0),
                account_id: Some("synthetic-account-secret".into()),
            },
        );
        auth.providers.insert(
            "anthropic".into(),
            AuthProviderRecord::ApiKey {
                key: "synthetic-api-secret".into(),
            },
        );
        write_auth(&paths, &auth).unwrap();
        let mut coordinator = ServiceCoordinator::new(
            runtime,
            ServiceTransportCapabilities::new(vec!["stdio".into()]),
        );
        assert_eq!(
            dispatch_value(&mut coordinator, "before", "auth.status", json!({}), None)["error"]["code"],
            "not_initialized"
        );
        dispatch_value(
            &mut coordinator,
            "init",
            "initialize",
            json!({"supported_protocol_versions":[1]}),
            None,
        );
        let status = dispatch_value(&mut coordinator, "status", "auth.status", json!({}), None);
        assert_eq!(status["payload"]["providers"][0]["readiness"], "ready");
        let encoded = status.to_string();
        for secret in [
            "synthetic-access-secret",
            "synthetic-refresh-secret",
            "synthetic-account-secret",
            "synthetic-api-secret",
            "account_id",
            "auth.json",
        ] {
            assert!(!encoded.contains(secret));
        }
        let denied = dispatch_value(
            &mut coordinator,
            "deny",
            "auth.logout",
            json!({"provider_id":"openai-codex"}),
            None,
        );
        assert_eq!(denied["error"]["code"], "confirmation_required");
        assert_eq!(crate::config::read_auth(&paths).unwrap(), auth);
        let removed = dispatch_value(
            &mut coordinator,
            "logout",
            "auth.logout",
            json!({"provider_id":"openai-codex", "confirmed":true}),
            None,
        );
        assert_eq!(removed["payload"]["removed"], true);
        let stored = crate::config::read_auth(&paths).unwrap();
        assert!(!stored.providers.contains_key("openai-codex"));
        assert_eq!(
            stored.providers.get("anthropic"),
            auth.providers.get("anthropic")
        );
        assert_eq!(
            dispatch_value(&mut coordinator, "after", "auth.status", json!({}), None)["payload"]["providers"]
                [0]["readiness"],
            "missing"
        );
    }

    #[test]
    fn auth_login_url_callback_failure_and_cancel_are_correlated_and_joined() {
        let temp = TempDir::new().unwrap();
        let mut coordinator = coordinator(&temp);
        dispatch_value(
            &mut coordinator,
            "init",
            "initialize",
            json!({"supported_protocol_versions":[1]}),
            None,
        );
        for (id, cancel) in [("login-failure", false), ("login-cancel", true)] {
            let start = dispatch_value(
                &mut coordinator,
                id,
                "auth.login.start",
                json!({"provider_id":"openai-codex"}),
                None,
            );
            let login_id = start["payload"]["login_id"].as_str().unwrap();
            assert_eq!(
                dispatch_value(
                    &mut coordinator,
                    "busy",
                    "auth.login.start",
                    json!({"provider_id":"openai-codex"}),
                    None
                )["error"]["code"],
                "auth_busy"
            );
            assert_eq!(
                dispatch_value(&mut coordinator, id, "auth.status", json!({}), None)["error"]["code"],
                "duplicate_request_id"
            );
            for method in ["auth.login.callback", "auth.login.cancel"] {
                let payload = if method == "auth.login.callback" {
                    json!({"login_id": "other-login", "input": "synthetic callback"})
                } else {
                    json!({"login_id": "other-login"})
                };
                assert_eq!(
                    dispatch_value(&mut coordinator, "wrong-login", method, payload, None)["error"]
                        ["code"],
                    "unknown_login"
                );
            }
            let mut states = Vec::new();
            for _ in 0..2 {
                let message = coordinator
                    .auth_receiver()
                    .recv_timeout(std::time::Duration::from_secs(2))
                    .unwrap();
                let output = coordinator.auth_output(message).unwrap();
                let value = serde_json::to_value(&output.messages()[0]).unwrap();
                assert_eq!(value["request_id"], id);
                states.push(value["payload"]["state"].as_str().unwrap().to_string());
                if value["payload"]["state"] == "authorization_url" {
                    assert!(
                        value["payload"]["url"]
                            .as_str()
                            .unwrap()
                            .starts_with("https://auth.openai.com/oauth/authorize?")
                    );
                    assert_eq!(value["payload"]["open_browser"], true);
                }
            }
            let result = if cancel {
                dispatch_value(
                    &mut coordinator,
                    "cancel",
                    "auth.login.cancel",
                    json!({"login_id":login_id}),
                    None,
                )
            } else {
                dispatch_value(
                    &mut coordinator,
                    "callback",
                    "auth.login.callback",
                    json!({"login_id":login_id,"input":"http://localhost:1455/auth/callback?code=secret-code&state=wrong"}),
                    None,
                )
            };
            assert!(result["error"].is_null());
            assert!(!result.to_string().contains("secret-code"));
            loop {
                let message = coordinator
                    .auth_receiver()
                    .recv_timeout(std::time::Duration::from_secs(2))
                    .unwrap();
                let output = coordinator.auth_output(message).unwrap();
                let value = serde_json::to_value(&output.messages()[0]).unwrap();
                assert_eq!(value["request_id"], id);
                assert_eq!(value["payload"]["login_id"], login_id);
                assert_eq!(value["payload"]["provider_id"], "openai-codex");
                assert!(value["session_id"].is_null());
                assert!(!value.to_string().contains("secret-code"));
                states.push(value["payload"]["state"].as_str().unwrap().to_string());
                if value["event"] == "auth.login.terminal" {
                    assert_eq!(value["payload"]["cleanup_complete"], true);
                    assert_eq!(
                        dispatch_value(&mut coordinator, id, "auth.status", json!({}), None)["error"]
                            ["code"],
                        "duplicate_request_id"
                    );
                    drop(output);
                    break;
                }
            }
            if cancel {
                assert_eq!(states, ["authorization_url", "waiting", "cancelled"]);
            } else {
                assert_eq!(
                    states,
                    [
                        "authorization_url",
                        "waiting",
                        "callback_received",
                        "callback_failed"
                    ]
                );
            }
            assert_eq!(
                dispatch_value(
                    &mut coordinator,
                    "stale-login",
                    "auth.login.cancel",
                    json!({"login_id": login_id}),
                    None
                )["error"]["code"],
                "unknown_login"
            );
            assert!(
                dispatch_value(&mut coordinator, id, "auth.status", json!({}), None)["error"]
                    .is_null()
            );
        }
        assert!(coordinator.shutdown_next().unwrap().is_none());
    }

    #[test]
    fn stdio_eof_cancels_auth_worker_and_writes_one_terminal() {
        let temp = TempDir::new().unwrap();
        let input = format!(
            "{}\n{}\n",
            request(
                "init",
                "initialize",
                json!({"supported_protocol_versions":[1]})
            ),
            request(
                "start",
                "auth.login.start",
                json!({"provider_id":"openai-codex"})
            )
        );
        let mut output = Vec::new();
        run_stdio_with_io(Cursor::new(input), &mut output, coordinator(&temp)).unwrap();
        let records: Vec<Value> = String::from_utf8(output)
            .unwrap()
            .lines()
            .map(|line| serde_json::from_str(line).unwrap())
            .collect();
        let terminal: Vec<_> = records
            .iter()
            .filter(|record| record["event"] == "auth.login.terminal")
            .collect();
        assert_eq!(terminal.len(), 1);
        assert_eq!(terminal[0]["payload"]["state"], "cancelled");
        assert_eq!(terminal[0]["payload"]["cleanup_complete"], true);
    }

    #[test]
    fn auth_status_reports_environment_and_no_auth_without_exposing_values() {
        let env = crate::test_support::env::env_lock();
        let _anthropic = env.save("ANTHROPIC_API_KEY");
        let _custom = env.save("MC_SERVICE_AUTH_TEST_KEY");
        env.set_var("ANTHROPIC_API_KEY", "synthetic-anthropic-secret");
        env.set_var("MC_SERVICE_AUTH_TEST_KEY", "synthetic-custom-secret");
        let temp = TempDir::new().unwrap();
        let mut runtime = runtime(&temp);
        let config = &mut Arc::get_mut(&mut runtime).unwrap().config;
        config.custom_providers.insert(
            "local".into(),
            crate::config::make_custom_provider_config("Local", "http://localhost:9999/v1", "")
                .unwrap(),
        );
        config.custom_providers.insert(
            "remote".into(),
            crate::config::make_custom_provider_config(
                "Remote",
                "https://example.test/v1",
                "MC_SERVICE_AUTH_TEST_KEY",
            )
            .unwrap(),
        );
        let paths = config.paths.clone();
        let mut coordinator = ServiceCoordinator::new(
            runtime,
            ServiceTransportCapabilities::new(vec!["stdio".into()]),
        );
        dispatch_value(
            &mut coordinator,
            "init",
            "initialize",
            json!({"supported_protocol_versions":[1]}),
            None,
        );
        let status = dispatch_value(&mut coordinator, "status", "auth.status", json!({}), None);
        let providers = status["payload"]["providers"].as_array().unwrap();
        assert_eq!(providers[1]["source"], "environment");
        assert_eq!(providers[1]["ready"], true);
        assert_eq!(providers[2]["auth_method"], "none");
        assert_eq!(providers[2]["ready"], true);
        assert_eq!(providers[3]["source"], "environment");
        assert_eq!(providers[3]["ready"], true);
        for forbidden in [
            "synthetic-anthropic-secret",
            "synthetic-custom-secret",
            "MC_SERVICE_AUTH_TEST_KEY",
            "http://localhost",
        ] {
            assert!(!status.to_string().contains(forbidden));
        }
        env.remove_var("MC_SERVICE_AUTH_TEST_KEY");
        assert_eq!(
            dispatch_value(&mut coordinator, "missing", "auth.status", json!({}), None)["payload"]
                ["providers"][3]["readiness"],
            "missing"
        );
        std::fs::create_dir_all(&paths.root).unwrap();
        std::fs::write(&paths.auth_file, "malformed-synthetic-secret").unwrap();
        let failed = dispatch_value(&mut coordinator, "failed", "auth.status", json!({}), None);
        assert_eq!(failed["error"]["code"], "internal_error");
        assert!(!failed.to_string().contains("malformed-synthetic-secret"));
    }
}