anyagent 0.0.3

//! The ACP adapter driven end to end through the public interface, against
//! the recorded fixture agent (tests/fixtures/acp/fixture.mjs; needs `node`).
//! The catalog wrappers are `sh` scripts, so the file is unix-only for now.

use std::path::Path;
use std::time::Duration;

use futures::StreamExt;

use anyagent::{
    AgentError, AgentInstallation, Answer, AuthKind, AuthStatus, Capability, ChoiceId, ConfigId,
    ConfigValue, DeliveryKind, Event, EventKind, Events, Input, LoginMethod, McpServer,
    McpTransport, PermissionChoice, QuestionAnswer, Request, ResumeToken, Runtime, Session,
    SessionOptions, StopReason,
};

mod common;

fn fixture(extra: &[&str]) -> AgentInstallation {
    let script = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/acp/fixture.mjs");
    let mut args = vec![script.to_string_lossy().into_owned()];
    args.extend(extra.iter().map(|s| s.to_string()));
    AgentInstallation::acp("fixture", "node", args)
}

async fn open(extra: &[&str]) -> (Session, Events) {
    let runtime = Runtime::new();
    runtime
        .open(
            &fixture(extra),
            SessionOptions::in_dir(std::env::temp_dir()),
        )
        .await
        .unwrap()
}

async fn next(events: &mut Events) -> Event {
    tokio::time::timeout(Duration::from_secs(10), events.next())
        .await
        .expect("timed out waiting for an event")
        .expect("stream ended")
        .expect("stream error")
}

fn allow() -> Answer {
    Answer::Permission(PermissionChoice::AllowOnce)
}

/// A catalog-agent stand-in: a script that execs the fixture with scenario
/// flags, ignoring the profile's protocol args appended after them. Needed
/// because auth truth (open-proves-auth, logged-out hints) is catalog data.
fn catalog_wrapper(agent: &str, name: &str, flags: &str) -> AgentInstallation {
    AgentInstallation::at(agent, common::wrapper("acp", agent, name, flags))
}

/// Full turn maps every ACP update kind: text, reasoning, tool, plan, usage, SessionUpdated, permission request, and agent-originated turn.
#[tokio::test]
async fn a_full_turn_maps_every_update_kind() {
    let (session, mut events) = open(&[]).await;
    session.prompt("hi").await.unwrap();

    let mut text = String::new();
    let mut thoughts = String::new();
    let mut open_messages = std::collections::BTreeSet::new();
    let mut tool_states = Vec::new();
    let mut plan_steps = Vec::new();
    let mut usage = None;
    let mut diagnostics = 0;
    let mut raw_update = false;
    let mut commands_seen = false;
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::TextDelta {
                text: t,
                message_id,
            } => {
                text.push_str(&t);
                open_messages.insert(message_id);
            }
            EventKind::ReasoningDelta {
                text: t,
                message_id,
            } => {
                thoughts.push_str(&t);
                open_messages.insert(message_id);
            }
            EventKind::MessageEnded { message_id } => {
                open_messages.remove(&message_id);
            }
            EventKind::ToolUpdated(tool) => tool_states.push(tool),
            EventKind::PlanUpdated { entries } => {
                plan_steps = entries.into_iter().map(|e| e.text).collect();
            }
            EventKind::ContextUsage {
                used_tokens,
                window_tokens,
                cost_usd,
            } => usage = Some((used_tokens, window_tokens, cost_usd)),
            EventKind::SessionUpdated(info) => {
                commands_seen |= info.details.commands.iter().any(|c| c.name == "compact");
            }
            EventKind::Diagnostic(_) => {
                diagnostics += 1;
                raw_update |= event.extensions.contains_key("acp/raw_update");
            }
            EventKind::RequestOpened(Request::Permission(request)) => {
                assert_eq!(request.tool.title, "Run tests");
                assert_eq!(
                    request.options,
                    vec![PermissionChoice::AllowOnce, PermissionChoice::DenyOnce]
                );
                session.answer(request.id, allow()).await.unwrap();
            }
            EventKind::TurnEnded { stop, .. } => {
                assert_eq!(
                    stop,
                    StopReason::Completed {
                        source: anyagent::CompletionSource::Protocol
                    }
                );
                // ACP has no end-of-message signal; the engine closes every
                // streamed message before ending the turn.
                assert!(open_messages.is_empty(), "unended: {open_messages:?}");
                break;
            }
            _ => {}
        }
    }
    assert_eq!(text, "Hello perm=selected ");
    assert_eq!(thoughts, "thinking…");
    assert_eq!(plan_steps, vec!["step 1"]);
    assert_eq!(usage, Some((1200, Some(200_000), Some(0.01))));
    assert!(commands_seen, "available commands became SessionUpdated");
    assert!(raw_update, "unknown update kind kept raw in extensions");
    assert!(diagnostics >= 2, "unknown kind + extension notification");

    // The edit tool: pending with a diff, then completed with output.
    let edit: Vec<_> = tool_states
        .iter()
        .filter(|t| t.title == "Edit main.rs")
        .collect();
    assert_eq!(edit[0].diffs[0].new_text, "b");
    assert_eq!(edit[0].locations, vec![std::path::PathBuf::from("main.rs")]);
    assert!(edit.last().unwrap().output.as_deref() == Some("done"));

    // Late noise after the turn opens an agent-originated turn.
    loop {
        if let EventKind::TurnStarted { origin } = next(&mut events).await.kind {
            assert_eq!(origin, anyagent::TurnOrigin::Agent);
            break;
        }
    }
    session.close().await.unwrap();
}

/// Handshake fills SessionInfo with version 0.0.1, capabilities, config options, and resume token.
#[tokio::test]
async fn the_handshake_fills_session_info() {
    let (session, _events) = open(&[]).await;
    let info = session.info();
    assert_eq!(info.details.version.as_deref(), Some("0.0.1"));
    for capability in [
        Capability::Permissions,
        Capability::Steer,
        Capability::Images,
        Capability::Resume,
    ] {
        assert!(info.details.capabilities.supports(capability));
    }
    let ids: Vec<_> = info
        .details
        .config_options
        .iter()
        .map(|o| o.id.as_str().to_owned())
        .collect();
    assert_eq!(ids, vec!["mode", "model"]);
    assert_eq!(info.resume_token.unwrap().as_str(), "sess-1");
    session.close().await.unwrap();
}

/// Probe waits for late available-commands SessionUpdated instead of returning empty commands.
#[tokio::test]
async fn probe_reports_details_and_waits_for_late_commands() {
    let details = Runtime::new()
        .probe(&fixture(&["--commands-on-open"]))
        .await
        .unwrap();
    // The same details an open reports: version and config options.
    assert_eq!(details.version.as_deref(), Some("0.0.1"));
    let ids: Vec<_> = details
        .config_options
        .iter()
        .map(|o| o.id.as_str().to_owned())
        .collect();
    assert_eq!(ids, vec!["mode", "model"]);
    // The command list only arrives as a `SessionUpdated` after session/new;
    // seeing it proves probe waited for it instead of returning empty.
    assert!(
        details.commands.iter().any(|c| c.name == "compact"),
        "probe must wait for the ACP available-commands update"
    );
}

/// Cancel mid-permission reaches the agent and ends the turn as Cancelled.
#[tokio::test]
async fn cancel_reaches_the_agent_and_ends_the_turn() {
    let (session, mut events) = open(&[]).await;
    session.prompt("hi").await.unwrap();
    loop {
        if let EventKind::RequestOpened(_) = next(&mut events).await.kind {
            break;
        }
    }
    session.cancel(false).await.unwrap();
    loop {
        if let EventKind::TurnEnded { stop, .. } = next(&mut events).await.kind {
            assert_eq!(stop, StopReason::Cancelled);
            break;
        }
    }
    session.close().await.unwrap();
}

/// Mid-turn steer is accepted and delivered as Steered with same turn_id.
#[tokio::test]
async fn steering_mid_turn_is_accepted() {
    let (session, mut events) = open(&[]).await;
    let first = session.prompt("one").await.unwrap();
    let DeliveryKind::Started { turn_id } = first.kind else {
        panic!("expected Started");
    };
    let steered = session.prompt("two").await.unwrap();
    assert_eq!(steered.kind, DeliveryKind::Steered { turn_id });
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::RequestOpened(request) => {
                session.answer(request.id(), allow()).await.unwrap()
            }
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    session.close().await.unwrap();
}

/// Attachments inline PNG as image block, PDF/missing as path refs, unreadable emits diagnostic.
#[tokio::test]
async fn attachments_inline_images_and_reference_paths() {
    let dir = std::env::temp_dir().join(format!("anyagent-att-acp-{}", std::process::id()));
    std::fs::create_dir_all(&dir).unwrap();
    std::fs::write(dir.join("shot.png"), b"\x89PNG\r\n\x1a\ndata").unwrap();
    std::fs::write(dir.join("report.pdf"), b"%PDF-1.7 data").unwrap();
    let (session, mut events) = open(&[]).await;
    session
        .prompt(
            Input::text("look")
                .attach(dir.join("shot.png"))
                .attach(dir.join("report.pdf"))
                .attach(dir.join("missing.png")),
        )
        .await
        .unwrap();
    let mut text = String::new();
    let mut unreadable = 0;
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::TextDelta { text: t, .. } => text.push_str(&t),
            EventKind::Diagnostic(d) => {
                unreadable += usize::from(d.message.contains("unreadable"));
            }
            EventKind::RequestOpened(request) => {
                session.answer(request.id(), allow()).await.unwrap()
            }
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    // One inlined image; the pdf (not advertised) and the unreadable file
    // ride as path refs only.
    assert!(
        text.contains("att=1 aud=0 res=0 ref=1"),
        "wire shape wrong: {text:?}"
    );
    assert_eq!(unreadable, 1);
    session.close().await.unwrap();
}

/// Audio and PDF attachments ride as `audio` and embedded `resource` blocks
/// once the agent advertises those prompt capabilities.
#[tokio::test]
async fn attachments_inline_audio_and_pdf_when_advertised() {
    let dir = std::env::temp_dir().join(format!("anyagent-att-media-{}", std::process::id()));
    std::fs::create_dir_all(&dir).unwrap();
    std::fs::write(dir.join("shot.png"), b"\x89PNG\r\n\x1a\ndata").unwrap();
    std::fs::write(dir.join("voice.wav"), b"RIFF\x00\x00\x00\x00WAVEfmt data").unwrap();
    std::fs::write(dir.join("report.pdf"), b"%PDF-1.7 data").unwrap();
    let (session, mut events) = open(&["--media"]).await;
    session
        .prompt(
            Input::text("listen")
                .attach(dir.join("shot.png"))
                .attach(dir.join("voice.wav"))
                .attach(dir.join("report.pdf")),
        )
        .await
        .unwrap();
    let mut text = String::new();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::TextDelta { text: t, .. } => text.push_str(&t),
            EventKind::RequestOpened(request) => {
                session.answer(request.id(), allow()).await.unwrap()
            }
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    assert!(
        text.contains("att=1 aud=1 res=1 ref=1"),
        "wire shape wrong: {text:?}"
    );
    session.close().await.unwrap();
}

/// Agent exit mid-turn fails the turn and surfaces ProcessExited with stderr.
#[tokio::test]
async fn agent_death_mid_turn_fails_the_turn() {
    let (session, mut events) = open(&["--eof"]).await;
    session.prompt("hi").await.unwrap();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::RequestOpened(request) => {
                session.answer(request.id(), allow()).await.unwrap()
            }
            EventKind::TurnEnded { stop, .. } => {
                assert!(matches!(stop, StopReason::Failed { .. }));
                break;
            }
            _ => {}
        }
    }
    let error = tokio::time::timeout(Duration::from_secs(10), events.next())
        .await
        .unwrap()
        .unwrap();
    match error {
        Err(AgentError::ProcessExited { status, stderr }) => {
            assert_eq!(status, "exit status: 3");
            assert!(stderr.contains("boom"), "stderr not carried: {stderr:?}");
        }
        other => panic!("expected ProcessExited, got {other:?}"),
    }
}

/// Configure mode round-trips via SessionUpdated; invalid value/unknown id rejected typed locally.
#[tokio::test]
async fn configuring_the_mode_round_trips_and_updates_the_session() {
    let (session, mut events) = open(&[]).await;
    session.configure("mode", "plan").await.unwrap();
    loop {
        let event = next(&mut events).await;
        if let EventKind::SessionUpdated(info) = event.kind {
            assert_eq!(
                info.configuration.options.get(&ConfigId::new("mode")),
                Some(&ConfigValue::Text("plan".into()))
            );
            break;
        }
    }
    // A value the agent never offered, and an option it never advertised,
    // are typed rejections that never reach the wire.
    let bad_value = session.configure("mode", "yolo").await;
    assert!(matches!(
        bad_value,
        Err(AgentError::InvalidConfiguration(_))
    ));
    let unknown = session.configure("nope", "x").await;
    assert!(matches!(unknown, Err(AgentError::InvalidConfiguration(_))));
    session.close().await.unwrap();
}

/// Kiro probe: successful open proves login -> Authenticated Subscription.
#[tokio::test]
async fn probe_confirms_login_when_an_open_proves_it() {
    // kiro refuses to open logged out (probed 2026-08-28), so a successful
    // open is proof of login.
    let agent = catalog_wrapper("kiro", "auth-in", "--commands-on-open");
    let details = Runtime::new().probe(&agent).await.unwrap();
    assert_eq!(
        details.auth,
        AuthStatus::Authenticated {
            kind: AuthKind::Subscription,
            account: None
        }
    );
}

/// ACP cannot say which credential opened the session: with the profile's documented API key in the env, a proven open is reported as an API-key login, not the catalog's subscription.
#[tokio::test]
async fn an_api_key_in_the_env_is_reported_as_one() {
    // std guards env access with its own lock, and no other test in this
    // binary reads XAI_API_KEY.
    unsafe { std::env::set_var("XAI_API_KEY", "xai-test") };
    let agent = catalog_wrapper("grok", "api-key", "");
    let details = Runtime::new().probe(&agent).await.unwrap();
    assert_eq!(
        details.auth,
        AuthStatus::Authenticated {
            kind: AuthKind::ApiKey,
            account: None
        }
    );
}

/// Pre-protocol exit (kiro not logged in) mapped to Unauthenticated with terminal login method.
#[tokio::test]
async fn probe_maps_a_pre_protocol_exit_to_logged_out() {
    // The kiro logged-out shape: complain on stderr, exit before speaking ACP.
    let agent = catalog_wrapper("kiro", "auth-exit", "--die-not-logged-in");
    let details = Runtime::new().probe(&agent).await.unwrap();
    let AuthStatus::Unauthenticated { login } = details.auth else {
        panic!("expected Unauthenticated, got {:?}", details.auth);
    };
    let LoginMethod::Terminal { command, .. } = &login[0] else {
        panic!("expected a terminal login method");
    };
    assert_eq!(command[1..], ["login"]);
}

/// Session/new internal error hinted 'No LLM provider' mapped to Unauthenticated.
#[tokio::test]
async fn probe_maps_a_hinted_wire_error_to_logged_out() {
    // The hermes logged-out shape: session/new fails with a plain internal
    // error whose `data` carries "No LLM provider configured".
    let agent = catalog_wrapper("hermes", "auth-hint", "--auth-hint-error");
    let details = Runtime::new().probe(&agent).await.unwrap();
    let AuthStatus::Unauthenticated { login } = details.auth else {
        panic!("expected Unauthenticated, got {:?}", details.auth);
    };
    let LoginMethod::Terminal { command, .. } = &login[0] else {
        panic!("expected a terminal login method");
    };
    assert_eq!(command[1..], ["login"]);
}

/// Empty auth methods fall back to profile executable + EnvVar XAI_API_KEY alternative.
#[tokio::test]
async fn probe_uses_the_profile_fallback_when_auth_methods_are_empty() {
    // Grok returns the auth code without a runnable method. Its profile maps
    // the error and offers the bare executable, which opens Grok's login TUI.
    let agent = catalog_wrapper(
        "grok",
        "auth-code",
        "--auth-required --capitalized-auth --no-auth-methods",
    );
    let details = Runtime::new().probe(&agent).await.unwrap();
    let AuthStatus::Unauthenticated { login } = details.auth else {
        panic!("expected Unauthenticated, got {:?}", details.auth);
    };
    let LoginMethod::Terminal { command, .. } = &login[0] else {
        panic!("expected a terminal login method");
    };
    assert_eq!(command, &[agent.executable_path.to_string_lossy()]);
    // The profile's documented API key rides along as the alternative.
    assert!(
        login
            .iter()
            .any(|m| matches!(m, LoginMethod::EnvVar { name } if name == "XAI_API_KEY")),
        "expected the XAI_API_KEY EnvVar alternative, got {login:?}"
    );
}

/// Meta-shaped auth method still yields runnable terminal login command.
#[tokio::test]
async fn probe_reads_a_meta_shaped_login_method() {
    // The qwen shape: the auth method's type/args ride in `_meta`, so the
    // typed Terminal variant never matches — the meta shape must still
    // yield a runnable login command.
    let details = Runtime::new()
        .probe(&fixture(&["--auth-required", "--meta-auth-methods"]))
        .await
        .unwrap();
    let AuthStatus::Unauthenticated { login } = details.auth else {
        panic!("expected Unauthenticated, got {:?}", details.auth);
    };
    let LoginMethod::Terminal { command, .. } = &login[0] else {
        panic!("expected a terminal login method");
    };
    assert_eq!(
        command.last().map(String::as_str),
        Some("--auth-type=openai")
    );
}

/// Two overlapping configures (slow model reply) both apply, not just the last.
#[tokio::test]
async fn overlapping_configures_both_apply() {
    // The model reply is delayed past the mode round-trip, so both changes
    // are in flight at once; each confirmation must land, not just the last.
    let (session, mut events) = open(&["--config-slow=150"]).await;
    session.configure("model", "opus").await.unwrap();
    session.configure("mode", "plan").await.unwrap();
    let (mut mode, mut model) = (false, false);
    while !(mode && model) {
        let event = next(&mut events).await;
        if let EventKind::SessionUpdated(info) = event.kind {
            let get = |id: &str| info.configuration.options.get(&ConfigId::new(id)).cloned();
            mode |= get("mode") == Some(ConfigValue::Text("plan".into()));
            model |= get("model") == Some(ConfigValue::Text("opus".into()));
        }
    }
    session.close().await.unwrap();
}

/// A thought-level option under another name (qwen: `reasoning_effort`) is advertised as `effort`, and a switch reaches the wire under the agent's own id; a replacement list that names it plainly resets that.
#[tokio::test]
async fn a_thought_level_option_under_another_name_is_effort() {
    let (session, mut events) = open(&["--qwen"]).await;
    let effort = option(&session, "effort").expect("an effort option");
    assert_eq!(effort.category.as_deref(), Some("thought_level"));
    assert_eq!(effort.current, Some(ConfigValue::Text("default".into())));
    assert!(
        option(&session, "reasoning_effort").is_none(),
        "the wire id leaked"
    );
    // The fixture only knows `reasoning_effort`: the switch applying
    // proves the wire id was mapped back.
    session.configure("effort", "high").await.unwrap();
    wait_options(&session, &mut events, |s| {
        option(s, "effort").and_then(|o| o.current) == Some(ConfigValue::Text("high".into()))
    })
    .await;
    // A model whose list names it `effort` outright: no stale wire id.
    session.configure("model", "opus").await.unwrap();
    wait_options(&session, &mut events, |s| {
        option(s, "model").and_then(|o| o.current) == Some(ConfigValue::Text("opus".into()))
    })
    .await;
    session.configure("effort", "default").await.unwrap();
    wait_options(&session, &mut events, |s| {
        option(s, "effort").and_then(|o| o.current) == Some(ConfigValue::Text("default".into()))
    })
    .await;
}

/// MCP http+stdio servers forwarded; capabilities advertised and wire contains http:voice/stdio:tool.
#[tokio::test]
async fn mcp_servers_forward_when_the_transport_is_supported() {
    let runtime = Runtime::new();
    let (session, mut events) = runtime
        .open(
            &fixture(&[]),
            SessionOptions::in_dir(std::env::temp_dir())
                .mcp_server(
                    McpServer::http("voice", "http://127.0.0.1:1/mcp")
                        .with("Authorization", "Bearer x"),
                )
                .mcp_server(McpServer::stdio("tool", "/bin/echo", ["hi"])),
        )
        .await
        .unwrap();
    assert_eq!(
        session.info().details.capabilities.mcp_transports,
        vec![McpTransport::Stdio, McpTransport::Http]
    );
    session.prompt("hi").await.unwrap();
    let mut text = String::new();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::TextDelta { text: t, .. } => text.push_str(&t),
            EventKind::RequestOpened(request) => {
                session.answer(request.id(), allow()).await.unwrap()
            }
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    assert!(text.contains("http:voice"), "declaration lost: {text:?}");
    assert!(text.contains("stdio:tool"), "declaration lost: {text:?}");
    session.close().await.unwrap();
}

/// Unsupported SSE transport refused typed UnsupportedFeature at open.
#[tokio::test]
async fn an_unsupported_mcp_transport_is_refused() {
    let runtime = Runtime::new();
    let error = runtime
        .open(
            &fixture(&[]),
            SessionOptions::in_dir(std::env::temp_dir())
                .mcp_server(McpServer::sse("events", "http://127.0.0.1:1/sse")),
        )
        .await
        .err()
        .expect("an sse declaration must be refused, not dropped");
    assert!(matches!(error, AgentError::UnsupportedFeature(_)));
}

/// Resume without load-session support fails typed ResumeFailed.
#[tokio::test]
async fn resuming_an_agent_without_load_session_fails() {
    let runtime = Runtime::new();
    let error = runtime
        .open(
            &fixture(&["--no-load"]),
            SessionOptions::in_dir(std::env::temp_dir()).resume(ResumeToken::new("sess-1")),
        )
        .await
        .err()
        .expect("resume must be refused, not silently replaced");
    assert!(matches!(error, AgentError::ResumeFailed(_)));
}

/// Fork on ACP agent fails typed UnsupportedFeature.
#[tokio::test]
async fn forking_an_acp_agent_fails_typed() {
    let runtime = Runtime::new();
    let error = runtime
        .open(
            &fixture(&[]),
            SessionOptions::in_dir(std::env::temp_dir())
                .fork_from(ResumeToken::new("sess-1"), None),
        )
        .await
        .err()
        .expect("fork must be refused, not silently replaced");
    assert!(matches!(error, AgentError::UnsupportedFeature(_)));
}

/// Auth-required open fails typed AuthRequired carrying terminal login command.
#[tokio::test]
async fn auth_required_carries_runnable_login_methods() {
    let runtime = Runtime::new();
    let result = runtime
        .open(
            &fixture(&["--auth-required"]),
            SessionOptions::in_dir(std::env::temp_dir()),
        )
        .await;
    let Err(AgentError::AuthRequired { login }) = result else {
        panic!("expected AuthRequired, got {:?}", result.err());
    };
    let LoginMethod::Terminal { command, .. } = &login[0] else {
        panic!("expected a terminal method");
    };
    assert_eq!(command[0], "node");
    assert_eq!(command[1..], ["auth".to_string(), "login".to_string()]);
}

/// Probe on logged-out reports Unauthenticated with empty config/commands instead of failing.
#[tokio::test]
async fn probe_reports_a_logged_out_agent_instead_of_failing() {
    // `open` refuses a logged-out agent; `probe` inspects it and answers.
    let details = Runtime::new()
        .probe(&fixture(&["--auth-required"]))
        .await
        .expect("probe must report a missing login, not fail on it");
    let AuthStatus::Unauthenticated { login } = &details.auth else {
        panic!("expected Unauthenticated, got {:?}", details.auth);
    };
    let LoginMethod::Terminal { command, .. } = &login[0] else {
        panic!("expected a terminal method");
    };
    assert_eq!(command[1..], ["auth".to_string(), "login".to_string()]);
    // Nothing is advertised until the agent has a login.
    assert!(details.config_options.is_empty());
    assert!(details.commands.is_empty());
}

/// Flood of 500 chunks arrives completely and in order by sequence.
#[tokio::test]
async fn a_flood_of_chunks_arrives_completely_and_in_order() {
    let (session, mut events) = open(&["--flood=500"]).await;
    session.prompt("hi").await.unwrap();
    let mut chunks = 0;
    let mut last_seq = 0;
    loop {
        let event = next(&mut events).await;
        assert!(event.sequence > last_seq);
        last_seq = event.sequence;
        match event.kind {
            EventKind::TextDelta { text, .. } if text.starts_with('x') => chunks += 1,
            EventKind::RequestOpened(request) => {
                session.answer(request.id(), allow()).await.unwrap()
            }
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    assert_eq!(chunks, 500);
    session.close().await.unwrap();
}

/// Losing auth mid-session fails turn and closes stream with AuthRequired + SessionClosed.
#[tokio::test]
async fn losing_auth_mid_session_fails_the_turn_and_closes() {
    let (session, mut events) = open(&[]).await;
    session.prompt("die-auth").await.unwrap();
    loop {
        let event = next(&mut events).await;
        if let EventKind::TurnEnded { stop, .. } = event.kind {
            assert!(matches!(stop, StopReason::Failed { .. }));
            break;
        }
    }
    let error = tokio::time::timeout(Duration::from_secs(10), events.next())
        .await
        .unwrap()
        .unwrap();
    let Err(AgentError::AuthRequired { login }) = error else {
        panic!("expected AuthRequired, got {error:?}");
    };
    let LoginMethod::Terminal { command, .. } = &login[0] else {
        panic!("expected a terminal method");
    };
    assert_eq!(command[1..], ["auth".to_string(), "login".to_string()]);
    // The stream closes (tolerating a trailing status flip) and the session
    // is gone.
    loop {
        let end = tokio::time::timeout(Duration::from_secs(10), events.next())
            .await
            .unwrap();
        match end {
            None => break,
            Some(Ok(event)) if matches!(event.kind, EventKind::StatusChanged(_)) => {}
            other => panic!("stream should close, got {other:?}"),
        }
    }
    assert!(matches!(
        session.prompt("hi").await,
        Err(AgentError::SessionClosed)
    ));
}

/// Plain prompt RPC error fails turn as Failed but session survives for next turn.
#[tokio::test]
async fn a_plain_prompt_error_fails_the_turn_and_the_session_survives() {
    let (session, mut events) = open(&[]).await;
    session.prompt("die-rpc").await.unwrap();
    loop {
        let event = next(&mut events).await;
        if let EventKind::TurnEnded { stop, .. } = event.kind {
            let StopReason::Failed { message } = stop else {
                panic!("expected Failed, got {stop:?}");
            };
            assert_eq!(message, "kaput (-32603)");
            break;
        }
    }
    // The session is still usable.
    session.prompt("hi").await.unwrap();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::RequestOpened(request) => {
                session.answer(request.id(), allow()).await.unwrap()
            }
            EventKind::TurnEnded { stop, .. } => {
                assert!(matches!(stop, StopReason::Completed { .. }));
                break;
            }
            _ => {}
        }
    }
    session.close().await.unwrap();
}

/// Creation-time configure applies at open; unknown id refused typed InvalidConfiguration.
#[tokio::test]
async fn creation_time_config_applies_at_open_and_a_refusal_fails_it() {
    let runtime = Runtime::new();
    let (session, _events) = runtime
        .open(
            &fixture(&[]),
            SessionOptions::in_dir(std::env::temp_dir())
                .configure("mode", "plan")
                .configure("model", "sonnet"),
        )
        .await
        .unwrap();
    let options = session.info().configuration.options;
    assert_eq!(
        options.get(&ConfigId::new("mode")),
        Some(&ConfigValue::Text("plan".into()))
    );
    session.close().await.unwrap();

    let refused = runtime
        .open(
            &fixture(&[]),
            SessionOptions::in_dir(std::env::temp_dir()).configure("bogus", "x"),
        )
        .await
        .map(|_| ());
    let Err(AgentError::InvalidConfiguration(message)) = refused else {
        panic!("expected InvalidConfiguration");
    };
    assert!(message.contains("bogus"), "got: {message}");
}

/// Grok prompt_complete settles hung turn; stale promptId frame ignored; session reusable.
#[tokio::test]
async fn grok_prompt_complete_ends_a_hung_turn_and_stale_ids_are_ignored() {
    let (session, mut events) = open(&[]).await;
    session.prompt("grok-hang").await.unwrap();
    let mut text = String::new();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::TextDelta { text: t, .. } => text.push_str(&t),
            EventKind::TurnEnded { stop, .. } => {
                // The stale frame carried `refusal`; only the frame echoing
                // our promptId may settle the turn.
                assert_eq!(
                    stop,
                    StopReason::Completed {
                        source: anyagent::CompletionSource::Protocol
                    }
                );
                break;
            }
            _ => {}
        }
    }
    assert_eq!(text, "grok ");

    // The abandoned session/prompt RPC never responds; the session still
    // runs full turns afterwards.
    session.prompt("hi").await.unwrap();
    let mut text = String::new();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::TextDelta { text: t, .. } => text.push_str(&t),
            EventKind::RequestOpened(Request::Permission(request)) => {
                session.answer(request.id, allow()).await.unwrap();
            }
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    assert!(text.starts_with("Hello"), "got: {text}");
}

/// Grok questions surface typed with header/choices; answer returns labels keyed by question text; the `ask_user` tool call around it stays hidden.
#[tokio::test]
async fn grok_questions_surface_typed_and_answers_return_labels() {
    let (session, mut events) = open(&[]).await;
    session.prompt("grok-question").await.unwrap();
    let mut text = String::new();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::ToolUpdated(tool) => panic!("question leaked as a tool: {}", tool.title),
            EventKind::RequestOpened(Request::Question(request)) => {
                let q = &request.questions[0];
                assert_eq!(q.text, "Pick a fruit");
                assert!(!q.multi_select);
                assert_eq!(q.choices[0].label, "Grape");
                assert_eq!(q.choices[0].description.as_deref(), Some("purple"));
                // Mango has no id: the label doubles as the choice id.
                assert_eq!(q.choices[1].id.as_str(), "Mango");
                session
                    .answer(
                        request.id,
                        Answer::Question(vec![QuestionAnswer::Choices(vec![ChoiceId::new("g")])]),
                    )
                    .await
                    .unwrap();
            }
            EventKind::TextDelta { text: t, .. } => text.push_str(&t),
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    // The answer went back keyed by question text, valued by option LABEL.
    assert_eq!(text, r#"q={"Pick a fruit":["Grape"]} "#);
}

/// Grok first-class models surface as live model select and switch via session/set_model.
#[tokio::test]
async fn grok_first_class_models_surface_as_the_model_option_and_switch_via_set_model() {
    let (session, mut events) = open(&["--grok-models", "--config-slow=50"]).await;
    let details = session.info().details;
    let model = details
        .config_options
        .iter()
        .find(|o| o.id.as_str() == "model")
        .expect("model option from the first-class models state");
    assert!(model.live);
    assert_eq!(model.current, Some(ConfigValue::Text("grok-4.5".into())));
    let anyagent::ConfigKind::Select { choices } = &model.kind else {
        panic!("expected select");
    };
    assert_eq!(choices.len(), 3);
    assert_eq!(choices[0].label, "Grok 4.5");
    assert_eq!(choices[0].description.as_deref(), Some("fast"));

    // Effort comes from the current model's `_meta`, with grok's labels.
    let effort_option = |session: &Session| {
        session
            .info()
            .details
            .config_options
            .into_iter()
            .find(|o| o.id.as_str() == "effort")
    };
    let effort = effort_option(&session).expect("effort from the model meta");
    assert!(effort.live);
    assert_eq!(effort.current, Some(ConfigValue::Text("high".into())));
    let anyagent::ConfigKind::Select { choices } = &effort.kind else {
        panic!("expected select");
    };
    assert_eq!(
        choices.iter().map(|c| c.label.as_str()).collect::<Vec<_>>(),
        ["Low Effort", "High Effort"]
    );
    assert_eq!(choices[1].description.as_deref(), Some("default"));

    // An effort switch rides session/set_model with `_meta.reasoningEffort`.
    session.configure("effort", "low").await.unwrap();
    while effort_option(&session).and_then(|o| o.current) != Some(ConfigValue::Text("low".into())) {
        next(&mut events).await;
    }

    // The fixture rejects session/set_config_option for models under
    // --grok-models, so this passing proves the session/set_model route.
    // The models/update that follows brings the new model's effort levels.
    session.configure("model", "grok-4.6").await.unwrap();
    session.configure("effort", "high").await.unwrap();
    loop {
        let event = next(&mut events).await;
        if let EventKind::SessionUpdated(info) = event.kind
            && info.configuration.options.get(&ConfigId::new("model"))
                == Some(&ConfigValue::Text("grok-4.6".into()))
        {
            break;
        }
    }
    let levels = |session: &Session| -> Vec<String> {
        match effort_option(session).map(|o| o.kind) {
            Some(anyagent::ConfigKind::Select { choices }) => {
                choices.into_iter().map(|c| c.value).collect()
            }
            _ => Vec::new(),
        }
    };
    while levels(&session) != ["low", "high", "xhigh"] {
        next(&mut events).await;
    }
    // The effort request sent before the model receipt must keep the new model.
    while effort_option(&session).and_then(|o| o.current) != Some(ConfigValue::Text("high".into()))
    {
        next(&mut events).await;
    }
    assert_eq!(
        session
            .info()
            .configuration
            .options
            .get(&ConfigId::new("model")),
        Some(&ConfigValue::Text("grok-4.6".into()))
    );
    session.prompt("model-state").await.unwrap();
    let mut actual = String::new();
    loop {
        match next(&mut events).await.kind {
            EventKind::TextDelta { text, .. } => actual.push_str(&text),
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    assert_eq!(
        actual, "grok-4.6:high",
        "the effort request reverted the model"
    );
    session.configure("effort", "xhigh").await.unwrap();
    while effort_option(&session).and_then(|o| o.current) != Some(ConfigValue::Text("xhigh".into()))
    {
        next(&mut events).await;
    }
    session.configure("model", "grok-basic").await.unwrap();
    while effort_option(&session).is_some() {
        next(&mut events).await;
    }
    assert!(
        !session
            .info()
            .configuration
            .options
            .contains_key(&ConfigId::new("effort"))
    );
    assert!(matches!(
        session.configure("effort", "low").await,
        Err(AgentError::InvalidConfiguration(_))
    ));
    session.close().await.unwrap();
}

/// Config_home on agent without known var refused typed InvalidConfiguration.
#[tokio::test]
async fn config_home_on_an_agent_without_a_known_var_is_refused() {
    let runtime = Runtime::new();
    let err = runtime
        .open(
            &fixture(&[]),
            SessionOptions::in_dir(std::env::temp_dir()).config_home(std::env::temp_dir()),
        )
        .await
        .err();
    assert!(
        matches!(err, Some(AgentError::InvalidConfiguration(_))),
        "expected InvalidConfiguration, got {err:?}"
    );
}

/// `record_wire` tees the ACP JSON-RPC wire too, both directions and including
/// the handshake, as one valid JSON object per line.
/// Record_wire tees both directions including initialize handshake as JSONL per direction.
#[tokio::test]
async fn record_wire_tees_the_acp_wire_including_handshake() {
    let dir = tempfile::tempdir().unwrap();
    let log = dir.path().join("wire.jsonl");
    let runtime = Runtime::new();
    let (session, mut events) = runtime
        .open(
            &fixture(&[]),
            SessionOptions::in_dir(std::env::temp_dir()).record_wire(&log),
        )
        .await
        .unwrap();
    session.prompt("hi").await.unwrap();
    loop {
        match next(&mut events).await.kind {
            EventKind::RequestOpened(Request::Permission(request)) => {
                session.answer(request.id, allow()).await.unwrap();
            }
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    session.close().await.unwrap();

    let mut lines = Vec::new();
    for _ in 0..40 {
        let body = std::fs::read_to_string(&log).unwrap_or_default();
        if body.lines().count() >= 5 {
            lines = body
                .lines()
                .map(|l| serde_json::from_str::<serde_json::Value>(l).expect("valid JSON line"))
                .collect();
            break;
        }
        tokio::time::sleep(Duration::from_millis(25)).await;
    }
    assert!(lines.len() >= 5, "too few frames recorded: {}", lines.len());
    assert!(
        lines.iter().any(|f| f["dir"] == "out"),
        "no outbound frames recorded"
    );
    assert!(
        lines.iter().any(|f| f["dir"] == "in"),
        "no inbound frames recorded"
    );
    assert!(
        lines
            .iter()
            .any(|f| f["dir"] == "out" && f["frame"]["method"] == "initialize"),
        "handshake initialize was not recorded"
    );
}

/// ACP has no compaction on the wire, so no ACP agent advertises it.
#[tokio::test]
async fn compact_is_unsupported_on_acp() {
    let (session, _events) = open(&[]).await;
    assert!(matches!(
        session.compact().await,
        Err(AgentError::UnsupportedFeature(_))
    ));
}

/// Kiro effort: the shared levels for the model, none for models without effort, switched via a `/effort` prompt that queues behind a running turn, ack never leaks, settable at creation.
#[tokio::test]
async fn kiro_effort_is_a_live_option_backed_by_the_effort_prompt() {
    let (session, mut events) = open(&["--kiro", "--effort-slow=50"]).await;
    let effort = |session: &Session| {
        session
            .info()
            .configuration
            .options
            .get(&ConfigId::new("effort"))
            .cloned()
    };
    let option = session
        .info()
        .details
        .config_options
        .into_iter()
        .find(|o| o.id.as_str() == "effort")
        .expect("effort advertised");
    assert!(option.live);
    let anyagent::ConfigKind::Select { choices } = &option.kind else {
        panic!("effort is a select");
    };
    assert_eq!(
        choices.iter().map(|c| c.value.as_str()).collect::<Vec<_>>(),
        ["low", "medium", "high", "xhigh", "max"]
    );

    // The metadata frame right after session/new fills in the current level.
    while effort(&session) != Some(ConfigValue::Text("high".into())) {
        next(&mut events).await;
    }

    // A live switch rides a `/effort` prompt; its ack chunk stays internal
    // while the unrelated update sent alongside it still arrives.
    session.configure("effort", "low").await.unwrap();
    let mut usage_seen = false;
    // Read the stream, not the snapshot: the snapshot can already be
    // updated while earlier events are still queued.
    loop {
        let event = next(&mut events).await;
        assert!(
            !matches!(event.kind, EventKind::TextDelta { .. }),
            "the effort ack leaked as text"
        );
        usage_seen |= matches!(event.kind, EventKind::ContextUsage { used_tokens: 7, .. });
        if let EventKind::SessionUpdated(info) = &event.kind
            && info.configuration.options.get(&ConfigId::new("effort"))
                == Some(&ConfigValue::Text("low".into()))
        {
            break;
        }
    }
    assert!(usage_seen, "the usage update mid-switch was dropped");

    // A switch mid-turn waits for the turn, then applies.
    session.prompt("hi").await.unwrap();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::RequestOpened(Request::Permission(r)) => {
                session.configure("effort", "medium").await.unwrap();
                session.answer(r.id, allow()).await.unwrap();
            }
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    while effort(&session) != Some(ConfigValue::Text("medium".into())) {
        next(&mut events).await;
    }

    // A switch queued behind a turn that fails still goes out.
    session.prompt("hi die-late").await.unwrap();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::RequestOpened(Request::Permission(r)) => {
                session.configure("effort", "xhigh").await.unwrap();
                session.answer(r.id, allow()).await.unwrap();
            }
            EventKind::TurnEnded { stop, .. } => {
                assert!(matches!(stop, StopReason::Failed { .. }), "{stop:?}");
                break;
            }
            _ => {}
        }
    }
    while effort(&session) != Some(ConfigValue::Text("xhigh".into())) {
        next(&mut events).await;
    }

    // A prompt sent while a switch runs waits for it, then runs.
    session.configure("effort", "high").await.unwrap();
    // This unrelated receipt lands before the effort response and must not
    // clear its ID or strand the held prompt.
    session.configure("model", "opus").await.unwrap();
    session.prompt("hi").await.unwrap();
    let mut text = String::new();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::TextDelta { text: t, .. } => text.push_str(&t),
            EventKind::RequestOpened(Request::Permission(r)) => {
                session.answer(r.id, allow()).await.unwrap();
            }
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    assert!(text.starts_with("Hello"), "got {text:?}");
    assert!(
        !text.contains("Effort set"),
        "the effort ack leaked: {text:?}"
    );
    assert_eq!(effort(&session), Some(ConfigValue::Text("high".into())));

    // A model without effort drops the option; switching back restores it.
    let has_effort = |session: &Session| {
        session
            .info()
            .details
            .config_options
            .iter()
            .any(|o| o.id.as_str() == "effort")
    };
    session
        .configure("model", "claude-haiku-4.5")
        .await
        .unwrap();
    while has_effort(&session) {
        next(&mut events).await;
    }
    assert_eq!(effort(&session), None);
    assert!(matches!(
        session.configure("effort", "low").await,
        Err(AgentError::InvalidConfiguration(_))
    ));
    session.configure("model", "opus").await.unwrap();
    while !has_effort(&session) {
        next(&mut events).await;
    }
    session.close().await.unwrap();

    // Creation-time effort on a model without it fails the open.
    let refused = Runtime::new()
        .open(
            &fixture(&["--kiro"]),
            SessionOptions::in_dir(std::env::temp_dir())
                .configure("model", "claude-haiku-4.5")
                .configure("effort", "low"),
        )
        .await
        .map(|_| ());
    assert!(matches!(refused, Err(AgentError::InvalidConfiguration(_))));

    // Creation-time effort applies before the first turn.
    let (session, _events) = Runtime::new()
        .open(
            &fixture(&["--kiro"]),
            SessionOptions::in_dir(std::env::temp_dir()).configure("effort", "medium"),
        )
        .await
        .unwrap();
    assert_eq!(effort(&session), Some(ConfigValue::Text("medium".into())));
    session.close().await.unwrap();
}

/// Kiro's cancel can race the next prompt and end it "cancelled" with
/// nothing said; the adapter re-sends that prompt once, so the caller sees a
/// normal turn. A cancel of our own is never retried.
#[tokio::test]
async fn kiro_spurious_cancel_is_retried_once_inside_the_adapter() {
    let (session, mut events) = open(&["--kiro", "--spurious-cancel"]).await;
    session.prompt("one").await.unwrap();
    loop {
        if let EventKind::RequestOpened(_) = next(&mut events).await.kind {
            break;
        }
    }
    session.cancel(false).await.unwrap();
    loop {
        if let EventKind::TurnEnded { stop, .. } = next(&mut events).await.kind {
            assert_eq!(stop, StopReason::Cancelled);
            break;
        }
    }
    // The fixture kills this prompt "cancelled" first; the retry runs it.
    session.prompt("two").await.unwrap();
    let mut text = String::new();
    loop {
        match next(&mut events).await.kind {
            EventKind::TextDelta { text: t, .. } => text.push_str(&t),
            EventKind::RequestOpened(request) => {
                session.answer(request.id(), allow()).await.unwrap()
            }
            EventKind::TurnEnded { stop, .. } => {
                assert!(matches!(stop, StopReason::Completed { .. }), "{stop:?}");
                break;
            }
            _ => {}
        }
    }
    assert!(text.contains("Hello"), "{text:?}");
    session.close().await.unwrap();
}

// ---------------------------------------------------------------------------
// CURSOR: the cursor-agent shape (fixture `--cursor`, recorded 2026-09-07)
// ---------------------------------------------------------------------------

/// A cursor catalog agent driven by the fixture's `--cursor` scenario.
fn cursor(name: &str, flags: &str) -> AgentInstallation {
    catalog_wrapper("cursor", name, &format!("--cursor {flags}"))
}

/// The advertised option with this id, if any.
fn option(session: &Session, id: &str) -> Option<anyagent::ConfigOption> {
    session
        .info()
        .details
        .config_options
        .into_iter()
        .find(|o| o.id.as_str() == id)
}

/// Waits for events until the session's options satisfy `done`.
async fn wait_options(session: &Session, events: &mut Events, done: impl Fn(&Session) -> bool) {
    while !done(session) {
        let event = next(events).await;
        assert!(
            !matches!(
                event.kind,
                EventKind::Diagnostic(anyagent::Diagnostic {
                    level: anyagent::DiagnosticLevel::Warning,
                    ..
                })
            ),
            "warning while waiting: {:?}",
            event.kind
        );
    }
}

/// Open runs `about` then `authenticate`: version and account come from `about`, options from session/new with `mode` listed once.
#[tokio::test]
async fn cursor_open_authenticates_and_reports_version_account_and_options() {
    let runtime = Runtime::new();
    let (session, _events) = runtime
        .open(
            &cursor("open", ""),
            SessionOptions::in_dir(std::env::temp_dir()),
        )
        .await
        .unwrap();
    let info = session.info();
    assert_eq!(info.details.version.as_deref(), Some("0.0.1"));
    assert_eq!(
        info.details.auth,
        AuthStatus::Authenticated {
            kind: AuthKind::Subscription,
            account: Some(anyagent::AccountInfo {
                email: Some("dev@example.com".into()),
                plan: Some("Free".into()),
            }),
        }
    );
    let caps = &info.details.capabilities;
    for cap in [
        Capability::Permissions,
        Capability::Images,
        Capability::Resume,
    ] {
        assert!(caps.supports(cap.clone()), "missing {cap:?}");
    }
    assert!(
        !caps.supports(Capability::Steer),
        "cursor advertises no steering"
    );
    let ids: Vec<_> = info
        .details
        .config_options
        .iter()
        .map(|o| o.id.as_str().to_owned())
        .collect();
    assert_eq!(
        ids,
        vec!["mode", "model"],
        "mode listed once, no per-model options on Auto"
    );
    assert_eq!(
        option(&session, "mode").unwrap().current,
        Some(ConfigValue::Text("agent".into()))
    );
    assert_eq!(info.resume_token.unwrap().as_str(), "sess-1");
    session.close().await.unwrap();
}

/// Logged out, `about` reports no email: probe answers Unauthenticated with the catalog login before `authenticate` could start a browser login (the fixture hangs there).
#[tokio::test]
async fn cursor_logged_out_is_reported_before_the_browser_login() {
    let agent = cursor("out", "--logged-out");
    let details = tokio::time::timeout(Duration::from_secs(5), Runtime::new().probe(&agent))
        .await
        .expect("probe must not wait on the browser login")
        .unwrap();
    let AuthStatus::Unauthenticated { login } = details.auth else {
        panic!("expected Unauthenticated, got {:?}", details.auth);
    };
    let LoginMethod::Terminal { command, .. } = &login[0] else {
        panic!("expected a terminal login method");
    };
    assert_eq!(command[1..], ["login"]);
    assert!(
        login
            .iter()
            .any(|m| matches!(m, LoginMethod::EnvVar { name } if name == "CURSOR_API_KEY"))
    );
}

/// A model switch adopts the model's own options from the config response (fast for composer, thinking/context/effort for opus), and they leave with the model; creation-time config does the same.
#[tokio::test]
async fn cursor_model_switch_reveals_the_models_own_options() {
    let runtime = Runtime::new();
    let (session, mut events) = runtime
        .open(
            &cursor("models", ""),
            SessionOptions::in_dir(std::env::temp_dir()),
        )
        .await
        .unwrap();
    session.configure("model", "composer-2.5").await.unwrap();
    wait_options(&session, &mut events, |s| option(s, "fast").is_some()).await;
    let fast = option(&session, "fast").unwrap();
    assert!(fast.live);
    assert_eq!(fast.current, Some(ConfigValue::Text("true".into())));
    assert_eq!(
        option(&session, "model").unwrap().current,
        Some(ConfigValue::Text("composer-2.5".into()))
    );

    session.configure("fast", "false").await.unwrap();
    wait_options(&session, &mut events, |s| {
        option(s, "fast").and_then(|o| o.current) == Some(ConfigValue::Text("false".into()))
    })
    .await;

    session.configure("model", "default").await.unwrap();
    wait_options(&session, &mut events, |s| option(s, "fast").is_none()).await;
    assert_eq!(
        option(&session, "mode").unwrap().current,
        Some(ConfigValue::Text("agent".into()))
    );
    session.close().await.unwrap();

    // Creation-time: the requested model's options are there at open.
    let (session, _events) = runtime
        .open(
            &cursor("models-open", ""),
            SessionOptions::in_dir(std::env::temp_dir()).configure("model", "claude-opus-5"),
        )
        .await
        .unwrap();
    assert_eq!(
        option(&session, "context").unwrap().current,
        Some(ConfigValue::Text("300k".into()))
    );
    assert_eq!(
        option(&session, "effort").unwrap().current,
        Some(ConfigValue::Text("high".into()))
    );
    assert_eq!(
        option(&session, "thinking").unwrap().current,
        Some(ConfigValue::Text("true".into()))
    );
    session.close().await.unwrap();
}

/// `cursor/ask_question` surfaces typed (prompt/allowMultiple read), and the answer goes back as `answered` with the option ids.
/// Antigravity's ACP server asks through an `interaction_*` permission
/// whose options are the choices: it surfaces as a question, never as a
/// tool or a permission, and the answer is the chosen option id. Another
/// agent's `interaction_*` tool is an ordinary tool and permission.
#[tokio::test]
async fn interaction_permissions_are_questions() {
    let (session, mut events) = open(&[]).await;
    session.prompt("interaction-question").await.unwrap();
    let mut saw = (false, false);
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::RequestOpened(Request::Permission(request)) => {
                saw.0 = true;
                session.answer(request.id, allow()).await.unwrap();
            }
            EventKind::RequestOpened(other) => panic!("a question on a plain agent: {other:?}"),
            EventKind::ToolUpdated(tool) if tool.id.as_str() == "interaction_1" => saw.1 = true,
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    assert_eq!(saw, (true, true), "(permission, tool) on a plain agent");
    session.close().await.unwrap();

    let (session, mut events) = open(&["--antigravity"]).await;
    session.prompt("interaction-question").await.unwrap();
    let mut text = String::new();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::RequestOpened(Request::Question(request)) => {
                let q = &request.questions[0];
                assert_eq!(q.text, "Red or blue?");
                assert_eq!(
                    q.choices
                        .iter()
                        .map(|c| c.label.as_str())
                        .collect::<Vec<_>>(),
                    ["Red", "Blue"]
                );
                assert_eq!(q.choices[1].id.as_str(), "2");
                session
                    .answer(
                        request.id,
                        Answer::Question(vec![QuestionAnswer::Choices(vec![ChoiceId::new("2")])]),
                    )
                    .await
                    .unwrap();
            }
            EventKind::RequestOpened(other) => panic!("not a question: {other:?}"),
            EventKind::ToolUpdated(tool) => panic!("question surfaced as a tool: {}", tool.title),
            EventKind::TextDelta { text: t, .. } => text.push_str(&t),
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    assert_eq!(text, r#"q={"outcome":"selected","optionId":"2"} "#);
    assert!(
        session
            .info()
            .details
            .capabilities
            .supports(Capability::Questions)
    );
    session.close().await.unwrap();
}

#[tokio::test]
async fn cursor_questions_round_trip_in_cursors_shape() {
    let runtime = Runtime::new();
    let (session, mut events) = runtime
        .open(
            &cursor("question", ""),
            SessionOptions::in_dir(std::env::temp_dir()),
        )
        .await
        .unwrap();
    session.prompt("cursor-question").await.unwrap();
    let mut text = String::new();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::RequestOpened(Request::Question(request)) => {
                let q = &request.questions[0];
                assert_eq!(q.id.as_str(), "color");
                assert_eq!(q.text, "Red or blue?");
                assert!(!q.multi_select);
                assert_eq!(
                    q.choices.iter().map(|c| c.id.as_str()).collect::<Vec<_>>(),
                    ["r", "b"]
                );
                assert_eq!(q.choices[1].label, "Blue");
                session
                    .answer(
                        request.id,
                        Answer::Question(vec![QuestionAnswer::Choices(vec![ChoiceId::new("b")])]),
                    )
                    .await
                    .unwrap();
            }
            EventKind::TextDelta { text: t, .. } => text.push_str(&t),
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    assert_eq!(
        text,
        r#"q={"outcome":"answered","answers":[{"questionId":"color","selectedOptionIds":["b"]}]} "#
    );
    assert!(
        session
            .info()
            .details
            .capabilities
            .supports(Capability::Questions)
    );
    session.close().await.unwrap();
}

/// `cursor/update_todos` becomes the plan (merge honoured), `cursor/task` is acknowledged (a named tool, not a nested subagent), image generation is declined; none is answered "method not found".
#[tokio::test]
async fn cursor_todos_task_and_image_requests_are_answered() {
    let runtime = Runtime::new();
    let (session, mut events) = runtime
        .open(
            &cursor("todos", ""),
            SessionOptions::in_dir(std::env::temp_dir()),
        )
        .await
        .unwrap();
    session.prompt("cursor-todos").await.unwrap();
    let mut plans = Vec::new();
    let mut surfaced = Vec::new();
    let mut task_kind = None;
    let mut text = String::new();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::PlanUpdated { entries } => plans.push(
                entries
                    .into_iter()
                    .map(|e| (e.text, e.status))
                    .collect::<Vec<_>>(),
            ),
            EventKind::ToolUpdated(tool) if tool.title == "Task: List files" => {
                task_kind = Some((tool.kind, tool.raw.unwrap().name));
            }
            EventKind::Diagnostic(d) => {
                assert_ne!(d.level, anyagent::DiagnosticLevel::Warning, "{}", d.message);
                surfaced.extend(event.extensions.keys().cloned());
            }
            EventKind::TextDelta { text: t, .. } => text.push_str(&t),
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    use anyagent::PlanStatus as P;
    assert_eq!(
        plans,
        vec![
            vec![
                ("first".to_string(), P::Completed),
                ("second".to_string(), P::Pending)
            ],
            vec![
                ("first".to_string(), P::Completed),
                ("second".to_string(), P::InProgress)
            ],
        ]
    );
    assert_eq!(
        task_kind,
        Some((anyagent::ToolKind::Other, "task".to_string()))
    );
    assert!(
        surfaced.contains(&"cursor/task".to_string()),
        "{surfaced:?}"
    );
    assert!(
        surfaced.contains(&"cursor/generate_image".to_string()),
        "{surfaced:?}"
    );
    assert_eq!(text, "todos=2 task=completed image=rejected ");
    session.close().await.unwrap();
}

/// `cursor/create_plan` is accepted and surfaced raw with its markdown; the plan itself arrives as the standard plan update.
#[tokio::test]
async fn cursor_plan_proposals_are_accepted_and_surfaced() {
    let runtime = Runtime::new();
    let (session, mut events) = runtime
        .open(
            &cursor("plan", ""),
            SessionOptions::in_dir(std::env::temp_dir()),
        )
        .await
        .unwrap();
    session.prompt("cursor-plan").await.unwrap();
    let mut steps = Vec::new();
    let mut markdown = None;
    let mut text = String::new();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::PlanUpdated { entries } => {
                steps = entries.into_iter().map(|e| e.text).collect()
            }
            EventKind::Diagnostic(_) => {
                if let Some(plan) = event.extensions.get("cursor/create_plan") {
                    markdown = plan["plan"].as_str().map(str::to_owned);
                }
            }
            EventKind::TextDelta { text: t, .. } => text.push_str(&t),
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    assert_eq!(steps, vec!["write README"]);
    assert_eq!(markdown.as_deref(), Some("# Plan\n1. write"));
    assert_eq!(text, "plan=accepted ");
    session.close().await.unwrap();
}

/// Every turn ends with an estimated ContextUsage labelled `anyagent/estimated`; the window follows the model's `context` option (none on Auto).
#[tokio::test]
async fn cursor_context_usage_is_estimated_and_labelled() {
    async fn usage_after_turn(session: &Session, events: &mut Events) -> (u64, Option<u64>, bool) {
        session.prompt("hi").await.unwrap();
        let mut usage = None;
        loop {
            let event = next(events).await;
            match event.kind {
                EventKind::RequestOpened(Request::Permission(request)) => {
                    session.answer(request.id, allow()).await.unwrap();
                }
                EventKind::ContextUsage {
                    used_tokens,
                    window_tokens,
                    ..
                } => {
                    let estimated = event.extensions.get("anyagent/estimated")
                        == Some(&serde_json::Value::Bool(true));
                    usage = Some((used_tokens, window_tokens, estimated));
                }
                EventKind::TurnEnded { .. } => break,
                _ => {}
            }
        }
        usage.expect("a usage event before the turn ended")
    }
    let runtime = Runtime::new();
    let (session, mut events) = runtime
        .open(
            &cursor("usage", ""),
            SessionOptions::in_dir(std::env::temp_dir()),
        )
        .await
        .unwrap();
    assert!(
        !session
            .info()
            .details
            .capabilities
            .supports(Capability::ContextUsage)
    );
    let (used, window, estimated) = usage_after_turn(&session, &mut events).await;
    assert!(estimated);
    assert!(used > 0, "counted the prompt and reply");
    assert_eq!(window, None, "Auto has no context option");
    assert!(
        session
            .info()
            .details
            .capabilities
            .supports(Capability::ContextUsage)
    );
    // The count accumulates over the session.
    let (used_again, _, _) = usage_after_turn(&session, &mut events).await;
    assert!(used_again > used);
    session.close().await.unwrap();

    let (session, mut events) = runtime
        .open(
            &cursor("usage-opus", ""),
            SessionOptions::in_dir(std::env::temp_dir()).configure("model", "claude-opus-5"),
        )
        .await
        .unwrap();
    let (_, window, _) = usage_after_turn(&session, &mut events).await;
    assert_eq!(window, Some(300_000));
    session.close().await.unwrap();

    // An inlined image counts a flat allowance; a resumed session has no
    // baseline for what came before, so it estimates nothing.
    let dir = std::env::temp_dir().join(format!("anyagent-cursor-att-{}", std::process::id()));
    std::fs::create_dir_all(&dir).unwrap();
    std::fs::write(dir.join("shot.png"), b"\x89PNG\r\n\x1a\ndata").unwrap();
    let (session, mut events) = runtime
        .open(
            &cursor("usage-image", ""),
            SessionOptions::in_dir(std::env::temp_dir()),
        )
        .await
        .unwrap();
    session
        .prompt(Input::text("hi").attach(dir.join("shot.png")))
        .await
        .unwrap();
    let mut used = 0;
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::RequestOpened(request) => {
                session.answer(request.id(), allow()).await.unwrap()
            }
            EventKind::ContextUsage { used_tokens, .. } => used = used_tokens,
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    assert!(used >= 1_500, "image allowance missing: {used}");
    let token = session.info().resume_token.unwrap();
    session.close().await.unwrap();

    let (session, mut events) = runtime
        .open(
            &cursor("usage-resume", ""),
            SessionOptions::in_dir(std::env::temp_dir()).resume(token),
        )
        .await
        .unwrap();
    session.prompt("hi").await.unwrap();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::RequestOpened(request) => {
                session.answer(request.id(), allow()).await.unwrap()
            }
            EventKind::ContextUsage { .. } => panic!("a resumed session must not estimate"),
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    assert!(
        !session
            .info()
            .details
            .capabilities
            .supports(Capability::ContextUsage)
    );
    session.close().await.unwrap();
}

/// A cursor question without options takes free text, which rides back as the option id.
#[tokio::test]
async fn cursor_optionless_questions_take_free_text() {
    let runtime = Runtime::new();
    let (session, mut events) = runtime
        .open(
            &cursor("question-free", ""),
            SessionOptions::in_dir(std::env::temp_dir()),
        )
        .await
        .unwrap();
    session.prompt("cursor-question-free").await.unwrap();
    let mut text = String::new();
    loop {
        let event = next(&mut events).await;
        match event.kind {
            EventKind::RequestOpened(Request::Question(request)) => {
                let q = &request.questions[0];
                assert!(q.choices.is_empty());
                assert!(q.allows_free_text);
                session
                    .answer(
                        request.id,
                        Answer::Question(vec![QuestionAnswer::Text("purple".into())]),
                    )
                    .await
                    .unwrap();
            }
            EventKind::TextDelta { text: t, .. } => text.push_str(&t),
            EventKind::TurnEnded { .. } => break,
            _ => {}
        }
    }
    assert_eq!(
        text,
        r#"q={"outcome":"answered","answers":[{"questionId":"color","selectedOptionIds":["purple"]}]} "#
    );
    session.close().await.unwrap();
}