diff --git a/src/acp/contract_tests.rs b/src/acp/contract_tests.rs index 793acab..f1244f5 100644 --- a/src/acp/contract_tests.rs +++ b/src/acp/contract_tests.rs @@ -1,6 +1,7 @@ use std::collections::VecDeque; use std::future::Future; use std::sync::{Arc, Mutex}; +use std::time::Duration; use agent_client_protocol::{ self as acp_sdk, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, schema::v1 as acp, @@ -12,6 +13,7 @@ use tokio_tungstenite::tungstenite::Message; use super::commands; use super::connection::{AcpConnection, internal_error}; use super::context::CommandContext; +use super::elicitation::PendingResponse; use super::events::EventSink; use super::inbound; use super::runtime::RuntimeState; @@ -19,8 +21,10 @@ use super::transport::jsonrpc::Peer; use crate::acp_state::{AcpAppEvent, AcpSessionUpdate}; use crate::app::App; use crate::application::{self, AppEvent}; -use crate::command::{Command, SessionListRequest}; +use crate::command::{Command, PromptBlock, SessionListRequest}; +use crate::domain::auth::{AuthMethod, OAuthFlowKind, OAuthResultStatus, OAuthStatus}; use crate::domain::chat::ChatEntry; +use crate::domain::session::{ForkResult, RedoResult, UndoResult, UndoStackSnapshot}; use crate::runtime_events::ServerChannelMsg; #[derive(Debug, Clone, PartialEq)] @@ -176,16 +180,43 @@ async fn extension_router_pins_methods_underscore_prefix_and_explicit_nulls() { "agent_id": "coder", "model": null }), - json!({ "providers": [] }), - json!({ "node_id": "node", "sessions": [], "total_count": 0 }), + json!({ + "providers": [{ + "provider": "openai", + "display_name": "OpenAI", + "oauth_status": "connected", + "has_stored_api_key": false, + "has_env_api_key": true, + "env_var_name": "OPENAI_API_KEY", + "supports_oauth": true, + "preferred_method": "oauth" + }] + }), + json!({ + "node_id": "node", + "sessions": [{ + "id": "remote", + "node_id": "node", + "node_label": "Remote Node", + "title": "Remote Session", + "cwd": "/remote", + "updated_at": "now", + "profile_id": "fast", + "model_id": "gpt-5" + }], + "next_offset": 50, + "total_count": 51 + }), json!({ "invite_id": "invite", "url": "qmt://invite", + "qr_code": "QR", "expires_at": 1, - "max_uses": 1 + "max_uses": 1, + "mesh_name": "Team" }), ]); - let (state, events, _rx) = harness(&connection); + let (state, events, mut rx) = harness(&connection); let commands = [ Command::SetDelegateModel { session_id: "parent".into(), @@ -211,6 +242,107 @@ async fn extension_router_pins_methods_underscore_prefix_and_explicit_nulls() { .expect("extension command"); } + assert_eq!( + connection.messages(), + vec![ + RecordedMessage { + method: "_querymt/session/setDelegateModel".into(), + params: json!({ + "session_id": "parent", + "agent_id": "coder", + "model_id": null, + "node_id": null + }), + }, + RecordedMessage { + method: "_querymt/auth/status".into(), + params: json!({}), + }, + RecordedMessage { + method: "_querymt/remote/sessions".into(), + params: json!({ "node_id": "node", "offset": 0, "limit": 50 }), + }, + RecordedMessage { + method: "_querymt/mesh/createInvite".into(), + params: json!({ "mesh_name": null, "ttl": null, "max_uses": null }), + }, + ] + ); + + assert!(matches!( + rx.try_recv().expect("delegate model event"), + ServerChannelMsg::Acp(AcpAppEvent::DelegateModelSet { + session_id, + agent_id, + model: None, + }) if session_id == "parent" && agent_id == "coder" + )); + assert!(matches!( + rx.try_recv().expect("auth providers event"), + ServerChannelMsg::Acp(AcpAppEvent::AuthProviders(providers)) + if providers.len() == 1 + && providers[0].provider == "openai" + && providers[0].display_name == "OpenAI" + && providers[0].oauth_status == Some(OAuthStatus::Connected) + && !providers[0].has_stored_api_key + && providers[0].has_env_api_key + && providers[0].env_var_name.as_deref() == Some("OPENAI_API_KEY") + && providers[0].supports_oauth + && providers[0].preferred_method == Some(AuthMethod::OAuth) + )); + assert!(matches!( + rx.try_recv().expect("remote sessions event"), + ServerChannelMsg::Acp(AcpAppEvent::RemoteSessions(list)) + if list.node_id == "node" + && list.next_offset == Some(50) + && list.total_count == 51 + && list.sessions.len() == 1 + && list.sessions[0].id == "remote" + && list.sessions[0].node_id == "node" + && list.sessions[0].node_label.as_deref() == Some("Remote Node") + && list.sessions[0].title.as_deref() == Some("Remote Session") + && list.sessions[0].cwd.as_deref() == Some("/remote") + && list.sessions[0].updated_at.as_deref() == Some("now") + && list.sessions[0].profile_id.as_deref() == Some("fast") + && list.sessions[0].model_id.as_deref() == Some("gpt-5") + )); + assert!(matches!( + rx.try_recv().expect("mesh invite event"), + ServerChannelMsg::Acp(AcpAppEvent::MeshInviteCreated(invite)) + if invite.invite_id == "invite" + && invite.url == "qmt://invite" + && invite.qr_code.as_deref() == Some("QR") + && invite.expires_at == 1 + && invite.max_uses == 1 + && invite.mesh_name.as_deref() == Some("Team") + )); + assert!(rx.try_recv().is_err()); +} + +#[tokio::test] +async fn initialize_pins_client_fields_gated_follow_ups_and_event_order() { + let connection = RecordingConnection::with_responses([ + json!({ + "protocolVersion": 1, + "agentCapabilities": {}, + "agentInfo": { "name": "querymt-agent", "title": "QueryMT Agent", "version": "1" } + }), + json!({ + "methods": ["querymt/mesh/nodes", "querymt/profiles"], + "features": { "profiles": true } + }), + json!({ "nodes": [{ "id": "node-1", "label": "Remote" }] }), + json!({ + "profiles": [{ "id": "fast", "name": "Fast" }], + "active_profile_id": "fast" + }), + ]); + let (state, events, mut rx) = harness(&connection); + + commands::dispatch(context(&connection, &state, &events), Command::Init) + .await + .expect("initialize"); + let messages = connection.messages(); assert_eq!( messages @@ -218,22 +350,328 @@ async fn extension_router_pins_methods_underscore_prefix_and_explicit_nulls() { .map(|message| message.method.as_str()) .collect::>(), [ - "_querymt/session/setDelegateModel", - "_querymt/auth/status", - "_querymt/remote/sessions", - "_querymt/mesh/createInvite", + "initialize", + "_querymt/capabilities", + "_querymt/mesh/nodes", + "_querymt/profiles", ] ); - assert_eq!(messages[0].params["model_id"], Value::Null); - assert_eq!(messages[0].params["node_id"], Value::Null); assert_eq!( - messages[2].params, - json!({ "node_id": "node", "offset": 0, "limit": 50 }) + messages[0].params, + json!({ + "protocolVersion": 1, + "clientCapabilities": { + "fs": { "readTextFile": false, "writeTextFile": false }, + "terminal": false, + "auth": { "terminal": false }, + "elicitation": { "form": {} } + }, + "clientInfo": { "name": "qmtui", "version": env!("CARGO_PKG_VERSION") } + }) ); + assert_eq!(messages[1].params, json!({})); + assert_eq!(messages[2].params, json!({})); + assert_eq!(messages[3].params, json!({})); + + assert!(matches!( + rx.try_recv().expect("initialized"), + ServerChannelMsg::Acp(AcpAppEvent::Initialized { + agent_id, + agent_name, + profiles, + active_profile_id: None, + agent_mode: Some(mode), + reasoning_effort: Some(None), + }) if agent_id == "querymt-agent" + && agent_name == "QueryMT Agent" + && profiles.is_empty() + && mode == "build" + )); + assert!(matches!( + rx.try_recv().expect("capabilities"), + ServerChannelMsg::Acp(AcpAppEvent::ControlCapabilities(value)) + if value["methods"][0] == "querymt/mesh/nodes" + )); + assert!(matches!( + rx.try_recv().expect("mesh nodes"), + ServerChannelMsg::Acp(AcpAppEvent::MeshNodes(nodes)) + if nodes.nodes[0].id == "node-1" + )); + assert!(matches!( + rx.try_recv().expect("profiles"), + ServerChannelMsg::Acp(AcpAppEvent::Profiles { profiles, active_profile_id }) + if profiles[0].id == "fast" && active_profile_id.as_deref() == Some("fast") + )); + assert!(rx.try_recv().is_err()); +} + +#[tokio::test] +async fn initialize_omits_unadvertised_follow_ups_and_emits_exact_empty_profiles() { + let connection = RecordingConnection::with_responses([ + json!({ + "protocolVersion": 1, + "agentCapabilities": {}, + "agentInfo": { "name": "querymt-agent", "title": "QueryMT Agent", "version": "1" } + }), + json!({ "methods": [], "features": { "profiles": false } }), + ]); + let (state, events, mut rx) = harness(&connection); + + commands::dispatch(context(&connection, &state, &events), Command::Init) + .await + .expect("initialize without optional capabilities"); + + let messages = connection.messages(); + assert_eq!(messages.len(), 2); + assert_eq!(messages[0].method, "initialize"); assert_eq!( - messages[3].params, - json!({ "mesh_name": null, "ttl": null, "max_uses": null }) + messages[1], + RecordedMessage { + method: "_querymt/capabilities".into(), + params: json!({}), + } ); + assert!(messages.iter().all(|message| { + message.method != "_querymt/mesh/nodes" && message.method != "_querymt/profiles" + })); + assert!(matches!( + rx.try_recv().expect("initialized"), + ServerChannelMsg::Acp(AcpAppEvent::Initialized { + agent_id, + agent_name, + profiles, + active_profile_id: None, + agent_mode: Some(mode), + reasoning_effort: Some(None), + }) if agent_id == "querymt-agent" + && agent_name == "QueryMT Agent" + && profiles.is_empty() + && mode == "build" + )); + assert!(matches!( + rx.try_recv().expect("capabilities"), + ServerChannelMsg::Acp(AcpAppEvent::ControlCapabilities(value)) + if value == json!({ "methods": [], "features": { "profiles": false } }) + )); + assert!(matches!( + rx.try_recv().expect("empty profiles"), + ServerChannelMsg::Acp(AcpAppEvent::Profiles { + profiles, + active_profile_id: None, + }) if profiles.is_empty() + )); + assert!(rx.try_recv().is_err()); +} + +#[tokio::test] +async fn standard_session_router_pins_new_prompt_delete_payloads_and_events() { + let connection = RecordingConnection::with_responses([ + json!({ "sessionId": "session-1" }), + json!({ "stopReason": "end_turn" }), + json!({}), + ]); + let (state, events, mut rx) = harness(&connection); + + commands::dispatch( + context(&connection, &state, &events), + Command::NewSession { + cwd: Some("/repo".into()), + profile_id: Some("fast".into()), + }, + ) + .await + .expect("new session"); + commands::dispatch( + context(&connection, &state, &events), + Command::Prompt { + prompt: vec![ + PromptBlock::Text { + text: "hello".into(), + }, + PromptBlock::ResourceLink { + name: "guide".into(), + uri: "file:///repo/guide.md".into(), + }, + ], + local_id: "local-1".into(), + }, + ) + .await + .expect("prompt dispatch"); + + let created = rx.recv().await.expect("created"); + let started = rx.recv().await.expect("turn started"); + let finished = tokio::time::timeout(Duration::from_secs(1), rx.recv()) + .await + .expect("prompt completion timeout") + .expect("finished"); + assert!(matches!( + created, + ServerChannelMsg::Acp(AcpAppEvent::SessionCreated { + agent_id, + session_id, + profile_id: Some(profile_id), + }) if agent_id == "querymt" && session_id == "session-1" && profile_id == "fast" + )); + assert!(matches!( + started, + ServerChannelMsg::Acp(AcpAppEvent::SessionUpdate { + session_id, + update: AcpSessionUpdate::TurnStarted, + is_replay: false, + }) if session_id == "session-1" + )); + assert!(matches!( + finished, + ServerChannelMsg::Acp(AcpAppEvent::SessionUpdate { + session_id, + update: AcpSessionUpdate::Finished { finish_reason }, + is_replay: false, + }) if session_id == "session-1" && finish_reason == "EndTurn" + )); + + commands::dispatch( + context(&connection, &state, &events), + Command::DeleteSession { + session_id: "old-session".into(), + }, + ) + .await + .expect("delete session"); + + let messages = connection.messages(); + assert_eq!(messages.len(), 3); + assert_eq!(messages[0].method, "session/new"); + assert_eq!( + messages[0].params, + json!({ + "cwd": "/repo", + "mcpServers": [], + "_meta": { "querymt": { "profile_id": "fast" } } + }) + ); + assert_eq!(messages[1].method, "session/prompt"); + assert_eq!( + messages[1].params, + json!({ + "sessionId": "session-1", + "prompt": [ + { "type": "text", "text": "hello" }, + { "type": "resource_link", "name": "guide", "uri": "file:///repo/guide.md" } + ] + }) + ); + assert_eq!( + messages[2], + RecordedMessage { + method: "session/delete".into(), + params: json!({ "sessionId": "old-session" }), + } + ); + assert!(rx.try_recv().is_err()); +} + +#[tokio::test] +async fn catalog_router_pins_config_model_refresh_payloads_and_events() { + let option = json!({ + "id": "mode", + "name": "Mode", + "type": "select", + "currentValue": "plan", + "options": [{ "value": "plan", "name": "Plan" }] + }); + let connection = RecordingConnection::with_responses([ + json!({ "configOptions": [option.clone()] }), + json!({ "configOptions": [] }), + json!({ "configOptions": [] }), + json!({ + "models": [{ + "id": "openai/gpt-5", + "label": "GPT-5", + "provider": "openai", + "model": "gpt-5" + }] + }), + ]); + let (state, events, mut rx) = harness(&connection); + state.set_current_session_id("session-1").await; + + for command in [ + Command::SetAgentMode { + mode: "plan".into(), + }, + Command::SetReasoningEffort { + reasoning_effort: "high".into(), + }, + Command::SetSessionModel { + session_id: "session-1".into(), + model_id: "openai/gpt-5".into(), + node_id: Some("node-1".into()), + }, + Command::ListAllModels { refresh: true }, + ] { + commands::dispatch(context(&connection, &state, &events), command) + .await + .expect("catalog command"); + } + + let messages = connection.messages(); + assert_eq!( + messages + .iter() + .map(|message| message.method.as_str()) + .collect::>(), + [ + "session/set_config_option", + "session/set_config_option", + "session/set_config_option", + "_querymt/refreshModels", + ] + ); + assert_eq!( + messages[0].params, + json!({ "sessionId": "session-1", "configId": "mode", "value": "plan" }) + ); + assert_eq!( + messages[1].params, + json!({ "sessionId": "session-1", "configId": "reasoning_effort", "value": "high" }) + ); + assert_eq!(messages[2].params["sessionId"], "session-1"); + assert_eq!(messages[2].params["configId"], "model"); + assert_eq!(messages[2].params["value"], "openai/gpt-5"); + assert_eq!( + messages[2].params["_meta"]["querymt"]["modelEntry"]["node_id"], + "node-1" + ); + assert_eq!(messages[3].params, json!({ "wait_for_completion": true })); + + assert!(matches!( + rx.try_recv().expect("mode"), + ServerChannelMsg::Acp(AcpAppEvent::AgentMode { mode }) if mode == "plan" + )); + assert!(matches!( + rx.try_recv().expect("model log"), + ServerChannelMsg::Acp(AcpAppEvent::InfoLog { target: "acp", message }) + if message.contains("model=gpt-5") + && message.contains("id=openai/gpt-5") + && message.contains("node=node-1") + )); + assert!(matches!( + rx.try_recv().expect("selected provider"), + ServerChannelMsg::Acp(AcpAppEvent::ProviderChanged { provider, model, .. }) + if provider == "openai" && model == "gpt-5" + )); + assert!(matches!( + rx.try_recv().expect("models"), + ServerChannelMsg::Acp(AcpAppEvent::Models { models, .. }) + if models[0].id == "openai/gpt-5" + )); + assert!(matches!( + rx.try_recv().expect("default provider"), + ServerChannelMsg::Acp(AcpAppEvent::ProviderChanged { provider, model, .. }) + if provider == "openai" && model == "gpt-5" + )); + assert!(rx.try_recv().is_err()); } #[tokio::test] @@ -270,6 +708,369 @@ async fn standard_router_pins_list_and_cancel_wire_shapes() { assert_eq!(messages[1].params["sessionId"], "session"); } +#[tokio::test] +async fn extension_families_pin_auth_history_mesh_profile_methods_payloads_and_events() { + let connection = RecordingConnection::with_responses([ + json!({ + "flow_id": "flow-1", + "provider": "openai", + "authorization_url": "https://example.test/authorize", + "flow_kind": "redirect_code" + }), + json!({ "provider": "openai", "success": true, "message": "connected" }), + json!({ "provider": "openai", "success": false, "message": "disconnect failed" }), + json!({ + "success": true, + "message_id": "u1", + "reverted_files": ["src/lib.rs"], + "message": "undone", + "undo_stack": [{ "message_id": "u1" }] + }), + json!({ "success": true, "message": "redone", "undo_stack": [] }), + json!({ "sessionId": "forked" }), + json!({ + "enabled": true, + "peer_id": "peer-1", + "transport": "relay", + "known_peer_count": 1, + "has_invite_store": true, + "has_mesh_state_store": false, + "scopes": [{ "kind": "team", "id": "scope-1" }] + }), + json!({ + "nodes": [{ + "id": "node-1", + "label": "Remote", + "capabilities": ["sessions"], + "active_sessions": 2, + "transport": "relay", + "last_seen_at": "now" + }] + }), + json!({ + "session_id": "remote-1", + "node_id": "node-1", + "attached": true, + "config_options": [{ "id": "mode" }], + "snapshot": { "cursor": 1 } + }), + json!({ + "session_id": "remote-2", + "node_id": "node-1", + "attached": false, + "config_options": [], + "snapshot": null + }), + json!({ + "profiles": [{ + "id": "fast", + "name": "Fast", + "description": "Fast profile", + "tags": ["quick"], + "source": "local", + "config_kind": "inline", + "fingerprint": "abc" + }], + "active_profile_id": "fast" + }), + json!({ + "profile_id": "fast", + "agents": [{ + "id": "coder", + "name": "Coder", + "description": "Writes code", + "capabilities": ["edit"] + }] + }), + ]); + let (state, events, mut rx) = harness(&connection); + state.set_current_session_id("session-1").await; + + for command in [ + Command::StartOAuthLogin { + provider: "openai".into(), + }, + Command::CompleteOAuthLogin { + flow_id: "flow-1".into(), + response: "code-1".into(), + }, + Command::DisconnectOAuth { + provider: "openai".into(), + }, + Command::Undo { + message_id: "u1".into(), + }, + Command::Redo, + Command::ForkSession { + message_id: "u1".into(), + }, + Command::ListRemoteNodes, + Command::CreateRemoteSession { + node_id: "node-1".into(), + cwd: None, + }, + Command::AttachRemoteSession { + node_id: "node-1".into(), + session_id: "remote-2".into(), + }, + Command::ListProfiles, + Command::ListProfileAgents { + profile_id: "fast".into(), + }, + ] { + commands::dispatch(context(&connection, &state, &events), command) + .await + .expect("extension command"); + } + + assert_eq!( + connection.messages(), + vec![ + RecordedMessage { + method: "_querymt/auth/start".into(), + params: json!({ "provider": "openai" }), + }, + RecordedMessage { + method: "_querymt/auth/complete".into(), + params: json!({ "flow_id": "flow-1", "response": "code-1" }), + }, + RecordedMessage { + method: "_querymt/auth/logout".into(), + params: json!({ "provider": "openai" }), + }, + RecordedMessage { + method: "_querymt/session/undo".into(), + params: json!({ "session_id": "session-1", "message_id": "u1" }), + }, + RecordedMessage { + method: "_querymt/session/redo".into(), + params: json!({ "session_id": "session-1" }), + }, + RecordedMessage { + method: "session/fork".into(), + params: json!({ + "sessionId": "session-1", + "cwd": "/launch", + "_meta": { "querymt": { "message_id": "u1" } } + }), + }, + RecordedMessage { + method: "_querymt/mesh/status".into(), + params: json!({}), + }, + RecordedMessage { + method: "_querymt/mesh/nodes".into(), + params: json!({}), + }, + RecordedMessage { + method: "_querymt/remote/createSession".into(), + params: json!({ "node_id": "node-1", "cwd": null, "attach": true }), + }, + RecordedMessage { + method: "_querymt/remote/attachSession".into(), + params: json!({ "node_id": "node-1", "session_id": "remote-2" }), + }, + RecordedMessage { + method: "_querymt/profiles".into(), + params: json!({}), + }, + RecordedMessage { + method: "_querymt/profile/agents".into(), + params: json!({ "profile_id": "fast" }), + }, + ] + ); + + let emitted = std::iter::from_fn(|| rx.try_recv().ok()).collect::>(); + assert_eq!(emitted.len(), 12); + assert!(matches!( + &emitted[0], + ServerChannelMsg::Acp(AcpAppEvent::OAuthFlowStarted(flow)) + if flow.flow_id == "flow-1" + && flow.provider == "openai" + && flow.authorization_url == "https://example.test/authorize" + && flow.flow_kind == OAuthFlowKind::RedirectCode + )); + assert!(matches!( + &emitted[1], + ServerChannelMsg::Acp(AcpAppEvent::OAuthResult(result)) + if result.provider == "openai" + && result.status == OAuthResultStatus::Success + && result.message == "connected" + )); + assert!(matches!( + &emitted[2], + ServerChannelMsg::Acp(AcpAppEvent::OAuthResult(result)) + if result.provider == "openai" + && result.status == OAuthResultStatus::Failure + && result.message == "disconnect failed" + )); + assert!(matches!( + &emitted[3], + ServerChannelMsg::Acp(AcpAppEvent::UndoResult(result)) + if result == &UndoResult::Applied { + target_message_id: Some("u1".into()), + reverted_files: vec!["src/lib.rs".into()], + message: Some("undone".into()), + stack: UndoStackSnapshot { message_ids: vec!["u1".into()] }, + } + )); + assert!(matches!( + &emitted[4], + ServerChannelMsg::Acp(AcpAppEvent::RedoResult(result)) + if result == &RedoResult::Applied { + message: Some("redone".into()), + stack: UndoStackSnapshot::default(), + } + )); + assert!(matches!( + &emitted[5], + ServerChannelMsg::Acp(AcpAppEvent::ForkResult(result)) + if result == &ForkResult::Succeeded { + source_session_id: Some("session-1".into()), + forked_session_id: Some("forked".into()), + message: None, + } + )); + assert!(matches!( + &emitted[6], + ServerChannelMsg::Acp(AcpAppEvent::MeshStatus(status)) + if status.enabled + && status.peer_id.as_deref() == Some("peer-1") + && status.transport.as_deref() == Some("relay") + && status.known_peer_count == 1 + && status.has_invite_store + && !status.has_mesh_state_store + && status.scopes.len() == 1 + && status.scopes[0].kind == "team" + && status.scopes[0].id == "scope-1" + )); + assert!(matches!( + &emitted[7], + ServerChannelMsg::Acp(AcpAppEvent::MeshNodes(nodes)) + if nodes.nodes.len() == 1 + && nodes.nodes[0].id == "node-1" + && nodes.nodes[0].label == "Remote" + && nodes.nodes[0].capabilities == ["sessions"] + && nodes.nodes[0].active_sessions == 2 + && nodes.nodes[0].transport == "relay" + && nodes.nodes[0].last_seen_at.as_deref() == Some("now") + )); + assert!(matches!( + &emitted[8], + ServerChannelMsg::Acp(AcpAppEvent::RemoteSessionAttached(info)) + if info.session_id == "remote-1" + && info.node_id == "node-1" + && info.attached + && info.config_options == [json!({ "id": "mode" })] + && info.snapshot == Some(json!({ "cursor": 1 })) + )); + assert!(matches!( + &emitted[9], + ServerChannelMsg::Acp(AcpAppEvent::RemoteSessionAttached(info)) + if info.session_id == "remote-2" + && info.node_id == "node-1" + && !info.attached + && info.config_options.is_empty() + && info.snapshot.is_none() + )); + assert!(matches!( + &emitted[10], + ServerChannelMsg::Acp(AcpAppEvent::Profiles { + profiles, + active_profile_id, + }) if profiles.len() == 1 + && profiles[0].id == "fast" + && profiles[0].name == "Fast" + && profiles[0].description.as_deref() == Some("Fast profile") + && profiles[0].tags == ["quick"] + && profiles[0].source.as_deref() == Some("local") + && profiles[0].config_kind.as_deref() == Some("inline") + && profiles[0].fingerprint.as_deref() == Some("abc") + && active_profile_id.as_deref() == Some("fast") + )); + assert!(matches!( + &emitted[11], + ServerChannelMsg::Acp(AcpAppEvent::ProfileAgents { profile_id, agents }) + if profile_id == "fast" + && agents.len() == 1 + && agents[0].id == "coder" + && agents[0].name == "Coder" + && agents[0].description.as_deref() == Some("Writes code") + && agents[0].capabilities == ["edit"] + )); +} + +#[tokio::test] +async fn extension_response_strictness_and_tolerance_are_preserved() { + let strict = RecordingConnection::with_responses([json!({})]); + let (strict_state, strict_events, mut strict_rx) = harness(&strict); + commands::dispatch( + context(&strict, &strict_state, &strict_events), + Command::ListProfiles, + ) + .await + .expect("profile errors are reported as availability events"); + assert!(matches!( + strict_rx.try_recv().expect("strict profile error"), + ServerChannelMsg::Acp(AcpAppEvent::InfoLog { target: "profiles", message }) + if message.starts_with("profile catalog unavailable: ") + )); + + let tolerant = RecordingConnection::with_responses([json!({ "malformed": true })]); + let (tolerant_state, tolerant_events, mut tolerant_rx) = harness(&tolerant); + commands::dispatch( + context(&tolerant, &tolerant_state, &tolerant_events), + Command::StartOAuthLogin { + provider: "openai".into(), + }, + ) + .await + .expect("malformed optional auth result is tolerated"); + assert!(tolerant_rx.try_recv().is_err()); +} + +#[tokio::test] +async fn elicitation_response_dispatches_the_registered_websocket_shape_once() { + let connection = RecordingConnection::default(); + let (state, events, _rx) = harness(&connection); + let (wire_tx, mut wire_rx) = mpsc::unbounded_channel(); + state + .elicitations + .insert( + "e1".into(), + PendingResponse::WebSocketResponse { + peer: Peer::new(wire_tx), + id: json!(41), + }, + ) + .await; + + commands::dispatch( + context(&connection, &state, &events), + Command::ElicitationResponse { + elicitation_id: "e1".into(), + action: "accept".into(), + content: Some(json!({ "selection": "yes" })), + }, + ) + .await + .expect("elicitation response"); + let Message::Text(text) = wire_rx.try_recv().expect("wire response") else { + panic!("text response"); + }; + assert_eq!( + serde_json::from_str::(&text).expect("response JSON"), + json!({ + "jsonrpc": "2.0", + "id": 41, + "result": { "action": "accept", "content": { "selection": "yes" } } + }) + ); + assert!(wire_rx.try_recv().is_err()); + assert!(connection.messages().is_empty()); +} + #[tokio::test] async fn load_emits_loaded_replay_delegation_provider_and_stack_in_order() { let load_response = json!({ @@ -350,14 +1151,19 @@ async fn load_emits_loaded_replay_delegation_provider_and_stack_in_order() { if stack.message_ids == ["u1"] )); assert!(rx.try_recv().is_err()); + let messages = connection.messages(); assert_eq!( - connection - .messages() + messages .iter() .map(|message| message.method.as_str()) .collect::>(), ["session/load", "_querymt/session/undoStack"] ); + assert_eq!( + messages[0].params, + json!({ "sessionId": "session", "cwd": "/repo", "mcpServers": [] }) + ); + assert_eq!(messages[1].params, json!({ "session_id": "session" })); } #[tokio::test] @@ -412,50 +1218,188 @@ async fn session_notification_reaches_application_reducer_coordination() { assert!(rx.try_recv().is_err()); } +fn normalized_event(message: ServerChannelMsg) -> Value { + match message { + ServerChannelMsg::Acp(AcpAppEvent::SessionUpdate { + session_id, + update, + is_replay, + }) => { + let update = match update { + AcpSessionUpdate::AssistantContentDelta { + content, + message_id, + } => { + json!({ "kind": "assistant", "content": content, "messageId": message_id }) + } + AcpSessionUpdate::AssistantThinkingDelta { + content, + message_id, + } => { + json!({ "kind": "thought", "content": content, "messageId": message_id }) + } + AcpSessionUpdate::AssistantMessage { + content, + thinking, + message_id, + } => json!({ + "kind": "assistantMessage", "content": content, + "thinking": thinking, "messageId": message_id + }), + AcpSessionUpdate::ToolCallStart { + tool_call_id, + name, + arguments, + } => json!({ + "kind": "toolStart", "toolCallId": tool_call_id, + "name": name, "arguments": arguments + }), + AcpSessionUpdate::ToolCallEnd { + tool_call_id, + name, + is_error, + result, + } => json!({ + "kind": "toolEnd", "toolCallId": tool_call_id, + "name": name, "isError": is_error, "result": result + }), + AcpSessionUpdate::UsageUpdate { + used, + size, + cost_usd, + } => json!({ + "kind": "usage", "used": used, "size": size, "costUsd": cost_usd + }), + other => panic!("unexpected normalized session update: {other:?}"), + }; + json!({ + "event": "sessionUpdate", "sessionId": session_id, + "update": update, "isReplay": is_replay + }) + } + ServerChannelMsg::Acp(AcpAppEvent::AgentMode { mode }) => { + json!({ "event": "agentMode", "mode": mode }) + } + ServerChannelMsg::Acp(AcpAppEvent::ReasoningEffort { reasoning_effort }) => { + json!({ "event": "reasoningEffort", "value": reasoning_effort }) + } + other => panic!("unexpected normalized event: {other:?}"), + } +} + #[tokio::test] async fn stdio_and_websocket_standard_inbound_share_normalized_events() { - let notification = acp::SessionNotification::new( - "session", - acp::SessionUpdate::UserMessageChunk( - acp::ContentChunk::new(acp::ContentBlock::Text(acp::TextContent::new("hello"))) - .message_id(Some(acp::MessageId::from("u1"))), - ), + let mode_option = acp::SessionConfigOption::select( + "reasoning_effort", + "Reasoning effort", + "high", + vec![acp::SessionConfigSelectOption::new("high", "High")], ); + let notifications = vec![ + acp::SessionNotification::new( + "session", + acp::SessionUpdate::AgentThoughtChunk( + acp::ContentChunk::new(acp::ContentBlock::Text(acp::TextContent::new("think"))) + .message_id(Some(acp::MessageId::from("a1"))), + ), + ), + acp::SessionNotification::new( + "session", + acp::SessionUpdate::AgentMessageChunk( + acp::ContentChunk::new(acp::ContentBlock::Text(acp::TextContent::new("answer"))) + .message_id(Some(acp::MessageId::from("a1"))), + ), + ), + acp::SessionNotification::new( + "session", + acp::SessionUpdate::ToolCall( + acp::ToolCall::new("tool-1", "Run shell").raw_input(json!({ "cmd": "pwd" })), + ), + ), + acp::SessionNotification::new( + "session", + acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new( + "tool-1", + acp::ToolCallUpdateFields::new() + .title("Run shell".to_string()) + .status(acp::ToolCallStatus::Completed) + .raw_output(json!("/repo")), + )), + ), + acp::SessionNotification::new( + "session", + acp::SessionUpdate::CurrentModeUpdate(acp::CurrentModeUpdate::new("plan")), + ), + acp::SessionNotification::new( + "session", + acp::SessionUpdate::ConfigOptionUpdate(acp::ConfigOptionUpdate::new(vec![mode_option])), + ), + acp::SessionNotification::new( + "session", + acp::SessionUpdate::UsageUpdate( + acp::UsageUpdate::new(8, 16).cost(acp::Cost::new(0.5, "usd")), + ), + ), + acp::SessionNotification::new( + "session", + acp::SessionUpdate::SessionInfoUpdate(acp::SessionInfoUpdate::new().title("ignored")), + ), + ]; let (stdio_tx, mut stdio_rx) = mpsc::unbounded_channel(); let stdio_events = EventSink::new(stdio_tx); let stdio_state = Arc::new(RuntimeState::new(None)); - inbound::session_notification(&stdio_state, &stdio_events, notification.clone()).await; - let (wire_tx, _wire_rx) = mpsc::unbounded_channel::(); let peer = Peer::new(wire_tx); let connection = RecordingConnection::default(); let (ws_tx, mut ws_rx) = mpsc::unbounded_channel(); let ws_events = EventSink::new(ws_tx); let ws_state = Arc::new(RuntimeState::new(None)); - let text = serde_json::to_string(&json!({ - "jsonrpc": "2.0", - "method": "session/update", - "params": serde_json::to_value(notification).expect("notification params") - })) - .expect("websocket envelope"); - inbound::websocket_text(&peer, &connection, &ws_state, &ws_events, &text) - .await - .expect("websocket inbound"); - let extract = |message| match message { - ServerChannelMsg::Acp(AcpAppEvent::SessionUpdate { - session_id, - update: - AcpSessionUpdate::UserMessage { - content, - message_id, - }, - is_replay, - }) => (session_id, content, message_id, is_replay), - other => panic!("unexpected event: {other:?}"), - }; + for notification in notifications { + inbound::session_notification(&stdio_state, &stdio_events, notification.clone()).await; + let text = serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": serde_json::to_value(notification).expect("notification params") + })) + .expect("websocket envelope"); + inbound::websocket_text(&peer, &connection, &ws_state, &ws_events, &text) + .await + .expect("websocket inbound"); + } + + let stdio = std::iter::from_fn(|| stdio_rx.try_recv().ok()) + .map(normalized_event) + .collect::>(); + let websocket = std::iter::from_fn(|| ws_rx.try_recv().ok()) + .map(normalized_event) + .collect::>(); + assert_eq!(stdio, websocket); assert_eq!( - extract(stdio_rx.try_recv().expect("stdio event")), - extract(ws_rx.try_recv().expect("websocket event")) + stdio, + vec![ + json!({ "event": "sessionUpdate", "sessionId": "session", "update": { + "kind": "thought", "content": "think", "messageId": "a1" + }, "isReplay": false }), + json!({ "event": "sessionUpdate", "sessionId": "session", "update": { + "kind": "assistant", "content": "answer", "messageId": "a1" + }, "isReplay": false }), + json!({ "event": "sessionUpdate", "sessionId": "session", "update": { + "kind": "assistantMessage", "content": "answer", "thinking": "think", "messageId": "a1" + }, "isReplay": false }), + json!({ "event": "sessionUpdate", "sessionId": "session", "update": { + "kind": "toolStart", "toolCallId": "tool-1", "name": "shell", + "arguments": { "cmd": "pwd" } + }, "isReplay": false }), + json!({ "event": "sessionUpdate", "sessionId": "session", "update": { + "kind": "toolEnd", "toolCallId": "tool-1", "name": "shell", + "isError": false, "result": "/repo" + }, "isReplay": false }), + json!({ "event": "agentMode", "mode": "plan" }), + json!({ "event": "reasoningEffort", "value": "high" }), + json!({ "event": "sessionUpdate", "sessionId": "session", "update": { + "kind": "usage", "used": 8, "size": 16, "costUsd": 0.5 + }, "isReplay": false }), + ] ); } diff --git a/src/acp/inbound.rs b/src/acp/inbound.rs index 7633b2a..450c751 100644 --- a/src/acp/inbound.rs +++ b/src/acp/inbound.rs @@ -243,9 +243,110 @@ fn string_alias(params: &Value, camel: &str, snake: &str) -> Option { #[cfg(test)] mod tests { + use std::collections::VecDeque; + use std::future::Future; + use std::sync::Mutex; + use std::time::Duration; + + use agent_client_protocol::{JsonRpcNotification, JsonRpcRequest, JsonRpcResponse}; use tokio::sync::mpsc; + use tokio_tungstenite::tungstenite::Message; use super::*; + use crate::runtime_events::ServerChannelMsg; + + #[derive(Clone, Default)] + struct TestConnection { + messages: Arc>>, + responses: Arc>>, + } + + impl TestConnection { + fn with_responses(responses: impl IntoIterator) -> Self { + Self { + responses: Arc::new(Mutex::new(responses.into_iter().collect())), + ..Self::default() + } + } + + fn messages(&self) -> Vec<(String, Value)> { + self.messages + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } + } + + impl AcpConnection for TestConnection { + async fn request(&self, request: R) -> Result + where + R: JsonRpcRequest + Send + Sync + 'static, + R::Response: Send + 'static, + { + let message = request.to_untyped_message()?; + self.messages + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push((message.method.clone(), message.params)); + let response = self + .responses + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .pop_front() + .unwrap_or_else(|| json!({})); + R::Response::from_value(&message.method, response) + } + + fn notify(&self, notification: N) -> Result<(), acp_sdk::Error> + where + N: JsonRpcNotification + Send + Sync + 'static, + { + let message = notification.to_untyped_message()?; + self.messages + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push((message.method, message.params)); + Ok(()) + } + + fn spawn( + &self, + future: impl Future> + Send + 'static, + ) -> Result<(), acp_sdk::Error> { + tokio::spawn(async move { + let _ = future.await; + }); + Ok(()) + } + } + + fn websocket_harness() -> ( + Peer, + mpsc::UnboundedReceiver, + Arc, + EventSink, + mpsc::UnboundedReceiver, + ) { + let (wire_tx, wire_rx) = mpsc::unbounded_channel(); + let (event_tx, event_rx) = mpsc::unbounded_channel(); + ( + Peer::new(wire_tx), + wire_rx, + Arc::new(RuntimeState::new(None)), + EventSink::new(event_tx), + event_rx, + ) + } + + fn envelope(method: &str, id: Option, params: Value) -> String { + serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "id": id, + "method": method, + "params": params, + })) + .expect("envelope JSON") + } fn delegation_value(version: u32) -> Value { json!({ @@ -274,10 +375,16 @@ mod tests { extension_notification("querymt/models/changed"), Some(ExtensionNotification::ModelsChanged) ); - assert_eq!( - extension_notification("querymt/mesh/joined"), - Some(ExtensionNotification::MeshChanged) - ); + for method in [ + "querymt/mesh/nodesChanged", + "querymt/mesh/joined", + "querymt/mesh/peerExpired", + ] { + assert_eq!( + extension_notification(method), + Some(ExtensionNotification::MeshChanged) + ); + } assert_eq!(extension_notification("session/update"), None); } @@ -315,4 +422,366 @@ mod tests { && message.starts_with("invalid delegation notification: ") )); } + + #[tokio::test] + async fn models_changed_refreshes_normalized_state_and_event() { + let connection = TestConnection::with_responses([json!({ + "models": [{ + "id": "openai/gpt-5", + "label": "GPT-5", + "source": "remote", + "provider": "openai", + "model": "gpt-5", + "node_id": "node-1", + "node_label": "Remote Node", + "family": "gpt", + "quant": "fp16" + }], + "meta": { + "stale": false, + "refresh_in_progress": false, + "remote_node_count": 2, + "remote_timeout_count": 1 + } + })]); + let (peer, _wire_rx, state, events, mut event_rx) = websocket_harness(); + + websocket_text( + &peer, + &connection, + &state, + &events, + &envelope("querymt/models/changed", None, json!({})), + ) + .await + .expect("models changed notification"); + + let event = tokio::time::timeout(Duration::from_secs(1), event_rx.recv()) + .await + .expect("model refresh timeout") + .expect("models event"); + assert_eq!( + connection.messages(), + vec![("_querymt/models".into(), json!({}))] + ); + assert!(matches!( + event, + ServerChannelMsg::Acp(AcpAppEvent::Models { models, meta: Some(meta) }) + if models.len() == 1 + && models[0].id == "openai/gpt-5" + && models[0].label == "GPT-5" + && models[0].provider == "openai" + && models[0].model == "gpt-5" + && models[0].node_id.as_deref() == Some("node-1") + && models[0].node_label.as_deref() == Some("Remote Node") + && models[0].family.as_deref() == Some("gpt") + && models[0].quant.as_deref() == Some("fp16") + && meta.remote_node_count == 2 + && meta.remote_timeout_count == 1 + )); + let stored = state + .model_by_id("openai/gpt-5") + .await + .expect("normalized model state"); + assert_eq!(stored.id, "openai/gpt-5"); + assert_eq!(stored.label, "GPT-5"); + assert_eq!(stored.source.as_deref(), Some("remote")); + assert_eq!(stored.provider, "openai"); + assert_eq!(stored.model, "gpt-5"); + assert_eq!(stored.node_id.as_deref(), Some("node-1")); + assert_eq!(stored.node_label.as_deref(), Some("Remote Node")); + assert_eq!(stored.family.as_deref(), Some("gpt")); + assert_eq!(stored.quant.as_deref(), Some("fp16")); + assert!(event_rx.try_recv().is_err()); + } + + #[tokio::test] + async fn mesh_aliases_route_to_the_same_refresh_request_and_event() { + for method in [ + "querymt/mesh/nodesChanged", + "querymt/mesh/joined", + "querymt/mesh/peerExpired", + ] { + let connection = TestConnection::with_responses([json!({ + "nodes": [{ "id": "node-1", "label": "Remote" }] + })]); + let (peer, _wire_rx, state, events, mut event_rx) = websocket_harness(); + websocket_text( + &peer, + &connection, + &state, + &events, + &envelope(method, None, json!({})), + ) + .await + .expect("mesh notification"); + + let event = tokio::time::timeout(Duration::from_secs(1), event_rx.recv()) + .await + .expect("mesh refresh timeout") + .expect("mesh event"); + assert_eq!( + connection.messages(), + vec![("_querymt/mesh/nodes".into(), json!({}))] + ); + assert!(matches!( + event, + ServerChannelMsg::Acp(AcpAppEvent::MeshNodes(nodes)) + if nodes.nodes[0].id == "node-1" + )); + assert!(event_rx.try_recv().is_err()); + } + } + + #[tokio::test] + async fn malformed_known_input_errors_while_unknown_methods_are_ignored() { + let connection = TestConnection::default(); + let (peer, mut wire_rx, state, events, mut event_rx) = websocket_harness(); + assert!( + websocket_text(&peer, &connection, &state, &events, "{") + .await + .is_err() + ); + assert!( + websocket_text( + &peer, + &connection, + &state, + &events, + &envelope("session/update", None, json!({ "sessionId": 7 })), + ) + .await + .is_err() + ); + websocket_text( + &peer, + &connection, + &state, + &events, + &envelope("querymt/unknown", Some(json!(9)), json!({ "bad": true })), + ) + .await + .expect("unknown ignored"); + + assert!(connection.messages().is_empty()); + assert!(wire_rx.try_recv().is_err()); + assert!(event_rx.try_recv().is_err()); + } + + #[tokio::test] + async fn response_envelopes_resolve_the_matching_peer_request() { + let connection = TestConnection::default(); + let (peer, mut wire_rx, state, events, _event_rx) = websocket_harness(); + let request_peer = peer.clone(); + let pending = tokio::spawn(async move { + request_peer + .request("querymt/test", json!({ "value": 1 })) + .await + }); + let Message::Text(text) = wire_rx.recv().await.expect("request frame") else { + panic!("text request"); + }; + let request: Value = serde_json::from_str(&text).expect("request JSON"); + + websocket_text( + &peer, + &connection, + &state, + &events, + &serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "id": request["id"], + "result": { "ok": true } + })) + .expect("response JSON"), + ) + .await + .expect("response dispatch"); + assert_eq!( + pending + .await + .expect("request task") + .expect("request result"), + json!({ "ok": true }) + ); + } + + #[tokio::test] + async fn permission_request_responds_with_allow_once_wire_shape() { + let connection = TestConnection::default(); + let (peer, mut wire_rx, state, events, mut event_rx) = websocket_harness(); + let request = acp::RequestPermissionRequest::new( + "session-1", + acp::ToolCallUpdate::new("tool-1", acp::ToolCallUpdateFields::new()), + vec![ + acp::PermissionOption::new( + "reject", + "Reject", + acp::PermissionOptionKind::RejectOnce, + ), + acp::PermissionOption::new("allow", "Allow", acp::PermissionOptionKind::AllowOnce), + ], + ); + websocket_text( + &peer, + &connection, + &state, + &events, + &envelope( + "session/request_permission", + Some(json!(7)), + serde_json::to_value(request).expect("permission params"), + ), + ) + .await + .expect("permission request"); + + let Message::Text(text) = wire_rx.try_recv().expect("permission response") else { + panic!("text response"); + }; + assert_eq!( + serde_json::from_str::(&text).expect("response JSON"), + json!({ + "jsonrpc": "2.0", + "id": 7, + "result": { "outcome": { "outcome": "selected", "optionId": "allow" } } + }) + ); + assert!(event_rx.try_recv().is_err()); + } + + fn direct_elicitation() -> acp::CreateElicitationRequest { + acp::CreateElicitationRequest::new( + acp::ElicitationFormMode::new( + acp::ElicitationSessionScope::new("session-1"), + acp::ElicitationSchema::new().string("selection", true), + ), + "Choose", + ) + .meta(serde_json::Map::from_iter([( + "querymt".to_string(), + json!({ "source": "test", "allow_custom": true }), + )])) + } + + #[tokio::test] + async fn direct_elicitation_registers_and_dispatches_exact_response() { + let connection = TestConnection::default(); + let (peer, mut wire_rx, state, events, mut event_rx) = websocket_harness(); + websocket_text( + &peer, + &connection, + &state, + &events, + &envelope( + "elicitation/create", + Some(json!("e-direct")), + serde_json::to_value(direct_elicitation()).expect("elicitation params"), + ), + ) + .await + .expect("direct elicitation"); + assert!(matches!( + event_rx.try_recv().expect("elicitation event"), + ServerChannelMsg::Acp(AcpAppEvent::SessionUpdate { + session_id, + update: AcpSessionUpdate::ElicitationRequested { + elicitation_id, + message, + source, + allow_custom: true, + .. + }, + is_replay: false, + }) if session_id == "session-1" + && elicitation_id == "e-direct" + && message == "Choose" + && source == "test" + )); + + state + .elicitations + .respond("e-direct", "accept", Some(json!({ "selection": "yes" }))) + .await; + let Message::Text(text) = wire_rx.try_recv().expect("elicitation response") else { + panic!("text response"); + }; + assert_eq!( + serde_json::from_str::(&text).expect("response JSON"), + json!({ + "jsonrpc": "2.0", + "id": "e-direct", + "result": { "action": "accept", "content": { "selection": "yes" } } + }) + ); + } + + #[tokio::test] + async fn requested_elicitation_accepts_aliases_and_exact_custom_metadata() { + for params in [ + json!({ + "elicitationId": "e-camel", + "sessionId": "session-camel", + "message": "Camel", + "requestedSchema": { "type": "object" }, + "source": "extension", + "allowCustom": true + }), + json!({ + "elicitation_id": "e-snake", + "session_id": "session-snake", + "message": "Snake", + "requested_schema": { "type": "string" }, + "source": "extension", + "_meta": { "querymt": { "allow_custom": true } } + }), + ] { + let expected_id = params + .get("elicitationId") + .or_else(|| params.get("elicitation_id")) + .and_then(Value::as_str) + .expect("elicitation id") + .to_string(); + let expected_session = params + .get("sessionId") + .or_else(|| params.get("session_id")) + .and_then(Value::as_str) + .expect("session id") + .to_string(); + let expected_schema = params + .get("requestedSchema") + .or_else(|| params.get("requested_schema")) + .cloned() + .expect("schema"); + let connection = TestConnection::default(); + let (peer, _wire_rx, state, events, mut event_rx) = websocket_harness(); + websocket_text( + &peer, + &connection, + &state, + &events, + &envelope("elicitation/requested", None, params), + ) + .await + .expect("requested elicitation"); + + assert!(matches!( + event_rx.try_recv().expect("elicitation event"), + ServerChannelMsg::Acp(AcpAppEvent::SessionUpdate { + session_id, + update: AcpSessionUpdate::ElicitationRequested { + elicitation_id, + requested_schema, + source, + allow_custom: true, + .. + }, + is_replay: false, + }) if session_id == expected_session + && elicitation_id == expected_id + && requested_schema == expected_schema + && source == "extension" + )); + } + } } diff --git a/src/acp/notification.rs b/src/acp/notification.rs index ce07a28..72b00c7 100644 --- a/src/acp/notification.rs +++ b/src/acp/notification.rs @@ -315,17 +315,178 @@ mod tests { } #[test] - fn usage_translation_keeps_only_usd_cost() { - let (_, translated) = translate(notification(acp::SessionUpdate::UsageUpdate( - acp::UsageUpdate::new(5, 10).cost(acp::Cost::new(0.25, "USD")), + fn tool_start_and_pending_updates_preserve_names_and_arguments() { + let (_, started) = translate(notification(acp::SessionUpdate::ToolCall( + acp::ToolCall::new("tool-1", "Run shell") + .raw_input(serde_json::json!({ "cmd": "cargo test" })), ))); assert!(matches!( - translated, - Translation::Update(AcpSessionUpdate::UsageUpdate { - used: 5, - size: 10, - cost_usd: Some(0.25) - }) + started, + Translation::ToolStart(AcpSessionUpdate::ToolCallStart { + tool_call_id: Some(id), + name, + arguments: Some(arguments), + }) if id == "tool-1" + && name == "shell" + && arguments == serde_json::json!({ "cmd": "cargo test" }) + )); + + let (_, pending) = translate(notification(acp::SessionUpdate::ToolCallUpdate( + acp::ToolCallUpdate::new( + "tool-2", + acp::ToolCallUpdateFields::new().raw_input(serde_json::json!({ "path": "src" })), + ), + ))); + assert!(matches!( + pending, + Translation::Update(AcpSessionUpdate::ToolCallStart { + tool_call_id: Some(id), + name, + arguments: Some(arguments), + }) if id == "tool-2" + && name == "tool" + && arguments == serde_json::json!({ "path": "src" }) )); } + + #[test] + fn completed_and_failed_tool_updates_choose_raw_output_then_content() { + let (_, completed) = translate(notification(acp::SessionUpdate::ToolCallUpdate( + acp::ToolCallUpdate::new( + "tool-1", + acp::ToolCallUpdateFields::new() + .title("Run shell".to_string()) + .status(acp::ToolCallStatus::Completed) + .raw_output(serde_json::json!({ "exit": 0 })) + .content(vec![ + acp::ContentBlock::Text(acp::TextContent::new("fallback")).into(), + ]), + ), + ))); + assert!(matches!( + completed, + Translation::Update(AcpSessionUpdate::ToolCallEnd { + tool_call_id: Some(id), + name, + is_error: false, + result: Some(result), + }) if id == "tool-1" && name == "shell" && result == r#"{"exit":0}"# + )); + + let (_, failed) = translate(notification(acp::SessionUpdate::ToolCallUpdate( + acp::ToolCallUpdate::new( + "tool-2", + acp::ToolCallUpdateFields::new() + .status(acp::ToolCallStatus::Failed) + .content(vec![ + acp::ContentBlock::Text(acp::TextContent::new("failure")).into(), + acp::ContentBlock::ResourceLink(acp::ResourceLink::new( + "log", + "file:///tmp/error.log", + )) + .into(), + ]), + ), + ))); + let Translation::Update(AcpSessionUpdate::ToolCallEnd { + tool_call_id: Some(id), + name, + is_error, + result: Some(result), + }) = failed + else { + panic!("unexpected failed tool translation: {failed:?}"); + }; + assert_eq!(id, "tool-2"); + assert_eq!(name, "tool"); + assert!(is_error); + assert_eq!( + result, + "{\"type\":\"content\",\"content\":{\"type\":\"text\",\"text\":\"failure\"}}\n{\"type\":\"content\",\"content\":{\"type\":\"resource_link\",\"name\":\"log\",\"uri\":\"file:///tmp/error.log\"}}" + ); + } + + #[test] + fn assistant_content_converts_resource_links_and_non_text_blocks() { + let (_, resource) = translate(notification(acp::SessionUpdate::AgentMessageChunk( + acp::ContentChunk::new(acp::ContentBlock::ResourceLink(acp::ResourceLink::new( + "guide", + "file:///repo/guide.md", + ))), + ))); + assert!(matches!( + resource, + Translation::AssistantChunk { text, thinking: false, .. } + if text == "file:///repo/guide.md" + )); + + let (_, image) = translate(notification(acp::SessionUpdate::AgentThoughtChunk( + acp::ContentChunk::new(acp::ContentBlock::Image(acp::ImageContent::new( + "YWJj", + "image/png", + ))), + ))); + assert!(matches!( + image, + Translation::AssistantChunk { text, thinking: true, .. } + if text == r#"{"type":"image","data":"YWJj","mimeType":"image/png"}"# + )); + } + + #[test] + fn mode_config_and_ignored_variants_stay_at_the_translation_boundary() { + let option = acp::SessionConfigOption::select( + "mode", + "Mode", + "plan", + vec![acp::SessionConfigSelectOption::new("plan", "Plan")], + ); + let (_, mode) = translate(notification(acp::SessionUpdate::CurrentModeUpdate( + acp::CurrentModeUpdate::new("plan"), + ))); + assert!(matches!(mode, Translation::AgentMode(value) if value == "plan")); + + let (_, config) = translate(notification(acp::SessionUpdate::ConfigOptionUpdate( + acp::ConfigOptionUpdate::new(vec![option.clone()]), + ))); + assert!(matches!( + config, + Translation::ConfigOptions(options) if options == vec![option] + )); + + for update in [ + acp::SessionUpdate::SessionInfoUpdate(acp::SessionInfoUpdate::new().title("ignored")), + acp::SessionUpdate::Plan(acp::Plan::new(vec![])), + acp::SessionUpdate::AvailableCommandsUpdate(acp::AvailableCommandsUpdate::new(vec![])), + ] { + assert!(matches!( + translate(notification(update)).1, + Translation::Ignore + )); + } + } + + #[test] + fn usage_translation_keeps_only_usd_cost() { + for (cost, expected) in [ + (Some(acp::Cost::new(0.25, "USD")), Some(0.25)), + (Some(acp::Cost::new(1.5, "eur")), None), + (None, None), + ] { + let update = if let Some(cost) = cost { + acp::UsageUpdate::new(5, 10).cost(cost) + } else { + acp::UsageUpdate::new(5, 10) + }; + let (_, translated) = translate(notification(acp::SessionUpdate::UsageUpdate(update))); + assert!(matches!( + translated, + Translation::Update(AcpSessionUpdate::UsageUpdate { + used: 5, + size: 10, + cost_usd, + }) if cost_usd == expected + )); + } + } } diff --git a/src/acp/transport/jsonrpc.rs b/src/acp/transport/jsonrpc.rs index 5302d15..913f494 100644 --- a/src/acp/transport/jsonrpc.rs +++ b/src/acp/transport/jsonrpc.rs @@ -190,6 +190,51 @@ mod tests { (task, envelope) } + fn wire_value(rx: &mut mpsc::UnboundedReceiver) -> Value { + let Message::Text(text) = rx.try_recv().expect("wire message") else { + panic!("expected text frame"); + }; + serde_json::from_str(&text).expect("wire JSON") + } + + #[test] + fn notify_omits_id_and_respond_selects_result_or_error() { + let (tx, mut rx) = mpsc::unbounded_channel(); + let peer = Peer::new(tx); + + peer.notify("session/update", json!({ "value": 1 })) + .expect("notification"); + peer.respond(json!(7), Ok(json!({ "accepted": true }))) + .expect("successful response"); + peer.respond(json!(8), Err(internal_error("permission denied"))) + .expect("error response"); + + assert_eq!( + wire_value(&mut rx), + json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { "value": 1 } + }) + ); + assert_eq!( + wire_value(&mut rx), + json!({ + "jsonrpc": "2.0", + "id": 7, + "result": { "accepted": true } + }) + ); + let error = wire_value(&mut rx); + assert_eq!(error["jsonrpc"], "2.0"); + assert_eq!(error["id"], 8); + assert!(error.get("result").is_none()); + assert_eq!(error["error"]["code"], -32603); + assert_eq!(error["error"]["message"], "Internal error"); + assert_eq!(error["error"]["data"], "permission denied"); + assert!(rx.try_recv().is_err()); + } + #[tokio::test] async fn ids_start_at_one_and_out_of_order_responses_correlate() { let (tx, mut rx) = mpsc::unbounded_channel(); diff --git a/src/acp/transport/websocket.rs b/src/acp/transport/websocket.rs index 4e5b574..a2bd923 100644 --- a/src/acp/transport/websocket.rs +++ b/src/acp/transport/websocket.rs @@ -347,6 +347,97 @@ mod tests { server.abort(); } + #[tokio::test] + async fn socket_close_fails_pending_prompt_once_before_connection_error() { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind WebSocket listener"); + let url = format!("ws://{}", listener.local_addr().expect("listener address")); + let (prompt_seen_tx, prompt_seen_rx) = oneshot::channel(); + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.expect("accept WebSocket client"); + let mut socket = accept_async(stream).await.expect("accept WebSocket"); + + let Message::Text(text) = socket + .next() + .await + .expect("new-session request") + .expect("valid new-session message") + else { + panic!("text new-session request"); + }; + let request: serde_json::Value = serde_json::from_str(&text).expect("request JSON"); + socket + .send(Message::Text( + serde_json::json!({ + "jsonrpc": "2.0", + "id": request["id"], + "result": { "sessionId": "session-1" } + }) + .to_string() + .into(), + )) + .await + .expect("send new-session response"); + + let Message::Text(text) = socket + .next() + .await + .expect("prompt request") + .expect("valid prompt message") + else { + panic!("text prompt request"); + }; + let request: serde_json::Value = serde_json::from_str(&text).expect("prompt JSON"); + assert_eq!(request["method"], "session/prompt"); + prompt_seen_tx.send(()).expect("signal pending prompt"); + socket.close(None).await.expect("close socket"); + }); + + let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel(); + let (srv_tx, mut srv_rx) = mpsc::unbounded_channel(); + let (conn_tx, _conn_rx) = mpsc::unbounded_channel(); + let connection_task = + tokio::spawn(async move { run(url, &mut cmd_rx, srv_tx, conn_tx, None).await }); + cmd_tx + .send(Command::NewSession { + cwd: None, + profile_id: None, + }) + .expect("send new-session command"); + cmd_tx + .send(Command::Prompt { + prompt: vec![PromptBlock::Text { + text: "hello".to_string(), + }], + local_id: "local-close".to_string(), + }) + .expect("send prompt command"); + tokio::time::timeout(Duration::from_secs(1), prompt_seen_rx) + .await + .expect("prompt reached server") + .expect("prompt signal"); + + let error = tokio::time::timeout(Duration::from_secs(1), connection_task) + .await + .expect("connection returned") + .expect("connection task") + .expect_err("socket closure error"); + assert!(error.to_string().contains("connection closed")); + let failures = std::iter::from_fn(|| srv_rx.try_recv().ok()) + .filter_map(|message| match message { + ServerChannelMsg::Acp(AcpAppEvent::PromptFailed { local_id, message }) => { + Some((local_id, message)) + } + _ => None, + }) + .collect::>(); + assert_eq!(failures.len(), 1); + assert_eq!(failures[0].0, "local-close"); + assert!(failures[0].1.contains("socket closed")); + server.await.expect("server task"); + } + #[tokio::test] async fn abort_spawned_cancels_owned_prompt_work() { let (tx, _rx) = mpsc::unbounded_channel();